CLIO Core

Docs · Context Transfer Engine

Context Transfer Engine API reference.

Signatures below are taken from context-transfer-engine/core/include/clio_cte/core/core_client.h, the Python bindings in context-transfer-engine/wrapper/python/core_bindings.cc, and the parser in context-transfer-engine/core/src/core_config.cc. Types are abbreviated: Future is clio::run::Future, PoolQuery is clio::run::PoolQuery, u32 and u64 are clio::run::u32 and clio::run::u64.

Namespace
clio::cte::core
Header
clio_cte/core/core_client.h
Python
clio_cte_core_ext
Module name
clio_cte_core

Initialization

cpp
bool clio::run::CLIO_INIT(clio::run::RuntimeMode mode, bool default_with_runtime = false, bool is_restart = false);
bool clio::cte::core::CLIO_CTE_CLIENT_INIT(const std::string &config_path = "",
                                          const PoolQuery &pool_query = PoolQuery::Dynamic());
#define CLIO_CTE_CLIENT  /* clio::cte::core::Client * bound by CLIO_CTE_CLIENT_INIT */
  • RuntimeMode is kClient, kServer or kRuntime.
  • CLIO_CTE_CLIENT_INIT binds the shared client to pool 512.0 (kCtePoolId), or to the pool named by the CLIO_CTE_POOL environment variable as major.minor. A malformed value is logged and ignored.
  • Client() uses the default pool; Client(const PoolId &) addresses any pool that speaks the core interface.

C++: Tag

A thin wrapper over the shared client for one tag. The blocking put from a private buffer defaults the score to 1.0. The asynchronous and shared-memory forms default to -1.0, which keeps the blob's existing score.

cpp
class Tag {
 public:
  explicit Tag(const std::string &tag_name);   // GetOrCreateTag
  explicit Tag(const TagId &tag_id);           // wrap an existing id, no RPC

  void PutBlob(const std::string &blob_name, const char *data, size_t data_size,
               size_t off = 0, float score = 1.0f, const Context &context = Context());
  void PutBlob(const std::string &blob_name, const ctp::ipc::ShmPtr<> &data, size_t data_size,
               size_t off = 0, float score = -1.0f, const Context &context = Context());
  Future<PutBlobTask> AsyncPutBlob(const std::string &blob_name, const char *data, size_t data_size,
               size_t off = 0, float score = -1.0f, const Context &context = Context());
  Future<PutBlobTask> AsyncPutBlob(const std::string &blob_name, const ctp::ipc::ShmPtr<> &data,
               size_t data_size, size_t off = 0, float score = -1.0f, const Context &context = Context());

  void GetBlob(const std::string &blob_name, char *data, size_t data_size, size_t off = 0);
  void GetBlob(const std::string &blob_name, ctp::ipc::ShmPtr<> data, size_t data_size, size_t off = 0);
  Future<GetBlobTask> AsyncGetBlob(const std::string &blob_name, char *data, size_t data_size, size_t off = 0);

  float GetBlobScore(const std::string &blob_name);
  u64 GetBlobSize(const std::string &blob_name);
  std::vector<std::string> GetContainedBlobs();
  void ReorganizeBlob(const std::string &blob_name, float new_score);
  const TagId &GetTagId() const;
};
  • The blocking PutBlob(const char *) throws on a non-zero return code. The private-buffer forms need no shared-memory management; the ShmPtr forms take a buffer from CLIO_IPC->AllocateBuffer that the caller keeps alive until the task completes.
  • A private-buffer AsyncGetBlob served from the shared-memory cache returns an empty future: Wait() returns at once and the future must not be dereferenced. Otherwise GetReturnCode() is 0 on success after Wait().

C++: Client

Every verb is asynchronous. Results live on the task after Wait(): return_code_ or GetReturnCode() is 0 on success; tag_id_ for GetOrCreateTag; size_ for GetBlobSize; results_ for TagQuery, SemanticSearch and TemporalSearch; parallel tag_names_ and blob_names_ for BlobQuery; blob_names_ for GetContainedBlobs; entries_ for PollTelemetryLog; target_names_ for ListTargets.

Tags

