~hc/RK356X_SDK_RELEASE.git

..	..	@@ -47,12 +47,15 @@
47	47	#include <net/checksum.h>
48	48	#include <net/udp.h>
49	49	#include <net/tcp.h>
	50	+#include <linux/bvec.h>
	51	+#include <linux/highmem.h>
	52	+#include <linux/uio.h>
	53	+#include <linux/sched/mm.h>
50	54
51	55	#include <trace/events/sunrpc.h>
52	56
	57	+#include "socklib.h"
53	58	#include "sunrpc.h"
54		-
55		-#define RPC_TCP_READ_CHUNK_SZ (35121024)
56	59
57	60	static void xs_close(struct rpc_xprt *xprt);
58	61	static void xs_tcp_set_socket_timeouts(struct rpc_xprt *xprt,
..	..	@@ -67,8 +70,6 @@
67	70
68	71	static unsigned int xprt_min_resvport = RPC_DEF_MIN_RESVPORT;
69	72	static unsigned int xprt_max_resvport = RPC_DEF_MAX_RESVPORT;
70		-
71		-#if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
72	73
73	74	#define XS_TCP_LINGER_TO (15U * HZ)
74	75	static unsigned int xs_tcp_fin_timeout __read_mostly = XS_TCP_LINGER_TO;
..	..	@@ -158,8 +159,6 @@
158	159	},
159	160	{ },
160	161	};
161		-
162		-#endif
163	162
164	163	/*
165	164	* Wait duration for a reply from the RPC portmapper.
..	..	@@ -325,160 +324,452 @@
325	324	}
326	325	}
327	326
	327	+static size_t
	328	+xs_alloc_sparse_pages(struct xdr_buf *buf, size_t want, gfp_t gfp)
	329	+{
	330	+ size_t i,n;
	331	+
	332	+ if (!want \|\| !(buf->flags & XDRBUF_SPARSE_PAGES))
	333	+ return want;
	334	+ n = (buf->page_base + want + PAGE_SIZE - 1) >> PAGE_SHIFT;
	335	+ for (i = 0; i < n; i++) {
	336	+ if (buf->pages[i])
	337	+ continue;
	338	+ buf->bvec[i].bv_page = buf->pages[i] = alloc_page(gfp);
	339	+ if (!buf->pages[i]) {
	340	+ i *= PAGE_SIZE;
	341	+ return i > buf->page_base ? i - buf->page_base : 0;
	342	+ }
	343	+ }
	344	+ return want;
	345	+}
	346	+
	347	+static ssize_t
	348	+xs_sock_recvmsg(struct socket sock, struct msghdr msg, int flags, size_t seek)
	349	+{
	350	+ ssize_t ret;
	351	+ if (seek != 0)
	352	+ iov_iter_advance(&msg->msg_iter, seek);
	353	+ ret = sock_recvmsg(sock, msg, flags);
	354	+ return ret > 0 ? ret + seek : ret;
	355	+}
	356	+
	357	+static ssize_t
	358	+xs_read_kvec(struct socket sock, struct msghdr msg, int flags,
	359	+ struct kvec *kvec, size_t count, size_t seek)
	360	+{
	361	+ iov_iter_kvec(&msg->msg_iter, READ, kvec, 1, count);
	362	+ return xs_sock_recvmsg(sock, msg, flags, seek);
	363	+}
	364	+
	365	+static ssize_t
	366	+xs_read_bvec(struct socket sock, struct msghdr msg, int flags,
	367	+ struct bio_vec *bvec, unsigned long nr, size_t count,
	368	+ size_t seek)
	369	+{
	370	+ iov_iter_bvec(&msg->msg_iter, READ, bvec, nr, count);
	371	+ return xs_sock_recvmsg(sock, msg, flags, seek);
	372	+}
	373	+
	374	+static ssize_t
	375	+xs_read_discard(struct socket sock, struct msghdr msg, int flags,
	376	+ size_t count)
	377	+{
	378	+ iov_iter_discard(&msg->msg_iter, READ, count);
	379	+ return sock_recvmsg(sock, msg, flags);
	380	+}
	381	+
	382	+#if ARCH_IMPLEMENTS_FLUSH_DCACHE_PAGE
	383	+static void
	384	+xs_flush_bvec(const struct bio_vec *bvec, size_t count, size_t seek)
	385	+{
	386	+ struct bvec_iter bi = {
	387	+ .bi_size = count,
	388	+ };
	389	+ struct bio_vec bv;
	390	+
	391	+ bvec_iter_advance(bvec, &bi, seek & PAGE_MASK);
	392	+ for_each_bvec(bv, bvec, bi, bi)
	393	+ flush_dcache_page(bv.bv_page);
	394	+}
	395	+#else
	396	+static inline void
	397	+xs_flush_bvec(const struct bio_vec *bvec, size_t count, size_t seek)
	398	+{
	399	+}
	400	+#endif
	401	+
	402	+static ssize_t
	403	+xs_read_xdr_buf(struct socket sock, struct msghdr msg, int flags,
	404	+ struct xdr_buf buf, size_t count, size_t seek, size_t read)
	405	+{
	406	+ size_t want, seek_init = seek, offset = 0;
	407	+ ssize_t ret;
	408	+
	409	+ want = min_t(size_t, count, buf->head[0].iov_len);
	410	+ if (seek < want) {
	411	+ ret = xs_read_kvec(sock, msg, flags, &buf->head[0], want, seek);
	412	+ if (ret <= 0)
	413	+ goto sock_err;
	414	+ offset += ret;
	415	+ if (offset == count \|\| msg->msg_flags & (MSG_EOR\|MSG_TRUNC))
	416	+ goto out;
	417	+ if (ret != want)
	418	+ goto out;
	419	+ seek = 0;
	420	+ } else {
	421	+ seek -= want;
	422	+ offset += want;
	423	+ }
	424	+
	425	+ want = xs_alloc_sparse_pages(buf,
	426	+ min_t(size_t, count - offset, buf->page_len),
	427	+ GFP_KERNEL);
	428	+ if (seek < want) {
	429	+ ret = xs_read_bvec(sock, msg, flags, buf->bvec,
	430	+ xdr_buf_pagecount(buf),
	431	+ want + buf->page_base,
	432	+ seek + buf->page_base);
	433	+ if (ret <= 0)
	434	+ goto sock_err;
	435	+ xs_flush_bvec(buf->bvec, ret, seek + buf->page_base);
	436	+ ret -= buf->page_base;
	437	+ offset += ret;
	438	+ if (offset == count \|\| msg->msg_flags & (MSG_EOR\|MSG_TRUNC))
	439	+ goto out;
	440	+ if (ret != want)
	441	+ goto out;
	442	+ seek = 0;
	443	+ } else {
	444	+ seek -= want;
	445	+ offset += want;
	446	+ }
	447	+
	448	+ want = min_t(size_t, count - offset, buf->tail[0].iov_len);
	449	+ if (seek < want) {
	450	+ ret = xs_read_kvec(sock, msg, flags, &buf->tail[0], want, seek);
	451	+ if (ret <= 0)
	452	+ goto sock_err;
	453	+ offset += ret;
	454	+ if (offset == count \|\| msg->msg_flags & (MSG_EOR\|MSG_TRUNC))
	455	+ goto out;
	456	+ if (ret != want)
	457	+ goto out;
	458	+ } else if (offset < seek_init)
	459	+ offset = seek_init;
	460	+ ret = -EMSGSIZE;
	461	+out:
	462	+ *read = offset - seek_init;
	463	+ return ret;
	464	+sock_err:
	465	+ offset += seek;
	466	+ goto out;
	467	+}
	468	+
	469	+static void
	470	+xs_read_header(struct sock_xprt transport, struct xdr_buf buf)
	471	+{
	472	+ if (!transport->recv.copied) {
	473	+ if (buf->head[0].iov_len >= transport->recv.offset)
	474	+ memcpy(buf->head[0].iov_base,
	475	+ &transport->recv.xid,
	476	+ transport->recv.offset);
	477	+ transport->recv.copied = transport->recv.offset;
	478	+ }
	479	+}
	480	+
	481	+static bool
	482	+xs_read_stream_request_done(struct sock_xprt *transport)
	483	+{
	484	+ return transport->recv.fraghdr & cpu_to_be32(RPC_LAST_STREAM_FRAGMENT);
	485	+}
	486	+
	487	+static void
	488	+xs_read_stream_check_eor(struct sock_xprt *transport,
	489	+ struct msghdr *msg)
	490	+{
	491	+ if (xs_read_stream_request_done(transport))
	492	+ msg->msg_flags \|= MSG_EOR;
	493	+}
	494	+
	495	+static ssize_t
	496	+xs_read_stream_request(struct sock_xprt transport, struct msghdr msg,
	497	+ int flags, struct rpc_rqst *req)
	498	+{
	499	+ struct xdr_buf *buf = &req->rq_private_buf;
	500	+ size_t want, read;
	501	+ ssize_t ret;
	502	+
	503	+ xs_read_header(transport, buf);
	504	+
	505	+ want = transport->recv.len - transport->recv.offset;
	506	+ if (want != 0) {
	507	+ ret = xs_read_xdr_buf(transport->sock, msg, flags, buf,
	508	+ transport->recv.copied + want,
	509	+ transport->recv.copied,
	510	+ &read);
	511	+ transport->recv.offset += read;
	512	+ transport->recv.copied += read;
	513	+ }
	514	+
	515	+ if (transport->recv.offset == transport->recv.len)
	516	+ xs_read_stream_check_eor(transport, msg);
	517	+
	518	+ if (want == 0)
	519	+ return 0;
	520	+
	521	+ switch (ret) {
	522	+ default:
	523	+ break;
	524	+ case -EFAULT:
	525	+ case -EMSGSIZE:
	526	+ msg->msg_flags \|= MSG_TRUNC;
	527	+ return read;
	528	+ case 0:
	529	+ return -ESHUTDOWN;
	530	+ }
	531	+ return ret < 0 ? ret : read;
	532	+}
	533	+
	534	+static size_t
	535	+xs_read_stream_headersize(bool isfrag)
	536	+{
	537	+ if (isfrag)
	538	+ return sizeof(__be32);
	539	+ return 3 * sizeof(__be32);
	540	+}
	541	+
	542	+static ssize_t
	543	+xs_read_stream_header(struct sock_xprt transport, struct msghdr msg,
	544	+ int flags, size_t want, size_t seek)
	545	+{
	546	+ struct kvec kvec = {
	547	+ .iov_base = &transport->recv.fraghdr,
	548	+ .iov_len = want,
	549	+ };
	550	+ return xs_read_kvec(transport->sock, msg, flags, &kvec, want, seek);
	551	+}
	552	+
	553	+#if defined(CONFIG_SUNRPC_BACKCHANNEL)
	554	+static ssize_t
	555	+xs_read_stream_call(struct sock_xprt transport, struct msghdr msg, int flags)
	556	+{
	557	+ struct rpc_xprt *xprt = &transport->xprt;
	558	+ struct rpc_rqst *req;
	559	+ ssize_t ret;
	560	+
	561	+ /* Look up and lock the request corresponding to the given XID */
	562	+ req = xprt_lookup_bc_request(xprt, transport->recv.xid);
	563	+ if (!req) {
	564	+ printk(KERN_WARNING "Callback slot table overflowed\n");
	565	+ return -ESHUTDOWN;
	566	+ }
	567	+ if (transport->recv.copied && !req->rq_private_buf.len)
	568	+ return -ESHUTDOWN;
	569	+
	570	+ ret = xs_read_stream_request(transport, msg, flags, req);
	571	+ if (msg->msg_flags & (MSG_EOR\|MSG_TRUNC))
	572	+ xprt_complete_bc_request(req, transport->recv.copied);
	573	+ else
	574	+ req->rq_private_buf.len = transport->recv.copied;
	575	+
	576	+ return ret;
	577	+}
	578	+#else /* CONFIG_SUNRPC_BACKCHANNEL */
	579	+static ssize_t
	580	+xs_read_stream_call(struct sock_xprt transport, struct msghdr msg, int flags)
	581	+{
	582	+ return -ESHUTDOWN;
	583	+}
	584	+#endif /* CONFIG_SUNRPC_BACKCHANNEL */
	585	+
	586	+static ssize_t
	587	+xs_read_stream_reply(struct sock_xprt transport, struct msghdr msg, int flags)
	588	+{
	589	+ struct rpc_xprt *xprt = &transport->xprt;
	590	+ struct rpc_rqst *req;
	591	+ ssize_t ret = 0;
	592	+
	593	+ /* Look up and lock the request corresponding to the given XID */
	594	+ spin_lock(&xprt->queue_lock);
	595	+ req = xprt_lookup_rqst(xprt, transport->recv.xid);
	596	+ if (!req \|\| (transport->recv.copied && !req->rq_private_buf.len)) {
	597	+ msg->msg_flags \|= MSG_TRUNC;
	598	+ goto out;
	599	+ }
	600	+ xprt_pin_rqst(req);
	601	+ spin_unlock(&xprt->queue_lock);
	602	+
	603	+ ret = xs_read_stream_request(transport, msg, flags, req);
	604	+
	605	+ spin_lock(&xprt->queue_lock);
	606	+ if (msg->msg_flags & (MSG_EOR\|MSG_TRUNC))
	607	+ xprt_complete_rqst(req->rq_task, transport->recv.copied);
	608	+ else
	609	+ req->rq_private_buf.len = transport->recv.copied;
	610	+ xprt_unpin_rqst(req);
	611	+out:
	612	+ spin_unlock(&xprt->queue_lock);
	613	+ return ret;
	614	+}
	615	+
	616	+static ssize_t
	617	+xs_read_stream(struct sock_xprt *transport, int flags)
	618	+{
	619	+ struct msghdr msg = { 0 };
	620	+ size_t want, read = 0;
	621	+ ssize_t ret = 0;
	622	+
	623	+ if (transport->recv.len == 0) {
	624	+ want = xs_read_stream_headersize(transport->recv.copied != 0);
	625	+ ret = xs_read_stream_header(transport, &msg, flags, want,
	626	+ transport->recv.offset);
	627	+ if (ret <= 0)
	628	+ goto out_err;
	629	+ transport->recv.offset = ret;
	630	+ if (transport->recv.offset != want)
	631	+ return transport->recv.offset;
	632	+ transport->recv.len = be32_to_cpu(transport->recv.fraghdr) &
	633	+ RPC_FRAGMENT_SIZE_MASK;
	634	+ transport->recv.offset -= sizeof(transport->recv.fraghdr);
	635	+ read = ret;
	636	+ }
	637	+
	638	+ switch (be32_to_cpu(transport->recv.calldir)) {
	639	+ default:
	640	+ msg.msg_flags \|= MSG_TRUNC;
	641	+ break;
	642	+ case RPC_CALL:
	643	+ ret = xs_read_stream_call(transport, &msg, flags);
	644	+ break;
	645	+ case RPC_REPLY:
	646	+ ret = xs_read_stream_reply(transport, &msg, flags);
	647	+ }
	648	+ if (msg.msg_flags & MSG_TRUNC) {
	649	+ transport->recv.calldir = cpu_to_be32(-1);
	650	+ transport->recv.copied = -1;
	651	+ }
	652	+ if (ret < 0)
	653	+ goto out_err;
	654	+ read += ret;
	655	+ if (transport->recv.offset < transport->recv.len) {
	656	+ if (!(msg.msg_flags & MSG_TRUNC))
	657	+ return read;
	658	+ msg.msg_flags = 0;
	659	+ ret = xs_read_discard(transport->sock, &msg, flags,
	660	+ transport->recv.len - transport->recv.offset);
	661	+ if (ret <= 0)
	662	+ goto out_err;
	663	+ transport->recv.offset += ret;
	664	+ read += ret;
	665	+ if (transport->recv.offset != transport->recv.len)
	666	+ return read;
	667	+ }
	668	+ if (xs_read_stream_request_done(transport)) {
	669	+ trace_xs_stream_read_request(transport);
	670	+ transport->recv.copied = 0;
	671	+ }
	672	+ transport->recv.offset = 0;
	673	+ transport->recv.len = 0;
	674	+ return read;
	675	+out_err:
	676	+ return ret != 0 ? ret : -ESHUTDOWN;
	677	+}
	678	+
	679	+static __poll_t xs_poll_socket(struct sock_xprt *transport)
	680	+{
	681	+ return transport->sock->ops->poll(transport->file, transport->sock,
	682	+ NULL);
	683	+}
	684	+
	685	+static bool xs_poll_socket_readable(struct sock_xprt *transport)
	686	+{
	687	+ __poll_t events = xs_poll_socket(transport);
	688	+
	689	+ return (events & (EPOLLIN \| EPOLLRDNORM)) && !(events & EPOLLRDHUP);
	690	+}
	691	+
	692	+static void xs_poll_check_readable(struct sock_xprt *transport)
	693	+{
	694	+
	695	+ clear_bit(XPRT_SOCK_DATA_READY, &transport->sock_state);
	696	+ if (!xs_poll_socket_readable(transport))
	697	+ return;
	698	+ if (!test_and_set_bit(XPRT_SOCK_DATA_READY, &transport->sock_state))
	699	+ queue_work(xprtiod_workqueue, &transport->recv_worker);
	700	+}
	701	+
	702	+static void xs_stream_data_receive(struct sock_xprt *transport)
	703	+{
	704	+ size_t read = 0;
	705	+ ssize_t ret = 0;
	706	+
	707	+ mutex_lock(&transport->recv_mutex);
	708	+ if (transport->sock == NULL)
	709	+ goto out;
	710	+ for (;;) {
	711	+ ret = xs_read_stream(transport, MSG_DONTWAIT);
	712	+ if (ret < 0)
	713	+ break;
	714	+ read += ret;
	715	+ cond_resched();
	716	+ }
	717	+ if (ret == -ESHUTDOWN)
	718	+ kernel_sock_shutdown(transport->sock, SHUT_RDWR);
	719	+ else
	720	+ xs_poll_check_readable(transport);
	721	+out:
	722	+ mutex_unlock(&transport->recv_mutex);
	723	+ trace_xs_stream_read_data(&transport->xprt, ret, read);
	724	+}
	725	+
	726	+static void xs_stream_data_receive_workfn(struct work_struct *work)
	727	+{
	728	+ struct sock_xprt *transport =
	729	+ container_of(work, struct sock_xprt, recv_worker);
	730	+ unsigned int pflags = memalloc_nofs_save();
	731	+
	732	+ xs_stream_data_receive(transport);
	733	+ memalloc_nofs_restore(pflags);
	734	+}
	735	+
	736	+static void
	737	+xs_stream_reset_connect(struct sock_xprt *transport)
	738	+{
	739	+ transport->recv.offset = 0;
	740	+ transport->recv.len = 0;
	741	+ transport->recv.copied = 0;
	742	+ transport->xmit.offset = 0;
	743	+}
	744	+
	745	+static void
	746	+xs_stream_start_connect(struct sock_xprt *transport)
	747	+{
	748	+ transport->xprt.stat.connect_count++;
	749	+ transport->xprt.stat.connect_start = jiffies;
	750	+}
	751	+
