Add spsc_q data structure library

This data structure is designed to support the use cases of the L1,
i.e., provide a queue which can be quickly used to order L1 jobs in FIFO
order with simple iteration of jobs to treat. It is a Single-Consumer
Single-Producer queue library.

Signed-off-by: Robert Schmidt <robert.schmidt@openairinterface.org>

common/utils/ds/CMakeLists.txt

@@ -6,6 +6,7 @@ add_library(ds OBJECT
             seq_arr.c
             hashtable.c
             obj_hashtable.c
+            spsc_q.c
 )
 
 target_include_directories(ds PUBLIC ${CMAKE_CURRENT_SOURCE_DIR})

common/utils/ds/spsc_q.c

@@ -0,0 +1,112 @@
+/*
+ * SPDX-License-Identifier: LicenseRef-CSSL-1.0
+ */
+
+#include <stdlib.h>
+#include <string.h>
+#include <assert.h>
+
+#include "spsc_q.h"
+
+#define rb_lock(LOCK) if (pthread_mutex_lock(LOCK) != 0) abort()
+#define rb_unlock(LOCK) if (pthread_mutex_unlock(LOCK) != 0) abort()
+
+spsc_q_t spsc_q_alloc(size_t cnt, size_t elsiz)
+{
+  /* internally, use one element more: the ringbuffer is full if
+   * (write_idx + 1) % cnt == read_idx, so it's full if one element is still
+   * free (because otherwise write_idx == read_idx, but this is the empty
+   * condition) */
+  cnt += 1;
+  spsc_q_t rb = {.cnt = cnt, .elsiz = elsiz};
+  rb.buf = calloc(cnt, elsiz);
+  pthread_mutex_init(&rb.mtx, NULL);
+  return rb;
+}
+
+void spsc_q_free(spsc_q_t *rb)
+{
+  free(rb->buf);
+  pthread_mutex_destroy(&rb->mtx);
+  memset(rb, 0, sizeof(*rb));
+}
+
+bool spsc_q_put(spsc_q_t *rb, const void *src, size_t elsiz)
+{
+  assert(elsiz == rb->elsiz);
+  rb_lock(&rb->mtx);
+  if ((rb->write_idx + 1) % rb->cnt == rb->read_idx) {
+    rb_unlock(&rb->mtx);
+    return false;
+  }
+
+  uint8_t *bufpos = &rb->buf[rb->write_idx * rb->elsiz];
+  memcpy(bufpos, src, elsiz);
+  rb->write_idx = (rb->write_idx + 1) % rb->cnt;
+  rb_unlock(&rb->mtx);
+  return true;
+}
+
+bool spsc_q_get_if(spsc_q_t *rb, pred p, void *user, void *dest, size_t elsiz)
+{
+  assert(elsiz == rb->elsiz);
+  rb_lock(&rb->mtx);
+  if (rb->write_idx == rb->read_idx) {
+    rb_unlock(&rb->mtx);
+    return false;
+  }
+
+  uint8_t *bufpos = &rb->buf[rb->read_idx * rb->elsiz];
+  if (!p(bufpos, user)) {
+    rb_unlock(&rb->mtx);
+    return false;
+  }
+
+  memcpy(dest, bufpos, elsiz);
+  rb->read_idx = (rb->read_idx + 1) % rb->cnt;
+  rb_unlock(&rb->mtx);
+  return true;
+}
+
+static bool always(const void *data, void *user)
+{
+  (void) data;
+  (void) user;
+  return true;
+}
+bool spsc_q_get(spsc_q_t *rb, void *dest, size_t elsiz)
+{
+  return spsc_q_get_if(rb, always, NULL, dest, elsiz);
+}
+
+int spsc_q_get_while(spsc_q_t *rb, pred p, void *user, void *dest, size_t elsiz, size_t max_len)
+{
+  assert(elsiz == rb->elsiz);
+  rb_lock(&rb->mtx);
+  size_t count = 0;
+  while (rb->write_idx != rb->read_idx && count < max_len) {
+    uint8_t *bufpos = &rb->buf[rb->read_idx * rb->elsiz];
+    if (!p(bufpos, user))
+      break;
+    memcpy(dest + count * elsiz, bufpos, elsiz);
+    count++;
+    rb->read_idx = (rb->read_idx + 1) % rb->cnt;
+  }
+  rb_unlock(&rb->mtx);
+  return count;
+}
+
+int spsc_q_drop_while(spsc_q_t *rb, pred p, void *user)
+{
+  rb_lock(&rb->mtx);
+  size_t dropped = 0;
+  while (rb->write_idx != rb->read_idx) {
+    uint8_t *bufpos = &rb->buf[rb->read_idx * rb->elsiz];
+    if (!p(bufpos, user))
+      break;
+    dropped++;
+    rb->read_idx = (rb->read_idx + 1) % rb->cnt;
+  }
+  rb_unlock(&rb->mtx);
+  return dropped;
+}

