rxrpc: Only disconnect calls in the I/O thread

Only perform call disconnection in the I/O thread to reduce the locking
requirement.

This is the first part of a fix for a race that exists between call
connection and call disconnection whereby the data transmission code adds
the call to the peer error distribution list after the call has been
disconnected (say by the rxrpc socket getting closed).

The fix is to complete the process of moving call connection, data
transmission and call disconnection into the I/O thread and thus forcibly
serialising them.

Note that the issue may predate the overhaul to an I/O thread model that
were included in the merge window for v6.2, but the timing is very much
changed by the change given below.

Fixes: cf37b5987508 ("rxrpc: Move DATA transmission into call processor work item")
Reported-by: syzbot+c22650d2844392afdcfd@syzkaller.appspotmail.com
Signed-off-by: David Howells <dhowells@redhat.com>
cc: Marc Dionne <marc.dionne@auristor.com>
cc: linux-afs@lists.infradead.org

diff --git a/include/trace/events/rxrpc.h b/include/trace/events/rxrpc.h
index c44cc01de7501b5106373ea62aef3314a5f9624a..eac513668e3338385857c47cd3acff03f3d19ea5 100644 (file)
--- a/include/trace/events/rxrpc.h
+++ b/include/trace/events/rxrpc.h
@@ -18,6 +18,7 @@
  */
 #define rxrpc_call_poke_traces \
        EM(rxrpc_call_poke_abort,               "Abort")        \
+       EM(rxrpc_call_poke_complete,            "Compl")        \
        EM(rxrpc_call_poke_error,               "Error")        \
        EM(rxrpc_call_poke_idle,                "Idle")         \
        EM(rxrpc_call_poke_start,               "Start")        \

diff --git a/net/rxrpc/call_event.c b/net/rxrpc/call_event.c
index b7efecf5ccfc01e9ac5f3a1180ae425408dc792b..b2fc3fa686ece31074d02188860420a508d5cf17 100644 (file)
--- a/net/rxrpc/call_event.c
+++ b/net/rxrpc/call_event.c
@@ -484,8 +484,13 @@ void rxrpc_input_call_event(struct rxrpc_call *call, struct sk_buff *skb)
        }
 
 out:
-       if (call->state == RXRPC_CALL_COMPLETE)
+       if (call->state == RXRPC_CALL_COMPLETE) {
                del_timer_sync(&call->timer);
+               if (!test_bit(RXRPC_CALL_DISCONNECTED, &call->flags))
+                       rxrpc_disconnect_call(call);
+               if (call->security)
+                       call->security->free_call_crypto(call);
+       }
        if (call->acks_hard_ack != call->tx_bottom)
                rxrpc_shrink_call_tx_buffer(call);
        _leave("");

diff --git a/net/rxrpc/call_object.c b/net/rxrpc/call_object.c
index 298b7c465d7eea5981d185ce8a23e0be37b74e8d..13aac3ca03a036322c3b94b96ca21ba618842f96 100644 (file)
--- a/net/rxrpc/call_object.c
+++ b/net/rxrpc/call_object.c
@@ -50,7 +50,7 @@ void rxrpc_poke_call(struct rxrpc_call *call, enum rxrpc_call_poke_trace what)
        struct rxrpc_local *local = call->local;
        bool busy;
 
-       if (call->state < RXRPC_CALL_COMPLETE) {
+       if (!test_bit(RXRPC_CALL_DISCONNECTED, &call->flags)) {
                spin_lock_bh(&local->lock);
                busy = !list_empty(&call->attend_link);
                trace_rxrpc_poke_call(call, busy, what);
@@ -533,13 +533,10 @@ void rxrpc_release_call(struct rxrpc_sock *rx, struct rxrpc_call *call)
        trace_rxrpc_call(call->debug_id, refcount_read(&call->ref),
                         call->flags, rxrpc_call_see_release);
 
-       ASSERTCMP(call->state, ==, RXRPC_CALL_COMPLETE);
-
        if (test_and_set_bit(RXRPC_CALL_RELEASED, &call->flags))
                BUG();
 
        rxrpc_put_call_slot(call);
-       del_timer_sync(&call->timer);
 
        /* Make sure we don't get any more notifications */
        write_lock(&rx->recvmsg_lock);
@@ -572,10 +569,6 @@ void rxrpc_release_call(struct rxrpc_sock *rx, struct rxrpc_call *call)
 
        _debug("RELEASE CALL %p (%d CONN %p)", call, call->debug_id, conn);
 
-       if (conn && !test_bit(RXRPC_CALL_DISCONNECTED, &call->flags))
-               rxrpc_disconnect_call(call);
-       if (call->security)
-               call->security->free_call_crypto(call);
        _leave("");
 }
 

diff --git a/net/rxrpc/input.c b/net/rxrpc/input.c
index 1f03a286620d5680051ba6d4b03085880eb3fadf..bb4beb4453259e31f12f861cbaee78bf7ad620a2 100644 (file)
--- a/net/rxrpc/input.c
+++ b/net/rxrpc/input.c
@@ -997,8 +997,6 @@ void rxrpc_input_call_packet(struct rxrpc_call *call, struct sk_buff *skb)
  */
 void rxrpc_implicit_end_call(struct rxrpc_call *call, struct sk_buff *skb)
 {
-       struct rxrpc_connection *conn = call->conn;
-
        switch (READ_ONCE(call->state)) {
        case RXRPC_CALL_SERVER_AWAIT_ACK:
                rxrpc_call_completed(call);
@@ -1012,8 +1010,4 @@ void rxrpc_implicit_end_call(struct rxrpc_call *call, struct sk_buff *skb)
        }
 
        rxrpc_input_call_event(call, skb);
-
-       spin_lock(&conn->bundle->channel_lock);
-       __rxrpc_disconnect_call(conn, call);
-       spin_unlock(&conn->bundle->channel_lock);
 }

diff --git a/net/rxrpc/recvmsg.c b/net/rxrpc/recvmsg.c
index a4ccdc006d0fee22d1c9d3e02a0a2d46098af560..8d5fe65f5951cb228e23ce1ae67df757c1986d6a 100644 (file)
--- a/net/rxrpc/recvmsg.c
+++ b/net/rxrpc/recvmsg.c
@@ -201,6 +201,7 @@ static void rxrpc_end_rx_phase(struct rxrpc_call *call, rxrpc_serial_t serial)
        case RXRPC_CALL_CLIENT_RECV_REPLY:
                __rxrpc_call_completed(call);
                write_unlock(&call->state_lock);
+               rxrpc_poke_call(call, rxrpc_call_poke_complete);
                break;
 
        case RXRPC_CALL_SERVER_RECV_REQUEST: