CLIO Core

Docs · Context Transfer Engine

Using the Context Transfer Engine.

This page shows how to do each common job with the CTE client, in C++ and in Python. The API reference has the full signatures; the overview explains tags, blobs, scores and tiers.

C++
clio::cte::core::Tag, Client
Python
clio_cte_core_ext
Default pool
512.0, or CLIO_CTE_POOL

Connect a client

Every program first initializes the runtime client, then the CTE client. The CTE client binds to the core pool 512.0 by default. The environment variable CLIO_CTE_POOL, as major.minor, binds it to another pool that speaks the core interface, such as the indexer at 564.0 or the cache at 563.0. Set it before initialization.

cpp
#include <clio_runtime/clio_runtime.h>
#include <clio_cte/core/core_client.h>

clio::run::CLIO_INIT(clio::run::RuntimeMode::kClient);   // attach to a running clio_run
clio::cte::core::CLIO_CTE_CLIENT_INIT();                  // bind the shared client
auto *client = CLIO_CTE_CLIENT;                           // clio::cte::core::Client *
python
import os
os.environ["CLIO_WITH_RUNTIME"] = "0"     # attach to a running clio_run
os.environ["CLIO_CTE_POOL"] = "564.0"     # optional: the indexer, so SemanticSearch works

import clio_cte_core_ext as cte
cte.clio_init(cte.RuntimeMode.kClient, False)    # (mode, default_with_runtime=False, is_restart=False)
cte.initialize_cte("", cte.PoolQuery.Dynamic())  # (config_path, pool_query)
client = cte.get_cte_client()

To address a specific pool from C++ without the environment variable, construct clio::cte::core::Client with a PoolId. In Python, cte.Client() and cte.Client(pool_id) do the same.

Put and get a blob

The Tag wrapper is the simplest path. Its constructor runs GetOrCreateTag. The blocking PutBlob from a private buffer writes at offset 0 with score 1.0 unless told otherwise, and throws on a non-zero return code. GetBlob reads into a buffer you own.

cpp
clio::cte::core::Tag tag("experiment/run1");
std::string payload = "Arctic sea ice extent fell to a record low.";
tag.PutBlob("note0", payload.data(), payload.size());          // off 0, score 1.0

std::string out(tag.GetBlobSize("note0"), '\0');
tag.GetBlob("note0", out.data(), out.size());

tag.PutBlob("note0", "!", 1, /*off=*/payload.size());           // append one byte
python
tag = cte.Tag("experiment/run1")
tag.PutBlob("note0", b"Arctic sea ice extent fell to a record low.", 0)
data = tag.GetBlob("note0", tag.GetBlobSize("note0"), 0)      # bytes

Python PutBlob has no score argument; a blob written from Python lands at 1.0 and can be moved afterwards with ReorganizeBlob. GetBlob raises RuntimeError when the read fails. A range whose bytes lived only in RAM before a runtime restart fails with an I/O error rather than returning zeros.

Work with tags

cpp
auto *client = CLIO_CTE_CLIENT;
const auto &id = tag.GetTagId();

std::vector<std::string> names = tag.GetContainedBlobs();

auto rn = client->AsyncRenameTag("experiment/run1", "experiment/run1-final", id);
rn.Wait();

auto sync = client->AsyncSyncTag(id);        // flush to a persistent tier, like fsync
sync.Wait();

auto del = client->AsyncDelTag("experiment/run1-final");
del.Wait();
python
names = tag.GetContainedBlobs()
tid = tag.GetTagId()                                  # UniqueId with major_, minor_
later = cte.Tag(tid)                                  # wrap an id, no RPC
fut = client.AsyncGetOrCreateTag("experiment/run2")   # .result() -> UniqueId
run2 = cte.Tag(fut.result())

The Python module exposes blob deletion (DelBlob) but not tag deletion, renaming or SyncTag.

Query by name

TagQuery and BlobQuery match whole names with std::regex_match: "doc.*" matches doc7, "doc" does not, and ".*ice.*" matches a substring. Both default to PoolQuery::Broadcast(), so every node answers for its own blobs and the results are merged. A max of 0 means no limit.

cpp
auto tq = client->AsyncTagQuery("experiment/.*");
tq.Wait();
for (const auto &name : tq->results_) { /* tag names */ }

auto bq = client->AsyncBlobQuery("experiment/.*", "note.*");
bq.Wait();
for (size_t i = 0; i < bq->blob_names_.size(); ++i) {
  // bq->tag_names_[i], bq->blob_names_[i]
}
python
tags = client.TagQuery("experiment/.*")                       # list[str]
for tag_name, blob_name in client.BlobQuery("experiment/.*", "note.*"):
    print(tag_name, blob_name)

Search by keyword

SemanticSearch is BM25 keyword search over blob contents. The core does not implement it; the indexer module at 564.0 does. The client must therefore be bound to the indexer or to a pool that forwards to it, through CLIO_CTE_POOL=564.0 or a Client constructed on that pool. Results come back best first, with score 0 for a blob that matched the patterns but contains no query word. k defaults to 10; 0 means no cap.

