From aa4d07187b66bf83bfddfa6d964f7c141518ba7b Mon Sep 17 00:00:00 2001 From: "Jason A. Donenfeld" Date: Wed, 4 Oct 2017 05:05:51 +0200 Subject: queueing: use ptr_ring instead of linked lists --- src/device.c | 12 ++++++------ src/device.h | 5 ++--- src/peer.c | 6 ++++-- src/queueing.c | 19 +++++++++++++++---- src/queueing.h | 56 ++++++++------------------------------------------------ src/receive.c | 24 +++++++++++++++++------- src/send.c | 21 +++++++++++++++------ 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 #include #include +#include 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); -- cgit v1.2.3-59-g8ed1b