Minki/linux
1
0
Fork 1
mirror of https://github.com/torvalds/linux.git synced 2024-11-23 04:31:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

SUNRPC: RPC callbacks may be split across several TCP segments

Browse Source 
Since TCP is a stream protocol, our callback read code needs to take into
account the fact that RPC callbacks are not always confined to a single
TCP segment.
This patch adds support for multiple TCP segments by ensuring that we
only remove the rpc_rqst structure from the 'free backchannel requests'
list once the data has been completely received. We rely on the fact
that TCP data is ordered for the duration of the connection.

Reported-by: shaobingqing <shaobingqing@bwstor.com.cn>
Signed-off-by: Trond Myklebust <trond.myklebust@primarydata.com>
This commit is contained in:
Trond Myklebust 2014-02-10 11:18:39 -05:00
parent 628356791b
commit 2ea24497a1
3 changed files with 74 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
include/linux/sunrpc/bc_xprt.h
View File

@ -32,7 +32,8 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#include <linux/sunrpc/sched.h> #include <linux/sunrpc/sched.h>
#ifdef CONFIG_SUNRPC_BACKCHANNEL #ifdef CONFIG_SUNRPC_BACKCHANNEL
struct rpc_rqst *xprt_alloc_bc_request(struct rpc_xprt *xprt); struct rpc_rqst *xprt_lookup_bc_request(struct rpc_xprt *xprt, __be32 xid);
void xprt_complete_bc_request(struct rpc_rqst *req, uint32_t copied);
void xprt_free_bc_request(struct rpc_rqst *req); void xprt_free_bc_request(struct rpc_rqst *req);
int xprt_setup_backchannel(struct rpc_xprt *, unsigned int min_reqs); int xprt_setup_backchannel(struct rpc_xprt *, unsigned int min_reqs);
void xprt_destroy_backchannel(struct rpc_xprt *, unsigned int max_reqs); void xprt_destroy_backchannel(struct rpc_xprt *, unsigned int max_reqs);

93
net/sunrpc/backchannel_rqst.c
View File

@ -212,39 +212,23 @@ out:
} }
EXPORT_SYMBOL_GPL(xprt_destroy_backchannel); EXPORT_SYMBOL_GPL(xprt_destroy_backchannel);
/* static struct rpc_rqst *xprt_alloc_bc_request(struct rpc_xprt *xprt, __be32 xid)
* One or more rpc_rqst structure have been preallocated during the
* backchannel setup. Buffer space for the send and private XDR buffers
* has been preallocated as well. Use xprt_alloc_bc_request to allocate
* to this request. Use xprt_free_bc_request to return it.
*
* We know that we're called in soft interrupt context, grab the spin_lock
* since there is no need to grab the bottom half spin_lock.
*
* Return an available rpc_rqst, otherwise NULL if non are available.
*/
struct rpc_rqst *xprt_alloc_bc_request(struct rpc_xprt *xprt)
{ {
struct rpc_rqst *req; struct rpc_rqst *req = NULL;
dprintk("RPC: allocate a backchannel request\n"); dprintk("RPC: allocate a backchannel request\n");
spin_lock(&xprt->bc_pa_lock); if (list_empty(&xprt->bc_pa_list))
if (!list_empty(&xprt->bc_pa_list)) { goto not_found;
req = list_first_entry(&xprt->bc_pa_list, struct rpc_rqst,
rq_bc_pa_list);
list_del(&req->rq_bc_pa_list);
} else {
req = NULL;
}
spin_unlock(&xprt->bc_pa_lock);
if (req != NULL) { req = list_first_entry(&xprt->bc_pa_list, struct rpc_rqst,
set_bit(RPC_BC_PA_IN_USE, &req->rq_bc_pa_state); rq_bc_pa_list);
req->rq_reply_bytes_recvd = 0; req->rq_reply_bytes_recvd = 0;
req->rq_bytes_sent = 0; req->rq_bytes_sent = 0;
memcpy(&req->rq_private_buf, &req->rq_rcv_buf, memcpy(&req->rq_private_buf, &req->rq_rcv_buf,
sizeof(req->rq_private_buf)); sizeof(req->rq_private_buf));
} req->rq_xid = xid;
req->rq_connect_cookie = xprt->connect_cookie;
not_found:
dprintk("RPC: backchannel req=%p\n", req); dprintk("RPC: backchannel req=%p\n", req);
return req; return req;
} }
@ -259,6 +243,7 @@ void xprt_free_bc_request(struct rpc_rqst *req)
dprintk("RPC: free backchannel req=%p\n", req); dprintk("RPC: free backchannel req=%p\n", req);
req->rq_connect_cookie = xprt->connect_cookie - 1;
smp_mb__before_clear_bit(); smp_mb__before_clear_bit();
WARN_ON_ONCE(!test_bit(RPC_BC_PA_IN_USE, &req->rq_bc_pa_state)); WARN_ON_ONCE(!test_bit(RPC_BC_PA_IN_USE, &req->rq_bc_pa_state));
clear_bit(RPC_BC_PA_IN_USE, &req->rq_bc_pa_state); clear_bit(RPC_BC_PA_IN_USE, &req->rq_bc_pa_state);
@ -281,7 +266,57 @@ void xprt_free_bc_request(struct rpc_rqst *req)
* may be reused by a new callback request. * may be reused by a new callback request.
*/ */
spin_lock_bh(&xprt->bc_pa_lock); spin_lock_bh(&xprt->bc_pa_lock);
list_add(&req->rq_bc_pa_list, &xprt->bc_pa_list); list_add_tail(&req->rq_bc_pa_list, &xprt->bc_pa_list);
spin_unlock_bh(&xprt->bc_pa_lock); spin_unlock_bh(&xprt->bc_pa_lock);
} }
/*
* One or more rpc_rqst structure have been preallocated during the
* backchannel setup. Buffer space for the send and private XDR buffers
* has been preallocated as well. Use xprt_alloc_bc_request to allocate
* to this request. Use xprt_free_bc_request to return it.
*
* We know that we're called in soft interrupt context, grab the spin_lock
* since there is no need to grab the bottom half spin_lock.
*
* Return an available rpc_rqst, otherwise NULL if non are available.
*/
struct rpc_rqst *xprt_lookup_bc_request(struct rpc_xprt *xprt, __be32 xid)
{
struct rpc_rqst *req;
spin_lock(&xprt->bc_pa_lock);
list_for_each_entry(req, &xprt->bc_pa_list, rq_bc_pa_list) {
if (req->rq_connect_cookie != xprt->connect_cookie)
continue;
if (req->rq_xid == xid)
goto found;
}
req = xprt_alloc_bc_request(xprt, xid);
found:
spin_unlock(&xprt->bc_pa_lock);
return req;
}
/*
* Add callback request to callback list. The callback
* service sleeps on the sv_cb_waitq waiting for new
* requests. Wake it up after adding enqueing the
* request.
*/
void xprt_complete_bc_request(struct rpc_rqst *req, uint32_t copied)
{
struct rpc_xprt *xprt = req->rq_xprt;
struct svc_serv *bc_serv = xprt->bc_serv;
req->rq_private_buf.len = copied;
set_bit(RPC_BC_PA_IN_USE, &req->rq_bc_pa_state);
dprintk("RPC: add callback request to list\n");
spin_lock(&bc_serv->sv_cb_lock);
list_del(&req->rq_bc_pa_list);
list_add(&req->rq_bc_list, &bc_serv->sv_cb_list);
wake_up(&bc_serv->sv_cb_waitq);
spin_unlock(&bc_serv->sv_cb_lock);
}