cpp
clio::cte::core::Client indexer(clio::run::PoolId(564, 0));
auto hits = indexer.AsyncSemanticSearch("experiment/.*", ".*", "sea ice", 5);
hits.Wait();
for (const auto &r : hits->results_) {
  // r.tag_id_, r.blob_name_, r.score_
}
python
for hit in client.SemanticSearch("experiment/.*", ".*", "sea ice", k=5):
    print(hit.blob_name, hit.score)

For files in a Context FS mount, pass "[0-9]+" as the blob pattern to search the page blobs and skip each file's attribute blob.

Search by time

TemporalSearch returns blobs whose last-modified timestamp lies in the inclusive range [time_begin, time_end], sorted ascending; 0 for either bound means unbounded.

python
for r in client.TemporalSearch("experiment/.*", ".*", time_begin=0, time_end=0, max_entries=20):
    print(r.blob_name, r.last_modified)
The timestamp is not Unix time

The docstrings describe last_modified as epoch nanoseconds, but the core stamps it from GetCurrentTimeNs(), which reads std::chrono::steady_clock. A bound computed from time.time() matches nothing. Treat last_modified as an ordering key within one run of the runtime. This is a suspected bug in the source, documented here as it behaves.

Move a blob between tiers

A blob's score decides its tier: 1.0 is the fastest, 0.0 the slowest. ReorganizeBlob gives the blob a new score and moves it to the tier that score selects. Evict frees the lowest-scored blobs off a tier, down to a minimum tier score and a byte count. With the default organizer: none these two calls are the only things that move data after a put.

cpp
tag.ReorganizeBlob("note0", 0.2f);                 // demote
float s = tag.GetBlobScore("note0");               // 0.2

auto ev = client->AsyncEvict(/*min_tier_score=*/0.9f, /*bytes=*/64 << 20);
ev.Wait();
python
tag.ReorganizeBlob("note0", 0.0)                   # slowest tier
ok = client.ReorganizeBlob(tag.GetTagId(), "note1", 1.0)   # True on success

Which tiers a score can reach depends on the tiers' own scores and the placement engine. The placement page explains how max_bw chooses.

Run operations asynchronously

Every Client method is asynchronous and returns a clio::run::Future over the task. Submit several, then Wait() on each and read the task's fields. The Tag wrapper adds AsyncPutBlob and AsyncGetBlob from private buffers. The buffer must stay valid until Wait() returns.

cpp
std::vector<std::string> bufs(4, std::string(4096, '\0'));
std::vector<clio::run::Future<clio::cte::core::GetBlobTask>> futs;
for (int i = 0; i < 4; ++i) {
  futs.push_back(tag.AsyncGetBlob("page" + std::to_string(i), bufs[i].data(), bufs[i].size()));
}
for (auto &f : futs) f.Wait();   // served from the shared-memory cache: the future is empty, do not dereference it

In Python every Async* method returns a Future with wait(max_sec=-1), done() and result(max_sec=-1). The value of result() depends on the call: bytes for a get, a list for a query, a dict for the conditional puts, None for status-only operations.

python
tid = tag.GetTagId()
puts = [client.AsyncPutBlob(tid, f"page{i}", bytes(4096), 0) for i in range(4)]
assert all(f.wait() == 0 for f in puts)

gets = [client.AsyncGetBlob(tid, f"page{i}", 4096, 0) for i in range(4)]
pages = [f.result() for f in gets]

Deferred puts

A deferred put stages a copy of the data in shared memory and returns a submission code immediately, without a future. The put is tracked in a process-wide registry. When shared memory is exhausted the next submission waits for the oldest puts to finish; max_inflight_bytes adds an explicit ceiling. Drain with AwaitPutsUntilSpace(0) and check DeferErrorCount() for puts that failed after submission. AsyncGetBlobDefer completes any pending deferred put for the blob before reading it.

python
for i in range(1000):
    rc = client.AsyncPutBlobDefer(tid, f"rec{i}", record(i), 0, max_inflight_bytes=256 << 20)
    assert rc == 0
client.AwaitPutsUntilSpace(0)
assert client.DeferErrorCount() == 0

Conditional puts

Two Context flags make a put conditional. kPutIfAbsent creates the blob only if it does not exist and fails with kPutExistsRc otherwise. kPutIfVersion applies the put only when the blob is still at the version the caller read, and fails with a version-mismatch code otherwise. Python exposes them as AsyncPutBlobIfAbsent and AsyncPutBlobIfVersion, whose result() is a dict with code and version, and AsyncGetBlobVersioned returns code, version and data. Conditional puts need a non-empty payload, and a put overwrites a range without truncating, so use fixed-size records.

Delete

cpp
auto del = client->AsyncDelBlob(tag.GetTagId(), "note0");
del.Wait();
int rc = del->return_code_;        // 0 on success
python
assert client.DelBlob(tag.GetTagId(), "note0")

Inspect targets

From Python, AsyncListTargets() returns the registered target names on this node, AsyncGetTargetInfo(name) returns a TargetInfo with the target's score, remaining space and byte and operation counters, and AsyncGetCapacity() returns total and remaining bytes. All three default to PoolQuery.Local(); Broadcast() sums or concatenates across nodes. A failed query raises RuntimeError from result().

python
for name in client.AsyncListTargets().result():
    info = client.AsyncGetTargetInfo(name).result()
    print(name, info.target_score, info.remaining_space)

Where to go next