diff --git a/README.md b/README.md index 6939aab..621fcce 100644 --- a/README.md +++ b/README.md @@ -102,6 +102,12 @@ if (_tasks.DequeueTaskDisposedDataModel(out var record) == TaskManagementStatus. } ``` +## Examples + +| Example | Description | +|---|---| +| [Financial Order Processing](examples/FinancialOrderProcessing/README.md) | Multi-threaded market order pipeline — producer, 5 trading desk workers, live monitor, graceful shutdown, and observability queue drain | + ## License MIT © Jose Luis Guerra Infante diff --git a/examples/FinancialOrderProcessing/FinancialOrderProcessing.csproj b/examples/FinancialOrderProcessing/FinancialOrderProcessing.csproj new file mode 100644 index 0000000..749ba46 --- /dev/null +++ b/examples/FinancialOrderProcessing/FinancialOrderProcessing.csproj @@ -0,0 +1,17 @@ + + + + Exe + net8.0 + enable + disable + FinancialOrderProcessing + + + + + + + + + diff --git a/examples/FinancialOrderProcessing/MarketMonitor.cs b/examples/FinancialOrderProcessing/MarketMonitor.cs new file mode 100644 index 0000000..9760b4f --- /dev/null +++ b/examples/FinancialOrderProcessing/MarketMonitor.cs @@ -0,0 +1,56 @@ +using System; +using System.Collections.Concurrent; +using System.Threading; +using NetFramework.Tasks.Management.Abstractions.Interfaces; + +namespace FinancialOrderProcessing +{ + /// + /// Runs every 2 seconds and prints a live status table by combining: + /// - GetTasksStatus() → actual TPL task state from the library + /// - WorkerMetrics → application-level order statistics + /// + public static class MarketMonitor + { + public static Action CreateAction( + ITaskManagement taskManagement, + ConcurrentDictionary metrics, + ConcurrentQueue queue) + { + return (state) => + { + var cts = (CancellationTokenSource)state; + + while (!cts.IsCancellationRequested) + { + Thread.Sleep(2_000); + + if (cts.IsCancellationRequested) break; + + var tplStatuses = taskManagement.GetTasksStatus(); + var now = DateTime.Now; + + Console.WriteLine(); + Console.WriteLine($"[{now:HH:mm:ss}] ── LIVE STATUS ───────────────────────────────────────────────────────"); + Console.WriteLine($" Queue depth : {queue.Count,8:N0} orders pending"); + Console.WriteLine(); + Console.WriteLine($" {"Task",-22} {"TPL State",-14} {"Orders",9} {"Avg ms",8} {"Min ms",8} {"Max ms",8}"); + Console.WriteLine($" {new string('─', 76)}"); + + foreach (var (name, m) in metrics) + { + tplStatuses.TryGetValue(name, out var tplState); + string avg = m.OrdersHandled > 0 ? $"{m.AverageMs:F1}" : "─"; + string min = m.OrdersHandled > 0 ? m.MinMs.ToString() : "─"; + string max = m.OrdersHandled > 0 ? m.MaxMs.ToString() : "─"; + + Console.WriteLine( + $" {name,-22} {tplState,-14} {m.OrdersHandled,9:N0} {avg,8} {min,8} {max,8}"); + } + + Console.WriteLine($" {new string('─', 76)}"); + } + }; + } + } +} diff --git a/examples/FinancialOrderProcessing/MarketOrder.cs b/examples/FinancialOrderProcessing/MarketOrder.cs new file mode 100644 index 0000000..38b6853 --- /dev/null +++ b/examples/FinancialOrderProcessing/MarketOrder.cs @@ -0,0 +1,28 @@ +using System; + +namespace FinancialOrderProcessing +{ + public enum OrderType { Buy, Sell } + + public class MarketOrder + { + public int Id { get; } + public string Symbol { get; } + public OrderType Type { get; } + public int Quantity { get; } + public decimal Price { get; } + public DateTime CreatedAt { get; } + + public MarketOrder(int id, string symbol, OrderType type, int quantity, decimal price, DateTime createdAt) + { + Id = id; + Symbol = symbol; + Type = type; + Quantity = quantity; + Price = price; + CreatedAt = createdAt; + } + + public decimal NotionalValue => Quantity * Price; + } +} diff --git a/examples/FinancialOrderProcessing/OrderProducer.cs b/examples/FinancialOrderProcessing/OrderProducer.cs new file mode 100644 index 0000000..37e51cd --- /dev/null +++ b/examples/FinancialOrderProcessing/OrderProducer.cs @@ -0,0 +1,54 @@ +using System; +using System.Collections.Concurrent; +using System.Threading; + +namespace FinancialOrderProcessing +{ + /// + /// Generates market orders at ~1 000 orders/sec into the shared queue. + /// Works in bursts of 20 orders followed by a 20 ms pause so the OS + /// scheduler can give CPU time to the worker desks. + /// + public static class OrderProducer + { + private static readonly string[] Symbols = + { + "AAPL", "MSFT", "GOOGL", "TSLA", "AMZN", + "META", "NVDA", "JPM", "BAC", "GS", + "BRK.B", "V", "MA", "UNH", "HD" + }; + + public static Action CreateAction( + ConcurrentQueue queue, + WorkerMetrics metrics) + { + return (state) => + { + var cts = (CancellationTokenSource)state; + var rng = new Random(); + int orderId = 0; + + while (!cts.IsCancellationRequested) + { + // Burst: enqueue 20 orders then yield briefly + for (int i = 0; i < 20 && !cts.IsCancellationRequested; i++) + { + var order = new MarketOrder( + id: ++orderId, + symbol: Symbols[rng.Next(Symbols.Length)], + type: rng.Next(2) == 0 ? OrderType.Buy : OrderType.Sell, + quantity: rng.Next(100, 10_001), + price: Math.Round((decimal)(rng.NextDouble() * 999 + 1), 2), + createdAt: DateTime.UtcNow + ); + + queue.Enqueue(order); + metrics.IncrementCount(); + } + + Thread.Sleep(20); // ~1 000 orders/sec sustained + } + }; + } + } +} diff --git a/examples/FinancialOrderProcessing/Program.cs b/examples/FinancialOrderProcessing/Program.cs new file mode 100644 index 0000000..8a9df4a --- /dev/null +++ b/examples/FinancialOrderProcessing/Program.cs @@ -0,0 +1,175 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; +using System.Threading; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using NetFramework.Tasks.Management; +using NetFramework.Tasks.Management.Abstractions.Enums; +using NetFramework.Tasks.Management.Abstractions.Interfaces; + +namespace FinancialOrderProcessing +{ + internal class Program + { + private const int SimulationSeconds = 15; + + private static readonly string[] DeskNames = + { + "desk-equities", + "desk-bonds", + "desk-fx", + "desk-derivatives", + "desk-commodities", + }; + + static void Main() + { + // ── Dependency Injection ────────────────────────────────────────────── + // AddTaskManagement() needs a plain ILogger (non-generic) from the container. + var services = new ServiceCollection(); + services.AddLogging(b => b + .AddConsole() + .SetMinimumLevel(LogLevel.Error)); // suppress library warn/info noise during demo + services.AddSingleton(sp => + sp.GetRequiredService() + .CreateLogger("NetFramework.Tasks.Management")); + services.AddTaskManagement(); + + var provider = services.BuildServiceProvider(); + var tasks = provider.GetRequiredService(); + + // ── Shared state ────────────────────────────────────────────────────── + var orderQueue = new ConcurrentQueue(); + var metrics = new ConcurrentDictionary(); + + // ── Register & start: Producer ──────────────────────────────────────── + var producerMetrics = new WorkerMetrics { StartedAt = DateTime.UtcNow }; + metrics["producer"] = producerMetrics; + + var producerCts = new CancellationTokenSource(); + tasks.RegisterTask("producer", + OrderProducer.CreateAction(orderQueue, producerMetrics), + producerCts); + tasks.StartTask("producer"); + + // ── Register & start: Trading desk workers ──────────────────────────── + foreach (var name in DeskNames) + { + var workerMetrics = new WorkerMetrics { StartedAt = DateTime.UtcNow }; + metrics[name] = workerMetrics; + + var cts = new CancellationTokenSource(); + tasks.RegisterTask(name, + TradingDeskWorker.CreateAction(name, orderQueue, workerMetrics), + cts); + tasks.StartTask(name); + } + + // ── Register & start: Market monitor ───────────────────────────────── + var monitorCts = new CancellationTokenSource(); + tasks.RegisterTask("monitor", + MarketMonitor.CreateAction(tasks, metrics, orderQueue), + monitorCts); + tasks.StartTask("monitor"); + + // ── Simulation run ──────────────────────────────────────────────────── + Console.WriteLine("╔══════════════════════════════════════════════════════════════════════╗"); + Console.WriteLine("║ Financial Order Processing — Simulation ║"); + Console.WriteLine($"║ Duration : {SimulationSeconds}s | Desks : {DeskNames.Length} | Press any key to stop ║"); + Console.WriteLine("╚══════════════════════════════════════════════════════════════════════╝"); + Console.WriteLine(); + Console.WriteLine($" Tasks registered and started:"); + Console.WriteLine($" producer — generating ~1 000 orders/sec"); + foreach (var n in DeskNames) + Console.WriteLine($" {n} — executing orders (5–75 ms latency model)"); + Console.WriteLine($" monitor — live status table every 2 s"); + + var simulationClock = Stopwatch.StartNew(); + + while (simulationClock.Elapsed.TotalSeconds < SimulationSeconds) + { + try + { + if (Console.KeyAvailable) + { + Console.ReadKey(intercept: true); + break; + } + } + catch (InvalidOperationException) { /* stdin redirected — ignore */ } + + Thread.Sleep(200); + } + + simulationClock.Stop(); + + // ── Graceful shutdown ───────────────────────────────────────────────── + Console.WriteLine(); + Console.WriteLine($"[{DateTime.Now:HH:mm:ss}] Shutdown initiated — cancelling all tasks..."); + + var cancelFailures = new ConcurrentDictionary(); + tasks.CancelAllTasks(except: null, tasksCancelPetitionFailedRef: ref cancelFailures); + + var completionStatuses = new ConcurrentDictionary(); + tasks.CheckAllTaskStatusCompleted( + except: null, + tasksCompletedStatusRef: ref completionStatuses, + retry: 3, + millisecondsCancellationWait: 8_000); + + foreach (var m in metrics.Values) + m.FinishedAt = DateTime.UtcNow; + + // Delete every task and push its disposal record onto the observability queue + foreach (var name in DeskNames.Concat(new[] { "producer", "monitor" })) + tasks.DeleteTask(name, sendDisposedDataToInternalQueue: true); + + // ── Final report ────────────────────────────────────────────────────── + long totalProcessed = DeskNames.Sum(n => metrics[n].OrdersHandled); + long totalGenerated = metrics["producer"].OrdersHandled; + int remainingInQueue = orderQueue.Count; + + Console.WriteLine(); + Console.WriteLine("╔══════════════════════════════════════════════════════════════════════╗"); + Console.WriteLine("║ FINAL REPORT ║"); + Console.WriteLine("╚══════════════════════════════════════════════════════════════════════╝"); + Console.WriteLine(); + Console.WriteLine($" Simulation wall time : {simulationClock.Elapsed:mm\\:ss\\.fff}"); + Console.WriteLine($" Orders generated : {totalGenerated,10:N0}"); + Console.WriteLine($" Orders processed : {totalProcessed,10:N0}"); + Console.WriteLine($" Orders remaining : {remainingInQueue,10:N0} ({remainingInQueue * 100.0 / Math.Max(totalGenerated, 1):F1}% unfilled)"); + Console.WriteLine(); + Console.WriteLine($" {"Desk",-22} {"Orders",9} {"Avg ms",8} {"Min ms",8} {"Max ms",8} {"Throughput",13}"); + Console.WriteLine($" {new string('─', 78)}"); + + foreach (var name in DeskNames) + { + var m = metrics[name]; + var elapsed = (m.FinishedAt == default ? DateTime.UtcNow : m.FinishedAt) - m.StartedAt; + var throughput = elapsed.TotalSeconds > 0 + ? m.OrdersHandled / elapsed.TotalSeconds + : 0; + + Console.WriteLine( + $" {name,-22} {m.OrdersHandled,9:N0} {m.AverageMs,8:F1} {m.MinMs,8} {m.MaxMs,8} {throughput,12:F1}/s"); + } + + // ── Observability queue drain ───────────────────────────────────────── + Console.WriteLine(); + Console.WriteLine($" {"─── Observability queue",78}"); + int dequeued = 0; + while (tasks.DequeueTaskDisposedDataModel(out var record) == TaskManagementStatus.ObjectInfoDequeued) + { + Console.WriteLine( + $" Task '{record.TaskName,-20}' id={record.TaskId,5}" + + $" final status = {record.TaskStatus,-18} disposed = {record.IsDisposed}"); + dequeued++; + } + Console.WriteLine($" Total records dequeued: {dequeued}"); + Console.WriteLine(); + } + } +} diff --git a/examples/FinancialOrderProcessing/README.md b/examples/FinancialOrderProcessing/README.md new file mode 100644 index 0000000..1362177 --- /dev/null +++ b/examples/FinancialOrderProcessing/README.md @@ -0,0 +1,136 @@ +# Financial Order Processing — Example + +A runnable simulation that demonstrates every major feature of `NetFramework.Tasks.Management` under realistic multi-threaded load, modelled on a financial market order processing pipeline. + +## What it simulates + +A trading platform receives a continuous stream of market orders (Buy/Sell across 15 symbols) and routes them through five specialised trading desks running in parallel. A market monitor reports live progress every two seconds. After 15 seconds the coordinator initiates a graceful shutdown and prints a full performance report. + +``` +┌─────────────┐ ConcurrentQueue +│ Producer │ ──────────────────────────────────────────────┐ +│ ~1 000/sec │ │ +└─────────────┘ ▼ + ┌───────────────────────────┐ + │ desk-equities │ + │ desk-bonds │ + │ desk-fx workers │ + │ desk-derivatives (~20/s │ + │ desk-commodities each) │ + └───────────────────────────┘ +┌─────────────┐ +│ Monitor │ GetTasksStatus() + WorkerMetrics → live table every 2 s +└─────────────┘ + +┌─────────────┐ +│ Coordinator │ CancelAllTasks → CheckAllTaskStatusCompleted +│ (Main) │ → DeleteTask (sendToQueue: true) × 7 +│ │ → DequeueTaskDisposedDataModel → final report +└─────────────┘ +``` + +## Execution latency model + +Each worker simulates realistic exchange round-trip times: + +| Factor | Cost | +|---|---| +| Base network RTT | 5 ms | +| Exchange queue jitter | 0 – 40 ms | +| Large order (> 5 000 shares) | +20 ms | +| High-price security (> $500) | +10 ms | +| **Worst case** | **75 ms** | + +At ~47 ms average, five desks process roughly **100 orders/sec combined** against a producer rate of **~1 000 orders/sec**, so the queue grows steadily — intentional, to demonstrate backpressure. + +## Library features exercised + +| Feature | Where | +|---|---| +| `RegisterTask` / `StartTask` | All 7 tasks (1 producer, 5 desks, 1 monitor) | +| `GetTasksStatus()` | `MarketMonitor` — live TPL state per task | +| `CancelAllTasks` | Coordinator graceful shutdown | +| `CheckAllTaskStatusCompleted` | Coordinator waits for all tasks to drain | +| `DeleteTask(sendToQueue: true)` | Coordinator cleanup — feeds observability queue | +| `DequeueTaskDisposedDataModel` | Final report — drains disposal records | + +## Project structure + +``` +FinancialOrderProcessing/ +├── Program.cs DI setup, lifecycle coordination, final report +├── MarketOrder.cs Order record and OrderType enum +├── WorkerMetrics.cs Lock-free thread-safe stats (Interlocked / CAS) +├── OrderProducer.cs Burst-generates market orders into the shared queue +├── TradingDeskWorker.cs Dequeues and executes orders with latency model +└── MarketMonitor.cs Prints live status table via GetTasksStatus() +``` + +## Running + +From the repository root: + +```bash +dotnet run --project examples/FinancialOrderProcessing/FinancialOrderProcessing.csproj +``` + +Press any key to stop early. The default duration is 15 seconds. + +## Sample output + +``` +╔══════════════════════════════════════════════════════════════════════╗ +║ Financial Order Processing — Simulation ║ +║ Duration : 15s | Desks : 5 | Press any key to stop ║ +╚══════════════════════════════════════════════════════════════════════╝ + +[02:16:11] ── LIVE STATUS ─────────────────────────────────────────────────────── + Queue depth : 1.102 orders pending + + Task TPL State Orders Avg ms Min ms Max ms + ──────────────────────────────────────────────────────────────────────────── + desk-derivatives Running 43 45,5 15 78 + desk-equities Running 43 45,6 14 77 + producer Running 1.320 0,0 0 0 + desk-bonds Running 41 48,6 14 78 + desk-commodities Running 45 43,8 16 79 + desk-fx Running 41 48,2 15 78 + ──────────────────────────────────────────────────────────────────────────── + +[02:16:24] Shutdown initiated — cancelling all tasks... + +╔══════════════════════════════════════════════════════════════════════╗ +║ FINAL REPORT ║ +╚══════════════════════════════════════════════════════════════════════╝ + + Simulation wall time : 00:15.020 + Orders generated : 9.660 + Orders processed : 1.579 + Orders remaining : 8.081 (83,7% unfilled) + + Desk Orders Avg ms Min ms Max ms Throughput + ────────────────────────────────────────────────────────────────────────────── + desk-equities 317 47,1 14 79 19,7/s + desk-bonds 328 45,4 14 79 20,4/s + desk-fx 314 47,4 14 79 19,5/s + desk-derivatives 309 48,2 14 79 19,2/s + desk-commodities 311 48,0 15 79 19,3/s + + ─── Observability queue + Task 'desk-equities ' id=1 final status = RanToCompletion disposed = True + Task 'desk-bonds ' id=2 final status = RanToCompletion disposed = True + Task 'desk-fx ' id=3 final status = RanToCompletion disposed = True + Task 'desk-derivatives ' id=4 final status = RanToCompletion disposed = True + Task 'desk-commodities ' id=5 final status = RanToCompletion disposed = True + Task 'producer ' id=6 final status = RanToCompletion disposed = True + Task 'monitor ' id=7 final status = RanToCompletion disposed = True + Total records dequeued: 7 +``` + +## Dependencies + +| Package | Purpose | +|---|---| +| `NetFramework.Tasks.Management` | Task lifecycle management (project reference) | +| `Microsoft.Extensions.DependencyInjection` | DI container setup | +| `Microsoft.Extensions.Logging.Console` | Console log sink (set to Error level to suppress library noise) | diff --git a/examples/FinancialOrderProcessing/TradingDeskWorker.cs b/examples/FinancialOrderProcessing/TradingDeskWorker.cs new file mode 100644 index 0000000..37ffa78 --- /dev/null +++ b/examples/FinancialOrderProcessing/TradingDeskWorker.cs @@ -0,0 +1,56 @@ +using System; +using System.Collections.Concurrent; +using System.Diagnostics; +using System.Threading; + +namespace FinancialOrderProcessing +{ + /// + /// Simulates a trading desk that dequeues and executes market orders. + /// + /// Execution latency model (realistic): + /// base 5 ms — network round-trip to exchange + /// jitter 0–40 ms — exchange queue depth / matching engine variance + /// size +20 ms — large orders (> 5 000 shares) need partial fills + /// price +10 ms — high-price securities route through a slower ECN + /// + /// Total worst case: ~75 ms per order. + /// + public static class TradingDeskWorker + { + public static Action CreateAction( + string deskName, + ConcurrentQueue queue, + WorkerMetrics metrics) + { + return (state) => + { + var cts = (CancellationTokenSource)state; + var rng = new Random(); + + while (!cts.IsCancellationRequested) + { + if (!queue.TryDequeue(out MarketOrder? order)) + { + // Queue is empty — back off 1 ms before retrying + Thread.Sleep(1); + continue; + } + + var sw = Stopwatch.StartNew(); + + int latencyMs = + 5 // base: network RTT + + rng.Next(0, 41) // jitter: exchange variance + + (order.Quantity > 5_000 ? 20 : 0) // size factor + + (order.Price > 500m ? 10 : 0); // price/ECN factor + + Thread.Sleep(latencyMs); + + sw.Stop(); + metrics.RecordExecution(sw.ElapsedMilliseconds); + } + }; + } + } +} diff --git a/examples/FinancialOrderProcessing/WorkerMetrics.cs b/examples/FinancialOrderProcessing/WorkerMetrics.cs new file mode 100644 index 0000000..64d3095 --- /dev/null +++ b/examples/FinancialOrderProcessing/WorkerMetrics.cs @@ -0,0 +1,62 @@ +using System; +using System.Threading; + +namespace FinancialOrderProcessing +{ + /// + /// Lock-free, thread-safe statistics for a single task. + /// Uses Interlocked and CAS loops so multiple worker threads can update + /// without taking a lock. + /// + public class WorkerMetrics + { + private long _ordersHandled; + private long _totalProcessingMs; + private long _minMs = long.MaxValue; + private long _maxMs; + + public DateTime StartedAt { get; set; } + public DateTime FinishedAt { get; set; } + + // ── Read-only snapshots ─────────────────────────────────────────────── + + public long OrdersHandled => Volatile.Read(ref _ordersHandled); + public long TotalProcessingMs => Volatile.Read(ref _totalProcessingMs); + public long MaxMs => Volatile.Read(ref _maxMs); + + public long MinMs + { + get + { + long v = Volatile.Read(ref _minMs); + return v == long.MaxValue ? 0 : v; + } + } + + public double AverageMs => + OrdersHandled > 0 ? TotalProcessingMs / (double)OrdersHandled : 0; + + // ── Writers (called from worker threads) ────────────────────────────── + + /// Counts one order without recording latency (used by the producer). + public void IncrementCount() => Interlocked.Increment(ref _ordersHandled); + + /// Records one completed order with its processing time. + public void RecordExecution(long processingMs) + { + Interlocked.Increment(ref _ordersHandled); + Interlocked.Add(ref _totalProcessingMs, processingMs); + + // CAS loop for min + long current; + do { current = Volatile.Read(ref _minMs); } + while (processingMs < current && + Interlocked.CompareExchange(ref _minMs, processingMs, current) != current); + + // CAS loop for max + do { current = Volatile.Read(ref _maxMs); } + while (processingMs > current && + Interlocked.CompareExchange(ref _maxMs, processingMs, current) != current); + } + } +}