rxrpc: Define rxrpc_txbuf struct to carry data to be transmitted

Define a struct, rxrpc_txbuf, to carry data to be transmitted instead of a
socket buffer so that it can be placed onto multiple queues at once.  This
also allows the data buffer to be in the same allocation as the internal
data.
Signed-off-by: David Howells <dhowells@redhat.com>
cc: Marc Dionne <marc.dionne@auristor.com>
cc: linux-afs@lists.infradead.org

include/trace/events/rxrpc.h

@@ -252,6 +252,18 @@
 	E_(rxrpc_reqack_small_txwin,		"SMALL-TXWN")
 /* ---- Must update size of stat_why_req_ack[] if more are added! */
 
+#define rxrpc_txbuf_traces \
+	EM(rxrpc_txbuf_alloc_ack,		"ALLOC ACK  ")	\
+	EM(rxrpc_txbuf_alloc_data,		"ALLOC DATA ")	\
+	EM(rxrpc_txbuf_free,			"FREE       ")	\
+	EM(rxrpc_txbuf_get_trans,		"GET TRANS  ")	\
+	EM(rxrpc_txbuf_get_retrans,		"GET RETRANS")	\
+	EM(rxrpc_txbuf_put_cleaned,		"PUT CLEANED")	\
+	EM(rxrpc_txbuf_put_rotated,		"PUT ROTATED")	\
+	EM(rxrpc_txbuf_put_send_aborted,	"PUT SEND-X ")	\
+	EM(rxrpc_txbuf_see_send_more,		"SEE SEND+  ")	\
+	E_(rxrpc_txbuf_see_unacked,		"SEE UNACKED")
+
 /*
  * Generate enums for tracing information.
  */
@@ -280,6 +292,7 @@ enum rxrpc_skb_trace		{ rxrpc_skb_traces } __mode(byte);
 enum rxrpc_timer_trace		{ rxrpc_timer_traces } __mode(byte);
 enum rxrpc_transmit_trace	{ rxrpc_transmit_traces } __mode(byte);
 enum rxrpc_tx_point		{ rxrpc_tx_points } __mode(byte);
+enum rxrpc_txbuf_trace		{ rxrpc_txbuf_traces } __mode(byte);
 
 #endif /* end __RXRPC_DECLARE_TRACE_ENUMS_ONCE_ONLY */
 
@@ -308,6 +321,7 @@ rxrpc_skb_traces;
 rxrpc_timer_traces;
 rxrpc_transmit_traces;
 rxrpc_tx_points;
+rxrpc_txbuf_traces;
 
 /*
  * Now redefine the EM() and E_() macros to map the enums to the strings that
@@ -1469,6 +1483,37 @@ TRACE_EVENT(rxrpc_req_ack,
 		      __print_symbolic(__entry->why, rxrpc_req_ack_traces))
 	    );
 
+TRACE_EVENT(rxrpc_txbuf,
+	    TP_PROTO(unsigned int debug_id,
+		     unsigned int call_debug_id, rxrpc_seq_t seq,
+		     int ref, enum rxrpc_txbuf_trace what),
+
+	    TP_ARGS(debug_id, call_debug_id, seq, ref, what),
+
+	    TP_STRUCT__entry(
+		    __field(unsigned int,		debug_id	)
+		    __field(unsigned int,		call_debug_id	)
+		    __field(rxrpc_seq_t,		seq		)
+		    __field(int,			ref		)
+		    __field(enum rxrpc_txbuf_trace,	what		)
+			     ),
+
+	    TP_fast_assign(
+		    __entry->debug_id = debug_id;
+		    __entry->call_debug_id = call_debug_id;
+		    __entry->seq = seq;
+		    __entry->ref = ref;
+		    __entry->what = what;
+			   ),
+
+	    TP_printk("B=%08x c=%08x q=%08x %s r=%d",
+		      __entry->debug_id,
+		      __entry->call_debug_id,
+		      __entry->seq,
+		      __print_symbolic(__entry->what, rxrpc_txbuf_traces),
+		      __entry->ref)
+	    );
+
 #undef EM
 #undef E_
 #endif /* _TRACE_RXRPC_H */

net/rxrpc/Makefile

@@ -30,6 +30,7 @@ rxrpc-y := \
 	sendmsg.o \
 	server_key.o \
 	skbuff.o \
