1/* @title: Multi-Producer Multi-Consumer FIFO Ring */
2#pragma once
3#include <atomic.h>
4#include <compiler/core.h>
5#include <stdbool.h>
6#include <stddef.h>
7#include <stdint.h>
8
9/* Bounded Vyukov MPMC FIFO ring, like
10 *
11 * MPMC_FIFO_RING_DECLARE(foo_ring, struct foo);
12 *
13 * emits `struct foo_ring` and foo_ring_{init,destroy,enqueue,dequeue,
14 * empty,capacity,drain,produced,consumed} */
15
16struct mpmc_fifo_ring_hdr {
17 ca_cacheline_member(size_t, mask); /* capacity - 1, a power of two */
18 ca_cacheline_member(atomic_uint64_t, head);
19 atomic_uint64_t tail;
20};
21
22/* `cap` rounds up to next_pow2 */
23void *mpmc_fifo_ring_slots_alloc(struct mpmc_fifo_ring_hdr *h, size_t cap,
24 size_t stride);
25void mpmc_fifo_ring_slots_free(struct mpmc_fifo_ring_hdr *h, void *slots);
26
27static inline cc_always_inline bool
28mpmc_fifo_ring_claim(atomic_uint64_t *cursor, size_t mask, void *slots,
29 size_t stride, uint64_t lag, uint64_t *pos_out) {
30 while (true) {
31 uint64_t pos = atomic_load_rlx(cursor);
32 atomic_uint64_t *seq =
33 (atomic_uint64_t *) ((uint8_t *) slots + (pos & mask) * stride);
34 int64_t diff = (int64_t) atomic_load_acq(seq) - (int64_t) (pos + lag);
35
36 if (diff == 0) {
37 if (atomic_cas_weak(cursor, &pos, pos + 1, mo_acq_rel, mo_rlx)) {
38 *pos_out = pos;
39 return true;
40 }
41 } else if (diff < 0) {
42 return false;
43 }
44 }
45}
46
47static inline cc_always_inline bool
48mpmc_fifo_ring_claim_range(struct mpmc_fifo_ring_hdr *h, void *slots,
49 size_t stride, uint64_t *first_out,
50 uint64_t *count_out) {
51 while (true) {
52 uint64_t t = atomic_load_rlx(&h->tail);
53 uint64_t end = atomic_load_acq(&h->head);
54
55 uint64_t p = t;
56 while (p != end) {
57 atomic_uint64_t *seq = (atomic_uint64_t *) ((uint8_t *) slots +
58 (p & h->mask) * stride);
59 if (atomic_load_acq(seq) != p + 1)
60 break;
61 p++;
62 }
63
64 if (p == t) {
65 if (atomic_load_rlx(&h->tail) != t)
66 continue;
67 return false;
68 }
69
70 uint64_t expected = t;
71 if (atomic_cas_weak(&h->tail, &expected, p, mo_acq_rel, mo_rlx)) {
72 *first_out = t;
73 *count_out = p - t;
74 return true;
75 }
76 }
77}
78
79static inline bool
80mpmc_fifo_ring_hdr_empty(const struct mpmc_fifo_ring_hdr *h) {
81 uint64_t hd = atomic_load_rlx(&h->head);
82 uint64_t tl = atomic_load_rlx(&h->tail);
83 return hd == tl;
84}
85
86#define MPMC_FIFO_RING_DECLARE(name, T) \
87 struct name##_slot { \
88 atomic_uint64_t seq; \
89 T val; \
90 }; \
91 \
92 struct name { \
93 struct name##_slot *slots; \
94 struct mpmc_fifo_ring_hdr hdr; \
95 }; \
96 \
97 static inline bool name##_init(struct name *q, size_t cap) { \
98 q->slots = \
99 mpmc_fifo_ring_slots_alloc(&q->hdr, cap, sizeof(*q->slots)); \
100 return q->slots != NULL; \
101 } \
102 \
103 static inline void name##_destroy(struct name *q) { \
104 mpmc_fifo_ring_slots_free(&q->hdr, q->slots); \
105 q->slots = NULL; \
106 } \
107 \
108 static inline bool name##_enqueue(struct name *q, T v) { \
109 uint64_t pos; \
110 if (!mpmc_fifo_ring_claim(&q->hdr.head, q->hdr.mask, q->slots, \
111 sizeof(*q->slots), 0, &pos)) \
112 return false; \
113 \
114 struct name##_slot *s = &q->slots[pos & q->hdr.mask]; \
115 s->val = v; \
116 atomic_store_release(&s->seq, pos + 1); \
117 return true; \
118 } \
119 \
120 static inline bool name##_dequeue(struct name *q, T *out) { \
121 uint64_t pos; \
122 if (!mpmc_fifo_ring_claim(&q->hdr.tail, q->hdr.mask, q->slots, \
123 sizeof(*q->slots), 1, &pos)) \
124 return false; \
125 \
126 struct name##_slot *s = &q->slots[pos & q->hdr.mask]; \
127 if (out) \
128 *out = s->val; \
129 \
130 s->val = (T) {0}; \
131 atomic_store_release(&s->seq, pos + q->hdr.mask + 1); \
132 return true; \
133 } \
134 \
135 /* Drain and return how many were drained */ \
136 static inline size_t name##_drain(struct name *q) { \
137 uint64_t first, count; \
138 if (!mpmc_fifo_ring_claim_range(&q->hdr, q->slots, sizeof(*q->slots), \
139 &first, &count)) \
140 return 0; \
141 \
142 for (uint64_t pos = first; pos != first + count; pos++) { \
143 struct name##_slot *s = &q->slots[pos & q->hdr.mask]; \
144 s->val = (T) {0}; \
145 atomic_store_release(&s->seq, pos + q->hdr.mask + 1); \
146 } \
147 return count; \
148 } \
149 \
150 static inline bool name##_empty(const struct name *q) { \
151 return mpmc_fifo_ring_hdr_empty(&q->hdr); \
152 } \
153 \
154 static inline size_t name##_capacity(const struct name *q) { \
155 return q->hdr.mask + 1; \
156 } \
157 \
158 static inline cc_unused uint64_t name##_produced(struct name *q) { \
159 return atomic_load_acq(&q->hdr.head); \
160 } \
161 \
162 static inline cc_unused uint64_t name##_consumed(struct name *q) { \
163 return atomic_load_acq(&q->hdr.tail); \
164 } \
165 \
166 static_assert(ct_field_offset(struct name##_slot, seq) == 0, \
167 #name ": seq must lead the slot")
168