mptcp: dedicated request sock for subflow in v6

[platform/kernel/linux-rpi.git] / net / sunrpc / xprtsock.c

diff --git a/net/sunrpc/xprtsock.c b/net/sunrpc/xprtsock.c
index 04f1b78..897dfce 100644 (file)
--- a/net/sunrpc/xprtsock.c
+++ b/net/sunrpc/xprtsock.c
@@ -763,12 +763,12 @@ xs_stream_start_connect(struct sock_xprt *transport)
 /**
  * xs_nospace - handle transmit was incomplete
  * @req: pointer to RPC request
+ * @transport: pointer to struct sock_xprt
  *
  */
-static int xs_nospace(struct rpc_rqst *req)
+static int xs_nospace(struct rpc_rqst *req, struct sock_xprt *transport)
 {
-       struct rpc_xprt *xprt = req->rq_xprt;
-       struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
+       struct rpc_xprt *xprt = &transport->xprt;
        struct sock *sk = transport->inet;
        int ret = -EAGAIN;
 
@@ -779,25 +779,49 @@ static int xs_nospace(struct rpc_rqst *req)
 
        /* Don't race with disconnect */
        if (xprt_connected(xprt)) {
+               struct socket_wq *wq;
+
+               rcu_read_lock();
+               wq = rcu_dereference(sk->sk_wq);
+               set_bit(SOCKWQ_ASYNC_NOSPACE, &wq->flags);
+               rcu_read_unlock();
+
                /* wait for more buffer space */
+               set_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
                sk->sk_write_pending++;
                xprt_wait_for_buffer_space(xprt);
        } else
                ret = -ENOTCONN;
 
        spin_unlock(&xprt->transport_lock);
+       return ret;
+}
 
-       /* Race breaker in case memory is freed before above code is called */
-       if (ret == -EAGAIN) {
-               struct socket_wq *wq;
+static int xs_sock_nospace(struct rpc_rqst *req)
+{
+       struct sock_xprt *transport =
+               container_of(req->rq_xprt, struct sock_xprt, xprt);
+       struct sock *sk = transport->inet;
+       int ret = -EAGAIN;
 
-               rcu_read_lock();
-               wq = rcu_dereference(sk->sk_wq);
-               set_bit(SOCKWQ_ASYNC_NOSPACE, &wq->flags);
-               rcu_read_unlock();
+       lock_sock(sk);
+       if (!sock_writeable(sk))
+               ret = xs_nospace(req, transport);
+       release_sock(sk);
+       return ret;
+}
 
-               sk->sk_write_space(sk);
-       }
+static int xs_stream_nospace(struct rpc_rqst *req)
+{
+       struct sock_xprt *transport =
+               container_of(req->rq_xprt, struct sock_xprt, xprt);
+       struct sock *sk = transport->inet;
+       int ret = -EAGAIN;
+
+       lock_sock(sk);
+       if (!sk_stream_memory_free(sk))
+               ret = xs_nospace(req, transport);
+       release_sock(sk);
        return ret;
 }
 
@@ -856,7 +880,7 @@ static int xs_local_send_request(struct rpc_rqst *req)
 
        /* Close the stream if the previous transmission was incomplete */
        if (xs_send_request_was_aborted(transport, req)) {
-               xs_close(xprt);
+               xprt_force_disconnect(xprt);
                return -ENOTCONN;
        }
 
@@ -887,14 +911,14 @@ static int xs_local_send_request(struct rpc_rqst *req)
        case -ENOBUFS:
                break;
        case -EAGAIN:
-               status = xs_nospace(req);
+               status = xs_stream_nospace(req);
                break;
        default:
                dprintk("RPC:       sendmsg returned unrecognized error %d\n",
                        -status);
                fallthrough;
        case -EPIPE:
-               xs_close(xprt);
+               xprt_force_disconnect(xprt);
                status = -ENOTCONN;
        }
 
