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>

Author: almusil

lib/automake.mk

@@ -50,6 +50,8 @@ lib_libovn_la_SOURCES = \
 	lib/lb.h \
 	lib/sparse-array.c \
 	lib/sparse-array.h \
+	lib/spsc-ring.c \
+	lib/spsc-ring.h \
 	lib/stopwatch-names.h \
 	lib/vec.c \
 	lib/vec.h \

lib/spsc-ring.c

@@ -0,0 +1,80 @@
+/* Copyright (c) 2026, Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at:
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <config.h>
+#include <string.h>
+
+#include "spsc-ring.h"
+#include "util.h"
+
+/* Initializes the SPSC ring, the capacity must be power of 2. */
+void
+spsc_ring_init(struct spsc_ring *r, uint32_t capacity, size_t esize)
+{
+    ovs_assert(IS_POW2(capacity));
+    r->buffer = xmalloc(capacity * esize);
+    r->esize = esize;
+    r->mask = capacity - 1;
+    atomic_init(&r->read, 0);
+    atomic_init(&r->write, 0);
+}
+
+void
+spsc_ring_destroy(struct spsc_ring *r)
+{
+    free(r->buffer);
+    r->buffer = NULL;
+}
+
+/* Producer: copy 'data' into the next available slot.
+ * Returns true on success, false if the ring is full. */
+bool
+spsc_ring_push(struct spsc_ring *r, const void *data)
+{
+    uint32_t wr, rd;
+
+    atomic_read(&r->write, &wr);
+    atomic_read(&r->read, &rd);
+
+    if (wr - rd > r->mask) {
+        return false;
+    }
+
+    memcpy((uint8_t *) r->buffer + (wr & r->mask) * r->esize,
+           data, r->esize);
+    atomic_store(&r->write, wr + 1);
+    return true;
+}
+
+/* Consumer: copy the oldest slot's data into 'data'.
+ * Returns true on success, false if the ring is empty. */
+bool
+spsc_ring_pop(struct spsc_ring *r, void *data)
+{
+    uint32_t rd, wr;
+
+    atomic_read(&r->read, &rd);
+    atomic_read(&r->write, &wr);
+
+    if (rd == wr) {
+        return false;
+    }
+
+    memcpy(data,
+           (uint8_t *) r->buffer + (rd & r->mask) * r->esize,
+           r->esize);
+    atomic_store(&r->read, rd + 1);
+    return true;
+}

lib/spsc-ring.h

@@ -0,0 +1,67 @@
+/* Copyright (c) 2026, Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at:
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef SPSC_RING_H
+#define SPSC_RING_H
+
+#include <stdbool.h>
+#include <stdint.h>
+
+#include "openvswitch/util.h"
+#include "ovs-atomic.h"
+
+/* Single-Producer, Single-Consumer lock-free ring buffer.
+ *
+ * Thread-safety
+ * =============
+ *
+ * Exactly one thread (the producer) may call spsc_ring_push().
+ * Exactly one thread (the consumer) may call spsc_ring_pop().
+ * These two threads may be different and may operate concurrently
+ * without any external synchronization.
+ *
+ * Memory ordering: sequential consistency (the default for OVS
+ * atomic_read/atomic_store) on the read and write indices ensures
+ * that slot data written by the producer is visible to the consumer
+ * after it observes the updated write index (and vice versa for
+ * read index updates freeing slots for reuse).
+ *
+ * Slot data is copied in and out by value (memcpy), so this is
+ * best suited for small, fixed-size structs.
+ *
+ * Capacity must be a power of two.  The ring uses unsigned 32-bit
+ * wraparound arithmetic on read/write indices, which is correct as
+ * long as capacity is much smaller than 2^32.
+ */
+struct spsc_ring {
+    void *buffer;           /* Pre-allocated slot array. */
+    size_t esize;           /* Size of each element in bytes. */
+    uint32_t mask;          /* capacity - 1 (for power-of-two modulo). */
+    atomic_uint32_t read;   /* Next slot to consume (advanced by consumer). */
+    atomic_uint32_t write;  /* Next slot to fill (advanced by producer). */
+};
+
+void spsc_ring_init(struct spsc_ring *, uint32_t capacity, size_t esize);
+void spsc_ring_destroy(struct spsc_ring *);
+bool spsc_ring_push(struct spsc_ring *, const void *data);
+bool spsc_ring_pop(struct spsc_ring *, void *data);
+
+/* Pop each element into VAR until the ring is empty. */
+#define SPSC_RING_FOR_EACH_POP(RING, VAR)                          \
+    for (bool ITER_VAR(VAR) = spsc_ring_pop(RING, &(VAR));         \
+         ITER_VAR(VAR);                                            \
+         ITER_VAR(VAR) = spsc_ring_pop(RING, &(VAR)))
+
+#endif /* lib/spsc-ring.h */

tests/automake.mk

@@ -280,6 +280,7 @@ tests_ovstest_SOURCES = \
 	tests/test-utils.h \
 	tests/test-ovn.c \
 	tests/test-sparse-array.c \
+	tests/test-spsc-ring.c \
 	tests/test-vector.c \
 	controller/test-lflow-cache.c \
 	controller/test-vif-plug.c \

tests/ovn.at

@@ -2451,6 +2451,15 @@ check ovstest test-sparse-array add
 check ovstest test-sparse-array remove-replace
 AT_CLEANUP
 
+AT_SETUP([SPSC ring buffer])
+check ovstest test-spsc-ring basic
+check ovstest test-spsc-ring fifo
+check ovstest test-spsc-ring full
+check ovstest test-spsc-ring wraparound
+check ovstest test-spsc-ring for-each-pop
+check ovstest test-spsc-ring struct
+AT_CLEANUP
+
 AT_SETUP([Parse MAC])
 AT_CHECK([ovstest test-ovn parse-eth-addr 01:02:03:04:05:xx], [1])
 AT_CHECK([ovstest test-ovn parse-eth-addr 01:02:03:04:05:06], [0], [dnl