벌크헤드 패턴

using System;
using System.Threading;
using System.Threading.Tasks;

var shared = new Bulkhead("shared", 2);
await Scenario("one shared bulkhead", payments: shared, catalog: shared);
await Scenario("separate bulkheads", payments: new Bulkhead("payments", 2), catalog: new Bulkhead("catalog", 2));

static async Task Scenario(string title, Bulkhead payments, Bulkhead catalog)
{
    Console.WriteLine($"== {title} ==");
    // 결제 서비스가 멈춘 상황: 게이트를 열기 전까지 결제 호출은 끝나지 않는다.
    var paymentGate = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
    var pay1 = payments.RunAsync(() => paymentGate.Task);
    var pay2 = payments.RunAsync(() => paymentGate.Task);
    Console.WriteLine("payment 3: " + await payments.RunAsync(() => paymentGate.Task));
    Console.WriteLine("catalog: " + await catalog.RunAsync(() => Task.FromResult("ok")));
    paymentGate.SetResult("ok");
    Console.WriteLine("payment 1: " + await pay1);
    Console.WriteLine("payment 2: " + await pay2);
}

sealed class Bulkhead(string name, int slots)
{
    private readonly SemaphoreSlim semaphore = new(slots, slots);

    public async Task<string> RunAsync(Func<Task<string>> call)
    {
        // 빈 칸이 없으면 기다리지 않고 바로 거절한다. 칸을 얻지 못했으니 반납할 것도 없다.
        if (!semaphore.Wait(0)) return "rejected by " + name;
        try
        {
            return await call();
        }
        finally
        {
            semaphore.Release();
        }
    }
}

== one shared bulkhead ==
payment 3: rejected by shared
catalog: rejected by shared
payment 1: ok
payment 2: ok
== separate bulkheads ==
payment 3: rejected by payments
catalog: ok
payment 1: ok
payment 2: ok

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

var payments = new Bulkhead("payments", maxConcurrent: 2, maxQueue: 1);
var catalog = new Bulkhead("catalog", maxConcurrent: 2, maxQueue: 0);

// 가짜 결제 게이트웨이: 호출마다 게이트가 하나 있고, 프로그램이 직접 열거나 실패시킨다.
var gates = new Dictionary<string, TaskCompletionSource<string>>();
foreach (var id in new[] { "pay-1", "pay-2", "pay-3", "pay-4", "pay-5" })
    gates[id] = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
Task<string> Pay(string id) => gates[id].Task;

var pay1 = payments.RunAsync(() => Pay("pay-1"));
var pay2 = payments.RunAsync(() => Pay("pay-2"));
var pay3 = payments.RunAsync(() => Pay("pay-3")); // 칸이 없어 대기열에서 기다린다.
var pay4 = payments.RunAsync(() => Pay("pay-4")); // 대기열도 차서 바로 거절된다.
Console.WriteLine("after 4 payment calls: " + payments.Describe());
Console.WriteLine("pay-4: " + Describe(await pay4));
Console.WriteLine("catalog-1: " + Describe(await catalog.RunAsync(() => Task.FromResult("item list"))));

gates["pay-2"].SetException(new InvalidOperationException("gateway reset"));
Console.WriteLine("pay-2: " + Describe(await pay2));
gates["pay-3"].SetResult("charged pay-3"); // pay-2가 반납한 칸으로 실행된다.
Console.WriteLine("pay-3: " + Describe(await pay3));
gates["pay-1"].SetResult("charged pay-1");
Console.WriteLine("pay-1: " + Describe(await pay1));
Console.WriteLine("after all calls: " + payments.Describe());

gates["pay-5"].SetResult("charged pay-5");
Console.WriteLine("pay-5: " + Describe(await payments.RunAsync(() => Pay("pay-5"))));

static string Describe(BulkheadResult<string> result) => result switch
{
    BulkheadResult<string>.Ok ok => "ok: " + ok.Value,
    BulkheadResult<string>.Rejected r => $"rejected: {r.Bulkhead} full (running {r.Running}, queued {r.Queued})",
    BulkheadResult<string>.Failed f => "failed: " + f.Error.Message,
    _ => "unknown result",
};

abstract record BulkheadResult<T>
{
    public sealed record Ok(T Value) : BulkheadResult<T>;
    public sealed record Rejected(string Bulkhead, int Running, int Queued) : BulkheadResult<T>;
    public sealed record Failed(Exception Error) : BulkheadResult<T>;
}

