Threading & Multitasking In Depth

// CONCURRENCY — multiple tasks in progress (may share one CPU core)
//   Web server handling 1 000 requests: most are awaiting I/O
//   async/await releases the thread while waiting for SQL/HTTP

// PARALLELISM — multiple tasks running SIMULTANEOUSLY on multiple cores
//   Image resize of 500 files: use all CPU cores
//   Parallel.ForEachAsync, PLINQ, or Task.Run for CPU-bound work

// I/O-bound  → async/await (don't block threads)
// CPU-bound  → Parallel.* or Task.Run to ThreadPool (use all cores)
// Mixed      → async wrapper around CPU work via Task.Run

// Bottom → top (prefer higher layers when they fit)

Thread          // OS thread — expensive; avoid creating manually
ThreadPool      // Reused worker threads — .NET schedules work here
Task / ValueTask // Promise of future result — composable, awaitable
async/await     // Compiler sugar over Tasks — non-blocking I/O
Parallel.*      // Data parallelism — fan-out CPU work with limits
Channels        // Producer/consumer queues — structured multitasking
IAsyncEnumerable // Async streams — yield items without buffering all

// Rare in modern .NET — prefer Task.Run unless you need thread affinity
var thread = new Thread(() =>
{
    Console.WriteLine($"Worker on thread {Environment.CurrentManagedThreadId}");
    // Long-running dedicated work (e.g. legacy COM apartment)
})
{
    IsBackground = true,   // dies when main app exits
    Name         = "MyWorker"
};
thread.Start();
thread.Join();             // block until finished

// Thread-local storage — data private to one thread
var slot = new ThreadLocal<int>(() => Environment.CurrentManagedThreadId);
Parallel.For(0, 4, _ => Console.WriteLine(slot.Value));

// Queue work to the pool — returns immediately
ThreadPool.QueueUserWorkItem(_ =>
{
    Console.WriteLine($"Pool thread {Thread.CurrentThread.ManagedThreadId}");
});

// Inspect pool health (diagnostics)
ThreadPool.GetAvailableThreads(out int workerAvail, out int ioAvail);
ThreadPool.GetMaxThreads(out int workerMax, out int ioMax);
Console.WriteLine($"Workers: {workerMax - workerAvail}/{workerMax} busy");

// ThreadPool starvation — classic production bug:
// All pool threads blocked on .Result / .Wait() / lock contention
// New requests queue up → app hangs → timeouts cascade
// Fix: async all the way, never block pool threads

// Task.Run — offload CPU-bound work to ThreadPool
var result = await Task.Run(() => HeavyComputation(data));

// Task.FromResult — already-completed task (no thread spawned)
Task<int> cached = Task.FromResult(42);

// TaskCompletionSource — manual control over task completion
var tcs = new TaskCompletionSource<string>();
_ = Task.Run(async () =>
{
    await Task.Delay(1000);
    tcs.SetResult("done");   // completes the Task returned by tcs.Task
});
string value = await tcs.Task;

// ValueTask<T> — zero allocation when result is often synchronous
public ValueTask<User?> GetCachedAsync(int id) =>
    _cache.TryGetValue(id, out var u)
        ? ValueTask.FromResult<User?>(u)
        : new ValueTask<User?>(FetchFromDbAsync(id));

// Exceptions are stored on the Task — surfaced at await
try
{
    await FailingOperationAsync();
}
catch (InvalidOperationException ex)
{
    // Handle — Task.Exception is AggregateException internally
}

// Observe without await (event handlers only — prefer async Task)
task.ContinueWith(t =>
{
    if (t.IsFaulted) log.LogError(t.Exception, "Background failed");
}, TaskScheduler.Default);

// Task.WhenAll propagates first exception as AggregateException
try
{
    await Task.WhenAll(task1, task2, task3);
}
catch (Exception ex)
{
    // Only the FIRST fault is thrown — inspect all via:
    // Task.WhenAll(...).ContinueWith(...) or try/catch per task
}

// You write:
public async Task<string> FetchAsync(string url)
{
    var response = await http.GetAsync(url);
    return await response.Content.ReadAsStringAsync();
}

// Compiler roughly produces:
// 1. IAsyncStateMachine with states: Start, AfterGetAsync, AfterReadAsync, Done
// 2. At each await: register continuation, return incomplete Task
// 3. When I/O completes: thread pool runs continuation, advances state
// 4. No thread is blocked during the HTTP wait — that's the win

// Golden rules:
// ✅ Return Task / Task<T> (never void except event handlers)
// ✅ Accept and forward CancellationToken
// ✅ await all the way — never .Result / .Wait()
// ✅ async all the way in ASP.NET Core request pipeline

// By default, await tries to resume on the "captured context"
// UI apps (WPF/WinForms): must resume on UI thread → leave default
// ASP.NET Core: NO SynchronizationContext → ConfigureAwait irrelevant

// Library code (class libraries, NuGet packages):
await SomeIoAsync().ConfigureAwait(false);
// "I don't care which thread resumes — slightly more efficient"

