rxrpc: Move packet reception processing into I/O thread

Split the packet input handler to make the softirq side just dump the
received packet into the local endpoint receive queue and then call the
remainder of the input function from the I/O thread.

Signed-off-by: David Howells <dhowells@redhat.com>
cc: Marc Dionne <marc.dionne@auristor.com>
cc: linux-afs@lists.infradead.org

diff --git a/net/rxrpc/ar-internal.h b/net/rxrpc/ar-internal.h
index de82c25956a6f24507844fe337f028096128fec1..044815ba2b493afc3f808bc9f2c5e0b939b2e6f6 100644 (file)
--- a/net/rxrpc/ar-internal.h
+++ b/net/rxrpc/ar-internal.h
@@ -36,6 +36,7 @@ struct rxrpc_txbuf;
  * to pass supplementary information.
  */
 enum rxrpc_skb_mark {
+       RXRPC_SKB_MARK_PACKET,          /* Received packet */
        RXRPC_SKB_MARK_REJECT_BUSY,     /* Reject with BUSY */
        RXRPC_SKB_MARK_REJECT_ABORT,    /* Reject with ABORT (code in skb->priority) */
 };
@@ -957,7 +958,7 @@ void rxrpc_input_implicit_end_call(struct rxrpc_sock *, struct rxrpc_connection
 /*
  * io_thread.c
  */
-int rxrpc_input_packet(struct sock *, struct sk_buff *);
+int rxrpc_encap_rcv(struct sock *, struct sk_buff *);
 int rxrpc_io_thread(void *data);
 static inline void rxrpc_wake_up_io_thread(struct rxrpc_local *local)
 {

diff --git a/net/rxrpc/call_event.c b/net/rxrpc/call_event.c
index 049b92b1c0406415297a302598b0c3f37e46ab0b..3925b55e20645411064e9c0cbdc71add217aaf45 100644 (file)
--- a/net/rxrpc/call_event.c
+++ b/net/rxrpc/call_event.c
@@ -83,7 +83,7 @@ void rxrpc_send_ACK(struct rxrpc_call *call, u8 ack_reason,
        rxrpc_inc_stat(call->rxnet, stat_tx_acks[ack_reason]);
 
        txb = rxrpc_alloc_txbuf(call, RXRPC_PACKET_TYPE_ACK,
-                               in_softirq() ? GFP_ATOMIC | __GFP_NOWARN : GFP_NOFS);
+                               rcu_read_lock_held() ? GFP_ATOMIC | __GFP_NOWARN : GFP_NOFS);
        if (!txb) {
                kleave(" = -ENOMEM");
                return;
@@ -111,7 +111,7 @@ void rxrpc_send_ACK(struct rxrpc_call *call, u8 ack_reason,
        spin_unlock_bh(&local->ack_tx_lock);
        trace_rxrpc_send_ack(call, why, ack_reason, serial);
 
-       if (in_task()) {
+       if (!rcu_read_lock_held()) {
                rxrpc_transmit_ack_packets(call->peer->local);
        } else {
                rxrpc_get_local(local, rxrpc_local_get_queue);

diff --git a/net/rxrpc/call_object.c b/net/rxrpc/call_object.c
index 9cd7e0190ef4c35c764a852d727bcdf9879b54b8..57c8d4cc900a1169832f9dac8c0e8289f93a745c 100644 (file)
--- a/net/rxrpc/call_object.c
+++ b/net/rxrpc/call_object.c
@@ -632,7 +632,7 @@ void rxrpc_cleanup_call(struct rxrpc_call *call)
        del_timer_sync(&call->timer);
        cancel_work(&call->processor);
 
-       if (in_softirq() || work_busy(&call->processor))
+       if (rcu_read_lock_held() || work_busy(&call->processor))
                /* Can't use the rxrpc workqueue as we need to cancel/flush
                 * something that may be running/waiting there.
                 */

diff --git a/net/rxrpc/io_thread.c b/net/rxrpc/io_thread.c
index 0b3e096e3d50e3da5ec451e7ce5422928758241b..ee2e36c46ae26dfbb15a050fa8a4f27f739cecae 100644 (file)
--- a/net/rxrpc/io_thread.c
+++ b/net/rxrpc/io_thread.c
@@ -9,6 +9,34 @@
 
 #include "ar-internal.h"
 
+/*
+ * handle data received on the local endpoint
+ * - may be called in interrupt context
+ *
+ * [!] Note that as this is called from the encap_rcv hook, the socket is not
+ * held locked by the caller and nothing prevents sk_user_data on the UDP from
+ * being cleared in the middle of processing this function.
+ *
+ * Called with the RCU read lock held from the IP layer via UDP.
+ */
+int rxrpc_encap_rcv(struct sock *udp_sk, struct sk_buff *skb)
+{
+       struct rxrpc_local *local = rcu_dereference_sk_user_data(udp_sk);
+
+       if (unlikely(!local)) {
+               kfree_skb(skb);
+               return 0;
+       }
+       if (skb->tstamp == 0)
+               skb->tstamp = ktime_get_real();
+
+       skb->mark = RXRPC_SKB_MARK_PACKET;
+       rxrpc_new_skb(skb, rxrpc_skb_new_encap_rcv);
+       skb_queue_tail(&local->rx_queue, skb);
+       rxrpc_wake_up_io_thread(local);
+       return 0;
+}
+
 /*
  * post connection-level events to the connection
  * - this includes challenges, responses, some aborts and call terminal packet
@@ -98,18 +126,10 @@ static bool rxrpc_extract_abort(struct sk_buff *skb)
 }
 
 /*
- * handle data received on the local endpoint
- * - may be called in interrupt context
- *
- * [!] Note that as this is called from the encap_rcv hook, the socket is not
- * held locked by the caller and nothing prevents sk_user_data on the UDP from
- * being cleared in the middle of processing this function.
- *
- * Called with the RCU read lock held from the IP layer via UDP.
+ * Process packets received on the local endpoint
  */
-int rxrpc_input_packet(struct sock *udp_sk, struct sk_buff *skb)
+static int rxrpc_input_packet(struct rxrpc_local *local, struct sk_buff *skb)
 {
-       struct rxrpc_local *local = rcu_dereference_sk_user_data(udp_sk);
        struct rxrpc_connection *conn;
        struct rxrpc_channel *chan;
        struct rxrpc_call *call = NULL;
@@ -118,17 +138,9 @@ int rxrpc_input_packet(struct sock *udp_sk, struct sk_buff *skb)
        struct rxrpc_sock *rx = NULL;
        unsigned int channel;
 
-       _enter("%p", udp_sk);
-
-       if (unlikely(!local)) {
-               kfree_skb(skb);
-               return 0;
-       }
        if (skb->tstamp == 0)
                skb->tstamp = ktime_get_real();
 
-       rxrpc_new_skb(skb, rxrpc_skb_new_encap_rcv);
-
        skb_pull(skb, sizeof(struct udphdr));
 
        /* The UDP protocol already released all skb resources;
@@ -387,8 +399,17 @@ int rxrpc_io_thread(void *data)
 
                /* Process received packets and errors. */
                if ((skb = __skb_dequeue(&rx_queue))) {
-                       // TODO: Input packet
-                       rxrpc_free_skb(skb, rxrpc_skb_put_input);
+                       switch (skb->mark) {
+                       case RXRPC_SKB_MARK_PACKET:
+                               rcu_read_lock();
+                               rxrpc_input_packet(local, skb);
+                               rcu_read_unlock();
+                               break;
+                       default:
+                               WARN_ON_ONCE(1);
+                               rxrpc_free_skb(skb, rxrpc_skb_put_unknown);
+                               break;
+                       }
                        continue;
                }
 

diff --git a/net/rxrpc/local_object.c b/net/rxrpc/local_object.c
index 7c61349984e3a044ab72129ff1d24e8a2f98e496..6b4d77219f3679c64ab47f3cb0299b7f6ce1f52d 100644 (file)
--- a/net/rxrpc/local_object.c
+++ b/net/rxrpc/local_object.c
@@ -154,7 +154,7 @@ static int rxrpc_open_socket(struct rxrpc_local *local, struct net *net)
        }
 
        tuncfg.encap_type = UDP_ENCAP_RXRPC;
-       tuncfg.encap_rcv = rxrpc_input_packet;
+       tuncfg.encap_rcv = rxrpc_encap_rcv;
        tuncfg.encap_err_rcv = rxrpc_encap_err_rcv;
        tuncfg.sk_user_data = local;
        setup_udp_tunnel_sock(net, local->socket, &tuncfg);