328	752	#define XS_SENDMSG_FLAGS (MSG_DONTWAIT \| MSG_NOSIGNAL)
329	753
330		-static int xs_send_kvec(struct socket sock, struct sockaddr addr, int addrlen, struct kvec *vec, unsigned int base, int more)
331		-{
332		- struct msghdr msg = {
333		- .msg_name = addr,
334		- .msg_namelen = addrlen,
335		- .msg_flags = XS_SENDMSG_FLAGS \| (more ? MSG_MORE : 0),
336		- };
337		- struct kvec iov = {
338		- .iov_base = vec->iov_base + base,
339		- .iov_len = vec->iov_len - base,
340		- };
341		-
342		- if (iov.iov_len != 0)
343		- return kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len);
344		- return kernel_sendmsg(sock, &msg, NULL, 0, 0);
345		-}
346		-
347		-static int xs_send_pagedata(struct socket sock, struct xdr_buf xdr, unsigned int base, int more, bool zerocopy, int *sent_p)
348		-{
349		- ssize_t (do_sendpage)(struct socket sock, struct page *page,
350		- int offset, size_t size, int flags);
351		- struct page **ppage;
352		- unsigned int remainder;
353		- int err;
354		-
355		- remainder = xdr->page_len - base;
356		- base += xdr->page_base;
357		- ppage = xdr->pages + (base >> PAGE_SHIFT);
358		- base &= ~PAGE_MASK;
359		- do_sendpage = sock->ops->sendpage;
360		- if (!zerocopy)
361		- do_sendpage = sock_no_sendpage;
362		- for(;;) {
363		- unsigned int len = min_t(unsigned int, PAGE_SIZE - base, remainder);
364		- int flags = XS_SENDMSG_FLAGS;
365		-
366		- remainder -= len;
367		- if (more)
368		- flags \|= MSG_MORE;
369		- if (remainder != 0)
370		- flags \|= MSG_SENDPAGE_NOTLAST \| MSG_MORE;
371		- err = do_sendpage(sock, *ppage, base, len, flags);
372		- if (remainder == 0 \|\| err != len)
373		- break;
374		- *sent_p += err;
375		- ppage++;
376		- base = 0;
377		- }
378		- if (err > 0) {
379		- *sent_p += err;
380		- err = 0;
381		- }
382		- return err;
383		-}
384		-
385	754	/**
386		- * xs_sendpages - write pages directly to a socket
387		- * @sock: socket to send on
388		- * @addr: UDP only -- address of destination
389		- * @addrlen: UDP only -- length of destination address
390		- * @xdr: buffer containing this request
391		- * @base: starting position in the buffer
392		- * @zerocopy: true if it is safe to use sendpage()
393		- * @sent_p: return the total number of bytes successfully queued for sending
	755	+ * xs_nospace - handle transmit was incomplete
	756	+ * @req: pointer to RPC request
	757	+ * @transport: pointer to struct sock_xprt
394	758	*
395	759	*/
396		-static int xs_sendpages(struct socket sock, struct sockaddr addr, int addrlen, struct xdr_buf xdr, unsigned int base, bool zerocopy, int sent_p)
	760	+static int xs_nospace(struct rpc_rqst req, struct sock_xprt transport)
397	761	{
398		- unsigned int remainder = xdr->len - base;
399		- int err = 0;
400		- int sent = 0;
401		-
402		- if (unlikely(!sock))
403		- return -ENOTSOCK;
404		-
405		- if (base != 0) {
406		- addr = NULL;
407		- addrlen = 0;
408		- }
409		-
410		- if (base < xdr->head[0].iov_len \|\| addr != NULL) {
411		- unsigned int len = xdr->head[0].iov_len - base;
412		- remainder -= len;
413		- err = xs_send_kvec(sock, addr, addrlen, &xdr->head[0], base, remainder != 0);
414		- if (remainder == 0 \|\| err != len)
415		- goto out;
416		- *sent_p += err;
417		- base = 0;
418		- } else
419		- base -= xdr->head[0].iov_len;
420		-
421		- if (base < xdr->page_len) {
422		- unsigned int len = xdr->page_len - base;
423		- remainder -= len;
424		- err = xs_send_pagedata(sock, xdr, base, remainder != 0, zerocopy, &sent);
425		- *sent_p += sent;
426		- if (remainder == 0 \|\| sent != len)
427		- goto out;
428		- base = 0;
429		- } else
430		- base -= xdr->page_len;
431		-
432		- if (base >= xdr->tail[0].iov_len)
433		- return 0;
434		- err = xs_send_kvec(sock, NULL, 0, &xdr->tail[0], base, 0);
435		-out:
436		- if (err > 0) {
437		- *sent_p += err;
438		- err = 0;
439		- }
440		- return err;
441		-}
442		-
443		-static void xs_nospace_callback(struct rpc_task *task)
444		-{
445		- struct sock_xprt *transport = container_of(task->tk_rqstp->rq_xprt, struct sock_xprt, xprt);
446		-
447		- transport->inet->sk_write_pending--;
448		-}
449		-
450		-/**
451		- * xs_nospace - place task on wait queue if transmit was incomplete
452		- * @task: task to put to sleep
453		- *
454		- */
455		-static int xs_nospace(struct rpc_task *task)
456		-{
457		- struct rpc_rqst *req = task->tk_rqstp;
458		- struct rpc_xprt *xprt = req->rq_xprt;
459		- struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
	762	+ struct rpc_xprt *xprt = &transport->xprt;
460	763	struct sock *sk = transport->inet;
461	764	int ret = -EAGAIN;
462	765
463		- dprintk("RPC: %5u xmit incomplete (%u left of %u)\n",
464		- task->tk_pid, req->rq_slen - req->rq_bytes_sent,
465		- req->rq_slen);
	766	+ trace_rpc_socket_nospace(req, transport);
466	767
467	768	/* Protect against races with write_space */
468		- spin_lock_bh(&xprt->transport_lock);
	769	+ spin_lock(&xprt->transport_lock);
469	770
470	771	/* Don't race with disconnect */
471	772	if (xprt_connected(xprt)) {
472		- /* wait for more buffer space */
473		- sk->sk_write_pending++;
474		- xprt_wait_for_buffer_space(task, xs_nospace_callback);
475		- } else
476		- ret = -ENOTCONN;
477		-
478		- spin_unlock_bh(&xprt->transport_lock);
479		-
480		- /* Race breaker in case memory is freed before above code is called */
481		- if (ret == -EAGAIN) {
482	773	struct socket_wq *wq;
483	774
484	775	rcu_read_lock();
..	..	@@ -486,24 +777,76 @@
486	777	set_bit(SOCKWQ_ASYNC_NOSPACE, &wq->flags);
487	778	rcu_read_unlock();
488	779
489		- sk->sk_write_space(sk);
490		- }
	780	+ /* wait for more buffer space */
	781	+ set_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
	782	+ sk->sk_write_pending++;
	783	+ xprt_wait_for_buffer_space(xprt);
	784	+ } else
	785	+ ret = -ENOTCONN;
	786	+
	787	+ spin_unlock(&xprt->transport_lock);
491	788	return ret;
492	789	}
493	790
494		-/*
495		- * Construct a stream transport record marker in @buf.
496		- */
497		-static inline void xs_encode_stream_record_marker(struct xdr_buf *buf)
	791	+static int xs_sock_nospace(struct rpc_rqst *req)
