Context Assimilation Engine API reference.

OMNI file schema

An OMNI file is YAML. The loader reads one top-level key, transfers, which must be a sequence. Any other top-level key is ignored. Each transfer is a map; src, dst and format are required, and a transfer that omits one fails the load with a message naming it by 1-based position. A per-transfer key not in the list below is logged as a warning and ignored.

Fields of AssimilationCtx that have no OMNI key (src_data, s3_region, s3_profile, on_fault, chunk_size, blob_prefix) can only be set from code.

clio_cae tool

$ clio_cae <omni_file_path>

One positional argument, the OMNI file. Any other argument count prints the usage and exits 1. clio_cae_omni is an installed alias. The tool:

1. Initializes a runtime client (CLIO_INIT(kClient, false)) and exits 1 if the runtime is unreachable.
2. Loads the file with LoadOmni, collecting run steps.
3. Executes the run steps in order with ExecuteRunSteps. On failure it exits 1 having submitted nothing.
4. Submits one ParseOmni task to pool 400.0 and waits.
5. Exits 1 if result_code_ or the task framework code is non-zero, printing error_message_, or if the request was non-empty and num_tasks_scheduled_ is 0. Otherwise prints the count and exits 0.

Environment: the tool honors the runtime client variables (CLIO_SERVER_CONF, CLIO_WITH_RUNTIME, and on Windows CLIO_SERVER_ADDR=127.0.0.1 with CLIO_BIND_ADDR=127.0.0.1). It reads no CAE-specific variables itself.

ParseOmni result codes

The ParseOmniTask outputs are num_tasks_scheduled_ (u32), result_code_ (u32, printed signed) and error_message_. Transfers are processed in order and the task stops at the first failure, so num_tasks_scheduled_ is the count completed before it.

• 0: every transfer ran.
• -1: the serialized context list could not be deserialized.
• -2: no assimilator for a source scheme, including a scheme whose support was not compiled in.
• -3: a depends_on that cannot be honored.
• -9: HDF5 include patterns matched no dataset.
• -20: an on_fault transfer whose assimilator cannot ingest per blob.
• Other non-zero: the assimilator's own error code, with error_message_ set to Assimilator failed, or a CTE return code during on_fault setup with on_fault assimilation setup failed.

AssimilationCtx

Declared in clio_cae/core/factory/assimilation_ctx.h, namespace clio::cae::core. It is serialized into the ParseOmni task.

struct AssimilationCtx {
  std::string src;          // source URL
  std::string dst;          // iowarp::<tag>
  std::string format;       // binary, hdf5, hdf5_vol, globus
  std::string depends_on;   // "" if none
  size_t range_off;         // 0
  size_t range_size;        // 0 = whole source
  std::string src_token;
  std::string dst_token;
  std::string s3_region;    // -> AWS_DEFAULT_REGION for the S3 helper; "" = inherit
  std::string s3_profile;   // -> AWS_PROFILE for the S3 helper; "" = inherit
  std::string src_data;     // payload for string:: sources
  bool on_fault = false;    // register as fault handler instead of ingesting now
  size_t chunk_size = 0;    // bytes per blob; 0 = assimilator default (1 MiB for files)
  std::string blob_prefix = "chunk_";   // "" names blobs "0", "1", ...
  std::vector<std::string> include_patterns;
  std::vector<std::string> exclude_patterns;

  AssimilationCtx();
  AssimilationCtx(const std::string& src_url, const std::string& dst_url,
                  const std::string& data_format, const std::string& dependency = "",
                  size_t offset = 0, size_t size = 0,
                  const std::string& source_token = "", const std::string& dest_token = "");
};

// Returns "" when every depends_on names an earlier context, else a message.
std::string CheckDependencies(const std::vector<AssimilationCtx>& ctxs);

The Python binding of this struct, with two constructors and the first nine fields as attributes, is on the CEE API page.

OMNI loader functions

Linked into clio_cae from the static library cae_omni_loader. Headers clio_cae/core/omni_loader.h and clio_cae/core/omni_run.h.

