C# Async & Multithreading

// ======================================================================
// Q: IAsyncEnumerable and await foreach: when and how?
//
// When data arrives gradually and you don't want to load it all into memory:
// exports, large queries, message consumers, paging through an external API.
// - [EnumeratorCancellation] is required for .WithCancellation(ct) to work.
// - An IAsyncEnumerable from EF Core keeps the connection open while you read.
//   Don't make slow calls inside the loop.
// - In ASP.NET Core, an action that returns IAsyncEnumerable<T> streams JSON
//   (demo: GET /api/orders/export).
// ======================================================================

public async IAsyncEnumerable<ProductDto> GetAllFromPimAsync(
    [EnumeratorCancellation] CancellationToken ct = default)
{
    string? cursor = null;
    do
    {
        var page = await pim.GetProductsAsync(cursor, ct);
        foreach (var p in page.Items) yield return p;    // the consumer gets the first items before all of them arrive
        cursor = page.NextCursor;
    } while (cursor is not null);
}

await foreach (var p in GetAllFromPimAsync().WithCancellation(ct)) { /* ... */ }

// ======================================================================
// Q: How do you implement timeouts and cooperative cancellation correctly?
//
// - Put a using on the CancellationTokenSource, because it holds a timer.
// - await task.WaitAsync(timeout, ct) stops the waiting, not the operation,
//   which keeps running. Use it only when the API doesn't accept a token.
// - An OperationCanceledException because the client closed the connection is
//   not an error. Don't log it as a 500.
// - For HTTP calls: put the timeout in the resilience pipeline (Polly, Q11), not
//   by hand everywhere.
// ======================================================================

// Timeout for a single call that also respects request cancellation
using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
cts.CancelAfter(TimeSpan.FromSeconds(5));
try
{
    return await erp.GetWarehouseStockAsync(sku, cts.Token);
}
catch (OperationCanceledException) when (!ct.IsCancellationRequested)
{
    throw new TimeoutException($"ERP stock lookup for {sku} timed out");   // timeout, not cancellation by the client
}

// CPU loop: cancellation is cooperative, you have to check it yourself
foreach (var row in rows)
{
    ct.ThrowIfCancellationRequested();
    Process(row);
}

// ======================================================================
// Q: How do you implement producer/consumer? Why Channel<T>?
//
// - Channel is async: no thread blocks while waiting, unlike BlockingCollection.
// - Bounded = backpressure, so memory doesn't fill up if the consumer is slow.
//   Unbounded only when the rate is under control.
// - It is in-process: if the process dies, the messages are lost. Anything that
//   must not be lost goes to an outbox + broker (Q30).
// ======================================================================

var channel = Channel.CreateBounded<OrderEvent>(new BoundedChannelOptions(1_000)
{
    FullMode = BoundedChannelFullMode.Wait     // backpressure: the producer waits when it is full
});

// Producer (e.g. an endpoint)
await channel.Writer.WriteAsync(evt, ct);

// Consumers: 4 workers reading in parallel (e.g. inside a BackgroundService)
var workers = Enumerable.Range(0, 4).Select(_ => Task.Run(async () =>
{
    await foreach (var e in channel.Reader.ReadAllAsync(ct))
        await HandleAsync(e, ct);
}));
await Task.WhenAll(workers);

// ======================================================================
// Q: ConcurrentDictionary: is GetOrAdd really thread-safe?
//
// The dictionary itself won't get corrupted, but the valueFactory can run
// several times concurrently (only one result is kept). If the factory is
// expensive or has side effects (e.g. a token request), wrap it in Lazy, as
// shown in the code.
// - A failed Task stays cached. On error, call TryRemove, otherwise the error
//   "sticks" forever.
// - dict[key]++ on a ConcurrentDictionary is not atomic. Use AddOrUpdate(key, 1,
//   (_, v) => v + 1).
// - For a cache with expiration: HybridCache, which also has stampede protection
//   (Q42).
// ======================================================================

private readonly ConcurrentDictionary<string, Lazy<Task<ErpToken>>> _tokens = new();

public Task<ErpToken> GetTokenAsync(string tenant) =>
    _tokens.GetOrAdd(tenant, t => new Lazy<Task<ErpToken>>(() => FetchTokenAsync(t))).Value;
// Lazy (default: ExecutionAndPublication): only one FetchTokenAsync per tenant

// ======================================================================
// Q: How do lock and deadlocks work? Why can't you await inside a lock?
//
// - lock (obj) = Monitor.Enter / Exit inside try/finally. In .NET 9+ you write
//   private readonly Lock _lock = new();, which is faster.
// - await inside a lock: compile error. The lock belongs to a thread, and the
//   continuation may run on a different thread. For an async critical section:
//   SemaphoreSlim(1, 1) (Q89).
// - Don't lock on this, typeof(X) or a string: other code may lock on the same
//   object. Always lock on a private readonly field.
// - Deadlock: thread A holds L1 and waits for L2, while B holds L2 and waits for
//   L1. Prevention: a consistent lock order, small critical sections, no
//   external calls (HTTP, database, callbacks) inside a lock, timeouts
//   (Monitor.TryEnter, WaitAsync(timeout)).
// - Many readers, rare changes: instead of a lock, an immutable snapshot that is
//   replaced as a whole, as shown in the code.
// ======================================================================

// Immutable FrozenDictionary snapshot instead of a lock
private volatile FrozenDictionary<string, decimal> _prices = FrozenDictionary<string, decimal>.Empty;

// Reload: build a new dictionary and swap the reference (atomic). Readers never lock.
public void Reload(IEnumerable<PriceRow> rows) => _prices = rows.ToFrozenDictionary(r => r.Sku, r => r.Amount);
public decimal? Get(string sku) => _prices.TryGetValue(sku, out var p) ? p : null;