diff --git a/common/utils/ds/CMakeLists.txt b/common/utils/ds/CMakeLists.txt index 7007d77fac..6f7053a1be 100644 --- a/common/utils/ds/CMakeLists.txt +++ b/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}) diff --git a/common/utils/ds/spsc_q.c b/common/utils/ds/spsc_q.c new file mode 100644 index 0000000000..fdc97f00b8 --- /dev/null +++ b/common/utils/ds/spsc_q.c @@ -0,0 +1,112 @@ +/* + * SPDX-License-Identifier: LicenseRef-CSSL-1.0 + */ + +#include +#include +#include + +#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; +} diff --git a/common/utils/ds/spsc_q.h b/common/utils/ds/spsc_q.h new file mode 100644 index 0000000000..d00cb4b77d --- /dev/null +++ b/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 +#include +#include + +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_ */ diff --git a/common/utils/ds/tests/CMakeLists.txt b/common/utils/ds/tests/CMakeLists.txt index a38cf159cb..1a2e5277a1 100644 --- a/common/utils/ds/tests/CMakeLists.txt +++ b/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) diff --git a/common/utils/ds/tests/test_spsc_q.cpp b/common/utils/ds/tests/test_spsc_q.cpp new file mode 100644 index 0000000000..0c1638d490 --- /dev/null +++ b/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(); +} diff --git a/common/utils/ds/tests/test_spsc_q_perf.cpp b/common/utils/ds/tests/test_spsc_q_perf.cpp new file mode 100644 index 0000000000..6487e9e6c3 --- /dev/null +++ b/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();