rpcrdma_remove_one() can race with unregister and invoke a notification callback after its owner is freed. XArray RCU does not pin stored objects. Pin notification owners under rpcrdma_device_mutex and serialize unregister with the same lock. Also pin client RPC transports. Skip owners in final release with kref_get_unless_zero(). Invoke rn_done and release references outside the mutex. Track references separately so partial acquisition can be unwound without accessing freed owners or waiting for XPRT_LOCKED under the device mutex. Fixes: 7e86845a0346 ("rpcrdma: Implement generic device removal") Signed-off-by: Li Xiasong --- include/linux/sunrpc/rdma_rn.h | 29 ++++++++++++- net/sunrpc/xprtrdma/ib_client.c | 54 +++++++++++++++++++++--- net/sunrpc/xprtrdma/svc_rdma_transport.c | 21 ++++++++- net/sunrpc/xprtrdma/verbs.c | 34 ++++++++++++++- 4 files changed, 128 insertions(+), 10 deletions(-) diff --git a/include/linux/sunrpc/rdma_rn.h b/include/linux/sunrpc/rdma_rn.h index 7d032ca057af..1b3e0989b3a5 100644 --- a/include/linux/sunrpc/rdma_rn.h +++ b/include/linux/sunrpc/rdma_rn.h @@ -8,17 +8,42 @@ #include +struct rpc_xprt; + +/** + * struct rpcrdma_notification_ref - references acquired for a notification + * @rr_owner: pinned notification owner, or NULL if not acquired + * @rr_xprt: optional RPC transport reference, or NULL if not acquired + * + * The record is initialized to zero before rn_get(). Record all acquired + * references here, including on partial failure, for release by rn_put(). + */ +struct rpcrdma_notification_ref { + void *rr_owner; + struct rpc_xprt *rr_xprt; +}; + /** - * rpcrdma_notification - request removal notification + * struct rpcrdma_notification - request removal notification + * @rn_done: device removal callback + * @rn_get: acquire references as described in rpcrdma_rn_register() + * @rn_put: release references as described in rpcrdma_rn_register() + * @rn_index: index of this notification in the device's XArray */ struct rpcrdma_notification { void (*rn_done)(struct rpcrdma_notification *rn); + bool (*rn_get)(struct rpcrdma_notification *rn, + struct rpcrdma_notification_ref *ref); + void (*rn_put)(struct rpcrdma_notification_ref *ref); u32 rn_index; }; int rpcrdma_rn_register(struct ib_device *device, struct rpcrdma_notification *rn, - void (*done)(struct rpcrdma_notification *rn)); + void (*done)(struct rpcrdma_notification *rn), + bool (*get)(struct rpcrdma_notification *rn, + struct rpcrdma_notification_ref *ref), + void (*put)(struct rpcrdma_notification_ref *ref)); void rpcrdma_rn_unregister(struct ib_device *device, struct rpcrdma_notification *rn); int rpcrdma_ib_client_register(void); diff --git a/net/sunrpc/xprtrdma/ib_client.c b/net/sunrpc/xprtrdma/ib_client.c index 23ab1dfcb230..6fa72996209a 100644 --- a/net/sunrpc/xprtrdma/ib_client.c +++ b/net/sunrpc/xprtrdma/ib_client.c @@ -52,6 +52,18 @@ static struct rpcrdma_device *rpcrdma_get_client_data(struct ib_device *device) * @device: device to monitor * @rn: notification object that wishes to be notified * @done: callback to notify caller of device removal + * @get: callback to acquire a notification object reference + * @put: callback to release references acquired by @get + * + * @get runs under rpcrdma_device_mutex. On success, it must pin both + * the object embedding @rn and its registration until the returned + * reference is released. The owner must not call rpcrdma_rn_unregister() + * while that pin is held. @get must record all acquired references, + * including on partial failure, in &struct rpcrdma_notification_ref. + * + * @get returns true if @done can be invoked. @put releases any acquired + * references outside rpcrdma_device_mutex, even if @get returns false, + * and must not access an unpinned @rn. * * Returns zero on success. The callback in rn_done is guaranteed * to be invoked when the device is removed, unless this notification @@ -67,7 +79,10 @@ static struct rpcrdma_device *rpcrdma_get_client_data(struct ib_device *device) */ int rpcrdma_rn_register(struct ib_device *device, struct rpcrdma_notification *rn, - void (*done)(struct rpcrdma_notification *rn)) + void (*done)(struct rpcrdma_notification *rn), + bool (*get)(struct rpcrdma_notification *rn, + struct rpcrdma_notification_ref *ref), + void (*put)(struct rpcrdma_notification_ref *ref)) { struct rpcrdma_device *rd; int ret = -ENETUNREACH; @@ -87,8 +102,12 @@ int rpcrdma_rn_register(struct ib_device *device, * value would unregister an @rn that was never inserted. */ rn->rn_done = done; + rn->rn_get = get; + rn->rn_put = put; if (xa_alloc(&rd->rd_xa, &rn->rn_index, rn, xa_limit_32b, GFP_KERNEL) < 0) { rn->rn_done = NULL; + rn->rn_get = NULL; + rn->rn_put = NULL; ret = -ENOMEM; goto out_unlock; } @@ -117,14 +136,18 @@ static void rpcrdma_rn_release(struct kref *kref) * It is safe to call this on an @rn whose registration never * completed or failed; rn_done == NULL is treated as * never-registered and the call is a no-op. + * + * Context: May sleep. Takes rpcrdma_device_mutex. */ void rpcrdma_rn_unregister(struct ib_device *device, struct rpcrdma_notification *rn) { - struct rpcrdma_device *rd = rpcrdma_get_client_data(device); + struct rpcrdma_device *rd; + mutex_lock(&rpcrdma_device_mutex); + rd = rpcrdma_get_client_data(device); if (!rd) - return; + goto out_unlock; /* * rn_done is the registration sentinel: rpcrdma_rn_register @@ -138,12 +161,16 @@ void rpcrdma_rn_unregister(struct ib_device *device, * rd_kref. */ if (!rn->rn_done) - return; + goto out_unlock; rn->rn_done = NULL; + rn->rn_get = NULL; + rn->rn_put = NULL; trace_rpcrdma_client_unregister(device, rn); xa_erase(&rd->rd_xa, rn->rn_index); kref_put(&rd->rd_kref, rpcrdma_rn_release); +out_unlock: + mutex_unlock(&rpcrdma_device_mutex); } /** @@ -186,15 +213,30 @@ static void rpcrdma_remove_one(struct ib_device *device, { struct rpcrdma_device *rd = client_data; struct rpcrdma_notification *rn; + void (*done)(struct rpcrdma_notification *rn); + void (*put)(struct rpcrdma_notification_ref *ref); unsigned long index; trace_rpcrdma_client_remove_one(device); mutex_lock(&rpcrdma_device_mutex); set_bit(RPCRDMA_RD_F_REMOVING, &rd->rd_flags); + xa_for_each(&rd->rd_xa, index, rn) { + struct rpcrdma_notification_ref ref = {}; + bool notify; + + notify = rn->rn_get(rn, &ref); + if (!ref.rr_owner && !ref.rr_xprt) + continue; + done = notify ? rn->rn_done : NULL; + put = rn->rn_put; + mutex_unlock(&rpcrdma_device_mutex); + if (done) + done(rn); + put(&ref); + mutex_lock(&rpcrdma_device_mutex); + } mutex_unlock(&rpcrdma_device_mutex); - xa_for_each(&rd->rd_xa, index, rn) - rn->rn_done(rn); /* * Wait only if there are still outstanding notification diff --git a/net/sunrpc/xprtrdma/svc_rdma_transport.c b/net/sunrpc/xprtrdma/svc_rdma_transport.c index f949601b2144..ff3545570859 100644 --- a/net/sunrpc/xprtrdma/svc_rdma_transport.c +++ b/net/sunrpc/xprtrdma/svc_rdma_transport.c @@ -424,6 +424,24 @@ static void svc_rdma_xprt_done(struct rpcrdma_notification *rn) svc_xprt_close(&rdma->sc_xprt); } +static void svc_rdma_xprt_put(struct rpcrdma_notification_ref *ref) +{ + if (ref->rr_owner) + svc_xprt_put(ref->rr_owner); +} + +static bool svc_rdma_xprt_get(struct rpcrdma_notification *rn, + struct rpcrdma_notification_ref *ref) +{ + struct svcxprt_rdma *rdma = container_of(rn, struct svcxprt_rdma, + sc_rn); + + if (!kref_get_unless_zero(&rdma->sc_xprt.xpt_ref)) + return false; + ref->rr_owner = &rdma->sc_xprt; + return true; +} + /* * This is the xpo_recvfrom function for listening endpoints. Its * purpose is to accept incoming connections. The CMA callback handler @@ -464,7 +482,8 @@ static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt) dev = newxprt->sc_cm_id->device; newxprt->sc_port_num = newxprt->sc_cm_id->port_num; - if (rpcrdma_rn_register(dev, &newxprt->sc_rn, svc_rdma_xprt_done)) + if (rpcrdma_rn_register(dev, &newxprt->sc_rn, svc_rdma_xprt_done, + svc_rdma_xprt_get, svc_rdma_xprt_put)) goto errout; newxprt->sc_max_req_size = svcrdma_max_req_size; diff --git a/net/sunrpc/xprtrdma/verbs.c b/net/sunrpc/xprtrdma/verbs.c index 54957295a631..69d1796bb4f3 100644 --- a/net/sunrpc/xprtrdma/verbs.c +++ b/net/sunrpc/xprtrdma/verbs.c @@ -297,6 +297,35 @@ static void rpcrdma_ep_removal_done(struct rpcrdma_notification *rn) xprt_force_disconnect(ep->re_xprt); } +static void rpcrdma_ep_removal_put(struct rpcrdma_notification_ref *ref) +{ + struct rpcrdma_ep *ep = ref->rr_owner; + struct rpc_xprt *xprt = ref->rr_xprt; + + if (ep) { + struct rdma_cm_id *id = ep->re_id; + + if (rpcrdma_ep_put(ep)) + rdma_destroy_id(id); + } + xprt_put(xprt); +} + +static bool rpcrdma_ep_removal_get(struct rpcrdma_notification *rn, + struct rpcrdma_notification_ref *ref) +{ + struct rpcrdma_ep *ep = container_of(rn, struct rpcrdma_ep, re_rn); + struct rpc_xprt *xprt = ep->re_xprt; + + if (!xprt_get(xprt)) + return false; + ref->rr_xprt = xprt; + if (!kref_get_unless_zero(&ep->re_kref)) + return false; + ref->rr_owner = ep; + return true; +} + static struct rdma_cm_id *rpcrdma_create_id(struct rpcrdma_xprt *r_xprt, struct rpcrdma_ep *ep) { @@ -337,7 +366,10 @@ static struct rdma_cm_id *rpcrdma_create_id(struct rpcrdma_xprt *r_xprt, goto out; ep->re_id = id; - rc = rpcrdma_rn_register(id->device, &ep->re_rn, rpcrdma_ep_removal_done); + rc = rpcrdma_rn_register(id->device, &ep->re_rn, + rpcrdma_ep_removal_done, + rpcrdma_ep_removal_get, + rpcrdma_ep_removal_put); if (rc) goto out; -- 2.34.1