queueing: use ptr_ring instead of linked lists

7 files changed, 67 insertions, 76 deletions

diff --git a/src/device.c b/src/device.c
index fdade83..5102acc 100644
--- a/src/device.c
+++ b/src/device.c
@@ -209,8 +209,8 @@ static void destruct(struct net_device *dev)
 	wg->incoming_port = 0;
 	destroy_workqueue(wg->handshake_receive_wq);
 	destroy_workqueue(wg->handshake_send_wq);
-	free_percpu(wg->decrypt_queue.worker);
-	free_percpu(wg->encrypt_queue.worker);
+	packet_queue_free(&wg->decrypt_queue, true);
+	packet_queue_free(&wg->encrypt_queue, true);
 	destroy_workqueue(wg->packet_crypt_wq);
 	routing_table_free(&wg->peer_routing_table);
 	ratelimiter_uninit();
@@ -293,10 +293,10 @@ static int newlink(struct net *src_net, struct net_device *dev, struct nlattr *t
 	if (!wg->packet_crypt_wq)
 		goto error_5;
 
-	if (packet_queue_init(&wg->encrypt_queue, packet_encrypt_worker, true) < 0)
+	if (packet_queue_init(&wg->encrypt_queue, packet_encrypt_worker, true, MAX_QUEUED_PACKETS) < 0)
 		goto error_6;
 
-	if (packet_queue_init(&wg->decrypt_queue, packet_decrypt_worker, true) < 0)
+	if (packet_queue_init(&wg->decrypt_queue, packet_decrypt_worker, true, MAX_QUEUED_PACKETS) < 0)
 		goto error_7;
 
 	ret = ratelimiter_init();
@@ -319,9 +319,9 @@ static int newlink(struct net *src_net, struct net_device *dev, struct nlattr *t
 error_9:
 	ratelimiter_uninit();
 error_8:
-	free_percpu(wg->decrypt_queue.worker);
+	packet_queue_free(&wg->decrypt_queue, true);
 error_7:
-	free_percpu(wg->encrypt_queue.worker);
+	packet_queue_free(&wg->encrypt_queue, true);
 error_6:
 	destroy_workqueue(wg->packet_crypt_wq);
 error_5:
diff --git a/src/device.h b/src/device.h
index 538a14e..4932477 100644
--- a/src/device.h
+++ b/src/device.h
@@ -13,6 +13,7 @@
 #include <linux/workqueue.h>
 #include <linux/mutex.h>
 #include <linux/net.h>
+#include <linux/ptr_ring.h>
 
 struct wireguard_device;
 
@@ -22,9 +23,7 @@ struct multicore_worker {
 };
 
 struct crypt_queue {
-	spinlock_t lock;
-	int len;
-	struct list_head queue;
+	struct ptr_ring ring;
 	union {
 		struct {
 			struct multicore_worker __percpu *worker;
diff --git a/src/peer.c b/src/peer.c
index 7357db5..4408201 100644
--- a/src/peer.c
+++ b/src/peer.c
@@ -46,8 +46,8 @@ struct wireguard_peer *peer_create(struct wireguard_device *wg, const u8 public_
 	INIT_WORK(&peer->transmit_handshake_work, packet_handshake_send_worker);
 	rwlock_init(&peer->endpoint_lock);
 	kref_init(&peer->refcount);
-	packet_queue_init(&peer->tx_queue, packet_tx_worker, false);
-	packet_queue_init(&peer->rx_queue, packet_rx_worker, false);
+	packet_queue_init(&peer->tx_queue, packet_tx_worker, false, MAX_QUEUED_PACKETS);
+	packet_queue_init(&peer->rx_queue, packet_rx_worker, false, MAX_QUEUED_PACKETS);
 	skb_queue_head_init(&peer->staged_packet_queue);
 	list_add_tail(&peer->peer_list, &wg->peer_list);
 	pubkey_hashtable_add(&wg->peer_hashtable, peer);
@@ -97,6 +97,8 @@ static void rcu_release(struct rcu_head *rcu)
 	struct wireguard_peer *peer = container_of(rcu, struct wireguard_peer, rcu);
 	pr_debug("%s: Peer %Lu (%pISpfsc) destroyed\n", peer->device->dev->name, peer->internal_id, &peer->endpoint.addr);
 	dst_cache_destroy(&peer->endpoint_cache);
+	packet_queue_free(&peer->rx_queue, false);
+	packet_queue_free(&peer->tx_queue, false);
 	kzfree(peer);
 }
 
diff --git a/src/queueing.c b/src/queueing.c
index 2e00d63..f1ae4f1 100644
--- a/src/queueing.c
+++ b/src/queueing.c
@@ -20,11 +20,14 @@ struct multicore_worker __percpu *packet_alloc_percpu_multicore_worker(work_func
 	return worker;
 }
 
-int packet_queue_init(struct crypt_queue *queue, work_func_t function, bool multicore)
+int packet_queue_init(struct crypt_queue *queue, work_func_t function, bool multicore, unsigned int len)
 {
-	INIT_LIST_HEAD(&queue->queue);
-	queue->len = 0;
-	spin_lock_init(&queue->lock);
+	int ret;
+
+	memset(queue, 0, sizeof(*queue));
+	ret = ptr_ring_init(&queue->ring, len, GFP_KERNEL);
+	if (ret)
+		return ret;
 	if (multicore) {
 		queue->worker = packet_alloc_percpu_multicore_worker(function, queue);
 		if (!queue->worker)
@@ -34,6 +37,14 @@ int packet_queue_init(struct crypt_queue *queue, work_func_t function, bool mult
 	return 0;
 }
 
+void packet_queue_free(struct crypt_queue *queue, bool multicore)
+{
+	if (multicore)
+		free_percpu(queue->worker);
+	WARN_ON(!ptr_ring_empty_bh(&queue->ring));
+	ptr_ring_cleanup(&queue->ring, NULL);
+}
+
 int __init crypt_ctx_cache_init(void)
 {
 	crypt_ctx_cache = KMEM_CACHE(crypt_ctx, 0);
diff --git a/src/queueing.h b/src/queueing.h
index 503445c..68640b7 100644
--- a/src/queueing.h
+++ b/src/queueing.h
@@ -19,7 +19,8 @@ struct sk_buff;
 extern struct kmem_cache *crypt_ctx_cache __read_mostly;
 int crypt_ctx_cache_init(void);
 void crypt_ctx_cache_uninit(void);
-int packet_queue_init(struct crypt_queue *queue, work_func_t function, bool multicore);
+int packet_queue_init(struct crypt_queue *queue, work_func_t function, bool multicore, unsigned int len);
+void packet_queue_free(struct crypt_queue *queue, bool multicore);
 struct multicore_worker __percpu *packet_alloc_percpu_multicore_worker(work_func_t function, void *ptr);
 
 /* receive.c APIs: */
@@ -47,7 +48,6 @@ struct packet_cb {
 #define PACKET_CB(skb) ((struct packet_cb *)skb->cb)
 
 struct crypt_ctx {
-	struct list_head per_peer_node, per_device_node;
 	union {
 		struct sk_buff_head packets;
 		struct sk_buff *skb;
@@ -131,57 +131,17 @@ static inline int cpumask_next_online(int *next)
 	return cpu;
 }
 
-static inline struct list_head *queue_dequeue(struct crypt_queue *queue)
-{
-	struct list_head *node;
-
-	spin_lock_bh(&queue->lock);
-	node = queue->queue.next;
-	if (&queue->queue == node) {
-		spin_unlock_bh(&queue->lock);
-		return NULL;
-	}
-	list_del(node);
-	--queue->len;
-	spin_unlock_bh(&queue->lock);
-	return node;
-}
-
-static inline bool queue_enqueue(struct crypt_queue *queue, struct list_head *node, int limit)
-{
-	spin_lock_bh(&queue->lock);
-	if (limit && queue->len >= limit) {
-		spin_unlock_bh(&queue->lock);
-		return false;
-	}
-	list_add_tail(node, &queue->queue);
-	++queue->len;
-	spin_unlock_bh(&queue->lock);
-	return true;
-}
-
-static inline struct crypt_ctx *queue_dequeue_per_device(struct crypt_queue *queue)
-{
-	struct list_head *node = queue_dequeue(queue);
-
-	return node ? list_entry(node, struct crypt_ctx, per_device_node) : NULL;
-}
-
-static inline struct crypt_ctx *queue_first_per_peer(struct crypt_queue *queue)
-{
-	return list_first_entry_or_null(&queue->queue, struct crypt_ctx, per_peer_node);
-}
-
-static inline bool queue_enqueue_per_device_and_peer(struct crypt_queue *device_queue, struct crypt_queue *peer_queue, struct crypt_ctx *ctx, struct workqueue_struct *wq, int *next_cpu)
+static inline int queue_enqueue_per_device_and_peer(struct crypt_queue *device_queue, struct crypt_queue *peer_queue, struct crypt_ctx *ctx, struct workqueue_struct *wq, int *next_cpu)
 {
 	int cpu;
 
-	if (unlikely(!queue_enqueue(peer_queue, &ctx->per_peer_node, MAX_QUEUED_PACKETS)))
-		return false;
+	if (unlikely(ptr_ring_produce_bh(&peer_queue->ring, ctx)))
+		return -ENOSPC;
 	cpu = cpumask_next_online(next_cpu);
-	queue_enqueue(device_queue, &ctx->per_device_node, 0);
+	if (unlikely(ptr_ring_produce_bh(&device_queue->ring, ctx)))
+		return -EPIPE;
 	queue_work_on(cpu, wq, &per_cpu_ptr(device_queue->worker, cpu)->work);
-	return true;
+	return 0;
 }
 
 #ifdef DEBUG
diff --git a/src/receive.c b/src/receive.c
index 504eb55..25349f9 100644
--- a/src/receive.c
+++ b/src/receive.c
@@ -353,8 +353,9 @@ void packet_rx_worker(struct work_struct *work)
 	struct crypt_queue *queue = container_of(work, struct crypt_queue, work);
 
 	local_bh_disable();
-	while ((ctx = queue_first_per_peer(queue)) != NULL && atomic_read(&ctx->is_finished)) {
-		queue_dequeue(queue);
+	spin_lock_bh(&queue->ring.consumer_lock);
+	while ((ctx = __ptr_ring_peek(&queue->ring)) != NULL && atomic_read(&ctx->is_finished)) {
+		__ptr_ring_discard_one(&queue->ring);
 		if (likely(ctx->skb)) {
 			if (likely(counter_validate(&ctx->keypair->receiving.counter, PACKET_CB(ctx->skb)->nonce))) {
 				skb_reset(ctx->skb);
@@ -369,6 +370,7 @@ void packet_rx_worker(struct work_struct *work)
 		peer_put(ctx->peer);
 		kmem_cache_free(crypt_ctx_cache, ctx);
 	}
+	spin_unlock_bh(&queue->ring.consumer_lock);
 	local_bh_enable();
 }
 
@@ -378,7 +380,7 @@ void packet_decrypt_worker(struct work_struct *work)
 	struct crypt_queue *queue = container_of(work, struct multicore_worker, work)->ptr;
 	struct wireguard_peer *peer;
 
-	while ((ctx = queue_dequeue_per_device(queue)) != NULL) {
+	while ((ctx = ptr_ring_consume_bh(&queue->ring)) != NULL) {
 		if (unlikely(socket_endpoint_from_skb(&ctx->endpoint, ctx->skb) < 0 || !skb_decrypt(ctx->skb, &ctx->keypair->receiving))) {
 			dev_kfree_skb(ctx->skb);
 			ctx->skb = NULL;
@@ -397,6 +399,7 @@ static void packet_consume_data(struct wireguard_device *wg, struct sk_buff *skb
 	struct crypt_ctx *ctx;
 	struct noise_keypair *keypair;
 	__le32 idx = ((struct message_data *)skb->data)->key_idx;
+	int ret;
 
 	rcu_read_lock_bh();
 	keypair = noise_keypair_get((struct noise_keypair *)index_hashtable_lookup(&wg->index_hashtable, INDEX_HASHTABLE_KEYPAIR, idx));
@@ -419,13 +422,20 @@ static void packet_consume_data(struct wireguard_device *wg, struct sk_buff *skb
 	/* We already have a reference to peer from index_hashtable_lookup. */
 	ctx->peer = ctx->keypair->entry.peer;
 
-	if (likely(queue_enqueue_per_device_and_peer(&wg->decrypt_queue, &ctx->peer->rx_queue, ctx, wg->packet_crypt_wq, &wg->decrypt_queue.last_cpu)))
+	ret = queue_enqueue_per_device_and_peer(&wg->decrypt_queue, &ctx->peer->rx_queue, ctx, wg->packet_crypt_wq, &wg->decrypt_queue.last_cpu);
+	if (likely(!ret))
 		return; /* Successful. No need to drop references below. */
 
-	noise_keypair_put(ctx->keypair);
-	peer_put(ctx->peer);
 	dev_kfree_skb(ctx->skb);
-	kmem_cache_free(crypt_ctx_cache, ctx);
+	if (ret == -EPIPE) {
+		ctx->skb = NULL;
+		atomic_set(&ctx->is_finished, true);
+		queue_work_on(cpumask_choose_online(&ctx->peer->serial_work_cpu, ctx->peer->internal_id), ctx->peer->device->packet_crypt_wq, &ctx->peer->rx_queue.work);
+	} else {
+		noise_keypair_put(ctx->keypair);
+		peer_put(ctx->peer);
+		kmem_cache_free(crypt_ctx_cache, ctx);
+	}
 }
 
 void packet_receive(struct wireguard_device *wg, struct sk_buff *skb)
diff --git a/src/send.c b/src/send.c
index d1fe150..016e438 100644
--- a/src/send.c
+++ b/src/send.c
@@ -190,12 +190,15 @@ void packet_tx_worker(struct work_struct *work)
 	struct crypt_queue *queue = container_of(work, struct crypt_queue, work);
 	struct crypt_ctx *ctx;
 
-	while ((ctx = queue_first_per_peer(queue)) != NULL && atomic_read(&ctx->is_finished)) {
-		queue_dequeue(queue);
+	spin_lock_bh(&queue->ring.consumer_lock);
+	while ((ctx = __ptr_ring_peek(&queue->ring)) != NULL && atomic_read(&ctx->is_finished)) {
+		__ptr_ring_discard_one(&queue->ring);
 		packet_create_data_done(&ctx->packets, ctx->peer);
+		noise_keypair_put(ctx->keypair);
 		peer_put(ctx->peer);
 		kmem_cache_free(crypt_ctx_cache, ctx);
 	}
+	spin_unlock_bh(&queue->ring.consumer_lock);
 }
 
 void packet_encrypt_worker(struct work_struct *work)
@@ -206,7 +209,7 @@ void packet_encrypt_worker(struct work_struct *work)
 	struct wireguard_peer *peer;
 	bool have_simd = chacha20poly1305_init_simd();
 
-	while ((ctx = queue_dequeue_per_device(queue)) != NULL) {
+	while ((ctx = ptr_ring_consume_bh(&queue->ring)) != NULL) {
 		skb_queue_walk_safe(&ctx->packets, skb, tmp) {
 			if (likely(skb_encrypt(skb, ctx->keypair, have_simd))) {
 				skb_reset(skb);
@@ -215,7 +218,6 @@ void packet_encrypt_worker(struct work_struct *work)
 				dev_kfree_skb(skb);
 			}
 		}
-		noise_keypair_put(ctx->keypair);
 		/* Dereferencing ctx is unsafe once ctx->is_finished == true, so
 		 * we grab an additional reference to peer. */
 		peer = peer_rcu_get(ctx->peer);
@@ -230,6 +232,7 @@ static void packet_create_data(struct wireguard_peer *peer, struct sk_buff_head
 {
 	struct crypt_ctx *ctx;
 	struct wireguard_device *wg = peer->device;
+	int ret;
 
 	ctx = kmem_cache_alloc(crypt_ctx_cache, GFP_ATOMIC);
 	if (unlikely(!ctx)) {
@@ -242,11 +245,17 @@ static void packet_create_data(struct wireguard_peer *peer, struct sk_buff_head
 	ctx->peer = peer;
 	__skb_queue_head_init(&ctx->packets);
 	skb_queue_splice_tail(packets, &ctx->packets);
-	if (likely(queue_enqueue_per_device_and_peer(&wg->encrypt_queue, &peer->tx_queue, ctx, wg->packet_crypt_wq, &wg->encrypt_queue.last_cpu)))
+	ret = queue_enqueue_per_device_and_peer(&wg->encrypt_queue, &peer->tx_queue, ctx, wg->packet_crypt_wq, &wg->encrypt_queue.last_cpu);
+	if (likely(!ret))
 		return; /* Successful. No need to fall through to drop references below. */
 
 	__skb_queue_purge(&ctx->packets);
-	kmem_cache_free(crypt_ctx_cache, ctx);
+	if (ret == -EPIPE) {
+		atomic_set(&ctx->is_finished, true);
+		queue_work_on(cpumask_choose_online(&peer->serial_work_cpu, peer->internal_id), peer->device->packet_crypt_wq, &peer->tx_queue.work);
+		return;
+	} else
+		kmem_cache_free(crypt_ctx_cache, ctx);
 
 err_drop_refs:
 	noise_keypair_put(keypair);