Commit 3a2886cc authored by Gerd Rausch's avatar Gerd Rausch Committed by David S. Miller
Browse files

net/rds: Keep track of and wait for FRWR segments in use upon shutdown 



Since "rds_ib_free_frmr" and "rds_ib_free_frmr_list" simply put
the FRMR memory segments on the "drop_list" or "free_list",
and it is the job of "rds_ib_flush_mr_pool" to reap those entries
by ultimately issuing a "IB_WR_LOCAL_INV" work-request,
we need to trigger and then wait for all those memory segments
attached to a particular connection to be fully released before
we can move on to release the QP, CQ, etc.

So we make "rds_ib_conn_path_shutdown" wait for one more
atomic_t called "i_fastreg_inuse_count" that keeps track of how
many FRWR memory segments are out there marked "FRMR_IS_INUSE"
(and also wake_up rds_ib_ring_empty_wait, as they go away).

Signed-off-by: default avatarGerd Rausch <gerd.rausch@oracle.com>
Acked-by: default avatarSantosh Shilimkar <santosh.shilimkar@oracle.com>
Signed-off-by: default avatarDavid S. Miller <davem@davemloft.net>
parent 9547dff1

net/rds/ib.h

+1 −0
Original line number Diff line number Diff line
@@ -156,6 +156,7 @@ struct rds_ib_connection { 


	/* To control the number of wrs from fastreg */

	atomic_t		i_fastreg_wrs;

	atomic_t		i_fastreg_inuse_count;



	/* interrupt handling */

	struct tasklet_struct	i_send_tasklet;

net/rds/ib_cm.c

+7 −0
Original line number Diff line number Diff line
@@ -40,6 +40,7 @@ 
#include "rds_single_path.h"

#include "rds.h"

#include "ib.h"

#include "ib_mr.h"



/*

 * Set the selected protocol version
@@ -993,6 +994,11 @@ void rds_ib_conn_path_shutdown(struct rds_conn_path *cp) 
				ic->i_cm_id, err);

		}



		/* kick off "flush_worker" for all pools in order to reap

		 * all FRMR registrations that are still marked "FRMR_IS_INUSE"

		 */

		rds_ib_flush_mrs();



		/*

		 * We want to wait for tx and rx completion to finish

		 * before we tear down the connection, but we have to be
@@ -1005,6 +1011,7 @@ void rds_ib_conn_path_shutdown(struct rds_conn_path *cp) 
		wait_event(rds_ib_ring_empty_wait,

			   rds_ib_ring_empty(&ic->i_recv_ring) &&

			   (atomic_read(&ic->i_signaled_sends) == 0) &&

			   (atomic_read(&ic->i_fastreg_inuse_count) == 0) &&

			   (atomic_read(&ic->i_fastreg_wrs) == RDS_IB_DEFAULT_FR_WR));

		tasklet_kill(&ic->i_send_tasklet);

		tasklet_kill(&ic->i_recv_tasklet);

net/rds/ib_frmr.c

+37 −6
Original line number Diff line number Diff line
@@ -32,6 +32,24 @@ 


#include "ib_mr.h"



static inline void

rds_transition_frwr_state(struct rds_ib_mr *ibmr,

			  enum rds_ib_fr_state old_state,

			  enum rds_ib_fr_state new_state)

{

	if (cmpxchg(&ibmr->u.frmr.fr_state,

		    old_state, new_state) == old_state &&

	    old_state == FRMR_IS_INUSE) {

		/* enforce order of ibmr->u.frmr.fr_state update

		 * before decrementing i_fastreg_inuse_count

		 */

		smp_mb__before_atomic();

		atomic_dec(&ibmr->ic->i_fastreg_inuse_count);

		if (waitqueue_active(&rds_ib_ring_empty_wait))

			wake_up(&rds_ib_ring_empty_wait);

	}

}



static struct rds_ib_mr *rds_ib_alloc_frmr(struct rds_ib_device *rds_ibdev,

					   int npages)

