GitHub

Process Managers

Use a ProcessManager when you already have a worker definition and want to run many independent jobs through a bounded set of reusable workers. For normal Processes.jl usage, each worker is usually a reusable Process, but managers can also wrap custom worker objects.

The current manager API separates two ideas:

  • a portable recipe core that describes how one job is loaded, run, read, and synchronized;
  • an execution mode stored on the manager that decides how workers are scheduled.
manager = ProcessManager(recipe; nworkers = 4, execution = ThreadedWorkers(Dynamic()))
run!(manager, jobs)

Do not pass execution modes to run!. Construct the manager with the execution mode you want, then call run!(manager, jobs).

Terms

  • A job is one item of work, such as one sample, trajectory, or simulation case.
  • A worker is the reusable object that handles one job at a time.
  • A slot is the manager record around one worker. It stores the worker, the current job, the latest result, and the latest error.
  • A recipe tells the manager how to create workers and how each job should interact with one worker.
  • manager.state is manager-owned runtime data, commonly used for output buffers, counters, shared parameters, and optimizer state.

Portable Recipe Core

The standard lifecycle calls these recipe callbacks:

loadjob!(recipe, slot, job, manager)
providearguments(recipe, slot, job, manager)
afterjob!(recipe, slot, job, manager)
sync_to_state!(recipe, manager)

As named tuple fields:

recipe = (;
    makeworker = (idx, manager) -> Process(MyAlgo; repeats = 1),

    loadjob! = (slot, job, manager) -> begin
        ctx = context(slot.worker)[MyAlgo]
        ctx.value[] = job.value
        resetworker!(slot)
    end,

    providearguments = (slot, job, manager) -> (;),

    afterjob! = (slot, job, manager) -> begin
        ctx = context(slot.worker)[MyAlgo]
        push!(ctx.local_outputs, ctx.output[])
    end,

    sync_to_state! = manager -> begin
        for slot in slots(manager)
            ctx = context(slot.worker)[MyAlgo]
            append!(manager.state.outputs, ctx.local_outputs)
            empty!(ctx.local_outputs)
        end
    end,
)

loadjob! is the normal place to write job data into context(slot.worker) and reset the worker. providearguments returns runtime keyword arguments for the worker run; return nothing or omit it if no runtime arguments are needed. afterjob! reads one finished worker. sync_to_state! merges worker-local state into manager.state or another shared destination according to the manager's sync policy.

The standard lifecycle does not fall back to old names such as prepare!, runarguments, consume!, or flush!. Those names are kept only as compatibility lookup functions and should not be used for new portable recipes.

Execution Modes

Execution mode is stored on the manager:

ProcessManager(recipe; execution = PollingWorkers())
ProcessManager(recipe; execution = ThreadedWorkers(Dynamic()))
ProcessManager(recipe; execution = ThreadedWorkers(Static()))
ProcessManager(recipe; execution = ThreadedWorkers(Greedy()))
ProcessManager(recipe; execution = ChannelWorkers(; channel_size = 8))

If execution is omitted, the manager uses PollingWorkers(). A recipe can also provide a default execution field:

recipe = (;
    execution = ChannelWorkers(),
    makeworker = (idx, manager) -> Process(MyAlgo; repeats = 1),
    loadjob! = (slot, job, manager) -> nothing,
)

manager = ProcessManager(recipe)

PollingWorkers() uses the original dispatch/poll/drain scheduler. ThreadedWorkers(schedule) runs jobs through Threads.@threads and runs default Process workers inline to avoid one task spawn per job. ChannelWorkers() starts one long-lived worker task per slot and lets workers pull jobs from a channel until the job stream closes.

run!(manager, jobs, Dynamic()), run!(manager, jobs, Static()), run!(manager, jobs, Greedy()), and run!(manager, jobs, ChannelWorkers()) throw an ArgumentError. Put the mode in ProcessManager(...; execution = ...) instead.

The direct helpers runthreaded! and runchannel! remain available as lower level escape hatches, but normal user-facing code should prefer manager-owned execution and plain run!(manager, jobs).

Sync Policies

sync_policy controls when sync_to_state!(recipe, manager) is called:

ProcessManager(recipe; sync_policy = SyncAtEnd())
ProcessManager(recipe; sync_policy = NoSync())
ProcessManager(recipe; sync_policy = SyncEvery(32; drain = true))
  • SyncAtEnd() is the default. It runs all jobs, drains active workers, then calls sync_to_state! once.
  • NoSync() never calls sync_to_state! automatically.
  • SyncEvery(n; drain = true) calls sync_to_state! after every n completed worker runs. With drain = true, active workers are finalized before syncing.

