xtc/tests/unit/test_realtime.c

#define _POSIX_C_SOURCE 200809L
#include "xtc/xtc_realtime.h"
#include "xtc/xtc_native.h"
#include <assert.h>
#include <pthread.h>
#include <sched.h>
#include <stdio.h>
#include <stdlib.h>
#define CHECK(c) do { if (!(c)) { fprintf(stderr,"FAIL %s:%d: %s\n",__FILE__,__LINE__,#c); return 1; } } while (0)

/* Count native lifetimes on the serialized producer while a real consumer
 * traverses plans. Destruction on that consumer is always a contract violation. */
static pthread_t control_thread;
static unsigned int prepared, destroyed;
static xtc_result prepare(double rate, size_t frames, float gain, void **state)
{
    (void)rate; (void)frames;
    assert(pthread_equal(pthread_self(), control_thread));
    *state = malloc(sizeof(float));
    if (!*state) return XTC_ERR_OUT_OF_MEMORY;
    *(float *)*state = gain;
    ++prepared;
    return XTC_OK;
}
static void reset(void *state) { (void)state; }
static xtc_result process(void *state, const xtc_native_block *block)
{
    size_t i;
    for (i = 0; i < block->frames; ++i)
        block->output[i] = block->input[i] * *(float *)state;
    return XTC_OK;
}
static void destroy(void *state)
{
    assert(pthread_equal(pthread_self(), control_thread));
    if (state) ++destroyed;
    free(state);
}
static const xtc_native_descriptor descriptor = {
    XTC_NATIVE_ABI_VERSION, sizeof(xtc_native_descriptor), "concurrent.gain",
    XTC_NATIVE_EFFECT, prepare, reset, process, destroy
};

typedef struct { xtc_realtime *engine; unsigned int stop, failed; } worker_state;
static void *worker(void *context)
{
    worker_state *state = context;
    float input[32], output[32];
    size_t i;
    for (i = 0U; i < 32U; ++i) input[i] = 1.0f;
    while (!__atomic_load_n(&state->stop, __ATOMIC_ACQUIRE)) {
        if (xtc_realtime_process(state->engine, input, output, 32U) != XTC_OK)
            __atomic_store_n(&state->failed, 1U, __ATOMIC_RELAXED);
        xtc_realtime_status status;
        (void)xtc_realtime_get_status(state->engine, &status);
        float expected = status.active_revision == 2U ? 2.0f : status.active_revision % 2U ? 2.0f : 1.0f;
        if (output[0] != expected) __atomic_store_n(&state->failed, 1U, __ATOMIC_RELAXED);
        /* A whole block must come from one plan. No partially edited gains. */
        for (i = 0U; i < 32U; ++i)
            if ((output[i] != 0.0f && output[i] != 1.0f && output[i] != 2.0f) || output[i] != output[0])
                __atomic_store_n(&state->failed, 1U, __ATOMIC_RELAXED);
        sched_yield();
    }
    return NULL;
}
static int run(void)
{
    xtc_realtime *engine = NULL;
    xtc_graph *one = NULL, *two = NULL;
    xtc_realtime_status status;
    size_t node, i;
    float input[5] = {1,1,1,1,1}, output[5] = {9,9,9,9,9};
    pthread_t thread;
    worker_state state;
    CHECK(xtc_realtime_create(48000.0, 32U, &engine) == XTC_OK);
    CHECK(xtc_realtime_process(engine, input, output, 4U) == XTC_OK && output[0] == 0.0f);
    CHECK(xtc_graph_create(1U, &one) == XTC_OK);
    CHECK(xtc_graph_add_native(one, &descriptor, 1.0f, &node) == XTC_OK);
    CHECK(xtc_graph_create(1U, &two) == XTC_OK);
    CHECK(xtc_graph_add_native(two, &descriptor, 2.0f, &node) == XTC_OK);
    CHECK(xtc_realtime_submit(engine, one) == XTC_OK);
    CHECK(xtc_realtime_submit(engine, two) == XTC_ERR_CAPACITY);
    CHECK(xtc_realtime_get_status(engine, &status) == XTC_OK && status.submitted_revision == 1U && status.active_revision == 0U);
    CHECK(xtc_realtime_process(engine, input, output, 4U) == XTC_OK && output[0] == 1.0f);
    CHECK(xtc_realtime_submit(engine, two) == XTC_OK);
    CHECK(xtc_realtime_process(engine, input, output, 4U) == XTC_OK && output[0] == 2.0f);
    xtc_realtime_collect(engine);
    CHECK(xtc_realtime_get_status(engine, &status) == XTC_OK && status.active_revision == 2U && status.submitted_revision == 2U);
    CHECK(xtc_realtime_process(engine, input, output, 0U) == XTC_OK && output[0] == 2.0f);
    CHECK(xtc_realtime_get_status(engine, &status) == XTC_OK && status.processed_blocks == 3U);
    state.engine = engine; state.stop = 0U; state.failed = 0U;
    CHECK(pthread_create(&thread, NULL, worker, &state) == 0);
    for (i = 0U; i < 2000U; ++i) {
        xtc_result result;
        do {
            result = xtc_realtime_submit(engine, i % 2U ? one : two);
            if (result == XTC_ERR_CAPACITY) sched_yield();
        } while (result == XTC_ERR_CAPACITY);
        CHECK(result == XTC_OK);
    }
    __atomic_store_n(&state.stop, 1U, __ATOMIC_RELEASE);
    CHECK(pthread_join(thread, NULL) == 0);
    CHECK(state.failed == 0U);
    CHECK(xtc_realtime_get_status(engine, &status) == XTC_OK && status.processed_blocks > 3U);
    xtc_realtime_destroy(engine); /* Also cleans pending and retired plans. */
    CHECK(prepared == destroyed);
    xtc_graph_destroy(one); xtc_graph_destroy(two);
    for (i = 0U; i < 100U; ++i) {
        CHECK(xtc_realtime_create(48000.0, 4U, &engine) == XTC_OK);
        CHECK(xtc_realtime_process(engine, input, output, 5U) == XTC_ERR_CAPACITY);
        CHECK(output[0] == 0.0f && output[4] == 0.0f);
        CHECK(xtc_realtime_get_status(engine, &status) == XTC_OK && status.last_result == XTC_ERR_CAPACITY);
        xtc_realtime_destroy(engine);
    }
    return 0;
}
int main(void)
{
    control_thread = pthread_self();
    CHECK(run() == 0);
    puts("Realtime handoff and concurrent replacement tests passed.");
    return 0;
}