From 5fbe4e1f7346f7a0ee280def82f61c9c4fe970c5 Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Fri, 28 Aug 2026 10:27:18 -0400 Subject: [PATCH 01/14] Harden transaction processing pool shutdown Bound the pool's dispose so a request handler pinned in a blocking call cannot hold up the engine shutdown, and report the requests that were dropped without being processed: - Join the worker threads against a single shared deadline instead of a full timeout per thread, and interrupt the ones that do not stop - Cancel up front so workers exit instead of draining a backlog that should not be sent to the brokerage while shutting down - Drop the requests still queued or parked once the workers are stopped, reporting each one so the transaction handler invalidates its order before the final results are sent, instead of leaving it as New --- Common/Interfaces/IBusyCollection.cs | 7 + Common/Util/BusyBlockingCollection.cs | 10 + Common/Util/BusyCollection.cs | 10 + .../BrokerageTransactionHandler.cs | 21 +- .../OrderRequestProcessingPool.cs | 128 +++++++++-- .../OrderRequestProcessingPoolTests.cs | 200 ++++++++++++++++++ 6 files changed, 358 insertions(+), 18 deletions(-) create mode 100644 Tests/Engine/BrokerageTransactionHandlerTests/OrderRequestProcessingPoolTests.cs diff --git a/Common/Interfaces/IBusyCollection.cs b/Common/Interfaces/IBusyCollection.cs index 29b5d73d9fec..f62eb3fcaf06 100644 --- a/Common/Interfaces/IBusyCollection.cs +++ b/Common/Interfaces/IBusyCollection.cs @@ -72,5 +72,12 @@ public interface IBusyCollection : IDisposable /// A cancellation token to observer /// An enumerable that removes and returns items from the collection IEnumerable GetConsumingEnumerable(CancellationToken cancellationToken); + + /// + /// Attempts to take an item from this collection without blocking + /// + /// The item taken, or default when none was available + /// True if an item was taken + bool TryTake(out T item); } } diff --git a/Common/Util/BusyBlockingCollection.cs b/Common/Util/BusyBlockingCollection.cs index 70053bb275ac..7fb94abd8b5e 100644 --- a/Common/Util/BusyBlockingCollection.cs +++ b/Common/Util/BusyBlockingCollection.cs @@ -124,6 +124,16 @@ public void CompleteAdding() _collection.CompleteAdding(); } + /// + /// Attempts to take an item from this collection without blocking + /// + /// The item taken, or default when none was available + /// True if an item was taken + public bool TryTake(out T item) + { + return _collection.TryTake(out item); + } + /// /// Provides a consuming enumerable for items in this collection. /// diff --git a/Common/Util/BusyCollection.cs b/Common/Util/BusyCollection.cs index d321b713cfd8..0d63c311952f 100644 --- a/Common/Util/BusyCollection.cs +++ b/Common/Util/BusyCollection.cs @@ -128,5 +128,15 @@ public void CompleteAdding() { _completedAdding = true; } + + /// + /// Attempts to take an item from this collection without blocking + /// + /// The item taken, or default when none was available + /// True if an item was taken + public bool TryTake(out T item) + { + return _collection.TryDequeue(out item); + } } } diff --git a/Engine/TransactionHandlers/BrokerageTransactionHandler.cs b/Engine/TransactionHandlers/BrokerageTransactionHandler.cs index fa9616d65623..b8aeb17f95b1 100644 --- a/Engine/TransactionHandlers/BrokerageTransactionHandler.cs +++ b/Engine/TransactionHandlers/BrokerageTransactionHandler.cs @@ -276,8 +276,9 @@ protected virtual void InitializeTransactionThread() // backtesting drains a single queue synchronously on the algorithm thread, live deployments use // background worker threads: a single one, or growing on demand up to the maximum when concurrent. _threadPool = SynchronousProcessing - ? OrderRequestProcessingPool.Synchronous(processRequest, onError) - : new OrderRequestProcessingPool(ConcurrencyEnabled, MinimumTransactionThreads, MaximumTransactionThreads, processRequest, onError); + ? OrderRequestProcessingPool.Synchronous(processRequest, onError, InvalidateDroppedRequest) + : new OrderRequestProcessingPool(ConcurrencyEnabled, MinimumTransactionThreads, MaximumTransactionThreads, + processRequest, onError, InvalidateDroppedRequest); } #region Order Request Processing @@ -1930,6 +1931,22 @@ private void InvalidateOrders(List orders, string message) } } + /// + /// Handles a request the processing pool dropped at shutdown without processing it. Invalidates a dropped + /// submit's order so it doesn't linger as new in the final results, before they are sent. + /// + private void InvalidateDroppedRequest(OrderRequest request) + { + var message = "The order was never sent to the brokerage: the engine was stopped while the request was queued"; + request.SetResponse(OrderResponse.Error(request, OrderResponseErrorCode.ProcessingError, message)); + + // only a dropped submit leaves an order that was never placed, updates and cancels target one already sent + if (request.OrderRequestType == OrderRequestType.Submit && _openOrders.TryGetValue(request.OrderId, out var openOrder)) + { + InvalidateOrders(new List { openOrder.Order }, message); + } + } + private void SendWarningOnPriceChange(string priceType, decimal priceRound, decimal priceOriginal) { if (!priceOriginal.Equals(priceRound) && !_hasLoggedPriceRoundingWarning) diff --git a/Engine/TransactionHandlers/OrderRequestProcessingPool.cs b/Engine/TransactionHandlers/OrderRequestProcessingPool.cs index 1174e6fe3ca7..d5c1d4fd66cd 100644 --- a/Engine/TransactionHandlers/OrderRequestProcessingPool.cs +++ b/Engine/TransactionHandlers/OrderRequestProcessingPool.cs @@ -15,6 +15,8 @@ using System; using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; using System.Threading; using QuantConnect.Interfaces; using QuantConnect.Logging; @@ -37,8 +39,16 @@ namespace QuantConnect.Lean.Engine.TransactionHandlers /// public class OrderRequestProcessingPool : IDisposable { - // maximum time to wait for each worker thread to stop when disposing the pool - private static readonly TimeSpan ShutdownTimeout = TimeSpan.FromSeconds(60); + /// + /// Maximum total time to wait for all the worker threads to stop when disposing the pool. Settable for tests. + /// + internal TimeSpan ShutdownTimeout { get; set; } = TimeSpan.FromSeconds(60); + + /// + /// Time granted to each interrupted worker to observe the interrupt when disposing. Settable for tests. + /// + internal TimeSpan InterruptTimeout { get; set; } = TimeSpan.FromSeconds(5); + // the shared queue of requests cleared to run. every worker pulls from here so the load stays balanced private readonly IBusyCollection _readyQueue; private readonly List _threads; @@ -54,13 +64,15 @@ public class OrderRequestProcessingPool : IDisposable // true when a single consumer drains the queue (synchronous or a single fixed worker), which already // preserves arrival order across all orders so the per-order serialization is skipped entirely private readonly bool _singleConsumer; - // set under the lock when shutting down so the pool stops growing while the queue drains, before the - // cancellation token is cancelled as the final hard stop - private bool _shuttingDown; + // set under the lock when shutting down so the pool stops growing while the pending requests are dropped, + // before the cancellation token is cancelled. volatile, the workers read it lock free while draining + private volatile bool _shuttingDown; // number of workers currently processing a request, used to decide when the pool is saturated private int _busyWorkers; private readonly Action _processRequest; private readonly Action _onError; + // reports each request dropped at shutdown without being processed, may be null + private readonly Action _onDropped; private readonly CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource(); /// @@ -92,12 +104,14 @@ public int ThreadCount /// The maximum number of worker threads the pool can grow to on demand /// Handles a single order request /// Invoked when processing fails unexpectedly + /// Invoked for each request dropped at shutdown without being processed public OrderRequestProcessingPool(bool concurrencyEnabled, int minimumThreads, int maximumThreads, - Action processRequest, Action onError) + Action processRequest, Action onError, Action onDropped = null) { _synchronous = false; _processRequest = processRequest; _onError = onError; + _onDropped = onDropped; // concurrency grows the pool minimum..maximum on demand, otherwise a single fixed thread is used _maximumThreads = concurrencyEnabled ? Math.Max(1, maximumThreads) : 1; _singleConsumer = _maximumThreads == 1; @@ -115,11 +129,13 @@ public OrderRequestProcessingPool(bool concurrencyEnabled, int minimumThreads, i /// /// Private constructor for the synchronous pool, a single non blocking queue and no worker threads. /// - private OrderRequestProcessingPool(Action processRequest, Action onError) + private OrderRequestProcessingPool(Action processRequest, Action onError, + Action onDropped) { _synchronous = true; _processRequest = processRequest; _onError = onError; + _onDropped = onDropped; _maximumThreads = 1; _singleConsumer = true; @@ -134,9 +150,11 @@ private OrderRequestProcessingPool(Action processRequest, Action /// Handles a single order request /// Invoked when processing fails unexpectedly - public static OrderRequestProcessingPool Synchronous(Action processRequest, Action onError) + /// Invoked for each request dropped at shutdown without being processed + public static OrderRequestProcessingPool Synchronous(Action processRequest, Action onError, + Action onDropped = null) { - return new OrderRequestProcessingPool(processRequest, onError); + return new OrderRequestProcessingPool(processRequest, onError, onDropped); } /// @@ -239,7 +257,8 @@ private bool IsProcessing() } /// - /// Stops every worker thread and waits for them to terminate, then releases the pool resources. + /// Stops the pool: stops every worker thread against a shared deadline, interrupting those that won't finish, + /// then drops the requests that never started processing, reporting each one, and releases the resources. /// public void Dispose() { @@ -254,18 +273,22 @@ public void Dispose() _shuttingDown = true; } - // let the workers drain whatever is queued and parked: once adding is complete their consuming - // enumerables finish naturally when the queue empties, so join before cancelling anything. Only - // escalate to StopSafely, which cancels the shared token and drops pending requests, on timeout + // stop accepting work and cancel right away so the workers exit their consuming enumerable instead + // of draining a backlog that shouldn't be sent to the brokerage while shutting down _readyQueue.CompleteAdding(); + _cancellationTokenSource.Cancel(); + + // join every worker against a single shared deadline instead of a full timeout per thread + var stopwatch = Stopwatch.StartNew(); + List stuckThreads = null; foreach (var thread in _threads) { try { - if (thread != null && !thread.Join(ShutdownTimeout)) + var remaining = ShutdownTimeout - stopwatch.Elapsed; + if (thread != null && !thread.Join(remaining > TimeSpan.Zero ? remaining : TimeSpan.Zero)) { - Log.Error($"OrderRequestProcessingPool.Dispose(): Exceeded timeout: {(int)ShutdownTimeout.TotalSeconds} seconds waiting for '{thread.Name}' to finish processing"); - thread.StopSafely(ShutdownTimeout, _cancellationTokenSource); + (stuckThreads ??= new()).Add(thread); } } catch (ThreadStateException) @@ -274,11 +297,75 @@ public void Dispose() } } + if (stuckThreads != null) + { + // a worker pinned in a blocking call, even a native wait, can only be freed by interrupting it + Log.Error($"OrderRequestProcessingPool.Dispose(): workers did not stop within {(int)ShutdownTimeout.TotalSeconds} seconds, interrupting: {string.Join(", ", stuckThreads.Select(thread => thread.Name))}"); + foreach (var thread in stuckThreads) + { + thread.Interrupt(); + } + foreach (var thread in stuckThreads) + { + if (!thread.Join(InterruptTimeout)) + { + Log.Error($"OrderRequestProcessingPool.Dispose(): '{thread.Name}' is still running after being interrupted"); + } + } + } + + // with the workers stopped, drop everything that never started processing, queued or parked, reporting + // each request so the caller can invalidate its order before the final results go out + DropPendingRequests(); + IsActive = false; _readyQueue.DisposeSafely(); _cancellationTokenSource.DisposeSafely(); } + /// + /// Drops every request that never started processing, from the ready queue and the parked queues, reporting + /// each one through the drop callback. Runs once the workers have stopped: a request a worker grabbed before + /// the cancellation belongs to that worker, whatever is still here was never taken. + /// + private void DropPendingRequests() + { + var dropped = new List(); + while (_readyQueue.TryTake(out var item)) + { + dropped.Add(item.Request); + } + + lock (_lock) + { + // drain the parked queues but keep the entries, so a worker finishing its current request still + // finds its order's key and cleans it up itself + foreach (var parked in _inFlight.Values) + { + while (parked != null && parked.Count > 0) + { + dropped.Add(parked.Dequeue()); + } + } + } + + if (_onDropped == null) + { + return; + } + foreach (var request in dropped) + { + try + { + _onDropped(request); + } + catch (Exception err) + { + Log.Error(err); + } + } + } + /// /// Creates and registers a worker thread without starting it, so callers can start it outside the lock. /// Callers growing the pool on demand must hold . @@ -340,6 +427,15 @@ private void Drain(Action process) process(item); } } + catch (OperationCanceledException) when (_shuttingDown) + { + // the shutdown cancelled the shared token, exit quietly + } + catch (ThreadInterruptedException) when (_shuttingDown) + { + // interrupted by Dispose after the shutdown deadline, exit quietly + Log.Trace($"OrderRequestProcessingPool.Drain(): '{Thread.CurrentThread.Name}' interrupted while shutting down"); + } catch (Exception err) { // unexpected error, we need to close down shop diff --git a/Tests/Engine/BrokerageTransactionHandlerTests/OrderRequestProcessingPoolTests.cs b/Tests/Engine/BrokerageTransactionHandlerTests/OrderRequestProcessingPoolTests.cs new file mode 100644 index 000000000000..5b030c516052 --- /dev/null +++ b/Tests/Engine/BrokerageTransactionHandlerTests/OrderRequestProcessingPoolTests.cs @@ -0,0 +1,200 @@ +/* + * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. + * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. +*/ + +using NUnit.Framework; +using QuantConnect.Lean.Engine.TransactionHandlers; +using QuantConnect.Orders; +using QuantConnect.Util; +using System; +using System.Collections.Concurrent; +using System.Diagnostics; +using System.Linq; +using System.Threading; + +namespace QuantConnect.Tests.Engine.BrokerageTransactionHandlerTests +{ + [TestFixture, Parallelizable(ParallelScope.Fixtures)] + public class OrderRequestProcessingPoolTests + { + [Test] + public void DisposeDropsQueuedAndParkedRequestsBeforeStopping() + { + using var gate = new ManualResetEventSlim(false); + var processed = new ConcurrentQueue(); + var dropped = new ConcurrentQueue(); + Exception processingError = null; + var pool = new OrderRequestProcessingPool(concurrencyEnabled: true, minimumThreads: 1, maximumThreads: 2, + request => + { + processed.Enqueue(request); + gate.Wait(); + }, + exception => processingError = exception, + dropped.Enqueue); + // shrink the shutdown budget so the test doesn't wait out the production timeout on the blocked workers + pool.ShutdownTimeout = TimeSpan.FromMilliseconds(500); + + try + { + var symbol = Symbols.SPY; + var reference = new DateTime(2025, 07, 03, 10, 0, 0); + + // the order we track, its submit claims a worker and blocks on the gate + var submit = new SubmitOrderRequest(OrderType.Market, symbol.SecurityType, symbol, 1, 0, 0, reference, ""); + submit.SetOrderId(1); + var order = Order.CreateOrder(submit); + pool.Dispatch(submit, order); + + // keep feeding unrelated orders until both workers are blocked inside the handler + var fillerId = 1000; + var saturated = SpinWait.SpinUntil(() => + { + var filler = new SubmitOrderRequest(OrderType.Market, symbol.SecurityType, symbol, 1, 0, 0, reference, ""); + filler.SetOrderId(++fillerId); + pool.Dispatch(filler, Order.CreateOrder(filler)); + return processed.Count >= 2; + }, 10000); + Assert.IsTrue(saturated, "the workers never got busy"); + + // these can never be processed: the submit waits in the ready queue, the update and cancel wait parked + var queuedSubmit = new SubmitOrderRequest(OrderType.Market, symbol.SecurityType, symbol, 1, 0, 0, reference, ""); + queuedSubmit.SetOrderId(2); + pool.Dispatch(queuedSubmit, Order.CreateOrder(queuedSubmit)); + var update = new UpdateOrderRequest(reference, order.Id, new UpdateOrderFields()); + var cancel = new CancelOrderRequest(reference, order.Id, ""); + pool.Dispatch(update, order); + pool.Dispatch(cancel, order); + + pool.Dispose(); + + var droppedRequests = dropped.ToList(); + Assert.Contains(queuedSubmit, droppedRequests); + Assert.Contains(update, droppedRequests); + Assert.Contains(cancel, droppedRequests); + // whoever takes a request owns it: nothing is both processed and dropped + CollectionAssert.IsEmpty(droppedRequests.Intersect(processed)); + Assert.IsNull(processingError, $"the pool reported an error: {processingError}"); + } + finally + { + gate.Set(); + pool.DisposeSafely(); + } + } + + [Test] + public void DisposeInterruptsWorkersStuckInTheRequestHandler() + { + using var gate = new ManualResetEventSlim(false); + using var workerBlocked = new ManualResetEventSlim(false); + using var workerInterrupted = new ManualResetEventSlim(false); + var dropped = new ConcurrentQueue(); + Exception processingError = null; + var pool = new OrderRequestProcessingPool(concurrencyEnabled: true, minimumThreads: 1, maximumThreads: 2, + request => + { + workerBlocked.Set(); + try + { + // simulates a brokerage call pinned in a wait that only a thread interrupt can free + gate.Wait(); + } + catch (ThreadInterruptedException) + { + workerInterrupted.Set(); + throw; + } + }, + exception => processingError = exception, + dropped.Enqueue); + pool.ShutdownTimeout = TimeSpan.FromMilliseconds(500); + + try + { + var symbol = Symbols.SPY; + var reference = new DateTime(2025, 07, 03, 10, 0, 0); + var submit = new SubmitOrderRequest(OrderType.Market, symbol.SecurityType, symbol, 1, 0, 0, reference, ""); + submit.SetOrderId(1); + pool.Dispatch(submit, Order.CreateOrder(submit)); + Assert.IsTrue(workerBlocked.Wait(10000), "the worker never picked up the request"); + + var stopwatch = Stopwatch.StartNew(); + pool.Dispose(); + stopwatch.Stop(); + + Assert.IsTrue(workerInterrupted.IsSet, "the stuck worker was not interrupted"); + Assert.Less(stopwatch.Elapsed, TimeSpan.FromSeconds(10), "Dispose did not return within the shutdown budget"); + CollectionAssert.IsEmpty(dropped); + Assert.IsNull(processingError, $"the pool reported an error: {processingError}"); + } + finally + { + gate.Set(); + pool.DisposeSafely(); + } + } + + [Test] + public void DisposeOfAnIdlePoolIsFastAndQuiet() + { + var dropped = new ConcurrentQueue(); + Exception processingError = null; + var pool = new OrderRequestProcessingPool(concurrencyEnabled: true, minimumThreads: 2, maximumThreads: 10, + _ => { }, + exception => processingError = exception, + dropped.Enqueue); + + // keeps the production shutdown timeout: idle workers must exit on cancellation, not wait it out + var stopwatch = Stopwatch.StartNew(); + pool.Dispose(); + stopwatch.Stop(); + + Assert.Less(stopwatch.Elapsed, TimeSpan.FromSeconds(10)); + CollectionAssert.IsEmpty(dropped); + Assert.IsNull(processingError, $"the pool reported an error: {processingError}"); + Assert.IsFalse(pool.IsActive); + } + + [Test] + public void SynchronousPoolDisposeDropsPendingRequests() + { + var dropped = new ConcurrentQueue(); + var processedCount = 0; + Exception processingError = null; + var pool = OrderRequestProcessingPool.Synchronous( + _ => processedCount++, + exception => processingError = exception, + dropped.Enqueue); + + var symbol = Symbols.SPY; + var reference = new DateTime(2025, 07, 03, 10, 0, 0); + var submit = new SubmitOrderRequest(OrderType.Market, symbol.SecurityType, symbol, 1, 0, 0, reference, ""); + submit.SetOrderId(1); + var order = Order.CreateOrder(submit); + pool.Dispatch(submit, order); + pool.ProcessPending(); + Assert.AreEqual(1, processedCount); + + // queued after the last drain, so it is never processed and must be dropped at dispose + var update = new UpdateOrderRequest(reference, order.Id, new UpdateOrderFields()); + pool.Dispatch(update, order); + pool.Dispose(); + + Assert.AreEqual(1, processedCount); + CollectionAssert.AreEquivalent(new OrderRequest[] { update }, dropped); + Assert.IsNull(processingError, $"the pool reported an error: {processingError}"); + } + } +} From c4bf3beebb7cc79a8abd2dbb35405a04916e2455 Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Fri, 28 Aug 2026 13:26:44 -0400 Subject: [PATCH 02/14] Drain dropped requests through the normal processing loop Address review feedback: instead of the pool draining its queues from the outside and reporting each request through a callback, the handler sets a flag at exit and the requests still queued flow through the normal request handling, which invalidates them instead of processing: - Parked per-order requests are re-enqueued onto the shared queue at dispose, and the surviving workers drain the backlog naturally - If every worker was stuck, a background drainer thread chews the backlog with a bounded join, so Dispose cannot block even when invalidating an order itself blocks - Reverts the IBusyCollection.TryTake addition, no longer needed --- Common/Interfaces/IBusyCollection.cs | 7 - Common/Util/BusyBlockingCollection.cs | 10 -- Common/Util/BusyCollection.cs | 10 -- .../BrokerageTransactionHandler.cs | 17 ++- .../OrderRequestProcessingPool.cs | 131 ++++++++---------- .../OrderRequestProcessingPoolTests.cs | 129 ++++++++++------- 6 files changed, 150 insertions(+), 154 deletions(-) diff --git a/Common/Interfaces/IBusyCollection.cs b/Common/Interfaces/IBusyCollection.cs index f62eb3fcaf06..29b5d73d9fec 100644 --- a/Common/Interfaces/IBusyCollection.cs +++ b/Common/Interfaces/IBusyCollection.cs @@ -72,12 +72,5 @@ public interface IBusyCollection : IDisposable /// A cancellation token to observer /// An enumerable that removes and returns items from the collection IEnumerable GetConsumingEnumerable(CancellationToken cancellationToken); - - /// - /// Attempts to take an item from this collection without blocking - /// - /// The item taken, or default when none was available - /// True if an item was taken - bool TryTake(out T item); } } diff --git a/Common/Util/BusyBlockingCollection.cs b/Common/Util/BusyBlockingCollection.cs index 7fb94abd8b5e..70053bb275ac 100644 --- a/Common/Util/BusyBlockingCollection.cs +++ b/Common/Util/BusyBlockingCollection.cs @@ -124,16 +124,6 @@ public void CompleteAdding() _collection.CompleteAdding(); } - /// - /// Attempts to take an item from this collection without blocking - /// - /// The item taken, or default when none was available - /// True if an item was taken - public bool TryTake(out T item) - { - return _collection.TryTake(out item); - } - /// /// Provides a consuming enumerable for items in this collection. /// diff --git a/Common/Util/BusyCollection.cs b/Common/Util/BusyCollection.cs index 0d63c311952f..d321b713cfd8 100644 --- a/Common/Util/BusyCollection.cs +++ b/Common/Util/BusyCollection.cs @@ -128,15 +128,5 @@ public void CompleteAdding() { _completedAdding = true; } - - /// - /// Attempts to take an item from this collection without blocking - /// - /// The item taken, or default when none was available - /// True if an item was taken - public bool TryTake(out T item) - { - return _collection.TryDequeue(out item); - } } } diff --git a/Engine/TransactionHandlers/BrokerageTransactionHandler.cs b/Engine/TransactionHandlers/BrokerageTransactionHandler.cs index b8aeb17f95b1..9893a3344ca9 100644 --- a/Engine/TransactionHandlers/BrokerageTransactionHandler.cs +++ b/Engine/TransactionHandlers/BrokerageTransactionHandler.cs @@ -76,6 +76,8 @@ public class BrokerageTransactionHandler : ITransactionHandler /// requests in order while growing the pool on demand as the threads get saturated. /// private OrderRequestProcessingPool _threadPool; + // set once the handler starts exiting, so requests still queued are invalidated instead of processed + private volatile bool _droppingQueuedRequests; private readonly ConcurrentQueue _orderEvents = new ConcurrentQueue(); @@ -268,6 +270,12 @@ protected virtual void InitializeTransactionThread() { Action processRequest = request => { + // at exit the pool still drains its queue through here, fail the request instead of processing it + if (_droppingQueuedRequests) + { + InvalidateDroppedRequest(request); + return; + } HandleOrderRequest(request); ProcessAsynchronousEvents(); }; @@ -276,9 +284,8 @@ protected virtual void InitializeTransactionThread() // backtesting drains a single queue synchronously on the algorithm thread, live deployments use // background worker threads: a single one, or growing on demand up to the maximum when concurrent. _threadPool = SynchronousProcessing - ? OrderRequestProcessingPool.Synchronous(processRequest, onError, InvalidateDroppedRequest) - : new OrderRequestProcessingPool(ConcurrencyEnabled, MinimumTransactionThreads, MaximumTransactionThreads, - processRequest, onError, InvalidateDroppedRequest); + ? OrderRequestProcessingPool.Synchronous(processRequest, onError) + : new OrderRequestProcessingPool(ConcurrencyEnabled, MinimumTransactionThreads, MaximumTransactionThreads, processRequest, onError); } #region Order Request Processing @@ -803,6 +810,8 @@ public void AddOpenOrder(Order order, IAlgorithm algorithm) /// public void Exit() { + // from here on the requests still queued get invalidated as they drain instead of sent to the brokerage + _droppingQueuedRequests = true; // Dispose drains the queued requests (CompleteAdding) and waits for the threads before stopping _threadPool.DisposeSafely(); } @@ -1932,7 +1941,7 @@ private void InvalidateOrders(List orders, string message) } /// - /// Handles a request the processing pool dropped at shutdown without processing it. Invalidates a dropped + /// Fails a request still queued while the handler exits, instead of processing it. Invalidates a dropped /// submit's order so it doesn't linger as new in the final results, before they are sent. /// private void InvalidateDroppedRequest(OrderRequest request) diff --git a/Engine/TransactionHandlers/OrderRequestProcessingPool.cs b/Engine/TransactionHandlers/OrderRequestProcessingPool.cs index d5c1d4fd66cd..e5d7d72c88de 100644 --- a/Engine/TransactionHandlers/OrderRequestProcessingPool.cs +++ b/Engine/TransactionHandlers/OrderRequestProcessingPool.cs @@ -64,15 +64,13 @@ public class OrderRequestProcessingPool : IDisposable // true when a single consumer drains the queue (synchronous or a single fixed worker), which already // preserves arrival order across all orders so the per-order serialization is skipped entirely private readonly bool _singleConsumer; - // set under the lock when shutting down so the pool stops growing while the pending requests are dropped, - // before the cancellation token is cancelled. volatile, the workers read it lock free while draining + // set under the lock when shutting down so the pool stops growing while the backlog is drained. + // volatile, the workers read it lock free while draining private volatile bool _shuttingDown; // number of workers currently processing a request, used to decide when the pool is saturated private int _busyWorkers; private readonly Action _processRequest; private readonly Action _onError; - // reports each request dropped at shutdown without being processed, may be null - private readonly Action _onDropped; private readonly CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource(); /// @@ -104,14 +102,12 @@ public int ThreadCount /// The maximum number of worker threads the pool can grow to on demand /// Handles a single order request /// Invoked when processing fails unexpectedly - /// Invoked for each request dropped at shutdown without being processed public OrderRequestProcessingPool(bool concurrencyEnabled, int minimumThreads, int maximumThreads, - Action processRequest, Action onError, Action onDropped = null) + Action processRequest, Action onError) { _synchronous = false; _processRequest = processRequest; _onError = onError; - _onDropped = onDropped; // concurrency grows the pool minimum..maximum on demand, otherwise a single fixed thread is used _maximumThreads = concurrencyEnabled ? Math.Max(1, maximumThreads) : 1; _singleConsumer = _maximumThreads == 1; @@ -129,13 +125,11 @@ public OrderRequestProcessingPool(bool concurrencyEnabled, int minimumThreads, i /// /// Private constructor for the synchronous pool, a single non blocking queue and no worker threads. /// - private OrderRequestProcessingPool(Action processRequest, Action onError, - Action onDropped) + private OrderRequestProcessingPool(Action processRequest, Action onError) { _synchronous = true; _processRequest = processRequest; _onError = onError; - _onDropped = onDropped; _maximumThreads = 1; _singleConsumer = true; @@ -150,11 +144,9 @@ private OrderRequestProcessingPool(Action processRequest, Action /// Handles a single order request /// Invoked when processing fails unexpectedly - /// Invoked for each request dropped at shutdown without being processed - public static OrderRequestProcessingPool Synchronous(Action processRequest, Action onError, - Action onDropped = null) + public static OrderRequestProcessingPool Synchronous(Action processRequest, Action onError) { - return new OrderRequestProcessingPool(processRequest, onError, onDropped); + return new OrderRequestProcessingPool(processRequest, onError); } /// @@ -257,8 +249,9 @@ private bool IsProcessing() } /// - /// Stops the pool: stops every worker thread against a shared deadline, interrupting those that won't finish, - /// then drops the requests that never started processing, reporting each one, and releases the resources. + /// Stops the pool. The requests still queued or parked are drained through the normal processing loop, by + /// the surviving workers or by a last resort drainer thread, so the caller's request handler decides their + /// fate. Workers that won't stop within the shared deadline are interrupted. /// public void Dispose() { @@ -271,14 +264,58 @@ public void Dispose() } // stop growing so the threads list is frozen and safe to iterate without taking a snapshot _shuttingDown = true; + + // move the parked requests back onto the shared queue, keeping their key, so whoever drains the + // backlog picks them up. the entries stay so a worker finishing its request still finds its key + foreach (var (key, parked) in _inFlight) + { + while (parked != null && parked.Count > 0) + { + _readyQueue.Add(new WorkItem(parked.Dequeue(), key)); + } + } } - // stop accepting work and cancel right away so the workers exit their consuming enumerable instead - // of draining a backlog that shouldn't be sent to the brokerage while shutting down + // stop accepting work without cancelling the token: the workers still alive keep chewing the backlog + // through the normal loop, and their consuming enumerable ends on its own once the queue empties _readyQueue.CompleteAdding(); + + if (_synchronous) + { + // backtesting only, no workers and no bounded time concern: drain the backlog on the caller thread + ProcessPending(); + } + else + { + JoinWorkers(); + + // if every worker got stuck the backlog was never drained, hand it to a fresh background thread and + // wait a bounded time, so Dispose stays bounded even when processing a request itself blocks + if (_readyQueue.Count > 0) + { + var drainer = new Thread(Run) { IsBackground = true, Name = "Transaction Thread Drainer" }; + drainer.Start(); + if (!drainer.Join(ShutdownTimeout)) + { + Log.Error($"OrderRequestProcessingPool.Dispose(): the drainer thread did not finish within {(int)ShutdownTimeout.TotalSeconds} seconds, abandoning it"); + } + } + } + + // release anything still lingering on the shared token before it is disposed _cancellationTokenSource.Cancel(); - // join every worker against a single shared deadline instead of a full timeout per thread + IsActive = false; + _readyQueue.DisposeSafely(); + _cancellationTokenSource.DisposeSafely(); + } + + /// + /// Joins every worker thread against a single shared deadline instead of a full timeout per thread, + /// interrupting, and briefly re-joining, the ones that won't stop in time. + /// + private void JoinWorkers() + { var stopwatch = Stopwatch.StartNew(); List stuckThreads = null; foreach (var thread in _threads) @@ -313,57 +350,6 @@ public void Dispose() } } } - - // with the workers stopped, drop everything that never started processing, queued or parked, reporting - // each request so the caller can invalidate its order before the final results go out - DropPendingRequests(); - - IsActive = false; - _readyQueue.DisposeSafely(); - _cancellationTokenSource.DisposeSafely(); - } - - /// - /// Drops every request that never started processing, from the ready queue and the parked queues, reporting - /// each one through the drop callback. Runs once the workers have stopped: a request a worker grabbed before - /// the cancellation belongs to that worker, whatever is still here was never taken. - /// - private void DropPendingRequests() - { - var dropped = new List(); - while (_readyQueue.TryTake(out var item)) - { - dropped.Add(item.Request); - } - - lock (_lock) - { - // drain the parked queues but keep the entries, so a worker finishing its current request still - // finds its order's key and cleans it up itself - foreach (var parked in _inFlight.Values) - { - while (parked != null && parked.Count > 0) - { - dropped.Add(parked.Dequeue()); - } - } - } - - if (_onDropped == null) - { - return; - } - foreach (var request in dropped) - { - try - { - _onDropped(request); - } - catch (Exception err) - { - Log.Error(err); - } - } } /// @@ -459,8 +445,9 @@ private void ProcessInOrder(WorkItem item) lock (_lock) { - var parked = _inFlight[item.Key]; - if (parked != null && parked.Count > 0) + // the key may already be gone: at shutdown parked requests go back to the shared queue with + // their key, so another thread can finish the order and clean up first. missing means done + if (_inFlight.TryGetValue(item.Key, out var parked) && parked != null && parked.Count > 0) { request = parked.Dequeue(); } diff --git a/Tests/Engine/BrokerageTransactionHandlerTests/OrderRequestProcessingPoolTests.cs b/Tests/Engine/BrokerageTransactionHandlerTests/OrderRequestProcessingPoolTests.cs index 5b030c516052..b76557ad21ce 100644 --- a/Tests/Engine/BrokerageTransactionHandlerTests/OrderRequestProcessingPoolTests.cs +++ b/Tests/Engine/BrokerageTransactionHandlerTests/OrderRequestProcessingPoolTests.cs @@ -29,11 +29,11 @@ namespace QuantConnect.Tests.Engine.BrokerageTransactionHandlerTests public class OrderRequestProcessingPoolTests { [Test] - public void DisposeDropsQueuedAndParkedRequestsBeforeStopping() + public void DisposeDrainsQueuedAndParkedRequestsThroughTheProcessingLoop() { using var gate = new ManualResetEventSlim(false); + var dispatched = new ConcurrentQueue(); var processed = new ConcurrentQueue(); - var dropped = new ConcurrentQueue(); Exception processingError = null; var pool = new OrderRequestProcessingPool(concurrencyEnabled: true, minimumThreads: 1, maximumThreads: 2, request => @@ -41,10 +41,9 @@ public void DisposeDropsQueuedAndParkedRequestsBeforeStopping() processed.Enqueue(request); gate.Wait(); }, - exception => processingError = exception, - dropped.Enqueue); - // shrink the shutdown budget so the test doesn't wait out the production timeout on the blocked workers - pool.ShutdownTimeout = TimeSpan.FromMilliseconds(500); + exception => processingError = exception); + // shrink the shutdown budget so a regression doesn't wait out the production timeout + pool.ShutdownTimeout = TimeSpan.FromSeconds(5); try { @@ -55,6 +54,7 @@ public void DisposeDropsQueuedAndParkedRequestsBeforeStopping() var submit = new SubmitOrderRequest(OrderType.Market, symbol.SecurityType, symbol, 1, 0, 0, reference, ""); submit.SetOrderId(1); var order = Order.CreateOrder(submit); + dispatched.Enqueue(submit); pool.Dispatch(submit, order); // keep feeding unrelated orders until both workers are blocked inside the handler @@ -63,28 +63,34 @@ public void DisposeDropsQueuedAndParkedRequestsBeforeStopping() { var filler = new SubmitOrderRequest(OrderType.Market, symbol.SecurityType, symbol, 1, 0, 0, reference, ""); filler.SetOrderId(++fillerId); + dispatched.Enqueue(filler); pool.Dispatch(filler, Order.CreateOrder(filler)); return processed.Count >= 2; }, 10000); Assert.IsTrue(saturated, "the workers never got busy"); - // these can never be processed: the submit waits in the ready queue, the update and cancel wait parked + // left behind when Dispose starts: the submit waits in the ready queue, the update and cancel parked var queuedSubmit = new SubmitOrderRequest(OrderType.Market, symbol.SecurityType, symbol, 1, 0, 0, reference, ""); queuedSubmit.SetOrderId(2); + dispatched.Enqueue(queuedSubmit); pool.Dispatch(queuedSubmit, Order.CreateOrder(queuedSubmit)); var update = new UpdateOrderRequest(reference, order.Id, new UpdateOrderFields()); var cancel = new CancelOrderRequest(reference, order.Id, ""); + dispatched.Enqueue(update); + dispatched.Enqueue(cancel); pool.Dispatch(update, order); pool.Dispatch(cancel, order); + // free the workers shortly after Dispose has re-queued the parked backlog, so they chew it all + var release = new Thread(() => { Thread.Sleep(300); gate.Set(); }) { IsBackground = true }; + release.Start(); + var stopwatch = Stopwatch.StartNew(); pool.Dispose(); + stopwatch.Stop(); - var droppedRequests = dropped.ToList(); - Assert.Contains(queuedSubmit, droppedRequests); - Assert.Contains(update, droppedRequests); - Assert.Contains(cancel, droppedRequests); - // whoever takes a request owns it: nothing is both processed and dropped - CollectionAssert.IsEmpty(droppedRequests.Intersect(processed)); + // every dispatched request reached the handler exactly once, none lost and none duplicated + CollectionAssert.AreEquivalent(dispatched, processed); + Assert.Less(stopwatch.Elapsed, TimeSpan.FromSeconds(10), "Dispose did not return within the shutdown budget"); Assert.IsNull(processingError, $"the pool reported an error: {processingError}"); } finally @@ -95,48 +101,72 @@ public void DisposeDropsQueuedAndParkedRequestsBeforeStopping() } [Test] - public void DisposeInterruptsWorkersStuckInTheRequestHandler() + public void DisposeInterruptsStuckWorkersAndDrainsTheBacklogOnANewThread() { using var gate = new ManualResetEventSlim(false); - using var workerBlocked = new ManualResetEventSlim(false); - using var workerInterrupted = new ManualResetEventSlim(false); - var dropped = new ConcurrentQueue(); + var processed = new ConcurrentQueue<(OrderRequest Request, string Thread)>(); + var interruptedWorkers = 0; Exception processingError = null; - var pool = new OrderRequestProcessingPool(concurrencyEnabled: true, minimumThreads: 1, maximumThreads: 2, + var pool = new OrderRequestProcessingPool(concurrencyEnabled: true, minimumThreads: 2, maximumThreads: 2, request => { - workerBlocked.Set(); - try - { - // simulates a brokerage call pinned in a wait that only a thread interrupt can free - gate.Wait(); - } - catch (ThreadInterruptedException) + processed.Enqueue((request, Thread.CurrentThread.Name)); + if (request.Tag == "poison") { - workerInterrupted.Set(); - throw; + try + { + // simulates a brokerage call pinned in a wait that only a thread interrupt can free + gate.Wait(); + } + catch (ThreadInterruptedException) + { + Interlocked.Increment(ref interruptedWorkers); + throw; + } } }, - exception => processingError = exception, - dropped.Enqueue); + exception => processingError = exception); pool.ShutdownTimeout = TimeSpan.FromMilliseconds(500); try { var symbol = Symbols.SPY; var reference = new DateTime(2025, 07, 03, 10, 0, 0); - var submit = new SubmitOrderRequest(OrderType.Market, symbol.SecurityType, symbol, 1, 0, 0, reference, ""); - submit.SetOrderId(1); - pool.Dispatch(submit, Order.CreateOrder(submit)); - Assert.IsTrue(workerBlocked.Wait(10000), "the worker never picked up the request"); + SubmitOrderRequest CreateSubmit(int orderId, string tag) + { + var request = new SubmitOrderRequest(OrderType.Market, symbol.SecurityType, symbol, 1, 0, 0, reference, tag); + request.SetOrderId(orderId); + return request; + } + + // both workers swallow a poison pill and get stuck inside the handler + var poison = CreateSubmit(1, "poison"); + var poisonOrder = Order.CreateOrder(poison); + pool.Dispatch(poison, poisonOrder); + var otherPoison = CreateSubmit(2, "poison"); + pool.Dispatch(otherPoison, Order.CreateOrder(otherPoison)); + Assert.IsTrue(SpinWait.SpinUntil(() => processed.Count >= 2, 10000), "the workers never got stuck"); + + // stranded until the drainer takes over: two queued submits and an update parked behind a poison + var queued1 = CreateSubmit(3, ""); + pool.Dispatch(queued1, Order.CreateOrder(queued1)); + var queued2 = CreateSubmit(4, ""); + pool.Dispatch(queued2, Order.CreateOrder(queued2)); + var update = new UpdateOrderRequest(reference, poisonOrder.Id, new UpdateOrderFields()); + pool.Dispatch(update, poisonOrder); var stopwatch = Stopwatch.StartNew(); pool.Dispose(); stopwatch.Stop(); - Assert.IsTrue(workerInterrupted.IsSet, "the stuck worker was not interrupted"); + Assert.AreEqual(2, interruptedWorkers, "the stuck workers were not interrupted"); + foreach (var request in new OrderRequest[] { queued1, queued2, update }) + { + var passes = processed.Where(pair => pair.Request == request).ToList(); + Assert.AreEqual(1, passes.Count, $"the stranded request was processed {passes.Count} times"); + Assert.AreEqual("Transaction Thread Drainer", passes[0].Thread, "the backlog was not drained on a new thread"); + } Assert.Less(stopwatch.Elapsed, TimeSpan.FromSeconds(10), "Dispose did not return within the shutdown budget"); - CollectionAssert.IsEmpty(dropped); Assert.IsNull(processingError, $"the pool reported an error: {processingError}"); } finally @@ -149,34 +179,31 @@ public void DisposeInterruptsWorkersStuckInTheRequestHandler() [Test] public void DisposeOfAnIdlePoolIsFastAndQuiet() { - var dropped = new ConcurrentQueue(); + var processedCount = 0; Exception processingError = null; var pool = new OrderRequestProcessingPool(concurrencyEnabled: true, minimumThreads: 2, maximumThreads: 10, - _ => { }, - exception => processingError = exception, - dropped.Enqueue); + _ => Interlocked.Increment(ref processedCount), + exception => processingError = exception); - // keeps the production shutdown timeout: idle workers must exit on cancellation, not wait it out + // keeps the production shutdown timeout: idle workers must exit once adding completes, not wait it out var stopwatch = Stopwatch.StartNew(); pool.Dispose(); stopwatch.Stop(); Assert.Less(stopwatch.Elapsed, TimeSpan.FromSeconds(10)); - CollectionAssert.IsEmpty(dropped); + Assert.AreEqual(0, processedCount); Assert.IsNull(processingError, $"the pool reported an error: {processingError}"); Assert.IsFalse(pool.IsActive); } [Test] - public void SynchronousPoolDisposeDropsPendingRequests() + public void SynchronousPoolDisposeProcessesPendingRequests() { - var dropped = new ConcurrentQueue(); - var processedCount = 0; + var processed = new ConcurrentQueue(); Exception processingError = null; var pool = OrderRequestProcessingPool.Synchronous( - _ => processedCount++, - exception => processingError = exception, - dropped.Enqueue); + processed.Enqueue, + exception => processingError = exception); var symbol = Symbols.SPY; var reference = new DateTime(2025, 07, 03, 10, 0, 0); @@ -185,16 +212,16 @@ public void SynchronousPoolDisposeDropsPendingRequests() var order = Order.CreateOrder(submit); pool.Dispatch(submit, order); pool.ProcessPending(); - Assert.AreEqual(1, processedCount); + CollectionAssert.AreEqual(new OrderRequest[] { submit }, processed); - // queued after the last drain, so it is never processed and must be dropped at dispose + // queued after the last drain: Dispose pumps it through the processing loop on the caller thread var update = new UpdateOrderRequest(reference, order.Id, new UpdateOrderFields()); pool.Dispatch(update, order); pool.Dispose(); - Assert.AreEqual(1, processedCount); - CollectionAssert.AreEquivalent(new OrderRequest[] { update }, dropped); + CollectionAssert.AreEqual(new OrderRequest[] { submit, update }, processed); Assert.IsNull(processingError, $"the pool reported an error: {processingError}"); + Assert.IsFalse(pool.IsActive); } } } From a553cf114f6c3b3983908f3154b9b55cd3944f48 Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Fri, 28 Aug 2026 13:45:05 -0400 Subject: [PATCH 03/14] Add handler level test for requests still queued at exit A cancel issued after the last pump, e.g. from OnEndOfAlgorithm, is still queued when the handler exits: assert it gets the "never sent to the brokerage" error response instead of being silently discarded, the order is not canceled and no extra order events are emitted. --- .../BrokerageTransactionHandlerTests.cs | 48 ++++++++++++++++++- 1 file changed, 47 insertions(+), 1 deletion(-) diff --git a/Tests/Engine/BrokerageTransactionHandlerTests/BrokerageTransactionHandlerTests.cs b/Tests/Engine/BrokerageTransactionHandlerTests/BrokerageTransactionHandlerTests.cs index a7c22d903ec1..09eeac7ee3e0 100644 --- a/Tests/Engine/BrokerageTransactionHandlerTests/BrokerageTransactionHandlerTests.cs +++ b/Tests/Engine/BrokerageTransactionHandlerTests/BrokerageTransactionHandlerTests.cs @@ -1,4 +1,4 @@ -/* +/* * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation. * @@ -1399,6 +1399,46 @@ public void CancelOrderTicket() Assert.AreEqual(orderTicket.Status, OrderStatus.CancelPending); } + [Test] + public void ExitInvalidatesRequestsStillQueued() + { + _transactionHandler = new TestBrokerageTransactionHandler(); + using var brokerage = new NoSubmitTestBrokerage(_algorithm); + _transactionHandler.Initialize(_algorithm, brokerage, new BacktestingResultHandler()); + + var security = _algorithm.Securities[_symbol]; + var price = 1.12m; + security.SetMarketPrice(new Tick(DateTime.Now, security.Symbol, price, price, price)); + var orderRequest = new SubmitOrderRequest(OrderType.Limit, security.Type, security.Symbol, 1000, 0, 1.11m, DateTime.Now, ""); + + _algorithm.Transactions.SetOrderProcessor(_transactionHandler); + + // submit a limit order and drain the queue like the engine would, then move it to submitted + var orderTicket = _transactionHandler.Process(orderRequest); + _transactionHandler.PumpPendingRequests(); + Assert.IsTrue(orderRequest.Response.IsSuccess); + var submitted = new OrderEvent(_algorithm.Transactions.GetOpenOrders().Single(), _algorithm.UtcTime, OrderFee.Zero) + { Status = OrderStatus.Submitted }; + brokerage.PublishOrderEvent(submitted); + Assert.AreEqual(OrderStatus.Submitted, orderTicket.Status); + + // queue a cancel without draining, like one issued in OnEndOfAlgorithm after the last pump + var cancelResponse = orderTicket.Cancel(); + Assert.IsFalse(cancelResponse.IsError); + var cancelRequest = orderTicket.CancelRequest; + Assert.AreEqual(OrderRequestStatus.Processing, cancelRequest.Status); + var orderEventsCount = _algorithm.OrderEvents.Count; + + // at exit the queued cancel is invalidated instead of silently discarded or sent to the brokerage + _transactionHandler.Exit(); + Assert.IsTrue(cancelRequest.Response.IsError); + Assert.AreEqual(OrderResponseErrorCode.ProcessingError, cancelRequest.Response.ErrorCode); + StringAssert.Contains("never sent to the brokerage", cancelRequest.Response.ErrorMessage); + // the cancel never reached the brokerage and only a submit drop emits an order event + Assert.AreNotEqual(OrderStatus.Canceled, orderTicket.Status); + Assert.AreEqual(orderEventsCount, _algorithm.OrderEvents.Count); + } + [Test] public void CancelOrderRequestShouldFailForFilledOrder() { @@ -3123,6 +3163,12 @@ public class TestBrokerageTransactionHandler : BrokerageTransactionHandler // no worker thread: these tests drive HandleOrderRequest manually protected override bool SynchronousProcessing => true; + // drains the queued requests like the engine would, so tests can leave the queue truly empty or not + public void PumpPendingRequests() + { + ProcessPendingRequests(); + } + public override void Initialize(IAlgorithm algorithm, IBrokerage brokerage, IResultHandler resultHandler) { _brokerage = brokerage; From 32100d2bebecda00042911d5f3fad994b5ce6d8b Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Fri, 28 Aug 2026 16:55:44 -0400 Subject: [PATCH 04/14] Never throw while joining the workers at dispose Both join loops now handle the disposing thread being interrupted while waiting, so an exception cannot abort the engine shutdown mid-way, and a trace at the end of Dispose records that it completed. --- .../OrderRequestProcessingPool.cs | 31 ++++++++++++++----- 1 file changed, 23 insertions(+), 8 deletions(-) diff --git a/Engine/TransactionHandlers/OrderRequestProcessingPool.cs b/Engine/TransactionHandlers/OrderRequestProcessingPool.cs index e5d7d72c88de..ad9b6fa3a89b 100644 --- a/Engine/TransactionHandlers/OrderRequestProcessingPool.cs +++ b/Engine/TransactionHandlers/OrderRequestProcessingPool.cs @@ -1,4 +1,4 @@ -/* +/* * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation. * @@ -308,6 +308,7 @@ public void Dispose() IsActive = false; _readyQueue.DisposeSafely(); _cancellationTokenSource.DisposeSafely(); + Log.Trace("OrderRequestProcessingPool.Dispose(): completed"); } /// @@ -332,22 +333,36 @@ private void JoinWorkers() { // registered by a concurrent Dispatch but not started yet, nothing to drain on it } + catch (ThreadInterruptedException) + { + // the disposing thread itself was interrupted while joining, keep shutting down + Log.Error($"OrderRequestProcessingPool.Dispose(): interrupted while joining '{thread.Name}'"); + } } if (stuckThreads != null) { // a worker pinned in a blocking call, even a native wait, can only be freed by interrupting it Log.Error($"OrderRequestProcessingPool.Dispose(): workers did not stop within {(int)ShutdownTimeout.TotalSeconds} seconds, interrupting: {string.Join(", ", stuckThreads.Select(thread => thread.Name))}"); - foreach (var thread in stuckThreads) - { - thread.Interrupt(); - } - foreach (var thread in stuckThreads) + try { - if (!thread.Join(InterruptTimeout)) + foreach (var thread in stuckThreads) { - Log.Error($"OrderRequestProcessingPool.Dispose(): '{thread.Name}' is still running after being interrupted"); + thread.Interrupt(); } + foreach (var thread in stuckThreads) + { + if (!thread.Join(InterruptTimeout)) + { + Log.Error($"OrderRequestProcessingPool.Dispose(): '{thread.Name}' is still running after being interrupted"); + } + } + } + catch (ThreadInterruptedException) + { + // the disposing thread itself was interrupted while joining, keep shutting down. + // the stuck list only holds started threads, so no other exception is expected here + Log.Error("OrderRequestProcessingPool.Dispose(): interrupted while joining the stuck workers"); } } } From 1067a43ab179decc46a504731741137ad8dcdfdd Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Fri, 28 Aug 2026 17:00:48 -0400 Subject: [PATCH 05/14] Handle the join interrupts per thread One interrupted join no longer skips joining the remaining stuck workers. Interrupting a started thread cannot throw, so the interrupt loop needs no guard. --- .../OrderRequestProcessingPool.cs | 24 +++++++++---------- 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/Engine/TransactionHandlers/OrderRequestProcessingPool.cs b/Engine/TransactionHandlers/OrderRequestProcessingPool.cs index ad9b6fa3a89b..83455694a528 100644 --- a/Engine/TransactionHandlers/OrderRequestProcessingPool.cs +++ b/Engine/TransactionHandlers/OrderRequestProcessingPool.cs @@ -344,25 +344,25 @@ private void JoinWorkers() { // a worker pinned in a blocking call, even a native wait, can only be freed by interrupting it Log.Error($"OrderRequestProcessingPool.Dispose(): workers did not stop within {(int)ShutdownTimeout.TotalSeconds} seconds, interrupting: {string.Join(", ", stuckThreads.Select(thread => thread.Name))}"); - try + // interrupting a started thread cannot throw, only the joins below can + foreach (var thread in stuckThreads) { - foreach (var thread in stuckThreads) - { - thread.Interrupt(); - } - foreach (var thread in stuckThreads) + thread.Interrupt(); + } + foreach (var thread in stuckThreads) + { + try { if (!thread.Join(InterruptTimeout)) { Log.Error($"OrderRequestProcessingPool.Dispose(): '{thread.Name}' is still running after being interrupted"); } } - } - catch (ThreadInterruptedException) - { - // the disposing thread itself was interrupted while joining, keep shutting down. - // the stuck list only holds started threads, so no other exception is expected here - Log.Error("OrderRequestProcessingPool.Dispose(): interrupted while joining the stuck workers"); + catch (ThreadInterruptedException) + { + // the disposing thread itself was interrupted while joining, keep going with the rest + Log.Error($"OrderRequestProcessingPool.Dispose(): interrupted while joining '{thread.Name}'"); + } } } } From af0e38a326baf065054c7bde20364f47840dc149 Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Fri, 28 Aug 2026 17:33:02 -0400 Subject: [PATCH 06/14] Warn once when the OnOrderEvent handler is slow A slow handler holds the order event lock, and for Python the GIL, stalling the other transaction threads and the order status processing. Handlers taking over 5 seconds get the user warned, once; after that the handler is no longer measured. --- .../BrokerageTransactionHandler.cs | 28 ++++++++++++++++++- .../BrokerageTransactionHandlerTests.cs | 28 +++++++++++++++++++ 2 files changed, 55 insertions(+), 1 deletion(-) diff --git a/Engine/TransactionHandlers/BrokerageTransactionHandler.cs b/Engine/TransactionHandlers/BrokerageTransactionHandler.cs index 9893a3344ca9..ead0e4fe3db0 100644 --- a/Engine/TransactionHandlers/BrokerageTransactionHandler.cs +++ b/Engine/TransactionHandlers/BrokerageTransactionHandler.cs @@ -16,6 +16,7 @@ using System; using System.Collections.Concurrent; using System.Collections.Generic; +using System.Diagnostics; using System.Linq; using System.Runtime.CompilerServices; using System.Threading; @@ -79,6 +80,13 @@ public class BrokerageTransactionHandler : ITransactionHandler // set once the handler starts exiting, so requests still queued are invalidated instead of processed private volatile bool _droppingQueuedRequests; + /// + /// An OnOrderEvent handler slower than this gets the user warned, once. Settable for tests. + /// + internal TimeSpan SlowOnOrderEventThreshold { get; set; } = TimeSpan.FromSeconds(5); + // once the warning is sent the handler is no longer measured. Only written under the order event lock + private bool _slowOnOrderEventWarningSent; + private readonly ConcurrentQueue _orderEvents = new ConcurrentQueue(); /// @@ -1376,7 +1384,25 @@ private void HandleOrderEvents(List orderEvents) try { //Trigger our order event handler - _algorithm.OnOrderEvent(orderEvent); + if (_slowOnOrderEventWarningSent) + { + _algorithm.OnOrderEvent(orderEvent); + } + else + { + // a slow handler holds the order event lock, and for Python the GIL, stalling the + // other transaction threads and the order status processing. + var stopwatch = Stopwatch.StartNew(); + _algorithm.OnOrderEvent(orderEvent); + stopwatch.Stop(); + if (stopwatch.Elapsed > SlowOnOrderEventThreshold) + { + _slowOnOrderEventWarningSent = true; + Log.Trace($"BrokerageTransactionHandler.HandleOrderEvents(): the OnOrderEvent handler took {stopwatch.Elapsed.TotalSeconds:0.##}s: " + + "while it runs, order status updates and the other transaction threads are blocked. Warning the user once"); + _algorithm.Debug($"Warning: The OnOrderEvent handler took {stopwatch.Elapsed.TotalSeconds:0.##} seconds to run, which can delay order processing. Keep the handler fast."); + } + } } catch (Exception err) { diff --git a/Tests/Engine/BrokerageTransactionHandlerTests/BrokerageTransactionHandlerTests.cs b/Tests/Engine/BrokerageTransactionHandlerTests/BrokerageTransactionHandlerTests.cs index 09eeac7ee3e0..b48851c810ed 100644 --- a/Tests/Engine/BrokerageTransactionHandlerTests/BrokerageTransactionHandlerTests.cs +++ b/Tests/Engine/BrokerageTransactionHandlerTests/BrokerageTransactionHandlerTests.cs @@ -1439,6 +1439,34 @@ public void ExitInvalidatesRequestsStillQueued() Assert.AreEqual(orderEventsCount, _algorithm.OrderEvents.Count); } + // A slow OnOrderEvent handler blocks the order event lock, and for Python the GIL, so the user is + // warned once and the measurement stops afterwards. + [Test] + public void SlowOnOrderEventHandlerWarnsOnlyOnce() + { + _transactionHandler = new TestBrokerageTransactionHandler(); + using var brokerage = new NoSubmitTestBrokerage(_algorithm); + _transactionHandler.Initialize(_algorithm, brokerage, new BacktestingResultHandler()); + // any handler run exceeds a zero threshold + _transactionHandler.SlowOnOrderEventThreshold = TimeSpan.Zero; + + var security = _algorithm.Securities[_symbol]; + var price = 1.12m; + security.SetMarketPrice(new Tick(DateTime.Now, security.Symbol, price, price, price)); + var orderRequest = new SubmitOrderRequest(OrderType.Limit, security.Type, security.Symbol, 1000, 0, 1.11m, DateTime.Now, ""); + _algorithm.Transactions.SetOrderProcessor(_transactionHandler); + _transactionHandler.Process(orderRequest); + _transactionHandler.PumpPendingRequests(); + + var order = _algorithm.Transactions.GetOpenOrders().Single(); + brokerage.PublishOrderEvent(new OrderEvent(order, _algorithm.UtcTime, OrderFee.Zero) { Status = OrderStatus.Submitted }); + Assert.AreEqual(1, _algorithm.DebugMessages.Count(message => message.Contains("OnOrderEvent"))); + + // the warning was sent, later slow handler runs are neither measured nor reported again + brokerage.PublishOrderEvent(new OrderEvent(order, _algorithm.UtcTime, OrderFee.Zero) { Status = OrderStatus.PartiallyFilled, FillQuantity = 1, FillPrice = price }); + Assert.AreEqual(1, _algorithm.DebugMessages.Count(message => message.Contains("OnOrderEvent"))); + } + [Test] public void CancelOrderRequestShouldFailForFilledOrder() { From 0aaee76f0a591a941e7685ffcc239993f873acd4 Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Fri, 28 Aug 2026 17:41:17 -0400 Subject: [PATCH 07/14] Raise the slow OnOrderEvent handler threshold to ten seconds --- Engine/TransactionHandlers/BrokerageTransactionHandler.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Engine/TransactionHandlers/BrokerageTransactionHandler.cs b/Engine/TransactionHandlers/BrokerageTransactionHandler.cs index ead0e4fe3db0..f46fb6cadc29 100644 --- a/Engine/TransactionHandlers/BrokerageTransactionHandler.cs +++ b/Engine/TransactionHandlers/BrokerageTransactionHandler.cs @@ -83,7 +83,7 @@ public class BrokerageTransactionHandler : ITransactionHandler /// /// An OnOrderEvent handler slower than this gets the user warned, once. Settable for tests. /// - internal TimeSpan SlowOnOrderEventThreshold { get; set; } = TimeSpan.FromSeconds(5); + internal TimeSpan SlowOnOrderEventThreshold { get; set; } = TimeSpan.FromSeconds(10); // once the warning is sent the handler is no longer measured. Only written under the order event lock private bool _slowOnOrderEventWarningSent; From 036f59d5abec9c9c15626f0687d8c8371e3cdde2 Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Fri, 28 Aug 2026 17:51:29 -0400 Subject: [PATCH 08/14] Measure the OnOrderEvent handler without allocating --- .../BrokerageTransactionHandler.cs | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/Engine/TransactionHandlers/BrokerageTransactionHandler.cs b/Engine/TransactionHandlers/BrokerageTransactionHandler.cs index f46fb6cadc29..af5692f6d1e4 100644 --- a/Engine/TransactionHandlers/BrokerageTransactionHandler.cs +++ b/Engine/TransactionHandlers/BrokerageTransactionHandler.cs @@ -1,4 +1,4 @@ -/* +/* * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation. * @@ -1392,15 +1392,15 @@ private void HandleOrderEvents(List orderEvents) { // a slow handler holds the order event lock, and for Python the GIL, stalling the // other transaction threads and the order status processing. - var stopwatch = Stopwatch.StartNew(); + var start = Stopwatch.GetTimestamp(); _algorithm.OnOrderEvent(orderEvent); - stopwatch.Stop(); - if (stopwatch.Elapsed > SlowOnOrderEventThreshold) + var elapsed = Stopwatch.GetElapsedTime(start); + if (elapsed > SlowOnOrderEventThreshold) { _slowOnOrderEventWarningSent = true; - Log.Trace($"BrokerageTransactionHandler.HandleOrderEvents(): the OnOrderEvent handler took {stopwatch.Elapsed.TotalSeconds:0.##}s: " + + Log.Trace($"BrokerageTransactionHandler.HandleOrderEvents(): the OnOrderEvent handler took {elapsed.TotalSeconds:0.##}s: " + "while it runs, order status updates and the other transaction threads are blocked. Warning the user once"); - _algorithm.Debug($"Warning: The OnOrderEvent handler took {stopwatch.Elapsed.TotalSeconds:0.##} seconds to run, which can delay order processing. Keep the handler fast."); + _algorithm.Debug($"Warning: The OnOrderEvent handler took {elapsed.TotalSeconds:0.##} seconds to run, which can delay order processing. Keep the handler fast."); } } } From 1a2e531d66c11312358b4dcd4526b329ff5be251 Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Mon, 31 Aug 2026 11:03:04 -0400 Subject: [PATCH 09/14] Drop parked requests with their owning worker at dispose Only the requests still in the ready queue are drained, and invalidated, when the pool is disposed; parked follow up requests are left with their owning worker and are dropped with it. --- .../OrderRequestProcessingPool.cs | 19 ++--- .../OrderRequestProcessingPoolTests.cs | 84 ++++++++++--------- 2 files changed, 48 insertions(+), 55 deletions(-) diff --git a/Engine/TransactionHandlers/OrderRequestProcessingPool.cs b/Engine/TransactionHandlers/OrderRequestProcessingPool.cs index 83455694a528..47d99a95417b 100644 --- a/Engine/TransactionHandlers/OrderRequestProcessingPool.cs +++ b/Engine/TransactionHandlers/OrderRequestProcessingPool.cs @@ -249,9 +249,10 @@ private bool IsProcessing() } /// - /// Stops the pool. The requests still queued or parked are drained through the normal processing loop, by + /// Stops the pool. The requests still in the ready queue are drained through the normal processing loop, by /// the surviving workers or by a last resort drainer thread, so the caller's request handler decides their - /// fate. Workers that won't stop within the shared deadline are interrupted. + /// fate. Parked follow up requests are left with their owning worker and are dropped with it. Workers that + /// won't stop within the shared deadline are interrupted. /// public void Dispose() { @@ -264,16 +265,6 @@ public void Dispose() } // stop growing so the threads list is frozen and safe to iterate without taking a snapshot _shuttingDown = true; - - // move the parked requests back onto the shared queue, keeping their key, so whoever drains the - // backlog picks them up. the entries stay so a worker finishing its request still finds its key - foreach (var (key, parked) in _inFlight) - { - while (parked != null && parked.Count > 0) - { - _readyQueue.Add(new WorkItem(parked.Dequeue(), key)); - } - } } // stop accepting work without cancelling the token: the workers still alive keep chewing the backlog @@ -460,8 +451,8 @@ private void ProcessInOrder(WorkItem item) lock (_lock) { - // the key may already be gone: at shutdown parked requests go back to the shared queue with - // their key, so another thread can finish the order and clean up first. missing means done + // the key may already be gone: a worker interrupted at shutdown abandons its keys, so + // treat a missing key as the order being done if (_inFlight.TryGetValue(item.Key, out var parked) && parked != null && parked.Count > 0) { request = parked.Dequeue(); diff --git a/Tests/Engine/BrokerageTransactionHandlerTests/OrderRequestProcessingPoolTests.cs b/Tests/Engine/BrokerageTransactionHandlerTests/OrderRequestProcessingPoolTests.cs index b76557ad21ce..48b28b95cca3 100644 --- a/Tests/Engine/BrokerageTransactionHandlerTests/OrderRequestProcessingPoolTests.cs +++ b/Tests/Engine/BrokerageTransactionHandlerTests/OrderRequestProcessingPoolTests.cs @@ -29,73 +29,73 @@ namespace QuantConnect.Tests.Engine.BrokerageTransactionHandlerTests public class OrderRequestProcessingPoolTests { [Test] - public void DisposeDrainsQueuedAndParkedRequestsThroughTheProcessingLoop() + public void DisposeDrainsQueuedRequestsOnceAndDropsParkedRequests() { - using var gate = new ManualResetEventSlim(false); - var dispatched = new ConcurrentQueue(); + using var stuckGate = new ManualResetEventSlim(false); + using var slowGate = new ManualResetEventSlim(false); var processed = new ConcurrentQueue(); Exception processingError = null; - var pool = new OrderRequestProcessingPool(concurrencyEnabled: true, minimumThreads: 1, maximumThreads: 2, + var pool = new OrderRequestProcessingPool(concurrencyEnabled: true, minimumThreads: 2, maximumThreads: 2, request => { processed.Enqueue(request); - gate.Wait(); + if (request.Tag == "stuck") + { + // pinned until interrupted, so its parked follow ups never get their turn + stuckGate.Wait(); + } + else if (request.Tag == "slow") + { + slowGate.Wait(); + } }, exception => processingError = exception); // shrink the shutdown budget so a regression doesn't wait out the production timeout - pool.ShutdownTimeout = TimeSpan.FromSeconds(5); + pool.ShutdownTimeout = TimeSpan.FromSeconds(2); try { var symbol = Symbols.SPY; var reference = new DateTime(2025, 07, 03, 10, 0, 0); - - // the order we track, its submit claims a worker and blocks on the gate - var submit = new SubmitOrderRequest(OrderType.Market, symbol.SecurityType, symbol, 1, 0, 0, reference, ""); - submit.SetOrderId(1); - var order = Order.CreateOrder(submit); - dispatched.Enqueue(submit); - pool.Dispatch(submit, order); - - // keep feeding unrelated orders until both workers are blocked inside the handler - var fillerId = 1000; - var saturated = SpinWait.SpinUntil(() => + SubmitOrderRequest CreateSubmit(int orderId, string tag) { - var filler = new SubmitOrderRequest(OrderType.Market, symbol.SecurityType, symbol, 1, 0, 0, reference, ""); - filler.SetOrderId(++fillerId); - dispatched.Enqueue(filler); - pool.Dispatch(filler, Order.CreateOrder(filler)); - return processed.Count >= 2; - }, 10000); - Assert.IsTrue(saturated, "the workers never got busy"); + var request = new SubmitOrderRequest(OrderType.Market, symbol.SecurityType, symbol, 1, 0, 0, reference, tag); + request.SetOrderId(orderId); + return request; + } + + // one worker gets pinned on the tracked order, the other blocks until released mid-Dispose + var stuckSubmit = CreateSubmit(1, "stuck"); + var stuckOrder = Order.CreateOrder(stuckSubmit); + pool.Dispatch(stuckSubmit, stuckOrder); + var slowSubmit = CreateSubmit(2, "slow"); + pool.Dispatch(slowSubmit, Order.CreateOrder(slowSubmit)); + Assert.IsTrue(SpinWait.SpinUntil(() => processed.Count >= 2, 10000), "the workers never got busy"); // left behind when Dispose starts: the submit waits in the ready queue, the update and cancel parked - var queuedSubmit = new SubmitOrderRequest(OrderType.Market, symbol.SecurityType, symbol, 1, 0, 0, reference, ""); - queuedSubmit.SetOrderId(2); - dispatched.Enqueue(queuedSubmit); + var queuedSubmit = CreateSubmit(3, ""); pool.Dispatch(queuedSubmit, Order.CreateOrder(queuedSubmit)); - var update = new UpdateOrderRequest(reference, order.Id, new UpdateOrderFields()); - var cancel = new CancelOrderRequest(reference, order.Id, ""); - dispatched.Enqueue(update); - dispatched.Enqueue(cancel); - pool.Dispatch(update, order); - pool.Dispatch(cancel, order); - - // free the workers shortly after Dispose has re-queued the parked backlog, so they chew it all - var release = new Thread(() => { Thread.Sleep(300); gate.Set(); }) { IsBackground = true }; + var update = new UpdateOrderRequest(reference, stuckOrder.Id, new UpdateOrderFields()); + var cancel = new CancelOrderRequest(reference, stuckOrder.Id, ""); + pool.Dispatch(update, stuckOrder); + pool.Dispatch(cancel, stuckOrder); + + // free the slow worker shortly into Dispose so it drains the ready queue; the stuck one stays pinned + var release = new Thread(() => { Thread.Sleep(300); slowGate.Set(); }) { IsBackground = true }; release.Start(); var stopwatch = Stopwatch.StartNew(); pool.Dispose(); stopwatch.Stop(); - // every dispatched request reached the handler exactly once, none lost and none duplicated - CollectionAssert.AreEquivalent(dispatched, processed); + // the ready queue requests reached the handler exactly once, the parked ones were dropped untouched + CollectionAssert.AreEquivalent(new OrderRequest[] { stuckSubmit, slowSubmit, queuedSubmit }, processed); Assert.Less(stopwatch.Elapsed, TimeSpan.FromSeconds(10), "Dispose did not return within the shutdown budget"); Assert.IsNull(processingError, $"the pool reported an error: {processingError}"); } finally { - gate.Set(); + stuckGate.Set(); + slowGate.Set(); pool.DisposeSafely(); } } @@ -147,7 +147,8 @@ SubmitOrderRequest CreateSubmit(int orderId, string tag) pool.Dispatch(otherPoison, Order.CreateOrder(otherPoison)); Assert.IsTrue(SpinWait.SpinUntil(() => processed.Count >= 2, 10000), "the workers never got stuck"); - // stranded until the drainer takes over: two queued submits and an update parked behind a poison + // stranded until the drainer takes over: two queued submits. the update parked behind a poison + // stays with its stuck worker and is dropped var queued1 = CreateSubmit(3, ""); pool.Dispatch(queued1, Order.CreateOrder(queued1)); var queued2 = CreateSubmit(4, ""); @@ -160,12 +161,13 @@ SubmitOrderRequest CreateSubmit(int orderId, string tag) stopwatch.Stop(); Assert.AreEqual(2, interruptedWorkers, "the stuck workers were not interrupted"); - foreach (var request in new OrderRequest[] { queued1, queued2, update }) + foreach (var request in new OrderRequest[] { queued1, queued2 }) { var passes = processed.Where(pair => pair.Request == request).ToList(); Assert.AreEqual(1, passes.Count, $"the stranded request was processed {passes.Count} times"); Assert.AreEqual("Transaction Thread Drainer", passes[0].Thread, "the backlog was not drained on a new thread"); } + Assert.IsFalse(processed.Any(pair => pair.Request == update), "the parked update should have been dropped"); Assert.Less(stopwatch.Elapsed, TimeSpan.FromSeconds(10), "Dispose did not return within the shutdown budget"); Assert.IsNull(processingError, $"the pool reported an error: {processingError}"); } From b7424970d65f925183f411b0aa683e0e4b70baf7 Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Mon, 31 Aug 2026 11:03:04 -0400 Subject: [PATCH 10/14] Guard interrupting the stuck workers Interrupt is not documented to throw on modern .NET, but one worker must never abort the shutdown or skip interrupting the rest. --- .../OrderRequestProcessingPool.cs | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/Engine/TransactionHandlers/OrderRequestProcessingPool.cs b/Engine/TransactionHandlers/OrderRequestProcessingPool.cs index 47d99a95417b..ee219346ceef 100644 --- a/Engine/TransactionHandlers/OrderRequestProcessingPool.cs +++ b/Engine/TransactionHandlers/OrderRequestProcessingPool.cs @@ -335,10 +335,18 @@ private void JoinWorkers() { // a worker pinned in a blocking call, even a native wait, can only be freed by interrupting it Log.Error($"OrderRequestProcessingPool.Dispose(): workers did not stop within {(int)ShutdownTimeout.TotalSeconds} seconds, interrupting: {string.Join(", ", stuckThreads.Select(thread => thread.Name))}"); - // interrupting a started thread cannot throw, only the joins below can + // interrupting a started thread is not documented to throw on modern .NET, but guard it + // anyway so one worker can never abort the shutdown or skip interrupting the rest foreach (var thread in stuckThreads) { - thread.Interrupt(); + try + { + thread.Interrupt(); + } + catch (Exception exception) + { + Log.Error(exception, $"OrderRequestProcessingPool.Dispose(): failed to interrupt '{thread.Name}'"); + } } foreach (var thread in stuckThreads) { From 2d355812f70dec9550a9454f0373674372d9b8f7 Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Mon, 31 Aug 2026 11:14:22 -0400 Subject: [PATCH 11/14] Restore the parked queue lookup by indexer With the parked requests no longer re-enqueued at dispose, at most one thread holds a work item per key, so the key is always present: keep the indexer so an invariant break surfaces instead of being swallowed. --- Engine/TransactionHandlers/OrderRequestProcessingPool.cs | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/Engine/TransactionHandlers/OrderRequestProcessingPool.cs b/Engine/TransactionHandlers/OrderRequestProcessingPool.cs index ee219346ceef..cf144bd432b0 100644 --- a/Engine/TransactionHandlers/OrderRequestProcessingPool.cs +++ b/Engine/TransactionHandlers/OrderRequestProcessingPool.cs @@ -459,9 +459,8 @@ private void ProcessInOrder(WorkItem item) lock (_lock) { - // the key may already be gone: a worker interrupted at shutdown abandons its keys, so - // treat a missing key as the order being done - if (_inFlight.TryGetValue(item.Key, out var parked) && parked != null && parked.Count > 0) + var parked = _inFlight[item.Key]; + if (parked != null && parked.Count > 0) { request = parked.Dequeue(); } From 296f67da95407aebec11dce3c8a6ae2a445ee2f0 Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Mon, 31 Aug 2026 13:03:15 -0400 Subject: [PATCH 12/14] Give the workers a grace window before dropping the queued requests At exit the workers keep processing the queued requests normally, as they always have, until their shared deadline: only past it does the pool flag ShutdownDeadlineReached and the request handling invalidates whatever is drained afterwards, instead of sending it to the brokerage. --- .../BrokerageTransactionHandler.cs | 15 +++--- .../OrderRequestProcessingPool.cs | 19 +++++-- .../BrokerageTransactionHandlerTests.cs | 22 +++----- .../OrderRequestProcessingPoolTests.cs | 52 ++++++++++++++++++- 4 files changed, 80 insertions(+), 28 deletions(-) diff --git a/Engine/TransactionHandlers/BrokerageTransactionHandler.cs b/Engine/TransactionHandlers/BrokerageTransactionHandler.cs index af5692f6d1e4..d87a5fbf77a3 100644 --- a/Engine/TransactionHandlers/BrokerageTransactionHandler.cs +++ b/Engine/TransactionHandlers/BrokerageTransactionHandler.cs @@ -77,8 +77,6 @@ public class BrokerageTransactionHandler : ITransactionHandler /// requests in order while growing the pool on demand as the threads get saturated. /// private OrderRequestProcessingPool _threadPool; - // set once the handler starts exiting, so requests still queued are invalidated instead of processed - private volatile bool _droppingQueuedRequests; /// /// An OnOrderEvent handler slower than this gets the user warned, once. Settable for tests. @@ -278,8 +276,8 @@ protected virtual void InitializeTransactionThread() { Action processRequest = request => { - // at exit the pool still drains its queue through here, fail the request instead of processing it - if (_droppingQueuedRequests) + // past the pool's shutdown deadline the requests still queued are invalidated instead of processed + if (_threadPool.ShutdownDeadlineReached) { InvalidateDroppedRequest(request); return; @@ -818,9 +816,8 @@ public void AddOpenOrder(Order order, IAlgorithm algorithm) /// public void Exit() { - // from here on the requests still queued get invalidated as they drain instead of sent to the brokerage - _droppingQueuedRequests = true; - // Dispose drains the queued requests (CompleteAdding) and waits for the threads before stopping + // Dispose drains the queued requests (CompleteAdding) and waits for the threads before stopping; + // past its deadline the requests still queued are invalidated instead of processed _threadPool.DisposeSafely(); } @@ -1967,8 +1964,8 @@ private void InvalidateOrders(List orders, string message) } /// - /// Fails a request still queued while the handler exits, instead of processing it. Invalidates a dropped - /// submit's order so it doesn't linger as new in the final results, before they are sent. + /// Fails a request still queued past the pool's shutdown deadline, instead of processing it. Invalidates + /// a dropped submit's order so it doesn't linger as new in the final results, before they are sent. /// private void InvalidateDroppedRequest(OrderRequest request) { diff --git a/Engine/TransactionHandlers/OrderRequestProcessingPool.cs b/Engine/TransactionHandlers/OrderRequestProcessingPool.cs index cf144bd432b0..f55da2408fb0 100644 --- a/Engine/TransactionHandlers/OrderRequestProcessingPool.cs +++ b/Engine/TransactionHandlers/OrderRequestProcessingPool.cs @@ -67,6 +67,14 @@ public class OrderRequestProcessingPool : IDisposable // set under the lock when shutting down so the pool stops growing while the backlog is drained. // volatile, the workers read it lock free while draining private volatile bool _shuttingDown; + // set once the workers had their shared deadline to drain normally. volatile, read lock free + private volatile bool _shutdownDeadlineReached; + + /// + /// True once disposing has given the workers their shared deadline to drain normally: the requests + /// drained after this point should be dropped by the request handler instead of processed + /// + public bool ShutdownDeadlineReached => _shutdownDeadlineReached; // number of workers currently processing a request, used to decide when the pool is saturated private int _busyWorkers; private readonly Action _processRequest; @@ -249,10 +257,11 @@ private bool IsProcessing() } /// - /// Stops the pool. The requests still in the ready queue are drained through the normal processing loop, by - /// the surviving workers or by a last resort drainer thread, so the caller's request handler decides their - /// fate. Parked follow up requests are left with their owning worker and are dropped with it. Workers that - /// won't stop within the shared deadline are interrupted. + /// Stops the pool. The requests still in the ready queue are drained through the normal processing loop: + /// the surviving workers process them normally until the shared deadline, after which a last resort + /// drainer thread drains the rest, dropped by the request handler through + /// . Parked follow up requests are left with their owning worker + /// and are dropped with it. Workers that won't stop within the deadline are interrupted. /// public void Dispose() { @@ -279,6 +288,8 @@ public void Dispose() else { JoinWorkers(); + // the workers had their chance to drain normally, whatever is left is dropped by the handler + _shutdownDeadlineReached = true; // if every worker got stuck the backlog was never drained, hand it to a fresh background thread and // wait a bounded time, so Dispose stays bounded even when processing a request itself blocks diff --git a/Tests/Engine/BrokerageTransactionHandlerTests/BrokerageTransactionHandlerTests.cs b/Tests/Engine/BrokerageTransactionHandlerTests/BrokerageTransactionHandlerTests.cs index b48851c810ed..77bbbcf8a68b 100644 --- a/Tests/Engine/BrokerageTransactionHandlerTests/BrokerageTransactionHandlerTests.cs +++ b/Tests/Engine/BrokerageTransactionHandlerTests/BrokerageTransactionHandlerTests.cs @@ -1400,7 +1400,7 @@ public void CancelOrderTicket() } [Test] - public void ExitInvalidatesRequestsStillQueued() + public void ExitProcessesRequestsStillQueued() { _transactionHandler = new TestBrokerageTransactionHandler(); using var brokerage = new NoSubmitTestBrokerage(_algorithm); @@ -1422,21 +1422,15 @@ public void ExitInvalidatesRequestsStillQueued() brokerage.PublishOrderEvent(submitted); Assert.AreEqual(OrderStatus.Submitted, orderTicket.Status); - // queue a cancel without draining, like one issued in OnEndOfAlgorithm after the last pump - var cancelResponse = orderTicket.Cancel(); - Assert.IsFalse(cancelResponse.IsError); - var cancelRequest = orderTicket.CancelRequest; - Assert.AreEqual(OrderRequestStatus.Processing, cancelRequest.Status); - var orderEventsCount = _algorithm.OrderEvents.Count; + // queue an update without draining, like one issued in OnEndOfAlgorithm after the last pump + var updateRequest = new UpdateOrderRequest(DateTime.UtcNow, orderTicket.OrderId, new UpdateOrderFields { Quantity = 2000 }); + _transactionHandler.Process(updateRequest); + Assert.AreEqual(OrderRequestStatus.Processing, updateRequest.Status); - // at exit the queued cancel is invalidated instead of silently discarded or sent to the brokerage + // the synchronous drain at exit has no deadline: the queued request is still processed normally _transactionHandler.Exit(); - Assert.IsTrue(cancelRequest.Response.IsError); - Assert.AreEqual(OrderResponseErrorCode.ProcessingError, cancelRequest.Response.ErrorCode); - StringAssert.Contains("never sent to the brokerage", cancelRequest.Response.ErrorMessage); - // the cancel never reached the brokerage and only a submit drop emits an order event - Assert.AreNotEqual(OrderStatus.Canceled, orderTicket.Status); - Assert.AreEqual(orderEventsCount, _algorithm.OrderEvents.Count); + Assert.IsTrue(updateRequest.Response.IsSuccess); + Assert.AreEqual(2000, orderTicket.Quantity); } // A slow OnOrderEvent handler blocks the order event lock, and for Python the GIL, so the user is diff --git a/Tests/Engine/BrokerageTransactionHandlerTests/OrderRequestProcessingPoolTests.cs b/Tests/Engine/BrokerageTransactionHandlerTests/OrderRequestProcessingPoolTests.cs index 48b28b95cca3..90f68f513cde 100644 --- a/Tests/Engine/BrokerageTransactionHandlerTests/OrderRequestProcessingPoolTests.cs +++ b/Tests/Engine/BrokerageTransactionHandlerTests/OrderRequestProcessingPoolTests.cs @@ -1,4 +1,4 @@ -/* +/* * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation. * @@ -100,6 +100,56 @@ SubmitOrderRequest CreateSubmit(int orderId, string tag) } } + // Disposing lets the workers drain normally first: only past the shared deadline are the drained + // requests flagged to be dropped by the request handler. + [Test] + public void FlagsTheShutdownDeadlineOnlyAfterJoiningTheWorkers() + { + using var gate = new ManualResetEventSlim(false); + var processed = new ConcurrentQueue<(OrderRequest Request, bool PastDeadline)>(); + Exception processingError = null; + OrderRequestProcessingPool pool = null; + pool = new OrderRequestProcessingPool(concurrencyEnabled: true, minimumThreads: 1, maximumThreads: 1, + request => + { + processed.Enqueue((request, pool.ShutdownDeadlineReached)); + if (request.Tag == "poison") + { + gate.Wait(); + } + }, + exception => processingError = exception); + pool.ShutdownTimeout = TimeSpan.FromMilliseconds(500); + + try + { + var symbol = Symbols.SPY; + var reference = new DateTime(2025, 07, 03, 10, 0, 0); + var poison = new SubmitOrderRequest(OrderType.Market, symbol.SecurityType, symbol, 1, 0, 0, reference, "poison"); + poison.SetOrderId(1); + pool.Dispatch(poison, Order.CreateOrder(poison)); + Assert.IsTrue(SpinWait.SpinUntil(() => processed.Count >= 1, 10000), "the worker never got stuck"); + var queued = new SubmitOrderRequest(OrderType.Market, symbol.SecurityType, symbol, 1, 0, 0, reference, ""); + queued.SetOrderId(2); + pool.Dispatch(queued, Order.CreateOrder(queued)); + + Assert.IsFalse(pool.ShutdownDeadlineReached); + pool.Dispose(); + Assert.IsTrue(pool.ShutdownDeadlineReached); + + // the pinned worker took its request before the deadline, the drainer took the rest after it + foreach (var pair in processed) + { + Assert.AreEqual(pair.Request == queued, pair.PastDeadline); + } + Assert.IsNull(processingError, $"the pool reported an error: {processingError}"); + } + finally + { + gate.Set(); + } + } + [Test] public void DisposeInterruptsStuckWorkersAndDrainsTheBacklogOnANewThread() { From 275e3fb2a63b3d6e895251dcdc555a3a036b19b3 Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Mon, 31 Aug 2026 16:38:39 -0400 Subject: [PATCH 13/14] Leave the synchronous pool dispose untouched The synchronous pool, used by backtesting, keeps its historical dispose behavior: pending requests are left unprocessed, only the threaded pool gets the shutdown drain, deadline and interrupts. --- Engine/TransactionHandlers/OrderRequestProcessingPool.cs | 8 ++------ .../BrokerageTransactionHandlerTests.cs | 8 ++++---- .../OrderRequestProcessingPoolTests.cs | 6 +++--- 3 files changed, 9 insertions(+), 13 deletions(-) diff --git a/Engine/TransactionHandlers/OrderRequestProcessingPool.cs b/Engine/TransactionHandlers/OrderRequestProcessingPool.cs index f55da2408fb0..2fb067748582 100644 --- a/Engine/TransactionHandlers/OrderRequestProcessingPool.cs +++ b/Engine/TransactionHandlers/OrderRequestProcessingPool.cs @@ -280,12 +280,8 @@ public void Dispose() // through the normal loop, and their consuming enumerable ends on its own once the queue empties _readyQueue.CompleteAdding(); - if (_synchronous) - { - // backtesting only, no workers and no bounded time concern: drain the backlog on the caller thread - ProcessPending(); - } - else + // the synchronous pool has no workers and keeps its historical dispose behavior untouched + if (!_synchronous) { JoinWorkers(); // the workers had their chance to drain normally, whatever is left is dropped by the handler diff --git a/Tests/Engine/BrokerageTransactionHandlerTests/BrokerageTransactionHandlerTests.cs b/Tests/Engine/BrokerageTransactionHandlerTests/BrokerageTransactionHandlerTests.cs index 77bbbcf8a68b..5f397e01b6f1 100644 --- a/Tests/Engine/BrokerageTransactionHandlerTests/BrokerageTransactionHandlerTests.cs +++ b/Tests/Engine/BrokerageTransactionHandlerTests/BrokerageTransactionHandlerTests.cs @@ -1400,7 +1400,7 @@ public void CancelOrderTicket() } [Test] - public void ExitProcessesRequestsStillQueued() + public void ExitLeavesQueuedRequestsUntouchedInSynchronousMode() { _transactionHandler = new TestBrokerageTransactionHandler(); using var brokerage = new NoSubmitTestBrokerage(_algorithm); @@ -1427,10 +1427,10 @@ public void ExitProcessesRequestsStillQueued() _transactionHandler.Process(updateRequest); Assert.AreEqual(OrderRequestStatus.Processing, updateRequest.Status); - // the synchronous drain at exit has no deadline: the queued request is still processed normally + // the synchronous dispose leaves the queued request untouched, as it always has in backtesting _transactionHandler.Exit(); - Assert.IsTrue(updateRequest.Response.IsSuccess); - Assert.AreEqual(2000, orderTicket.Quantity); + Assert.AreEqual(OrderRequestStatus.Processing, updateRequest.Status); + Assert.AreEqual(1000, orderTicket.Quantity); } // A slow OnOrderEvent handler blocks the order event lock, and for Python the GIL, so the user is diff --git a/Tests/Engine/BrokerageTransactionHandlerTests/OrderRequestProcessingPoolTests.cs b/Tests/Engine/BrokerageTransactionHandlerTests/OrderRequestProcessingPoolTests.cs index 90f68f513cde..60e1eb6eddfd 100644 --- a/Tests/Engine/BrokerageTransactionHandlerTests/OrderRequestProcessingPoolTests.cs +++ b/Tests/Engine/BrokerageTransactionHandlerTests/OrderRequestProcessingPoolTests.cs @@ -249,7 +249,7 @@ public void DisposeOfAnIdlePoolIsFastAndQuiet() } [Test] - public void SynchronousPoolDisposeProcessesPendingRequests() + public void SynchronousPoolDisposeLeavesPendingRequestsUnprocessed() { var processed = new ConcurrentQueue(); Exception processingError = null; @@ -266,12 +266,12 @@ public void SynchronousPoolDisposeProcessesPendingRequests() pool.ProcessPending(); CollectionAssert.AreEqual(new OrderRequest[] { submit }, processed); - // queued after the last drain: Dispose pumps it through the processing loop on the caller thread + // queued after the last drain: the synchronous dispose leaves it untouched, as it always has var update = new UpdateOrderRequest(reference, order.Id, new UpdateOrderFields()); pool.Dispatch(update, order); pool.Dispose(); - CollectionAssert.AreEqual(new OrderRequest[] { submit, update }, processed); + CollectionAssert.AreEqual(new OrderRequest[] { submit }, processed); Assert.IsNull(processingError, $"the pool reported an error: {processingError}"); Assert.IsFalse(pool.IsActive); } From c909a359db2d25dc36a74c2285763538aca8b206 Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Mon, 31 Aug 2026 16:38:40 -0400 Subject: [PATCH 14/14] Measure the OnOrderEvent handler with the tick count Environment.TickCount64 is a single read of the OS tick counter, its millisecond resolution is plenty against the ten second threshold. --- Engine/TransactionHandlers/BrokerageTransactionHandler.cs | 5 ++--- .../BrokerageTransactionHandlerTests.cs | 4 ++-- 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/Engine/TransactionHandlers/BrokerageTransactionHandler.cs b/Engine/TransactionHandlers/BrokerageTransactionHandler.cs index d87a5fbf77a3..f7d92acd4a3f 100644 --- a/Engine/TransactionHandlers/BrokerageTransactionHandler.cs +++ b/Engine/TransactionHandlers/BrokerageTransactionHandler.cs @@ -16,7 +16,6 @@ using System; using System.Collections.Concurrent; using System.Collections.Generic; -using System.Diagnostics; using System.Linq; using System.Runtime.CompilerServices; using System.Threading; @@ -1389,9 +1388,9 @@ private void HandleOrderEvents(List orderEvents) { // a slow handler holds the order event lock, and for Python the GIL, stalling the // other transaction threads and the order status processing. - var start = Stopwatch.GetTimestamp(); + var start = Environment.TickCount64; _algorithm.OnOrderEvent(orderEvent); - var elapsed = Stopwatch.GetElapsedTime(start); + var elapsed = TimeSpan.FromMilliseconds(Environment.TickCount64 - start); if (elapsed > SlowOnOrderEventThreshold) { _slowOnOrderEventWarningSent = true; diff --git a/Tests/Engine/BrokerageTransactionHandlerTests/BrokerageTransactionHandlerTests.cs b/Tests/Engine/BrokerageTransactionHandlerTests/BrokerageTransactionHandlerTests.cs index 5f397e01b6f1..c87dbeb488a0 100644 --- a/Tests/Engine/BrokerageTransactionHandlerTests/BrokerageTransactionHandlerTests.cs +++ b/Tests/Engine/BrokerageTransactionHandlerTests/BrokerageTransactionHandlerTests.cs @@ -1441,8 +1441,8 @@ public void SlowOnOrderEventHandlerWarnsOnlyOnce() _transactionHandler = new TestBrokerageTransactionHandler(); using var brokerage = new NoSubmitTestBrokerage(_algorithm); _transactionHandler.Initialize(_algorithm, brokerage, new BacktestingResultHandler()); - // any handler run exceeds a zero threshold - _transactionHandler.SlowOnOrderEventThreshold = TimeSpan.Zero; + // any handler run exceeds a negative threshold, even one under the millisecond tick resolution + _transactionHandler.SlowOnOrderEventThreshold = TimeSpan.FromTicks(-1); var security = _algorithm.Securities[_symbol]; var price = 1.12m;