Commit bdb0f159 authored by Stefan Metzmacher's avatar Stefan Metzmacher Committed by Steve French
Browse files

smb: server: make use of smbdirect_socket.recv_io.reassembly.* 



This is also used by the client and will allow us to introduce
common helper functions soon.

Cc: Namjae Jeon <linkinjeon@kernel.org>
Cc: Steve French <smfrench@gmail.com>
Cc: Tom Talpey <tom@talpey.com>
Cc: linux-cifs@vger.kernel.org
Cc: samba-technical@lists.samba.org
Signed-off-by: default avatarStefan Metzmacher <metze@samba.org>
Acked-by: default avatarNamjae Jeon <linkinjeon@kernel.org>
Signed-off-by: default avatarSteve French <stfrench@microsoft.com>
parent d9989207

fs/smb/server/transport_rdma.c

+43 −47
Original line number Diff line number Diff line
@@ -95,16 +95,8 @@ struct smb_direct_transport { 


	struct smbdirect_socket socket;



	bool			full_packet_received;

	wait_queue_head_t	wait_status;



	spinlock_t		reassembly_queue_lock;

	struct list_head	reassembly_queue;

	int			reassembly_data_length;

	int			reassembly_queue_length;

	int			first_entry_offset;

	wait_queue_head_t	wait_reassembly_queue;



	spinlock_t		receive_credit_lock;

	int			recv_credits;

	int			count_avail_recvmsg;
@@ -255,9 +247,11 @@ static void enqueue_reassembly(struct smb_direct_transport *t, 
			       struct smbdirect_recv_io *recvmsg,

			       int data_length)

{

	spin_lock(&t->reassembly_queue_lock);

	list_add_tail(&recvmsg->list, &t->reassembly_queue);

	t->reassembly_queue_length++;

	struct smbdirect_socket *sc = &t->socket;



	spin_lock(&sc->recv_io.reassembly.lock);

	list_add_tail(&recvmsg->list, &sc->recv_io.reassembly.list);

	sc->recv_io.reassembly.queue_length++;

	/*

	 * Make sure reassembly_data_length is updated after list and

	 * reassembly_queue_length are updated. On the dequeue side
@@ -265,14 +259,16 @@ static void enqueue_reassembly(struct smb_direct_transport *t, 
	 * if reassembly_queue_length and list is up to date

	 */

	virt_wmb();

	t->reassembly_data_length += data_length;

	spin_unlock(&t->reassembly_queue_lock);

	sc->recv_io.reassembly.data_length += data_length;

	spin_unlock(&sc->recv_io.reassembly.lock);

}



static struct smbdirect_recv_io *get_first_reassembly(struct smb_direct_transport *t)

{

	if (!list_empty(&t->reassembly_queue))

		return list_first_entry(&t->reassembly_queue,

	struct smbdirect_socket *sc = &t->socket;



	if (!list_empty(&sc->recv_io.reassembly.list))

		return list_first_entry(&sc->recv_io.reassembly.list,

				struct smbdirect_recv_io, list);

	else

		return NULL;
@@ -337,11 +333,11 @@ static struct smb_direct_transport *alloc_transport(struct rdma_cm_id *cm_id) 
	sc->status = SMBDIRECT_SOCKET_CREATED;

	init_waitqueue_head(&t->wait_status);



	spin_lock_init(&t->reassembly_queue_lock);

	INIT_LIST_HEAD(&t->reassembly_queue);

	t->reassembly_data_length = 0;

	t->reassembly_queue_length = 0;

	init_waitqueue_head(&t->wait_reassembly_queue);

	spin_lock_init(&sc->recv_io.reassembly.lock);

	INIT_LIST_HEAD(&sc->recv_io.reassembly.list);

	sc->recv_io.reassembly.data_length = 0;

	sc->recv_io.reassembly.queue_length = 0;

	init_waitqueue_head(&sc->recv_io.reassembly.wait_queue);

	init_waitqueue_head(&t->wait_send_credits);

	init_waitqueue_head(&t->wait_rw_credits);
@@ -398,17 +394,17 @@ static void free_transport(struct smb_direct_transport *t) 


	ksmbd_debug(RDMA, "drain the reassembly queue\n");

	do {

		spin_lock(&t->reassembly_queue_lock);

		spin_lock(&sc->recv_io.reassembly.lock);

		recvmsg = get_first_reassembly(t);

		if (recvmsg) {

			list_del(&recvmsg->list);

			spin_unlock(&t->reassembly_queue_lock);

			spin_unlock(&sc->recv_io.reassembly.lock);

			put_recvmsg(t, recvmsg);

		} else {

			spin_unlock(&t->reassembly_queue_lock);

			spin_unlock(&sc->recv_io.reassembly.lock);

		}

	} while (recvmsg);

	t->reassembly_data_length = 0;

	sc->recv_io.reassembly.data_length = 0;



	if (sc->ib.send_cq)

		ib_free_cq(sc->ib.send_cq);
@@ -542,7 +538,7 @@ static void recv_done(struct ib_cq *cq, struct ib_wc *wc) 
			return;

		}

		t->negotiation_requested = true;

		t->full_packet_received = true;

		sc->recv_io.reassembly.full_packet_received = true;

		sc->status = SMBDIRECT_SOCKET_CONNECTED;

		enqueue_reassembly(t, recvmsg, 0);

		wake_up_interruptible(&t->wait_status);
@@ -579,13 +575,13 @@ static void recv_done(struct ib_cq *cq, struct ib_wc *wc) 
		}



		if (data_length) {

			if (t->full_packet_received)

			if (sc->recv_io.reassembly.full_packet_received)

				recvmsg->first_segment = true;



			if (le32_to_cpu(data_transfer->remaining_data_length))

				t->full_packet_received = false;

				sc->recv_io.reassembly.full_packet_received = false;

			else

				t->full_packet_received = true;

				sc->recv_io.reassembly.full_packet_received = true;



			spin_lock(&t->receive_credit_lock);

			receive_credits = --(t->recv_credits);
@@ -615,7 +611,7 @@ static void recv_done(struct ib_cq *cq, struct ib_wc *wc) 


		if (data_length) {

			enqueue_reassembly(t, recvmsg, (int)data_length);

			wake_up_interruptible(&t->wait_reassembly_queue);

			wake_up_interruptible(&sc->recv_io.reassembly.wait_queue);

		} else

			put_recvmsg(t, recvmsg);
@@ -693,7 +689,7 @@ static int smb_direct_read(struct ksmbd_transport *t, char *buf, 
	 * the only one reading from the front of the queue. The transport

	 * may add more entries to the back of the queue at the same time

	 */

	if (st->reassembly_data_length >= size) {

	if (sc->recv_io.reassembly.data_length >= size) {

		int queue_length;

		int queue_removed = 0;
@@ -705,10 +701,10 @@ static int smb_direct_read(struct ksmbd_transport *t, char *buf, 
		 * updated in SOFTIRQ as more data is received

		 */

		virt_rmb();

		queue_length = st->reassembly_queue_length;

		queue_length = sc->recv_io.reassembly.queue_length;

		data_read = 0;

		to_read = size;

		offset = st->first_entry_offset;

		offset = sc->recv_io.reassembly.first_entry_offset;

		while (data_read < size) {

			recvmsg = get_first_reassembly(st);

			data_transfer = smbdirect_recv_io_payload(recvmsg);
@@ -751,9 +747,9 @@ static int smb_direct_read(struct ksmbd_transport *t, char *buf, 
				if (queue_length) {

					list_del(&recvmsg->list);

				} else {

					spin_lock_irq(&st->reassembly_queue_lock);

					spin_lock_irq(&sc->recv_io.reassembly.lock);

					list_del(&recvmsg->list);

					spin_unlock_irq(&st->reassembly_queue_lock);

					spin_unlock_irq(&sc->recv_io.reassembly.lock);

				}

				queue_removed++;

				put_recvmsg(st, recvmsg);
@@ -766,10 +762,10 @@ static int smb_direct_read(struct ksmbd_transport *t, char *buf, 
			data_read += to_copy;

		}



		spin_lock_irq(&st->reassembly_queue_lock);

		st->reassembly_data_length -= data_read;

		st->reassembly_queue_length -= queue_removed;

		spin_unlock_irq(&st->reassembly_queue_lock);

		spin_lock_irq(&sc->recv_io.reassembly.lock);

		sc->recv_io.reassembly.data_length -= data_read;

		sc->recv_io.reassembly.queue_length -= queue_removed;

		spin_unlock_irq(&sc->recv_io.reassembly.lock);



		spin_lock(&st->receive_credit_lock);

		st->count_avail_recvmsg += queue_removed;
@@ -780,18 +776,18 @@ static int smb_direct_read(struct ksmbd_transport *t, char *buf, 
			spin_unlock(&st->receive_credit_lock);

		}



		st->first_entry_offset = offset;

		sc->recv_io.reassembly.first_entry_offset = offset;

		ksmbd_debug(RDMA,

			    "returning to thread data_read=%d reassembly_data_length=%d first_entry_offset=%d\n",

			    data_read, st->reassembly_data_length,

			    st->first_entry_offset);

			    data_read, sc->recv_io.reassembly.data_length,

			    sc->recv_io.reassembly.first_entry_offset);

read_rfc1002_done:

		return data_read;

	}



	ksmbd_debug(RDMA, "wait_event on more data\n");

	rc = wait_event_interruptible(st->wait_reassembly_queue,

				      st->reassembly_data_length >= size ||

	rc = wait_event_interruptible(sc->recv_io.reassembly.wait_queue,

				      sc->recv_io.reassembly.data_length >= size ||

				       sc->status != SMBDIRECT_SOCKET_CONNECTED);

	if (rc)

		return -EINTR;
@@ -1598,7 +1594,7 @@ static int smb_direct_cm_handler(struct rdma_cm_id *cm_id, 


		sc->status = SMBDIRECT_SOCKET_DISCONNECTED;

		wake_up_interruptible(&t->wait_status);

		wake_up_interruptible(&t->wait_reassembly_queue);

		wake_up_interruptible(&sc->recv_io.reassembly.wait_queue);

		wake_up(&t->wait_send_credits);

		break;

	}
@@ -2065,10 +2061,10 @@ static int smb_direct_prepare(struct ksmbd_transport *t) 


	ret = smb_direct_send_negotiate_response(st, ret);

out:

	spin_lock_irq(&st->reassembly_queue_lock);

	st->reassembly_queue_length--;

	spin_lock_irq(&sc->recv_io.reassembly.lock);

	sc->recv_io.reassembly.queue_length--;

	list_del(&recvmsg->list);

	spin_unlock_irq(&st->reassembly_queue_lock);

	spin_unlock_irq(&sc->recv_io.reassembly.lock);

	put_recvmsg(st, recvmsg);



	return ret;