Concurrency Model — Trade Signal Terminal

Live Pipeline Counters live

The endpoint /api/concurrency/threads exposes an approximate demo breakdown. Feed and normalizer use generation rate. Risk/routing use market-data drain rate. UI coalescer uses SignalR snapshot cadence.

Pipeline stage msgs/sec

Feed—

Normalizer—

Risk—

Routing—

UI Coalescer—

Actual Server-Side Threading Model

ASP.NET Core host
│
├── BackgroundService: MarketDataSimulator
│   └── publishes TradeSignal into IMarketDataBus
│
├── Singleton: InMemoryMarketDataBus
│   ├── LosslessTickStore.OnMarketData(signal)
│   └── TradeQueueProcessor.OnMarketData(signal)
│
├── BackgroundService: LosslessTickStore
│   └── Channel<TradeSignal> capacity 1,000,000, FullMode=Wait
│       └── recent ring buffer 500,000 + optional journal batches
│
├── BackgroundService: TradeQueueProcessor
│   └── Channel<TradeSignal> capacity 100,000, FullMode=DropOldest
│       └── PeriodicTimer 33ms → drain all available → latest-by-symbol snapshot → SignalR
│
├── BackgroundService: OrderCommandProcessor
│   └── Channel<QueuedOrderCommand> capacity 4,096, FullMode=Wait
│       └── single-reader FIFO submit/cancel/modify dispatch
│
└── REST request threads
    ├── read-only endpoints query singleton state
    └── mutating order endpoints enqueue commands and await completion

Key correction: the current implementation is not one generic 10K queue with batch-of-50 SignalR sends. Market data has two separate subscriber queues with different backpressure policies, and mutating order operations have their own FIFO command queue.

Market Data Fan-Out

MarketDataSimulator generates synthetic ticks in a hosted loop and publishes each tick once to IMarketDataBus. The in-memory bus synchronously fans the signal out to the fixed subscriber set from dependency injection.

CURRENT FAN-OUT IMPLEMENTATION

public sealed class InMemoryMarketDataBus : IMarketDataBus
{
    private readonly IMarketDataSubscriber[] _subscribers;

    public InMemoryMarketDataBus(IEnumerable<IMarketDataSubscriber> subscribers)
    {
        _subscribers = subscribers.ToArray();
    }

    public void Publish(TradeSignal signal)
    {
        foreach (var subscriber in _subscribers)
        {
            subscriber.OnMarketData(signal);
        }
    }
}

The important concurrency detail is that subscribers own their own channels. The producer does not directly write to one shared queue, and it does not wait for SignalR clients.

System.Threading.Channels<T> in This Repo

The API uses bounded channels in three places, each tuned for a different correctness/latency tradeoff.

Channel owner	Capacity	Full mode	Concurrency semantics	Purpose
`TradeQueueProcessor`	`100_000`	`DropOldest`	single reader, multiple writers	Fresh coalesced UI snapshots
`LosslessTickStore`	`1_000_000`	`Wait`	single reader, multiple writers	Recent replay ring + optional journal
`OrderCommandQueue`	`4_096`	`Wait`	single reader, multiple writers	FIFO order submit/cancel/modify serialization

UI MARKET-DATA CHANNEL

_channel = Channel.CreateBounded<TradeSignal>(new BoundedChannelOptions(ChannelCapacity)
{
    FullMode = BoundedChannelFullMode.DropOldest,
    SingleReader = true,
    SingleWriter = false
});

ORDER COMMAND CHANNEL

_channel = Channel.CreateBounded<QueuedOrderCommand>(new BoundedChannelOptions(ChannelCapacity)
{
    FullMode = BoundedChannelFullMode.Wait,
    SingleReader = true,
    SingleWriter = false
});

UI Coalescing Loop

TradeQueueProcessor does not call SignalR once per raw tick. Every 33ms it drains whatever is available, overwrites a dictionary entry per symbol, and broadcasts one compact snapshot.

DRAIN-THEN-SNAPSHOT LOOP

using var timer = new PeriodicTimer(TimeSpan.FromMilliseconds(33));
var latestBySymbol = new Dictionary<string, TradeSignal>(capacity: 64);
var snapshot = new List<TradeSignal>(capacity: 64);

while (await timer.WaitForNextTickAsync(stoppingToken))
{
    var drained = 0;
    while (_channel.Reader.TryRead(out var signal))
    {
        latestBySymbol[signal.Symbol] = signal;
        drained++;
    }

    if (drained > 0)
    {
        Interlocked.Add(ref _processedCount, drained);
    }

    if (latestBySymbol.Count > 0)
    {
        snapshot.Clear();
        foreach (var signal in latestBySymbol.Values)
        {
            snapshot.Add(signal);
        }
        latestBySymbol.Clear();

        await _hubContext.Clients.All.SendAsync("TradeSignals", snapshot, stoppingToken);
        Interlocked.Increment(ref _broadcastCount);
    }
}

This keeps the browser and WPF terminal responsive under raw tick rates that are much higher than any UI should render directly.

Order Command Serialization

Mutating order requests are serialized through OrderCommandQueue. The HTTP handler enqueues a command and awaits its completion task; the hosted processor is the only reader.

HTTP MUTATION PATH

