rxrpc: Only transmit one ACK per jumbo packet received

Only generate one ACK packet for all the subpackets in a jumbo packet.  If
we would like to generate more than one ACK, we prioritise them base on
their reason code, in the order, highest first:

   OutOfSeq > NoSpace > ExceedsWin > Duplicate > Requested > Delay > Idle

For the first four, we reference the lowest offending subpacket; for the
last three, the highest.

This reduces the number of ACKs we end up transmitting to one per UDP
packet transmitted to reduce network loading and packet parsing.

Fixes: 5d7edbc9231e ("rxrpc: Get rid of the Rx ring")
Signed-off-by: David Howells <dhowells@redhat.com>
cc: Marc Dionne <marc.dionne@auristor.com>
cc: linux-afs@lists.infradead.org
Reviewed-by: Jeffrey Altman <jaltman@auristor.com <mailto:jaltman@auristor.com>>
Link: https://lore.kernel.org/r/20240503150749.1001323-3-dhowells@redhat.com
Signed-off-by: Jakub Kicinski <kuba@kernel.org>

diff --git a/net/rxrpc/input.c b/net/rxrpc/input.c
index 860075f1290b65457f9ebe46d5a24017675db668..16d49a861dbb582fa5055101f4e536bb5b70f71d 100644 (file)
--- a/net/rxrpc/input.c
+++ b/net/rxrpc/input.c
@@ -9,6 +9,17 @@
 
 #include "ar-internal.h"
 
+/* Override priority when generating ACKs for received DATA */
+static const u8 rxrpc_ack_priority[RXRPC_ACK__INVALID] = {
+       [RXRPC_ACK_IDLE]                = 1,
+       [RXRPC_ACK_DELAY]               = 2,
+       [RXRPC_ACK_REQUESTED]           = 3,
+       [RXRPC_ACK_DUPLICATE]           = 4,
+       [RXRPC_ACK_EXCEEDS_WINDOW]      = 5,
+       [RXRPC_ACK_NOSPACE]             = 6,
+       [RXRPC_ACK_OUT_OF_SEQUENCE]     = 7,
+};
+
 static void rxrpc_proto_abort(struct rxrpc_call *call, rxrpc_seq_t seq,
                              enum rxrpc_abort_reason why)
 {
@@ -365,7 +376,7 @@ static void rxrpc_input_queue_data(struct rxrpc_call *call, struct sk_buff *skb,
  * Process a DATA packet.
  */
 static void rxrpc_input_data_one(struct rxrpc_call *call, struct sk_buff *skb,
-                                bool *_notify)
+                                bool *_notify, rxrpc_serial_t *_ack_serial, int *_ack_reason)
 {
        struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
        struct sk_buff *oos;
@@ -418,8 +429,6 @@ static void rxrpc_input_data_one(struct rxrpc_call *call, struct sk_buff *skb,
                /* Send an immediate ACK if we fill in a hole */
                else if (!skb_queue_empty(&call->rx_oos_queue))
                        ack_reason = RXRPC_ACK_DELAY;
-               else
-                       call->ackr_nr_unacked++;
 
                window++;
                if (after(window, wtop)) {
@@ -497,12 +506,16 @@ static void rxrpc_input_data_one(struct rxrpc_call *call, struct sk_buff *skb,
        }
 
 send_ack:
-       if (ack_reason >= 0)
-               rxrpc_send_ACK(call, ack_reason, serial,
-                              rxrpc_propose_ack_input_data);
-       else
-               rxrpc_propose_delay_ACK(call, serial,
-                                       rxrpc_propose_ack_input_data);
+       if (ack_reason >= 0) {
+               if (rxrpc_ack_priority[ack_reason] > rxrpc_ack_priority[*_ack_reason]) {
+                       *_ack_serial = serial;
+                       *_ack_reason = ack_reason;
+               } else if (rxrpc_ack_priority[ack_reason] == rxrpc_ack_priority[*_ack_reason] &&
+                          ack_reason == RXRPC_ACK_REQUESTED) {
+                       *_ack_serial = serial;
+                       *_ack_reason = ack_reason;
+               }
+       }
 }
 
 /*
@@ -513,9 +526,11 @@ static bool rxrpc_input_split_jumbo(struct rxrpc_call *call, struct sk_buff *skb
        struct rxrpc_jumbo_header jhdr;
        struct rxrpc_skb_priv *sp = rxrpc_skb(skb), *jsp;
        struct sk_buff *jskb;
+       rxrpc_serial_t ack_serial = 0;
        unsigned int offset = sizeof(struct rxrpc_wire_header);
        unsigned int len = skb->len - offset;
        bool notify = false;
+       int ack_reason = 0;
 
        while (sp->hdr.flags & RXRPC_JUMBO_PACKET) {
                if (len < RXRPC_JUMBO_SUBPKTLEN)
@@ -535,7 +550,7 @@ static bool rxrpc_input_split_jumbo(struct rxrpc_call *call, struct sk_buff *skb
                jsp = rxrpc_skb(jskb);
                jsp->offset = offset;
                jsp->len = RXRPC_JUMBO_DATALEN;
-               rxrpc_input_data_one(call, jskb, &notify);
+               rxrpc_input_data_one(call, jskb, &notify, &ack_serial, &ack_reason);
                rxrpc_free_skb(jskb, rxrpc_skb_put_jumbo_subpacket);
 
                sp->hdr.flags = jhdr.flags;
@@ -548,7 +563,16 @@ static bool rxrpc_input_split_jumbo(struct rxrpc_call *call, struct sk_buff *skb
 
        sp->offset = offset;
        sp->len    = len;
-       rxrpc_input_data_one(call, skb, &notify);
+       rxrpc_input_data_one(call, skb, &notify, &ack_serial, &ack_reason);
+
+       if (ack_reason > 0) {
+               rxrpc_send_ACK(call, ack_reason, ack_serial,
+                              rxrpc_propose_ack_input_data);
+       } else {
+               call->ackr_nr_unacked++;
+               rxrpc_propose_delay_ACK(call, sp->hdr.serial,
+                                       rxrpc_propose_ack_input_data);
+       }
        if (notify) {
                trace_rxrpc_notify_socket(call->debug_id, sp->hdr.serial);
                rxrpc_notify_socket(call);