diff --git a/net/rds/connection.c b/net/rds/connection.c index 46ac72088f84..fbbac55a0e81 100644 --- a/net/rds/connection.c +++ b/net/rds/connection.c @@ -106,10 +106,12 @@ static struct rds_connection *rds_conn_lookup(struct net *net, } /* - * This is called by transports as they're bringing down a connection. - * It clears partial message state so that the transport can start sending - * and receiving over this connection again in the future. It is up to - * the transport to have serialized this call with its send and recv. + * This is called by rds_conn_shutdown() once the transport has brought + * a path down. It clears partial message state so that the transport + * can start sending and receiving over this path again in the future. + * The caller owns RDS_IN_XMIT and RDS_RECV_REFILL across this call, + * which is what serializes it against the send and receive-refill + * paths. */ static void rds_conn_path_reset(struct rds_conn_path *cp) { @@ -124,8 +126,9 @@ static void rds_conn_path_reset(struct rds_conn_path *cp) /* Clear the bits the reset is responsible for individually: a * blanket cp_flags = 0 is a plain store that can clobber a * concurrent atomic read-modify-write on the same word. - * RDS_IN_XMIT and RDS_RECV_REFILL belong to the caller, - * rds_conn_shutdown(), and are left alone here. + * RDS_IN_XMIT and RDS_RECV_REFILL are held as locks by the + * caller, rds_conn_shutdown(), which releases them once the + * teardown is complete. */ clear_bit(RDS_LL_SEND_FULL, &cp->cp_flags); clear_bit(RDS_RECONNECT_PENDING, &cp->cp_flags); @@ -414,14 +417,35 @@ void rds_conn_shutdown(struct rds_conn_path *cp) } mutex_unlock(&cp->cp_cm_lock); + /* Quiesce the transmit and receive-refill paths by + * acquiring their bit locks, not merely waiting for + * them to be released: with a plain wait, either path + * can re-take its lock the instant after we sample it + * clear and then run concurrently with the transport + * shutdown and the path reset below. Holding both + * locks across the teardown makes that structurally + * impossible. + */ wait_event(cp->cp_waitq, - !test_bit(RDS_IN_XMIT, &cp->cp_flags)); + !test_and_set_bit_lock(RDS_IN_XMIT, &cp->cp_flags)); wait_event(cp->cp_waitq, - !test_bit(RDS_RECV_REFILL, &cp->cp_flags)); + !test_and_set_bit(RDS_RECV_REFILL, &cp->cp_flags)); conn->c_trans->conn_path_shutdown(cp); rds_conn_path_reset(cp); + /* Release the two locks and wake any waiter (e.g. + * rds_tcp_reset_callbacks()) that blocked on them while + * we held them. The unlock orders the transport's ring + * re-initialization and the path reset above before + * either bit is seen clear. rds_conn_path_reset() leaves + * both bits alone: ownership ends here, not inside the + * reset. + */ + clear_bit_unlock(RDS_IN_XMIT, &cp->cp_flags); + clear_bit_unlock(RDS_RECV_REFILL, &cp->cp_flags); + wake_up_all(&cp->cp_waitq); + if (!rds_conn_path_transition(cp, RDS_CONN_DISCONNECTING, RDS_CONN_DOWN) && !rds_conn_path_transition(cp, RDS_CONN_ERROR, diff --git a/net/rds/ib_recv.c b/net/rds/ib_recv.c index a6983861eec7..bd6cb3ffaa57 100644 --- a/net/rds/ib_recv.c +++ b/net/rds/ib_recv.c @@ -391,7 +391,9 @@ void rds_ib_recv_refill(struct rds_connection *conn, int prefill, gfp_t gfp) /* the goal here is to just make sure that someone, somewhere * is posting buffers. If we can't get the refill lock, - * let them do their thing + * let them do their thing. The holder may also be + * rds_conn_shutdown() tearing the path down, in which case + * there is nothing to post. */ if (!acquire_refill(conn)) return; diff --git a/net/rds/send.c b/net/rds/send.c index 8aad185e4b1a..1afa981e5c06 100644 --- a/net/rds/send.c +++ b/net/rds/send.c @@ -244,8 +244,11 @@ int rds_send_xmit(struct rds_conn_path *cp) WRITE_ONCE(cp->cp_send_gen, send_gen); /* - * rds_conn_shutdown() sets the conn state and then tests RDS_IN_XMIT, - * we do the opposite to avoid races. + * rds_conn_shutdown() sets the conn state and then acquires + * RDS_IN_XMIT; we take the lock first and then check the state. + * Ownership is decided by the atomic RMW on the cp_flags word: + * if the teardown won the bit we back off here, and if we won + * it the teardown waits until we release it. */ if (!rds_conn_path_up(cp)) { release_in_xmit(cp); diff --git a/net/rds/tcp.c b/net/rds/tcp.c index f4c83e368390..69c6d3145b5a 100644 --- a/net/rds/tcp.c +++ b/net/rds/tcp.c @@ -144,8 +144,10 @@ void rds_tcp_reset_callbacks(struct socket *sock, * so we must quiesce any send threads before resetting * cp_transport_data. Setting cp_state to something other * than RDS_CONN_UP stops new senders, and owning RDS_IN_XMIT - * excludes any thread already inside rds_send_xmit() for the - * whole socket swap and the rds_send_path_reset() below. + * excludes any thread already inside rds_send_xmit() - or a + * teardown in rds_conn_shutdown(), which holds the same lock + * for the duration of the transport shutdown - for the whole + * socket swap and the rds_send_path_reset() below. * * An incoming syn-ack at this point would end up marking the * conn as RDS_CONN_UP, and would again permit rds_send_xmit() @@ -178,13 +180,22 @@ void rds_tcp_reset_callbacks(struct socket *sock, /* Read t_sock only while owning RDS_IN_XMIT, never before the * wait: the teardown in rds_conn_shutdown() releases the old * socket and clears t_sock, so a pointer sampled earlier can - * be stale by the time we wake up. + * be stale by the time we wake up. The teardown holds the + * same lock while it does so, so what we read here cannot + * change under us until we release it. */ osock = tc->t_sock; if (!osock) goto newsock; - /* reset receive side state for rds_tcp_data_recv() for osock */ + /* reset receive side state for rds_tcp_data_recv() for osock. + * + * The sync cancels while owning RDS_IN_XMIT rely on cp_wq + * being ordered: a teardown blocked on the bit occupies + * cp_wq's only execution slot, so cp_send_w and cp_recv_w are + * pending at most and the cancels never flush. Nothing here + * may flush or wait on cp_wq itself. + */ cancel_delayed_work_sync(&cp->cp_send_w); cancel_delayed_work_sync(&cp->cp_recv_w); lock_sock(osock->sk);