Skip to content

Commit b8cb977

Browse files
almusilalmusil
committed
spsc-ring: Add lock-free SPSC ring buffer.
Add a Single-Producer, Single-Consumer lock-free ring buffer to lib/ for use as a building block in removing pinctrl_mutex contention from packet-in processing. The ring buffer uses a pre-allocated array with atomic read and write indices for synchronization. Data is copied in and out by value (memcpy), making it well suited for small, fixed-size structs. Capacity must be a power of two; the implementation uses unsigned 32-bit wraparound arithmetic for correct index handling. Provide SPSC_RING_FOR_EACH_POP() for idiomatic drain loops. Include unit tests covering basic push/pop, FIFO ordering, full ring behavior, index wraparound, the FOR_EACH_POP macro, and multi-field struct elements. Assisted-by: Claude Opus 4.6, OpenCode Acked-by: Dumitru Ceara <dceara@redhat.com> Signed-off-by: Ales Musil <amusil@redhat.com> (cherry picked from commit ce5a8cd)
1 parent 608aa1c commit b8cb977

6 files changed

Lines changed: 383 additions & 0 deletions

File tree

‎lib/automake.mk‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,8 @@ lib_libovn_la_SOURCES = \
5050
lib/lb.h \
5151
lib/sparse-array.c \
5252
lib/sparse-array.h \
53+
lib/spsc-ring.c \
54+
lib/spsc-ring.h \
5355
lib/stopwatch-names.h \
5456
lib/vec.c \
5557
lib/vec.h \

‎lib/spsc-ring.c‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
/* Copyright (c) 2026, Red Hat, Inc.
2+
*
3+
* Licensed under the Apache License, Version 2.0 (the "License");
4+
* you may not use this file except in compliance with the License.
5+
* You may obtain a copy of the License at:
6+
*
7+
* http://www.apache.org/licenses/LICENSE-2.0
8+
*
9+
* Unless required by applicable law or agreed to in writing, software
10+
* distributed under the License is distributed on an "AS IS" BASIS,
11+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
* See the License for the specific language governing permissions and
13+
* limitations under the License.
14+
*/
15+
16+
#include <config.h>
17+
#include <string.h>
18+
19+
#include "spsc-ring.h"
20+
#include "util.h"
21+
22+
/* Initializes the SPSC ring, the capacity must be power of 2. */
23+
void
24+
spsc_ring_init(struct spsc_ring *r, uint32_t capacity, size_t esize)
25+
{
26+
ovs_assert(IS_POW2(capacity));
27+
r->buffer = xmalloc(capacity * esize);
28+
r->esize = esize;
29+
r->mask = capacity - 1;
30+
atomic_init(&r->read, 0);
31+
atomic_init(&r->write, 0);
32+
}
33+
34+
void
35+
spsc_ring_destroy(struct spsc_ring *r)
36+
{
37+
free(r->buffer);
38+
r->buffer = NULL;
39+
}
40+
41+
/* Producer: copy 'data' into the next available slot.
42+
* Returns true on success, false if the ring is full. */
43+
bool
44+
spsc_ring_push(struct spsc_ring *r, const void *data)
45+
{
46+
uint32_t wr, rd;
47+
48+
atomic_read(&r->write, &wr);
49+
atomic_read(&r->read, &rd);
50+
51+
if (wr - rd > r->mask) {
52+
return false;
53+
}
54+
55+
memcpy((uint8_t *) r->buffer + (wr & r->mask) * r->esize,
56+
data, r->esize);
57+
atomic_store(&r->write, wr + 1);
58+
return true;
59+
}
60+
61+
/* Consumer: copy the oldest slot's data into 'data'.
62+
* Returns true on success, false if the ring is empty. */
63+
bool
64+
spsc_ring_pop(struct spsc_ring *r, void *data)
65+
{
66+
uint32_t rd, wr;
67+
68+
atomic_read(&r->read, &rd);
69+
atomic_read(&r->write, &wr);
70+
71+
if (rd == wr) {
72+
return false;
73+
}
74+
75+
memcpy(data,
76+
(uint8_t *) r->buffer + (rd & r->mask) * r->esize,
77+
r->esize);
78+
atomic_store(&r->read, rd + 1);
79+
return true;
80+
}

