OrderRequestProcessingPool

QuantConnect.Lean.Engine.TransactionHandlers.OrderRequestProcessingPool

OrderRequestProcessingPool(
    concurrency_enabled: bool,
    minimum_threads: int,
    maximum_threads: int,
    process_request: Callable[[OrderRequest], Any],
    on_error: Callable[[Exception], Any],
)

Bases: Object, IDisposable

Runs order requests on background worker threads that pull from a single shared queue. The pool grows on demand when the workers get saturated and keeps every request of an order processed in order.

Creates a threaded pool and starts its initial worker threads. When concurrency is enabled the pool starts at minimum_threads and grows on demand up to maximum_threads, otherwise it runs a single fixed worker thread.

Parameters:

Name	Type	Description	Default
concurrency_enabled	bool	True to grow the pool on demand, false to run a single worker thread	required
minimum_threads	int	The number of worker threads the pool starts with when growing	required
maximum_threads	int	The maximum number of worker threads the pool can grow to on demand	required
process_request	Callable[[OrderRequest], Any]	Handles a single order request	required
on_error	Callable[[Exception], Any]	Invoked when processing fails unexpectedly	required

is_active

is_active: bool

True while the pool is processing order requests, false once it has been shut down.

thread_count

thread_count: int

The number of worker threads currently running.

dispatch

dispatch(request: OrderRequest, order: Order) -> None

Dispatches an order request to be processed. If the order already has a request in flight, the new one waits parked so its worker runs it next and the order stays in arrival order. Otherwise it is queued for any worker to pick up, growing the pool first when every worker is already busy.

Parameters:

Name	Type	Description	Default
request	OrderRequest	The order request to process	required
order	Order	The order the request belongs to, used to keep its requests ordered	required

dispose

dispose() -> None

Stops every worker thread and waits for them to terminate, then releases the pool resources.

process_pending

process_pending() -> None

Drains the pending order requests on the calling thread. Only used in synchronous mode, where there are no worker threads and the caller pumps the single queue itself.

synchronous

synchronous(
    process_request: Callable[[OrderRequest], Any],
    on_error: Callable[[Exception], Any],
) -> OrderRequestProcessingPool

Creates a synchronous pool with no worker threads. Its single queue is drained on the caller thread via process_pending.

Parameters:

Name	Type	Description	Default
process_request	Callable[[OrderRequest], Any]	Handles a single order request	required
on_error	Callable[[Exception], Any]	Invoked when processing fails unexpectedly	required

wait_for_processing

wait_for_processing(timeout: timedelta) -> bool

Waits until no order has requests in flight, up to the given timeout. In practice only the synchronous early return runs. The threaded branch below is defensive, since its callers only reach it in backtesting where the pool is synchronous, so it never runs in a live deployment.

Parameters:

Name	Type	Description	Default
timeout	timedelta	The maximum time to wait	required

Returns:

Type	Description
bool	True if the pool was still processing when the timeout elapsed.