Asynchronous Submit/Observe Model#

ovstage is an asynchronous, ordinal-keyed submit/observe system. State-mutating and data-producing calls enqueue synchronously — they return an ovstage_enqueue_result_t (status + op_index) immediately — while the real work runs later. This lets the CPU run ahead of execution: a producer can push many ordinals of data while consumers observe at their own cadence.

Enqueue vs. Execution#

  • An enqueue returning OVSTAGE_OK means the op was accepted, not completed. Check status == OVSTAGE_OK before using op_index; a non-OK enqueue yields OVSTAGE_INVALID_OP_ID.

  • Per-op execution errors surface later, at ovstage_wait_op (or the matching ovstage_fetch_*), not at enqueue time. Refer to Error Handling and Diagnostics.

Ordinal Ordering and Concurrency#

Writes, deletes, and map commits carry an explicit ordinal:

  • Same-ordinal ops execute in submission order.

  • Different-ordinal ops are independent and can run concurrently.

A ovstage_wait_op that returns OVSTAGE_OK does not imply all your enqueues finished — ops in other ordinal buckets can still be in flight. Wait on the specific op_id whose result you need. ovstage_wait_op waits for the whole dependency chain up to and including that op_id.

Blocking, Polling, and Timeouts#

ovstage_wait_op(instance, op_id, timeout, &wait_result) (and the ovstage_fetch_* calls) share one timeout convention:

timeout

Behavior

0

Poll — return immediately; OVSTAGE_ERROR_TIMEOUT means “not ready yet.”

OVSTAGE_TIMEOUT_INFINITE

Block until the op completes.

other (nanoseconds)

Wait up to that long, then report OVSTAGE_ERROR_TIMEOUT.

ovstage_timeout_ns_t is a uint64_t nanosecond count and OVSTAGE_TIMEOUT_INFINITE is ~0ULL. OVSTAGE_ERROR_TIMEOUT is a “not-ready” signal, distinct from the terminal OVSTAGE_ERROR_OP_FAILED. On a timeout, wait_result.lowest_pending_op_id reports the lowest still-pending op in the chain — a partial-progress cursor.

After an op completes, release its tracking with ovstage_release_op; the id must not be reused afterward.

Running the CPU Ahead#

To run ahead, keep enqueuing across ordinals without blocking, and only wait or fetch when you actually need a result. Python mirrors the same model — enqueue methods return handles whose .wait(timeout=...) polls (timeout=0) or blocks (TIMEOUT_INFINITE) and raises ovstage.OvstageError on op failure:

# Run the CPU ahead of execution: enqueue without blocking, then poll with
# timeout=0 and do other work between polls instead of stalling. The low-level
# Stage.wait_op returns (code, error_op_ids, lowest_pending_op_id); a TIMEOUT
# code means "not ready yet". Release the op once it completes.
def poll_until_done(stage, op) -> None:
    if not op.ok:
        raise OvstageError(op.status, op.error_message())  # enqueue was rejected
    while True:
        code, error_op_ids, lowest_pending = stage.wait_op(op.op_id, timeout=0)
        if code == ovstage.ErrorCode.TIMEOUT:
            continue  # not done yet — go do other CPU work / submit the next ordinal
        if code != ovstage.ErrorCode.OK:
            msg = op.error_message()  # read while the op is alive — it's released next
            stage.release_op(op.op_id)
            raise OvstageError(code, msg)
        stage.release_op(op.op_id)
        return

In C, the same shape is: enqueue, check the enqueue status, then ovstage_wait_op on the op_index and inspect per-op errors:

// Drive an async enqueue to completion. Enqueue success (OVSTAGE_OK) only
// means the op was accepted, so wait on the op id, report any per-op errors
// surfaced by the wait, and retire the op (ovstage_destroy_instance requires
// every op released first). The examples fail fast; a real application would
// propagate the errors instead.
static void waitOp(ovstage_instance_t* stage, ovstage_enqueue_result_t enq, const char* what)
{
    check(stage, enq.status, what);
    ovstage_op_wait_result_t wait{};
    const ovstage_api_status_t code = ovstage_wait_op(stage, enq.op_index, OVSTAGE_TIMEOUT_INFINITE, &wait);
    for (size_t i = 0; i < wait.error_op_id_count; ++i)
    {
        const ovx_string_t e = ovstage_get_last_op_error(stage, wait.error_op_ids[i]);
        std::fprintf(stderr, "ovstage %s op %llu failed: %.*s\n", what,
                     (unsigned long long)wait.error_op_ids[i], (int)e.length, e.ptr ? e.ptr : "");
    }
    if (wait.error_op_id_count != 0)
        std::exit(EXIT_FAILURE);
    ovstage_release_op(stage, enq.op_index);
    check(stage, code, what);
}

Note

This build retains only the latest committed state — do not design async flows that read back older ordinals.

Where to Go Next#