Use afterjob! for per-job result reads. Use sync_to_state! when results or worker-local buffers should be merged in batches.

The old names FlushAtEnd, NoFlush, FlushEvery, and the constructor keyword flush_policy are compatibility aliases. New code should use SyncAtEnd, NoSync, SyncEvery, and sync_policy.

Manager State

manager.state is shared manager-owned data. Worker context is slot-local:

  • use manager.state for shared parameters, output buffers, counters, optimizer state, and history;
  • use context(slot.worker) for the current job, worker-local buffers, and temporary state.

Build manager state with initstate:

recipe = (;
    initstate = config -> (;
        params = Ref(config.initial_params),
        outputs = Float64[],
    ),
    makeworker = (idx, manager) -> Process(MyAlgo; repeats = 1),
)

manager = ProcessManager(
    recipe;
    config = (; initial_params = (w = 0.0, b = 0.0)),
)

or pass it directly:

manager = ProcessManager(recipe; state = (; outputs = Float64[]))

Worker Ownership

The usual form lets the manager create and own workers:

recipe = (;
    makeworker = (idx, manager) -> Process(MyAlgo; repeats = 1),
    loadjob! = (slot, job, manager) -> begin
        ctx = context(slot.worker)[MyAlgo]
        ctx.value[] = job.value
        resetworker!(slot)
    end,
)

manager = ProcessManager(recipe; nworkers = 4)

If workers = ... is passed, the manager wraps existing workers and does not create new contexts:

manager = ProcessManager(recipe; workers = existing_workers)

For manager-owned Process workers, makecontext can build separate per-slot contexts while sharing one worker template:

recipe = (;
    makeworker = (idx, manager) -> Process(MyAlgo; repeats = 1),
    makecontext = (idx, manager, template) -> begin
        initialized = init(getalgo(template), Init(MyAlgo; seed = idx))
        context(initialized)
    end,
)

Inspect slots and workers with:

slots(manager)
workers(manager)

Closing an owning manager closes its workers:

close(manager)

On-Demand Workers

Use OnDemandWorkers() when each job should construct its own worker:

recipe = (;
    initstate = config -> (; results = []),
    workername = (idx, manager, job) -> job.name,
    makeworker = (idx, manager, job) -> Process(
        job.algo,
        Init(job.algo; seed = job.seed);
        repeats = 1,
    ),
    afterjob! = (slot, job, manager) -> push!(
        manager.state.results,
        (; name = slot.name, output = job.readout(slot.worker)),
    ),
)

manager = ProcessManager(
    recipe;
    nworkers = 4,
    worker_lifecycle = OnDemandWorkers(destroy_after_finalize = false),
    worker_type = Process,
)

Pass worker_type when different jobs may return different concrete worker types. Use destroy_after_finalize = false if the last finished worker should remain in the slot for inspection.

Execution-Specific Hooks

The portable recipe core works across polling, threaded, and channel execution modes. Advanced callbacks can take over parts of the execution protocol:

  • start!(slot, job, manager) replaces the default worker launch.
  • isdone(slot, manager) customizes polling completion checks.
  • finalize!(slot, job, manager) replaces default worker finalization.
  • workerfinalizer(slot, job, manager) chooses a per-job finalizer.

These hooks are execution-specific. In particular, isdone is valid only with PollingWorkers(), because threaded and channel workers own the job until it is complete and do not poll per-job completion.

For normal result accumulation, prefer afterjob! over finalize!. Use finalize! only when worker finalization itself must be customized.

Runtime Inputs And Reinitialization

Use loadjob! for persistent context changes:

loadjob! = (slot, job, manager) -> begin
    ctx = context(slot.worker)[MyAlgo]
    ctx.value[] = job.value
    resetworker!(slot)
end

Use providearguments for loop-level runtime @input values:

providearguments = (slot, job, manager) -> (; temperature = job.temperature)

For Process workers, lifetime, repeats, and repeat returned by providearguments are launch controls. Other keys are passed as runtime inputs.

Use reinitworker! when the whole worker context should be rebuilt through the normal init pipeline:

loadjob! = (slot, job, manager) -> reinitworker!(
    slot,
    Input(MyAlgo, :start => job.start),
)

Use partialinitworker! when only selected context targets should be rebuilt:

loadjob! = (slot, job, manager) -> partialinitworker!(
    slot,
    Init(MyStep; base = job.base),
)

Chunked Inline Workers

