Viele Keys, wenige Round-Trips: Wie Zaris eine Bulk-Operation per Fan-out verteilt

Fragen Sie einen verteilten Store nach einem Key, wird der Aufwand von einer einzigen Zahl dominiert: dem Round-Trip zum Node, der ihn besitzt. Der Lookup selbst dauert Nanosekunden; das Netzwerk Mikrosekunden. Fragen Sie nun nach tausend Keys. Wenn Sie in einer Schleife tausendmal Get aufrufen, zahlen Sie diesen Round-Trip tausendmal, seriell, und der Store verbringt fast die gesamte Wanduhrzeit untätig mit dem Warten auf Pakete. Die Keys könnten auf vier Nodes verteilt sein, die parallel hätten antworten können — aber eine naive Schleife gibt ihnen nie die Chance.

Zaris hat drei Bulk-APIs — GetManyAsync, PutManyAsync, DeleteManyAsync — und darunter ein gemischtes ExecuteBatchAsync, und ihr ganzer Daseinszweck ist, „tausend Keys" in „eine Anfrage pro besitzendem Node, parallel abgesetzt" zu verwandeln. Dieser Beitrag ist der Fan-out-Pfad von Anfang bis Ende: wie der Client Keys nach Owner in Buckets sortiert, warum alle vier APIs auf eine einzige Engine zusammenfallen, wie ein Index-Feld die Antworten durch zwei Schichten von Nebenläufigkeit in Reihenfolge hält und warum — anders als bei einer Multi-Key-Transaktion — eine Bulk-Operation pro Item arbeitet und ein Teilerfolg normal ist.

Das Kostenmodell: Round-Trips, nicht Operationen

Wogegen eine Bulk-API kämpft, ist nicht die CPU. Ein GET auf einen einzelnen Key ist auf einem besitzenden Node ein Dictionary-Lookup — der Latenzboden ist eine Mikrosekunde echter Arbeit in zehn Mikrosekunden Installation. Der teure Teil ist die Installation: Die Anfrage muss das Netzwerk zum Owner überqueren, und die Antwort muss zurückkommen. Nennen wir diese Round-Trip-Zeit R, in einem Datacenter-Netzwerk typischerweise zehn bis hundert Mikrosekunden, um Größenordnungen mehr als der Lookup.

Eine serielle Schleife über N Keys kostet etwa N × R, weil jeder Aufruf auf die Rückkehr des Pakets des vorherigen wartet, bevor er den nächsten sendet. Das ist die Falle. Die Lösung ist kein schnellerer Lookup — es ist, aufzuhören, R pro Key zu zahlen. Wenn die N Keys auf M besitzenden Nodes liegen, ist der Boden R, M-mal nebenläufig gezahlt, also insgesamt etwa ein R, unabhängig davon, wie viele Keys jeder Node hält. Bei tausend Keys auf vier Nodes ist das der Unterschied zwischen tausend seriellen Round-Trips und effektiv einem. Das gesamte Design unten existiert, um N × R auf ~R zusammenzufalten.

Eine Engine: alles ist ein Batch von Items

Das erste Überraschende im Cluster-Client ist, dass GetManyAsync, PutManyAsync und DeleteManyAsync nicht drei getrennte Implementierungen sind. Jede ist ein dünner Shim, der ihre Argumente in eine Liste von KvBatchItems verwandelt und an die eine Methode übergibt, die die eigentliche Arbeit erledigt, ExecuteBatchAsync. So macht GetManyAsync genau das:

var list = keys.Select((k, i) => new KvBatchItem
{
    Index = i,                 // caller's position — the correlation key
    Op    = KvBatchOp.Get,
    Key   = k,
    GetOptions = options?.ToGetOptions()
}).ToList();

var resp = await ExecuteBatchAsync<T>(
    new KvBatchRequest
    {
        Items = list,
        DefaultConsistency = options?.Consistency ?? Defaults.DefaultConsistency
    }, ct).ConfigureAwait(false);

PutManyAsync und DeleteManyAsync tun dasselbe mit KvBatchOp.Put und KvBatchOp.Delete. Die Unterscheidung, die die Samples zwischen Bulk (jeweils eine Operationsart) und Batch (eine Mischung aus PUT / GET / DELETE in einem Aufruf) ziehen, ist also eine Annehmlichkeit der Oberfläche: Die Bulk-Helfer sind nur ein Batch, in dem zufällig jedes Item dieselbe Operation hat. Es gibt genau eine Fan-out-Engine, und sie ist operationsagnostisch.

