Writing an assimilator.

Where the code lives

Assimilators are part of the CAE core module under context-assimilation-engine/core. Each has an interface header under include/clio_cae/core/factory/ and an implementation under src/factory/. The shipped classes, all in namespace clio::cae::core:

• BinaryFileAssimilator for file::. Implements ReadBlob and BlobBytes, so it supports ingest on fault.
• StringDataAssimilator for string::. Only Schedule.
• Hdf5FileAssimilator for hdf5::, behind CLIO_CAE_ENABLE_HDF5.
• HttpFileAssimilator for http:// and https://, behind CAE_ENABLE_HTTP (POSIX builds).
• S3FileAssimilator behind CAE_ENABLE_S3, GcsFileAssimilator behind CAE_ENABLE_GCS, GlobusFileAssimilator behind CAE_ENABLE_GLOBUS.

The interface

namespace clio::cae::core {
class BaseAssimilator {
 public:
  virtual ~BaseAssimilator() = default;

  // Required. A coroutine: co_await CTE operations, report failure in error_code.
  virtual clio::run::TaskResume Schedule(const AssimilationCtx& ctx, int& error_code) = 0;

  // Optional, for ingest on fault. The bytes of ONE destination blob, exactly as
  // Schedule would have stored them. Blocking I/O, no tasks. Returns the byte
  // count, 0 past the end of the source, or -errno. Default: -38 (ENOSYS).
  virtual long long ReadBlob(const AssimilationCtx& ctx, const std::string& blob_name,
                             char* dst, clio::run::u64 cap);

  // Optional. Bytes per blob for ctx. Default 0 = unsupported.
  virtual clio::run::u64 BlobBytes(const AssimilationCtx& ctx);

  // Optional, for pagify. The page-space layout of the source. Default: empty map.
  virtual PageMap BuildPageMap(const AssimilationCtx& ctx, clio::run::u64 page_size);

  // Helpers over a map, not virtual.
  std::vector<PageSlice> ToPage(const PageMap& map, clio::run::u64 page_id) const;
  std::vector<PageRef>   FromPage(const PageMap& map, const std::string& blob,
                                  clio::run::u64 off, clio::run::u64 size) const;
};
}

Only Schedule is required. The three hooks opt a source into the features on the on-demand page: ReadBlob and BlobBytes together make a source usable with on_fault, and BuildPageMap is where a source would publish its page layout. None of the shipped assimilators overrides BuildPageMap yet.

What Schedule must do

Schedule runs inside the CAE pool as a runtime coroutine. It receives the context the user built and a reference to an error code it must set. The pattern the shipped assimilators follow:

1. Parse ctx.src: the part after the scheme is the path, object key or payload name. Derive the tag name from ctx.dst by stripping iowarp::.
2. Open the source and apply ctx.range_off and ctx.range_size when they apply to the source.
3. Get or create the destination tag with cte_client_->AsyncGetOrCreateTag(tag_name) and CLIO_CO_AWAIT the future.
4. For each piece: allocate a buffer with CLIO_IPC->AllocateBuffer(size), fill it, and CLIO_CO_AWAIT an AsyncPutBlob(tag_id, blob_name, 0, size, buffer, score, context). Name blobs ctx.blob_prefix + index and size them by ctx.chunk_size (0 means your default) so the on_fault path can reproduce the same names and sizes.
5. Free the buffers, set error_code to 0, and CLIO_CO_RETURN. On any failure set a non-zero error_code; ParseOmni copies it into result_code_, sets error_message_ to Assimilator failed, and stops processing later transfers.

Blocking work that takes long, such as a remote download, should not block a runtime worker. The HTTP assimilator spawns a child process and the S3 assimilator streams through Poco inside a coroutine; the fault path runs ReadBlob on a dedicated thread pool for the same reason. The constructor receives a std::shared_ptr<clio::cte::core::Client>; the S3 assimilator also takes the shared connection pool.

Register the scheme in the factory

AssimilatorFactory::Get(src) in src/factory/assimilator_factory.cc extracts the scheme with GetUrlProtocol (the text before :// or ::) and returns a std::unique_ptr<BaseAssimilator>, or nullptr with a logged error. One special case comes first: a URL starting with https://app.globus.org goes to the Globus assimilator before the scheme is examined. Add a branch:

} else if (protocol == "myscheme") {
#ifdef CAE_ENABLE_MYSCHEME
  return std::make_unique<MySchemeAssimilator>(cte_client_);
#else
  HLOG(kError, "AssimilatorFactory: myscheme requested but support not compiled in. "
               "Rebuild with -DCAE_ENABLE_MYSCHEME=ON.");
  return nullptr;
#endif
}

Returning nullptr makes ParseOmni fail that transfer with result code -2 and the message No assimilator found for source, which is what clio_cae prints. Keep the #ifdef for any source that needs an SDK so a build without it still compiles and reports the missing option at run time.

Build wiring

• Declare the option in the top-level CMakeLists.txt next to CAE_ENABLE_S3, find the dependency there, and turn the option off with a warning when it is missing. That is the pattern S3 and GCS follow.
• Add the source file to the clio_cae_core_runtime target in core/CMakeLists.txt under the option, and add a compile definition with the option's name so the factory branch and the include see it.
• If the SDK must not be loaded into the runtime process, as with the AWS SDK, follow cae_s3_tool: a separate executable that links the SDK, spawned by the assimilator.
• Add a unit test under test/unit/. test_cloud_factory_guard.cc checks that each scheme returns an assimilator or nullptr according to the build flags, and is the place to add the new scheme.

Adding an OMNI key

If the source needs a new per-transfer key, add the field to AssimilationCtx and its serialize list, read it in LoadOmni in src/omni_loader.cc, and add the key name to KnownTransferKeys in the same file so the loader stops warning about it. If the field must survive an on_fault registration, add it to EncodeCtx and DecodeCtx in core_runtime.cc, which carry the context to other nodes. If Python callers need it, add a def_rw in context-exploration-engine/api/src/python_bindings.cc.