[PATCH v2 02/10] tap: Make L4 sequence pools per-qpair for thread safety

[Laurent Vivier <lvivier@redhat.com>]

The L4 sequence arrays tap4_l4[] and tap6_l4[] are used to batch
packets with the same L4 tuple within a single tap_handler() call.
They are global, but tap_handler() can be called concurrently from
different worker threads with different qpairs in vhost-user mode.

Make these arrays per-qpair by adding a VHOST_USER_MAX_VQS/2 first
dimension, indexed by the qpair parameter already available in
tap4_handler() and tap6_handler().

Update tap_sock_update_pool() to initialize all qpair*seq entries.

Signed-off-by: Laurent Vivier <lvivier@redhat.com>
---
 tap.c | 32 +++++++++++++++++++-------------
 1 file changed, 19 insertions(+), 13 deletions(-)

diff --git a/tap.c b/tap.c
index 6bba436f148f..25cde67538ac 100644
--- a/tap.c
+++ b/tap.c
@@ -591,7 +591,7 @@ static struct tap4_l4_t {
 	struct in_addr daddr;
 
 	struct pool_l4_t p;
-} tap4_l4[TAP_SEQS /* Arbitrary: TAP_MSGS in theory, so limit in users */];
+} tap4_l4[VHOST_USER_MAX_VQS / 2][TAP_SEQS /* Arbitrary: TAP_MSGS in theory, so limit in users */];
 
 /**
  * struct l4_seq6_t - Message sequence for one protocol handler call, IPv6
@@ -618,7 +618,7 @@ static struct tap6_l4_t {
 	uint8_t hop_limit;
 
 	struct pool_l4_t p;
-} tap6_l4[TAP_SEQS /* Arbitrary: TAP_MSGS in theory, so limit in users */];
+} tap6_l4[VHOST_USER_MAX_VQS / 2][TAP_SEQS /* Arbitrary: TAP_MSGS in theory, so limit in users */];
 
 /**
  * tap_packet_debug() - Print debug message for packet(s) from guest/tap
@@ -838,7 +838,7 @@ resume:
 		if (seq_count == TAP_SEQS)
 			break;	/* Resume after flushing if i < pool_tap4[qpair]->count */
 
-		for (seq = tap4_l4 + seq_count - 1; seq >= tap4_l4; seq--) {
+		for (seq = tap4_l4[qpair] + seq_count - 1; seq >= tap4_l4[qpair]; seq--) {
 			if (L4_MATCH(iph, uh, seq)) {
 				if (seq->p.count >= UIO_MAXIOV)
 					seq = NULL;
@@ -846,8 +846,8 @@ resume:
 			}
 		}
 
-		if (!seq || seq < tap4_l4) {
-			seq = tap4_l4 + seq_count++;
+		if (!seq || seq < tap4_l4[qpair]) {
+			seq = tap4_l4[qpair] + seq_count++;
 			L4_SET(iph, uh, seq);
 			pool_flush((struct pool *)&seq->p);
 		}
@@ -859,7 +859,7 @@ append:
 		packet_add((struct pool *)&seq->p, &data);
 	}
 
-	for (j = 0, seq = tap4_l4; j < seq_count; j++, seq++) {
+	for (j = 0, seq = tap4_l4[qpair]; j < seq_count; j++, seq++) {
 		const struct pool *p = (const struct pool *)&seq->p;
 		size_t k;
 
@@ -1088,7 +1088,7 @@ resume:
 		if (seq_count == TAP_SEQS)
 			break;	/* Resume after flushing if i < pool_tap6[qpair]->count */
 
-		for (seq = tap6_l4 + seq_count - 1; seq >= tap6_l4; seq--) {
+		for (seq = tap6_l4[qpair] + seq_count - 1; seq >= tap6_l4[qpair]; seq--) {
 			if (L4_MATCH(ip6h, proto, uh, seq)) {
 				if (seq->p.count >= UIO_MAXIOV)
 					seq = NULL;
@@ -1096,8 +1096,8 @@ resume:
 			}
 		}
 
-		if (!seq || seq < tap6_l4) {
-			seq = tap6_l4 + seq_count++;
+		if (!seq || seq < tap6_l4[qpair]) {
+			seq = tap6_l4[qpair] + seq_count++;
 			L4_SET(ip6h, proto, uh, seq);
 			pool_flush((struct pool *)&seq->p);
 		}
@@ -1109,7 +1109,7 @@ append:
 		packet_add((struct pool *)&seq->p, &data);
 	}
 
-	for (j = 0, seq = tap6_l4; j < seq_count; j++, seq++) {
+	for (j = 0, seq = tap6_l4[qpair]; j < seq_count; j++, seq++) {
 		const struct pool *p = (const struct pool *)&seq->p;
 		size_t k;
 
@@ -1602,9 +1602,15 @@ static void tap_sock_update_pool(void *base, size_t size)
 		pool_tap6[i] = (struct pool *)&pool_tap6_storage[i];
 	}
 
-	for (i = 0; i < TAP_SEQS; i++) {
-		tap4_l4[i].p = PACKET_INIT(pool_l4, UIO_MAXIOV, base, size);
-		tap6_l4[i].p = PACKET_INIT(pool_l4, UIO_MAXIOV, base, size);
+	for (i = 0; i < VHOST_USER_MAX_VQS / 2; i++) {
+		unsigned int j;
+
+		for (j = 0; j < TAP_SEQS; j++) {
+			tap4_l4[i][j].p = PACKET_INIT(pool_l4, UIO_MAXIOV,
+						      base, size);
+			tap6_l4[i][j].p = PACKET_INIT(pool_l4, UIO_MAXIOV,
+						      base, size);
+		}
 	}
 }
 
-- 
2.54.0