cpp
Future<GetOrCreateTagTask<CreateParams>> AsyncGetOrCreateTag(const std::string &tag_name,
    const TagId &tag_id = TagId::GetNull(), const PoolQuery &pool_query = PoolQuery::Dynamic());
Future<GetOrCreateTagTask<CreateParams>> AsyncGetOrCreateTag(const std::string &tag_name,
    const clio::run::PoolId &fault_pool_id, const std::string &fault_pool_name, const std::string &fault_params,
    const TagId &tag_id = TagId::GetNull(), const PoolQuery &pool_query = PoolQuery::Dynamic());
Future<DelTagTask>    AsyncDelTag(const TagId &tag_id, const PoolQuery &pool_query = PoolQuery::Dynamic());
Future<DelTagTask>    AsyncDelTag(const std::string &tag_name, const PoolQuery &pool_query = PoolQuery::Dynamic(),
    bool posix_unlink = false);
Future<RenameTagTask> AsyncRenameTag(const std::string &old_name, const std::string &new_name,
    const TagId &tag_id = TagId::GetNull(), const PoolQuery &pool_query = PoolQuery::Broadcast());
Future<SyncTagTask>   AsyncSyncTag(const TagId &tag_id, const PoolQuery &pool_query = PoolQuery::Broadcast(),
    int min_persistence = -1);
Future<GetContainedBlobsTask> AsyncGetContainedBlobs(const TagId &tag_id,
    const PoolQuery &pool_query = PoolQuery::Dynamic());

The second AsyncGetOrCreateTag registers another pool as the tag's fault handler: a get of a blob the tag does not hold is dispatched to that pool with fault_params in the context. The Context Assimilation Engine and the paged vectors' Copy use it.

Blobs

cpp
Future<PutBlobTask> AsyncPutBlob(const TagId &tag_id, const std::string &blob_name,
    u64 offset, u64 size, const char *priv_data, float score = -1.0f,
    const Context &context = Context(), u32 flags = 0, const PoolQuery &pool_query = PoolQuery::Dynamic());
Future<PutBlobTask> AsyncPutBlob(const TagId &tag_id, const std::string &blob_name,
    u64 offset, u64 size, ctp::ipc::ShmPtr<> blob_data, float score = -1.0f,
    const Context &context = Context(), u32 flags = 0, const PoolQuery &pool_query = PoolQuery::Dynamic());
int AsyncPutBlobDefer(const TagId &tag_id, const std::string &blob_name, u64 offset, u64 size,
    const char *priv_data, float score = -1.0f, const Context &context = Context(), u32 flags = 0,
    const PoolQuery &pool_query = PoolQuery::Dynamic(), u64 max_inflight_bytes = 0);
static u64 AwaitPutsUntilSpace(u64 max_inflight_bytes, u64 max_inflight_count = std::numeric_limits<u64>::max());
static u64 DeferErrorCount();

Future<GetBlobTask> AsyncGetBlob(const TagId &tag_id, const std::string &blob_name,
    u64 offset, u64 size, u32 flags, char *priv_data,
    const PoolQuery &pool_query = PoolQuery::Dynamic(), const Context &context = Context());
Future<GetBlobTask> AsyncGetBlob(const TagId &tag_id, const std::string &blob_name,
    u64 offset, u64 size, u32 flags, ctp::ipc::ShmPtr<> blob_data,
    const PoolQuery &pool_query = PoolQuery::Dynamic(), const Context &context = Context());
Future<GetBlobTask> AsyncGetBlobDefer(const TagId &tag_id, const std::string &blob_name,
    u64 offset, u64 size, char *priv_data, u32 flags = 0,
    const PoolQuery &pool_query = PoolQuery::Dynamic(), const Context &context = Context());

Future<DelBlobTask>        AsyncDelBlob(const TagId &tag_id, const std::string &blob_name,
    const PoolQuery &pool_query = PoolQuery::Dynamic(), u32 del_flags = 0);
Future<TruncateBlobTask>   AsyncTruncateBlob(const TagId &tag_id, const std::string &blob_name, u64 new_size,
    const PoolQuery &pool_query = PoolQuery::Dynamic());
Future<GetBlobSizeTask>    AsyncGetBlobSize(const TagId &tag_id, const std::string &blob_name,
    const PoolQuery &pool_query = PoolQuery::Dynamic(), int replica = 0);