Beachten Sie Index = i. Jedes Item wird mit seiner Position in der Liste des Aufrufers gestempelt, bevor irgendetwas gruppiert, umsortiert oder gesendet wird. Dieser Stempel überlebt die zwei nebenläufigen Hops, die noch kommen, und erlaubt es, die endgültige Antwort wieder in Reihenfolge zu bringen. Merken Sie sich das.

Der Fan-out: nach Owner in Buckets sortieren, je eine Anfrage

Hier ist das Herz des ganzen Beitrags — der Batch-Dispatch des Cluster-Clients, im Wesentlichen vollständig:

var table   = _routing;
var buckets = new Dictionary<ZarisRemoteClient, List<KvBatchItem>>();

foreach (var item in batch.Items)
{
    var seg    = SegmentMapper.ComputeSegmentId(item.Key, table.TotalSegments);
    var client = GetClientForSegment(seg, table);

    if (!buckets.TryGetValue(client, out var list))
        buckets[client] = list = new List<KvBatchItem>();

    list.Add(item);
}

var tasks = buckets.Select(kv =>
    kv.Key.ExecuteBatchAsync<T>(new KvBatchRequest
    {
        Items = kv.Value,
        DefaultConsistency = batch.DefaultConsistency
    }, ct));

var perOwner = await Task.WhenAll(tasks).ConfigureAwait(false);

return new KvBatchResponse
{
    Results = perOwner.SelectMany(r => r.Results)
                      .OrderBy(r => r.Index)
                      .ToList()
};

Lesen Sie es als vier Züge:

1. Jeden Key lokal zu seinem Owner auflösen. SegmentMapper.ComputeSegmentId ist dieselbe Platzierungsfunktion, die der Single-Key-Pfad nutzt — CRC16 über den Hash-Tag, modulo Segmentanzahl —, und GetClientForSegment bildet dieses Segment auf die bereits geöffnete Verbindung zu seinem besitzenden Node ab. Kein Proxy, kein Koordinator: Der Client berechnet das Ownership selbst, genau wie in Ein Hop zum Owner beschrieben. Der Unterschied ist, dass er das hier für jeden Key der Menge auf einmal tut.

2. Bucketing. Keys, die demselben Node gehören, häufen sich in dessen Liste. Tausend Keys, verteilt auf vier Owner, werden zu vier Listen. Hier werden aus N Operationen M Anfragen.

3. Nebenläufig absetzen. buckets.Select(...) baut pro Owner einen ExecuteBatchAsync-Aufruf, und Task.WhenAll lässt alle gleichzeitig starten. Vier Nodes antworten parallel; die Wanduhrzeit ist die Antwort des langsamsten einzelnen Nodes, nicht die Summe.

4. In Reihenfolge neu zusammensetzen. Jeder Owner liefert seinen Anteil der Ergebnisse; SelectMany flacht sie ab, und OrderBy(r => r.Index) stellt die ursprüngliche Reihenfolge des Aufrufers wieder her. Der Aufrufer erhält eine Liste, in der Position i die Antwort für Key i ist, obwohl die Keys zu verschiedenen Nodes gereist und in der Reihenfolge zurückgekehrt sind, in der das Netzwerk sie lieferte.

Das ist die vollständige Amortisationsbehauptung, konkret gemacht: Die Anzahl der Round-Trips entspricht der Anzahl der verschiedenen besitzenden Nodes in der Key-Menge, nicht der Anzahl der Keys.

Innerhalb eines Owners: wieder nebenläufig, wieder geordnet

Jeder Sub-Batch pro Owner landet im BatchCommandHandler dieses Nodes. Auch er führt die Items nicht nacheinander aus — er führt sie alle gleichzeitig aus und sortiert dann neu:

var items = batchReq.Request.Items.Cast<KvBatchItemWire>().ToArray();

// Execute every item in the sub-batch concurrently.
var tasks   = items.Select(item => ExecuteItemAsync(item)).ToArray();
var results = await Task.WhenAll(tasks).ConfigureAwait(false);

return new ZarisResponse
{
    Payload = new BatchResponsePayload
    {
        Response = new KvBatchResponse
        {
            Results = results.OrderBy(r => r.Index).ToList()
        }
    }
};