@@ -963,7 +987,7 @@ process_status:
                /* Should we call xs_close() here? */
                break;
        case -EAGAIN:
-               status = xs_nospace(req);
+               status = xs_sock_nospace(req);
                break;
        case -ENETUNREACH:
        case -ENOBUFS:
@@ -1083,7 +1107,7 @@ static int xs_tcp_send_request(struct rpc_rqst *req)
                /* Should we call xs_close() here? */
                break;
        case -EAGAIN:
-               status = xs_nospace(req);
+               status = xs_stream_nospace(req);
                break;
        case -ECONNRESET:
        case -ECONNREFUSED:
@@ -1181,6 +1205,16 @@ static void xs_reset_transport(struct sock_xprt *transport)
 
        if (sk == NULL)
                return;
+       /*
+        * Make sure we're calling this in a context from which it is safe
+        * to call __fput_sync(). In practice that means rpciod and the
+        * system workqueue.
+        */
+       if (!(current->flags & PF_WQ_WORKER)) {
+               WARN_ON_ONCE(1);
+               set_bit(XPRT_CLOSE_WAIT, &xprt->state);
+               return;
+       }
 
        if (atomic_read(&transport->xprt.swapper))
                sk_clear_memalloc(sk);
@@ -1204,7 +1238,7 @@ static void xs_reset_transport(struct sock_xprt *transport)
        mutex_unlock(&transport->recv_mutex);
 
        trace_rpc_socket_close(xprt, sock);
-       fput(filp);
+       __fput_sync(filp);
 
        xprt_disconnect_done(xprt);
 }
@@ -2233,6 +2267,7 @@ static int xs_tcp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
                fallthrough;
        case -EINPROGRESS:
                /* SYN_SENT! */
+               set_bit(XPRT_SOCK_CONNECT_SENT, &transport->sock_state);
                if (xprt->reestablish_timeout < XS_TCP_INIT_REEST_TO)
                        xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
                break;
@@ -2258,10 +2293,14 @@ static void xs_tcp_setup_socket(struct work_struct *work)
        struct rpc_xprt *xprt = &transport->xprt;
        int status = -EIO;
 
-       if (!sock) {
-               sock = xs_create_sock(xprt, transport,
-                               xs_addr(xprt)->sa_family, SOCK_STREAM,
-                               IPPROTO_TCP, true);
+       if (xprt_connected(xprt))
+               goto out;
+       if (test_and_clear_bit(XPRT_SOCK_CONNECT_SENT,
+                              &transport->sock_state) ||
+           !sock) {
+               xs_reset_transport(transport);
+               sock = xs_create_sock(xprt, transport, xs_addr(xprt)->sa_family,
+                                     SOCK_STREAM, IPPROTO_TCP, true);
                if (IS_ERR(sock)) {
                        status = PTR_ERR(sock);
                        goto out;
@@ -2343,11 +2382,7 @@ static void xs_connect(struct rpc_xprt *xprt, struct rpc_task *task)
 
        if (transport->sock != NULL) {
                dprintk("RPC:       xs_connect delayed xprt %p for %lu "
-                               "seconds\n",
-                               xprt, xprt->reestablish_timeout / HZ);
-
-               /* Start by resetting any existing state */
-               xs_reset_transport(transport);
+                       "seconds\n", xprt, xprt->reestablish_timeout / HZ);
 
                delay = xprt_reconnect_delay(xprt);
                xprt_reconnect_backoff(xprt, XS_TCP_INIT_REEST_TO);
@@ -2823,9 +2858,6 @@ static struct rpc_xprt *xs_setup_local(struct xprt_create *args)
                }
                xprt_set_bound(xprt);
                xs_format_peer_addresses(xprt, "local", RPCBIND_NETID_LOCAL);
-               ret = ERR_PTR(xs_local_setup_socket(transport));
-               if (ret)
-                       goto out_err;
                break;
        default:
                ret = ERR_PTR(-EAFNOSUPPORT);