‎lib/spsc-ring.h‎

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,67 @@
1+
/* Copyright (c) 2026, Red Hat, Inc.
2+
*
3+
* Licensed under the Apache License, Version 2.0 (the "License");
4+
* you may not use this file except in compliance with the License.
5+
* You may obtain a copy of the License at:
6+
*
7+
* http://www.apache.org/licenses/LICENSE-2.0
8+
*
9+
* Unless required by applicable law or agreed to in writing, software
10+
* distributed under the License is distributed on an "AS IS" BASIS,
11+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
* See the License for the specific language governing permissions and
13+
* limitations under the License.
14+
*/
15+
16+
#ifndef SPSC_RING_H
17+
#define SPSC_RING_H
18+
19+
#include <stdbool.h>
20+
#include <stdint.h>
21+
22+
#include "openvswitch/util.h"
23+
#include "ovs-atomic.h"
24+
25+
/* Single-Producer, Single-Consumer lock-free ring buffer.
26+
*
27+
* Thread-safety
28+
* =============
29+
*
30+
* Exactly one thread (the producer) may call spsc_ring_push().
31+
* Exactly one thread (the consumer) may call spsc_ring_pop().
32+
* These two threads may be different and may operate concurrently
33+
* without any external synchronization.
34+
*
35+
* Memory ordering: sequential consistency (the default for OVS
36+
* atomic_read/atomic_store) on the read and write indices ensures
37+
* that slot data written by the producer is visible to the consumer
38+
* after it observes the updated write index (and vice versa for
39+
* read index updates freeing slots for reuse).
40+
*
41+
* Slot data is copied in and out by value (memcpy), so this is
42+
* best suited for small, fixed-size structs.
43+
*
44+
* Capacity must be a power of two. The ring uses unsigned 32-bit
45+
* wraparound arithmetic on read/write indices, which is correct as
46+
* long as capacity is much smaller than 2^32.
47+
*/
48+
struct spsc_ring {
49+
void *buffer; /* Pre-allocated slot array. */
50+
size_t esize; /* Size of each element in bytes. */
51+
uint32_t mask; /* capacity - 1 (for power-of-two modulo). */
52+
atomic_uint32_t read; /* Next slot to consume (advanced by consumer). */
53+
atomic_uint32_t write; /* Next slot to fill (advanced by producer). */
54+
};
55+
56+
void spsc_ring_init(struct spsc_ring *, uint32_t capacity, size_t esize);
57+
void spsc_ring_destroy(struct spsc_ring *);
58+
bool spsc_ring_push(struct spsc_ring *, const void *data);
59+
bool spsc_ring_pop(struct spsc_ring *, void *data);
60+
61+
/* Pop each element into VAR until the ring is empty. */
62+
#define SPSC_RING_FOR_EACH_POP(RING, VAR) \
63+
for (bool ITER_VAR(VAR) = spsc_ring_pop(RING, &(VAR)); \
64+
ITER_VAR(VAR); \
65+
ITER_VAR(VAR) = spsc_ring_pop(RING, &(VAR)))
66+
67+
#endif /* lib/spsc-ring.h */

‎tests/automake.mk‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -289,6 +289,7 @@ tests_ovstest_SOURCES = \
289289
tests/test-utils.h \
290290
tests/test-ovn.c \
291291
tests/test-sparse-array.c \
292+
tests/test-spsc-ring.c \
292293
tests/test-vector.c \
293294
controller/test-lflow-cache.c \
294295
controller/test-vif-plug.c \

‎tests/ovn.at‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2451,6 +2451,15 @@ check ovstest test-sparse-array add
24512451
check ovstest test-sparse-array remove-replace
24522452
AT_CLEANUP
24532453

2454+
AT_SETUP([SPSC ring buffer])
2455+
check ovstest test-spsc-ring basic
2456+
check ovstest test-spsc-ring fifo
2457+
check ovstest test-spsc-ring full
2458+
check ovstest test-spsc-ring wraparound
2459+
check ovstest test-spsc-ring for-each-pop
2460+
check ovstest test-spsc-ring struct
2461+
AT_CLEANUP
2462+
24542463
AT_SETUP([Parse MAC])
24552464
AT_CHECK([ovstest test-ovn parse-eth-addr 01:02:03:04:05:xx], [1])
24562465
AT_CHECK([ovstest test-ovn parse-eth-addr 01:02:03:04:05:06], [0], [dnl

0 commit comments

Comments
 (0)