ExecuteItemAsync ist ein kleiner Switch über die Operation, der jedes Item an den Store weiterleitet — GetRawAsync, PutRawAsync oder DeleteRawAsync —, sodass ein gemischter Batch aus Lese- und Schreibzugriffen Item für Item innerhalb derselben Anfrage verarbeitet wird. (Für die In-Process- und Single-Node-Router erscheint dieselbe Form als Parallel.ForEachAsync über die Keys, begrenzt durch MaxDegreeOfParallelism — standardmäßig eine Option MaxParallelism oder Environment.ProcessorCount —, weshalb Lesezugriffe mit Kernen skalieren: Ein dicker Sub-Batch hält jeden Kern beschäftigt statt nur einen.)

Der Index-Stempel verdient sich seinen Lohn also zweimal. Nebenläufigkeit bringt die Reihenfolge auf dem Node durcheinander, daher sortiert der Handler vor der Antwort nach Index. Fan-out bringt die Reihenfolge auf dem Client durcheinander, daher sortiert der Client nach dem Zusammenführen nach Index. Der Aufrufer sieht nie eines der beiden Durcheinander. Eine kleine Ganzzahl, ab dem Moment, in dem die Anfrage gebaut wird, an jedem Item mitgeführt, ist der gesamte Mechanismus, der es sicher macht, „wirf alles parallel auf den Cluster" als geordnete Liste bereitzustellen.

Eine Bulk-Operation ist keine Transaktion

Das ist die Unterscheidung, die die Leute am häufigsten falsch verstehen, daher soll sie scharf formuliert werden. Jedes Item eines Batches trägt sein eigenes Ergebnis. Der Ergebnistyp pro Item hält Erfolg, Status und — für die Operationen, die sie haben — Revision, Version und TTL dieses einen Keys fest:

private static KvBatchItemResult MapKvToBatch(int index, KvBatchOp op, KvResult r) => new()
{
    Index   = index,
    Op      = op,
    Success = r.IsSuccess,
    Status  = r.Status,          // e.g. Success, NotFound, Moved
    Revision = r.Revision?.Value,
    ItemVersion = r.Version?.Version,
    Ttl     = r.TimeToLive
};

Es gibt keinen gemeinsamen Commit-Punkt. Ein GetMany über hundert Keys, von denen dreißig nicht existieren, liefert siebzig Success-Ergebnisse und dreißig NotFound-Ergebnisse, nach Index verzahnt — keinen Fehler, keine leere Liste. Ein PutMany, bei dem das Segment eines Keys gerade migriert, liefert dieses eine Item mit Status Moved und den Rest mit Success. Teilerfolg ist die normale, erwartete Form der Antwort, und der Aufrufer soll jedes Item prüfen.

Das ist das genaue Gegenteil einer Multi-Key-Transaktion, die per Konstruktion alles-oder-nichts ist: ein clientkoordinierter Two-Phase-Commit über optimistische Versionen, bei dem entweder jeder Key sich bewegt oder keiner. Die beiden Features sehen von außen ähnlich aus — beide nehmen viele Keys in einem Aufruf — und sind für entgegengesetzte Bedürfnisse gebaut. Greifen Sie zu einem Batch, wenn die Keys unabhängig sind und Sie Durchsatz wollen; greifen Sie zu einer Transaktion, wenn eine Invariante die Keys umspannt und ein halb angewendetes Ergebnis sie verletzen würde. Zaris hält sie als getrennte APIs, genau damit die Semantik nie mehrdeutig ist: Ein Batch rollt nie stillschweigend zurück, und eine Transaktion committet nie stillschweigend halb.

Überleben, wenn das Ownership mitten im Batch wandert

Eine Bulk-Operation ist ein breiteres Ziel als ein einzelnes GET für das Problem, dass sich das Ownership darunter ändert. Zwischen dem Moment, in dem der Client einen Key in den Bucket von Node A sortiert, und dem Moment, in dem Node A den Sub-Batch verarbeitet, kann ein Partition-Cut-over dieses Segment zu Node B verschoben haben. Node A beantwortet diese Items dann mit dem Status Moved, statt zu raten.

Der Cluster-Client umhüllt den gesamten Fan-out für genau diesen Fall mit einem begrenzten Retry:

