[PATCH] RPC: expose API for serializing access to RPC transports

 The next several patches introduce an API that allows transports to
 choose whether the RPC client provides congestion control or whether
 the transport itself provides it.

 The first method we abstract is the one that serializes access to the
 RPC transport to prevent the bytes from different requests from mingling
 together.  This method provides proper request serialization and the
 opportunity to prevent new requests from being started because the
 transport is congested.

 The normal situation is for the transport to handle congestion control
 itself.  Although NFS over UDP was first, it has been recognized after
 years of experience that having the transport provide congestion control
 is much better than doing it in the RPC client.  Thus TCP, and probably
 every future transport implementation, will use the default method,
 xprt_lock_write, provided in xprt.c, which does not provide any kind
 of congestion control.  UDP can continue using the xprt.c-provided
 Van Jacobson congestion avoidance implementation.

 Test-plan:
 Use WAN simulation to cause sporadic bursty packet loss.  Look for significant
 regression in performance or client stability.

 Signed-off-by: Chuck Lever <cel@netapp.com>
 Signed-off-by: Trond Myklebust <Trond.Myklebust@netapp.com>

diff --git a/include/linux/sunrpc/xprt.h b/include/linux/sunrpc/xprt.h
index ac08e99a81cbd1f1fb8f7fc383ce327f4cde98b5..eee1c68778518c3861d68ab50f6b4608287026de 100644 (file)
--- a/include/linux/sunrpc/xprt.h
+++ b/include/linux/sunrpc/xprt.h
@@ -132,6 +132,7 @@ struct rpc_xprt;
 
 struct rpc_xprt_ops {
        void            (*set_buffer_size)(struct rpc_xprt *xprt);
+       int             (*reserve_xprt)(struct rpc_task *task);
        void            (*connect)(struct rpc_task *task);
        int             (*send_request)(struct rpc_task *task);
        void            (*set_retrans_timeout)(struct rpc_task *task);
@@ -232,6 +233,8 @@ void                        xprt_set_timeout(struct rpc_timeout *to, unsigned int retr, unsigned long
  */
 void                   xprt_connect(struct rpc_task *task);
 void                   xprt_reserve(struct rpc_task *task);
+int                    xprt_reserve_xprt(struct rpc_task *task);
+int                    xprt_reserve_xprt_cong(struct rpc_task *task);
 int                    xprt_prepare_transmit(struct rpc_task *task);
 void                   xprt_transmit(struct rpc_task *task);
 int                    xprt_adjust_timeout(struct rpc_rqst *req);

diff --git a/net/sunrpc/xprt.c b/net/sunrpc/xprt.c
index 1ac2fbe05102a59e1e7f077ef0f203a3c0c943e3..2d1e8b83dd685203bd51d7944db51df45fcb25b6 100644 (file)
--- a/net/sunrpc/xprt.c
+++ b/net/sunrpc/xprt.c
@@ -64,14 +64,56 @@ static int      __xprt_get_cong(struct rpc_xprt *, struct rpc_task *);
 
 static int     xprt_clear_backlog(struct rpc_xprt *xprt);
 
+/**
+ * xprt_reserve_xprt - serialize write access to transports
+ * @task: task that is requesting access to the transport
+ *
+ * This prevents mixing the payload of separate requests, and prevents
+ * transport connects from colliding with writes.  No congestion control
+ * is provided.
+ */
+int xprt_reserve_xprt(struct rpc_task *task)
+{
+       struct rpc_xprt *xprt = task->tk_xprt;
+       struct rpc_rqst *req = task->tk_rqstp;
+
+       if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) {
+               if (task == xprt->snd_task)
+                       return 1;
+               if (task == NULL)
+                       return 0;
+               goto out_sleep;
+       }
+       xprt->snd_task = task;
+       if (req) {
+               req->rq_bytes_sent = 0;
+               req->rq_ntrans++;
+       }
+       return 1;
+
+out_sleep:
+       dprintk("RPC: %4d failed to lock transport %p\n",
+                       task->tk_pid, xprt);
+       task->tk_timeout = 0;
+       task->tk_status = -EAGAIN;
+       if (req && req->rq_ntrans)
+               rpc_sleep_on(&xprt->resend, task, NULL, NULL);
+       else
+               rpc_sleep_on(&xprt->sending, task, NULL, NULL);
+       return 0;
+}
+
 /*
- * Serialize write access to transports, in order to prevent different
- * requests from interfering with each other.
- * Also prevents transport connects from colliding with writes.
+ * xprt_reserve_xprt_cong - serialize write access to transports
+ * @task: task that is requesting access to the transport
+ *
+ * Same as xprt_reserve_xprt, but Van Jacobson congestion control is
+ * integrated into the decision of whether a request is allowed to be
+ * woken up and given access to the transport.
  */
-static int
-__xprt_lock_write(struct rpc_xprt *xprt, struct rpc_task *task)
+int xprt_reserve_xprt_cong(struct rpc_task *task)
 {
+       struct rpc_xprt *xprt = task->tk_xprt;
        struct rpc_rqst *req = task->tk_rqstp;
 
        if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) {
@@ -79,7 +121,7 @@ __xprt_lock_write(struct rpc_xprt *xprt, struct rpc_task *task)
                        return 1;
                goto out_sleep;
        }
-       if (xprt->nocong || __xprt_get_cong(xprt, task)) {
+       if (__xprt_get_cong(xprt, task)) {
                xprt->snd_task = task;
                if (req) {
                        req->rq_bytes_sent = 0;
@@ -101,20 +143,18 @@ out_sleep:
        return 0;
 }
 
-static inline int
-xprt_lock_write(struct rpc_xprt *xprt, struct rpc_task *task)
+static inline int xprt_lock_write(struct rpc_xprt *xprt, struct rpc_task *task)
 {
        int retval;
 
        spin_lock_bh(&xprt->transport_lock);
-       retval = __xprt_lock_write(xprt, task);
+       retval = xprt->ops->reserve_xprt(task);
        spin_unlock_bh(&xprt->transport_lock);
        return retval;
 }
 
 
-static void
-__xprt_lock_write_next(struct rpc_xprt *xprt)
+static void __xprt_lock_write_next(struct rpc_xprt *xprt)
 {
        struct rpc_task *task;
 
@@ -598,7 +638,7 @@ int xprt_prepare_transmit(struct rpc_task *task)
                err = req->rq_received;
                goto out_unlock;
        }
-       if (!__xprt_lock_write(xprt, task)) {
+       if (!xprt->ops->reserve_xprt(task)) {
                err = -EAGAIN;
                goto out_unlock;
        }

diff --git a/net/sunrpc/xprtsock.c b/net/sunrpc/xprtsock.c
index 79433ffd1df0e3bf688fb07ccee6789b06ec0e4a..fc4fbe8ea346d2554cbf5bd87b079ddd84d3deed 100644 (file)
--- a/net/sunrpc/xprtsock.c
+++ b/net/sunrpc/xprtsock.c
@@ -1045,6 +1045,7 @@ static void xs_connect(struct rpc_task *task)
 
 static struct rpc_xprt_ops xs_udp_ops = {
        .set_buffer_size        = xs_udp_set_buffer_size,
+       .reserve_xprt           = xprt_reserve_xprt_cong,
        .connect                = xs_connect,
        .send_request           = xs_udp_send_request,
        .set_retrans_timeout    = xprt_set_retrans_timeout_rtt,
@@ -1054,6 +1055,7 @@ static struct rpc_xprt_ops xs_udp_ops = {
 
 static struct rpc_xprt_ops xs_tcp_ops = {
        .set_buffer_size        = xs_tcp_set_buffer_size,
+       .reserve_xprt           = xprt_reserve_xprt,
        .connect                = xs_connect,
        .send_request           = xs_tcp_send_request,
        .set_retrans_timeout    = xprt_set_retrans_timeout_def,