+	txbuf.o \
 	utils.o
 
 rxrpc-$(CONFIG_PROC_FS) += proc.o

net/rxrpc/ar-internal.h

@@ -29,6 +29,7 @@ struct rxrpc_crypt {
 
 struct key_preparsed_payload;
 struct rxrpc_connection;
+struct rxrpc_txbuf;
 
 /*
  * Mark applied to socket buffers in skb->mark.  skb->priority is used
@@ -759,6 +760,48 @@ struct rxrpc_send_params {
 	bool			upgrade;	/* If the connection is upgradeable */
 };
 
+/*
+ * Buffer of data to be output as a packet.
+ */
+struct rxrpc_txbuf {
+	struct rcu_head		rcu;
+	struct list_head	call_link;	/* Link in call->tx_queue */
+	struct list_head	tx_link;	/* Link in live Enc queue or Tx queue */
+	struct rxrpc_call	*call;		/* Call to which belongs */
+	ktime_t			last_sent;	/* Time at which last transmitted */
+	refcount_t		ref;
+	rxrpc_seq_t		seq;		/* Sequence number of this packet */
+	unsigned int		call_debug_id;
+	unsigned int		debug_id;
+	unsigned int		len;		/* Amount of data in buffer */
+	unsigned int		space;		/* Remaining data space */
+	unsigned int		offset;		/* Offset of fill point */
+	unsigned long		flags;
+#define RXRPC_TXBUF_ACKED	0		/* Set if ACK'd */
+#define RXRPC_TXBUF_NACKED	1		/* Set if NAK'd */
+#define RXRPC_TXBUF_LAST	2		/* Set if last packet in Tx phase */
+#define RXRPC_TXBUF_RESENT	3		/* Set if has been resent */
+#define RXRPC_TXBUF_RETRANS	4		/* Set if should be retransmitted */
+	struct {
+		/* The packet for encrypting and DMA'ing.  We align it such
+		 * that data[] aligns correctly for any crypto blocksize.
+		 */
+		u8		pad[64 - sizeof(struct rxrpc_wire_header)];
+		struct rxrpc_wire_header wire;	/* Network-ready header */
+		u8		data[RXRPC_JUMBO_DATALEN]; /* Data packet */
+	} __aligned(64);
+};
+
+static inline bool rxrpc_sending_to_server(const struct rxrpc_txbuf *txb)
+{
+	return txb->wire.flags & RXRPC_CLIENT_INITIATED;
+}
+
+static inline bool rxrpc_sending_to_client(const struct rxrpc_txbuf *txb)
+{
+	return !rxrpc_sending_to_server(txb);
+}
+
 #include <trace/events/rxrpc.h>
 
 /*
@@ -1125,6 +1168,16 @@ static inline int __init rxrpc_sysctl_init(void) { return 0; }
 static inline void rxrpc_sysctl_exit(void) {}
 #endif
 
+/*
+ * txbuf.c
+ */
+extern atomic_t rxrpc_nr_txbuf;
+struct rxrpc_txbuf *rxrpc_alloc_txbuf(struct rxrpc_call *call, u8 packet_type,
+				      gfp_t gfp);
+void rxrpc_get_txbuf(struct rxrpc_txbuf *txb, enum rxrpc_txbuf_trace what);
+void rxrpc_see_txbuf(struct rxrpc_txbuf *txb, enum rxrpc_txbuf_trace what);
+void rxrpc_put_txbuf(struct rxrpc_txbuf *txb, enum rxrpc_txbuf_trace what);
+
 /*
  * utils.c
  */

net/rxrpc/proc.c

@@ -458,7 +458,8 @@ int rxrpc_stats_show(struct seq_file *seq, void *v)
 		   atomic_read(&rxnet->stat_why_req_ack[rxrpc_reqack_slow_start]),
 		   atomic_read(&rxnet->stat_why_req_ack[rxrpc_reqack_small_txwin]));
 	seq_printf(seq,
-		   "Buffers  : txb=%u rxb=%u\n",
+		   "Buffers  : txb=%u,%u rxb=%u\n",
+		   atomic_read(&rxrpc_nr_txbuf),
 		   atomic_read(&rxrpc_n_tx_skbs),
 		   atomic_read(&rxrpc_n_rx_skbs));
 	return 0;

