myxce/src/harness/interruptcall.c

#include "interruptcall.h"

#include <assert.h>
#include <errno.h>
#include <pthread.h>
#include <semaphore.h>
#include <stdlib.h>
#include <time.h>

#include "xcetest_interrupt.h"

/* design.md's Interruptible Call Adapter component: poll interval is a
 * fixed 50ms, per the component's own Implementation Notes ("poll
 * break_out every fifty milliseconds"). Fast enough that an operator's
 * Ctrl-C feels immediate; slow enough not to spin the calling thread. */
#define XCETEST_POLL_INTERVAL_NS (50L * 1000L * 1000L) /* 50ms */

/* The heap-allocated call block backing one xcetest_run_interruptible()
 * call. `state` is the single atomic word whose CAS transitions decide
 * who frees this block -- see interruptcall.h's xcetest_call_state_t
 * comment and xcetest_run_interruptible()'s own header comment for the
 * full protocol. `lock` is the per-connection lock (task 1.5) this call
 * acquired before spawning its worker; whichever side -- caller or
 * worker -- ends up owning the block also ends up owning the job of
 * releasing `lock`, per the same protocol. It is a sem_t*, not a
 * pthread_mutex_t* -- see interruptcall.h's "WHY THE LOCK IS A POSIX
 * SEMAPHORE" section for why: the release below is a genuine cross-
 * thread handoff, which sem_t (unlike pthread_mutex_t, of any type)
 * defines cleanly. */
typedef struct {
    xcetest_blocking_fn          fn;
    void*                        ctx;
    int                          rc;
    sem_t*                       lock;
    _Atomic xcetest_call_state_t state;
} xcetest_call_block_t;

#ifdef XCE_TESTING
_Atomic int xcetest_test_abandoned_free_count = 0;
#endif

/* --- Per-connection lock registry (task 1.5) ---
 *
 * See interruptcall.h's "PER-CONNECTION LOCK REGISTRY" and "WHY THE LOCK
 * IS A POSIX SEMAPHORE" file-header sections for the full rationale,
 * including why this is a singly-linked list (pointer stability for a
 * sem_t* a still-running abandoned worker may hold onto indefinitely)
 * rather than a growable array, and why each entry is a sem_t rather
 * than a pthread_mutex_t.
 */
typedef struct xcetest_conn_lock_node {
    void*                          key;
    sem_t                          sem;
    struct xcetest_conn_lock_node* next;
} xcetest_conn_lock_node_t;

/* Only ever read/written from the single REPL thread that issues
 * top-level commands -- see interruptcall.h's "CONCURRENCY OF THE
 * REGISTRY ITSELF" note. Neither variable needs its own lock. */
static xcetest_conn_lock_node_t* g_conn_lock_registry = NULL;
static void*                     g_active_connection  = NULL;

void xcetest_set_active_connection(void* conn_key)
{
    g_active_connection = conn_key;
}

/* Looks up (by pointer equality) or lazily creates the lock entry for
 * `key` in the process-lifetime registry. Returns NULL only if the new
 * node's allocation fails. Nodes are prepended to the list and NEVER
 * freed or moved for the rest of the process's life, so a pointer this
 * function has already returned to a caller stays valid forever,
 * regardless of how many later, unrelated connections are subsequently
 * registered. */
static sem_t* xcetest_conn_lock_for(void* key)
{
    xcetest_conn_lock_node_t* node;

    for (node = g_conn_lock_registry; node; node = node->next) {
        if (node->key == key)
            return &node->sem;
    }

    node = malloc(sizeof(*node));
    if (!node)
        return NULL;

    node->key = key;
    /* A binary semaphore (initial value 1), used as a mutual-exclusion
     * lock, NOT a pthread_mutex_t: this design's own ownership-handoff
     * protocol (see xcetest_run_interruptible()'s and
     * xcetest_worker_main()'s comments) REQUIRES a cross-thread release
     * on the abandonment path -- the calling thread acquires the lock,
     * but the still-running WORKER thread is the one that releases it
     * once fn() finishes, since ownership of "release the lock" passed
     * to the worker at the same moment ownership of the call block did.
     * A pthread_mutex_t (ERRORCHECK/_RECURSIVE, or even the DEFAULT/
     * NORMAL type) is the wrong primitive for this: POSIX classifies an
     * unlock from a non-owning thread as undefined behavior for every
     * mutex type -- ERRORCHECK/_RECURSIVE are simply *required* to
     * detect and report it (which is why an earlier version of this code
     * hit an immediate EPERM here), while NORMAL/DEFAULT are merely
     * *permitted* not to detect it, not declared well-defined for it.
     * sem_t has no ownership concept at all -- sem_wait()/sem_trywait()/
     * sem_post() are fully defined regardless of which thread calls
     * which -- so it is the textbook-correct, portable primitive for
     * this exact cross-thread signal, with no UB reliance whatsoever.
     * pshared=0 since this registry is intra-process only. */
    sem_init(&node->sem, 0, 1);

    node->next           = g_conn_lock_registry;
    g_conn_lock_registry = node;
    return &node->sem;
}