namespace clio::cae::core {
std::vector<AssimilationCtx> LoadOmni(const std::string& omni_path,
                                      std::vector<RunStep>* runs = nullptr);
// Throws std::runtime_error on an unreadable file, a missing or non-sequence
// `transfers`, a missing src/dst/format, or a `run` key when runs == nullptr.

struct RunStep { size_t transfer = 0; std::string script; std::string out; };
std::string RunSourceArg(const std::string& src);     // path for file::/hdf5::, else the URL
std::string RunOutputSrc(const std::string& format, const std::string& out);  // hdf5:: or file::
int RunScript(const std::string& script, const std::vector<std::string>& args, std::string* err);
std::string ExecuteRunSteps(const std::vector<RunStep>& steps,
                            std::vector<AssimilationCtx>* contexts);  // "" on success
}

C++ client

Header clio_cae/core/core_client.h, class clio::cae::core::Client, derived from clio::run::ContainerClient. The CAE pool id constant is clio::cae::core::kCaePoolId, which is 400.0. Link against clio_cae_core_client. Every method returns a clio::run::Future<Task>; call Wait() and read the task's output fields.

namespace clio::cae::core {
class Client : public clio::run::ContainerClient {
 public:
  Client() = default;
  explicit Client(const clio::run::PoolId& pool_id);

  Future<CreateTask>  AsyncCreate(const PoolQuery& pool_query, const std::string& pool_name,
                                  const PoolId& custom_pool_id, const CreateParams& params = CreateParams());
  Future<MonitorTask> AsyncMonitor(const PoolQuery& pool_query, const std::string& query);

  // Submit a list of transfers. Outputs: num_tasks_scheduled_, result_code_, error_message_.
  Future<ParseOmniTask> AsyncParseOmni(const std::vector<AssimilationCtx>& contexts);

  // Write every blob of a tag to one file. format: "hdf5" or "binary".
  // Outputs: result_code_, error_message_, bytes_exported_.
  Future<ExportDataTask> AsyncExportData(const std::string& tag_name, const std::string& output_path,
                                         const std::string& format,
                                         const PoolQuery& pool_query = PoolQuery::Local());

  // Read an HDF5 dataset (path == tag_name) from a file into a tag. format: "hdf5".
  // Outputs: result_code_, error_message_, bytes_imported_.
  Future<ImportDataTask> AsyncImportData(const std::string& tag_name, const std::string& input_path,
                                         const std::string& format,
                                         const PoolQuery& pool_query = PoolQuery::Local());

  // One HDF5 dataset into tags under tag_prefix, routable to a node.
  Future<ProcessHdf5DatasetTask> AsyncProcessHdf5Dataset(const PoolQuery& pool_query,
                                         const std::string& file_path, const std::string& dataset_path,
                                         const std::string& tag_prefix);

  // CTE passthroughs: same arguments as clio::cte::core::Client, forwarded to next_pool_id.
  Future<PutBlobTask>        AsyncPutBlob(const TagId&, const std::string& blob_name, u64 offset, u64 size,
                                          ShmPtr<> blob_data, float score = -1.0f,
                                          const Context& = Context(), u32 flags = 0,
                                          const PoolQuery& = PoolQuery::Dynamic());
  Future<GetBlobTask>        AsyncGetBlob(const TagId&, const std::string& blob_name, u64 offset, u64 size,
                                          u32 flags, ShmPtr<> blob_data,
                                          const PoolQuery& = PoolQuery::Dynamic());
  Future<GetOrCreateTagTask> AsyncGetOrCreateTag(const std::string& tag_name,
                                          const TagId& = TagId::GetNull(),
                                          const PoolQuery& = PoolQuery::Dynamic());
  Future<SemanticSearchTask> AsyncSemanticSearch(const std::string& tag_regex, const std::string& blob_regex,
                                          const std::string& query_text, u32 k = 10,
                                          const PoolQuery& = PoolQuery::Local());
};
}

// Singleton. Initializes the CTE client first. Returns false on failure.
bool CLIO_CAE_CLIENT_INIT(const std::string& config_path = "",
                          const clio::run::PoolQuery& pool_query = clio::run::PoolQuery::Dynamic());
#define CLIO_CAE_CLIENT  /* clio::cae::core::Client* */

#include <clio_cae/core/core_client.h>
#include <clio_cae/core/constants.h>

clio::run::CLIO_INIT(clio::run::RuntimeMode::kClient, false);
CLIO_CAE_CLIENT_INIT();
clio::cae::core::Client cae(clio::cae::core::kCaePoolId);