app.MapPost("/api/orders", async (Order order, IOrderCommandQueue orderQueue, CancellationToken ct) =>
{
    var outcome = await orderQueue.EnqueueAsync(new SubmitOrderCommand(order), ct);
    return outcome.Status switch
    {
        OrderCommandOutcomeStatus.Completed => Results.Ok(outcome.Result),
        OrderCommandOutcomeStatus.Canceled => Results.StatusCode(StatusCodes.Status499ClientClosedRequest),
        _ => Results.Problem(outcome.ErrorMessage ?? "Order command failed")
    };
});

SINGLE-READER COMMAND LOOP

await foreach (var queued in _queue.Reader.ReadAllAsync(stoppingToken))
{
    _queue.RecordDequeued();

    var queueWait = DateTime.UtcNow - queued.Command.CreatedAt;
    var outcome = await DispatchAsync(queued.Command, queueWait, stoppingToken);

    queued.Complete(outcome);
    _queue.RecordCompleted(outcome);
}

This gives deterministic in-memory state transitions for submit, cancel, and modify operations without pretending to be a separate production OMS.

Synchronization Patterns

Component	Pattern	Why
`PerformanceMetrics`	`Interlocked` + fixed sample array + `ArrayPool<long>`	Low-overhead latency counters and percentile snapshots
`GenerationStats`	`Interlocked` + `Volatile`	Cheap generator throughput accounting
`TradeQueueProcessor`	`Interlocked` counters	Metrics without locking the enqueue/drain hot path
`OrderCommandQueue`	`Interlocked` counters + channel serialization	Queue stats and FIFO mutation ordering
`ExchangeSimulator`	`lock` around in-memory order/depth state	Correctness for shared demo state; not the raw market-data hot path
`PositionManager`	`lock` around position state	Consistent position/PnL updates
WPF client	`ConcurrentQueue<TradeSignalDto>` + dispatcher timers	SignalR callbacks enqueue; UI thread drains at bounded cadence

LOCK-FREE PEAK LATENCY TRACKING

public void RecordLatency(long elapsedTicks)
{
    var idx = Interlocked.Increment(ref _sampleIndex) & (SampleBufferSize - 1);
    _latencySamples[idx] = elapsedTicks;
    Interlocked.Increment(ref _totalMessages);
    Interlocked.Increment(ref _windowMessageCount);

    long current = Volatile.Read(ref _peakLatencyTicks);
    while (elapsedTicks > current)
    {
        long next = Interlocked.CompareExchange(ref _peakLatencyTicks, elapsedTicks, current);
        if (next == current)
        {
            break;
        }
        current = next;
    }
}

Memory & Allocation Strategy

The high-volume market-data code minimizes allocation and avoids locks where it matters most. The current implementation also deliberately uses ordinary locks in lower-volume stateful simulation code where correctness and simplicity are more important.

Area	Current implementation
Trade signal model	`readonly record struct TradeSignal`
UI snapshot buffer	Reusable `List<TradeSignal>(64)` inside `TradeQueueProcessor`
Recent tick store	Fixed `TradeSignal[500_000]` ring buffer
Latency samples	Fixed `long[4096]`, power-of-two index mask
Percentile sort buffer	`ArrayPool<long>.Shared.Rent(...)` and return
GC mode	`GCSettings.LatencyMode = GCLatencyMode.SustainedLowLatency` in `Program.cs`

PERCENTILE SNAPSHOT BUFFERING

var sampleCount = Math.Min(Interlocked.Read(ref _sampleIndex), SampleBufferSize);
var rentedBuffer = ArrayPool<long>.Shared.Rent((int)sampleCount);
try
{
    Array.Copy(_latencySamples, rentedBuffer, (int)sampleCount);
    Array.Sort(rentedBuffer, 0, (int)sampleCount);

    var p50Ticks = sampleCount > 0 ? rentedBuffer[(int)(sampleCount * 0.50)] : 0;
    var p95Ticks = sampleCount > 0 ? rentedBuffer[(int)(sampleCount * 0.95)] : 0;
    var p99Ticks = sampleCount > 0 ? rentedBuffer[(int)(sampleCount * 0.99)] : 0;
}
finally
{
    ArrayPool<long>.Shared.Return(rentedBuffer);
}

Client-Side Concurrency

The browser client receives SignalR snapshots and polls selected REST views. The WPF client uses SignalR callbacks to enqueue received DTOs, then drains them on a dispatcher timer so the UI thread does bounded work per frame.

WPF CLIENT DRAIN QUEUE

public FeedDrainResult Drain(int maxFeedRowsPerFrame)
{
    var latestBySymbol = new Dictionary<string, TradeSignalDto>(StringComparer.OrdinalIgnoreCase);
    var feedWindow = new Queue<TradeSignalDto>(maxFeedRowsPerFrame);
    var dequeued = 0;

    while (_pendingSignals.TryDequeue(out var signal))
    {
        dequeued++;
        latestBySymbol[signal.Symbol] = signal;

        if (feedWindow.Count >= maxFeedRowsPerFrame)
        {
            feedWindow.Dequeue();
        }

        feedWindow.Enqueue(signal);
    }

    return new FeedDrainResult(dequeued, latestBySymbol.Values.ToArray(), feedWindow.Reverse().ToArray());
}