Application Flow#

This page shows how to create an ovstage instance, write attributes, and read them back. See ovstage Overview for the data model and Asynchronous Operations for waiting on operations.

Create, Write, and Read#

  1. Create the stage with the ovstage.Stage context manager.

  2. Obtain its ovstage.PathDictionary and build a list of prims.

  3. Open a query over the list with ovstage.Stage.query_from_path_list().

  4. Write values at an ordinal that clears the current watermark and satisfies Ordinal Admission Policies. Wait for the write to complete and check for errors.

  5. Advance the watermark and wait for completion.

  6. Read the values, then release the results before updating the same data.

  7. Release the read and query handles and the path list before destroying the stage. In C, also release tracking for completed operations. Do not destroy the instance-owned path dictionary separately.

In Python, Operation.ok checks only enqueue acceptance. Call wait() on mutations and observe query/read/map operations inside their context managers. Cleanup is not a substitute for checking completion errors.

The snippets below write three temperature values at ordinal 1, advance the watermark, and read them back. query selects three prims; attr identifies temperature. Python imports NumPy as np and OrdinalRange from ovstage. The C API example uses C++; its waitOp() helper waits, checks errors, and releases operation tracking.

# Write one value per prim, then advance the watermark before reading.
# wait() reports asynchronous failures; the context manager and finally
# block release the read and its result even if consuming the data fails.
stage.write_attribute(
    query, attr, ordinal=1, tensors=np.array([1.0, 2.0, 3.0], np.float32), is_array=False
).wait()
stage.advance_write_floor(ordinal=1).wait()

with stage.read_attributes(query, [attr], OrdinalRange.latest(1)) as read:
    read.wait()
    group = read.fetch_next()
    if group is None:
        raise SystemExit("read returned no group at ordinal 1")
    try:
        # group.array(i) is a zero-copy numpy view of tensor i (CPU).
        print("read back ordinal", group.ordinal, group.array(0))  # -> [1. 2. 3.]
    finally:
        stage.release_group(group)

// Write one float per prim into the "temperature" column (UPSERT creates
// the prims on first write), advance the watermark to ordinal 1,
// then read the column back.
// Describe three contiguous CPU floats, one per prim in query order.
// The tensor references this array; keep it alive until the write completes.
float values[] = { 1.0f, 2.0f, 3.0f };
int64_t shape[] = { 3 };
int64_t strides[] = { 1 };
DLTensor tensor{};
tensor.data = values;
tensor.device = { kDLCPU, 0 };
tensor.ndim = 1;
tensor.dtype = { kDLFloat, 32, 1 }; // {code, bits, lanes}
tensor.shape = shape;
tensor.strides = strides;

// This is one scalar per prim, rather than a variable-length array per prim.
ovstage_write_data_t write{};
write.tensors = &tensor;
write.tensor_count = 1;
write.is_array = false;

// Submit the values at ordinal 1. UPSERT creates missing prims/attributes.
// waitOp checks both acceptance and completion before we proceed.
ovstage_enqueue_result_t enq =
    ovstage_write_attribute(stage, query, attrArg, /*ordinal*/ 1, write, OVSTAGE_PRIM_MODE_UPSERT);
waitOp(stage, enq, "write_attribute");

// Finish writes through ordinal 1 for all attributes. Once this completes,
// the values can be read and further writes at ordinals <= 1 are rejected.
ovstage_write_floor_desc_t writeFloor{};
writeFloor.ordinal = 1;
writeFloor.scope = OVSTAGE_SCOPE_ALL;
enq = ovstage_advance_write_floor(stage, &writeFloor);
waitOp(stage, enq, "advance_write_floor");

// Read the values at ordinal 1 (a snapshot: no start ordinal). A later change
// to this attribute would make this read fail with OUT_OF_RANGE; use
// UINT64_MAX as the end to read current values.
ovstage_ordinal_range_t range{};
range.end_ordinal = 1;
range.has_start_ordinal = false;
ovstage_read_handle_t read = OVSTAGE_INVALID_READ_HANDLE;
enq = ovstage_read_attributes(stage, query, &attr, 1, range, &read);
waitOp(stage, enq, "read_attributes");

// Fetch the result data after the read completes. This uniform example
// returns one group; general readers fetch until END_OF_ITERATION.
ovstage_read_group_t group{};
status = ovstage_fetch_read_next(stage, read, OVSTAGE_TIMEOUT_INFINITE, &group);
check(stage, status, "fetch_read_next");
if (group.data.tensor_count != 1 || !group.data.tensors[0].data)
{
    std::fprintf(stderr, "unexpected read layout\n");
    return EXIT_FAILURE;
}
// Borrow the returned floats while the group is alive. Releasing the group
// invalidates this pointer; copy the values first if they must live longer.
const float* out = static_cast<const float*>(group.data.tensors[0].data);
std::printf("read back ordinal %llu: %.1f %.1f %.1f\n", (unsigned long long)group.ordinal, out[0], out[1], out[2]);
ovstage_release_group(stage, &group); // the tensor data is only valid until the group is released

The C example releases the read handle separately during cleanup:

// Release the read handle after releasing its returned data.
enq = ovstage_release_read(stage, read);
waitOp(stage, enq, "release_read");

Python releases the read when its with block exits. Within that block, read.wait().fetch_next() can combine the wait and fetch; wait() returns the read handle. In both languages, releasing a read handle does not release its returned groups of data. Release those explicitly, as shown above. This example returns one group; general readers must fetch all groups. See Reading Attributes.

Further Reading#