reactor_tests.h File Reference

#include <sched.h>
#include <stdint.h>
#include <serene/rt/engine.h>
#include <serene/rt/fiber.h>
#include <serene/rt/reactor.h>
#include "acutest.h"
#include "base.h"
#include "rt/reactor/backend.h"
#include "rt/reactor/internal.h"
#include "serene/rt/impl/spsc_ring.h"

Include dependency graph for reactor_tests.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Macros
#define	REACTOR_TESTS(X)

Functions
static void	reactor_test_submit (srn_reactor_t *r, size_t channel, const srn_reactor_io_request_t *req)
static srn_reactor_io_completion_t	reactor_test_await (srn_reactor_t *r, size_t channel)
static void	reactor_test_start (srn_reactor_t *r, size_t nworkers)
static void	test_reactor_lifecycle ()
static void	test_reactor_activate_shutdown ()
static void	test_reactor_fresh_state ()
static void	test_reactor_nop_roundtrip ()
static void	test_reactor_channel_routing ()
static void	test_reactor_sleep ()
static void	test_reactor_cq_spsc_stress ()

Macro Definition Documentation

REACTOR_TESTS

#define REACTOR_TESTS

(

X

)

Value:

X("reactor::lifecycle", test_reactor_lifecycle),                   \
  X("reactor::activate_shutdown", test_reactor_activate_shutdown), \
  X("reactor::fresh_state", test_reactor_fresh_state),             \
  X("reactor::nop_roundtrip", test_reactor_nop_roundtrip),         \
  X("reactor::channel_routing", test_reactor_channel_routing),     \
  X("reactor::cq_spsc_stress", test_reactor_cq_spsc_stress),       \
  X("reactor::sleep", test_reactor_sleep)

Definition at line 38 of file reactor_tests.h.

#define REACTOR_TESTS(X)                                             \
  X("reactor::lifecycle", test_reactor_lifecycle),                   \
    X("reactor::activate_shutdown", test_reactor_activate_shutdown), \
    X("reactor::fresh_state", test_reactor_fresh_state),             \
    X("reactor::nop_roundtrip", test_reactor_nop_roundtrip),         \
    X("reactor::channel_routing", test_reactor_channel_routing),     \
    X("reactor::cq_spsc_stress", test_reactor_cq_spsc_stress),       \
    X("reactor::sleep", test_reactor_sleep)

Function Documentation

reactor_test_await()

srn_reactor_io_completion_t reactor_test_await ( srn_reactor_t * r, size_t channel )

static

Definition at line 60 of file reactor_tests.h.

                                                                                        {
  srn_reactor_io_completion_t c;
  while (srn_reactor_reap(r, channel, &c, 1) == 0) {
    // Wait for the reactor thread to post the completion. Yield so valgrind's
    // serial scheduler runs the reactor thread instead of spinning here (with
    // the reactor's indefinite wait, nothing else hands it the CPU).
    sched_yield();
  }
  return c;
}

Here is the call graph for this function:

Here is the caller graph for this function:

reactor_test_start()

void reactor_test_start ( srn_reactor_t * r, size_t nworkers )

static

Definition at line 73 of file reactor_tests.h.

                                                                  {
  srn_reactor_activate(r, nworkers, srn_sched_wake_worker);
}

Here is the call graph for this function:

Here is the caller graph for this function:

reactor_test_submit()

void reactor_test_submit ( srn_reactor_t * r, size_t channel, const srn_reactor_io_request_t * req )

static

Definition at line 50 of file reactor_tests.h.

                                                                                           {
  while (!srn_spsc_ring_push(r->channels[channel].sq, req)) {
    // SQ full, the reactor thread will drain it; retry. Yield so valgrind's
    // serial scheduler can run the reactor thread rather than spin here.
    sched_yield();
  }
  r->backend->wake(r); // the reactor waits indefinitely; nudge it to drain
}

Here is the call graph for this function:

Here is the caller graph for this function:

test_reactor_activate_shutdown()

void test_reactor_activate_shutdown ( )

static

Definition at line 91 of file reactor_tests.h.

                                             {
  MAKE_ENGINE(mm, engine);
  reactor_test_start(engine->reactor, 4);
  TEST_CHECK(engine->reactor->nchannels == 4);
  SHUTDOWN_ENGINE(mm, engine); // joins the reactor thread
}

Here is the call graph for this function:

test_reactor_channel_routing()

void test_reactor_channel_routing ( )

static

Definition at line 137 of file reactor_tests.h.

                                           {
  MAKE_ENGINE(mm, engine);
  srn_reactor_t *r = engine->reactor;
  reactor_test_start(r, 2);
 
  int marker                   = 0;
  srn_reactor_io_request_t req = {.op = SRN_REACTOR_IO_NOP, .fiber_data = &marker};
  reactor_test_submit(r, 1, &req);
 
  srn_reactor_io_completion_t c = reactor_test_await(r, 1);
  TEST_CHECK(c.fiber_data == &marker);
  TEST_CHECK(!srn_reactor_channel_has_completions(r, 0)); // channel 0 untouched
 
  SHUTDOWN_ENGINE(mm, engine);
}