/* Runs on the detached worker thread. Invokes fn(ctx), stores its result,
 * then resolves ownership of `block` via the RUNNING -> DONE_UNCLAIMED
 * CAS: on success, the caller (still polling, or about to read the result
 * of its own successful poll) owns the block from here and this function
 * must not touch it again. On failure, the caller has already forced
 * RUNNING -> ABANDONED and returned -EINTR without touching the block
 * again -- this thread now owns it and frees it. */
static void* xcetest_worker_main(void* arg)
{
    xcetest_call_block_t* block = (xcetest_call_block_t*)arg;
    xcetest_call_state_t  expected;
    sem_t*                lock = block->lock;

    block->rc = block->fn(block->ctx);

    expected = XCT_RUNNING;
    if (atomic_compare_exchange_strong(&block->state, &expected,
                                       XCT_DONE_UNCLAIMED)) {
        /* Won the CAS: the caller still owns the block AND the
         * per-connection lock. Do not touch either again from this
         * thread. */
        return NULL;
    }

    /* CAS failed. The caller is the only other party that ever writes
     * block->state, and the only value it ever writes is XCT_ABANDONED --
     * so `expected` (and the block's actual state) can only be
     * XCT_ABANDONED here. The caller returned -EINTR without touching
     * the block or the lock again; this thread now owns both. Free the
     * block (and record the test-only counter) BEFORE releasing the
     * lock, so that any other thread waiting on this same connection's
     * lock only ever observes it become available once this cleanup has
     * already fully happened. */
    free(block);
#ifdef XCE_TESTING
    atomic_fetch_add(&xcetest_test_abandoned_free_count, 1);
#endif
    {
        /* sem_post(), unlike pthread_mutex_unlock(), is well-defined from
         * any thread regardless of which thread called sem_wait()/
         * sem_trywait() -- see interruptcall.h's "WHY THE LOCK IS A
         * POSIX SEMAPHORE" section. This cross-thread release is exactly
         * the case that primitive was chosen for. */
        int post_rc = sem_post(lock);
        (void)post_rc; /* checked via assert() below, never inside it --
                         * an assert() argument's side effects vanish
                         * under NDEBUG, which must never apply to the
                         * post call itself. NOTE: this project's
                         * `release` build profile compiles with -DNDEBUG
                         * (unlike asan/debug), so this assert() -- like
                         * every other one in this file -- compiles out
                         * entirely there; it is a debug/asan-only safety
                         * net, not a guarantee in a release-profile
                         * xce_test binary. */
        assert(post_rc == 0);
    }
    return NULL;
}

/* Sleeps one poll interval. A short nanosleep() legitimately interrupted
 * early by a delivered signal (e.g. SIGINT, once xcetest_interrupt.h's
 * SA_RESTART-cleared installation is in effect) just means the next
 * poll-loop iteration happens slightly sooner than 50ms -- harmless, so
 * the return value is deliberately ignored. */
static void xcetest_sleep_poll_interval(void)
{
    struct timespec ts;
    ts.tv_sec  = 0;
    ts.tv_nsec = XCETEST_POLL_INTERVAL_NS;
    nanosleep(&ts, NULL);
}

