dista（dista-mojo tin、バージョン0.2.0、Mojo 1.2.0）を使用すると、Mojo関数をリモートとしてマークし、呼び出して、他のプロセスやKubernetesポッドで実行できます。すべてのプロセスは 同じコンパイル済みバイナリ を実行するため、タスクメッセージは関数IDとエンコードされた引数のみで構成されます。コードの出荷はなく、ピクル化も行われず、カラムナデータはArrow形式で転送されます。
distaが ではないもの: クラスタマネージャです。ポッドやマシンを作成しません。参加したワーカーを使用し、停止指示を受けたワーカーをドレインし、必要なワーカー数を宣言します。Kubernetes、シェルループ、またはorcaがワーカーの起動・停止を行います。また、ワーカーがダウンして失われた結果を再構築することもありません（lineage再構築なし）。ジョブは失敗し、結果を特定して、オーケストレーターが再実行します。
インストールされた環境には、コンパイル済みの dista.mojoc としてのみdistaが含まれています。ソースはGitHubの magmalake/dista.mojo で、pixi.toml で指定した rev タグ、またはpixiのgitキャッシュ（macOS: ls -d ~/Library/Caches/rattler/cache/git-v0/checkouts/*/<revの最初の7文字>*、Linux: ~/.cache/rattler/...）で確認できます。ここで引用するファイルパスは、そのチェックアウト内のものです。
pixi shelf add dista-mojo # arrow-mlake, memory-region, objectstore を追加
Mojo自体の記述には mojo-syntax スキルをロードしてください。タスク内での高速な行単位の処理については、mojo-data-performance スキルを参照してください。
タスクを宣言し、呼び出す
タスクは トップレベルの def であり、引数と結果の型を指定して一度宣言します。Mojo 1.2では関数の結果型を推論できないためです。レジストリは Runtime のパラメータリストです。このファイル内のすべての完全なサンプルは、dista 0.2.0 に対してビルドおよび実行されています。
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) # ObjectRef[Int]、即座に返る
print(rt.get(r)) # 49: 結果を待つ
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: 引数としてリファレンスを渡す
rt.release(maps)
- アリティ:
Task[A, R, f]は1つの引数、Task2[A, B, R, f]は2つ、Task3[A, B, C, R, f]は3つの引数を取ります。それ以上は1つの構造体として渡します。Task2を1つの引数で呼び出すと、呼び出し時にtakes 2 argumentsが発生します。 - 関数はエラーを発生させる場合とそうでない場合があります。 エラーを発生させるタスクは、
rt.get内でそのメッセージを再発生させ、ワーカーは処理を続行します。 - リファレンスは、ドライバが何もフェッチせずにタスクに供給されます。
ObjectRef[A]をタスクがAを受け取る場所に、またはList[ObjectRef[X]]をタスクがSeq[X]を受け取る場所に（順序通りに収集されて）渡します。タスクはすべての入力が存在すると一度実行され、いずれかの入力が失敗した場合はそのエラーで失敗します。 rt.release(ref)（またはリスト）は結果を解放します。保留中のタスクが不要になった時点で、ヘッドとワーカー上の結果が解放されます。その後getを呼び出すとエラーが発生します。結果はそれ以外の場合はランタイムが終了するまで保持されます。Runtimeのその他の機能:rt.ready(r)（待たずに終了したかどうか）、rt.location(r)（保持しているワーカーのip:portとノードストアファイル、結果がインラインで返された場合は""）、rt.stats()、rt.bytes_fetched()、およびrt.timeline(path)（Perfetto用のChromeトレースを書き出します）。
シャッフル: 複数の結果、またはリファレンスと並んだ値
TaskMany[A, X, f]（f が Seq[X] を返す）はマップタスクです。rt.remote_many[T](arg, n) は、パーティションごとに1つ、合計 n 個のリファレンスを返します。それぞれが独立して保存、フェッチ、解放されます。f は正確に n 個のアイテムを返す必要があります。TaskWith[A, X, R, f]（f が A と Seq[X] を受け取る）はリデューサーです。rt.remote[T](j, refs)。コンパイル可能なサンプルと詳細は references/shuffles.md にあります。
引数または結果として使用できるもの
コーデックはコンパイル時にエンコーディングを選択します:
Int、IntN、UIntN、Float32、Float64、Bool、StringSeq[T]（コーデックを通過するリスト。Listは 不可: リフレクションで要素型を復元できません）、構造体のフィールドとしても使用可能- 上記の型のみで構成されるプレーンな構造体（リフレクションによりフィールドごとに処理。フィールドは
Movableである必要があります。引数と結果の型はMovable & Deinitableが必要） Codecトレイトを実装する型（独自のencodeとdecodeを持つ）- arrow-mlake の
RecordBatch（Arrow IPCストリームとして送信）。大きなデータに使用します。Parquet行グループごとに1バッチが意図された粒度です。
構造体内や Seq 内の ObjectRef はコンパイルエラーになり、そのエラーは機能する3つの形状（引数としてのリファレンス、リファレンスのリスト、または TaskWith によるリファレンスと並んだ値）を示します。64 KiB以下の結果はヘッド経由で返されます。それより大きいものは、それを作成したワーカーに残り、同じホスト上の共有メモリからマッピングされるか、ホスト間でフェッチされます。
アクター
アクターは、ワーカー上の独立した状態です。その呼び出しは、行われた順序で1つずつ実行され、タスクと同様に ObjectRef を返します。
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) # 渡されたアクターを呼び出すタスク
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)
- 1つのアクターでワーカー全体を占有し、タスクはそのワーカーでは実行されません。すべてのワーカーが1つずつアクターをホストしている状態でアクターを作成すると、ワーカーが参加するのを待つため、ローカルモードではワーカー数をアクター数より多く設定してください。
- ハンドルは値です:
ActorHandleをCtxTask（def f(mut c: Client, a: A) raises -> R）に渡すか、別のアクターの状態に保持し、Client経由で呼び出します。CtxMethod（def f(mut s: S, mut c: Client, a: A) raises -> R）により、アクターは他のアクターを呼び出せます。アクター呼び出しの循環はデッドロックを引き起こします。 - チェックポイント:
createのcheckpoint_every=N、またはrt.checkpoint(c)。後者のObjectRef[Int]はカバーされた呼び出し数です。これらはコーデックを介してDISTA_CHECKPOINT_URI（ディレクトリまたはs3://）に送られます。これがないとrt.checkpointは失敗します。Codecを実装する状態型は、何を保存するかを選択します。アクターのワーカーがダウンすると、最後のチェックポイント（またはコンストラクタ）から別のワーカーで復活し、その後の呼び出しが再実行されるため、メソッドは繰り返し実行しても安全である必要があります。
ローカルモードとクラスタモード
Runtime(...) は環境を読み取り、プロセスのロールを選択します:
| 環境 | ロール |
|---|---|
| どちらの変数もなし | ローカルモード: ヘッドと workers 個のワーカーをフォークし、それらを駆動 |
DISTA_ROLE=head | DISTA_PORT（デフォルト6390、0の場合は1つを選んで表示）でヘッドを実行、戻らない |
DISTA_ROLE=worker、DISTA_HEAD=host:port | そのヘッドにサービスを提供、戻らない |
DISTA_HEAD=host:port | そのヘッドにジョブを送信するドライバ |
したがって、main の最初に Runtime を作成してください。ヘッドまたはワーカーではコンストラクタは戻らず、その前のコードはすべてのロールで実行されます。ローカルモードでは、ワーカーはコンストラクタ内でフォークされ、その前に設定された環境を継承します。手動でクラスタを実行するには、tests/cluster.sh のように:
DISTA_ROLE=head DISTA_PORT=6390 ./app # ヘッド
DISTA_ROLE=worker DISTA_HEAD=headhost:6390 ./app # ワーカーごとに1回
DISTA_HEAD=headhost:6390 ./app # ドライバ
ワーカーとドライバは、DISTA_CONNECT_TIMEOUT 秒（デフォルト60）の間、ヘッドへの到達を再試行するため、起動順序は重要ではありません。代わりに dista head で起動したヘッドは汎用的で、複数のプログラムを同時に提供し、各ドライバのタスクを同じバイナリからビルドされたワーカーにのみ割り当てます。
dista コマンド
distaリポジトリから pixi run cli でビルドされます（build/dista に出力）。各コマンドは --head、次に DISTA_HEAD、最後に localhost:6390 の順でヘッドを検出します。
| コマンド | 機能 |
|---|---|
dista head [--port 6390] [--pipeline 2] | 任意のプログラム用の汎用ヘッド |
dista status | レジストリハッシュを持つワーカー、ドライバ、アクター |
dista status --json | オートスケーラー向けに、需要を1行で出力 |
dista drain <ip:port> | そのワーカーをドレイン（SIGTERMと同様）。作業が残っている間は最後の1台を拒否 |
dista scale --once | 現在の需要に対するポリシーの決定を出力 |
dista timeline -o trace.json | すべてのドライバの完了タスクをChromeトレースとして出力 |
dista stop | ヘッドとそのワーカーをシャットダウン |
DISTA_METRICS_PORT が設定されている場合、ヘッドとすべてのワーカーは /metrics でPrometheusメトリクスを提供します。すべての変数、メトリクス名、およびKubernetesマニフェストは references/operations.md にあります。
ドレインとオートスケーリング
SIGTERM（Kubernetesがポッドを停止する方法）を送信されたワーカーは、即座に終了する代わりにドレインします。新しい作業を受け付けず、現在の作業を完了し、その結果は他のワーカーに移動し、アクターはチェックポイントを経由して移動し、その後終了コード0で終了します。何も失敗したり2回実行されたりすることはありません（tests/drain.sh）。DISTA_DRAIN_SECONDS（デフォルト25）が制限時間で、2回目のSIGTERMで即座に終了します。
dista status --json は、tasks_ready、tasks_waiting、tasks_running、actors_waiting、pipeline、checkpoints、およびワーカーごとの address、node、running、idle_seconds、held_bytes、actor、draining を提供します。dista.scale（dista.scale からインポート。dista から再エクスポートはされません）は、これを決定に変換します。これ自体はクロックやI/Oを持ちません:
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 のフィールドとデフォルト: min_workers=1（0も許可: タスクがなく何も保持していない場合、最後のワーカーがドレイン）、max_workers=4、tasks_per_worker=0（ヘッドのパイプライン深度）、up_after=5 秒間空きスロットなしで準備完了タスクがある場合、idle_before_down=60 秒、max_step=4 ワーカー（1回の決定で起動）。decide(demand, policy, now, pending=0, short_since=0) は start、drain（アドレス、保持バイト数の少ない順）、want、short_since を返します。プロバイダーは次回 pending（起動されたがまだ参加していないワーカー）とともにこれらを渡します。dista scale --once --min 0 --max 6 ... はこれを1行で出力します。tests/autoscale.sh は、これを使用してジョブを0ワーカーから6ワーカーに、そして0ワーカーに戻すように駆動します。
orca の下で
orcaパイプラインはタスクとワーカーを宣言し、orcaは実行ごとにランタイムを開始します。ステップが独自の Runtime を構築することはありません。実行のランタイムを借用します（抜粋）:
comptime Trips = Task[Day, Int, trips]
comptime FanoutTasks = Tasks[Trips] # 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)) # その後 rt.get、通常通り
...
comptime Fanout = Pipeline["fanout", MonthTrips, AllTrips, tasks=FanoutTasks, workers=3]
完全なプログラムは、orca.mojo の examples/fanout.mojo です。
orcaは、実行のステップを順番に実行し、マップされたステップのパーティションを1つずつ実行するため、1つのタスクを呼び出してその結果を待つステップは直列になります。ステップのすべての作業を、最初に rt.remote で送信し、その後各リファレンスを rt.get で取得します。
よくある間違い
registry mismatch: this peer was built with different tasks. ドライバとワーカーが異なるビルドから来ているか、Runtime[...]で異なるタスクをリストしています。レジストリハッシュはタスク名を順序でカバーするため、再ビルドしてその1つのバイナリをすべてのロールで実行してください。... is not registered with this runtime. 呼び出すすべてのタスク、アクター、メソッドはRuntime[...]に含まれている必要があります。CtxTaskのClientを通じた呼び出しは呼び出し時にチェックされないため、登録されていないものは代わりにワーカー上でunknown task IDまたはunknown method IDで失敗します。- タスクとしてのネストされた関数またはクロージャ。 トップレベルの
defを使用してください。 - タスクが1回実行されることを前提とする。 ダウンしたワーカーのタスクは
DISTA_TASK_RETRIES（デフォルト3）まで再実行されます。タスクとメソッドは繰り返し実行しても安全であるように保ってください。 - 失われた結果が戻ってくることを期待する。 ダウンしたワーカーのみが保持していた結果は、
result <id> lost: its worker went awayで失敗し、それを待っているすべてのタスクも同様に失敗します。ジョブを再実行するか、orcaにステップを再試行させてください。 - 解放されていないリファレンス は、小さな結果をヘッドのメモリに保持し続けます。また、
DISTA_MAX_QUEUED（デフォルト100,000）タスクを超えてremoteがブロックするのは、バックプレッシャーであり、ハングではありません。