Wprowadzenie

W poprzednim artykule o Jitter poznaliśmy, jak możemy uniknąć nagłego skoku obciążenia, gdy wiele procesów startuje i pobiera te same dane w jednym momencie.

Tym razem rozważymy, co zrobić, gdy zdarzenia z natury napływają nierównomiernie, a mimo to aplikacja musi przetwarzać je w stałym, przewidywalnym tempie.

Pomyślmy o systemie opartym na zdarzeniach, który generuje korespondencję w aplikacji finansowej. Inne aplikacje przekazują do niego komendy tworzenia i wysłania listów PDF do wskazanych odbiorców. Wszystko działa sprawnie, dopóki nie nadejdzie termin korespondencji okresowej. Firma jest zobowiązana dostarczyć ją klientom w konkretnym dniu. Wtedy w systemie, poza codziennymi zapytaniami, pojawiają się dodatkowe polecenia generowania listów, załóżmy dla 50 000 użytkowników naraz. Umówmy się, że nasz generator PDF-ów ma ograniczoną przepustowość. Potrafi przetworzyć maksymalnie 100 dokumentów na minutę. Gdy przekroczymy limit, zapytania zaczną być odrzucane, a w systemie pojawią się błędy oraz automatyczne ponowienia, które tylko pogłębią chaos. Potrzebujemy mechanizmu, który przyjmie wszystkie polecenia i odda je w tempie, które faktycznie jesteśmy w stanie obsłużyć.

Idea leaky bucket

Żeby obsłużyć gwałtowne obciążenie i nie przekroczyć limitu przepustowości naszego systemu, skorzystamy z algorytmu leaky bucket. Jego nazwa pochodzi od prostego obrazu – wiadra z dziurą w dnie. Gdy wlewamy do niego wodę, niezależnie od tempa, z dziury wypływa ona tym samym, stabilnym strumieniem.

W naszym przypadku „wodą" będą polecenia generowania korespondencji. Nie ma znaczenia, czy do systemu trafi ich 10, czy 50 000 w jednym momencie – mechanizm generowania PDF-ów i tak przetworzy je w tempie maksymalnie 100 na minutę.

Wiadro ma jednak swoją pojemność. Jeśli wlewamy wodę szybciej, niż zdąża ona wyciekać, w pewnym momencie pojemnik się przepełni, a nadmiar po prostu się rozleje. To istotne ograniczenie, które zagwarantuje, że przy trwale zbyt dużym napływie nasz system nie zostanie przeciążony.

Leaky bucket zakłada więc dwie rzeczy do ustalenia: tempo przetwarzania oraz pojemność wiadra. W naszym przypadku pierwsze z nich jest proste do wyznaczenia i wynika wprost z limitu tempa mechanizmu generowania PDF-ów. Pojemność to już decyzja biznesowa, która może wynikać z SLA - ustalenia w jakim maksymalnie czasie powinien zostać wygenerowany list.

Leaky bucket w praktyce – kontrola tempa generowania dokumentów

Żeby zademonstrować działanie leaky bucket, zasymulujemy przypadek generowania korespondencji na przykładzie w .NET. Kod będzie dostępny w repozytorium serii.

Limit mechanizmu generowania PDF-ów

Na początek modelujemy wymaganie stałego czasu przetwarzania - generator PDF-ów, który akceptuje co najwyżej 10 zapytań na sekundę. Skala jest celowo mniejsza niż scenariusz 50 000 listów i 100 PDF-ów na minutę. Dzięki temu możemy sprawdzić działanie algorytmu w rozsądnym czasie. Gdy limit generowania zostanie przekroczony, pojawi się wyjątek - w prawdziwym systemie mógłby to być status code 429 - throttling z zewnętrznego API.

public void GeneratePDF(CorrespondenceDocument _, CancellationToken cancellationToken = default)
{
    generateSemaphore.Wait(cancellationToken);
    try
    {
        var now = DateTime.UtcNow;
        var windowStart = now.AddSeconds(-1);

        while (callTimestamps.Count > 0 && callTimestamps.Peek() < windowStart)
        {
            callTimestamps.Dequeue();
        }

        if (callTimestamps.Count >= maxPerSecond)
        {
            throw PDFGenerationThroughputExceededException.PerSecondExceeded(maxPerSecond);
        }

        callTimestamps.Enqueue(now);
    }
    finally
    {
        generateSemaphore.Release();
    }
}

W przykładzie ustawiamy maxPerSecond: 10. To wygodniejszy do demonstracji odpowiednik limitu „100 na minutę” z wprowadzenia.

Dwa źródła ruchu: codzienność i nagły skok obciążeń

