nornir_pools

nornir_pools aims to provide a consistent interface around four different multi-threading and clustering libraries available to Python.

The use pattern for pools is:

1. Create a pool

2. add a task or process to the pool

3. save the task object returned

4. call wait or wait_return on the task object to fetch the output or raise exceptions

Steps 3 and 4 can be skipped if output is not required. In this case wait_completion can be called on the pool to delay until all tasks have completed. Note that in this pattern exceptions may be lost.

Pool Creation

Pool creation functions share a common signature

Get<X>Pool([Poolname=None, num_threads=None)

Return a pool of X type, listed below. Repeated calls using the same name returns the same pool

Parameters:

• Poolname (str) – Name of the pool to get or create. Passing “None” returns the global pool

• num_threads (int) – Number of tasks allowed to execute concurrently. Not honored by all pools at this time

Returns:

object derived from PoolBase

Return type:

PoolBase

nornir_pools.GetThreadPool(Poolname: str | None = None, num_threads: int | None = None) → IPool

Get or create a specific thread pool using vanilla python threads

nornir_pools.GetMultithreadingPool(Poolname: str | None = None, num_threads: int | None = None) → IPool

Get or create a specific thread pool to execute threads in other processes on the same computer using the multiprocessing library

nornir_pools.GetProcessPool(Poolname: str | None = None, num_threads: int | None = None) → IPool

Get or create a specific pool to invoke shell command processes on the same computer using the subprocess module

nornir_pools.GetParallelPythonPool(Poolname: str | None = None, num_threads: int | None = None) → IPool

Get or create a specific pool to invoke functions or shell command processes on a cluster using parallel python

Dashboard pool load

nornir_pools publishes pool_load MQTT events (via nornir_shared.pool_load.report_pool_load) so the build dashboard can show a thin Pools section. Ad hoc concurrent.futures work can use nornir_pools.InstrumentedThreadPoolExecutor or call report_pool_load directly.

Global pools

Most callers will not care about getting a specific pool. These functions always return the same pool.

nornir_pools.GetGlobalThreadPool() → IPool

Common pool for thread based tasks

nornir_pools.GetGlobalMultithreadingPool() → IPool

Common pool for multithreading module tasks, threads run in different python processes to work around the global interpreter lock

nornir_pools.GetGlobalProcessPool() → IPool

Common pool for processes on the local machine

nornir_pools.GetGlobalClusterPool() → IPool

Get the common pool for placing tasks on the cluster

Pool Objects

class nornir_pools.poolbase.ABC

Bases: object

Helper class that provides a standard way to create an ABC using inheritance.

class nornir_pools.poolbase.IPool

Bases: ABC

abstractmethod add_process(name: str, func: Callable[[...], Any] | str, *args, **kwargs) → TaskWithEvent

Invoke a process on the pool. This function creates a task using name and then invokes pythons subprocess

Parameters:

• name (str) – Friendly name of the task. Non-unique

• func (function) – Process name to invoke using subprocess

Returns:

task object

Return type:

task

abstractmethod add_task(name: str, func: Callable[[...], Any], *args, **kwargs) → Task

Call a python function on the pool

Parameters:

• name (str) – Friendly name of the task. Non-unique

• func (function) – Python function pointer to invoke on the pool

Returns:

task object

Return type:

task

abstract property name: str

abstract property num_active_tasks: int

abstractmethod shutdown()

The pool waits for all tasks to complete and frees any resources such as threads in a thread pool

abstractmethod wait_completion()

Blocks until all tasks have completed

class nornir_pools.poolbase.LocalThreadPoolBase(*args, **kwargs)

Bases: PoolBase, ABC

Base class for pools that rely on local threads and a queue to dispatch jobs

AtExitLock = <unlocked _thread.lock object>

AtExitRegisteredWaitTime = 0

classmethod TryRegisterAtExit(wait_time: float)

Register a wait atexit so we don’t leave threads alive when the program exits and get an error message

WorkerCheckInterval = 1

property active_tasks: int

Tasks currently executing. Override when in-flight is tracked.

add_threads_if_needed()

Grow the pool towards one thread per outstanding task, capped at num_threads.

Outstanding work is queued + inflight, and that choice is what makes this race-free. The previous version sized against qsize() + 1 and then re-checked not self.tasks.empty() before each creation, so a worker dequeueing the last item between the two cancelled a thread it had already decided was needed – and nothing revisited the decision until the next add_task, which for the final task of a batch is never. Measured on ThreadPool(4) with 4 blocking tasks, deterministically 5 of 5 trials: 3 threads, 1 task left queued, and only 3 of the 4 tasks running at once.

A dequeue moves a task from queued to inflight, so it leaves the sum unchanged: the event that used to invalidate the target now cannot. Counting inflight also fixes the sizing itself, since qsize() + 1 ignored busy threads and so could only ever approach capacity one thread per submission.

Reading qsize() before active_tasks is deliberate. The two reads are not atomic, so a dequeue landing between them counts one task in both terms; erring towards one extra thread is the direction that serves the throughput this method exists for, and min with max_t keeps the num_threads contract intact regardless. Idle threads are not over-created either: a pool of N idle workers receiving one task computes a target of 1 and creates nothing.

abstractmethod add_worker_thread()

called_from_pool_worker() → bool

True when the calling thread is one of this pool’s own workers.

Worker threads record the pool they serve, so this is exact rather than a guess based on thread names or identity reuse.

enqueue_task(entry) → None

Queue a task, applying capacity only to producers outside the pool.

A worker submitting onto its own pool is never made to wait. Throttling it would block the thread that has to drain the queue for the wait to end, which is a deadlock no timeout can recover from. Outside producers still get backpressure, which is what the bound is for.

The room check and the put are deliberately not atomic, so concurrent producers can overshoot the capacity slightly. It is a backpressure threshold, not an invariant, and making it exact would reintroduce a lock held across a put.

mark_task_finished() → None

Called by a worker when a dequeued task finishes.

mark_task_started() → None

Called by a worker when it dequeues a task to run.

property max_workers: int | None

Worker capacity when known.

nested_wait_guard(task_name: str)

Report a worker of this pool blocking on a task belonging to this same pool.

#85 removed the queue-capacity half of this hazard, so a worker submitting onto its own pool is no longer throttled. Submitting is only half the pattern: a worker that then waits for its child consumes the very worker the child needs, because the child is queued behind the parent that is still occupying its thread. With every worker doing this the pool cannot drain and is wedged permanently.

Nothing here changes that. Work-stealing or growing the pool would, but both are material behavioural changes – re-entrant execution on a thread that is mid-task can surprise anything holding a lock, and growing breaks the num_threads contract that add_threads_if_needed accounts against. This makes the fragility visible instead: a warning while it is merely wasteful, and an error once it is provably a deadlock, naming the pool so the report points at the fix.

A no-op unless the caller really is one of this pool’s workers, so the ordinary outside-caller wait pays only one attribute lookup.

property num_active_tasks: int

property queued_tasks: int

Tasks waiting to start. Override in subclasses with a queue.

remove_finished_threads()

shutdown()

The pool waits for all tasks to complete and frees any resources such as threads in a thread pool

wait_completion()

Wait for completion of all the tasks in the queue. Note that wait or wait_return must be called on each task to detect exceptions if there were any

class nornir_pools.poolbase.PoolBase(*args, **kwargs)

Bases: IPool

Pool objects provide the interface to create tasks on the pool.

PrintActiveTaskCount()

TryReportActiveTaskCount()

Report the current job count if we haven’t reported it recently

TryReportPoolLoad() → None

Publish dashboard pool_load for this pool (throttled in shared helper).

property active_tasks: int

Tasks currently executing. Override when in-flight is tracked.

property logger

property max_workers: int | None

Worker capacity when known.

property name: str

property queued_tasks: int

Tasks waiting to start. Override in subclasses with a queue.

nornir_pools.poolbase.abstractmethod(funcobj)

A decorator indicating abstract methods.

Requires that the metaclass is ABCMeta or derived from it. A class that has a metaclass derived from ABCMeta cannot be instantiated unless all of its abstract methods are overridden. The abstract methods can be called using any of the normal ‘super’ call mechanisms. abstractmethod() may be used to declare abstract methods for properties and descriptors.

Usage:

class C(metaclass=ABCMeta):

@abstractmethod def my_abstract_method(self, arg1, arg2, argN):

…

Task Objects

class nornir_pools.task.Task(name, *args, **kwargs)

Bases: ABC

Represents a task assigned to a pool. Responsible for allowing the caller to wait for task completion, raising any exceptions, and returning data from the call. Task objects are created by adding tasks or processes to the pools. They are not intended to be created directly by callers.

args: tuple[Any, ...]

property elapsed_time: float

If the task is completed, returns the time to completion. If the task is still running, returns the time from the start of the task to the current time

property elapsed_time_str: str

SS.ssss

Type:

Formats the elapsed time as a string in the format HH

Type:

MM

classmethod generate_id() → int

Returns the next unique ID for a task. Thread safe.

abstractmethod iscompleted() → bool

Non-blocking test to determine if task has completed. No exception is raised if the task raised an exception during execution until wait or wait_return is called.

Returns:

True if the task is completed, otherwise False

Return type:

bool

kwargs: dict[str, Any]

name: str

property pool

The pool that created this task, or None if it did not record itself.

Held weakly: a task object outliving its pool must not keep the pool, and its worker threads, alive. Set by the creating pool rather than passed to __init__, because *args/**kwargs there belong to the task’s function.

Knowing the owning pool is what lets a blocking wait notice that it is running on one of that same pool’s workers, which starves the worker the awaited task needs.

set_completion_time()

Marks the current time as the task completion time. Will only set completion time on the first call.

task_end_time: float | None = None

property task_id: int

Unique ID of task

task_start_time: float

abstractmethod wait()

Wait for task to complete, does not return a value

Raises:

Exception – Exceptions raised during task execution are re-raised on the thread calling wait

abstractmethod wait_return() → Any

Wait for task to complete and return the value

Returns:

The output of the task function or the stdout text of a called process

Raises:

Exception – Exceptions raised during task execution are re-raised on the thread calling wait_return

Pool lifecycle

It is not necessary to perform cleanup during normal scripting; ClosePools() runs automatically at process exit via atexit. Long-running pipelines (for example nornir_buildmanager) enqueue work on global pools across many stages. Waiting for tasks and shutting down pools are separate operations:

• Wait — block until queued tasks finish. Pools stay registered and accept new work.

• Close — shut down workers and remove the pool from the registry (after waiting, unless skip_wait is used).

Thread-kind pools (ThreadPool, subprocess ProcessPool, SerialPool) run inside the parent process. Process-kind pools (MultiprocessThreadPool, LocalMachinePool, ParallelPythonProcess_Pool, cluster pools) keep OS worker processes alive. Spawning those workers is expensive, so production code keeps process pools warm across pipeline stages and recreates thread pools at stage boundaries instead.

class nornir_pools.poolbase.PoolKind

Classifies a pool for selective wait/shutdown helpers. THREAD pools are in-process; PROCESS pools use separate worker processes.

Stage boundaries (recommended)

At the end of a pipeline stage that used pools (import finished, transform saved, overlay assembly complete, PipelineManager.Execute returned, etc.), call ReleaseStagePools():

1. Wait for all thread- and process-pool tasks to finish so stage outputs are safe to read.

2. Shut down thread-kind pools so idle in-process workers do not linger.

3. Leave process-kind pools registered so the next stage reuses warm workers.

Use WaitOnAllPools() when you only need synchronization and will enqueue more work immediately without releasing thread workers. Use ClosePools() once at process exit or test teardown when every pool must be destroyed.

Environment variables

NORNIR_POOL_DIAG

When set to 1, true, or yes, log pool names, kinds, and active task counts on each lifecycle call.

NORNIR_KEEP_PROCESS_POOLS

When set, CloseProcessPools() skips process-pool shutdown. Normally unnecessary because ReleaseStagePools() already preserves process pools across stages; use only for advanced debugging or custom teardown.

Optimization

On windows there is significant overhead to passing parameters to multiprocessing jobs. To address this I added pickle overrides to objects being marshalled. I also removed as many global initializations as I could from modules loaded by the tasks.

nornir_pools.ApplyOSThreadLimit(num_threads: int | None) → int | None

:return The minimum of the maximum number of threads on the OS, the MAX_PYTHON_THREADS environment variable, or the requested num_threads parameter

nornir_pools.ClosePools(timeout: float | None = None) → None

Shut down all known pools (wait for tasks, then destroy all workers).

Registered as an atexit handler. Tests and short scripts should also call this explicitly at teardown. Long pipelines should use ReleaseStagePools() between stages and reserve this for final cleanup.

Parameters:

timeout – Seconds to wait per pool before giving up. None (the default) waits indefinitely, matching historical behaviour. Tests should pass a bound so a leaked worker becomes a TimeoutError of that test rather than hanging the whole suite. When a timeout is given it is also applied to each pool’s shutdown drain (via NORNIR_POOL_SHUTDOWN_TIMEOUT), so a stuck worker is terminated rather than blocking teardown a second time. Pools are shut down even when the wait times out, so a later caller is not left facing the same stuck registry entry.

nornir_pools.CloseProcessPools() → None

Shut down process-kind pools after waiting for their tasks.

No-op when NORNIR_KEEP_PROCESS_POOLS is set. Production pipelines normally call ReleaseStagePools() at stage boundaries instead, which leaves process pools registered.

nornir_pools.CloseThreadPools(*, skip_wait: bool = False) → None

Shut down thread-kind pools after optionally waiting for their tasks.

Parameters:

skip_wait – When True, assume callers already waited (as ReleaseStagePools() does via WaitOnAllPools()).

class nornir_pools.Enum(new_class_name, /, names, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: object

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

• attribute access:

  >>> Color.RED
  <Color.RED: 1>
  

• value lookup:

  >>> Color(1)
  <Color.RED: 1>
  

• name lookup:

  >>> Color['RED']
  <Color.RED: 1>
  

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3

>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

nornir_pools.FastClosePools() → None

Shut down all pools, force-terminating process workers when supported.

Intended for faster test teardown when graceful multiprocessing join is too slow. Production pipelines should use ReleaseStagePools() between stages and ClosePools() at exit instead.

nornir_pools.GetAndCreateProfileDataFileName()

nornir_pools.GetAndCreateProfileDataPath()

nornir_pools.GetGlobalClusterPool() → IPool

Get the common pool for placing tasks on the cluster

nornir_pools.GetGlobalLocalMachinePool() → IPool

Common pool for launching other processes for threads or executables. Combines multithreading and process pool interface.

nornir_pools.GetGlobalMultithreadingPool() → IPool

Common pool for multithreading module tasks, threads run in different python processes to work around the global interpreter lock

nornir_pools.GetGlobalProcessPool() → IPool

Common pool for processes on the local machine

nornir_pools.GetGlobalSerialPool() → IPool

Common pool for processes on the local machine

nornir_pools.GetGlobalThreadPool() → IPool

Common pool for thread based tasks

nornir_pools.GetLocalMachinePool(Poolname: str | None = None, num_threads: int | None = None, is_global=False) → IPool

nornir_pools.GetMultithreadingPool(Poolname: str | None = None, num_threads: int | None = None) → IPool

Get or create a specific thread pool to execute threads in other processes on the same computer using the multiprocessing library

nornir_pools.GetParallelPythonPool(Poolname: str | None = None, num_threads: int | None = None) → IPool

Get or create a specific pool to invoke functions or shell command processes on a cluster using parallel python

nornir_pools.GetProcessPool(Poolname: str | None = None, num_threads: int | None = None) → IPool

Get or create a specific pool to invoke shell command processes on the same computer using the subprocess module

nornir_pools.GetSerialPool(Poolname: str | None = None, num_threads: int | None = None) → IPool

Get or create a specific thread pool using vanilla python threads

nornir_pools.GetThreadPool(Poolname: str | None = None, num_threads: int | None = None) → IPool

Get or create a specific thread pool using vanilla python threads

class nornir_pools.IPool

Bases: ABC

abstractmethod add_process(name: str, func: Callable[[...], Any] | str, *args, **kwargs) → TaskWithEvent

Invoke a process on the pool. This function creates a task using name and then invokes pythons subprocess

Parameters:

• name (str) – Friendly name of the task. Non-unique

• func (function) – Process name to invoke using subprocess

Returns:

task object

Return type:

task

abstractmethod add_task(name: str, func: Callable[[...], Any], *args, **kwargs) → Task

Call a python function on the pool

Parameters:

• name (str) – Friendly name of the task. Non-unique

• func (function) – Python function pointer to invoke on the pool

Returns:

task object

Return type:

task

abstract property name: str

abstract property num_active_tasks: int

abstractmethod shutdown()

The pool waits for all tasks to complete and frees any resources such as threads in a thread pool

abstractmethod wait_completion()

Blocks until all tasks have completed

class nornir_pools.InstrumentedThreadPoolExecutor(name: str, max_workers: int | None = None, *args: Any, **kwargs: Any)

Bases: ThreadPoolExecutor

ThreadPoolExecutor that reports load under a stable pool name.

Use this (or call nornir_shared.pool_load.report_pool_load() manually) so ad hoc executors appear in the dashboard Pools section alongside nornir_pools pools.

submit(fn: Callable[[...], Any], /, *args: Any, **kwargs: Any) → Future

Submits a callable to be executed with the given arguments.

Schedules the callable to be executed as fn(*args, **kwargs) and returns a Future instance representing the execution of the callable.

Returns:

A Future representing the given call.

nornir_pools.IsParallelPythonAvailable()

nornir_pools.MergeProfilerStats(root_output_dir: str, profile_dir: str, pool_name: str)

Called by atexit. Merges all *.profile files in the profile_dir into a single .profile file

class nornir_pools.ParamSpec

Bases: object

Parameter specification variable.

The preferred way to construct a parameter specification is via the dedicated syntax for generic functions, classes, and type aliases, where the use of ‘**’ creates a parameter specification:

type IntFunc[**P] = Callable[P, int]

The following syntax creates a parameter specification that defaults to a callable accepting two positional-only arguments of types int and str:

type IntFuncDefault[**P = [int, str]] = Callable[P, int]

For compatibility with Python 3.11 and earlier, ParamSpec objects can also be created as follows:

P = ParamSpec('P')
DefaultP = ParamSpec('DefaultP', default=[int, str])

Parameter specification variables exist primarily for the benefit of static type checkers. They are used to forward the parameter types of one callable to another callable, a pattern commonly found in higher-order functions and decorators. They are only valid when used in Concatenate, or as the first argument to Callable, or as parameters for user-defined Generics. See class Generic for more information on generic types.

An example for annotating a decorator:

def add_logging[**P, T](f: Callable[P, T]) -> Callable[P, T]:
    '''A type-safe decorator to add logging to a function.'''
    def inner(*args: P.args, **kwargs: P.kwargs) -> T:
        logging.info(f'{f.__name__} was called')
        return f(*args, **kwargs)
    return inner

@add_logging
def add_two(x: float, y: float) -> float:
    '''Add two numbers together.'''
    return x + y

Parameter specification variables can be introspected. e.g.:

>>> P = ParamSpec("P")
>>> P.__name__
'P'

Note that only parameter specification variables defined in the global scope can be pickled.

args

Represents positional arguments.

evaluate_default

has_default()

kwargs

Represents keyword arguments.

class nornir_pools.PoolFactory(*args, **kwargs)

Bases: Protocol[_PoolFactoryParams]

class nornir_pools.PoolKind(*values)

Bases: Enum

Whether a pool uses in-process workers or separate OS worker processes.

Used by selective wait/shutdown helpers. THREAD includes vanilla ThreadPool, subprocess ProcessPool, and SerialPool. PROCESS includes MultiprocessThreadPool, LocalMachinePool, and cluster backends.

PROCESS = 'process'

THREAD = 'thread'

class nornir_pools.Protocol

Bases: Generic

Base class for protocol classes.

Protocol classes are defined as:

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example:

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as:

class GenProto[T](Protocol):
    def meth(self) -> T:
        ...

nornir_pools.ReleaseStagePools() → None

Synchronize a pipeline stage and release in-process pool workers.

Waits for all thread- and process-pool tasks to complete, then shuts down thread-kind pools only. Process-kind pools stay registered so later stages reuse warm worker processes instead of paying fork/spawn cost again.

Call at production stage boundaries (import complete, registration transform saved, stos overlay assembly done, pipeline Execute finished). Full pool teardown belongs in ClosePools() at process or test exit.

class nornir_pools.Task(name, *args, **kwargs)

Bases: ABC

Represents a task assigned to a pool. Responsible for allowing the caller to wait for task completion, raising any exceptions, and returning data from the call. Task objects are created by adding tasks or processes to the pools. They are not intended to be created directly by callers.

args: tuple[Any, ...]

property elapsed_time: float

If the task is completed, returns the time to completion. If the task is still running, returns the time from the start of the task to the current time

property elapsed_time_str: str

SS.ssss

Type:

Formats the elapsed time as a string in the format HH

Type:

MM

classmethod generate_id() → int

Returns the next unique ID for a task. Thread safe.

abstractmethod iscompleted() → bool

Non-blocking test to determine if task has completed. No exception is raised if the task raised an exception during execution until wait or wait_return is called.

Returns:

True if the task is completed, otherwise False

Return type:

bool

kwargs: dict[str, Any]

name: str

property pool

The pool that created this task, or None if it did not record itself.

Held weakly: a task object outliving its pool must not keep the pool, and its worker threads, alive. Set by the creating pool rather than passed to __init__, because *args/**kwargs there belong to the task’s function.

Knowing the owning pool is what lets a blocking wait notice that it is running on one of that same pool’s workers, which starves the worker the awaited task needs.

set_completion_time()

Marks the current time as the task completion time. Will only set completion time on the first call.

task_end_time: float | None = None

property task_id: int

Unique ID of task

task_start_time: float

abstractmethod wait()

Wait for task to complete, does not return a value

Raises:

Exception – Exceptions raised during task execution are re-raised on the thread calling wait

abstractmethod wait_return() → Any

Wait for task to complete and return the value

Returns:

The output of the task function or the stdout text of a called process

Raises:

Exception – Exceptions raised during task execution are re-raised on the thread calling wait_return

nornir_pools.WaitOnAllPools() → None

Block until all known pools finish queued work without shutting them down.

Use when later code on the same thread will enqueue more tasks and you do not need to release idle thread-pool workers. Pipeline stage boundaries should prefer ReleaseStagePools() instead.

nornir_pools.WaitOnProcessPools() → None

Block until process-kind pools finish without shutting them down.

Process-kind pools use OS worker processes (see PoolKind). Waiting does not destroy workers; they remain available for new tasks.

nornir_pools.WaitOnThreadPools() → None

Block until thread-kind pools finish without shutting them down.

Thread-kind pools are in-process (see PoolKind). This does not wait on multiprocessing worker pools.

nornir_pools.aggregate_profiler_data(output_path)

nornir_pools.end_profiling()

nornir_pools.get_or_create_shared_memory_manager(authkey: bytes | None = None)

Obtain a SharedMemoryManager for inter-process buffers.

Call from the parent before workers start. The listener address/authkey are published via SHARED_MEMORY_SERVER_ADDRESS / SHARED_MEMORY_AUTHKEY so children can connect to the same manager.

nornir_pools.init_pool_process(logging_queue=None, logging_level=None, the_lock=None)

Worker initializer: configure queue logging and force the NumPy backend in workers.

Optional the_lock sets shared_lock (legacy API).

nornir_pools.invoke_with_profiler(func, *args, **kwargs)

nornir_pools.start_profiling()