Here is the call graph for this function:

test_reactor_cq_spsc_stress()

void test_reactor_cq_spsc_stress ( )

static

Definition at line 182 of file reactor_tests.h.

                                          {
  MAKE_ENGINE(mm, engine);
  srn_reactor_t *r = engine->reactor;
  reactor_test_start(r, 1);
 
  enum { N = 100000 };
  size_t sent = 0, got = 0;
 
  // Submit and reap are interleaved through the public API: the channel's
  // admission cap fills, so the producer must consume completions (and drop
  // them via srn_reactor_op_done) to reopen the window and keep making
  // progress. The test thread is both sides of channel 0.
  while (got < N) {
    bool progress = false;
 
    while (sent < N) {
      srn_reactor_io_request_t req = {
        .op = SRN_REACTOR_IO_NOP, .fiber_data = (void *)(uintptr_t)sent
      };
      if (!srn_reactor_submit(r, 0, &req)) {
        break; // channel at its in-flight cap; go drain the CQ
      }
      sent++;
      progress = true;
    }
 
    srn_reactor_io_completion_t c;
    while (srn_reactor_reap(r, 0, &c, 1) == 1) {
      if ((uintptr_t)c.fiber_data != got || c.result != 0) {
        TEST_CHECK(false); // out of order or bad result
        goto done;
      }
      srn_reactor_op_done(r, 0);
      got++;
      progress = true;
    }
 
    if (!progress) {
      // The reactor thread holds the outstanding ops; yield so valgrind's
      // serial scheduler can run it rather than spin here.
      sched_yield();
    }
  }
done:
  TEST_CHECK(got == N);
  TEST_CHECK(srn_reactor_idle(r));
  SHUTDOWN_ENGINE(mm, engine);
}

Here is the call graph for this function:

test_reactor_fresh_state()

void test_reactor_fresh_state ( )

static

Definition at line 99 of file reactor_tests.h.

                                       {
  MAKE_ENGINE(mm, engine);
  srn_reactor_t *r = engine->reactor;
  reactor_test_start(r, 2);
 
  TEST_CHECK(srn_reactor_idle(r));
  TEST_CHECK(!srn_reactor_channel_has_completions(r, 0));
  TEST_CHECK(!srn_reactor_channel_has_completions(r, 1));
 
  srn_reactor_io_completion_t c;
  TEST_CHECK(srn_reactor_reap(r, 0, &c, 1) == 0);
 
  SHUTDOWN_ENGINE(mm, engine);
}

Here is the call graph for this function:

test_reactor_lifecycle()

void test_reactor_lifecycle ( )

static

Definition at line 83 of file reactor_tests.h.

                                     {
  MAKE_ENGINE(mm, engine);
  TEST_ASSERT(engine->reactor != nullptr);
  SHUTDOWN_ENGINE(mm, engine);
}

test_reactor_nop_roundtrip()

void test_reactor_nop_roundtrip ( )

static

Definition at line 120 of file reactor_tests.h.

                                         {
  MAKE_ENGINE(mm, engine);
  srn_reactor_t *r = engine->reactor;
  reactor_test_start(r, 1);
 
  int marker                   = 0;
  srn_reactor_io_request_t req = {.op = SRN_REACTOR_IO_NOP, .fiber_data = &marker};
  reactor_test_submit(r, 0, &req);
 
  srn_reactor_io_completion_t c = reactor_test_await(r, 0);
  TEST_CHECK(c.fiber_data == &marker);
  TEST_CHECK(c.result == 0);
 
  SHUTDOWN_ENGINE(mm, engine);
}

Here is the call graph for this function:

test_reactor_sleep()

void test_reactor_sleep ( )

static

Definition at line 155 of file reactor_tests.h.

                                 {
  MAKE_ENGINE(mm, engine);
  srn_reactor_t *r = engine->reactor;
  reactor_test_start(r, 1);
 
  uint64_t before = srn_now_ns();
  uint64_t delay  = 50ULL * SRN_NS_PER_MS; // 50ms
 
  int marker                   = 0;
  srn_reactor_io_request_t req = {
    .op = SRN_REACTOR_IO_SLEEP, .deadline_ns = before + delay, .fiber_data = &marker
  };
  reactor_test_submit(r, 0, &req);
 
  srn_reactor_io_completion_t c = reactor_test_await(r, 0);
  TEST_CHECK(c.fiber_data == &marker);
  TEST_CHECK(c.result == 0);
 
  TEST_CHECK(srn_now_ns() - before >= delay); // fired no earlier than deadline
 
  SHUTDOWN_ENGINE(mm, engine);
}

Here is the call graph for this function: