Compute Fabric

Every effect in id_effect passes through Compute Fabric at the runtime boundary. You write lazy Effect descriptions and declare a ResourcePolicy; Fabric decides where, when, and how concurrently work runs.

Declarative policy

#![allow(unused)]
fn main() {
use id_effect::compute::{ResourcePolicy, MetricMode, MetricPolicy, RebalanceStrategy};

let policy = ResourcePolicy {
  memory: MetricPolicy::new(MetricMode::Max { ceiling: 0.85 }),
  cpu: MetricPolicy::new(MetricMode::Max { ceiling: 1.0 }),
  rebalance: RebalanceStrategy::ThrottleAdmission,
};
}

Memory capped at 85% while CPU may use all cores is:

#![allow(unused)]
fn main() {
let policy = ResourcePolicy::memory_cap_max_cpu(0.85);
}

Unlimited memory with even CPU spread across workers:

#![allow(unused)]
fn main() {
let policy = ResourcePolicy::unlimited_memory_cpu_spread(0.25);
}

Supervisor loop

ComputeSupervisor polls TelemetryEngine (live sysinfo on host; mock in tests) and compares readings to policy:

  1. Monitor — TelemetrySnapshot { cpu_pct, mem_pct }
  2. Admit — AdmissionController adjusts permit count
  3. Place — fibers on FiberPool, collections via Rayon, I/O via Tokio
  4. Rebalance — throttle, shed, scale-out, or scale-in

Each tick emits a ComputeEvent (SupervisorTick) for observability alongside FiberEvent.

Example: memory at 61% with an 85% ceiling → headroom → admit many fibers. Memory at 86% → throttle admission.

Installing Fabric

#![allow(unused)]
fn main() {
use id_effect::{ThreadSleepRuntime, run_fork, succeed};
use id_effect::compute::ComputeFabric;

let fabric = ComputeFabric::memory_cap_max_cpu(0.85);
let rt = ThreadSleepRuntime::with_fabric(fabric);
// fibers spawned via rt use the shared pool + admission
}

run_with installs a default fabric for the duration of each application run. run_blocking refreshes the thread-local AdaptiveContext on every entry.

See example 120_compute_fabric_memory_cap.rs.

Adaptive parallelism

AdaptiveContext holds the current admission budget, Rayon thread count, and auto-parallel element threshold. Bulk dispatch via parallel_if_profitable consults should_parallelize_current(len) instead of a fixed 1024 default when Fabric is installed.

See also Implicit Parallelism for the full 0.4.0 API surface (map / map_effect + *_serial).

Effect binds

Independent ~ steps in an effect! block may run concurrently when the Effect Dependency Graph finds no data or capability conflict. Dependent steps stay ordered. Use #[effect(serial)] to opt out.

Example: 122_compute_fabric_effect_parallel.rs (Stream::map_effect under admission budget).

Streams and collections

  • Element threshold and Rayon pool size follow AdaptiveContext
  • Stream::map_effect caps concurrent effect mappers from the admission budget
  • Pure chunk ops (map, filter, …) use the same Fabric snapshot via parallel_if_profitable
  • CPU spread mode caps per-worker share via CpuSpreadBucket

Cluster

When local Fabric saturates, RebalanceStrategy::ScaleOut builds a FabricJobSpec via ComputeSupervisor::scale_out. Convert with id_effect_jobs::JobSpec::from_fabric and enqueue on a FabricJobRunner. Durable cross-node steps hook through DistributedStepJournal (stub).

Example: 123_compute_fabric_cluster.rs (two-node offload with MemoryJobRunner).

ClusterResourcePolicy combines global and per-node caps with PlacementMode (LocalFirst, Spread, Affinity).

Further reading