Future<GetBlobScoreTask>   AsyncGetBlobScore(const TagId &tag_id, const std::string &blob_name,
    const PoolQuery &pool_query = PoolQuery::Dynamic());
Future<ReorganizeBlobTask> AsyncReorganizeBlob(const TagId &tag_id, const std::string &blob_name, float new_score,
    const PoolQuery &pool_query = PoolQuery::Dynamic(), int replica = 0);
Future<EvictTask>          AsyncEvict(float min_tier_score, u64 bytes,
    const PoolQuery &pool_query = PoolQuery::Broadcast(), u32 droppable_only = 0);

Context flags: kNoFault suppresses fault-handler dispatch for one operation; kPutIfAbsent fails with kPutExistsRc when the blob exists; kPutIfVersion applies the put only at the expected content version and fails with a version-mismatch code otherwise. A deferred put returns 0 when submitted. The private-buffer AsyncGetBlob has three paths: a copy out of the shared-memory cache with no task (the future is empty), a direct read when the caller is inside the runtime, and client staging.

Queries and telemetry

cpp
Future<TagQueryTask>       AsyncTagQuery(const std::string &tag_regex, u32 max_tags = 0,
    const PoolQuery &pool_query = PoolQuery::Broadcast());
Future<BlobQueryTask>      AsyncBlobQuery(const std::string &tag_regex, const std::string &blob_regex,
    u32 max_blobs = 0, const PoolQuery &pool_query = PoolQuery::Broadcast());
Future<SemanticSearchTask> AsyncSemanticSearch(const std::string &tag_regex, const std::string &blob_regex,
    const std::string &query_text, u32 k = 10, const PoolQuery &pool_query = PoolQuery::Broadcast());
Future<TemporalSearchTask> AsyncTemporalSearch(const std::string &tag_regex, const std::string &blob_regex,
    Timestamp time_begin = 0, Timestamp time_end = 0, u32 max_entries = 0,
    const PoolQuery &pool_query = PoolQuery::Broadcast());
Future<PollTelemetryLogTask> AsyncPollTelemetryLog(std::uint64_t minimum_logical_time,
    const PoolQuery &pool_query = PoolQuery::Dynamic());
Future<ListTargetsTask>    AsyncListTargets(const PoolQuery &pool_query = PoolQuery::Dynamic());

Patterns are matched against the whole name with std::regex_match. SemanticSearchResult has tag_id_, blob_name_ and score_; TemporalSearchResult has tag_id_, blob_name_ and last_modified_. Timestamp is a u64 of nanoseconds on std::chrono::steady_clock (see the note on the usage page). SemanticSearch is answered by the indexer module, so the client must be bound to it.

Pool queries

  • PoolQuery::Dynamic(): route to the owner of the tag or blob. The default for data operations.
  • PoolQuery::Broadcast(): every container answers and the runtime merges. The default for queries, RenameTag, SyncTag and Evict.
  • PoolQuery::Local(): only this node.

Python: clio_cte_core_ext

The module binds the same client. Payloads are bytes, every blocking call releases the GIL, and asynchronous calls return a Future. The Python module reference lists every call with its arguments, defaults and return value. In short:

  • Setup: clio_init(mode, default_with_runtime=False, is_restart=False), initialize_cte(config_path, pool_query), get_cte_client().
  • Tag: Tag(tag_name) or Tag(tag_id); PutBlob(blob_name, data, off=0), GetBlob(blob_name, data_size, off=0), GetBlobScore, GetBlobSize, GetContainedBlobs, ReorganizeBlob(blob_name, new_score), GetTagId.
  • Client, blocking: TagQuery, BlobQuery, SemanticSearch, TemporalSearch, ReorganizeBlob, DelBlob, PollTelemetryLog, RegisterTarget.
  • Client, asynchronous: AsyncGetOrCreateTag, AsyncPutBlob, AsyncPutBlobIfAbsent, AsyncPutBlobIfVersion, AsyncGetBlob, AsyncGetBlobVersioned, AsyncGetBlobDefer, AsyncPutBlobDefer (returns an int), AwaitPutsUntilSpace, DeferErrorCount, AsyncGetBlobSize, AsyncGetContainedBlobs, AsyncDelBlob, AsyncReorganizeBlob, AsyncTagQuery, AsyncBlobQuery, AsyncSemanticSearch, AsyncTemporalSearch, AsyncPollTelemetryLog, AsyncRegisterTarget, AsyncGetCapacity, AsyncListTargets, AsyncStatTargets, AsyncGetTargetInfo.
  • Types: UniqueId (the tag id), PoolQuery, Future, SemanticSearchResult, TemporalSearchResult, CteTelemetry, CapacityInfo, TargetInfo, and the enums RuntimeMode, BdevType (kFile, kRam) and CteOp.