498	792	{
499		- u32 reclen = buf->len - sizeof(rpc_fraghdr);
500		- rpc_fraghdr *base = buf->head[0].iov_base;
501		- *base = cpu_to_be32(RPC_LAST_STREAM_FRAGMENT \| reclen);
	793	+ struct sock_xprt *transport =
	794	+ container_of(req->rq_xprt, struct sock_xprt, xprt);
	795	+ struct sock *sk = transport->inet;
	796	+ int ret = -EAGAIN;
	797	+
	798	+ lock_sock(sk);
	799	+ if (!sock_writeable(sk))
	800	+ ret = xs_nospace(req, transport);
	801	+ release_sock(sk);
	802	+ return ret;
	803	+}
	804	+
	805	+static int xs_stream_nospace(struct rpc_rqst *req)
	806	+{
	807	+ struct sock_xprt *transport =
	808	+ container_of(req->rq_xprt, struct sock_xprt, xprt);
	809	+ struct sock *sk = transport->inet;
	810	+ int ret = -EAGAIN;
	811	+
	812	+ lock_sock(sk);
	813	+ if (!sk_stream_memory_free(sk))
	814	+ ret = xs_nospace(req, transport);
	815	+ release_sock(sk);
	816	+ return ret;
	817	+}
	818	+
	819	+static void
	820	+xs_stream_prepare_request(struct rpc_rqst *req)
	821	+{
	822	+ xdr_free_bvec(&req->rq_rcv_buf);
	823	+ req->rq_task->tk_status = xdr_alloc_bvec(&req->rq_rcv_buf, GFP_KERNEL);
	824	+}
	825	+
	826	+/*
	827	+ * Determine if the previous message in the stream was aborted before it
	828	+ * could complete transmission.
	829	+ */
	830	+static bool
	831	+xs_send_request_was_aborted(struct sock_xprt transport, struct rpc_rqst req)
	832	+{
	833	+ return transport->xmit.offset != 0 && req->rq_bytes_sent == 0;
	834	+}
	835	+
	836	+/*
	837	+ * Return the stream record marker field for a record of length < 2^31-1
	838	+ */
	839	+static rpc_fraghdr
	840	+xs_stream_record_marker(struct xdr_buf *xdr)
	841	+{
	842	+ if (!xdr->len)
	843	+ return 0;
	844	+ return cpu_to_be32(RPC_LAST_STREAM_FRAGMENT \| (u32)xdr->len);
502	845	}
503	846
504	847	/**
505	848	* xs_local_send_request - write an RPC request to an AF_LOCAL socket
506		- * @task: RPC task that manages the state of an RPC request
	849	+ * @req: pointer to RPC request
507	850	*
508	851	* Return values:
509	852	* 0: The request has been sent
..	..	@@ -512,35 +855,44 @@
512	855	* ENOTCONN: Caller needs to invoke connect logic then call again
513	856	* other: Some other error occured, the request was not sent
514	857	*/
515		-static int xs_local_send_request(struct rpc_task *task)
	858	+static int xs_local_send_request(struct rpc_rqst *req)
516	859	{
517		- struct rpc_rqst *req = task->tk_rqstp;
518	860	struct rpc_xprt *xprt = req->rq_xprt;
519	861	struct sock_xprt *transport =
520	862	container_of(xprt, struct sock_xprt, xprt);
521	863	struct xdr_buf *xdr = &req->rq_snd_buf;
	864	+ rpc_fraghdr rm = xs_stream_record_marker(xdr);
	865	+ unsigned int msglen = rm ? req->rq_slen + sizeof(rm) : req->rq_slen;
	866	+ struct msghdr msg = {
	867	+ .msg_flags = XS_SENDMSG_FLAGS,
	868	+ };
	869	+ unsigned int sent;
522	870	int status;
523		- int sent = 0;
524	871
525		- xs_encode_stream_record_marker(&req->rq_snd_buf);
	872	+ /* Close the stream if the previous transmission was incomplete */
	873	+ if (xs_send_request_was_aborted(transport, req)) {
	874	+ xprt_force_disconnect(xprt);
	875	+ return -ENOTCONN;
	876	+ }
526	877
527	878	xs_pktdump("packet data:",
528	879	req->rq_svec->iov_base, req->rq_svec->iov_len);
529	880
530	881	req->rq_xtime = ktime_get();
531		- status = xs_sendpages(transport->sock, NULL, 0, xdr, req->rq_bytes_sent,
532		- true, &sent);
	882	+ status = xprt_sock_sendmsg(transport->sock, &msg, xdr,
	883	+ transport->xmit.offset, rm, &sent);
533	884	dprintk("RPC: %s(%u) = %d\n",
534		- __func__, xdr->len - req->rq_bytes_sent, status);
	885	+ __func__, xdr->len - transport->xmit.offset, status);
535	886
536	887	if (status == -EAGAIN && sock_writeable(transport->inet))
537	888	status = -ENOBUFS;
538	889
539	890	if (likely(sent > 0) \|\| status == 0) {
540		- req->rq_bytes_sent += sent;
541		- req->rq_xmit_bytes_sent += sent;
542		- if (likely(req->rq_bytes_sent >= req->rq_slen)) {
543		- req->rq_bytes_sent = 0;
	891	+ transport->xmit.offset += sent;
	892	+ req->rq_bytes_sent = transport->xmit.offset;
	893	+ if (likely(req->rq_bytes_sent >= msglen)) {
	894	+ req->rq_xmit_bytes_sent += transport->xmit.offset;
	895	+ transport->xmit.offset = 0;
544	896	return 0;
545	897	}
546	898	status = -EAGAIN;
..	..	@@ -550,14 +902,14 @@
550	902	case -ENOBUFS:
551	903	break;
552	904	case -EAGAIN:
553		- status = xs_nospace(task);
	905	+ status = xs_stream_nospace(req);
554	906	break;
555	907	default:
556	908	dprintk("RPC: sendmsg returned unrecognized error %d\n",
557	909	-status);
558		- /* fall through */
	910	+ fallthrough;
559	911	case -EPIPE:
560		- xs_close(xprt);
	912	+ xprt_force_disconnect(xprt);
561	913	status = -ENOTCONN;
562	914	}
563	915
..	..	@@ -566,7 +918,7 @@
566	918
567	919	/**
568	920	* xs_udp_send_request - write an RPC request to a UDP socket
569		- * @task: address of RPC task that manages the state of an RPC request
	921	+ * @req: pointer to RPC request
570	922	*
571	923	* Return values:
572	924	* 0: The request has been sent
..	..	@@ -575,13 +927,17 @@
575	927	* ENOTCONN: Caller needs to invoke connect logic then call again
576	928	* other: Some other error occurred, the request was not sent
577	929	*/
578		-static int xs_udp_send_request(struct rpc_task *task)
	930	+static int xs_udp_send_request(struct rpc_rqst *req)
579	931	{
580		- struct rpc_rqst *req = task->tk_rqstp;
581	932	struct rpc_xprt *xprt = req->rq_xprt;
582	933	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
583	934	struct xdr_buf *xdr = &req->rq_snd_buf;
584		- int sent = 0;
	935	+ struct msghdr msg = {
	936	+ .msg_name = xs_addr(xprt),
	937	+ .msg_namelen = xprt->addrlen,
	938	+ .msg_flags = XS_SENDMSG_FLAGS,
	939	+ };
	940	+ unsigned int sent;
585	941	int status;
586	942
587	943	xs_pktdump("packet data:",
..	..	@@ -590,12 +946,15 @@
590	946
591	947	if (!xprt_bound(xprt))
592	948	return -ENOTCONN;
	949	+
	950	+ if (!xprt_request_get_cong(xprt, req))
	951	+ return -EBADSLT;
	952	+
593	953	req->rq_xtime = ktime_get();
594		- status = xs_sendpages(transport->sock, xs_addr(xprt), xprt->addrlen,
595		- xdr, req->rq_bytes_sent, true, &sent);
	954	+ status = xprt_sock_sendmsg(transport->sock, &msg, xdr, 0, 0, &sent);
596	955
597	956	dprintk("RPC: xs_udp_send_request(%u) = %d\n",
598		- xdr->len - req->rq_bytes_sent, status);
	957	+ xdr->len, status);
599	958
600	959	/* firewall is blocking us, don't return -EAGAIN or we end up looping */
601	960	if (status == -EPERM)
..	..	@@ -619,7 +978,7 @@
619	978	/* Should we call xs_close() here? */
620	979	break;
621	980	case -EAGAIN:
622		- status = xs_nospace(task);
	981	+ status = xs_sock_nospace(req);
623	982	break;
624	983	case -ENETUNREACH:
625	984	case -ENOBUFS:
..	..	@@ -639,7 +998,7 @@
639	998
640	999	/**
641	1000	* xs_tcp_send_request - write an RPC request to a TCP socket
642		- * @task: address of RPC task that manages the state of an RPC request
	1001	+ * @req: pointer to RPC request
643	1002	*
644	1003	* Return values:
645	1004	* 0: The request has been sent
..	..	@@ -651,28 +1010,30 @@
651	1010	* XXX: In the case of soft timeouts, should we eventually give up
652	1011	* if sendmsg is not able to make progress?
653	1012	*/
654		-static int xs_tcp_send_request(struct rpc_task *task)
	1013	+static int xs_tcp_send_request(struct rpc_rqst *req)
655	1014	{
656		- struct rpc_rqst *req = task->tk_rqstp;
657	1015	struct rpc_xprt *xprt = req->rq_xprt;
658	1016	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
659	1017	struct xdr_buf *xdr = &req->rq_snd_buf;
660		- bool zerocopy = true;
	1018	+ rpc_fraghdr rm = xs_stream_record_marker(xdr);
	1019	+ unsigned int msglen = rm ? req->rq_slen + sizeof(rm) : req->rq_slen;
	1020	+ struct msghdr msg = {
	1021	+ .msg_flags = XS_SENDMSG_FLAGS,
	1022	+ };
661	1023	bool vm_wait = false;
	1024	+ unsigned int sent;
662	1025	int status;
663		- int sent;
664	1026
665		- xs_encode_stream_record_marker(&req->rq_snd_buf);
	1027	+ /* Close the stream if the previous transmission was incomplete */
	1028	+ if (xs_send_request_was_aborted(transport, req)) {
	1029	+ if (transport->sock != NULL)
	1030	+ kernel_sock_shutdown(transport->sock, SHUT_RDWR);
	1031	+ return -ENOTCONN;
	1032	+ }
666	1033
667	1034	xs_pktdump("packet data:",
668	1035	req->rq_svec->iov_base,
669	1036	req->rq_svec->iov_len);
670		- /* Don't use zero copy if this is a resend. If the RPC call
671		- * completes while the socket holds a reference to the pages,
672		- * then we may end up resending corrupted data.
673		- */
674		- if (task->tk_flags & RPC_TASK_SENT)
675		- zerocopy = false;
676	1037
677	1038	if (test_bit(XPRT_SOCK_UPD_TIMEOUT, &transport->sock_state))
678	1039	xs_tcp_set_socket_timeouts(xprt, transport->sock);
..	..	@@ -682,19 +1043,19 @@
682	1043	* called sendmsg(). */
683	1044	req->rq_xtime = ktime_get();
684	1045	while (1) {
685		- sent = 0;
686		- status = xs_sendpages(transport->sock, NULL, 0, xdr,
687		- req->rq_bytes_sent, zerocopy, &sent);
	1046	+ status = xprt_sock_sendmsg(transport->sock, &msg, xdr,
	1047	+ transport->xmit.offset, rm, &sent);
688	1048
689	1049	dprintk("RPC: xs_tcp_send_request(%u) = %d\n",
690		- xdr->len - req->rq_bytes_sent, status);
	1050	+ xdr->len - transport->xmit.offset, status);
691	1051
692	1052	/* If we've sent the entire packet, immediately
693	1053	* reset the count of bytes sent. */
694		- req->rq_bytes_sent += sent;
695		- req->rq_xmit_bytes_sent += sent;
696		- if (likely(req->rq_bytes_sent >= req->rq_slen)) {
697		- req->rq_bytes_sent = 0;
	1054	+ transport->xmit.offset += sent;
	1055	+ req->rq_bytes_sent = transport->xmit.offset;
	1056	+ if (likely(req->rq_bytes_sent >= msglen)) {
	1057	+ req->rq_xmit_bytes_sent += transport->xmit.offset;
	1058	+ transport->xmit.offset = 0;
698	1059	return 0;
699	1060	}
700	1061
..	..	@@ -732,7 +1093,7 @@
732	1093	/* Should we call xs_close() here? */
733	1094	break;
734	1095	case -EAGAIN:
735		- status = xs_nospace(task);
	1096	+ status = xs_stream_nospace(req);
736	1097	break;
737	1098	case -ECONNRESET:
738	1099	case -ECONNREFUSED:
..	..	@@ -747,35 +1108,6 @@
747	1108	}
748	1109
749	1110	return status;
750		-}
751		-
752		-/**
753		- * xs_tcp_release_xprt - clean up after a tcp transmission
754		- * @xprt: transport
755		- * @task: rpc task
756		- *
757		- * This cleans up if an error causes us to abort the transmission of a request.
758		- * In this case, the socket may need to be reset in order to avoid confusing
759		- * the server.
760		- */
761		-static void xs_tcp_release_xprt(struct rpc_xprt xprt, struct rpc_task task)
762		-{
763		- struct rpc_rqst *req;
764		-
765		- if (task != xprt->snd_task)
766		- return;
767		- if (task == NULL)
768		- goto out_release;
769		- req = task->tk_rqstp;
770		- if (req == NULL)
771		- goto out_release;
772		- if (req->rq_bytes_sent == 0)
773		- goto out_release;
774		- if (req->rq_bytes_sent == req->rq_snd_buf.len)
775		- goto out_release;
776		- set_bit(XPRT_CLOSE_WAIT, &xprt->state);
777		-out_release:
778		- xprt_release_xprt(xprt, task);
779	1111	}
780	1112
781	1113	static void xs_save_old_callbacks(struct sock_xprt transport, struct sock sk)
..	..	@@ -799,6 +1131,15 @@
799	1131	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
800	1132
801	1133	clear_bit(XPRT_SOCK_DATA_READY, &transport->sock_state);
	1134	+ clear_bit(XPRT_SOCK_WAKE_ERROR, &transport->sock_state);
	1135	+ clear_bit(XPRT_SOCK_WAKE_WRITE, &transport->sock_state);
	1136	+ clear_bit(XPRT_SOCK_WAKE_DISCONNECT, &transport->sock_state);
	1137	+}
	1138	+
	1139	+static void xs_run_error_worker(struct sock_xprt *transport, unsigned int nr)
	1140	+{
	1141	+ set_bit(nr, &transport->sock_state);
	1142	+ queue_work(xprtiod_workqueue, &transport->error_worker);
802	1143	}
803	1144
804	1145	static void xs_sock_reset_connection_flags(struct rpc_xprt *xprt)
..	..	@@ -819,20 +1160,24 @@
819	1160	*/
820	1161	static void xs_error_report(struct sock *sk)
821	1162	{
	1163	+ struct sock_xprt *transport;
822	1164	struct rpc_xprt *xprt;
823		- int err;
824	1165
825	1166	read_lock_bh(&sk->sk_callback_lock);
826	1167	if (!(xprt = xprt_from_sock(sk)))
827	1168	goto out;
828	1169
829		- err = -sk->sk_err;
830		- if (err == 0)
	1170	+ transport = container_of(xprt, struct sock_xprt, xprt);
	1171	+ transport->xprt_err = -sk->sk_err;
	1172	+ if (transport->xprt_err == 0)
831	1173	goto out;
832	1174	dprintk("RPC: xs_error_report client %p, error=%d...\n",
833		- xprt, -err);
834		- trace_rpc_socket_error(xprt, sk->sk_socket, err);
835		- xprt_wake_pending_tasks(xprt, err);
	1175	+ xprt, -transport->xprt_err);
	1176	+ trace_rpc_socket_error(xprt, sk->sk_socket, transport->xprt_err);
	1177	+
	1178	+ /* barrier ensures xprt_err is set before XPRT_SOCK_WAKE_ERROR */
	1179	+ smp_mb__before_atomic();
	1180	+ xs_run_error_worker(transport, XPRT_SOCK_WAKE_ERROR);
836	1181	out:
837	1182	read_unlock_bh(&sk->sk_callback_lock);
838	1183	}
..	..	@@ -842,9 +1187,20 @@
842	1187	struct socket *sock = transport->sock;
843	1188	struct sock *sk = transport->inet;
844	1189	struct rpc_xprt *xprt = &transport->xprt;
	1190	+ struct file *filp = transport->file;