sealed class Bulkhead(string name, int maxConcurrent, int maxQueue)
{
    private readonly SemaphoreSlim slots = new(maxConcurrent, maxConcurrent);
    private int admitted; // 실행 중 + 대기 중인 호출 수

    public int Running => maxConcurrent - slots.CurrentCount;
    public int Queued => Volatile.Read(ref admitted) - Running;
    public string Describe() => $"{name} running={Running} queued={Queued}";

    public async Task<BulkheadResult<T>> RunAsync<T>(Func<Task<T>> call)
    {
        // 실행 칸과 대기열이 모두 찼으면 기다리지 않고 거절 결과를 돌려준다.
        if (Interlocked.Increment(ref admitted) > maxConcurrent + maxQueue)
        {
            Interlocked.Decrement(ref admitted);
            return new BulkheadResult<T>.Rejected(name, Running, Queued);
        }
        var acquired = false;
        try
        {
            // 칸이 없으면 대기열에서 기다린다. 대기 중 예외가 나도 아래 finally가 계수를 되돌린다.
            await slots.WaitAsync();
            acquired = true;
            return new BulkheadResult<T>.Ok(await call());
        }
        catch (Exception ex)
        {
            // 의존성 경계: 예외를 실패 결과로 바꾼다. 형식과 스택을 잃지 않도록 예외를 그대로 담는다.
            return new BulkheadResult<T>.Failed(ex);
        }
        finally
        {
            // admitted는 항상 줄이고, 칸은 실제로 얻었을 때만 반납한다.
            Interlocked.Decrement(ref admitted);
            if (acquired) slots.Release();
        }
    }
}

after 4 payment calls: payments running=2 queued=1
pay-4: rejected: payments full (running 2, queued 1)
catalog-1: ok: item list
pay-2: failed: gateway reset
pay-3: ok: charged pay-3
pay-1: ok: charged pay-1
after all calls: payments running=0 queued=0
pay-5: ok: charged pay-5

type CallResult =
  | { ok: true; value: string }
  | { ok: false; error: "BULKHEAD_FULL"; bulkhead: string };

// 남은 칸 수만 세는 작은 세마포어. 기다리지 않고 바로 성공 여부를 알려 준다.
class Semaphore {
  #available: number;
  constructor(slots: number) { this.#available = slots; }
  tryAcquire(): boolean {
    if (this.#available === 0) return false;
    this.#available -= 1;
    return true;
  }
  release(): void { this.#available += 1; }
}

class Bulkhead {
  readonly #slots: Semaphore;
  constructor(readonly name: string, slots: number) { this.#slots = new Semaphore(slots); }
  async run(call: () => Promise<string>): Promise<CallResult> {
    // 칸을 얻지 못하면 거절한다. 얻지 못한 칸은 반납하지 않는다.
    if (!this.#slots.tryAcquire()) return { ok: false, error: "BULKHEAD_FULL", bulkhead: this.name };
    try {
      return { ok: true, value: await call() };
    } finally {
      this.#slots.release();
    }
  }
}

// 직접 열 때까지 끝나지 않는 게이트. 멈춘 결제 서비스를 흉내 낸다.
function gate(): { promise: Promise<string>; open: (value: string) => void } {
  let open: (value: string) => void = () => {};
  const promise = new Promise<string>((resolve) => { open = resolve; });
  return { promise, open };
}

function describe(result: CallResult): string {
  return result.ok ? result.value : "rejected by " + result.bulkhead;
}

async function scenario(title: string, payments: Bulkhead, catalog: Bulkhead): Promise<void> {
  console.log(`== ${title} ==`);
  const paymentGate = gate();
  const pay1 = payments.run(() => paymentGate.promise);
  const pay2 = payments.run(() => paymentGate.promise);
  console.log("payment 3: " + describe(await payments.run(() => paymentGate.promise)));
  console.log("catalog: " + describe(await catalog.run(() => Promise.resolve("ok"))));
  paymentGate.open("ok");
  console.log("payment 1: " + describe(await pay1));
  console.log("payment 2: " + describe(await pay2));
}

async function main(): Promise<void> {
  const shared = new Bulkhead("shared", 2);
  await scenario("one shared bulkhead", shared, shared);
  await scenario("separate bulkheads", new Bulkhead("payments", 2), new Bulkhead("catalog", 2));
}

void main();

== one shared bulkhead ==
payment 3: rejected by shared
catalog: rejected by shared
payment 1: ok
payment 2: ok
== separate bulkheads ==
payment 3: rejected by payments
catalog: ok
payment 1: ok
payment 2: ok