magmalake/dista

dista (die dista-mojo Tin, 0.2.0, Mojo 1.2.0) ermöglicht es, eine Mojo-Funktion als remote zu markieren, sie aufzurufen und auf anderen Prozessen oder Kubernetes-Pods auszuführen. Jeder Prozess führt dieselbe kompilierte Binärdatei aus, daher ist eine Task-Nachricht eine Funktions-ID plus codierte Argumente: Es wird kein Code versendet, ichts wird gepickled, und spaltendatenbasierte Daten reisen als Arrow.

Was dista nicht ist: ein Cluster-Manager. Es erstellt keine Pods oder Maschinen. Es nutzt jeden Worker, der beitritt, stoppt diejenigen, die es angewiesen wurde zu beenden, und gibt an, wie viele es haben möchte. Kubernetes, eine Shell-Schleife oder orca starten und stoppen sie. Es baut auch keine verlorenen Ergebnisse eines toten Workers wieder auf (keine Lineage-Rekonstruktion): Der Job schlägt fehl, nennt das Ergebnis, und ein Orchestrator führt ihn erneut aus.

Die installierte Umgebung enthält dista nur als kompilierte dista.mojoc. Lesen Sie den Quellcode auf GitHub, magmalake/dista.mojo, unter der rev, die Ihre pixi.toml festlegt, oder im Git-Cache von pixi (ls -d ~/Library/Caches/rattler/cache/git-v0/checkouts/*/<erste 7 der rev>* auf macOS, ~/.cache/rattler/... auf Linux). Die hier zitierten Dateipfade befinden sich in diesem Checkout.

pixi shelf add dista-mojo    # bringt arrow-mlake, memory-region und objectstore

Schreiben Sie das Mojo selbst mit geladenem mojo-syntax-Skill; für schnelle pro-Zeilen-Arbeit innerhalb von Tasks siehe mojo-data-performance.

Tasks deklarieren, dann aufrufen

Ein Task ist ein Top-Level def, das einmal mit seinen Argument- und Ergebnis- typen deklariert wird, da Mojo 1.2 den Ergebnistyp einer Funktion nicht ableiten kann. Das Register ist die Parameterliste des Runtime. Jedes vollständige Beispiel in dieser Datei wurde gegen dista 0.2.0 gebaut und ausgeführt.

from dista import ObjectRef, Runtime, Seq, Task, Task2

def square(x: Int) -> Int:
    return x * x

def total(xs: Seq[Int]) -> Int:
    var t = 0
    for i in range(len(xs)):
        t += xs[i]
    return t

def scale(x: Int, factor: Float64) -> Float64:
    return Float64(x) * factor

comptime Square = Task[Int, Int, square]
comptime Total = Task[Seq[Int], Int, total]
comptime Scale = Task2[Int, Float64, Float64, scale]

def main() raises:
    var rt = Runtime[Square, Total, Scale](workers=4)
    var r = rt.remote[Square](7)              # ein ObjectRef[Int], sofort
    print(rt.get(r))                          # 49: wartet auf das Ergebnis
    print(rt.get(rt.remote[Scale](5, 2.5)))   # 12.5
    var maps = List[ObjectRef[Int]]()
    for i in range(1000):
        maps.append(rt.remote[Square](i))
    print(rt.get(rt.remote[Total](maps)))     # 332833500
    print(rt.get(rt.remote[Square](r)))       # 2401: ein Ref als Argument
    rt.release(maps)

• Stelligkeit: Task[A, R, f] nimmt ein Argument, Task2[A, B, R, f] zwei, Task3[A, B, C, R, f] drei. Übergeben Sie mehr als eine Struktur. Der Aufruf einer Task2 mit einem Argument löst nimmt 2 Argumente beim Aufruf aus.
• Die Funktion kann werfen oder nicht. Ein Task, der wirft, löst seine Nachricht in rt.get erneut aus, und der Worker arbeitet weiter.
• Refs füttern Tasks, ohne dass der Treiber etwas abruft. Übergeben Sie einen ObjectRef[A], wo der Task ein A erwartet, oder eine List[ObjectRef[X]], wo er ein Seq[X] erwartet (in der Reihenfolge gesammelt). Der Task wird einmal ausgeführt, sobald jeder Input existiert, und schlägt mit dem Fehler des Inputs fehl, wenn einer fehlschlug.
• rt.release(ref) (oder eine Liste) gibt ein Ergebnis am Head und auf seinem Worker frei, sobald kein anstehender Task es mehr benötigt; ein get darauf wirft dann. Ergebnisse werden ansonsten bis zum Ende der Laufzeit aufbewahrt.
• Auch auf Runtime: rt.ready(r) (ob fertig, ohne zu warten), rt.location(r) (der haltende Worker ip:port und die Node-Store-Datei, oder "", wenn das Ergebnis inline zurückkam), rt.stats(), rt.bytes_fetched() und rt.timeline(path), das eine Chrome-Trace für Perfetto schreibt.

Shuffles: mehrere Ergebnisse oder ein Wert neben Refs

Ein TaskMany[A, X, f], wobei f ein Seq[X] zurückgibt, ist ein Map-Task: rt.remote_many[T](arg, n) gibt n Refs zurück, eine pro Partition, jeweils eigenständig gespeichert, abgerufen und freigegeben; f muss genau n Elemente zurückgeben. Ein TaskWith[A, X, R, f], wobei f ein A und ein Seq[X] nimmt, ist ein Reducer: rt.remote[T](j, refs). Ein Beispiel, das kompiliert, und die Details finden Sie in references/shuffles.md.

Was kann ein Argument oder Ergebnis sein

Der Codec wählt zur Kompilierzeit eine Kodierung:

• Int, IntN, UIntN, Float32, Float64, Bool und String;
• Seq[T], eine Liste, die durch den Codec geht (List tut dies nicht: Reflection kann ihren Elementtyp nicht wiederherstellen), auch als Strukturfeld;
• jede einfache Struktur aus den obigen, Feld für Feld durch Reflection (Felder müssen Movable sein; Argument- und Ergebnistypen benötigen Movable & Deinitable);
• ein Typ, der das Codec-Trait implementiert (eigene encode und decode);
• arrow-mlakes RecordBatch, gesendet als Arrow IPC-Stream. Verwenden Sie es für alles Große; ein Batch pro Parquet-Zeilengruppe ist die beabsichtigte Körnung.

Ein ObjectRef innerhalb einer Struktur oder eines Seq ist ein Kompilierfehler, der die drei Formen nennt, die funktionieren: das Ref als Argument, eine Liste von Refs oder ein Wert neben Refs mit TaskWith. Ergebnisse von 64 KiB oder weniger kommen zurück durch den Head; größere bleiben auf dem Worker, der sie erstellt hat, und werden aus Shared Memory auf demselben Host gemappt oder über Hosts hinweg abgerufen.

Actors

Ein Actor ist ein Zustand auf einem eigenen Worker. Seine Aufrufe werden nacheinander in der Reihenfolge ihrer Erstellung ausgeführt und geben wie Tasks ObjectRefs zurück.

from dista import Actor, ActorHandle, Client, CtxTask, Method, ObjectRef, Runtime

@fieldwise_init
struct Counter(Movable):
    var total: Int

def new_counter(start: Int) -> Counter:
    return Counter(start)

def add(mut c: Counter, n: Int) -> Int:
    c.total += n
    return c.total

def bump(mut c: Client, h: ActorHandle[CounterActor]) raises -> Int:
    var r = c.call[Add](h, 1)    # ein Task, der den Actor aufruft, den er erhalten hat
    var n = c.get(r)
    c.release(r)
    return n

comptime CounterActor = Actor[Counter, Int, new_counter]
comptime Add = Method[CounterActor, Int, Int, add]
comptime Bump = CtxTask[ActorHandle[CounterActor], Int, bump]

def main() raises:
    var rt = Runtime[CounterActor, Add, Bump](workers=3)
    var c = rt.create[CounterActor](0, checkpoint_every=100)
    var refs = List[ObjectRef[Int]]()
    for _ in range(10):
        refs.append(rt.call[Add](c, 5))
    print(rt.get(refs[9]))           # 50
    print(rt.get(rt.remote[Bump](c)))  # 51
    rt.kill(c)

• Ein Actor belegt einen ganzen Worker, und Tasks laufen nie darauf. Ein Actor, der erstellt wird, wenn jeder Worker bereits einen beherbergt, wartet auf einen beitretenden Worker; geben Sie also im lokalen Modus mehr Worker als Actors an.
• Handles sind Werte: Übergeben Sie ein ActorHandle an einen CtxTask (def f(mut c: Client, a: A) raises -> R) oder behalten Sie es im Zustand eines anderen Actors und rufen Sie durch den Client auf. Ein CtxMethod (def f(mut s: S, mut c: Client, a: A) raises -> R) erlaubt einem Actor, andere aufzurufen; ein Zyklus von Actor-Aufrufen führt zu einem Deadlock.
• Checkpoints: checkpoint_every=N bei create, oder rt.checkpoint(c), dessen ObjectRef[Int] die Anzahl der abgedeckten Aufrufe ist. Sie gehen durch den Codec zu DISTA_CHECKPOINT_URI, einem Verzeichnis oder s3://; ohne dieses schlägt rt.checkpoint fehl. Ein Zustandstyp, der Codec implementiert, wählt aus, was gespeichert wird. Wenn der Worker eines Actors stirbt, kommt er auf einem anderen Worker von seinem letzten Checkpoint (oder seinem Konstruktor) zurück und die Aufrufe seitdem werden erneut ausgeführt, daher sollten Methoden sicher wiederholbar sein.

Lokaler Modus und Cluster-Modus

Runtime(...) liest die Umgebung, um die Rolle des Prozesses zu wählen:

Umgebung	Rolle
keine Variable	Lokaler Modus: forkt einen Head und workers Worker, steuert sie dann
DISTA_ROLE=head	Läuft den Head auf DISTA_PORT (Standard 6390; 0 wählt einen aus und gibt ihn aus), kehrt nie zurück
DISTA_ROLE=worker, DISTA_HEAD=host:port	Bedient diesen Head, kehrt nie zurück
DISTA_HEAD=host:port	Ein Treiber, der an diesen Head sendet

Erstellen Sie also das Runtime zuerst in main: In einem Head oder Worker kehrt der Konstruktor nie zurück, und alles davor läuft in jeder Rolle. Im lokalen Modus werden die Worker innerhalb des Konstruktors geforkt und erben die davor gesetzte Umgebung. Um einen Cluster von Hand zu starten, wie es tests/cluster.sh tut:

DISTA_ROLE=head DISTA_PORT=6390 ./app               # der Head
DISTA_ROLE=worker DISTA_HEAD=headhost:6390 ./app    # einmal pro Worker
DISTA_HEAD=headhost:6390 ./app                      # der Treiber

Worker und Treiber versuchen erneut, den Head für DISTA_CONNECT_TIMEOUT Sekunden zu erreichen (Standard 60), daher spielt die Startreihenfolge keine Rolle. Ein Head,

der mit dista head gestartet wurde, ist stattdessen generisch: Er bedient mehrere Programme gleichzeitig und gibt die Tasks jedes Treibers nur an Worker, die aus derselben Binärdatei erstellt wurden.

Der dista-Befehl

Erstellt aus dem dista-Repository mit pixi run cli (nach build/dista). Jeder Befehl findet seinen Head über --head, sonst DISTA_HEAD, sonst localhost:6390.

Befehl	Was er tut
dista head [--port 6390] [--pipeline 2]	Ein generischer Head für jedes Programm
dista status	Worker mit ihrem Registry-Hash, Treiber, Actors
dista status --json	Die Nachfrage, in einer Zeile, für einen Autoscaler
dista drain <ip:port>	Entleert diesen Worker, wie SIGTERM; verweigert den letzten, solange Arbeit übrig ist
dista scale --once	Gibt die Entscheidung der Richtlinie für die aktuelle Nachfrage aus
dista timeline -o trace.json	Die abgeschlossenen Tasks jedes Treibers, als Chrome-Trace
dista stop	Fährt den Head und seine Worker herunter

Mit gesetztem DISTA_METRICS_PORT bedienen der Head und jeder Worker Prometheus- Metriken unter /metrics. Jede Variable, jeder Metrikname und die Kubernetes- Manifeste sind in references/operations.md.

Entleeren und Autoscaling

Ein Worker, der SIGTERM erhält (wie Kubernetes einen Pod beendet), entleert, statt zu sterben: Er nimmt keine neue Arbeit an, beendet seine aktuelle, seine Ergebnisse werden zu anderen Workern verschoben und sein Actor bewegt sich über einen Checkpoint, dann beendet er mit 0. Nichts schlägt fehl oder läuft zweimal (tests/drain.sh). DISTA_DRAIN_SECONDS (Standard 25) begrenzt es; ein zweites SIGTERM beendet es sofort.

dista status --json liefert tasks_ready, tasks_waiting, tasks_running, actors_waiting, pipeline, checkpoints, und pro Worker address, node, running, idle_seconds, held_bytes, actor und draining. dista.scale (importieren Sie es von dista.scale; es wird nicht von dista reexportiert) verwandelt das in eine Entscheidung, ohne eigene Uhr oder E/A:

from dista.scale import Demand, Policy, decide

def main() raises:
    var d = Demand.parse('{"tasks_ready":100,"pipeline":2,"workers":[]}')
    var got = decide(d, Policy(min_workers=0, max_workers=6), now=10)
    print(got.want, got.start, len(got.drain))  # 6 4 0

Policy-Felder und Standardwerte: min_workers=1 (0 ist erlaubt: der letzte Worker wird entleert, sobald kein Task mehr übrig ist und er nichts hält), max_workers=4, tasks_per_worker=0 (die Pipeline-Tiefe des Heads), up_after=5 Sekunden an bereiten Tasks ohne freien Slot, idle_before_down=60 Sekunden und max_step=4 Worker, die pro Entscheidung gestartet werden. decide(demand, policy, now, pending=0, short_since=0) gibt start, drain (Adressen, zuerst die wenigsten belegten Bytes), want und short_since zurück, die der Anbieter beim nächsten Mal mit pending, den gestarteten, aber noch nicht beigetretenen Workern, zurückgibt. dista scale --once --min 0 --max 6 ... gibt es zeilenweise aus; tests/autoscale.sh treibt einen Job von 0 Workern auf 6 und zurück auf 0 damit.

Unter orca

Eine orca-Pipeline deklariert ihre Tasks und Worker, und orca startet die Laufzeit einmal pro Lauf. Ein Schritt baut nie ein eigenes Runtime; er leiht sich das des Laufs (ein Auszug):

comptime Trips = Task[Day, Int, trips]
comptime FanoutTasks = Tasks[Trips]                   # von orca

def month_trips(mut ctx: Ctx, m: Month) raises -> Int:
    ref rt = ctx.runtime[FanoutTasks]()
    var r = rt.remote[Trips](Day(m.copy(), 1))   # dann rt.get, wie üblich
    ...

comptime Fanout = Pipeline["fanout", MonthTrips, AllTrips, tasks=FanoutTasks, workers=3]

Das vollständige Programm ist examples/fanout.mojo in orca.mojo.

orca führt die Schritte eines Laufs in der Reihenfolge und die Partitionen eines gemappten Schritts nacheinander aus, daher ist ein Schritt, der einen Task aufruft und auf ihn wartet, seriell. Senden Sie die gesamte Arbeit eines Schritts zuerst mit rt.remote, dann rt.get für jedes Ref.

Häufige Fehler

• registry mismatch: this peer was built with different tasks. Der Treiber und die Worker stammen von unterschiedlichen Builds oder listen unterschiedliche Tasks in Runtime[...] auf. Der Registry-Hash deckt die Task-Namen in der Reihenfolge ab, daher neu bauen und diese eine Binärdatei in jeder Rolle ausführen.
• ... is not registered with this runtime. Jeder Task, Actor und jede Methode, die Sie aufrufen, muss in Runtime[...] sein. Aufrufe über den Client eines CtxTask werden beim Aufruf nicht überprüft, daher schlägt ein nicht registrierter stattdessen auf dem Worker fehl, mit unbekannte Task-ID oder unbekannte Methoden-ID.
• Eine verschachtelte Funktion oder Closure als Task. Verwenden Sie einen Top-Level def.
• Annahme, dass ein Task genau einmal läuft. Tasks eines toten Workers werden erneut ausgeführt, bis zu DISTA_TASK_RETRIES (Standard 3): Halten Sie Tasks und Methoden sicher wiederholbar.
• Erwartung, dass ein verlorenes Ergebnis zurückkommt. Ein Ergebnis, das nur von einem toten Worker gehalten wird, schlägt mit Ergebnis <id> verloren: sein Worker ist weg, ebenso wie jeder Task, der darauf wartet. Führen Sie den Job erneut aus, oder lassen Sie orca den Schritt wiederholen.
• Nicht freigegebene Refs halten kleine Ergebnisse im Speicher des Heads. Und remote, das über DISTA_MAX_QUEUED Tasks (Standard 100.000) hinaus blockiert, ist Gegendruck, kein Hängen.