845	1191
846	1192	if (sk == NULL)
847	1193	return;
	1194	+ /*
	1195	+ * Make sure we're calling this in a context from which it is safe
	1196	+ * to call __fput_sync(). In practice that means rpciod and the
	1197	+ * system workqueue.
	1198	+ */
	1199	+ if (!(current->flags & PF_WQ_WORKER)) {
	1200	+ WARN_ON_ONCE(1);
	1201	+ set_bit(XPRT_CLOSE_WAIT, &xprt->state);
	1202	+ return;
	1203	+ }
848	1204
849	1205	if (atomic_read(&transport->xprt.swapper))
850	1206	sk_clear_memalloc(sk);
..	..	@@ -855,6 +1211,7 @@
855	1211	write_lock_bh(&sk->sk_callback_lock);
856	1212	transport->inet = NULL;
857	1213	transport->sock = NULL;
	1214	+ transport->file = NULL;
858	1215
859	1216	sk->sk_user_data = NULL;
860	1217
..	..	@@ -862,10 +1219,14 @@
862	1219	xprt_clear_connected(xprt);
863	1220	write_unlock_bh(&sk->sk_callback_lock);
864	1221	xs_sock_reset_connection_flags(xprt);
	1222	+ /* Reset stream record info */
	1223	+ xs_stream_reset_connect(transport);
865	1224	mutex_unlock(&transport->recv_mutex);
866	1225
867	1226	trace_rpc_socket_close(xprt, sock);
868		- sock_release(sock);
	1227	+ __fput_sync(filp);
	1228	+
	1229	+ xprt_disconnect_done(xprt);
869	1230	}
870	1231
871	1232	/**
..	..	@@ -886,8 +1247,6 @@
886	1247
887	1248	xs_reset_transport(transport);
888	1249	xprt->reestablish_timeout = 0;
889		-
890		- xprt_disconnect_done(xprt);
891	1250	}
892	1251
893	1252	static void xs_inject_disconnect(struct rpc_xprt *xprt)
..	..	@@ -917,116 +1276,9 @@
917	1276	cancel_delayed_work_sync(&transport->connect_worker);
918	1277	xs_close(xprt);
919	1278	cancel_work_sync(&transport->recv_worker);
	1279	+ cancel_work_sync(&transport->error_worker);
920	1280	xs_xprt_free(xprt);
921	1281	module_put(THIS_MODULE);
922		-}
923		-
924		-static int xs_local_copy_to_xdr(struct xdr_buf xdr, struct sk_buff skb)
925		-{
926		- struct xdr_skb_reader desc = {
927		- .skb = skb,
928		- .offset = sizeof(rpc_fraghdr),
929		- .count = skb->len - sizeof(rpc_fraghdr),
930		- };
931		-
932		- if (xdr_partial_copy_from_skb(xdr, 0, &desc, xdr_skb_read_bits) < 0)
933		- return -1;
934		- if (desc.count)
935		- return -1;
936		- return 0;
937		-}
938		-
939		-/**
940		- * xs_local_data_read_skb
941		- * @xprt: transport
942		- * @sk: socket
943		- * @skb: skbuff
944		- *
945		- * Currently this assumes we can read the whole reply in a single gulp.
946		- */
947		-static void xs_local_data_read_skb(struct rpc_xprt *xprt,
948		- struct sock *sk,
949		- struct sk_buff *skb)
950		-{
951		- struct rpc_task *task;
952		- struct rpc_rqst *rovr;
953		- int repsize, copied;
954		- u32 _xid;
955		- __be32 *xp;
956		-
957		- repsize = skb->len - sizeof(rpc_fraghdr);
958		- if (repsize < 4) {
959		- dprintk("RPC: impossible RPC reply size %d\n", repsize);
960		- return;
961		- }
962		-
963		- /* Copy the XID from the skb... */
964		- xp = skb_header_pointer(skb, sizeof(rpc_fraghdr), sizeof(_xid), &_xid);
965		- if (xp == NULL)
966		- return;
967		-
968		- /* Look up and lock the request corresponding to the given XID */
969		- spin_lock(&xprt->recv_lock);
970		- rovr = xprt_lookup_rqst(xprt, *xp);
971		- if (!rovr)
972		- goto out_unlock;
973		- xprt_pin_rqst(rovr);
974		- spin_unlock(&xprt->recv_lock);
975		- task = rovr->rq_task;
976		-
977		- copied = rovr->rq_private_buf.buflen;
978		- if (copied > repsize)
979		- copied = repsize;
980		-
981		- if (xs_local_copy_to_xdr(&rovr->rq_private_buf, skb)) {
982		- dprintk("RPC: sk_buff copy failed\n");
983		- spin_lock(&xprt->recv_lock);
984		- goto out_unpin;
985		- }
986		-
987		- spin_lock(&xprt->recv_lock);
988		- xprt_complete_rqst(task, copied);
989		-out_unpin:
990		- xprt_unpin_rqst(rovr);
991		- out_unlock:
992		- spin_unlock(&xprt->recv_lock);
993		-}
994		-
995		-static void xs_local_data_receive(struct sock_xprt *transport)
996		-{
997		- struct sk_buff *skb;
998		- struct sock *sk;
999		- int err;
1000		-
1001		-restart:
1002		- mutex_lock(&transport->recv_mutex);
1003		- sk = transport->inet;
1004		- if (sk == NULL)
1005		- goto out;
1006		- for (;;) {
1007		- skb = skb_recv_datagram(sk, 0, 1, &err);
1008		- if (skb != NULL) {
1009		- xs_local_data_read_skb(&transport->xprt, sk, skb);
1010		- skb_free_datagram(sk, skb);
1011		- continue;
1012		- }
1013		- if (!test_and_clear_bit(XPRT_SOCK_DATA_READY, &transport->sock_state))
1014		- break;
1015		- if (need_resched()) {
1016		- mutex_unlock(&transport->recv_mutex);
1017		- cond_resched();
1018		- goto restart;
1019		- }
1020		- }
1021		-out:
1022		- mutex_unlock(&transport->recv_mutex);
1023		-}
1024		-
1025		-static void xs_local_data_receive_workfn(struct work_struct *work)
1026		-{
1027		- struct sock_xprt *transport =
1028		- container_of(work, struct sock_xprt, recv_worker);
1029		- xs_local_data_receive(transport);
1030	1282	}
1031	1283
1032	1284	/**
..	..	@@ -1058,13 +1310,13 @@
1058	1310	return;
1059	1311
1060	1312	/* Look up and lock the request corresponding to the given XID */
1061		- spin_lock(&xprt->recv_lock);
	1313	+ spin_lock(&xprt->queue_lock);
1062	1314	rovr = xprt_lookup_rqst(xprt, *xp);
1063	1315	if (!rovr)
1064	1316	goto out_unlock;
1065	1317	xprt_pin_rqst(rovr);
1066	1318	xprt_update_rtt(rovr->rq_task);
1067		- spin_unlock(&xprt->recv_lock);
	1319	+ spin_unlock(&xprt->queue_lock);
1068	1320	task = rovr->rq_task;
1069	1321
1070	1322	if ((copied = rovr->rq_private_buf.buflen) > repsize)
..	..	@@ -1072,22 +1324,22 @@
1072	1324
1073	1325	/* Suck it into the iovec, verify checksum if not done by hw. */
1074	1326	if (csum_partial_copy_to_xdr(&rovr->rq_private_buf, skb)) {
1075		- spin_lock(&xprt->recv_lock);
	1327	+ spin_lock(&xprt->queue_lock);
1076	1328	__UDPX_INC_STATS(sk, UDP_MIB_INERRORS);
1077	1329	goto out_unpin;
1078	1330	}
1079	1331
1080	1332
1081		- spin_lock_bh(&xprt->transport_lock);
	1333	+ spin_lock(&xprt->transport_lock);
1082	1334	xprt_adjust_cwnd(xprt, task, copied);
1083		- spin_unlock_bh(&xprt->transport_lock);
1084		- spin_lock(&xprt->recv_lock);
	1335	+ spin_unlock(&xprt->transport_lock);
	1336	+ spin_lock(&xprt->queue_lock);
1085	1337	xprt_complete_rqst(task, copied);
1086	1338	__UDPX_INC_STATS(sk, UDP_MIB_INDATAGRAMS);
1087	1339	out_unpin:
1088	1340	xprt_unpin_rqst(rovr);
1089	1341	out_unlock:
1090		- spin_unlock(&xprt->recv_lock);
	1342	+ spin_unlock(&xprt->queue_lock);
1091	1343	}
1092	1344
1093	1345	static void xs_udp_data_receive(struct sock_xprt *transport)
..	..	@@ -1096,26 +1348,19 @@
1096	1348	struct sock *sk;
1097	1349	int err;
1098	1350
1099		-restart:
1100	1351	mutex_lock(&transport->recv_mutex);
1101	1352	sk = transport->inet;
1102	1353	if (sk == NULL)
1103	1354	goto out;
1104	1355	for (;;) {
1105	1356	skb = skb_recv_udp(sk, 0, 1, &err);
1106		- if (skb != NULL) {
1107		- xs_udp_data_read_skb(&transport->xprt, sk, skb);
1108		- consume_skb(skb);
1109		- continue;
1110		- }
1111		- if (!test_and_clear_bit(XPRT_SOCK_DATA_READY, &transport->sock_state))
	1357	+ if (skb == NULL)
1112	1358	break;
1113		- if (need_resched()) {
1114		- mutex_unlock(&transport->recv_mutex);
1115		- cond_resched();
1116		- goto restart;
1117		- }
	1359	+ xs_udp_data_read_skb(&transport->xprt, sk, skb);
	1360	+ consume_skb(skb);
	1361	+ cond_resched();
1118	1362	}
	1363	+ xs_poll_check_readable(transport);