public override async Task<KvBatchResponse> ExecuteBatchAsync<T>(
    KvBatchRequest batch, CancellationToken ct = default)
{
    return await RetryHelper.RetryAsync(
        () => ExecuteBatchClusterAsync<T>(batch, ct),
        maxAttempts: 2,
        backoffMillis: 100,
        logger: _logger).ConfigureAwait(false);
}

Bei einem Retry hat sich die Routing-Tabelle meist aktualisiert, sodass das erneute Bucketing die verschobenen Keys an ihren neuen Owner sendet. Der Retry ist bewusst flach — zwei Versuche, ein kurzes Backoff —, weil das Ziel ist, eine kurzzeitig veraltete Routing-Tabelle zu überbrücken, nicht einen tatsächlich ausgefallenen Node zu übertünchen. Das ist dieselbe Disziplin wie in Retries bei Ownership-Churn beschrieben, angewendet auf eine ganze Key-Menge auf einmal: Der Retry führt die Gruppierung erneut aus, nicht nur das Senden, sodass jeder Versuch das neueste Ownership widerspiegelt statt einen veralteten Plan zu wiederholen.

Ein echter Bug, weil die Wire-Form zwei Gestalten hat

Es lohnt sich, eine Stelle zu zeigen, an der dieser Pfad brach, weil sie etwas speziell über Bulk-Operationen aussagt. Der Wert eines GET-Ergebnisses kann auf der Leitung in zwei Formen eintreffen — in einen ZarisEntry-Umschlag gewickelt oder als roher serialisierter Wert —, je nachdem, wie er gespeichert wurde (siehe Was über die Leitung geht). Der ursprüngliche Bulk-GET-Decoder nahm nur die Wrapper-Form an und warf bei der rohen eine Exception.

Bei einem einzelnen GET ist das ein Fehler für einen Key. Bei einem GetMany war es auf subtile Weise katastrophal: Jeder Key im Batch schlug fehl, weil die eine unbehandelte Kodierung eine Exception warf, bevor irgendein Ergebnis dekodiert wurde. Ein Zertifizierungslauf markierte es als „Bulk-Get-Fehler", und die Lösung war, die Dekodierung über einen gemeinsamen Helfer mit Doppel-Kodierung zu leiten, der beide Formen behandelt:

// Decode through the shared helper (handles both the ZarisEntry-wrapper and
// the raw-value forms the server can emit). The old inline code assumed only
// the wrapper and threw, failing every key of a bulk GET.
var obj = KvBatchItemResultExtensions.GetValue<T>(r);

Die Lehre, die sich verallgemeinern lässt: In einem Bulk-Pfad wird ein Defekt in der gemeinsamen Maschinerie pro Item mit der Batch-Größe multipliziert, und ein Fehler, der für einen Key lästig ist, ist für tausend ein Ausfall. Das Ergebnismodell pro Item begrenzt den Schadensradius — sobald die Dekodierung korrekt ist, scheitert ein fehlerhafter Wert an einem Item und lässt die anderen 999 intakt.

Was der Fan-out bringt

Fügt man die Teile zusammen, ist die Form des Gewinns klar. Ein Bulk-Aufruf stempelt jedes Item mit seinem Index, löst den Owner jedes Keys mit derselben Platzierungsmathematik auf, die der Single-Key-Pfad nutzt, sortiert die Keys in Buckets, sodass aus N Operationen eine Anfrage pro besitzendem Node wird, setzt diese Anfragen nebenläufig ab und setzt die Antworten in der Reihenfolge des Aufrufers wieder zusammen — überlebt eine Routing-Tabelle, die sich mitten im Flug verschiebt, und meldet das Schicksal jedes Keys unabhängig. Die Zahl der Round-Trips sinkt von der Zahl der Keys auf die Zahl der Nodes, und das ist die einzige Zahl, auf die es je ankommen konnte.

Nichts davon erforderte einen neuen Speicherpfad oder einen neuen Wire-Record. Die Bulk-APIs sind eine Scheduling-Schicht über den Primitiven, die Zaris bereits hat: die partitionsbewusste Platzierung, die dem Client sagt, wer was besitzt, den Vertrag pro Record, den jedes Item-Ergebnis wiederverwendet, und die kernparallele Ausführung, die einen Sub-Batch durchkaut, sobald er gelandet ist. Der schwere Teil des schnellen Umgangs mit vielen Keys waren nie die Keys — es war die Weigerung, das Netzwerk einen Key nach dem anderen zu bezahlen.