spsc_ring_tests.h

Go to the documentation of this file.

/* -*- C -*-
 * Serene programming language
 * Copyright (C) 2019-2026 Sameer Rahmani <[email protected]>
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <https://www.gnu.org/licenses/>.
 */
 
#pragma once
 
#include <sched.h>
#include <stdint.h>
 
#include <serene/rt/fiber/thread.h>
#include <serene/rt/impl/spsc_ring.h>
 
#include "acutest.h"
#include "base.h"
 
#define SPSC_RING_TESTS(X)                                      \
  X("spsc_ring::make_valid", test_spsc_ring_make_valid),        \
    X("spsc_ring::make_out_of_range", test_spsc_ring_make_oor), \
    X("spsc_ring::fill_drain_wrap", test_spsc_ring_fill_drain), \
    X("spsc_ring::elem_bytes", test_spsc_ring_elem_bytes),      \
    X("spsc_ring::spsc_stress", test_spsc_ring_spsc_stress)
 
// A valid magnitude yields a power-of-two capacity, no error, and an empty
// ring.
static void test_spsc_ring_make_valid() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
 
  srn_spsc_ring_t *r = srn_spsc_ring_make(ctx, 3, sizeof(uint64_t)); // cap 8
  TEST_ASSERT(r != nullptr);
  TEST_CHECK(!HAS_ERROR(*r));
  TEST_CHECK(r->cap == 8);
  TEST_CHECK(r->mask == 7);
  TEST_CHECK(r->elem_size == sizeof(uint64_t));
  TEST_CHECK(srn_spsc_ring_empty(r));
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// A magnitude below MIN_MAG or above MAX_MAG returns a value carrying an error,
// not an aborting failure.
static void test_spsc_ring_make_oor() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
 
  srn_spsc_ring_t *too_small = srn_spsc_ring_make(ctx, 1, sizeof(int));
  TEST_ASSERT(too_small != nullptr);
  TEST_CHECK(HAS_ERROR(*too_small));
  TEST_ASSERT(too_small->maybe_error->tag == WRONG_RANGE);
 
  srn_spsc_ring_t *too_big = srn_spsc_ring_make(ctx, 64, sizeof(int));
  TEST_ASSERT(too_big != nullptr);
  TEST_CHECK(HAS_ERROR(*too_big));
  TEST_ASSERT(too_big->maybe_error->tag == WRONG_RANGE);
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// Fill to capacity (push past it fails), drain, and refill so the indices wrap;
// elements still come out in order across the wrap point.
static void test_spsc_ring_fill_drain() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
 
  srn_spsc_ring_t *r = srn_spsc_ring_make(ctx, 2, sizeof(uint64_t)); // cap 4
  TEST_ASSERT(r != nullptr && !HAS_ERROR(*r));
 
  TEST_CHECK(srn_spsc_ring_empty(r));
 
  for (uint64_t i = 0; i < 4; i++) {
    TEST_CHECK(srn_spsc_ring_push(r, &i));
  }
 
  uint64_t overflow = 99;
  TEST_CHECK(!srn_spsc_ring_push(r, &overflow)); // full
  TEST_CHECK(!srn_spsc_ring_empty(r));
 
  uint64_t v = 0;
  TEST_CHECK(srn_spsc_ring_pop(r, &v) && v == 0);
  TEST_CHECK(srn_spsc_ring_pop(r, &v) && v == 1);
 
  uint64_t a = 4, b = 5; // these wrap into the freed slots
  TEST_CHECK(srn_spsc_ring_push(r, &a));
  TEST_CHECK(srn_spsc_ring_push(r, &b));
 
  for (uint64_t expect = 2; expect <= 5; expect++) {
    TEST_CHECK(srn_spsc_ring_pop(r, &v) && v == expect);
  }
  TEST_CHECK(srn_spsc_ring_empty(r));
  TEST_CHECK(!srn_spsc_ring_pop(r, &v)); // empty
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
typedef struct spsc_pair {
  uint32_t a;
  uint32_t b;
} spsc_pair;
 
// A struct element round trips field for field, confirming push/pop copy the
// whole element via elem_size, not just scalar values.
static void test_spsc_ring_elem_bytes() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
 
  srn_spsc_ring_t *r = srn_spsc_ring_make(ctx, 3, sizeof(spsc_pair));
  TEST_ASSERT(r != nullptr && !HAS_ERROR(*r));
 
  spsc_pair in = {.a = 0xDEADBEEFU, .b = 0x12345678U};
  TEST_CHECK(srn_spsc_ring_push(r, &in));
 
  spsc_pair out = {0};
  TEST_CHECK(srn_spsc_ring_pop(r, &out));
  TEST_CHECK(out.a == 0xDEADBEEFU && out.b == 0x12345678U);
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
typedef struct spsc_producer_arg {
  srn_spsc_ring_t *ring;
  uint64_t count;
} spsc_producer_arg;
 
static void spsc_producer(void *p) {
  spsc_producer_arg *a = (spsc_producer_arg *)p;
  for (uint64_t i = 0; i < a->count; i++) {
    while (!srn_spsc_ring_push(a->ring, &i)) {
      // Ring full, yield so the consumer gets the CPU. Without this a tight
      // spin starves the other thread under valgrind's serialized scheduler.
      sched_yield();
    }
  }
}
 
// One producer thread, one consumer (this one), a small ring forcing constant
// full/empty contention and wraparound. Every value arrives exactly once and in
// order. This is the case the memory ordering exists for; run it under TSan.
static void test_spsc_ring_spsc_stress() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
 
  srn_spsc_ring_t *r = srn_spsc_ring_make(ctx, 6, sizeof(uint64_t)); // cap 64
  TEST_ASSERT(r != nullptr && !HAS_ERROR(*r));
 
  constexpr uint64_t n  = 100000;
  spsc_producer_arg arg = {.ring = r, .count = n};
  srn_thread_t producer;
  TEST_ASSERT(srn_thread_spawn(&producer, spsc_producer, &arg) == SRN_THREAD_OK);
 
  bool in_order = true;
  for (uint64_t i = 0; i < n; i++) {
    uint64_t v;
    while (!srn_spsc_ring_pop(r, &v)) {
      // Ring empty, yield so the producer gets the CPU. Without this a tight
      // spin starves the producer under valgrind's serialized scheduler.
      sched_yield();
    }
    if (v != i) {
      in_order = false;
      break;
    }
  }
  TEST_CHECK(in_order);
  TEST_CHECK(srn_thread_join(&producer) == SRN_THREAD_OK);
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}