Struct PythonProcess 

pub struct PythonProcess { /* private fields */ }

A persistent Python child process that executes filter commands.

Implementations

impl PythonProcess

pub fn spawn( python_path: &str, filters: &[(String, Vec<u8>, bool)], ) -> Result<Self>

Spawn a Python daemon and load filters into it.

pub fn fit( &mut self, node_id: &str, data: &Value, y: Option<&Value>, ) -> Result<Value>

Fit the filter loaded under node_id on data (and optional labels y), returning what its fit returned — the trained state.

pub fn forward( &mut self, node_id: &str, data: &Value, state: &Value, ) -> Result<Value>

Run the filter’s forward on data with a previously trained state, returning its output.

pub fn composite_fit( &mut self, node_ids: &[String], data: &Value, y: Option<&Value>, ) -> Result<(Value, HashMap<String, Value>)>

Fit a chain of filters in one command: each trainable filter fits, then forwards to feed the next; if torch is importable the daemon follows with one autograd forward/backward pass over the chain. Returns the chain’s output plus each node’s serialized state (torch state_dict bytes when available, cloudpickle otherwise). One round-trip — intermediate values never cross the process boundary, and the autograd graph stays whole.

pub fn batched_fit( &mut self, node_ids: &[String], data: &Value, y: Option<&Value>, batch_size: usize, ) -> Result<(Value, HashMap<String, Value>)>

Batched fit: send full dataset + batch_size, daemon splits internally. Model loaded ONCE, batches processed in a loop.

pub fn composite_forward( &mut self, node_ids: &[String], data: &Value, ) -> Result<Value>

Forward data through a chain of filters, in order, inside one command — the composite counterpart of PythonProcess::forward.

pub fn get_state(&mut self, node_id: &str) -> Result<Value>

Extract one filter’s state.

A materialized DifferentiableFilter answers with its own state convention — Value::Json({"weights_b64": …}), the dict its forward reads back and the local fit path writes — so a state read off a worker is loadable by a local graph. Anything else is opaque bytes: a torch state_dict when the filter has one, the cloudpickled filter otherwise.

pub fn set_state(&mut self, node_id: &str, state: &Value) -> Result<()>

Load what PythonProcess::get_state produced back into the filter — how FedAvg-style aggregated states reach a worker.

Mirrors get_state in both of its forms: a Value::Json state goes as-is (and {"weights_b64": …} is loaded into the filter’s module), bytes go base64. Anything else is an encoding error.

pub fn get_gradients(&mut self, node_id: &str) -> Result<Value>

Collect the filter’s current gradients, one nested list per named parameter, for AllReduce aggregation.

Returns Value::Json({param_name: nested list}) rather than the torch pickle this used to send. The aggregator is in Rust ([somatize_runtime::strategy]), and a pickle is opaque to it: the average of two Value::Bytes blobs is not a thing that can be computed, so the round died at the aggregation step having done all the work. Plain JSON also makes the average independent of the torch version each worker happens to have installed.

A filter with no parameters, or no gradient on them, is an error rather than Value::Empty. Returning empty meant the average was taken over nothing and applied as nothing, and the round reported success — a data-parallel step that trained no one. The daemon says which of the three it is; this passes that on.

pub fn apply_gradients( &mut self, node_id: &str, gradients: &Value, ) -> Result<()>

Hand aggregated gradients (the post-AllReduce mean of what PythonProcess::get_gradients returned) to the filter: the daemon writes them onto the matching parameters and steps the optimizer, so the replica actually moves.

pub fn shutdown(&mut self)

Ask the daemon to exit its command loop. Best-effort — the reply is ignored, and Drop kills the child regardless.

pub fn node_ids(&self) -> &[String]

The node ids of the filters loaded into this process, in load order.

Trait Implementations

impl Drop for PythonProcess

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.