net/rxrpc/txbuf.c

+// SPDX-License-Identifier: GPL-2.0-or-later
+/* RxRPC Tx data buffering.
+ *
+ * Copyright (C) 2022 Red Hat, Inc. All Rights Reserved.
+ * Written by David Howells (dhowells@redhat.com)
+ */
+
+#define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
+
+#include <linux/slab.h>
+#include "ar-internal.h"
+
+static atomic_t rxrpc_txbuf_debug_ids;
+atomic_t rxrpc_nr_txbuf;
+
+/*
+ * Allocate and partially initialise an I/O request structure.
+ */
+struct rxrpc_txbuf *rxrpc_alloc_txbuf(struct rxrpc_call *call, u8 packet_type,
+				      gfp_t gfp)
+{
+	struct rxrpc_txbuf *txb;
+
+	txb = kmalloc(sizeof(*txb), gfp);
+	if (txb) {
+		INIT_LIST_HEAD(&txb->call_link);
+		INIT_LIST_HEAD(&txb->tx_link);
+		refcount_set(&txb->ref, 1);
+		txb->call		= call;
+		txb->call_debug_id	= call->debug_id;
+		txb->debug_id		= atomic_inc_return(&rxrpc_txbuf_debug_ids);
+		txb->space		= sizeof(txb->data);
+		txb->len		= 0;
+		txb->offset		= 0;
+		txb->flags		= 0;
+		txb->seq		= call->tx_top + 1;
+		txb->wire.epoch		= htonl(call->conn->proto.epoch);
+		txb->wire.cid		= htonl(call->cid);
+		txb->wire.callNumber	= htonl(call->call_id);
+		txb->wire.seq		= htonl(txb->seq);
+		txb->wire.type		= packet_type;
+		txb->wire.flags		= call->conn->out_clientflag;
+		txb->wire.userStatus	= 0;
+		txb->wire.securityIndex	= call->security_ix;
+		txb->wire._rsvd		= 0;
+		txb->wire.serviceId	= htons(call->service_id);
+
+		trace_rxrpc_txbuf(txb->debug_id,
+				  txb->call_debug_id, txb->seq, 1,
+				  packet_type == RXRPC_PACKET_TYPE_DATA ?
+				  rxrpc_txbuf_alloc_data :
+				  rxrpc_txbuf_alloc_ack);
+		atomic_inc(&rxrpc_nr_txbuf);
+	}
+
+	return txb;
+}
+
+void rxrpc_get_txbuf(struct rxrpc_txbuf *txb, enum rxrpc_txbuf_trace what)
+{
+	int r;
+
+	__refcount_inc(&txb->ref, &r);
+	trace_rxrpc_txbuf(txb->debug_id, txb->call_debug_id, txb->seq, r + 1, what);
+}
+
+void rxrpc_see_txbuf(struct rxrpc_txbuf *txb, enum rxrpc_txbuf_trace what)
+{
+	int r = refcount_read(&txb->ref);
+
+	trace_rxrpc_txbuf(txb->debug_id, txb->call_debug_id, txb->seq, r, what);
+}
+
+static void rxrpc_free_txbuf(struct rcu_head *rcu)
+{
+	struct rxrpc_txbuf *txb = container_of(rcu, struct rxrpc_txbuf, rcu);
+
+	trace_rxrpc_txbuf(txb->debug_id, txb->call_debug_id, txb->seq, 0,
+			  rxrpc_txbuf_free);
+	kfree(txb);
+	atomic_dec(&rxrpc_nr_txbuf);
+}
+
+void rxrpc_put_txbuf(struct rxrpc_txbuf *txb, enum rxrpc_txbuf_trace what)
+{
+	unsigned int debug_id, call_debug_id;
+	rxrpc_seq_t seq;
+	bool dead;
+	int r;
+
+	if (txb) {
+		debug_id = txb->debug_id;
+		call_debug_id = txb->call_debug_id;
+		seq = txb->seq;
+		dead = __refcount_dec_and_test(&txb->ref, &r);
+		trace_rxrpc_txbuf(debug_id, call_debug_id, seq, r - 1, what);
+		if (dead)
+			call_rcu(&txb->rcu, rxrpc_free_txbuf);
+	}
+}