1119	1364	out:
1120	1365	mutex_unlock(&transport->recv_mutex);
1121	1366	}
..	..	@@ -1124,7 +1369,10 @@
1124	1369	{
1125	1370	struct sock_xprt *transport =
1126	1371	container_of(work, struct sock_xprt, recv_worker);
	1372	+ unsigned int pflags = memalloc_nofs_save();
	1373	+
1127	1374	xs_udp_data_receive(transport);
	1375	+ memalloc_nofs_restore(pflags);
1128	1376	}
1129	1377
1130	1378	/**
..	..	@@ -1163,416 +1411,12 @@
1163	1411	xprt_force_disconnect(xprt);
1164	1412	}
1165	1413
1166		-static inline void xs_tcp_read_fraghdr(struct rpc_xprt xprt, struct xdr_skb_reader desc)
1167		-{
1168		- struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1169		- size_t len, used;
1170		- char *p;
1171		-
1172		- p = ((char *) &transport->tcp_fraghdr) + transport->tcp_offset;
1173		- len = sizeof(transport->tcp_fraghdr) - transport->tcp_offset;
1174		- used = xdr_skb_read_bits(desc, p, len);
1175		- transport->tcp_offset += used;
1176		- if (used != len)
1177		- return;
1178		-
1179		- transport->tcp_reclen = ntohl(transport->tcp_fraghdr);
1180		- if (transport->tcp_reclen & RPC_LAST_STREAM_FRAGMENT)
1181		- transport->tcp_flags \|= TCP_RCV_LAST_FRAG;
1182		- else
1183		- transport->tcp_flags &= ~TCP_RCV_LAST_FRAG;
1184		- transport->tcp_reclen &= RPC_FRAGMENT_SIZE_MASK;
1185		-
1186		- transport->tcp_flags &= ~TCP_RCV_COPY_FRAGHDR;
1187		- transport->tcp_offset = 0;
1188		-
1189		- /* Sanity check of the record length */
1190		- if (unlikely(transport->tcp_reclen < 8)) {
1191		- dprintk("RPC: invalid TCP record fragment length\n");
1192		- xs_tcp_force_close(xprt);
1193		- return;
1194		- }
1195		- dprintk("RPC: reading TCP record fragment of length %d\n",
1196		- transport->tcp_reclen);
1197		-}
1198		-
1199		-static void xs_tcp_check_fraghdr(struct sock_xprt *transport)
1200		-{
1201		- if (transport->tcp_offset == transport->tcp_reclen) {
1202		- transport->tcp_flags \|= TCP_RCV_COPY_FRAGHDR;
1203		- transport->tcp_offset = 0;
1204		- if (transport->tcp_flags & TCP_RCV_LAST_FRAG) {
1205		- transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
1206		- transport->tcp_flags \|= TCP_RCV_COPY_XID;
1207		- transport->tcp_copied = 0;
1208		- }
1209		- }
1210		-}
1211		-
1212		-static inline void xs_tcp_read_xid(struct sock_xprt transport, struct xdr_skb_reader desc)
1213		-{
1214		- size_t len, used;
1215		- char *p;
1216		-
1217		- len = sizeof(transport->tcp_xid) - transport->tcp_offset;
1218		- dprintk("RPC: reading XID (%zu bytes)\n", len);
1219		- p = ((char *) &transport->tcp_xid) + transport->tcp_offset;
1220		- used = xdr_skb_read_bits(desc, p, len);
1221		- transport->tcp_offset += used;
1222		- if (used != len)
1223		- return;
1224		- transport->tcp_flags &= ~TCP_RCV_COPY_XID;
1225		- transport->tcp_flags \|= TCP_RCV_READ_CALLDIR;
1226		- transport->tcp_copied = 4;
1227		- dprintk("RPC: reading %s XID %08x\n",
1228		- (transport->tcp_flags & TCP_RPC_REPLY) ? "reply for"
1229		- : "request with",
1230		- ntohl(transport->tcp_xid));
1231		- xs_tcp_check_fraghdr(transport);
1232		-}
1233		-
1234		-static inline void xs_tcp_read_calldir(struct sock_xprt *transport,
1235		- struct xdr_skb_reader *desc)
1236		-{
1237		- size_t len, used;
1238		- u32 offset;
1239		- char *p;
1240		-
1241		- /*
1242		- * We want transport->tcp_offset to be 8 at the end of this routine
1243		- * (4 bytes for the xid and 4 bytes for the call/reply flag).
1244		- * When this function is called for the first time,
1245		- * transport->tcp_offset is 4 (after having already read the xid).
1246		- */
1247		- offset = transport->tcp_offset - sizeof(transport->tcp_xid);
1248		- len = sizeof(transport->tcp_calldir) - offset;
1249		- dprintk("RPC: reading CALL/REPLY flag (%zu bytes)\n", len);
1250		- p = ((char *) &transport->tcp_calldir) + offset;
1251		- used = xdr_skb_read_bits(desc, p, len);
1252		- transport->tcp_offset += used;
1253		- if (used != len)
1254		- return;
1255		- transport->tcp_flags &= ~TCP_RCV_READ_CALLDIR;
1256		- /*
1257		- * We don't yet have the XDR buffer, so we will write the calldir
1258		- * out after we get the buffer from the 'struct rpc_rqst'
1259		- */
1260		- switch (ntohl(transport->tcp_calldir)) {
1261		- case RPC_REPLY:
1262		- transport->tcp_flags \|= TCP_RCV_COPY_CALLDIR;
1263		- transport->tcp_flags \|= TCP_RCV_COPY_DATA;
1264		- transport->tcp_flags \|= TCP_RPC_REPLY;
1265		- break;
1266		- case RPC_CALL:
1267		- transport->tcp_flags \|= TCP_RCV_COPY_CALLDIR;
1268		- transport->tcp_flags \|= TCP_RCV_COPY_DATA;
1269		- transport->tcp_flags &= ~TCP_RPC_REPLY;
1270		- break;
1271		- default:
1272		- dprintk("RPC: invalid request message type\n");
1273		- xs_tcp_force_close(&transport->xprt);
1274		- }
1275		- xs_tcp_check_fraghdr(transport);
1276		-}
1277		-
1278		-static inline void xs_tcp_read_common(struct rpc_xprt *xprt,
1279		- struct xdr_skb_reader *desc,
1280		- struct rpc_rqst *req)
1281		-{
1282		- struct sock_xprt *transport =
1283		- container_of(xprt, struct sock_xprt, xprt);
1284		- struct xdr_buf *rcvbuf;
1285		- size_t len;
1286		- ssize_t r;
1287		-
1288		- rcvbuf = &req->rq_private_buf;
1289		-
1290		- if (transport->tcp_flags & TCP_RCV_COPY_CALLDIR) {
1291		- /*
1292		- * Save the RPC direction in the XDR buffer
1293		- */
1294		- memcpy(rcvbuf->head[0].iov_base + transport->tcp_copied,
1295		- &transport->tcp_calldir,
1296		- sizeof(transport->tcp_calldir));
1297		- transport->tcp_copied += sizeof(transport->tcp_calldir);
1298		- transport->tcp_flags &= ~TCP_RCV_COPY_CALLDIR;
1299		- }
1300		-
1301		- len = desc->count;
1302		- if (len > transport->tcp_reclen - transport->tcp_offset)
1303		- desc->count = transport->tcp_reclen - transport->tcp_offset;
1304		- r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
1305		- desc, xdr_skb_read_bits);
1306		-
1307		- if (desc->count) {
1308		- /* Error when copying to the receive buffer,
1309		- * usually because we weren't able to allocate
1310		- * additional buffer pages. All we can do now
1311		- * is turn off TCP_RCV_COPY_DATA, so the request
1312		- * will not receive any additional updates,
1313		- * and time out.
1314		- * Any remaining data from this record will
1315		- * be discarded.
1316		- */
1317		- transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
1318		- dprintk("RPC: XID %08x truncated request\n",
1319		- ntohl(transport->tcp_xid));
1320		- dprintk("RPC: xprt = %p, tcp_copied = %lu, "
1321		- "tcp_offset = %u, tcp_reclen = %u\n",
1322		- xprt, transport->tcp_copied,
1323		- transport->tcp_offset, transport->tcp_reclen);
1324		- return;
1325		- }
1326		-
1327		- transport->tcp_copied += r;
1328		- transport->tcp_offset += r;
1329		- desc->count = len - r;
1330		-
1331		- dprintk("RPC: XID %08x read %zd bytes\n",
1332		- ntohl(transport->tcp_xid), r);
1333		- dprintk("RPC: xprt = %p, tcp_copied = %lu, tcp_offset = %u, "
1334		- "tcp_reclen = %u\n", xprt, transport->tcp_copied,
1335		- transport->tcp_offset, transport->tcp_reclen);
1336		-
1337		- if (transport->tcp_copied == req->rq_private_buf.buflen)
1338		- transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
1339		- else if (transport->tcp_offset == transport->tcp_reclen) {
1340		- if (transport->tcp_flags & TCP_RCV_LAST_FRAG)
1341		- transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
1342		- }
1343		-}
1344		-
1345		-/*
1346		- * Finds the request corresponding to the RPC xid and invokes the common
1347		- * tcp read code to read the data.
1348		- */
1349		-static inline int xs_tcp_read_reply(struct rpc_xprt *xprt,
1350		- struct xdr_skb_reader *desc)
1351		-{
1352		- struct sock_xprt *transport =
1353		- container_of(xprt, struct sock_xprt, xprt);
1354		- struct rpc_rqst *req;
1355		-
1356		- dprintk("RPC: read reply XID %08x\n", ntohl(transport->tcp_xid));
1357		-
1358		- /* Find and lock the request corresponding to this xid */
1359		- spin_lock(&xprt->recv_lock);
1360		- req = xprt_lookup_rqst(xprt, transport->tcp_xid);
1361		- if (!req) {
1362		- dprintk("RPC: XID %08x request not found!\n",
1363		- ntohl(transport->tcp_xid));
1364		- spin_unlock(&xprt->recv_lock);
1365		- return -1;
1366		- }
1367		- xprt_pin_rqst(req);
1368		- spin_unlock(&xprt->recv_lock);
1369		-
1370		- xs_tcp_read_common(xprt, desc, req);
1371		-
1372		- spin_lock(&xprt->recv_lock);
1373		- if (!(transport->tcp_flags & TCP_RCV_COPY_DATA))
1374		- xprt_complete_rqst(req->rq_task, transport->tcp_copied);
1375		- xprt_unpin_rqst(req);
1376		- spin_unlock(&xprt->recv_lock);
1377		- return 0;
1378		-}
1379		-
1380	1414	#if defined(CONFIG_SUNRPC_BACKCHANNEL)
1381		-/*
1382		- * Obtains an rpc_rqst previously allocated and invokes the common
1383		- * tcp read code to read the data. The result is placed in the callback
1384		- * queue.
1385		- * If we're unable to obtain the rpc_rqst we schedule the closing of the
1386		- * connection and return -1.
1387		- */
1388		-static int xs_tcp_read_callback(struct rpc_xprt *xprt,
1389		- struct xdr_skb_reader *desc)
1390		-{
1391		- struct sock_xprt *transport =
1392		- container_of(xprt, struct sock_xprt, xprt);
1393		- struct rpc_rqst *req;
1394		-
1395		- /* Look up the request corresponding to the given XID */
1396		- req = xprt_lookup_bc_request(xprt, transport->tcp_xid);
1397		- if (req == NULL) {
1398		- printk(KERN_WARNING "Callback slot table overflowed\n");
1399		- xprt_force_disconnect(xprt);
1400		- return -1;
1401		- }
1402		-
1403		- dprintk("RPC: read callback XID %08x\n", ntohl(req->rq_xid));
1404		- xs_tcp_read_common(xprt, desc, req);
1405		-
1406		- if (!(transport->tcp_flags & TCP_RCV_COPY_DATA))
1407		- xprt_complete_bc_request(req, transport->tcp_copied);
1408		-
1409		- return 0;
1410		-}
1411		-
1412		-static inline int _xs_tcp_read_data(struct rpc_xprt *xprt,
1413		- struct xdr_skb_reader *desc)
1414		-{
1415		- struct sock_xprt *transport =
1416		- container_of(xprt, struct sock_xprt, xprt);
1417		-
1418		- return (transport->tcp_flags & TCP_RPC_REPLY) ?
1419		- xs_tcp_read_reply(xprt, desc) :
1420		- xs_tcp_read_callback(xprt, desc);
1421		-}
1422		-
1423		-static int xs_tcp_bc_up(struct svc_serv serv, struct net net)
1424		-{
1425		- int ret;
1426		-
1427		- ret = svc_create_xprt(serv, "tcp-bc", net, PF_INET, 0,
1428		- SVC_SOCK_ANONYMOUS);
1429		- if (ret < 0)
1430		- return ret;
1431		- return 0;
1432		-}
1433		-
1434	1415	static size_t xs_tcp_bc_maxpayload(struct rpc_xprt *xprt)
1435	1416	{
1436	1417	return PAGE_SIZE;
1437	1418	}
1438		-#else
1439		-static inline int _xs_tcp_read_data(struct rpc_xprt *xprt,
1440		- struct xdr_skb_reader *desc)
1441		-{
1442		- return xs_tcp_read_reply(xprt, desc);
1443		-}
1444	1419	#endif /* CONFIG_SUNRPC_BACKCHANNEL */
1445		-
1446		-/*
1447		- * Read data off the transport. This can be either an RPC_CALL or an
1448		- * RPC_REPLY. Relay the processing to helper functions.
1449		- */
1450		-static void xs_tcp_read_data(struct rpc_xprt *xprt,
1451		- struct xdr_skb_reader *desc)
1452		-{
1453		- struct sock_xprt *transport =
1454		- container_of(xprt, struct sock_xprt, xprt);
1455		-
1456		- if (_xs_tcp_read_data(xprt, desc) == 0)
1457		- xs_tcp_check_fraghdr(transport);
1458		- else {
1459		- /*
1460		- * The transport_lock protects the request handling.
1461		- * There's no need to hold it to update the tcp_flags.
1462		- */
1463		- transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
1464		- }
1465		-}
1466		-
1467		-static inline void xs_tcp_read_discard(struct sock_xprt transport, struct xdr_skb_reader desc)
1468		-{
1469		- size_t len;
1470		-
1471		- len = transport->tcp_reclen - transport->tcp_offset;
1472		- if (len > desc->count)
1473		- len = desc->count;
1474		- desc->count -= len;
1475		- desc->offset += len;
1476		- transport->tcp_offset += len;
1477		- dprintk("RPC: discarded %zu bytes\n", len);
1478		- xs_tcp_check_fraghdr(transport);
1479		-}
1480		-
1481		-static int xs_tcp_data_recv(read_descriptor_t rd_desc, struct sk_buff skb, unsigned int offset, size_t len)
1482		-{
1483		- struct rpc_xprt *xprt = rd_desc->arg.data;
1484		- struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1485		- struct xdr_skb_reader desc = {
1486		- .skb = skb,
1487		- .offset = offset,
1488		- .count = len,
1489		- };
1490		- size_t ret;
1491		-
1492		- dprintk("RPC: xs_tcp_data_recv started\n");
1493		- do {
1494		- trace_xs_tcp_data_recv(transport);
1495		- /* Read in a new fragment marker if necessary */
1496		- /* Can we ever really expect to get completely empty fragments? */
1497		- if (transport->tcp_flags & TCP_RCV_COPY_FRAGHDR) {
1498		- xs_tcp_read_fraghdr(xprt, &desc);
1499		- continue;
1500		- }
1501		- /* Read in the xid if necessary */
1502		- if (transport->tcp_flags & TCP_RCV_COPY_XID) {
1503		- xs_tcp_read_xid(transport, &desc);
1504		- continue;
1505		- }
1506		- /* Read in the call/reply flag */
1507		- if (transport->tcp_flags & TCP_RCV_READ_CALLDIR) {
1508		- xs_tcp_read_calldir(transport, &desc);
1509		- continue;
1510		- }
1511		- /* Read in the request data */
1512		- if (transport->tcp_flags & TCP_RCV_COPY_DATA) {
1513		- xs_tcp_read_data(xprt, &desc);
1514		- continue;
1515		- }
1516		- /* Skip over any trailing bytes on short reads */
1517		- xs_tcp_read_discard(transport, &desc);
1518		- } while (desc.count);
1519		- ret = len - desc.count;
1520		- if (ret < rd_desc->count)
1521		- rd_desc->count -= ret;
1522		- else
1523		- rd_desc->count = 0;
1524		- trace_xs_tcp_data_recv(transport);
1525		- dprintk("RPC: xs_tcp_data_recv done\n");
1526		- return ret;
1527		-}
1528		-
1529		-static void xs_tcp_data_receive(struct sock_xprt *transport)
1530		-{
1531		- struct rpc_xprt *xprt = &transport->xprt;
1532		- struct sock *sk;
1533		- read_descriptor_t rd_desc = {
1534		- .arg.data = xprt,
1535		- };
1536		- unsigned long total = 0;
1537		- int read = 0;
1538		-
1539		-restart:
1540		- mutex_lock(&transport->recv_mutex);
1541		- sk = transport->inet;
1542		- if (sk == NULL)
1543		- goto out;
1544		-
1545		- /* We use rd_desc to pass struct xprt to xs_tcp_data_recv */
1546		- for (;;) {
1547		- rd_desc.count = RPC_TCP_READ_CHUNK_SZ;
1548		- lock_sock(sk);
1549		- read = tcp_read_sock(sk, &rd_desc, xs_tcp_data_recv);
1550		- if (rd_desc.count != 0 \|\| read < 0) {
1551		- clear_bit(XPRT_SOCK_DATA_READY, &transport->sock_state);
1552		- release_sock(sk);
1553		- break;
1554		- }
1555		- release_sock(sk);
1556		- total += read;
1557		- if (need_resched()) {
1558		- mutex_unlock(&transport->recv_mutex);
1559		- cond_resched();
1560		- goto restart;
1561		- }
1562		- }
1563		- if (test_bit(XPRT_SOCK_DATA_READY, &transport->sock_state))
1564		- queue_work(xprtiod_workqueue, &transport->recv_worker);
1565		-out:
1566		- mutex_unlock(&transport->recv_mutex);
1567		- trace_xs_tcp_data_ready(xprt, read, total);
1568		-}
1569		-
1570		-static void xs_tcp_data_receive_workfn(struct work_struct *work)
1571		-{
1572		- struct sock_xprt *transport =
1573		- container_of(work, struct sock_xprt, recv_worker);
1574		- xs_tcp_data_receive(transport);
1575		-}
1576	1420
1577	1421	/**
1578	1422	* xs_tcp_state_change - callback to handle TCP socket state changes
..	..	@@ -1598,15 +1442,7 @@
1598	1442	trace_rpc_socket_state_change(xprt, sk->sk_socket);
1599	1443	switch (sk->sk_state) {
1600	1444	case TCP_ESTABLISHED:
1601		- spin_lock(&xprt->transport_lock);
1602	1445	if (!xprt_test_and_set_connected(xprt)) {
1603		-
1604		- /* Reset TCP record info */
1605		- transport->tcp_offset = 0;
1606		- transport->tcp_reclen = 0;
1607		- transport->tcp_copied = 0;
1608		- transport->tcp_flags =
1609		- TCP_RCV_COPY_FRAGHDR \| TCP_RCV_COPY_XID;
1610	1446	xprt->connect_cookie++;
1611	1447	clear_bit(XPRT_SOCK_CONNECTING, &transport->sock_state);
1612	1448	xprt_clear_connecting(xprt);
..	..	@@ -1614,9 +1450,8 @@
1614	1450	xprt->stat.connect_count++;
1615	1451	xprt->stat.connect_time += (long)jiffies -
1616	1452	xprt->stat.connect_start;
1617		- xprt_wake_pending_tasks(xprt, -EAGAIN);
	1453	+ xs_run_error_worker(transport, XPRT_SOCK_WAKE_PENDING);