28
net/sunrpc/xprtsock.c
View File

@ -1306,41 +1306,29 @@ static inline int xs_tcp_read_reply(struct rpc_xprt *xprt,
* If we're unable to obtain the rpc_rqst we schedule the closing of the * If we're unable to obtain the rpc_rqst we schedule the closing of the
* connection and return -1. * connection and return -1.
*/ */
static inline int xs_tcp_read_callback(struct rpc_xprt *xprt, static int xs_tcp_read_callback(struct rpc_xprt *xprt,
struct xdr_skb_reader *desc) struct xdr_skb_reader *desc)
{ {
struct sock_xprt *transport = struct sock_xprt *transport =
container_of(xprt, struct sock_xprt, xprt); container_of(xprt, struct sock_xprt, xprt);
struct rpc_rqst *req; struct rpc_rqst *req;
req = xprt_alloc_bc_request(xprt); /* Look up and lock the request corresponding to the given XID */
spin_lock(&xprt->transport_lock);
req = xprt_lookup_bc_request(xprt, transport->tcp_xid);
if (req == NULL) { if (req == NULL) {
spin_unlock(&xprt->transport_lock);
printk(KERN_WARNING "Callback slot table overflowed\n"); printk(KERN_WARNING "Callback slot table overflowed\n");
xprt_force_disconnect(xprt); xprt_force_disconnect(xprt);
return -1; return -1;
} }
req->rq_xid = transport->tcp_xid;
dprintk("RPC: read callback XID %08x\n", ntohl(req->rq_xid)); dprintk("RPC: read callback XID %08x\n", ntohl(req->rq_xid));
xs_tcp_read_common(xprt, desc, req); xs_tcp_read_common(xprt, desc, req);
if (!(transport->tcp_flags & TCP_RCV_COPY_DATA)) { if (!(transport->tcp_flags & TCP_RCV_COPY_DATA))
struct svc_serv *bc_serv = xprt->bc_serv; xprt_complete_bc_request(req, transport->tcp_copied);
spin_unlock(&xprt->transport_lock);
/*
* Add callback request to callback list. The callback
* service sleeps on the sv_cb_waitq waiting for new
* requests. Wake it up after adding enqueing the
* request.
*/
dprintk("RPC: add callback request to list\n");
spin_lock(&bc_serv->sv_cb_lock);
list_add(&req->rq_bc_list, &bc_serv->sv_cb_list);
spin_unlock(&bc_serv->sv_cb_lock);
wake_up(&bc_serv->sv_cb_waitq);
}
req->rq_private_buf.len = transport->tcp_copied;
return 0; return 0;
} }