// ❌ Deadlock pattern (WinForms / old ASP.NET — NOT Core):
button.Click += async (_, _) =>
{
    var data = GetDataAsync().Result;  // blocks UI thread
    // GetDataAsync tries to resume on UI thread → deadlock
};

// ✅ Always await in UI/event code:
button.Click += async (_, _) => label.Text = await GetDataAsync();

// I/O-bound — async without Task.Run (network, disk, SQL)
public async Task<Order?> GetOrderAsync(int id, CancellationToken ct)
    => await db.Orders.FindAsync([id], ct);

// CPU-bound inside async — offload to avoid blocking caller's thread
public async Task<byte[]> ResizeImageAsync(Stream input, CancellationToken ct)
{
    using var ms = new MemoryStream();
    await input.CopyToAsync(ms, ct);
    return await Task.Run(() => ImageSharpResize(ms.ToArray()), ct);
}

// ❌ async without await — compiler warning; use Task.FromResult
public Task<int> GetConstantAsync() => Task.FromResult(42);

// ❌ Task.Run for I/O — wastes a pool thread for no benefit
// await Task.Run(() => db.Orders.ToList());  // WRONG — blocks pool thread on SQL

// Fan-out independent I/O — total time = slowest task, not sum
public async Task<DashboardDto> GetDashboardAsync(int userId, CancellationToken ct)
{
    var statsTask  = _stats.GetAsync(userId, ct);
    var ordersTask = _orders.GetRecentAsync(userId, 5, ct);
    var notifTask  = _notif.GetUnreadAsync(userId, ct);

    await Task.WhenAll(statsTask, ordersTask, notifTask);

    return new DashboardDto(
        statsTask.Result,      // safe — already completed
        ordersTask.Result,
        notifTask.Result);
}

// First responder wins — fallback / race pattern
var winner = await Task.WhenAny(
    FetchPrimaryAsync(ct),
    FetchReplicaAsync(ct));
var data = await winner;

// Throttle concurrency — don't fire 10 000 tasks at once
var semaphore = new SemaphoreSlim(8);
var tasks = urls.Select(async url =>
{
    await semaphore.WaitAsync(ct);
    try { return await FetchAsync(url, ct); }
    finally { semaphore.Release(); }
});
var results = await Task.WhenAll(tasks);

// Parallel.ForEachAsync (.NET 6+) — bounded CPU + async body
await Parallel.ForEachAsync(
    files,
    new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount, CancellationToken = ct },
    async (file, token) =>
    {
        await ProcessFileAsync(file, token);
    });

// Parallel.For — synchronous CPU-only (no await in body)
Parallel.For(0, items.Length, new ParallelOptions { MaxDegreeOfParallelism = 4 },
    i => items[i] = Transform(items[i]));

// PLINQ — parallel LINQ for in-memory data
var primes = Enumerable.Range(2, 1_000_000)
    .AsParallel()
    .WithDegreeOfParallelism(4)
    .Where(IsPrime)
    .ToList();

private readonly object _gate = new();
private int _counter;

public void Increment()
{
    lock (_gate) { _counter++; }   // mutual exclusion
}

// SemaphoreSlim — limit concurrent access (async-friendly)
private readonly SemaphoreSlim _dbLimiter = new(10); // max 10 at once

public async Task<T> QueryAsync<T>(Func<Task<T>> query, CancellationToken ct)
{
    await _dbLimiter.WaitAsync(ct);
    try { return await query(); }
    finally { _dbLimiter.Release(); }
}

// ReaderWriterLockSlim — many readers OR one writer
private readonly ReaderWriterLockSlim _rw = new();

public T Read<T>(Func<T> read)  { _rw.EnterReadLock();  try { return read(); }  finally { _rw.ExitReadLock(); } }
public void Write(Action write) { _rw.EnterWriteLock(); try { write(); } finally { _rw.ExitWriteLock(); } }

// Interlocked — atomic operations without lock
int count = 0;
Interlocked.Increment(ref count);
Interlocked.CompareExchange(ref _state, newState, expectedState);

// volatile — ensure reads/writes aren't cached in CPU register
private volatile bool _shutdownRequested;

public void RequestShutdown() => _shutdownRequested = true;

public void WorkerLoop()
{
    while (!_shutdownRequested)
        ProcessNextItem();
}

// Prefer Interlocked over lock for simple counters/flags
// Still not a replacement for complex shared state

using System.Collections.Concurrent;

var cache = new ConcurrentDictionary<string, User>();

// GetOrAdd — atomic "get or create" (factory may run twice — rare)
var user = cache.GetOrAdd(email, e => LoadUserFromDb(e));

// AddOrUpdate — atomic upsert
cache.AddOrUpdate(key, addValue, (_, old) => UpdateUser(old));

// TryRemove / TryGetValue — safe under concurrent access
if (cache.TryGetValue(id, out var cached))
    return cached;

// Don't wrap Dictionary in lock {} when ConcurrentDictionary exists

