Struct Worker 

pub struct Worker {
    pub id: WorkerId,
    pub capabilities: Capabilities,
    /* private fields */
}

Worker state: manages execution of plans received from a coordinator.

Fields

id: WorkerId

The identity this worker registers and reports under.

capabilities: Capabilities

What this worker can run, announced to the coordinator at registration.

Implementations

impl Worker

pub fn new(id: impl Into<String>, capabilities: Capabilities) -> Self

A worker with an in-memory cache, an empty catalog, and per-worker temp/env directories derived from id. Filters arrive later, with the plans; the interpreter defaults to $SOMA_PYTHON, then python3 — see Worker::with_python for why that matters.

pub fn with_python(self, python: impl Into<String>) -> Self

Run filters in this interpreter rather than whatever python3 resolves to.

An embedding process should pass its own sys.executable: it is the interpreter that pickled the filters, so it is the only one certain to unpickle them.

pub fn with_cache(self, cache: Arc<dyn CacheStore>) -> Self

Set a custom cache store (e.g. tiered or shared).

pub fn with_data_store(self, store: Arc<dyn DataStore>) -> Self

Set a persistent DataStore (S3, Zarr, etc.) for large data references.

pub fn with_temp_dir(self, path: PathBuf) -> Self

Set a custom temp directory for HTTP bulk uploads.

pub fn temp_store(&self) -> &Arc<LocalDataStore>

Get the temp store (for HTTP upload endpoint).

pub fn register_filter( &mut self, node_id: impl Into<String>, filter: Box<dyn Filter>, )

Register a filter that this worker can execute.

pub fn get_filter(&self, node_id: &str) -> Option<Arc<dyn Filter>>

Get a filter by node_id.

pub fn catalog(&self) -> &NodeCatalog

The node catalog — what a stream driver is built over.

pub fn event_bus(&self) -> &Arc<EventBus>

The worker’s event bus.

pub fn cache(&self) -> &Arc<dyn CacheStore>

The worker’s cache store.

pub fn get_filter_state(&self, node_id: &str) -> Arc<Value>

Get trained state for a filter.

pub fn read_states(&self, node_ids: &[String]) -> Result<HashMap<String, Value>>

Trained state of one or more nodes, read from the Python process.

The four methods below back the wire messages of the same names. They existed on PythonProcess and in the daemon script from the start; what was missing was anything calling them, so soma-worker/src/server.rs answered all four with “not implemented for SubprocessFilter” and DataParallel could not run.

pub fn write_states(&mut self, states: &HashMap<String, Value>) -> Result<()>

Load states into the Python process (and the catalog beside it).

pub fn read_gradients( &self, node_ids: &[String], ) -> Result<HashMap<String, Value>>

Gradients currently held by each node’s parameters.

pub fn write_gradients(&self, gradients: &HashMap<String, Value>) -> Result<()>

Apply aggregated gradients to each node’s parameters.

pub fn set_filter_state(&mut self, node_id: &str, state: Value)

Set trained state for a filter.

pub fn wrap_output(&self, output: Value) -> OutputDelivery

Wrap output in the right delivery: inline for small, DataRef for large.

pub fn subscribe(&self) -> Receiver<Event>

Subscribe to execution events.

pub fn registration_message(&self) -> WorkerToCoordinator

Build a registration message.

pub fn execute_plan(&mut self, plan: &SerializedPlan) -> PlanResult

Execute a serialized plan.

If the plan contains serialized filter definitions, they are registered temporarily for this execution (alongside any pre-registered filters).

In Fit mode: fits each filter (topological order), stores trained states, then forwards to propagate outputs. Returns states so the client can cache them.

In Forward mode: executes the compiled plan directly.

pub fn matches_target(&self, target: &RemoteTarget) -> bool

Check if this worker matches a remote target.