Environment variables

  • CLIO_CTE_POOL: pool the shared CTE client binds to, as major.minor. Default 512.0.
  • CLIO_WITH_RUNTIME: 0 attaches to a running runtime, 1 starts one in the process. Overrides the default_with_runtime argument of CLIO_INIT and clio_init.
  • CLIO_SERVER_CONF: path of the runtime configuration file. Unset means ~/.clio/clio.yaml, then built-in defaults.
  • CLIO_WAIT_SERVER: seconds a client waits for the runtime to answer before clio_init gives up. Default 30.
  • CLIO_BIND_ADDR, CLIO_SERVER_ADDR: the address the runtime binds and the address clients connect to. On Windows set both to 127.0.0.1.

Compose YAML: clio_cte_core

The CTE is one entry in the runtime's compose list. Sizes accept suffixes k to p (powers of 1024). This is the default file's entry with comments removed; the module page adds the behavior behind each key.

yaml
compose:
  - mod_name: clio_cte_core
    pool_name: cte_main
    pool_query: local
    pool_id: "512.0"
    storage:
      - path: "ram::cte_ram_tier1"
        bdev_type: "ram"
        capacity_limit: "0g"
        score: 1.0
      - path: "${CLIO_STORAGE_ROOT}/cte_disk_tier.dat"
        bdev_type: "file"
        capacity_limit: "10GB"
        score: 0.2
        persistence_level: "temporary"
    performance:
      metadata_log_path: "${CLIO_STORAGE_ROOT}/cte_metadata_log"
      transaction_log_capacity: "32MB"
    dpe:
      dpe_type: "max_bw"
    targets:
      neighborhood: 1
      default_target_timeout_ms: 30000
      poll_period_ms: 5000

storage (list)

  • path: ram::<name> for memory tiers, a file path for file tiers, a URL for s3 and gcs. Required unless binding to an existing pool.
  • bdev_type: file, ram, hbm, pinned, noop, s3 or gcs.
  • capacity_limit: size string. 0 is valid only for ram and means 80% of system DRAM. A file tier grows lazily up to this cap.
  • score: 0.0 to 1.0. Absent means automatic scoring from measured bandwidth.
  • persistence_level: volatile (default), temporary or long_term. Only a tier above volatile can hold the durable copy a flush or SyncTag asks for.
  • growth_unit: lazy-growth step of a file tier, default 1 GB.
  • existing_pool_id, existing_pool_module: bind an already composed pool, such as clio_safe_bdev, as a tier instead of creating a block device. The module name is for logging only.

performance

  • metadata_log_path: directory for the metadata write-ahead log and snapshots. Empty disables metadata durability.
  • transaction_log_capacity: WAL capacity, default 32 MB.
  • flush_metadata_period_ms (5000), flush_data_period_ms (10000), flush_data_min_persistence (1).
  • fsync_mode: durable (default) or deferred.
  • stat_targets_period_ms (1000, valid 10 to 60000), max_concurrent_operations (64).

targets

  • neighborhood: nodes, in ring order, a container registers persistent targets from. Built-in default 4; the shipped file sets 1.
  • default_target_timeout_ms (30000), poll_period_ms (5000).
  • failover_to_successor (false): route a dead owner's blobs to the first live successor.

dpe and organizer

  • dpe.dpe_type: max_bw (default), round_robin or random.
  • organizer: none (default), frecency, grayscott, cyclic, scatter or hotset. organizer_tasks (1) and organizer_period_ms (5000) size and pace the periodic rescoring.

Where to go next