common/utils/ds/spsc_q.h

@@ -0,0 +1,32 @@
+/*
+ * SPDX-License-Identifier: LicenseRef-CSSL-1.0
+ */
+
+#ifndef SPSC_Q_H_
+#define SPSC_Q_H_
+
+#include <stdint.h>
+#include <stdbool.h>
+#include <pthread.h>
+
+typedef struct spsc_q {
+  uint8_t *buf;
+  size_t elsiz;
+  size_t cnt;
+  size_t write_idx;
+  size_t read_idx;
+  pthread_mutex_t mtx;
+} spsc_q_t;
+
+spsc_q_t spsc_q_alloc(size_t cnt, size_t elsiz);
+void spsc_q_free(spsc_q_t *rb);
+
+bool spsc_q_put(spsc_q_t *rb, const void *src, size_t elsiz);
+bool spsc_q_get(spsc_q_t *rb, void *dest, size_t elsiz);
+
+typedef bool (*pred)(const void *data, void *user);
+bool spsc_q_get_if(spsc_q_t *rb, pred p, void *user, void *dest, size_t elsiz);
+int spsc_q_drop_while(spsc_q_t *rb, pred p, void *user);
+int spsc_q_get_while(spsc_q_t *rb, pred p, void *user, void *dest, size_t elsiz, size_t max_len);
+
+#endif /* SPSC_Q_H_ */

common/utils/ds/tests/CMakeLists.txt

@@ -8,5 +8,11 @@ add_test(NAME test_seq_arr COMMAND test_seq_arr) # no options required
 add_executable(test_hashtable test_hashtable.cpp)
 add_dependencies(tests test_hashtable)
 target_link_libraries(test_hashtable PRIVATE ds GTest::gtest)
-add_test(NAME test_hashtable
-  COMMAND ./test_hashtable)
+add_test(NAME test_hashtable COMMAND ./test_hashtable)
+
+add_executable(test_spsc_q test_spsc_q.cpp)
+target_link_libraries(test_spsc_q ds GTest::gtest)
+add_dependencies(tests test_spsc_q)
+add_test(NAME test_spsc_q COMMAND test_spsc_q)
+add_executable(test_spsc_q_perf test_spsc_q_perf.cpp)
+target_link_libraries(test_spsc_q_perf PRIVATE ds benchmark::benchmark)

common/utils/ds/tests/test_spsc_q.cpp

