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.

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. A timing is taken at a host read of the result, or after synchronize() on the producing stream: the return of the dispatch is not the end of the work, and a result nobody reads is not evidence of when it ran.

C++
Python

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({2048, 2048}, 0.001, DataType::Float32, worker);
    const Tensor w = ops::full({2048, 2048}, 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;
    // 48 dependent 2048 x 2048 products, about 0.8 TFLOP: the lane is still
    // at work when the query after the dispatch returns, on any machine.
    for (int i = 0; i < 48; ++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;
}

Every operator returns at once and its work rides the calling thread's lane on the input's device; a read waits for the value it needs. The contract, in full:

These settle (they block until the value exists): numpy(), bytes(), item(), tolist(), repr and print, bool(), __array__, __dlpack__, crt.save, crt.eval(*trees) and crt.synchronize(target); a truthful shape / numel() / nbytes settles only when the producer's shape depends on data.
These never settle: dtype, device, ndim, stride(), is_contiguous().
The scopes are thread-local context managers that also work as decorators: crt.synchronous(), crt.tracing(), crt.eager(), crt.meta_init() (the same as crt.device("meta")), crt.device(d) and crt.stream(s).
crt.async_eval(*trees) submits the work and returns without waiting, the one-token-lookahead idiom of a decode loop.
A data read under crt.tracing() or of a storage-free tensor raises crt.ClikaRTError; numpy() on a device tensor raises TypeError (move it to the cpu first); bool() of a tensor with more than one element raises RuntimeError.

settle.py
import numpy as np
import clika_runtime as crt

x = np.ones((64, 64), dtype=np.float32)
t = crt.tensor(x)
y = t
for _ in range(50):
    y = crt.add(crt.mul(y, 1.01), 0.01)   # fifty operators, dispatched without waiting

print(y.dtype)                              # clika_runtime.float32 (metadata never settles)
print(float(y.numpy()[0, 0]))               # 2.289262533187866 (the first read returns the settled value)

tree = {"a": crt.exp(t), "b": [crt.sum(t), None, "text"], "c": (crt.relu(t),)}
crt.eval(tree, crt.abs(t))                  # settles every tensor leaf of every tree, ignores the rest

z = crt.add(crt.mul(t, 2.0), 1.0)
crt.async_eval(z)                           # submitted, not waited for
print(float(z.numpy()[0, 0]))               # 3.0 (3.0, read later)

borrowed = crt.from_numpy(x)                # a zero-copy view over the array
borrowed.add_(1.0)
crt.synchronize()                           # the calling thread's lane; the array carries the write after this
print(x[0, 0])                              # 2.0 (2.0)

A raw read of a borrowed source array after an in-place operator settles first, as the last four lines do; a fresh numpy() call settles on its own.

dispatch returned after 23.5 ms, stream idle: no
ready after           24.2 ms, stream idle: yes
max(x) = 0.00176683

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.

C++
Python

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;
}

The Python package carries no completion callback; the push-style on_complete is the C++ arm's. The Python shape of the same idea is crt.async_eval to submit without waiting, and a read at the point of use: hand the tensor to whoever consumes it, and that consumer's numpy() or item() is the wait, on its own thread if the dispatching thread must not block.

main: dispatch done at 10.8 ms, registering the callback and doing other work
callback: fired after 11.8 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.

C++
Python

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());
    // One 4096 x 4096 product, about 137 GFLOP: still at work when the
    // query after the dispatch returns, on any machine.
    const Tensor a = ops::full({4096, 4096}, 0.001, DataType::Float32, worker);
    const Tensor w = ops::full({4096, 4096}, 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;
}

crt.synchronous() is the region form: inside it every operator the calling thread dispatches completes before the call returns, so a borrowed source array carries an in-place write the moment the line finishes. It works as a decorator too.

sync_scope.py
import numpy as np
import clika_runtime as crt

source = np.zeros((8, 8), dtype=np.float32)
borrowed = crt.from_numpy(source)
with crt.synchronous():
    borrowed.add_(3.0)
    print(source[0, 0])                     # 3.0 (3.0, written before the line returned)
    result = crt.mul(borrowed, 2.0)
print(float(result.numpy()[0, 0]))          # 6.0 (6.0)

@crt.synchronous()
def step(t: crt.Tensor) -> crt.Tensor:
    return crt.relu(t)

print(float(step(borrowed).numpy()[0, 0]))  # 3.0 (3.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.

C++
Python

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;
}

crt.tracing() flips the calling thread to lazy: operators return unscheduled tensors that carry their shape and dtype and no value, a data read inside the region raises crt.ClikaRTError, and crt.eval materializes the whole chain leaf-first. crt.eager() restores the default inside a tracing region for the one operator that must run at once.

tracing_scope.py
import numpy as np
import clika_runtime as crt

t = crt.tensor(np.ones((8, 8), dtype=np.float32))
with crt.tracing():
    y = crt.add(crt.mul(t, 2.0), 1.0)
    print(y.shape, y.dtype)                 # clika_runtime.Size([8, 8]) clika_runtime.float32 ((8, 8) clika_runtime.float32; no kernel has run)
    try:
        y.numpy()
    except crt.ClikaRTError:
        print("a value read under tracing raises")   # printed: the read raised
    with crt.eager():
        at_once = crt.mul(t, 2.0)
        print(float(at_once.numpy()[0, 0]))  # 2.0 (2.0, run at once)
crt.eval(y)                                 # materializes the traced chain
print(float(y.numpy()[0, 0]))               # 3.0 (3.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.

Dispatch is not execution​

React the moment a result is ready​

Step synchronously while debugging​

Build the whole graph first, run it once​

Dispatch is not execution

React the moment a result is ready

Step synchronously while debugging

Build the whole graph first, run it once