1618	1454	}
1619		- spin_unlock(&xprt->transport_lock);
1620	1455	break;
1621	1456	case TCP_FIN_WAIT1:
1622	1457	/* The client initiated a shutdown of the socket */
..	..	@@ -1632,8 +1467,8 @@
1632	1467	/* The server initiated a shutdown of the socket */
1633	1468	xprt->connect_cookie++;
1634	1469	clear_bit(XPRT_CONNECTED, &xprt->state);
1635		- xs_tcp_force_close(xprt);
1636		- /* fall through */
	1470	+ xs_run_error_worker(transport, XPRT_SOCK_WAKE_DISCONNECT);
	1471	+ fallthrough;
1637	1472	case TCP_CLOSING:
1638	1473	/*
1639	1474	* If the server closed down the connection, make sure that
..	..	@@ -1653,10 +1488,8 @@
1653	1488	&transport->sock_state))
1654	1489	xprt_clear_connecting(xprt);
1655	1490	clear_bit(XPRT_CLOSING, &xprt->state);
1656		- if (sk->sk_err)
1657		- xprt_wake_pending_tasks(xprt, -sk->sk_err);
1658	1491	/* Trigger the socket release */
1659		- xs_tcp_force_close(xprt);
	1492	+ xs_run_error_worker(transport, XPRT_SOCK_WAKE_DISCONNECT);
1660	1493	}
1661	1494	out:
1662	1495	read_unlock_bh(&sk->sk_callback_lock);
..	..	@@ -1665,6 +1498,7 @@
1665	1498	static void xs_write_space(struct sock *sk)
1666	1499	{
1667	1500	struct socket_wq *wq;
	1501	+ struct sock_xprt *transport;
1668	1502	struct rpc_xprt *xprt;
1669	1503
1670	1504	if (!sk->sk_socket)
..	..	@@ -1673,12 +1507,14 @@
1673	1507
1674	1508	if (unlikely(!(xprt = xprt_from_sock(sk))))
1675	1509	return;
	1510	+ transport = container_of(xprt, struct sock_xprt, xprt);
1676	1511	rcu_read_lock();
1677	1512	wq = rcu_dereference(sk->sk_wq);
1678	1513	if (!wq \|\| test_and_clear_bit(SOCKWQ_ASYNC_NOSPACE, &wq->flags) == 0)
1679	1514	goto out;
1680	1515
1681		- xprt_write_space(xprt);
	1516	+ xs_run_error_worker(transport, XPRT_SOCK_WAKE_WRITE);
	1517	+ sk->sk_write_pending--;
1682	1518	out:
1683	1519	rcu_read_unlock();
1684	1520	}
..	..	@@ -1765,15 +1601,16 @@
1765	1601
1766	1602	/**
1767	1603	* xs_udp_timer - called when a retransmit timeout occurs on a UDP transport
	1604	+ * @xprt: controlling transport
1768	1605	* @task: task that timed out
1769	1606	*
1770	1607	* Adjust the congestion window after a retransmit timeout has occurred.
1771	1608	*/
1772	1609	static void xs_udp_timer(struct rpc_xprt xprt, struct rpc_task task)
1773	1610	{
1774		- spin_lock_bh(&xprt->transport_lock);
	1611	+ spin_lock(&xprt->transport_lock);
1775	1612	xprt_adjust_cwnd(xprt, task, -ETIMEDOUT);
1776		- spin_unlock_bh(&xprt->transport_lock);
	1613	+ spin_unlock(&xprt->transport_lock);
1777	1614	}
1778	1615
1779	1616	static int xs_get_random_port(void)
..	..	@@ -1787,21 +1624,6 @@
1787	1624	range = max - min + 1;
1788	1625	rand = (unsigned short) prandom_u32() % range;
1789	1626	return rand + min;
1790		-}
1791		-
1792		-/**
1793		- * xs_set_reuseaddr_port - set the socket's port and address reuse options
1794		- * @sock: socket
1795		- *
1796		- * Note that this function has to be called on all sockets that share the
1797		- * same port, and it must be called before binding.
1798		- */
1799		-static void xs_sock_set_reuseport(struct socket *sock)
1800		-{
1801		- int opt = 1;
1802		-
1803		- kernel_setsockopt(sock, SOL_SOCKET, SO_REUSEPORT,
1804		- (char *)&opt, sizeof(opt));
1805	1627	}
1806	1628
1807	1629	static unsigned short xs_sock_getport(struct socket *sock)
..	..	@@ -1838,7 +1660,7 @@
1838	1660
1839	1661	static void xs_set_srcport(struct sock_xprt transport, struct socket sock)
1840	1662	{
1841		- if (transport->srcport == 0)
	1663	+ if (transport->srcport == 0 && transport->xprt.reuseport)
1842	1664	transport->srcport = xs_sock_getport(sock);
1843	1665	}
1844	1666
..	..	@@ -1850,6 +1672,13 @@
1850	1672	port = xs_get_random_port();
1851	1673	return port;
1852	1674	}
	1675	+
	1676	+unsigned short get_srcport(struct rpc_xprt *xprt)
	1677	+{
	1678	+ struct sock_xprt *sock = container_of(xprt, struct sock_xprt, xprt);
	1679	+ return xs_sock_getport(sock->sock);
	1680	+}
	1681	+EXPORT_SYMBOL(get_srcport);
1853	1682
1854	1683	static unsigned short xs_next_srcport(struct sock_xprt *transport, unsigned short port)
1855	1684	{
..	..	@@ -1892,7 +1721,8 @@
1892	1721	err = kernel_bind(sock, (struct sockaddr *)&myaddr,
1893	1722	transport->xprt.addrlen);
1894	1723	if (err == 0) {
1895		- transport->srcport = port;
	1724	+ if (transport->xprt.reuseport)
	1725	+ transport->srcport = port;
1896	1726	break;
1897	1727	}
1898	1728	last = port;
..	..	@@ -1983,6 +1813,7 @@
1983	1813	struct sock_xprt *transport, int family, int type,
1984	1814	int protocol, bool reuseport)
1985	1815	{
	1816	+ struct file *filp;
1986	1817	struct socket *sock;
1987	1818	int err;
1988	1819
..	..	@@ -1995,13 +1826,18 @@
1995	1826	xs_reclassify_socket(family, sock);
1996	1827
1997	1828	if (reuseport)
1998		- xs_sock_set_reuseport(sock);
	1829	+ sock_set_reuseport(sock->sk);
1999	1830
2000	1831	err = xs_bind(transport, sock);
2001	1832	if (err) {
2002	1833	sock_release(sock);
2003	1834	goto out;
2004	1835	}
	1836	+
	1837	+ filp = sock_alloc_file(sock, O_NONBLOCK, NULL);
	1838	+ if (IS_ERR(filp))
	1839	+ return ERR_CAST(filp);
	1840	+ transport->file = filp;
2005	1841
2006	1842	return sock;
2007	1843	out:
..	..	@@ -2026,7 +1862,6 @@
2026	1862	sk->sk_write_space = xs_udp_write_space;
2027	1863	sock_set_flag(sk, SOCK_FASYNC);
2028	1864	sk->sk_error_report = xs_error_report;
2029		- sk->sk_allocation = GFP_NOIO;
2030	1865
2031	1866	xprt_clear_connected(xprt);
2032	1867
..	..	@@ -2037,7 +1872,8 @@
2037	1872	write_unlock_bh(&sk->sk_callback_lock);
2038	1873	}
2039	1874
2040		- /* Tell the socket layer to start connecting... */
	1875	+ xs_stream_start_connect(transport);
	1876	+
2041	1877	return kernel_connect(sock, xs_addr(xprt), xprt->addrlen, 0);
2042	1878	}
2043	1879
..	..	@@ -2048,8 +1884,9 @@
2048	1884	static int xs_local_setup_socket(struct sock_xprt *transport)
2049	1885	{
2050	1886	struct rpc_xprt *xprt = &transport->xprt;
	1887	+ struct file *filp;
2051	1888	struct socket *sock;
2052		- int status = -EIO;
	1889	+ int status;
2053	1890
2054	1891	status = __sock_create(xprt->xprt_net, AF_LOCAL,
2055	1892	SOCK_STREAM, 0, &sock, 1);
..	..	@@ -2059,6 +1896,13 @@
2059	1896	goto out;
2060	1897	}
2061	1898	xs_reclassify_socket(AF_LOCAL, sock);
	1899	+
	1900	+ filp = sock_alloc_file(sock, O_NONBLOCK, NULL);
	1901	+ if (IS_ERR(filp)) {
	1902	+ status = PTR_ERR(filp);
	1903	+ goto out;
	1904	+ }
	1905	+ transport->file = filp;