{
@@ -118,13 +136,18 @@ static int rds_ib_post_reg_frmr(struct rds_ib_mr *ibmr) 
	if (unlikely(ret != ibmr->sg_len))

		return ret < 0 ? ret : -EINVAL;



	if (cmpxchg(&frmr->fr_state,

		    FRMR_IS_FREE, FRMR_IS_INUSE) != FRMR_IS_FREE)

		return -EBUSY;



	atomic_inc(&ibmr->ic->i_fastreg_inuse_count);



	/* Perform a WR for the fast_reg_mr. Each individual page

	 * in the sg list is added to the fast reg page list and placed

	 * inside the fast_reg_mr WR.  The key used is a rolling 8bit

	 * counter, which should guarantee uniqueness.

	 */

	ib_update_fast_reg_key(frmr->mr, ibmr->remap_count++);

	frmr->fr_state = FRMR_IS_INUSE;

	frmr->fr_reg = true;



	memset(&reg_wr, 0, sizeof(reg_wr));
@@ -141,7 +164,8 @@ static int rds_ib_post_reg_frmr(struct rds_ib_mr *ibmr) 
	ret = ib_post_send(ibmr->ic->i_cm_id->qp, &reg_wr.wr, NULL);

	if (unlikely(ret)) {

		/* Failure here can be because of -ENOMEM as well */

		frmr->fr_state = FRMR_IS_STALE;

		rds_transition_frwr_state(ibmr, FRMR_IS_INUSE, FRMR_IS_STALE);



		atomic_inc(&ibmr->ic->i_fastreg_wrs);

		if (printk_ratelimit())

			pr_warn("RDS/IB: %s returned error(%d)\n",
@@ -268,8 +292,12 @@ static int rds_ib_post_inv(struct rds_ib_mr *ibmr) 


	ret = ib_post_send(i_cm_id->qp, s_wr, NULL);

	if (unlikely(ret)) {

		frmr->fr_state = FRMR_IS_STALE;

		rds_transition_frwr_state(ibmr, FRMR_IS_INUSE, FRMR_IS_STALE);

		frmr->fr_inv = false;

		/* enforce order of frmr->fr_inv update

		 * before incrementing i_fastreg_wrs

		 */

		smp_mb__before_atomic();

		atomic_inc(&ibmr->ic->i_fastreg_wrs);

		pr_err("RDS/IB: %s returned error(%d)\n", __func__, ret);

		goto out;
@@ -297,7 +325,7 @@ void rds_ib_mr_cqe_handler(struct rds_ib_connection *ic, struct ib_wc *wc) 
	struct rds_ib_frmr *frmr = &ibmr->u.frmr;



	if (wc->status != IB_WC_SUCCESS) {

		frmr->fr_state = FRMR_IS_STALE;

		rds_transition_frwr_state(ibmr, FRMR_IS_INUSE, FRMR_IS_STALE);

		if (rds_conn_up(ic->conn))

			rds_ib_conn_error(ic->conn,

					  "frmr completion <%pI4,%pI4> status %u(%s), vendor_err 0x%x, disconnecting and reconnecting\n",
@@ -309,8 +337,7 @@ void rds_ib_mr_cqe_handler(struct rds_ib_connection *ic, struct ib_wc *wc) 
	}



	if (frmr->fr_inv) {

		if (frmr->fr_state == FRMR_IS_INUSE)

			frmr->fr_state = FRMR_IS_FREE;

		rds_transition_frwr_state(ibmr, FRMR_IS_INUSE, FRMR_IS_FREE);

		frmr->fr_inv = false;

		wake_up(&frmr->fr_inv_done);

	}
@@ -320,6 +347,10 @@ void rds_ib_mr_cqe_handler(struct rds_ib_connection *ic, struct ib_wc *wc) 
		wake_up(&frmr->fr_reg_done);

	}



	/* enforce order of frmr->{fr_reg,fr_inv} update

	 * before incrementing i_fastreg_wrs

	 */

	smp_mb__before_atomic();

	atomic_inc(&ic->i_fastreg_wrs);

}

CRA Git | Maintained and supported by SUSTech CRA and CCSE