Utwórzmy dwa źródła ruchu. Pierwsze - codzienne nazywane „Steady", w którym klasa Publisher będzie wysyłać żądanie wygenerowania pojedynczego dokumentu co 250 ms, imitując codzienne obciążenie. Drugie - „Burst" - odpowiada korespondencji okresowej i dostarcza 8 dokumentów co 2 sekundy, tworząc nagły skok obciążeń ponad limit.

var steadyPublisher = new Publisher<CorrespondenceDocument>(
    [correspondenceRef],
    tickDelay: TimeSpan.FromMilliseconds(250));

var burstPublisher = new Publisher<CorrespondenceDocument>(
    [correspondenceRef],
    tickDelay: TimeSpan.FromSeconds(2),
    batchSize: 8);

Generowanie PDF-ów bez leaky bucket

Jeśli oba źródła ruchu podłączymy wprost do GeneratePDF, nastąpi chwilowe przekroczenie przepustowości. W naszym wypadku prowadzi to do błędu PDFGenerationThroughputExceededException i kończy działanie programu. W scenariuszu produkcyjnym nieudane próby prawdopodobnie skutkowałyby powtarzaniem, które dodatkowo pogorszyłoby sytuację.

steadyPublisher.MessageStream.Subscribe(doc =>
{
    pdfGenerator.GeneratePDF(doc, cts.Token);
});

burstPublisher.MessageStream.Subscribe(doc =>
{
    pdfGenerator.GeneratePDF(doc, cts.Token);
});

var publishing = await Task.WhenAny(
    steadyPublisher.StartPublishingAsync(cts.Token),
    burstPublisher.StartPublishingAsync(cts.Token));

cts.Cancel();
await publishing; // PDFGenerationThroughputExceededException

Leaky bucket do kontrolowania tempa przepływu

By ograniczyć przepustowość tak, żeby nie przekraczała 10 zapytań na sekundę, wprowadzimy leaky bucket:

public void AddToBucket(CorrespondenceDocument document)
{
    if (completed)
    {
        throw new InvalidOperationException("Cannot add to a completed leaky bucket.");
    }

    var newCount = queue.Count + 1;
    if (newCount > capacity)
    {
        throw BucketOverflowException.CapacityExceeded(newCount, capacity);
    }

    queue.Enqueue(document);
    signal.Release();
}

public async Task StartLeakingAsync()
{
    while (true)
    {
        if (queue.TryDequeue(out var document))
        {
            subject.OnNext(document);
            await Task.Delay(leakInterval);
        }
        else if (completed)
        {
            subject.OnCompleted();
            return;
        }
        else
        {
            await signal.WaitAsync();
        }
    }
}

AddToBucket odpowiada wlewaniu wody do wiadra – dodaje dokument do kolejki, o ile starczy w niej miejsca. Kontrolę przepływu realizuje StartLeakingAsync, czyli nasza „dziura w dnie": co leakInterval (u nas 100 ms, czyli maksymalnie 10 dokumentów na sekundę) wypuszcza jeden dokument do strumienia LeakyStream.

Teraz oba źródła ruchu podłączamy do wiadra, a generator PDF-ów – już tylko do strumienia wycieku:

var bucket = new CorrespondenceLeakyBucket(leakInterval: TimeSpan.FromMilliseconds(100), capacity: 100);

steadyPublisher.MessageStream.Subscribe(bucket.AddToBucket);
burstPublisher.MessageStream.Subscribe(bucket.AddToBucket);
bucket.LeakyStream.Subscribe(doc => pdfGenerator.GeneratePDF(doc));

var leaking = bucket.StartLeakingAsync();

await Task.WhenAll(
    steadyPublisher.StartPublishingAsync(cts.Token),
    burstPublisher.StartPublishingAsync(cts.Token));

bucket.Complete();
await leaking; // bez wyjątku – tempo trzyma się limitu generatora

Burst nadal pojawia się w systemie gwałtownie, ale trafia do kolejki bucketa. Do PDF-ów dociera już równym strumieniem dopasowanym do limitu.

Podsumowanie

Leaky bucket sprowadza się do jednej zasady: przyjmij dowolny napływ zdarzeń, który mieści się w pojemności wiadra, i oddawaj go dalej w stałym, przewidywalnym tempie – niezależnie od tego, jak nierówno trafiają na wejście. W naszym przykładzie burst nałożony na stały ruch nie zachwiał systemem – trafiał do generatora PDF-ów w rytmie, jaki ten był w stanie faktycznie obsłużyć.

Ten artykuł jest częścią serii. Pełny kod przykładu znajdziesz w repozytorium: useful-async-algorithms.