2062	1906
2063	1907	dprintk("RPC: worker connecting xprt %p via AF_LOCAL to %s\n",
2064	1908	xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
..	..	@@ -2073,6 +1917,7 @@
2073	1917	xprt->stat.connect_time += (long)jiffies -
2074	1918	xprt->stat.connect_start;
2075	1919	xprt_set_connected(xprt);
	1920	+ break;
2076	1921	case -ENOBUFS:
2077	1922	break;
2078	1923	case -ENOENT:
..	..	@@ -2110,6 +1955,7 @@
2110	1955	* we'll need to figure out how to pass a namespace to
2111	1956	* connect.
2112	1957	*/
	1958	+ task->tk_rpc_status = -ENOTCONN;
2113	1959	rpc_exit(task, -ENOTCONN);
2114	1960	return;
2115	1961	}
..	..	@@ -2213,7 +2059,6 @@
2213	2059	sk->sk_data_ready = xs_data_ready;
2214	2060	sk->sk_write_space = xs_udp_write_space;
2215	2061	sock_set_flag(sk, SOCK_FASYNC);
2216		- sk->sk_allocation = GFP_NOIO;
2217	2062
2218	2063	xprt_set_connected(xprt);
2219	2064
..	..	@@ -2291,30 +2136,24 @@
2291	2136	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2292	2137	unsigned int keepidle;
2293	2138	unsigned int keepcnt;
2294		- unsigned int opt_on = 1;
2295	2139	unsigned int timeo;
2296	2140
2297		- spin_lock_bh(&xprt->transport_lock);
	2141	+ spin_lock(&xprt->transport_lock);
2298	2142	keepidle = DIV_ROUND_UP(xprt->timeout->to_initval, HZ);
2299	2143	keepcnt = xprt->timeout->to_retries + 1;
2300	2144	timeo = jiffies_to_msecs(xprt->timeout->to_initval) *
2301	2145	(xprt->timeout->to_retries + 1);
2302	2146	clear_bit(XPRT_SOCK_UPD_TIMEOUT, &transport->sock_state);
2303		- spin_unlock_bh(&xprt->transport_lock);
	2147	+ spin_unlock(&xprt->transport_lock);
2304	2148
2305	2149	/* TCP Keepalive options */
2306		- kernel_setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE,
2307		- (char *)&opt_on, sizeof(opt_on));
2308		- kernel_setsockopt(sock, SOL_TCP, TCP_KEEPIDLE,
2309		- (char *)&keepidle, sizeof(keepidle));
2310		- kernel_setsockopt(sock, SOL_TCP, TCP_KEEPINTVL,
2311		- (char *)&keepidle, sizeof(keepidle));
2312		- kernel_setsockopt(sock, SOL_TCP, TCP_KEEPCNT,
2313		- (char *)&keepcnt, sizeof(keepcnt));
	2150	+ sock_set_keepalive(sock->sk);
	2151	+ tcp_sock_set_keepidle(sock->sk, keepidle);
	2152	+ tcp_sock_set_keepintvl(sock->sk, keepidle);
	2153	+ tcp_sock_set_keepcnt(sock->sk, keepcnt);
2314	2154
2315	2155	/* TCP user timeout (see RFC5482) */
2316		- kernel_setsockopt(sock, SOL_TCP, TCP_USER_TIMEOUT,
2317		- (char *)&timeo, sizeof(timeo));
	2156	+ tcp_sock_set_user_timeout(sock->sk, timeo);
