並行実行:@task の future オーケストレーション
役割
@task の実行時セマンティクスは PregelRunner が引き受けます。task 呼び出し自体は即座に実行されるわけではなく、関数と引数を現在の runner に渡し、runner がいつ・どのスレッド/コルーチンで動かすかを決めます。この層は3つの仕事を担います。task 呼び出しを PUSH タスクに変換してキューに積む、同一スーパーステップ (superstep) 内でトリガーされた複数タスクを並行実行する、future で結果を呼び出し元に返す。PregelRunner(PregelRunner:135)がこのオーケストレーションの中心です。
位置としては PregelLoop の下、BackgroundExecutor / AsyncBackgroundExecutor の上です。loop.tick が当ステップで動かすべきタスクを計算し、runner.tick / atick(PregelRunner.tick:176)に渡します。runner は submit で各タスクを executor に投げ、自身は FuturesDict で全 in-flight future を追跡し、done になったら writes を書き戻し、例外をエラー処理ノードにルーティングします。
設計動機
- task 呼び出し即ち PUSH タスク:entrypoint 関数本体に
task_a(); task_b()と書いても、この2つの task に明示的なエッジは要りません。_call_with_optionsは config からCONFIG_KEY_CALLコールバック(runner が注入した_call/_acall)を取り出し、task をCallデータ構造に包んでschedule_taskでTASKSチャネルに積みます(schedule PUSH task:713)。次回のprepare_next_tasksが自然にそれを見つけます。 - 同期と非同期の統一抽象:
SyncAsyncFuture(SyncAsyncFuture:253)はconcurrent.futures.Futureであると同時に__await__を持つオブジェクトです。その__await__は単に1回yieldして制御を外側の runner に返します。同期 entrypoint ではtask().result()でブロックして値を取り、非同期 entrypoint ではawait task()でコルーチンを譲ります。両者は同一の未来オブジェクトを共有します。 - next_tick スケジューリング:子 task はスケジュールされた瞬間に動かず、
__next_tick__=True(next_tick:768)で executor に「次の tick で動かす」ことを伝えます。これにより現在の tick の writes が先に commit され、stream イベントが先に流れてから子タスクが始まるため、stream の順序がユーザーがコードに書いた呼び出し順と一致します。 - 再入の重複排除:同一 task が複数回呼ばれたとき、
_callはまずfuturesから同じ id の future を探し、あれば再利用します。task がすでに完了していれば(next_task.writesが非空)、writes からRETURN値を取り出して完了済み future を構築して返します(dedup already ran:724)。親タスク再試行時に子タスクが重複実行されるのを防ぎます。 max_concurrencyセマフォ:非同期パスではAsyncBackgroundExecutorがconfig["max_concurrency"]を取り、asyncio.Semaphoreを起動します。すべての task coro をgated(semaphore, coro)で1層包み(gated:214)、同一グラフ内で同時に in-flight な task 数の上限を制御します。LLM 呼び出しでイベントループが圧倒されるのを防ぐためです。
主要ファイル
PregelRunner:135— 並行実行器。submit/put_writes/node_error_handler_map/schedule_error_handlerを保持。PregelRunner.tick:176— タスクのバッチを同期的に実行。concurrent.futures.wait(FIRST_COMPLETED)のループで done future を取り出し、writes を commit。PregelRunner.atick:360— 非同期版。asyncio.waitが同期waitを置き換える以外はイテレータセマンティクスが同じ。_call:700— 同期パスのCONFIG_KEY_CALLコールバック。task 呼び出しを PUSH タスクに変換し、重複排除と future のチェーンを行う。_acall:789— 非同期パスのコールバック。セマンティクスは上と同じで coroutine を扱う。schedule PUSH task:713—schedule_task(実体はloop.accept_push/aaccept_push)経由でCallデータ構造をTASKSチャネルに積む。SyncAsyncFuture:253— sync Future でもあり async awaitable でもある橋渡しオブジェクト。__await__が単一のyieldで制御を返す。_call_with_options:276—_TaskFunction.__call__の実装。config からCONFIG_KEY_CALLを取り出して呼ぶ。BackgroundExecutor:40— 同期スレッドプールのコンテキストマネージャ。submitはcopy_context()で子スレッドに contextvar を継承。AsyncBackgroundExecutor:122— 非同期コンテキストマネージャ。asyncio.get_running_loop()ベース。オプションでmax_concurrencyセマフォ。gated:214—async def gated(semaphore, coro):まずacquire()してからrelease()し、並行上限を task coro の外に被せる。
データフロー
_call は task 呼び出しの中核で、「entrypoint 関数内で task_fn(arg) を呼ぶ」を「runner 内で PUSH タスクをスケジュールする」に変換します:
# schedule the next task, if the callback returns one
if next_task := schedule_task(
task(),
scratchpad.call_counter(),
Call(
func,
input,
retry_policy=retry_policy,
cache_policy=cache_policy,
callbacks=callbacks,
timeout=timeout,
),
):
if fut := next(
(
f
for f, t in list(futures().items())
if t is not None and t == next_task.id
),
None,
):
# if the parent task was retried,
# the next task might already be running
pass
elif next_task.writes:
# if it already ran, return the result
fut = concurrent.futures.Future()
ret = next((v for c, v in next_task.writes if c == RETURN), MISSING)
# ...
else:
# schedule the next task
fut = submit()(
run_with_retry,
next_task,
retry_policy,
configurable={
CONFIG_KEY_CALL: partial(
_call,
weakref.ref(next_task),
futures=futures,
# ...
),
},
__reraise_on_exit__=False,
__next_tick__=True,
)
SKIP_RERAISE_SET.add(fut)
futures()[fut] = next_task3つの分岐を順に判定します。(1) 同じ id の task が実行中なら future を再利用。(2) task がすでに実行済み(writes が非空)なら writes から RETURN 値を取り出して完了済み future を構築。(3) 全新の task なら submit() で run_with_retry を BackgroundExecutor または AsyncBackgroundExecutor に投げ、再帰的な _call を CONFIG_KEY_CALL として子タスクの config に注入します。これにより子タスク内でさらに task() を呼んでも同じスケジュール機構を通ります。
PregelRunner.tick 自身の中核は FuturesDict と concurrent.futures.wait(FIRST_COMPLETED) ループです(wait loop:281)。future が done になるたびに対応する task を取り出し、commit でその writes を loop に書き戻します。task にエラー処理ノードが設定されていれば error handler task を submit で追加します。ループは全 in-flight future が完了する(または get_waiter のプレースホルダ future だけが残る)まで回り続け、終わると制御を loop.after_tick に返してチャネルに落とします。
境界と失敗
- 同期コンテキストで async task を呼ぶとエラー:
_callの冒頭でif inspect.iscoroutinefunction(func): raise RuntimeError(sync async mismatch:716)。同期 entrypoint ではawaitできず、非同期 task は動かせません。 - 子 task の例外は親に伝播:
SKIP_RERAISE_SET.add(fut)で子 task の future を「tick 終了時に re-raise しない」マーク付けします。例外はchainFutureで親 future に伝播し、親 task がawaitしたときに初めて見えます(skip reraise:781)。これによりエラー伝播経路が呼び出し階層と一致し、runner が直接例外を投げて tick 全体を中断するのを防ぎます。 __next_tick__は同期モードで遅延実行:BackgroundExecutor.submitはnext_tick(ctx.run, fn, ...)(next_tick sync:66)でタスクを次回の tick に並べます。非同期 executor では__next_tick__は noop で、asyncio タスクがもともと非同期にスケジュールされるためです。max_concurrencyはノードレベルの並行度には影響しない:AsyncBackgroundExecutorのセマフォは task レベルの coro(gated)だけを制限し、PregelRunner.tickが同時にいくつのノードをスケジュールするかには影響しません。ノードの並行度はprepare_next_tasksが決め、セマフォは単にスロットルします(semaphore:153)。- 親タスク再試行時に子タスクは再実行されない:親タスクが retry すると、
_callは「dedup already ran」分岐に入り、既存の future を返すかwritesから結果を取り出します(dedup already ran:724)。冪等性を保証します。 - contextvar の継承:
BackgroundExecutor.submitはcopy_context()(copy_context:62)で子スレッドに親スレッドの contextvar(RunnableConfigなど)を継承させ、子タスクが config を取り損なうのを防ぎます。
まとめ
@task の並行性は PregelRunner の future オーケストレーションに由来します。中核の抽象は SyncAsyncFuture で同期/非同期を橋渡しし、_call / _acall が task 呼び出しを PUSH タスクに変換し、BackgroundExecutor / AsyncBackgroundExecutor が実行スレッド/コルーチンを提供します。PUSH タスクのスケジューリング詳細は /subgraph/send-command を、entrypoint 内での task のセマンティクスは /func/entrypoint-task を参照してください。
公式資料:LangGraph 文档 · README