runchunks! remains a specialized batching path for InlineChunkWorker. It is not part of the portable manager recipe core. Chunked recipes keep their chunk-specific callback names:

  • beforechunk!
  • resetexample!
  • loadexample!
  • afterexample!
  • afterchunk!

Use chunked inline workers when a chunk is semantically meaningful or when an InlineProcess should process many examples inside one manager job. If the goal is only to avoid per-job task spawning for ordinary Process workers, prefer ChannelWorkers() or ThreadedWorkers(...).

worker = InlineChunkWorker(InlineProcess(MyInlineAlgo; repeats = 1))

manager = ProcessManager(
    recipe;
    workers = (worker,),
    sync_policy = NoSync(),
    job_type = Vector{Int},
)

runchunks!(manager, dataset; chunksize = 128)

Threaded chunk scheduling still accepts a thread schedule directly:

runchunks!(manager, dataset, Dynamic(); chunksize = 128)
runchunks!(manager, dataset, Static(); chunksize = 128)
runchunks!(manager, dataset, Greedy(); chunksize = 128)

Public API Reference

Processes.ProcessManager — Type
ProcessManager(recipe; nworkers = Threads.nthreads(), workers = nothing,
               config = nothing, state = nothing,
               sync_policy = SyncAtEnd(), flush_policy = nothing,
               execution = nothing,
               worker_lifecycle = ReuseWorker(),
               worker_init = CopyFirstWorker(),
               worker_init_data = nothing,
               worker_type = nothing,
               name_type = Symbol,
               throw = true, poll_interval = 0.0,
               job_type = Any, scratch_type = Any,
               result_type = Any, error_type = Any)

Flexible worker orchestrator.

Recipes may be named tuples containing callbacks, or concrete objects that overload the callback functions below. The default worker protocol supports Process workers.

For each job, the portable lifecycle assigns a free slot, calls loadjob!, calls providearguments, runs the worker through the manager's execution mode, finalizes the worker, calls afterjob!, and eventually calls sync_to_state! according to the manager's sync policy.

When workers is omitted, the recipe must define makeworker. The manager calls makeworker once to create a template worker, then copies that template for the remaining slots. The default Process copy reuses the template task description and deep-copies the runtime context. Recipes can define makecontext(idx, manager, template) to build a separate Process context for each slot while keeping the template task description, or copyworker(template, idx, manager) when the whole worker copy must be custom. When workers is passed, the manager wraps those existing workers in slots and does not create new worker contexts.

Set worker_init = MakeEachWorker() to call makeworker(idx, manager) for every owned slot instead of copying slot 1. This avoids the manager's deepcopy-based default Process copy path entirely. Pass worker_init_data = data to provide one value per worker; callbacks may accept it as makeworker(idx, manager, data).

Use worker_lifecycle = OnDemandWorkers() when workers should be constructed from job data instead of upfront. That mode lives in OnDemandWorkers.jl and uses makeworker(idx, manager, job) for each dispatched job. Pass worker_type when the job may choose among different concrete worker types.

Slots have a public name field for logs and result lookup. Missing workername callbacks use names such as :worker_1. Pass name_type if names should use a type other than Symbol.

The job_type, scratch_type, result_type, and error_type keywords let latency-sensitive code make worker slot fields concrete. Leaving them as Any keeps the manager fully flexible.

source
Processes.WorkerSlot — Type
WorkerSlot

Transparent manager-owned slot around a reusable worker.

The worker field is intentionally public so recipes can inspect and mutate the underlying worker context directly.

source
Processes.PollingWorkers — Type
PollingWorkers()

Run the manager through the default polling scheduler, which starts worker jobs and polls active slots for completion.

source
Processes.ChannelWorkers — Type
ChannelWorkers(; channel_size = nothing)
ChannelWorkers(channel_size)

Manager execution mode that starts one long-lived worker task per slot and has those workers pull jobs from a channel until it closes. When channel_size is nothing, the internal channel uses one slot per manager worker.

source
Processes.SyncEvery — Type
SyncEvery(n; drain = true)

Invoke the recipe sync_to_state! callback after every n completed worker runs. When drain is true, all active workers are finalized before syncing.

source
Processes.loadjob! — Function
loadjob!(recipe, slot, job, manager)

Portable per-job callback that loads job into worker-local state before the worker is run.

source
Processes.providearguments — Function
providearguments(recipe, slot, job, manager)

Portable per-job callback that returns runtime keyword arguments for the implicit worker launch. Return a NamedTuple for run(slot.worker; kwargs...), or return nothing for no runtime keyword arguments.

