Concurrency control

Tune how many processing components can run at the same time with `max_inflight_components` or the `COCOINDEX_MAX_INFLIGHT_COMPONENTS` environment variable, and understand how the limit bounds a hierarchy of components without deadlocking it.

Version

v 1.0.14

Last reviewed

Nov 4, 2026

CocoIndex executes each processing component as a unit of concurrent work. By default, up to 1024 processing components can run at the same time per app. When components perform resource-intensive work (e.g., calling external APIs, running ML models), or each holds a sizeable piece of data (e.g., a file’s contents) while it runs, you may want a tighter limit.

Setting the limit

Set max_inflight_components on the App builder:

let app = App::builder("MyPipeline")
    .db_path("./coco_state")
    .max_inflight_components(4)
    .build()
    .await?;

EnvironmentBuilder has the same max_inflight_components setter; the limit is per app, and every app of the environment gets it.

With a limit of 4, at most 4 processing components execute at the same time. When a component finishes, the next pending one starts. A ctx.mount, ctx.scope / use_mount! or ctx.mount_each call waits while no slot is free, so a parent cannot run ahead of its children without bound: a loop that mounts a child per source item advances only as fast as children finish.

ctx.mount_each and ctx.mount_each_stream mount each item as a background child and admit the items at the pace slots free up; ctx.use_mount_each runs the items in the foreground and returns their values in input order. Both draw on the same limit.

A limit of 1 serializes all components — only one runs at a time.

You can also set the limit via the COCOINDEX_MAX_INFLIGHT_COMPONENTS environment variable:

export COCOINDEX_MAX_INFLIGHT_COMPONENTS=4

Precedence: the builder’s max_inflight_components > environment variable > default (1024). The variable is read when an app is created.

Nested mounts: a parent lends its slot

A parent component usually waits for its children: use_mount! / ctx.scope returns the child’s result, and a component’s own target states are committed only once all its children are done. If a waiting parent simply kept its slot, a small limit could fill up with waiting parents and leave no slot for the children they wait for.

Instead, a parent keeps its slot while it waits and lends it to one child at a time:

• The first child a parent mounts runs on the parent’s slot; it needs no free slot of its own.
• Further children take free slots from the pool, and their mount calls wait while the pool is empty.
• When the child running on the lent slot finishes, the slot passes to the parent’s next waiting child.

So a parent that is blocked — on its children, or on anything else after mounting one, such as an application-level tokio::sync::Semaphore — always has one child that can run, and a hierarchy of any depth completes even with a limit of 1. The limit bounds the components that execute at the same time; counting the parents waiting on them, the components in flight are bounded by the limit times the depth of the component hierarchy.

A few related rules:

• A component that awaits a function call — including a batched one (#[cocoindex::function(batching)]) — keeps its slot; only mounting a child lends it.
• When several parents mount the same component at once (two app updates overlapping, or several owners of a shared component), only the run that executes takes a slot. The others wait for it without a slot and reuse its result.
• Live components (ctx.auto_refresh, ctx.mount_live, ctx.mount_each_live) borrow the slot of the component that mounted them for their cycles and item updates, like its other children, for as long as it runs; afterwards they take slots from the pool like any other component.
• Deleting components (cleaning up a removed subtree, dropping an app) is bounded by the same limit.

Live components share the limit fairly

Live components update on their own schedules, and one of them can have far more work queued than another: a full re-index next to a refresh of a few changed items. The limit is shared fairly between independent updates, so a small update never waits for a large one’s whole backlog:

• An independent update is an app update, one cycle of a live component (each refresh of ctx.auto_refresh), or one item update of ctx.mount_each_live. Every component an update mounts belongs to that update.
• A component that finds a free slot takes it at once, and no slot is reserved, so an update that runs alone can use the whole limit.
• When no slot is free, a slot that frees up goes to whichever waiting side holds the fewest slots: first among the app’s updates and its live components, then, inside the live component that wins, among its own updates and the live components they mounted. Among equals it goes to the one that started waiting first, and within one update components start in the order they asked.
• A live component mounted by another live component’s update counts as part of that live component.

Fairness only decides which waiting component gets a slot that frees up. It never takes a slot from a running component, and the lending above works the same way.

Other concurrency controls

The component limit bounds how many components run at once. Four more controls each act on a smaller unit of work.

Concurrent items with ctx.map

ctx.map(items, f) runs f for every item at once, inside the current component, and collects the results in input order. Every started item runs to completion; then the first error in input order is returned. Use ctx.mount_each instead when each item should be its own component.

let results = ctx.map(tasks, |task| async move {
    Ok(format!("processed {task}"))
}).await?;

Batch size of a batched function

A batched function with max_batch_size = N receives up to N items at once; concurrent single-item calls are grouped into physical batches of at most that size. See Batching for the attribute, and Error::retry_with_smaller_batch for a batch the backend rejects.

#[cocoindex::function(memo, batching, max_batch_size = 32)]
async fn embed(ctx: &Ctx, texts: Vec<String>) -> Result<Vec<Vec<f32>>> { /* ... */ }

// Called with one item; the engine groups concurrent calls into batches.
let embedding: Vec<f32> = embed(ctx, text).await?;

GPU runners

runner = cocoindex::GPU makes each call hold one whole GPU from the process-wide pool while its body runs, so bodies on one GPU run one at a time. runner = cocoindex::GPURunner::fraction(0.5) holds half a GPU, so two calls share one. With batching, each physical batch holds the share instead. Inside the body, cocoindex::current_gpu(), current_gpus() and current_gpu_fraction() report the assignment.

The pool size is detected from COCOINDEX_NUM_GPUS, then CUDA_VISIBLE_DEVICES, then nvidia-smi, and falls back to one. configure_gpu_pool(n) overrides it; call it before any GPU function runs.

use std::num::NonZeroUsize;
use cocoindex::{GPU, GPURunner, configure_gpu_pool};

#[cocoindex::function(runner = GPU)]
async fn whole_gpu(_ctx: &Ctx, item: i64) -> Result<i64> {
    let device = cocoindex::current_gpu(); // Some(index) inside the body
    // ...
    Ok(item + 1)
}

#[cocoindex::function(memo, runner = GPURunner::fraction(0.5))]
async fn half_gpu(_ctx: &Ctx, item: i64) -> Result<i64> {
    assert_eq!(cocoindex::current_gpu_fraction(), Some(0.5));
    Ok(item * 10)
}

#[cocoindex::function(batching, max_batch_size = 2, runner = GPU)]
async fn gpu_batch(_ctx: &Ctx, items: Vec<i64>) -> Result<Vec<i64>> {
    Ok(items.into_iter().map(|item| item * 3).collect())
}

// Before the first GPU function runs: the pool has two GPUs.
configure_gpu_pool(NonZeroUsize::new(2).unwrap());

Rate limiting outbound calls

RateLimiter is a token bucket for calls to an external service, for example a source or target connector that must stay under an API’s rate limit. acquire(n) waits until n tokens are available; concurrent callers are served in FIFO order. Clones share one bucket.

use std::time::Duration;
use cocoindex::RateLimiter;

// 10 calls per second, with up to one second of tokens saved up while idle.
let limiter = RateLimiter::new(10.0, Duration::from_secs(1))?;
limiter.acquire(1).await?;