Skip to main content

Control asynchronous execution

An ops:: call dispatches work and returns; the kernels run behind it. Where the work lands follows one law: a stream you pass is used as given; an operation placed on a bare Device resolves to the ambient placement scope's stream for that device when one is set, otherwise to the calling thread's asynchronous stream; and an operation with no placement of its own follows its inputs. Most programs never notice any of this, because reads wait for the result. This guide is for when you need control anyway: measuring where time goes, reacting the moment a result is ready, stepping op by op while debugging, or building a whole graph before running any of it.

Everything below runs on CPU streams, so it behaves the same on any machine; the same rules apply to CUDA, Vulkan and Metal streams. The timing numbers are from one real run and vary with the machine; the ordering they show does not. The C samples abbreviate the api-table bootstrap that tutorial part 1 shows in full.

Dispatch is not execution​

An ops:: call returns as soon as the work is queued, and the result knows where it queued: Tensor::stream() names the stream the operation rode (under the asynchronous default, stream().is_default() is false), which is the stream to query and synchronize. Tensor::status() names where a result is in that lifecycle, Stream::query_idle() asks a stream without blocking, and Stream::synchronize() blocks until everything queued has run. Before dispatching anything, StreamOrDevice(device).resolve() reads where a bare-Device placement would land. Host reads (to_string, item, item_as_vec) wait on the producing work themselves, so a read is always safe; what you never observe is unfinished bytes.

dispatch_vs_ready.cpp
#include <chrono>
#include <cstdint>
#include <cstdio>

#include <ClikaRT/clika_rt.h>

using ClikaRT::DataType;
using ClikaRT::Device;
using ClikaRT::Stream;
using ClikaRT::Tensor;
namespace ops = ClikaRT::ops;

namespace {

double ms_since(std::chrono::steady_clock::time_point t0) {
    return std::chrono::duration<double, std::milli>(
               std::chrono::steady_clock::now() - t0).count();
}

}  // namespace


int main() {
    // The inputs settle on a stream of their own. The chain below carries no
    // placement, so it rides the calling thread's asynchronous lane, and its
    // result names that stream: x.stream() is what to query and synchronize.
    Stream worker = Stream::create(Device::cpu());

    const Tensor a = ops::full({1024, 1024}, 0.001, DataType::Float32, worker);
    const Tensor w = ops::full({1024, 1024}, 0.001, DataType::Float32, worker);
    worker.synchronize();  // settle the inputs so only the chain is measured

    const auto t0 = std::chrono::steady_clock::now();
    Tensor x = a;
    for (int i = 0; i < 24; ++i) x = ops::matmul(x, w);
    const double dispatch_ms = ms_since(t0);
    Stream lane = x.stream();  // the stream the chain actually ran on

    std::printf("dispatch returned after %.1f ms, stream idle: %s\n",
                dispatch_ms, lane.query_idle() ? "yes" : "no");

    lane.synchronize();
    std::printf("ready after           %.1f ms, stream idle: %s\n",
                ms_since(t0), lane.query_idle() ? "yes" : "no");

    // A host read needs none of the above; it waits on its producer alone.
    // (item<T>() reads a single-element tensor, so reduce first.)
    std::printf("max(x) = %g\n", ops::amax(x).item<float>());
    return 0;
}
dispatch returned after 23.5 ms, stream idle: no
ready after           24.2 ms, stream idle: yes
max(x) = 0.00176685

The dispatch and the ready walls sit close together on a CPU stream, and the gap between them is the point: dispatch is still not execution (the stream is not idle when the loop returns), but memory on an asynchronous CPU stream follows execution, so a dependent chain is issued one operation ahead of the one executing and the dispatch loop is paced by the work itself. On a CUDA stream the guarantee is the enqueue rather than the run, so the same loop returns in well under a millisecond while the device works behind it. An op placed on a bare Device overlaps with the caller; code that needs it synchronous opts in explicitly, with Stream::default_stream(device) as the placement or a SynchronousStreamScope around the region.

React the moment a result is ready​

Tensor::on_complete registers a callback on a result; it fires the moment the producing kernel finishes, on ClikaRT's callback pool while the producer is still running, or inline at registration when the result has already settled. No polling and no blocked thread, the push-style inverse of synchronize(). The tensor's storage is kept alive for the callback, which receives it by const reference. Every binding carries the same member (the Go, Rust and Kotlin tabs), so a consumer in any language reacts the same way.

on_complete.cpp
#include <atomic>
#include <chrono>
#include <cstdint>
#include <cstdio>
#include <thread>

#include <ClikaRT/clika_rt.h>

using ClikaRT::DataType;
using ClikaRT::Device;
using ClikaRT::Stream;
using ClikaRT::Tensor;
namespace ops = ClikaRT::ops;

namespace {

double ms_since(std::chrono::steady_clock::time_point t0) {
    return std::chrono::duration<double, std::milli>(
               std::chrono::steady_clock::now() - t0).count();
}

}  // namespace


