[PATCH net 7/9] rxrpc: Add a shadow refcount in the skb private data

[David Howells]

Add a shadow refcount to count pins from the Rx/Tx ring on an sk_buff so
that we can hold multiple refs on it without causing skb_cow_data() to
throw an assertion.

This is stored in the private part of the sk_buff as laid out in struct
rxrpc_skb_priv.

Add two accessor functions for pinning (adding) or unpinning (discarding) a
shadow ref.

Signed-off-by: David Howells <dhowells@xxxxxxxxxx>
---

 include/trace/events/rxrpc.h |   15 ++++++++---
 net/rxrpc/ar-internal.h      |    2 +
 net/rxrpc/skbuff.c           |   57 ++++++++++++++++++++++++++++++++++++++----
 3 files changed, 65 insertions(+), 9 deletions(-)

diff --git a/include/trace/events/rxrpc.h b/include/trace/events/rxrpc.h
index e2356c51883b..34237ea8ceb0 100644
--- a/include/trace/events/rxrpc.h
+++ b/include/trace/events/rxrpc.h
@@ -28,10 +28,12 @@ enum rxrpc_skb_trace {
 	rxrpc_skb_got,
 	rxrpc_skb_lost,
 	rxrpc_skb_new,
+	rxrpc_skb_pin,
 	rxrpc_skb_purged,
 	rxrpc_skb_received,
 	rxrpc_skb_rotated,
 	rxrpc_skb_seen,
+	rxrpc_skb_unpin,
 };
 
 enum rxrpc_local_trace {
@@ -228,10 +230,12 @@ enum rxrpc_tx_point {
 	EM(rxrpc_skb_got,			"GOT") \
 	EM(rxrpc_skb_lost,			"*L*") \
 	EM(rxrpc_skb_new,			"NEW") \
+	EM(rxrpc_skb_pin,			"PIN") \
 	EM(rxrpc_skb_purged,			"PUR") \
 	EM(rxrpc_skb_received,			"RCV") \
 	EM(rxrpc_skb_rotated,			"ROT") \
-	E_(rxrpc_skb_seen,			"SEE")
+	EM(rxrpc_skb_seen,			"SEE") \
+	E_(rxrpc_skb_unpin,			"UPN")
 
 #define rxrpc_local_traces \
 	EM(rxrpc_local_got,			"GOT") \
@@ -633,14 +637,15 @@ TRACE_EVENT(rxrpc_call,
 
 TRACE_EVENT(rxrpc_skb,
 	    TP_PROTO(struct sk_buff *skb, enum rxrpc_skb_trace op,
-		     int usage, int mod_count, const void *where),
+		     int usage, int mod_count, int pins, const void *where),
 
-	    TP_ARGS(skb, op, usage, mod_count, where),
+	    TP_ARGS(skb, op, usage, mod_count, pins, where),
 
 	    TP_STRUCT__entry(
 		    __field(struct sk_buff *,		skb		)
 		    __field(enum rxrpc_skb_trace,	op		)
 		    __field(u8,				flags		)
+		    __field(u8,				pins		)
 		    __field(int,			usage		)
 		    __field(int,			mod_count	)
 		    __field(const void *,		where		)
@@ -651,16 +656,18 @@ TRACE_EVENT(rxrpc_skb,
 		    __entry->flags = rxrpc_skb(skb)->rx_flags;
 		    __entry->op = op;
 		    __entry->usage = usage;
+		    __entry->pins = pins;
 		    __entry->mod_count = mod_count;
 		    __entry->where = where;
 			   ),
 
-	    TP_printk("s=%p %cx %s u=%d m=%d p=%pSR",
+	    TP_printk("s=%p %cx %s u=%d m=%d r=%u p=%pSR",
 		      __entry->skb,
 		      __entry->flags & RXRPC_SKB_TX_BUFFER ? 'T' : 'R',
 		      __print_symbolic(__entry->op, rxrpc_skb_traces),
 		      __entry->usage,
 		      __entry->mod_count,
+		      __entry->pins,
 		      __entry->where)
 	    );
 
diff --git a/net/rxrpc/ar-internal.h b/net/rxrpc/ar-internal.h
index 2d5294f3e62f..d784d58e0a0d 100644
--- a/net/rxrpc/ar-internal.h
+++ b/net/rxrpc/ar-internal.h
@@ -1113,6 +1113,8 @@ void rxrpc_see_skb(struct sk_buff *, enum rxrpc_skb_trace);
 void rxrpc_get_skb(struct sk_buff *, enum rxrpc_skb_trace);
 void rxrpc_free_skb(struct sk_buff *, enum rxrpc_skb_trace);
 void rxrpc_purge_queue(struct sk_buff_head *);
+void rxrpc_pin_skb(struct sk_buff *, enum rxrpc_skb_trace);
+void rxrpc_unpin_skb(struct sk_buff *, enum rxrpc_skb_trace);
 
 /*
  * sysctl.c
diff --git a/net/rxrpc/skbuff.c b/net/rxrpc/skbuff.c
index 8e6f45f84b9b..f9986a1510d3 100644
--- a/net/rxrpc/skbuff.c
+++ b/net/rxrpc/skbuff.c
@@ -22,9 +22,12 @@
  */
 void rxrpc_new_skb(struct sk_buff *skb, enum rxrpc_skb_trace op)
 {
+	struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
 	const void *here = __builtin_return_address(0);
 	int n = atomic_inc_return(select_skb_count(skb));
-	trace_rxrpc_skb(skb, op, refcount_read(&skb->users), n, here);
+
+	atomic_set(&sp->nr_ring_pins, 1);
+	trace_rxrpc_skb(skb, op, refcount_read(&skb->users), n, 1, here);
 }
 
 /*
@@ -33,9 +36,12 @@ void rxrpc_new_skb(struct sk_buff *skb, enum rxrpc_skb_trace op)
 void rxrpc_see_skb(struct sk_buff *skb, enum rxrpc_skb_trace op)
 {
 	const void *here = __builtin_return_address(0);
+
 	if (skb) {
+		struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
 		int n = atomic_read(select_skb_count(skb));
-		trace_rxrpc_skb(skb, op, refcount_read(&skb->users), n, here);
+		trace_rxrpc_skb(skb, op, refcount_read(&skb->users), n,
+				atomic_read(&sp->nr_ring_pins), here);
 	}
 }
 
@@ -44,9 +50,11 @@ void rxrpc_see_skb(struct sk_buff *skb, enum rxrpc_skb_trace op)
  */
 void rxrpc_get_skb(struct sk_buff *skb, enum rxrpc_skb_trace op)
 {
+	struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
 	const void *here = __builtin_return_address(0);
 	int n = atomic_inc_return(select_skb_count(skb));
-	trace_rxrpc_skb(skb, op, refcount_read(&skb->users), n, here);
+	trace_rxrpc_skb(skb, op, refcount_read(&skb->users), n,
+			atomic_read(&sp->nr_ring_pins), here);
 	skb_get(skb);
 }
 
@@ -56,11 +64,14 @@ void rxrpc_get_skb(struct sk_buff *skb, enum rxrpc_skb_trace op)
 void rxrpc_free_skb(struct sk_buff *skb, enum rxrpc_skb_trace op)
 {
 	const void *here = __builtin_return_address(0);
+
 	if (skb) {
+		struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
 		int n;
 		CHECK_SLAB_OKAY(&skb->users);
 		n = atomic_dec_return(select_skb_count(skb));
-		trace_rxrpc_skb(skb, op, refcount_read(&skb->users), n, here);
+		trace_rxrpc_skb(skb, op, refcount_read(&skb->users), n,
+				atomic_read(&sp->nr_ring_pins), here);
 		kfree_skb(skb);
 	}
 }
@@ -72,10 +83,46 @@ void rxrpc_purge_queue(struct sk_buff_head *list)
 {
 	const void *here = __builtin_return_address(0);
 	struct sk_buff *skb;
+
 	while ((skb = skb_dequeue((list))) != NULL) {
+		struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
 		int n = atomic_dec_return(select_skb_count(skb));
 		trace_rxrpc_skb(skb, rxrpc_skb_purged,
-				refcount_read(&skb->users), n, here);
+				refcount_read(&skb->users), n,
+				atomic_read(&sp->nr_ring_pins), here);
 		kfree_skb(skb);
 	}
 }
+
+/*
+ * Add a secondary ref on the socket buffer.
+ */
+void rxrpc_pin_skb(struct sk_buff *skb, enum rxrpc_skb_trace op)
+{
+	struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
+	const void *here = __builtin_return_address(0);
+	int n = atomic_read(select_skb_count(skb));
+	int np;
+
+	np = atomic_inc_return(&sp->nr_ring_pins);
+	trace_rxrpc_skb(skb, op, refcount_read(&skb->users), n, np, here);
+}
+
+/*
+ * Remove a secondary ref on the socket buffer.
+ */
+void rxrpc_unpin_skb(struct sk_buff *skb, enum rxrpc_skb_trace op)
+{
+	const void *here = __builtin_return_address(0);
+
+	if (skb) {
+		struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
+		int n = atomic_read(select_skb_count(skb));
+		int np;
+
+		np = atomic_dec_return(&sp->nr_ring_pins);
+		trace_rxrpc_skb(skb, op, refcount_read(&skb->users), n, np, here);
+		if (np == 0)
+			rxrpc_free_skb(skb, op);
+	}
+}