// ConcurrentBag — unordered, thread-safe (great for Parallel.For results)
var results = new ConcurrentBag<int>();
Parallel.For(0, 100, i => results.Add(Compute(i)));

// ConcurrentQueue — FIFO work stealing between threads
var queue = new ConcurrentQueue<WorkItem>();
queue.Enqueue(new WorkItem(1));
if (queue.TryDequeue(out var item)) Process(item);

// BlockingCollection — backpressure + bounding (legacy but still used)
var bounded = new BlockingCollection<Job>(capacity: 100);
bounded.Add(job);                    // blocks when full
var job = bounded.Take();            // blocks when empty

// Modern alternative: Channel (section 8)

var channel = Channel.CreateUnbounded<WorkItem>();
var writer = channel.Writer;
var reader = channel.Reader;

// Producer
await writer.WriteAsync(new WorkItem("job-1"));
writer.Complete();   // signal no more items

// Consumer — async foreach
await foreach (var item in reader.ReadAllAsync(ct))
    await ProcessAsync(item, ct);

// Multiple consumers — each item processed by exactly one reader
// (competing consumers pattern, like Service Bus)

var channel = Channel.CreateBounded<OrderEvent>(new BoundedChannelOptions(500)
{
    FullMode = BoundedChannelFullMode.Wait   // backpressure when full
});

// Producer (API endpoint)
await channel.Writer.WriteAsync(new OrderEvent(orderId), ct);

// Worker pool — N consumers in parallel
var workers = Enumerable.Range(0, 4).Select(_ => Task.Run(async () =>
{
    await foreach (var evt in channel.Reader.ReadAllAsync(ct))
        await HandleOrderAsync(evt, ct);
}));
await Task.WhenAll(workers);

// Graceful shutdown:
channel.Writer.TryComplete();
await Task.WhenAll(workers);

// Register singleton channel
builder.Services.AddSingleton(Channel.CreateUnbounded<AuditEvent>());
builder.Services.AddHostedService<AuditConsumerService>();

public sealed class AuditConsumerService(
    Channel<AuditEvent> channel, IAuditWriter writer) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        await foreach (var evt in channel.Reader.ReadAllAsync(ct))
            await writer.WriteAsync(evt, ct);
    }
}

// Endpoint — non-blocking publish
app.MapPost("/orders", async (CreateOrderDto dto, Channel<AuditEvent> ch, ...) =>
{
    // ... save order ...
    await ch.Writer.WriteAsync(new AuditEvent("Created", order.Id), ct);
    return Results.Created(...);
});

// Yield items as they arrive — no List<T> buffering millions of rows
public async IAsyncEnumerable<OrderDto> StreamOrdersAsync(
    [EnumeratorCancellation] CancellationToken ct = default)
{
    await foreach (var order in db.Orders.AsAsyncEnumerable().WithCancellation(ct))
    {
        ct.ThrowIfCancellationRequested();
        yield return MapToDto(order);
    }
}

// Consumer
await foreach (var dto in StreamOrdersAsync(ct))
    await SendToClientAsync(dto, ct);

// Minimal API — .NET 7+ streams JSON as NDJSON/chunked
app.MapGet("/orders/stream", (AppDbContext db, CancellationToken ct) =>
    StreamOrdersAsync(db, ct));

// Always accept CancellationToken — ASP.NET Core injects it automatically
public async Task<List<Customer>> GetAllAsync(CancellationToken ct = default)
{
    return await db.Customers.ToListAsync(ct);
}

// Linked token — cancel when EITHER parent or timeout fires
using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
cts.CancelAfter(TimeSpan.FromSeconds(30));
await LongOperationAsync(cts.Token);

// PeriodicTimer + BackgroundService — modern scheduled loop
public sealed class CleanupService(
    IServiceScopeFactory scopes,
    ILogger<CleanupService> log) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        using var timer = new PeriodicTimer(TimeSpan.FromHours(1));
        while (await timer.WaitForNextTickAsync(ct))
        {
            using var scope = scopes.CreateScope();
            await scope.ServiceProvider.GetRequiredService<ICleaner>()
                .RunAsync(ct);
        }
    }
}

// ❌ lock + await = deadlock waiting to happen
lock (_gate)
{
    await SaveAsync();   // releases lock? NO — holds lock across await
}

// ✅ release lock before await, or use SemaphoreSlim
await _semaphore.WaitAsync(ct);
try { await SaveAsync(ct); }
finally { _semaphore.Release(); }

// Waiting for HTTP / SQL / file I/O?
//   → async/await with native *Async APIs

// Multiple independent I/O calls?
//   → Task.WhenAll

// CPU-bound work (image processing, crypto, parsing)?
//   → Task.Run or Parallel.For / Parallel.ForEachAsync

// Producer feeds consumer(s) in-process?
//   → Channel or BlockingCollection

// Stream large result sets without buffering?
//   → IAsyncEnumerable

// Cross-service async messaging?
//   → MassTransit / Azure Service Bus (not Channels)

// Scheduled / persistent background jobs?
//   → Hangfire / Quartz / Azure Functions timer