From fbefc71bf22a328446d34bc359ea1b17fd817a15 Mon Sep 17 00:00:00 2001 From: michael <78113554+chowhsu@users.noreply.github.com> Date: Tue, 20 Oct 2026 15:19:40 +0000 Subject: [PATCH 01/23] Thread preflight on by default for eval launches; log and record every loosened value The C1b eval scripts passed --preflight-threads off. They now keep the default (wait) and may loosen it only with explicit flags (cfg preflight_args / PREFLIGHT env); the launcher prints each loosened value with its default and records it under preflight.threads.relaxed. Co-Authored-By: Claude Opus 5.5 --- tests/test_launch_preflight_guards.py | 27 ++++++++++++++++++++++++++- tools/eval/miles.py | 8 +++++--- tools/eval/miles_head.sh | 7 +++++-- yeto/launch_preflight.py | 27 +++++++++++++++++++++++++-- 4 files changed, 61 insertions(+), 8 deletions(-) diff --git a/tests/test_launch_preflight_guards.py b/tests/test_launch_preflight_guards.py index ab9be398..c8d9ee6f 100644 --- a/tests/test_launch_preflight_guards.py +++ b/tests/test_launch_preflight_guards.py @@ -143,7 +143,32 @@ def test_threads_off_skips_and_warns(): out = io.StringIO() rec = lp.thread_preflight(_targs(preflight_threads="off"), counter=lambda: pytest.fail("off must not read"), out=out) - assert rec == {"mode": "off", "disabled": True} and "OFF" in out.getvalue() + assert rec == {"mode": "off", "disabled": True, "relaxed": {"mode": ["off", "wait"]}} + assert "OFF" in out.getvalue() and "RELAXED" in out.getvalue() + + +def test_threads_default_is_on_and_not_relaxed(): + out = io.StringIO() + rec = lp.thread_preflight(_targs(), counter=_counter([100]), out=out) + assert rec["mode"] == "wait" and rec["relaxed"] == {} and "RELAXED" not in out.getvalue() + + +def test_threads_loosened_values_are_logged_and_recorded(): + out = io.StringIO() + rec = lp.thread_preflight(_targs(preflight_thread_start=2950, preflight_thread_hard=3200), + counter=_counter([2900]), out=out) + assert rec["relaxed"] == {"start": [2950, 2800], "hard": [3200, 3000]} + assert "start 2950 (default 2800)" in out.getvalue() and "hard 3200 (default 3000)" in out.getvalue() + # a stricter setting is not a relaxation + assert lp.thread_relaxations(lp.ThreadConfig(mode="error", start=100, hard=200)) == {} + + +def test_eval_launch_args_keep_the_thread_preflight_on(): + from pathlib import Path + from tools.eval import miles + root = Path(__file__).resolve().parents[1] + assert "--preflight-threads off" not in (root / "tools/eval/miles_head.sh").read_text() + assert "--preflight-threads off" not in Path(miles.__file__).read_text() def test_thread_message_text(): diff --git a/tools/eval/miles.py b/tools/eval/miles.py index 3f9538e6..401e3fae 100644 --- a/tools/eval/miles.py +++ b/tools/eval/miles.py @@ -30,9 +30,11 @@ def _gpu(cfg: dict, n: int) -> str: def _common(cfg: dict, run_id: str, prefix: str) -> str: - # --preflight-threads off: Modal islands put no SkyPilot threads on this host; the scripts - # check the host thread count (< 3000) before the launch instead - return (f"launch --training-mode rl --rl-engine ports --preflight-threads off " + # the launcher thread preflight stays on (default wait); a cell may loosen it only with + # explicit flags in cfg["preflight_args"] (e.g. "--preflight-thread-start 2950"), which the + # launcher prints and records in the run manifest + preflight = cfg.get("preflight_args", "") + return (f"launch --training-mode rl --rl-engine ports {preflight + ' ' if preflight else ''}" f"--modal-gpu-exact --modal-retries 0 --modal-cpu 8 " f"--modal-memory 64 --modal-tape-volume {TAPE_VOLUME} --controller local --cluster-prefix {prefix} " f"--tuning lora --lora-r {cfg['lora_r']} --rl-observe-timeline {MODEL} {DATA} " diff --git a/tools/eval/miles_head.sh b/tools/eval/miles_head.sh index dac65373..e4ac5b03 100755 --- a/tools/eval/miles_head.sh +++ b/tools/eval/miles_head.sh @@ -8,6 +8,9 @@ set -u : "${P:?}" "${R:?}" "${REPO:?}" "${PY:?}" "${SKY:?}" HARD=${HARD:-2700}; WD=$(( HARD + 300 )); STEPS=${STEPS:-4}; SROUNDS=$(( 2 * STEPS + 6 )); GPU=${GPU:-a10g} +# launcher thread preflight stays on; loosen only with explicit flags, e.g. +# PREFLIGHT="--preflight-thread-start 2950" (the launcher logs and records the loosened values) +PREFLIGHT=${PREFLIGHT:-} SHA=${SHA:-HEAD}; MODAL="$PY -m modal" APP=yeto-$P; SP=$P-s; SAPP=yeto-$SP; TAPEVOL=yeto-event-tapes; STOREVOL=${STOREVOL:-yeto-eval-store}; HEAD_REGION=nebius/eu-north1 export CARGO_HOME=/home/michael/work/gpu-default-modal/cargo RUSTUP_HOME=/home/michael/work/gpu-default-modal/rustup @@ -30,8 +33,8 @@ LORA="--tuning lora --lora-r 16" MODALX="--modal-gpu-exact --modal-retries 0 --modal-cpu 8 --modal-memory 64 --modal-tape-volume $TAPEVOL" STORE="--rl-checkpoint-store modal-volume://$STOREVOL/$P" COMMON="--training-mode rl --rl-engine ports $MODALX $MODEL $DATA $LORA $STORE --rl-algorithm-spec examples/rl_algorithms/decoupled-ppo-c2.json --rl-max-policy-age 2 --rl-observe-timeline --reward-function gsm8k_reward:score --fragments 1 --pipeline 1 --rollout-batch-size 32 --n-samples-per-prompt 4 --rollout-max-response-len 1024 --seq-len 2048 --inner-lr 1e-5 --seed 17 --apply-chat-template-kwargs '{\"enable_thinking\": false}' --trust-remote-code" -TARGS="launch --controller head --syncer-region $HEAD_REGION --preflight-threads off --rl-stall-timeout 2400 --on-demand --no-island-relaunch --modal-timeout-s $HARD --cluster-prefix $P --gpu modal:1x$GPU,modal:1x$GPU --rl-island-scheduling elastic --rl-sample-role trainer --rl-sample-wait-s 900 $COMMON --total-steps $STEPS" -SARGS="launch --preflight-threads off --controller local --cluster-prefix $SP --gpu modal:1x$GPU --rl-single-island-no-sync --rl-sample-role sampler --rl-sampler-max-backlog-rounds 4 $COMMON --total-steps $SROUNDS" +TARGS="launch --controller head --syncer-region $HEAD_REGION $PREFLIGHT --rl-stall-timeout 2400 --on-demand --no-island-relaunch --modal-timeout-s $HARD --cluster-prefix $P --gpu modal:1x$GPU,modal:1x$GPU --rl-island-scheduling elastic --rl-sample-role trainer --rl-sample-wait-s 900 $COMMON --total-steps $STEPS" +SARGS="launch $PREFLIGHT --controller local --cluster-prefix $SP --gpu modal:1x$GPU --rl-single-island-no-sync --rl-sample-role sampler --rl-sampler-max-backlog-rounds 4 $COMMON --total-steps $SROUNDS" git -C $REPO rev-parse $SHA > $R/yeto_sha.txt; echo "$TARGS" > $R/args.txt; echo "$SARGS" > $R/sargs.txt echo "$APP $SAPP $HEAD_CL" > $R/resources.txt eval "$(/usr/bin/python3 - <<'PYC' diff --git a/yeto/launch_preflight.py b/yeto/launch_preflight.py index 539b07a0..c173c4da 100644 --- a/yeto/launch_preflight.py +++ b/yeto/launch_preflight.py @@ -185,6 +185,24 @@ def thread_message(count: ThreadCount, cfg: ThreadConfig, headline: str) -> str: ]) +def thread_relaxations(cfg: ThreadConfig) -> dict: + """The thread settings that are looser than the defaults, as ``{name: [value, default]}``. + + The check is on by default; a run may only loosen it with an explicit flag, and the + launcher prints and records every loosened value (empty dict: defaults or stricter).""" + + relaxed = {} + if cfg.mode == "off": + relaxed["mode"] = ["off", DEFAULT_THREAD_MODE] + if cfg.start > DEFAULT_THREAD_START: + relaxed["start"] = [cfg.start, DEFAULT_THREAD_START] + if cfg.hard > DEFAULT_THREAD_HARD: + relaxed["hard"] = [cfg.hard, DEFAULT_THREAD_HARD] + if cfg.mode == "wait" and cfg.wait_s > DEFAULT_THREAD_WAIT_S: + relaxed["wait_s"] = [cfg.wait_s, DEFAULT_THREAD_WAIT_S] + return relaxed + + def thread_preflight(args, *, counter: Callable[[], ThreadCount] | None = None, clock: Callable[[], float] = time.monotonic, sleep: Callable[[float], None] = time.sleep, @@ -197,10 +215,15 @@ def thread_preflight(args, *, counter: Callable[[], ThreadCount] | None = None, out = sys.stderr if out is None else out cfg = ThreadConfig.from_args(args) + relaxed = thread_relaxations(cfg) + if relaxed: + print("[preflight] WARNING: thread preflight RELAXED by explicit flags: " + + ", ".join(f"{k} {v[0]} (default {v[1]})" for k, v in relaxed.items()), + file=out, flush=True) if cfg.mode == "off": print("[preflight] WARNING: thread preflight is OFF (--preflight-threads off); the " "s1run thread guard may tear down this run", file=out, flush=True) - return {"mode": "off", "disabled": True} + return {"mode": "off", "disabled": True, "relaxed": relaxed} counter = counter or count_threads started = clock() count = counter() @@ -233,7 +256,7 @@ def thread_preflight(args, *, counter: Callable[[], ThreadCount] | None = None, f"(SkyPilot API server {count.sky_api if count.sky_api_found else 'not found'})", file=out, flush=True) return {"mode": cfg.mode, "start": cfg.start, "hard": cfg.hard, "wait_s": cfg.wait_s, - "waited": waited, "readings": readings, **count.to_dict()} + "waited": waited, "readings": readings, "relaxed": relaxed, **count.to_dict()} # --------------------------------------------------------------------------- From 59ab2b492d934e204783889282d4bdab521a3e13 Mon Sep 17 00:00:00 2001 From: michael <78113554+chowhsu@users.noreply.github.com> Date: Tue, 20 Oct 2026 15:28:54 +0000 Subject: [PATCH 02/23] Samplers draw disjoint data slots; trainer ends FINISHED on an exhausted buffer; launcher follows rescheduled Modal containers - sample_roles.DataSlot: sampler i of n keeps prompt rows r % n == i. The launcher sets --rl-sampler-data-slot per pair sampler island; the Miles buffer writer splits its RolloutDataSource before the first draw and any resume seek; the engine sampler spec takes prompts.data_slot (stage 1 sets it with several sampler nodes). - BatchReader raises SampleBufferExhausted when the buffer ended with no trainable group; IslandDriver ends the run FINISHED before that round (event rl_buffer_exhausted), except under a strict roster. - ContainerLogFollower: when a new container is expected, the launcher lists the app's containers and streams each new one's log ('modal container logs'), with a warning; the call log stream misses it. Co-Authored-By: Claude Opus 5.5 --- tests/test_s26_e9_data_slots.py | 273 ++++++++++++++++++ yeto/launch/modal_islands.py | 17 ++ yeto/launcher.py | 13 +- yeto/modal_runner.py | 101 +++++++ yeto/rl/adapters/miles/buffer_rollout.py | 44 ++- .../rl/adapters/miles/buffer_rollout_miles.py | 6 + yeto/rl/adapters/miles/island_entry.py | 2 +- yeto/rl/engine/bridges.py | 8 + yeto/rl/engine/buffer_payload.py | 24 ++ yeto/rl/engine/driver.py | 13 +- yeto/rl/engine/engine_sampler_entry.py | 26 +- yeto/rl/engine/run_flags.py | 3 + yeto/rl/engine/sample_roles.py | 51 ++++ yeto/rl/stage1.py | 12 +- 14 files changed, 575 insertions(+), 18 deletions(-) create mode 100644 tests/test_s26_e9_data_slots.py diff --git a/tests/test_s26_e9_data_slots.py b/tests/test_s26_e9_data_slots.py new file mode 100644 index 00000000..8609d4d8 --- /dev/null +++ b/tests/test_s26_e9_data_slots.py @@ -0,0 +1,273 @@ +"""S26 E9.3: disjoint sampler data slots, a normal trainer end on an exhausted +buffer, and the launcher following a rescheduled Modal container's log.""" + +from __future__ import annotations + +import io +import json +import random +import sys +from types import SimpleNamespace + +import pytest +import torch + +from yeto.rl.engine import sample_roles +from yeto.rl.engine.buffer_payload import SampleBufferExhausted, SampleBufferRolloutError, is_buffer_exhausted +from yeto.rl.engine.engine_sampler_entry import JsonlPrompts, SamplerSpecError +from yeto.rl.engine.sample_roles import DataSlot + + +# ------------------------------------------------------------------ DataSlot +def test_data_slot_parse_and_rows(): + assert DataSlot.parse(None) == DataSlot() and DataSlot().whole + slot = DataSlot.parse("1/3") + assert (slot.index, slot.count, slot.env()) == (1, 3, "1/3") + assert slot.rows(list(range(10))) == [1, 4, 7] + for bad in ("3/3", "-1/2", "1/0", "2"): + with pytest.raises(ValueError): + DataSlot.parse(bad) + with pytest.raises(ValueError, match="empty"): + DataSlot(2, 3).rows([0, 1]) + + +def test_slots_of_all_samplers_cover_the_data_once(): + rows = list(range(17)) + drawn = [r for k in range(3) for r in DataSlot(k, 3).rows(rows)] + assert sorted(drawn) == rows + + +def test_pair_island_args_give_each_sampler_its_slot_and_flag(): + pair = SimpleNamespace(rl_island_sample_role=["0:sampler", "1:sampler", "2:trainer"], + rl_sampler_max_backlog_rounds=None, rl_sample_wait_s=None, total_steps=8, + rl_sampler_spot=False) + a0, a1, t = (sample_roles.island_args(pair, m, 3) for m in range(3)) + assert (a0.rl_sampler_data_slot, a1.rl_sampler_data_slot) == ("0/2", "1/2") + assert getattr(t, "rl_sampler_data_slot", None) is None + assert " --rl-sampler-data-slot 1/2" in sample_roles.island_flags(a1) + one = SimpleNamespace(rl_island_sample_role=["0:sampler", "1:trainer"], rl_sampler_max_backlog_rounds=None, + rl_sample_wait_s=None, total_steps=8, rl_sampler_spot=False) + assert sample_roles.island_args(one, 0, 2).rl_sampler_data_slot is None # one sampler: the whole data + + +def test_data_slot_flag_on_both_layers(): + import argparse + + from yeto.rl.engine.run_flags import CLI, ISLAND, flags_for + + for layer in (CLI, ISLAND): + parser = argparse.ArgumentParser() + for flag in flags_for(layer): + if flag.options[0] == "--rl-sampler-data-slot": + parser.add_argument(*flag.options, **dict(flag.kwargs)) + assert parser.parse_args(["--rl-sampler-data-slot", "0/2"]).rl_sampler_data_slot == "0/2" + + +# ------------------------------------------------------------ engine sampler +def _prompts(tmp_path, n=10): + path = tmp_path / "p.jsonl" + path.write_text("".join(json.dumps({"q": f"q{i}", "a": str(i)}) + "\n" for i in range(n))) + return path + + +def test_engine_sampler_prompts_of_two_slots_are_disjoint(tmp_path): + path = _prompts(tmp_path) + ids = [] + for slot in ("0/2", "1/2"): + prompts = JsonlPrompts(path, input_key="q", label_key="a", groups_per_round=2, data_slot=slot) + ids.append([item.prompt_id for r in range(2) for item in prompts(r)]) + assert ids == [["row0", "row2", "row4", "row6"], ["row1", "row3", "row5", "row7"]] + whole = JsonlPrompts(path, input_key="q", label_key="a", groups_per_round=2) + assert [i.prompt_id for i in whole(1)] == ["row2", "row3"] # unchanged without a slot + with pytest.raises(SamplerSpecError, match="data_slot"): + JsonlPrompts(path, input_key="q", label_key="a", groups_per_round=2, data_slot="2/2") + + +def test_stage1_sampler_spec_carries_the_slot_only_with_several_samplers(): + from yeto.rl import stage1 + + src = open(stage1.__file__).read() + assert "slot=DataSlot(samplers.index(node)" in src and '"data_slot": slot.env()' in src + + +# --------------------------------------------------------------- Miles source +class _FakeDataset: + """The fields and shuffle of Miles' ``miles.utils.data.Dataset``.""" + + def __init__(self, rows, seed=3): + self.origin_samples = list(rows) + self.samples = list(rows) + self.seed, self.epoch_id = seed, -1 + + def shuffle(self, new_epoch_id): + if self.epoch_id == new_epoch_id: + return + random.seed(self.seed + new_epoch_id) + permutation = list(range(len(self.samples))) + random.shuffle(permutation) + self.samples = [self.origin_samples[i] for i in permutation] + self.epoch_id = new_epoch_id + + def __len__(self): + return len(self.samples) + + +def _source(rows, shuffle): + source = SimpleNamespace(dataset=_FakeDataset(rows), epoch_id=0, sample_offset=0, sample_group_index=0, + sample_index=0) + if shuffle: + source.dataset.shuffle(0) + return source + + +@pytest.mark.parametrize("shuffle", [False, True]) +def test_miles_data_source_slots_are_disjoint(shuffle): + from yeto.rl.adapters.miles.buffer_rollout import apply_data_slot + + rows = list(range(20)) + drawn = [] + for k in range(2): + source = _source(rows, shuffle) + assert apply_data_slot(source, DataSlot(k, 2), shuffle=shuffle) == 10 + assert sorted(source.dataset.samples) == rows[k::2] + drawn.append(set(source.dataset.samples)) + assert not drawn[0] & drawn[1] and drawn[0] | drawn[1] == set(rows) + + +def test_miles_data_slot_refuses_after_the_first_draw(): + from yeto.rl.adapters.miles.buffer_rollout import apply_data_slot + + source = _source(list(range(4)), False) + source.sample_offset = 2 + with pytest.raises(SampleBufferRolloutError, match="before the first draw"): + apply_data_slot(source, DataSlot(0, 2), shuffle=False) + whole = _source(list(range(4)), False) + assert apply_data_slot(whole, DataSlot(), shuffle=False) == 4 + + +def test_island_env_exports_the_slot_for_a_sampler_only(): + from yeto.rl.adapters.miles import buffer_rollout as br + + env = br.island_env("sampler", store_root="/s", sampler_id="island0", shards=["island0", "island1"], + data_slot="0/2") + assert env[br.DATA_SLOT_ENV] == "0/2" + with pytest.raises(ValueError, match="only a sampler"): + br.island_env("trainer", store_root="/s", max_policy_age=2, data_slot="0/2") + + +# ------------------------------------------------------- exhausted buffer end +def test_is_buffer_exhausted_follows_wrappers(): + inner = SampleBufferExhausted("ended") + assert is_buffer_exhausted(inner) + try: + try: + raise inner + except SampleBufferExhausted as e: + raise RuntimeError("ray task failed") from e + except RuntimeError as outer: + assert is_buffer_exhausted(outer) + ray_like = RuntimeError("RayTaskError") + ray_like.cause = inner + assert is_buffer_exhausted(ray_like) + dual = type("SampleBufferExhausted", (RuntimeError,), {})("dual class of another process") + assert is_buffer_exhausted(dual) + assert not is_buffer_exhausted(SampleBufferRolloutError("other")) + assert not is_buffer_exhausted(None) + + +def test_batch_reader_raises_exhausted_on_an_ended_empty_buffer(): + from yeto.rl.adapters.miles.buffer_rollout import BatchReader + + buffer = SimpleNamespace(read_cursor=5, data_ended=True, select=lambda consumer, n: []) + reader = BatchReader(buffer, max_policy_age=2, wait_s=0, poll_s=0) + with pytest.raises(SampleBufferExhausted, match="no trainable group"): + reader.take(3, 4, current_version=3) + + +def _fake_driver(tmp_path, sync, exhaust_at): + from yeto.rl.engine.algorithm import AlgorithmSpec + from yeto.rl.engine.driver import EventTape, IslandDriver + from yeto.rl.engine.fake import FakeEngine, fake_capabilities + + engine = FakeEngine(tensors={"base_model.model.layer.lora_A.weight": torch.zeros(1, 2)}, step_delta=1.0, capability_profile="miles-like") + generate = engine.rollout.generate + + def gen(rollout_id, **kw): + if rollout_id == exhaust_at: + raise RuntimeError("ray task failed") from SampleBufferExhausted("ended") + return generate(rollout_id, **kw) + + engine.rollout.generate = gen + return IslandDriver(learner_id=0, rollout=engine.rollout, trainer=engine.trainer, + policy_state=engine.policy_state, publisher=engine.publisher, + placement=engine.placement, algorithm=AlgorithmSpec(), sync=sync, + events=EventTape(tmp_path / "e.jsonl", 0), capabilities=fake_capabilities("miles-like")) + + +def test_driver_ends_finished_when_the_buffer_is_exhausted(tmp_path): + from yeto.rl.engine.bridges import LocalOnlySync + + final = _fake_driver(tmp_path, LocalOnlySync(5), exhaust_at=3).run() + assert final.policy_version == 3 + events = [json.loads(line) for line in (tmp_path / "e.jsonl").read_text().splitlines()] + names = [e["event"] for e in events] + assert "rl_buffer_exhausted" in names and "island_leave" not in names + assert next(e for e in events if e["event"] == "rl_buffer_exhausted")["rollout_id"] == 3 + + +def test_driver_still_fails_on_other_errors_and_on_a_strict_roster(tmp_path): + from yeto.rl.engine.bridges import LocalOnlySync, StrictAvgSync + + assert StrictAvgSync.ends_on_data_end is False + sync = LocalOnlySync(5) + sync.ends_on_data_end = False + with pytest.raises(RuntimeError, match="ray task failed"): + _fake_driver(tmp_path, sync, exhaust_at=2).run() + + +# ------------------------------------------------- rescheduled container log +def test_container_log_follower_streams_new_containers_only(): + from yeto.modal_runner import ContainerLogFollower + + lists = iter([{"ta-a"}, {"ta-a"}, {"ta-a", "ta-b"}, {"ta-b"}]) + streamed = [] + + def stream(cid): + streamed.append(cid) + yield "[modal-island 1] rank 0 container ta-b\n" + yield "round 2 done\nround 3 done\n" + + class Collector: + def __init__(self): + self.lines = [] + + def feed(self, line): + self.lines.append(line) + + collector = Collector() + out = io.StringIO() + follower = ContainerLogFollower(lambda: next(lists), stream, expected=1, + route=lambda learner: ("island1", collector) if learner == 1 else None, + out=out) + assert follower.poll_once() == [] and follower.baseline == {"ta-a"} + assert follower.poll_once() == [] + assert follower.poll_once() == ["ta-b"] + follower.threads[0].join(5) + assert follower.poll_once() == [] # followed once + text = out.getvalue() + assert "new Modal container ta-b" in text + assert "[island1] round 2 done" in text and "[island1] round 3 done" in text + assert streamed == ["ta-b"] and collector.lines[-1] == "round 2 done\nround 3 done" + + +def test_container_log_follower_only_when_new_containers_are_expected(): + from yeto.launch.modal_islands import container_log_follower_for + + ops = SimpleNamespace(list_container_ids=lambda: set(), stream_container_logs=lambda cid: iter(())) + cfgs = {"island0": SimpleNamespace(num_nodes=1, learner_id=0)} + assert container_log_follower_for(None, ops, cfgs, None, poller=object(), start=False) is None + assert container_log_follower_for(None, None, cfgs, None, poller=None, start=False) is None + follower = container_log_follower_for(None, ops, cfgs, None, poller=None, start=False) + assert follower is not None and follower.expected_count == 1 + assert container_log_follower_for(None, SimpleNamespace(list_container_ids=lambda: set()), cfgs, None, + poller=None, start=False) is None diff --git a/yeto/launch/modal_islands.py b/yeto/launch/modal_islands.py index a7f12af6..74649a6d 100644 --- a/yeto/launch/modal_islands.py +++ b/yeto/launch/modal_islands.py @@ -301,6 +301,23 @@ def container_set_poller_for(args, modal_ops, modal_cfgs, on_change, *, start: b return poller.start() if start else poller +def container_log_follower_for(args, modal_ops, modal_cfgs, route, *, poller, start: bool = True): + """The follower of rescheduled containers' logs (modal_runner.ContainerLogFollower), + only when a new container is expected (no failing container guard: ``poller`` + is None) and the ops can stream one container. None otherwise.""" + if poller is not None or modal_ops is None or not modal_cfgs: + return None + if not (callable(getattr(modal_ops, "list_container_ids", None)) + and callable(getattr(modal_ops, "stream_container_logs", None))): + return None + from ..modal_runner import ContainerLogFollower + + expected = sum(max(1, int(getattr(cfg, "num_nodes", 1) or 1)) for cfg in modal_cfgs.values()) + follower = ContainerLogFollower(modal_ops.list_container_ids, modal_ops.stream_container_logs, + expected, route=route) + return follower.start() if start else follower + + def effective_recover_timeout(args) -> float: """The fleet controller's learner relaunch budget. diff --git a/yeto/launcher.py b/yeto/launcher.py index 0abd2f4b..6ee357ca 100644 --- a/yeto/launcher.py +++ b/yeto/launcher.py @@ -60,7 +60,7 @@ ) from .launch.modal_islands import ( _modal_tape_run_dir, _tail_modal, _verify_modal_app_stopped, build_modal_island_config, - container_set_poller_for, effective_recover_timeout, pull_modal_tapes, + container_log_follower_for, container_set_poller_for, effective_recover_timeout, pull_modal_tapes, require_modal_for_gpu_exact, ) from .launch.fleet import ( @@ -618,6 +618,17 @@ def container_changed() -> bool: if container_poller is not None: container_guards.append(container_poller) + def route_container(learner_id: int): + # the island of a rescheduled container's lines (unique learner id only: + # the islands of a pair all run as learner 0) + names = [n for n, cfg in modal_cfgs.items() if int(cfg.learner_id) == int(learner_id)] + if len(names) != 1: + return None + return names[0], (event_collectors.get(names[0]) if names[0] in echo_names else None) + + # daemon threads; each container's log ends when the app stops + container_log_follower_for(args, modal_ops, modal_cfgs, route_container, poller=container_poller) + from . import runs from .modal_runner import RoutingOps diff --git a/yeto/modal_runner.py b/yeto/modal_runner.py index b25be3c4..19ba5353 100644 --- a/yeto/modal_runner.py +++ b/yeto/modal_runner.py @@ -1194,6 +1194,19 @@ def stream_logs(self, call_id: str): for entry in modal.FunctionCall.from_id(call_id).logs.stream(): yield getattr(entry, "message", str(entry)) + def stream_container_logs(self, container_id: str): + """The log lines of one container (``modal container logs``; follows while it + runs). The function-call stream of :meth:`stream_logs` does not deliver the + lines of a container that Modal started after a reschedule (S25 C1).""" + proc = subprocess.Popen([sys.executable, "-m", "modal", "container", "logs", str(container_id)], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) + try: + assert proc.stdout is not None + yield from proc.stdout + finally: + if proc.poll() is None: + proc.terminate() + CONTAINER_LINE_RE = re.compile(r"\[modal-island (\d+)\] rank (\d+) container (\S+)") @@ -1291,6 +1304,94 @@ def stop(self) -> None: self._stop.set() +class ContainerLogFollower: + """Follows the logs of the Modal containers the call log streams miss (S26 E9.3). + + After ``modal container stop`` (or a real reclaim) Modal runs the island's + input again in a new container, and the function-call log stream never + delivers that container's lines (S19, S25 C1), so the launcher showed only + the first container. Every ``interval_s`` this lists the app's running + containers; the first ``expected`` ids are the baseline (the call streams + show them); for every later id it prints one launcher warning and streams + that container's own log (``stream_container(id)``) on a daemon thread. + + Each line is printed with a ``[