clio::cae::core::AssimilationCtx item("file::/data/run42/field.bin", "iowarp::run42_field", "binary");
auto f = cae.AsyncParseOmni({item});
f.Wait();
if (f->result_code_ != 0) { /* f->error_message_ */ }

The ExportData, ImportData and ProcessHdf5Dataset handlers return result_code_ -1 with HDF5 support not compiled in when the build lacks CLIO_CORE_ENABLE_HDF5.

Compose keys: clio_cae_core

Module clio_cae_core, method ids in core/clio_mod.yaml. The CreateParams loader reads one key from the compose entry.

Methods the module handles: kParseOmni (10), kProcessHdf5Dataset (11), kExportData (12), kImportData (13), and the CTE passthroughs kGetOrCreateTag (14), kPutBlob (15), kGetBlob (16) and kSemanticSearch (35), whose ids equal the CTE core's.

Compose keys: clio_cae_summarizer

Module clio_cae_summarizer. It interposes kPutBlob (15) and forwards every other CTE core method to next_pool_id. With no rules it is a pure passthrough.

The label is stored as {blob_name}_label in the same tag with the same placement score. A payload larger than the budget is split into chunks of about 75% of context_length tokens at 3 bytes per token, minus the prompt template, and the chunk replies are concatenated.

Environment variables

Read by the runtime process unless noted. Values in parentheses are defaults.

Ingest on fault

• CLIO_CAE_FAULT_RESERVE_MB (512): megabytes kept free across the node's targets. A fault-ingested blob that would cross that line is served from the source without being stored.
• CLIO_CAE_FAULT_STORE (1): set to 0 to never store fault-ingested blobs, giving pure read-through.
• CLIO_CAE_FAULT_STATS (unset): set to 1 to log the fault stage counters every 5 seconds.
• CLIO_CAE_REF_IO_THREADS (32): size of the thread pool that performs blocking source I/O for faults. 0 is treated as 1.

HTTP sources

• CAE_HTTP_AUTH_FILE (~/.clio/cae_http_auth.ini): INI file of per-host or per-prefix sections with query.*, header.*, bearer_token, user, password and client (curl or wget). ${VAR} in a value is expanded; an unset variable is an error. A file holding a secret must be mode 600.
• CAE_HTTP_STREAMS (8, 1 to 64): parallel ranged GETs.
• CAE_HTTP_PART_MB (64, 1 to 4096): bytes per ranged GET, in MiB.
• CAE_HTTP_PART_RETRIES (5, 0 to 100) and CAE_HTTP_BACKOFF_MS (1000, 0 to 60000): per-part retry policy.
• CAE_HTTP_OVERWRITE (0): set to 1 to overwrite an existing download.
• CAE_HTTP_PROGRESS_S (30, 0 to 3600): seconds between progress log lines.
• CAE_HTTP_CURL, CAE_HTTP_WGET: paths of the client programs.
• CAE_HTTP_TOKEN: read by the standalone cae_http_tool only, in place of src_token.

S3 sources

• Standard AWS variables: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN, AWS_PROFILE, AWS_DEFAULT_REGION (us-east-1 in the helper), S3_ENDPOINT or AWS_ENDPOINT_URL for S3-compatible stores (path-style addressing), AWS_NO_SIGN_REQUEST=1 to force anonymous requests.
• CAE_S3_STREAMS (up to 64): parallel GET streams.
• CAE_S3_PART_MB (up to 1024): part size in MiB.
• CAE_S3_MAX_RESUMES (1 to 1000), CAE_S3_BACKOFF_MS (0 to 30000), CAE_S3_PART_RETRIES (0 to 100): resume policy for a GET stream that ends short.
• CAE_S3_TOOL: read by the S3 read benchmark to locate cae_s3_tool.

Other sources

• GLOBUS_ACCESS_TOKEN: Globus token when src_token is empty. GLOBUS_HTTPS_ACCESS_TOKEN is read for the HTTPS collection path.
• GCS_ENDPOINT: overrides the Google Cloud Storage endpoint.

Client side

• CLIO_CTE_POOL (major.minor): binds the process's CTE client, which CLIO_CAE_CLIENT_INIT creates, to an interposing pool. Needed for semantic queries; see the CEE usage page.
• CLIO_WITH_RUNTIME: 1 or true makes a client try to start a runtime in-process before attaching. clio_cae and the CEE default to attaching only.