source
Processes.afterjob! — Function
afterjob!(recipe, slot, job, manager)

Portable per-job callback for reading a finished worker and accumulating worker-local results.

source
Processes.sync_to_state! — Function
sync_to_state!(recipe, manager)

Portable manager-level callback controlled by the manager's SyncPolicy.

The manager calls this only after one or more worker runs have completed. It is the usual place to merge per-worker buffers into shared manager state, apply one batched parameter update, write accumulated results to an external destination, or clear worker-local accumulation buffers for the next batch of jobs.

sync_to_state! is not called per job. Its timing depends on sync_policy:

  • SyncAtEnd(): once after the run finishes.
  • SyncEvery(n; drain = true): after every n completed runs, with optional draining of active slots first.
  • NoSync(): never called automatically.
source
Processes.run! — Function
run!(manager, jobs)

Run jobs using the execution mode stored on manager.

source
run!(manager, jobs, mode)

Execution modes are manager-owned. Construct the manager with execution = mode instead of passing a mode to run!.

source
run!(manager, jobs, schedule)

Execution schedules are manager-owned through ThreadedWorkers(schedule).

source
Processes.runchannel! — Function
runchannel!(manager, jobs_channel)

Run a manager with one long-lived worker task per slot, pulling jobs from jobs_channel until that channel is closed and drained.

source
runchannel!(manager, jobs; channel_size = length(slots(manager)))

Run jobs through a bounded channel. The caller task feeds the channel once, while one spawned worker task per manager slot stays online until all jobs are consumed.

source
Processes.runchunks! — Function
runchunks!(manager, jobs; chunksize)

Dispatch jobs as vector chunks. Each chunk is one manager job, so an InlineChunkWorker processes many examples inside one spawned task.

source
runchunks!(manager, jobs, schedule; chunksize)

Dispatch jobs as chunks and run those chunks through runthreaded! with the given thread schedule. Each chunk is one manager job, and InlineChunkWorker slots execute their examples inline on the owning threaded iteration.

source
Processes.InlineChunkWorker — Type
InlineChunkWorker(process)

Worker wrapper for running a chunk of examples through one InlineProcess task.

The manager starts one task per chunk instead of one task per example. Recipe callbacks load each example into the inline process context, optionally reset selected state, and optionally consume the result after each inline run.

source
Processes.OnDemandWorkers — Type
OnDemandWorkers(; destroy_after_finalize = true)

Construct a fresh worker for each job by calling makeworker(idx, manager, job).

When destroy_after_finalize is true, the worker is cleaned up after consume! and release!. Recipe destroyworker! gets first chance to clean up; if it is missing, the manager tries recipe close!, then the default worker close behavior.

source
Processes.resetworker! — Function
resetworker!(slot)

Reset slot.worker by calling reset!(slot.worker) and return slot.

For a Process, this performs these exact mutations:

  • slot.worker.loopidx = 1
  • slot.worker.tickidx = 1
  • slot.worker.paused = false
  • slot.worker.shouldrun = true
  • slot.worker.starttime = nothing
  • slot.worker.endtime = nothing
  • reset!(getalgo(slot.worker))

It does not change slot.worker.runtime_context, slot.worker.task, or slot.worker.lastresult. It also does not rebuild or clear values stored in the process context. Context fields, arrays, buffers, and refs stay exactly as they were.

Call this from prepare! after you have loaded the next job into an existing context and want the next run to start from the first repeat/step again. Use reinitworker! or partialinitworker! when context values should be rebuilt through init.

source
Processes.reinitworker! — Function
reinitworker!(slot, inputs_overrides...; kwargs...)

Replace the whole context of slot.worker by running the normal process init pipeline, then return slot.

This is different from resetworker!: it rebuilds context values by calling init(getalgo(slot.worker), inputs_overrides...; lifetime = lifetime(slot.worker)) and then committing the resulting persistent context back onto the worker. Use it from prepare! when a job needs freshly initialized context state instead of reusing the previous context object.

source
Processes.partialinitworker! — Function
partialinitworker!(slot, inputs_overrides...)

Reinitialize only the context targets named by inputs_overrides, then return slot.

This keeps the existing process context as the starting point and runs partialinit for the affected algorithm or state entries. Use it when one part of the context should be rebuilt through its init method while the rest of the context should keep its current values. Concretely, it builds a lifecycle-wrapped algorithm from the current process context, runs partialinit(algo, inputs_overrides...), and assigns the returned context back to slot.worker.

source