@@ -0,0 +1,135 @@
+/*
+ * SPDX-License-Identifier: LicenseRef-CSSL-1.0
+ */
+
+#include "gtest/gtest.h"
+extern "C" {
+#include "spsc_q.h"
+}
+
+TEST(spsc_q, basic_test) {
+  spsc_q_t rb = spsc_q_alloc(2, sizeof(int));
+
+  int a = 1;
+  EXPECT_TRUE(spsc_q_put(&rb, &a, sizeof(a)));
+  EXPECT_TRUE(spsc_q_put(&rb, &a, sizeof(a)));
+  EXPECT_FALSE(spsc_q_put(&rb, &a, sizeof(a)));
+
+  int b;
+  EXPECT_TRUE(spsc_q_get(&rb, &b, sizeof(b)));
+  EXPECT_EQ(a, b);
+  EXPECT_TRUE(spsc_q_get(&rb, &b, sizeof(b)));
+  EXPECT_EQ(a, b);
+  EXPECT_FALSE(spsc_q_get(&rb, &b, sizeof(b)));
+
+  spsc_q_free(&rb);
+}
+
+TEST(spsc_q, cont_write) {
+  spsc_q_t rb = spsc_q_alloc(10, sizeof(int));
+
+  int a = 1;
+  for (int i = 0; i < 1111; ++i) {
+    EXPECT_TRUE(spsc_q_put(&rb, &a, sizeof(a)));
+    a++;
+    EXPECT_TRUE(spsc_q_put(&rb, &a, sizeof(a)));
+    a++;
+    EXPECT_TRUE(spsc_q_put(&rb, &a, sizeof(a)));
+    a++;
+
+    int b;
+    EXPECT_TRUE(spsc_q_get(&rb, &b, sizeof(b)));
+    EXPECT_EQ(b, a - 3);
+    EXPECT_TRUE(spsc_q_get(&rb, &b, sizeof(b)));
+    EXPECT_EQ(b, a - 2);
+    EXPECT_TRUE(spsc_q_get(&rb, &b, sizeof(b)));
+    EXPECT_EQ(b, a - 1);
+  }
+
+  spsc_q_free(&rb);
+}
+
+bool count(const void *data, void *user)
+{
+  EXPECT_TRUE(data != NULL);
+  int *c = (int *)user;
+  (*c)++;
+  return true; /* count all */
+}
+TEST(spsc_q, iterator) {
+  spsc_q_t rb = spsc_q_alloc(10, sizeof(int));
+  int n = 8;
+  for (int i = 0; i < n; ++i)
+    EXPECT_TRUE(spsc_q_put(&rb, &i, sizeof(i)));
+
+  int c = 0;
+  int results[n];
+  int ret = spsc_q_get_while(&rb, count, &c, results, sizeof(*results), n);
+  EXPECT_EQ(c, n);
+  EXPECT_EQ(ret, n);
+  for (int i = 0; i < n; ++i)
+    EXPECT_EQ(results[i], i);
+
+  int b;
+  EXPECT_FALSE(spsc_q_get(&rb, &b, sizeof(b))); /* empty! */
+
+  spsc_q_free(&rb);
+}
+
+bool drop(const void *data, void *user)
+{
+  const int *e = (const int *)data;
+  int *c = (int *)user;
+  EXPECT_EQ(*e, *c);
+  (*c)++;
+  return true; /* drop all */
+}
+TEST(spsc_q, drop) {
+  spsc_q_t rb = spsc_q_alloc(10, sizeof(int));
+  int n = 8;
+  for (int i = 0; i < n; ++i)
+    EXPECT_TRUE(spsc_q_put(&rb, &i, sizeof(i)));
+
+  int c = 0;
+  int ret = spsc_q_drop_while(&rb, drop, &c);
+  EXPECT_EQ(c, n);
+  EXPECT_EQ(ret, n);
+
+  int b;
+  EXPECT_FALSE(spsc_q_get(&rb, &b, sizeof(b))); /* empty! */
+
+  spsc_q_free(&rb);
+}
+
+
+TEST(spsc_q, full) {
+  int n = 3;
+  spsc_q_t rb = spsc_q_alloc(n, sizeof(int));
+  int a = 1;
+  EXPECT_TRUE(spsc_q_put(&rb, &a, sizeof(a)));
+  a++;
+  EXPECT_TRUE(spsc_q_put(&rb, &a, sizeof(a)));
+  a++;
+  EXPECT_TRUE(spsc_q_put(&rb, &a, sizeof(a)));
+  a++;
+  EXPECT_FALSE(spsc_q_put(&rb, &a, sizeof(a))); /* full! */
+
+  int b;
+  EXPECT_TRUE(spsc_q_get(&rb, &b, sizeof(b)));
+  EXPECT_EQ(b, a - n);
+
+  EXPECT_TRUE(spsc_q_put(&rb, &a, sizeof(a)));
+  a++;
+  EXPECT_FALSE(spsc_q_put(&rb, &a, sizeof(a))); /* full! */
+
+  EXPECT_TRUE(spsc_q_get(&rb, &b, sizeof(b)));
+  EXPECT_EQ(b, a - n);
+
+  spsc_q_free(&rb);
+}
+
+int main(int argc, char** argv)
+{
+  testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}

common/utils/ds/tests/test_spsc_q_perf.cpp

@@ -0,0 +1,75 @@
+#include "benchmark/benchmark.h"
+#include "stdio.h"
+#include "assert.h"
+#include "pthread.h"
+
+extern "C" {
+#include "spsc_q.h"
+}
+
+typedef struct q_args {
+  spsc_q_t q;
+  int range;
+} q_args_t;
+
+static void *producer(void *arg)
+{
+  q_args_t *a = (q_args_t *) arg;
+  int i = 0;
+  while (i <= a->range) {
+    if (spsc_q_put(&a->q, &i, sizeof(i)))
+      i++;
+  }
+  return NULL;
+}
+
+static void *consumer(void *arg)
+{
+  q_args_t *a = (q_args_t *) arg;
+  int i = 0;
+  while (i != a->range) {
+    int r;
+    if (spsc_q_get(&a->q, &r, sizeof(r))) {
+      assert(r == i);
+      i++;
+    }
+  }
+  return NULL;
+}
+
+static void run_test(int range, int q_size)
+{
+  q_args_t args = {
+    .q = spsc_q_alloc(q_size, sizeof(int)),
+    .range = range,
+  };
+
+  pthread_t p, c;
+  int ret;
+  // start
+  ret = pthread_create(&p, NULL, producer, &args);
+  assert(ret == 0);
+  ret = pthread_create(&c, NULL, consumer, &args);
+  assert(ret == 0);
+
+  ret = pthread_join(p, NULL);
+  assert(ret == 0);
+  ret = pthread_join(c, NULL);
+  assert(ret == 0);
+  // stop
+  // data/time => throughput
+
+  spsc_q_free(&args.q);
+}
+
+static void BM_spsc_q(benchmark::State &state)
+{
+  int range = 500000;
+  int q_size = state.range(0);
+  for (auto _ : state)
+    run_test(range, q_size);
+}
+
+BENCHMARK(BM_spsc_q)->RangeMultiplier(2)->Range(10, 160);
+
+BENCHMARK_MAIN();