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

4 files changed, 156 insertions, 1 deletions

diff --git a/net/rxrpc/Makefile b/net/rxrpc/Makefile
index b11281bed2a49..fdeba488fc6e1 100644
--- a/net/rxrpc/Makefile
+++ b/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
diff --git a/net/rxrpc/ar-internal.h b/net/rxrpc/ar-internal.h
index 282cb986f8a79..49e90614d5e5f 100644
--- a/net/rxrpc/ar-internal.h
+++ b/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>
 
 /*
@@ -1126,6 +1169,16 @@ 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
  */
 int rxrpc_extract_addr_from_skb(struct sockaddr_rxrpc *, struct sk_buff *);
diff --git a/net/rxrpc/proc.c b/net/rxrpc/proc.c
index 9bd357f39c39d..3877afd235982 100644
--- a/net/rxrpc/proc.c
+++ b/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;
diff --git a/net/rxrpc/txbuf.c b/net/rxrpc/txbuf.c
new file mode 100644
index 0000000000000..66d922ad65d0d
--- /dev/null
+++ b/net/rxrpc/txbuf.c
@@ -0,0 +1,100 @@
+// 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);
+	}
+}