2318	2157	}
2319	2158
2320	2159	static void xs_tcp_set_connect_timeout(struct rpc_xprt *xprt,
..	..	@@ -2325,7 +2164,7 @@
2325	2164	struct rpc_timeout to;
2326	2165	unsigned long initval;
2327	2166
2328		- spin_lock_bh(&xprt->transport_lock);
	2167	+ spin_lock(&xprt->transport_lock);
2329	2168	if (reconnect_timeout < xprt->max_reconnect_timeout)
2330	2169	xprt->max_reconnect_timeout = reconnect_timeout;
2331	2170	if (connect_timeout < xprt->connect_timeout) {
..	..	@@ -2342,7 +2181,7 @@
2342	2181	xprt->connect_timeout = connect_timeout;
2343	2182	}
2344	2183	set_bit(XPRT_SOCK_UPD_TIMEOUT, &transport->sock_state);
2345		- spin_unlock_bh(&xprt->transport_lock);
	2184	+ spin_unlock(&xprt->transport_lock);
2346	2185	}
2347	2186
2348	2187	static int xs_tcp_finish_connecting(struct rpc_xprt xprt, struct socket sock)
..	..	@@ -2352,7 +2191,6 @@
2352	2191
2353	2192	if (!transport->inet) {
2354	2193	struct sock *sk = sock->sk;
2355		- unsigned int addr_pref = IPV6_PREFER_SRC_PUBLIC;
2356	2194
2357	2195	/* Avoid temporary address, they are bad for long-lived
2358	2196	* connections such as NFS mounts.
..	..	@@ -2361,8 +2199,10 @@
2361	2199	* knowledge about the normal duration of connections,
2362	2200	* MAY override this as appropriate.
2363	2201	*/
2364		- kernel_setsockopt(sock, SOL_IPV6, IPV6_ADDR_PREFERENCES,
2365		- (char *)&addr_pref, sizeof(addr_pref));
	2202	+ if (xs_addr(xprt)->sa_family == PF_INET6) {
	2203	+ ip6_sock_set_addr_preferences(sk,
	2204	+ IPV6_PREFER_SRC_PUBLIC);
	2205	+ }
2366	2206
2367	2207	xs_tcp_set_socket_timeouts(xprt, sock);
2368	2208
..	..	@@ -2376,7 +2216,6 @@
2376	2216	sk->sk_write_space = xs_tcp_write_space;
2377	2217	sock_set_flag(sk, SOCK_FASYNC);
2378	2218	sk->sk_error_report = xs_error_report;
2379		- sk->sk_allocation = GFP_NOIO;
2380	2219
2381	2220	/* socket options */
2382	2221	sock_reset_flag(sk, SOCK_LINGER);
..	..	@@ -2396,13 +2235,15 @@
2396	2235
2397	2236	xs_set_memalloc(xprt);
2398	2237
	2238	+ xs_stream_start_connect(transport);
	2239	+
2399	2240	/* Tell the socket layer to start connecting... */
2400	2241	set_bit(XPRT_SOCK_CONNECTING, &transport->sock_state);
2401	2242	ret = kernel_connect(sock, xs_addr(xprt), xprt->addrlen, O_NONBLOCK);
2402	2243	switch (ret) {
2403	2244	case 0:
2404	2245	xs_set_srcport(transport, sock);
2405		- /* fall through */
	2246	+ fallthrough;
2406	2247	case -EINPROGRESS:
2407	2248	/* SYN_SENT! */
2408	2249	if (xprt->reestablish_timeout < XS_TCP_INIT_REEST_TO)
..	..	@@ -2418,6 +2259,7 @@
2418	2259
2419	2260	/**
2420	2261	* xs_tcp_setup_socket - create a TCP socket and connect to a remote endpoint
	2262	+ * @work: queued work item
2421	2263	*
2422	2264	* Invoked by a work queue tasklet.
2423	2265	*/
..	..	@@ -2429,10 +2271,14 @@
2429	2271	struct rpc_xprt *xprt = &transport->xprt;
2430	2272	int status = -EIO;
2431	2273
2432		- if (!sock) {
2433		- sock = xs_create_sock(xprt, transport,
2434		- xs_addr(xprt)->sa_family, SOCK_STREAM,
2435		- IPPROTO_TCP, true);
	2274	+ if (xprt_connected(xprt))
	2275	+ goto out;
	2276	+ if (test_and_clear_bit(XPRT_SOCK_CONNECT_SENT,
	2277	+ &transport->sock_state) \|\|
	2278	+ !sock) {
	2279	+ xs_reset_transport(transport);
	2280	+ sock = xs_create_sock(xprt, transport, xs_addr(xprt)->sa_family,
	2281	+ SOCK_STREAM, IPPROTO_TCP, true);
2436	2282	if (IS_ERR(sock)) {
2437	2283	status = PTR_ERR(sock);
2438	2284	goto out;
..	..	@@ -2454,7 +2300,7 @@
2454	2300	default:
2455	2301	printk("%s: connect returned unhandled error %d\n",
2456	2302	__func__, status);
2457		- /* fall through */
	2303	+ fallthrough;
2458	2304	case -EADDRNOTAVAIL:
2459	2305	/* We're probably in TIME_WAIT. Get rid of existing socket,
2460	2306	* and retry
..	..	@@ -2463,6 +2309,8 @@
2463	2309	break;
2464	2310	case 0:
2465	2311	case -EINPROGRESS:
	2312	+ set_bit(XPRT_SOCK_CONNECT_SENT, &transport->sock_state);
	2313	+ fallthrough;
2466	2314	case -EALREADY:
2467	2315	xprt_unlock_connect(xprt, transport);
2468	2316	return;
..	..	@@ -2493,25 +2341,6 @@
2493	2341	xprt_wake_pending_tasks(xprt, status);
2494	2342	}
2495	2343
2496		-static unsigned long xs_reconnect_delay(const struct rpc_xprt *xprt)
2497		-{
2498		- unsigned long start, now = jiffies;
2499		-
2500		- start = xprt->stat.connect_start + xprt->reestablish_timeout;
2501		- if (time_after(start, now))
2502		- return start - now;
2503		- return 0;
2504		-}
2505		-
2506		-static void xs_reconnect_backoff(struct rpc_xprt *xprt)
2507		-{
2508		- xprt->reestablish_timeout <<= 1;
2509		- if (xprt->reestablish_timeout > xprt->max_reconnect_timeout)
2510		- xprt->reestablish_timeout = xprt->max_reconnect_timeout;
2511		- if (xprt->reestablish_timeout < XS_TCP_INIT_REEST_TO)
2512		- xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
2513		-}
2514		-
2515	2344	/**
2516	2345	* xs_connect - connect a socket to a remote endpoint
2517	2346	* @xprt: pointer to transport structure
..	..	@@ -2535,14 +2364,10 @@
2535	2364
2536	2365	if (transport->sock != NULL) {
2537	2366	dprintk("RPC: xs_connect delayed xprt %p for %lu "
2538		- "seconds\n",
2539		- xprt, xprt->reestablish_timeout / HZ);
	2367	+ "seconds\n", xprt, xprt->reestablish_timeout / HZ);
2540	2368
2541		- /* Start by resetting any existing state */
2542		- xs_reset_transport(transport);
2543		-
2544		- delay = xs_reconnect_delay(xprt);
2545		- xs_reconnect_backoff(xprt);
	2369	+ delay = xprt_reconnect_delay(xprt);
	2370	+ xprt_reconnect_backoff(xprt, XS_TCP_INIT_REEST_TO);
2546	2371
2547	2372	} else
2548	2373	dprintk("RPC: xs_connect scheduled xprt %p\n", xprt);
..	..	@@ -2550,6 +2375,53 @@
2550	2375	queue_delayed_work(xprtiod_workqueue,
2551	2376	&transport->connect_worker,
2552	2377	delay);
	2378	+}
	2379	+
	2380	+static void xs_wake_disconnect(struct sock_xprt *transport)
	2381	+{
	2382	+ if (test_and_clear_bit(XPRT_SOCK_WAKE_DISCONNECT, &transport->sock_state))
	2383	+ xs_tcp_force_close(&transport->xprt);
	2384	+}
	2385	+
	2386	+static void xs_wake_write(struct sock_xprt *transport)
	2387	+{
	2388	+ if (test_and_clear_bit(XPRT_SOCK_WAKE_WRITE, &transport->sock_state))
	2389	+ xprt_write_space(&transport->xprt);
	2390	+}
	2391	+
	2392	+static void xs_wake_error(struct sock_xprt *transport)
	2393	+{
	2394	+ int sockerr;
	2395	+
	2396	+ if (!test_bit(XPRT_SOCK_WAKE_ERROR, &transport->sock_state))
	2397	+ return;
	2398	+ mutex_lock(&transport->recv_mutex);
	2399	+ if (transport->sock == NULL)
	2400	+ goto out;
	2401	+ if (!test_and_clear_bit(XPRT_SOCK_WAKE_ERROR, &transport->sock_state))
	2402	+ goto out;
	2403	+ sockerr = xchg(&transport->xprt_err, 0);
	2404	+ if (sockerr < 0)
	2405	+ xprt_wake_pending_tasks(&transport->xprt, sockerr);
	2406	+out:
	2407	+ mutex_unlock(&transport->recv_mutex);
	2408	+}
	2409	+
	2410	+static void xs_wake_pending(struct sock_xprt *transport)
	2411	+{
	2412	+ if (test_and_clear_bit(XPRT_SOCK_WAKE_PENDING, &transport->sock_state))
	2413	+ xprt_wake_pending_tasks(&transport->xprt, -EAGAIN);
	2414	+}
	2415	+
	2416	+static void xs_error_handle(struct work_struct *work)
	2417	+{
	2418	+ struct sock_xprt *transport = container_of(work,
	2419	+ struct sock_xprt, error_worker);
	2420	+
	2421	+ xs_wake_disconnect(transport);
	2422	+ xs_wake_write(transport);
	2423	+ xs_wake_error(transport);
	2424	+ xs_wake_pending(transport);
2553	2425	}
2554	2426
2555	2427	/**
..	..	@@ -2569,7 +2441,7 @@
2569	2441	"%llu %llu %lu %llu %llu\n",
2570	2442	xprt->stat.bind_count,
2571	2443	xprt->stat.connect_count,
2572		- xprt->stat.connect_time,
	2444	+ xprt->stat.connect_time / HZ,
2573	2445	idle_time,
2574	2446	xprt->stat.sends,
2575	2447	xprt->stat.recvs,
..	..	@@ -2624,7 +2496,7 @@
2624	2496	transport->srcport,
2625	2497	xprt->stat.bind_count,
2626	2498	xprt->stat.connect_count,
2627		- xprt->stat.connect_time,
	2499	+ xprt->stat.connect_time / HZ,
2628	2500	idle_time,
2629	2501	xprt->stat.sends,
2630	2502	xprt->stat.recvs,
..	..	@@ -2678,47 +2550,43 @@
2678	2550	free_page((unsigned long)buf);
2679	2551	}
2680	2552
2681		-/*
2682		- * Use the svc_sock to send the callback. Must be called with svsk->sk_mutex
2683		- * held. Borrows heavily from svc_tcp_sendto and xs_tcp_send_request.
2684		- */
2685	2553	static int bc_sendto(struct rpc_rqst *req)
2686	2554	{
2687		- int len;
2688		- struct xdr_buf *xbufp = &req->rq_snd_buf;
2689		- struct rpc_xprt *xprt = req->rq_xprt;
	2555	+ struct xdr_buf *xdr = &req->rq_snd_buf;
2690	2556	struct sock_xprt *transport =
2691		- container_of(xprt, struct sock_xprt, xprt);
2692		- struct socket *sock = transport->sock;
2693		- unsigned long headoff;
2694		- unsigned long tailoff;
	2557	+ container_of(req->rq_xprt, struct sock_xprt, xprt);
	2558	+ struct msghdr msg = {
	2559	+ .msg_flags = 0,
	2560	+ };
	2561	+ rpc_fraghdr marker = cpu_to_be32(RPC_LAST_STREAM_FRAGMENT \|
	2562	+ (u32)xdr->len);
	2563	+ unsigned int sent = 0;
	2564	+ int err;
2695	2565
2696		- xs_encode_stream_record_marker(xbufp);
2697		-
2698		- tailoff = (unsigned long)xbufp->tail[0].iov_base & ~PAGE_MASK;
2699		- headoff = (unsigned long)xbufp->head[0].iov_base & ~PAGE_MASK;
2700		- len = svc_send_common(sock, xbufp,
2701		- virt_to_page(xbufp->head[0].iov_base), headoff,
2702		- xbufp->tail[0].iov_base, tailoff);
2703		-
2704		- if (len != xbufp->len) {
2705		- printk(KERN_NOTICE "Error sending entire callback!\n");
2706		- len = -EAGAIN;
2707		- }
2708		-
2709		- return len;
	2566	+ req->rq_xtime = ktime_get();
	2567	+ err = xprt_sock_sendmsg(transport->sock, &msg, xdr, 0, marker, &sent);
	2568	+ xdr_free_bvec(xdr);
	2569	+ if (err < 0 \|\| sent != (xdr->len + sizeof(marker)))
	2570	+ return -EAGAIN;
	2571	+ return sent;
2710	2572	}
2711	2573
2712		-/*
2713		- * The send routine. Borrows from svc_send
	2574	+/**
	2575	+ * bc_send_request - Send a backchannel Call on a TCP socket
	2576	+ * @req: rpc_rqst containing Call message to be sent
	2577	+ *
	2578	+ * xpt_mutex ensures @rqstp's whole message is written to the socket
	2579	+ * without interruption.
	2580	+ *
	2581	+ * Return values:
	2582	+ * %0 if the message was sent successfully
	2583	+ * %ENOTCONN if the message was not sent
2714	2584	*/
2715		-static int bc_send_request(struct rpc_task *task)
	2585	+static int bc_send_request(struct rpc_rqst *req)
2716	2586	{
2717		- struct rpc_rqst *req = task->tk_rqstp;
2718	2587	struct svc_xprt *xprt;
2719	2588	int len;
2720	2589
2721		- dprintk("sending request with xid: %08x\n", ntohl(req->rq_xid));
2722	2590	/*
2723	2591	* Get the server socket associated with this callback xprt
2724	2592	*/
..	..	@@ -2728,12 +2596,7 @@
2728	2596	* Grab the mutex to serialize data as the connection is shared
2729	2597	* with the fore channel
2730	2598	*/
2731		- if (!mutex_trylock(&xprt->xpt_mutex)) {
2732		- rpc_sleep_on(&xprt->xpt_bc_pending, task, NULL);
2733		- if (!mutex_trylock(&xprt->xpt_mutex))
2734		- return -EAGAIN;
2735		- rpc_wake_up_queued_task(&xprt->xpt_bc_pending, task);
2736		- }
	2599	+ mutex_lock(&xprt->xpt_mutex);
2737	2600	if (test_bit(XPT_DEAD, &xprt->xpt_flags))
2738	2601	len = -ENOTCONN;
2739	2602	else
..	..	@@ -2752,6 +2615,7 @@
2752	2615
2753	2616	static void bc_close(struct rpc_xprt *xprt)
2754	2617	{
	2618	+ xprt_disconnect_done(xprt);
2755	2619	}
2756	2620
2757	2621	/*
..	..	@@ -2769,7 +2633,7 @@
2769	2633
2770	2634	static const struct rpc_xprt_ops xs_local_ops = {
2771	2635	.reserve_xprt = xprt_reserve_xprt,
2772		- .release_xprt = xs_tcp_release_xprt,
	2636	+ .release_xprt = xprt_release_xprt,
2773	2637	.alloc_slot = xprt_alloc_slot,
2774	2638	.free_slot = xprt_free_slot,
2775	2639	.rpcbind = xs_local_rpcbind,
..	..	@@ -2777,8 +2641,9 @@
2777	2641	.connect = xs_local_connect,
2778	2642	.buf_alloc = rpc_malloc,
2779	2643	.buf_free = rpc_free,
	2644	+ .prepare_request = xs_stream_prepare_request,
2780	2645	.send_request = xs_local_send_request,
2781		- .set_retrans_timeout = xprt_set_retrans_timeout_def,
	2646	+ .wait_for_reply_request = xprt_wait_for_reply_request_def,
2782	2647	.close = xs_close,
2783	2648	.destroy = xs_destroy,
2784	2649	.print_stats = xs_local_print_stats,
..	..	@@ -2798,7 +2663,7 @@
2798	2663	.buf_alloc = rpc_malloc,
2799	2664	.buf_free = rpc_free,
2800	2665	.send_request = xs_udp_send_request,
2801		- .set_retrans_timeout = xprt_set_retrans_timeout_rtt,
	2666	+ .wait_for_reply_request = xprt_wait_for_reply_request_rtt,
2802	2667	.timer = xs_udp_timer,
2803	2668	.release_request = xprt_release_rqst_cong,
2804	2669	.close = xs_close,
..	..	@@ -2811,16 +2676,17 @@
2811	2676
2812	2677	static const struct rpc_xprt_ops xs_tcp_ops = {
2813	2678	.reserve_xprt = xprt_reserve_xprt,
2814		- .release_xprt = xs_tcp_release_xprt,
2815		- .alloc_slot = xprt_lock_and_alloc_slot,
	2679	+ .release_xprt = xprt_release_xprt,
	2680	+ .alloc_slot = xprt_alloc_slot,
2816	2681	.free_slot = xprt_free_slot,
2817	2682	.rpcbind = rpcb_getport_async,
2818	2683	.set_port = xs_set_port,
2819	2684	.connect = xs_connect,
2820	2685	.buf_alloc = rpc_malloc,
2821	2686	.buf_free = rpc_free,
	2687	+ .prepare_request = xs_stream_prepare_request,
2822	2688	.send_request = xs_tcp_send_request,
2823		- .set_retrans_timeout = xprt_set_retrans_timeout_def,
	2689	+ .wait_for_reply_request = xprt_wait_for_reply_request_def,
2824	2690	.close = xs_tcp_shutdown,
2825	2691	.destroy = xs_destroy,
2826	2692	.set_connect_timeout = xs_tcp_set_connect_timeout,
..	..	@@ -2830,8 +2696,8 @@
2830	2696	.inject_disconnect = xs_inject_disconnect,
2831	2697	#ifdef CONFIG_SUNRPC_BACKCHANNEL
2832	2698	.bc_setup = xprt_setup_bc,
2833		- .bc_up = xs_tcp_bc_up,
2834	2699	.bc_maxpayload = xs_tcp_bc_maxpayload,
	2700	+ .bc_num_slots = xprt_bc_max_slots,
2835	2701	.bc_free_rqst = xprt_free_bc_rqst,
2836	2702	.bc_destroy = xprt_destroy_bc,
2837	2703	#endif
..	..	@@ -2849,7 +2715,7 @@
2849	2715	.buf_alloc = bc_malloc,
2850	2716	.buf_free = bc_free,
2851	2717	.send_request = bc_send_request,
2852		- .set_retrans_timeout = xprt_set_retrans_timeout_def,
	2718	+ .wait_for_reply_request = xprt_wait_for_reply_request_def,
2853	2719	.close = bc_close,
2854	2720	.destroy = bc_destroy,
2855	2721	.print_stats = xs_tcp_print_stats,
..	..	@@ -2950,7 +2816,6 @@
2950	2816	transport = container_of(xprt, struct sock_xprt, xprt);
2951	2817
2952	2818	xprt->prot = 0;
2953		- xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
2954	2819	xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
2955	2820
2956	2821	xprt->bind_timeout = XS_BIND_TO;
..	..	@@ -2960,9 +2825,9 @@
2960	2825	xprt->ops = &xs_local_ops;
2961	2826	xprt->timeout = &xs_local_default_timeout;
2962	2827
2963		- INIT_WORK(&transport->recv_worker, xs_local_data_receive_workfn);
2964		- INIT_DELAYED_WORK(&transport->connect_worker,
2965		- xs_dummy_setup_socket);
	2828	+ INIT_WORK(&transport->recv_worker, xs_stream_data_receive_workfn);
	2829	+ INIT_WORK(&transport->error_worker, xs_error_handle);
	2830	+ INIT_DELAYED_WORK(&transport->connect_worker, xs_dummy_setup_socket);
2966	2831
2967	2832	switch (sun->sun_family) {
2968	2833	case AF_LOCAL:
..	..	@@ -2974,9 +2839,6 @@
2974	2839	}
2975	2840	xprt_set_bound(xprt);
2976	2841	xs_format_peer_addresses(xprt, "local", RPCBIND_NETID_LOCAL);
2977		- ret = ERR_PTR(xs_local_setup_socket(transport));
2978		- if (ret)
2979		- goto out_err;
2980	2842	break;
2981	2843	default:
2982	2844	ret = ERR_PTR(-EAFNOSUPPORT);
..	..	@@ -3020,7 +2882,6 @@
3020	2882	transport = container_of(xprt, struct sock_xprt, xprt);
3021	2883
3022	2884	xprt->prot = IPPROTO_UDP;
3023		- xprt->tsh_size = 0;
3024	2885	/* XXX: header size can vary due to auth type, IPv6, etc. */
3025	2886	xprt->max_payload = (1U << 16) - (MAX_HEADER << 3);
3026	2887
..	..	@@ -3033,6 +2894,7 @@
3033	2894	xprt->timeout = &xs_udp_default_timeout;
3034	2895
3035	2896	INIT_WORK(&transport->recv_worker, xs_udp_data_receive_workfn);
	2897	+ INIT_WORK(&transport->error_worker, xs_error_handle);
3036	2898	INIT_DELAYED_WORK(&transport->connect_worker, xs_udp_setup_socket);
3037	2899
3038	2900	switch (addr->sa_family) {
..	..	@@ -3100,7 +2962,6 @@
3100	2962	transport = container_of(xprt, struct sock_xprt, xprt);
3101	2963
3102	2964	xprt->prot = IPPROTO_TCP;
3103		- xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
3104	2965	xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
3105	2966
3106	2967	xprt->bind_timeout = XS_BIND_TO;
..	..	@@ -3114,7 +2975,8 @@
3114	2975	xprt->connect_timeout = xprt->timeout->to_initval *
3115	2976	(xprt->timeout->to_retries + 1);
3116	2977
3117		- INIT_WORK(&transport->recv_worker, xs_tcp_data_receive_workfn);
	2978	+ INIT_WORK(&transport->recv_worker, xs_stream_data_receive_workfn);
	2979	+ INIT_WORK(&transport->error_worker, xs_error_handle);
3118	2980	INIT_DELAYED_WORK(&transport->connect_worker, xs_tcp_setup_socket);
3119	2981
3120	2982	switch (addr->sa_family) {
..	..	@@ -3173,7 +3035,6 @@
3173	3035	transport = container_of(xprt, struct sock_xprt, xprt);
3174	3036
3175	3037	xprt->prot = IPPROTO_TCP;
3176		- xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
3177	3038	xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
3178	3039	xprt->timeout = &xs_tcp_default_timeout;
3179	3040
..	..	@@ -3277,10 +3138,8 @@
3277	3138	*/
3278	3139	int init_socket_xprt(void)
3279	3140	{
3280		-#if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
3281	3141	if (!sunrpc_table_header)
3282	3142	sunrpc_table_header = register_sysctl_table(sunrpc_table);
3283		-#endif
3284	3143
3285	3144	xprt_register_transport(&xs_local_transport);
3286	3145	xprt_register_transport(&xs_udp_transport);
..	..	@@ -3296,12 +3155,10 @@
3296	3155	*/
3297	3156	void cleanup_socket_xprt(void)
3298	3157	{
3299		-#if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
3300	3158	if (sunrpc_table_header) {
3301	3159	unregister_sysctl_table(sunrpc_table_header);
3302	3160	sunrpc_table_header = NULL;
3303	3161	}
3304		-#endif
3305	3162
3306	3163	xprt_unregister_transport(&xs_local_transport);
3307	3164	xprt_unregister_transport(&xs_udp_transport);