int main() {
    Stream worker = Stream::create(Device::cpu());
    const Tensor a = ops::full({1024, 1024}, 0.001, DataType::Float32, worker);
    const Tensor w = ops::full({1024, 1024}, 0.001, DataType::Float32, worker);
    worker.synchronize();

    std::atomic<bool> fired{false};
    const auto t0 = std::chrono::steady_clock::now();

    Tensor x = a;
    for (int i = 0; i < 10; ++i) x = ops::matmul(x, w);

    std::printf("main: dispatch done at %.1f ms, registering the callback and doing other work\n",
                ms_since(t0));

    // Fires the moment the producing kernel finishes: on ClikaRT's callback
    // pool while the producer is still running, inline at registration when
    // the result has already settled. The storage stays alive for the call.
    x.on_complete([&](const Tensor& result) {
        std::printf("callback: fired after %.1f ms, x = %s\n",
                    ms_since(t0), result.to_string().c_str());
        fired = true;
    });

    while (!fired) std::this_thread::yield();
    return 0;
}
main: dispatch done at 12.1 ms, registering the callback and doing other work
callback: fired after 13.0 ms, x = Tensor(shape=[1024, 1024], dtype=Float32, device=CPU, numel=1048576, data=[0.001268, 0.001268, 0.001268, 0.001268, 0.001268, 0.001268, ...])

Use it to hand results to a queue, complete a request, or chain host-side work without dedicating a thread to waiting. Keep callbacks short; they share the callback pool.

Step synchronously while debugging​

SynchronousStreamScope is RAII: inside it, every op the calling thread dispatches to the stream completes before the call returns, so the program state after each line is exactly what the line computed. Deterministic and slow, which is the right trade while hunting a numeric bug or stepping in a debugger. Open it on a quiescent stream, and pipelining resumes when the scope closes. It is also the region-sized opt-in for code that needs synchronous bare-Device behavior.

sync_scope.cpp
#include <cstdint>
#include <cstdio>

#include <ClikaRT/clika_rt.h>

using ClikaRT::DataType;
using ClikaRT::Device;
using ClikaRT::Stream;
using ClikaRT::SynchronousStreamScope;
using ClikaRT::Tensor;
namespace ops = ClikaRT::ops;

int main() {
    Stream worker = Stream::create(Device::cpu());
    const Tensor a = ops::full({256, 256}, 0.001, DataType::Float32, worker);
    const Tensor w = ops::full({256, 256}, 0.001, DataType::Float32, worker);
    worker.synchronize();

    // An op with no placement rides the calling thread's asynchronous lane;
    // its result names that stream, which is the one to query.
    Tensor x = ops::matmul(a, w);  // async: dispatched, likely still running
    std::printf("no scope   : idle after dispatch: %s\n",
                x.stream().query_idle() ? "yes" : "no");
    x.stream().synchronize();

    {
        SynchronousStreamScope scope(worker);
        x = ops::matmul(a, w);  // completes before this line returns
        std::printf("sync scope : idle after dispatch: %s\n",
                    x.stream().query_idle() ? "yes" : "no");
    }
    return 0;
}
no scope   : idle after dispatch: no
sync scope : idle after dispatch: yes

Build the whole graph first, run it once​

TracingScope flips the calling thread the other way, to lazy: ops return Unscheduled placeholder tensors carrying lineage and no kernel runs. One synchronize() (or any read) materializes the graph leaf-first. Use it to declare a computation in full before spending anything, or to hand the runtime the widest possible scheduling view. The same idea with a reusable artifact is tracing eager code to a graph: capture a function once as a ModelGraph and run it repeatedly, instead of scoping one thread's dispatches.

tracing_scope.cpp
#include <cstdint>
#include <cstdio>

#include <ClikaRT/clika_rt.h>

using ClikaRT::DataType;
using ClikaRT::Device;
using ClikaRT::Tensor;
using TensorStatus = ClikaRT::Tensor::Status;
using ClikaRT::TracingScope;
namespace ops = ClikaRT::ops;

namespace {

const char* name(TensorStatus s) {
    switch (s) {
        case TensorStatus::Unscheduled: return "Unscheduled";
        case TensorStatus::Evaluated:   return "Evaluated";
        case TensorStatus::Available:   return "Available";
    }
    return "?";
}

}  // namespace


int main() {
    const Tensor a = ops::ones({4, 4}, DataType::Float32, Device::cpu());
    const Tensor b = ops::ones({4, 4}, DataType::Float32, Device::cpu());

    Tensor m, s;
    {
        TracingScope trace;
        m = ops::matmul(a, b);      // no kernel runs
        s = ops::add(m, a);
        std::printf("traced  : m=%s s=%s\n", name(m.status()), name(s.status()));

        s.synchronize();            // materialize the graph, leaf-first
    }
    // Outside the scope, ops run eagerly again: a one-element view of the
    // realized result and its value readback.
    std::printf("realized: m=%s s=%s, s[0] = %g (4 ones dot ones + 1 = 5)\n",
                name(m.status()), name(s.status()),
                ops::select(s.reshape({-1}), 0, 0).item<float>());
    return 0;
}
traced  : m=Unscheduled s=Unscheduled
realized: m=Evaluated s=Evaluated, s[0] = 5 (4 ones dot ones + 1 = 5)

The four tools compose into one rule of thumb: leave the asynchronous default alone for throughput, read results and let the reads wait, reach for on_complete when a thread should not wait, and reserve the two scopes for debugging (synchronous) and up-front graph building (tracing). The bundle's async example walks each in its own chapter, including safe-reads patterns this guide leaves implicit.