reactor_tests.h

Go to the documentation of this file.

/* -*- C -*-
 * Serene programming language
 * Copyright (C) 2019-2026 Sameer Rahmani <lxsameer@lxsameer.com>
 *
 * 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/engine.h>
#include <serene/rt/fiber.h>
#include <serene/rt/reactor.h>
 
// Mostly white-box, some tests reach into the channel rings directly, below
// srn_reactor_submit, to exercise ring-level behavior in isolation. Raw
// pushes bypass the admission accounting, so a test using them must keep its
// outstanding ops under the ring capacity.
#include "acutest.h"
#include "base.h"
#include "rt/reactor/backend.h"
#include "rt/reactor/internal.h"
#include "serene/rt/impl/spsc_ring.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),                         \
    X("reactor::static_errors_stay_immutable", test_reactor_static_errors_stay_immutable)
 
// The static IO errors are shared by every failure with the same tag, so
// taking a trace on one has to go through a copy and leave the shared error
// untouched.
static void test_reactor_static_errors_stay_immutable() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
 
  srn_error_t *shared = srn_reactor_errors_static(NOT_FOUND);
  ASSERT_NOT_NULL(shared);
  TEST_CHECK(shared->as.frames == nullptr);
 
  srn_error_t *owned = ERR_FROM(ctx, shared);
  ASSERT_NOT_NULL(owned);
  TEST_CHECK(owned != shared);
  TEST_CHECK(owned->tag == NOT_FOUND);
  TEST_CHECK(owned->msg == shared->msg);
  ASSERT_NOT_NULL(owned->as.frames);
  TEST_CHECK(owned->as.frames->count == 1);
  TEST_CHECK(shared->as.frames == nullptr);
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// Push a request straight onto a channel's SQ, the way srn_reactor_submit will.
// Spins if the SQ is momentarily full (the reactor thread is the consumer).
static void
reactor_test_submit(srn_reactor_t *r, size_t channel, const srn_reactor_io_request_t *req) {
  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
}
 
// Spin on reap until one completion arrives on `channel`.
static srn_reactor_io_completion_t reactor_test_await(srn_reactor_t *r, size_t channel) {
  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;
}
 
// Activate the reactor with one channel per worker, wiring the scheduler's wake
// seam -- the same activation srn_sched_run does for a real run.
static void reactor_test_start(srn_reactor_t *r, size_t nworkers) {
  srn_reactor_activate(r, nworkers, srn_sched_wake_worker);
}
 
// ---------------------------------------------------------------------------
// public-API lifecycle and fresh-state.
// ---------------------------------------------------------------------------
 
// The engine owns a reactor; tearing the engine down without activating the
// reactor is safe (shutdown returns early when it was never started).
static void test_reactor_lifecycle() {
  MAKE_ENGINE(mm, engine);
  TEST_ASSERT(engine->reactor != nullptr);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// Activating spawns the reactor thread and allocates one channel per worker;
// engine shutdown stops and joins it cleanly.
static void test_reactor_activate_shutdown() {
  MAKE_ENGINE(mm, engine);
  reactor_test_start(engine->reactor, 4);
  TEST_CHECK(engine->reactor->nchannels == 4);
  SHUTDOWN_ENGINE(mm, engine); // joins the reactor thread
}
 
// A freshly activated reactor has nothing in flight and empty channels.
static void test_reactor_fresh_state() {
  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);
}
 
// ---------------------------------------------------------------------------
// the submit -> reactor thread -> completion pipeline (white-box submit).
// ---------------------------------------------------------------------------
 
// A NOP round-trips, the reactor thread drains the SQ, completes it, and posts
// to the CQ with the same fiber_data and a zero result.
static void test_reactor_nop_roundtrip() {
  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);
}
 
// A completion comes back on the channel it was submitted on, not another.
static void test_reactor_channel_routing() {
  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);
}
 
// A SLEEP completes no earlier than its deadline, the reactor registers a timer
// and posts the completion once the monotonic clock reaches it.
static void test_reactor_sleep() {
  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);
}
 
// Stress the channel rings, the test thread is the SQ producer and the CQ
// consumer, the reactor thread is the SQ consumer and the CQ producer -- two
// SPSC rings under contention. Every submission comes back exactly once, in
// order (single channel drains FIFO). The case TSan exists for.
static void test_reactor_cq_spsc_stress() {
  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);
}