dista(dista-mojo tin, 0.2.0, Mojo 1.2.0)를 사용하면 Mojo 함수를 원격으로 표시하고 호출하여 다른 프로세스나 Kubernetes 포드에서 실행할 수 있습니다. 모든 프로세스는 동일한 컴파일된 바이너리를 실행하므로, 태스크 메시지는 함수 ID와 인코딩된 인자로 구성됩니다: 코드가 전송되지 않고, 아무것도 피클되지 않으며, 컬럼 데이터는 Arrow로 전송됩니다.
dista가 아닌 것: 클러스터 관리자입니다. 포드나 머신을 생성하지 않습니다. 조인하는 모든 워커를 사용하고, 중지하라고 지시받은 워커를 드레인하며, 원하는 워커 수를 알려줍니다. Kubernetes, 셸 루프 또는 orca가 워커를 시작하고 중지합니다. 또한 죽은 워커로 인해 손실된 결과를 재구성하지 않습니다(계보 재구성 없음): 작업이 실패하고 결과를 명명하며, 오케스트레이터가 다시 실행합니다.
설치된 환경에는 dista가 컴파일된 dista.mojoc으로만 존재합니다. 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: 인자로 ref 사용
rt.release(maps)
- 인자 개수:
Task[A, R, f]는 인자를 하나,Task2[A, B, R, f]는 두 개,Task3[A, B, C, R, f]는 세 개 받습니다. 더 많이 전달하려면 하나의 구조체로 전달하세요.Task2를 인자 하나로 호출하면 호출 시takes 2 arguments가 발생합니다. - 함수는 예외를 발생시키거나 발생시키지 않을 수 있습니다. 예외를 발생시키는 태스크는
rt.get에서 해당 메시지를 다시 발생시키고 워커는 계속 실행됩니다. - Ref는 드라이버가 아무것도 가져오지 않고도 태스크에 공급됩니다. 태스크가
A를 받는 곳에ObjectRef[A]를 전달하거나,Seq[X](순서대로 수집)를 받는 곳에List[ObjectRef[X]]를 전달하세요. 태스크는 모든 입력이 존재할 때 한 번 실행되며, 하나라도 실패하면 입력의 오류로 실패합니다. rt.release(ref)(또는 리스트)는 결과를 해제하여 보류 중인 태스크가 더 이상 필요하지 않으면 헤드와 해당 워커에서 결과를 해제합니다; 이후에get을 호출하면 예외가 발생합니다. 결과는 그렇지 않으면 런타임이 종료될 때까지 유지됩니다.Runtime의 추가 기능:rt.ready(r)(기다리지 않고 완료 여부 확인),rt.location(r)(보유 워커의ip:port및 노드 저장소 파일, 결과가 인라인으로 반환된 경우""),rt.stats(),rt.bytes_fetched()및rt.timeline(path)(Perfetto용 Chrome 트레이스 작성).
셔플: 여러 결과 또는 ref 옆의 값
f가 Seq[X]를 반환하는 TaskMany[A, X, f]는 맵 태스크입니다: rt.remote_many[T](arg, n)은 각각 별도로 저장, 가져오기 및 해제되는 파티션당 하나씩 n개의 ref를 반환합니다; f는 정확히 n개의 항목을 반환해야 합니다. A와 Seq[X]를 받는 f가 있는 TaskWith[A, X, R, f]는 리듀서입니다: rt.remote[T](j, refs). 컴파일되는 샘플과 자세한 내용은 references/shuffles.md에 있습니다.
인자 또는 결과가 될 수 있는 것
코덱은 컴파일 타임에 인코딩을 선택합니다:
Int,IntN,UIntN,Float32,Float64,Bool및String;Seq[T], 코덱을 통과하는 리스트 (List는 통과하지 않음: 리플렉션이 요소 타입을 복구할 수 없음), 구조체 필드로도 가능;- 위 항목의 일반 구조체, 리플렉션으로 필드별로 (필드는
Movable이어야 함; 인자와 결과 타입은Movable & Deinitable필요); Codec트레이트를 구현하는 타입 (자체encode및decode);- arrow-mlake의
RecordBatch, Arrow IPC 스트림으로 전송. 큰 데이터에 사용하세요; Parquet 행 그룹당 하나의 배치가 의도된 단위입니다.
구조체 또는 Seq 내부의 ObjectRef는 작동하는 세 가지 형태(ref를 인자로, ref 리스트, 또는 TaskWith로 ref 옆의 값)를 명명하는 컴파일 오류입니다. 64KiB 이하의 결과는 헤드를 통해 반환됩니다; 더 큰 결과는 해당 결과를 만든 워커에 남아 동일한 호스트의 공유 메모리에서 매핑되거나 호스트 간에 가져옵니다.
액터
액터는 자체 워커의 상태입니다. 해당 호출은 만들어진 순서대로 한 번에 하나씩 실행되며, 태스크와 마찬가지로 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)
- 하나의 액터는 하나의 전체 워커를 차지하며, 태스크는 그 위에서 실행되지 않습니다. 모든 워커가 하나를 호스팅할 때 생성된 액터는 워커가 조인할 때까지 기다리므로, 로컬 모드에서는 액터보다 더 많은 워커를 할당하세요.
- 핸들은 값입니다:
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은 하나를 선택하여 출력)에서 헤드 실행, 반환되지 않음 |
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 # 워커당 한 번
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 | 오토스케일러를 위한 수요를 한 줄로 표시 |
dista drain <ip:port> | 해당 워커를 드레인(SIGTERM과 동일); 작업이 남아 있는 동안 마지막 워커는 거부 |
dista scale --once | 현재 수요에 대한 정책의 결정 출력 |
dista timeline -o trace.json | Chrome 트레이스로 모든 드라이버의 완료된 태스크 |
dista stop | 헤드와 모든 워커 종료 |
DISTA_METRICS_PORT가 설정되면 헤드와 모든 워커가 /metrics에서 Prometheus 메트릭을 제공합니다. 모든 변수, 메트릭 이름 및 Kubernetes 매니페스트는 references/operations.md에 있습니다.
드레이닝 및 오토스케일링
SIGTERM(Kubernetes가 포드를 중지하는 방식)을 받은 워커는 죽는 대신 드레인됩니다: 새 작업을 받지 않고, 진행 중인 작업을 완료하며, 결과가 다른 워커로 이동하고 액터가 체크포인트를 통해 이동한 후 0으로 종료됩니다. 아무것도 실패하거나 두 번 실행되지 않습니다(tests/drain.sh). DISTA_DRAIN_SECONDS(기본값 25)가 시간을 제한하며; 두 번째 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에서 다시 내보내지 않으므로 dista.scale에서 가져옴)은 이를 자체적인 클록이나 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 (결정당 시작할 워커 수). decide(demand, policy, now, pending=0, short_since=0)은 start, drain (주소, 보유 바이트가 가장 적은 순), want 및 short_since를 반환하며, 제공자는 이를 다음 번에 pending(시작되었지만 아직 조인하지 않은 워커)과 함께 다시 전달합니다. dista scale --once --min 0 --max 6 ...은 이를 줄로 출력합니다; 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는 실행의 스텝을 순서대로 실행하고 매핑된 스텝의 파티션을 하나씩 실행하므로, 하나의 태스크를 호출하고 기다리는 스텝은 직렬입니다. 먼저 rt.remote로 스텝의 모든 작업을 보낸 다음 각 ref에 대해 rt.get을 호출하세요.
일반적인 실수
registry mismatch: this peer was built with different tasks. 드라이버와 워커가 다른 빌드에서 왔거나,Runtime[...]에 다른 태스크를 나열합니다. 레지스트리 해시는 태스크 이름을 순서대로 포함하므로, 다시 빌드하고 모든 역할에서 해당 하나의 바이너리를 실행하세요.... is not registered with this runtime. 호출하는 모든 태스크, 액터 및 메서드는Runtime[...]에 있어야 합니다.CtxTask의Client를 통한 호출은 호출 시 확인되지 않으므로, 등록되지 않은 것은 워커에서unknown task ID또는unknown method ID로 실패합니다.- 태스크로 중첩 함수 또는 클로저. 최상위
def를 사용하세요. - 태스크가 한 번 실행된다고 가정. 죽은 워커의 태스크는
DISTA_TASK_RETRIES(기본값 3)까지 다시 실행됩니다: 태스크와 메서드를 반복 실행에 안전하게 유지하세요. - 손실된 결과가 돌아올 것으로 예상. 죽은 워커만이 보유한 결과는
result <id> lost: its worker went away로 실패하며, 이를 기다리는 모든 태스크도 실패합니다. 작업을 다시 실행하거나 orca가 스텝을 재시도하도록 하세요. - 해제되지 않은 ref는 작은 결과를 헤드의 메모리에 유지합니다. 그리고
DISTA_MAX_QUEUED태스크(기본값 100,000)를 넘는remote차단은 중단이 아닌 역압력입니다.