int xcetest_run_interruptible(xcetest_blocking_fn fn, void* ctx)
{
    xcetest_call_block_t* block;
    sem_t*                lock;
    pthread_attr_t        attr;
    pthread_t             worker;
    int                   rc;

    if (!fn)
        return -EINVAL;

    lock = xcetest_conn_lock_for(g_active_connection);
    if (!lock)
        return -ENOMEM;

    /* Interruptible lock acquisition (task 1.5): poll sem_trywait() on
     * the same interval used for the completion-wait poll below,
     * checking xcetest_interrupted() between attempts. A plain
     * sem_wait() would block uninterruptibly, defeating the whole point
     * of letting a second call on this same connection be Ctrl-C'd out
     * of waiting for a prior abandoned call to release it.
     * sem_trywait() returns -1/EAGAIN when the semaphore is currently
     * at 0 (held) -- the direct equivalent of pthread_mutex_trylock()'s
     * EBUSY for this poll loop's purposes. If interrupted here, this
     * call never acquires the lock and never spawns a worker -- there
     * is nothing to release. */
    for (;;) {
        if (sem_trywait(lock) == 0)
            break; /* acquired */
        if (errno != EAGAIN)
            return -errno; /* unexpected sem_trywait() error; nothing held */

        if (xcetest_interrupted())
            return -EINTR; /* never acquired; nothing to release */

        xcetest_sleep_poll_interval();
    }

    block = malloc(sizeof(*block));
    if (!block) {
        rc = sem_post(lock);
        (void)rc;
        assert(rc == 0);
        return -ENOMEM;
    }

    block->fn   = fn;
    block->ctx  = ctx;
    block->rc   = 0;
    block->lock = lock;
    atomic_init(&block->state, XCT_RUNNING);

    rc = pthread_attr_init(&attr);
    if (rc != 0) {
        free(block);
        {
            /* NOTE: this assert(), like every other one in this file,
             * compiles out under -DNDEBUG (this project's `release`
             * build profile) -- a documented debug/asan-only safety net,
             * not a release-build guarantee. */
            int post_rc = sem_post(lock);
            (void)post_rc;
            assert(post_rc == 0);
        }
        return -rc;
    }
    /* Detached: no one ever joins this thread. That is precisely why
     * ownership of `block` (and of `lock`) must be resolved by the CAS
     * protocol below rather than by a join -- an abandoned caller
     * returns without waiting for this thread at all. */
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);

    rc = pthread_create(&worker, &attr, xcetest_worker_main, block);
    pthread_attr_destroy(&attr);
    if (rc != 0) {
        /* The worker never started, so fn() was never invoked and this
         * thread unambiguously still owns the block, and the lock, by
         * itself. */
        free(block);
        {
            /* NDEBUG caveat: see the pthread_attr_init() failure branch
             * above. */
            int post_rc = sem_post(lock);
            (void)post_rc;
            assert(post_rc == 0);
        }
        return -rc;
    }

    for (;;) {
        xcetest_call_state_t state = atomic_load(&block->state);

        if (state == XCT_DONE_UNCLAIMED) {
            int result = block->rc;
            free(block);
            {
                /* NDEBUG caveat: see the pthread_attr_init() failure
                 * branch above. */
                int post_rc = sem_post(lock);
                (void)post_rc;
                assert(post_rc == 0);
            }
            return result;
        }

        if (xcetest_interrupted()) {
            xcetest_call_state_t expected = XCT_RUNNING;
            if (atomic_compare_exchange_strong(&block->state, &expected,
                                               XCT_ABANDONED)) {
                /* Won the CAS: the worker (still running fn()) will see
                 * ABANDONED via its own failed CAS once fn() returns,
                 * free the block, and release `lock` itself. Must not
                 * touch `block` or `lock` again from here. */
                return -EINTR;
            }
            /* CAS failed: the worker is the only other party that ever
             * writes block->state, and the only value it ever writes is
             * XCT_DONE_UNCLAIMED -- so the block's actual state can only
             * be XCT_DONE_UNCLAIMED here. The worker won the race right
             * as the interrupt was noticed; fall through (via the loop's
             * next iteration) to the normal completion path above
             * instead of returning -EINTR, since the call genuinely
             * finished. */
            continue;
        }

        xcetest_sleep_poll_interval();
    }
}

/* See interruptcall.h for the full contract. Reuses xcetest_conn_lock_for()
 * -- the SAME lazy-lookup/create used by xcetest_run_interruptible() above
 * -- rather than a parallel registry, so this genuinely observes the same
 * per-connection semaphore any in-flight or abandoned worker on `conn_key`
 * is holding. */
void xcetest_wait_connection_idle(void* conn_key)
{
    sem_t* lock;
    int    rc;

    lock = xcetest_conn_lock_for(conn_key);
    if (!lock) {
        /* Allocation failure in the lazy-create path: this key has never
         * been registered, so by definition no worker could be holding a
         * lock for it via this registry. Nothing to wait for. */
        return;
    }

    /* Plain BLOCKING wait, deliberately not the interruptible
     * sem_trywait()-poll loop used above -- see interruptcall.h's
     * "CONTRACT" section on this function. A signal delivered while
     * blocked here can legitimately interrupt sem_wait() with EINTR even
     * though nothing else ever posts spuriously; retry rather than treat
     * that as "acquired" or bail out early, since either would defeat
     * this function's entire purpose (proving the wait genuinely
     * completed before the caller proceeds to tear the connection down). */
    do {
        rc = sem_wait(lock);
    } while ((rc != 0) && (errno == EINTR));
    assert(rc == 0); /* NDEBUG caveat: see xcetest_worker_main()'s comment. */

    /* Immediately release: the semantic is "prove idle right now," not
     * "hold the connection locked." */
    rc = sem_post(lock);
    (void)rc;
    assert(rc == 0);
}