From 5b979bc6ab1131ae6f5fa905a3d1f44c07a95783 Mon Sep 17 00:00:00 2001 From: Shang Yang Date: Fri, 3 Jul 2026 21:20:38 +0000 Subject: [PATCH 1/3] Backfill async rollouts on sample completions Add an opt-in --rollout-sample-completion-backfill mode that submits one replacement prompt group after n_samples_per_prompt individual samples finish, instead of waiting for an entire group task to return. With long-horizon agentic rollouts (SWE-agent style, 20-120 min per trial), the legacy group-level scheduler leaves engine capacity idle while the slowest sibling of each n-sample group finishes; sample-completion credits keep the pipeline saturated. The default path remains the legacy group-level scheduler. Wire the sample completion callback through both inference rollout implementations and the fully-async example worker, and add focused CPU tests for the default-disabled routing and replacement scheduling. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_013vyY58s2TFpCzqEP4KaSxC --- examples/fully_async/fully_async_rollout.py | 100 +++++++--- .../inference_rollout_common.py | 14 +- .../inference_rollout_train.py | 181 ++++++++++++++++-- miles/rollout/sglang_rollout.py | 13 +- miles/utils/arguments.py | 11 ++ .../test_sample_completion_backfill.py | 155 +++++++++++++++ 6 files changed, 419 insertions(+), 55 deletions(-) create mode 100644 tests/fast/rollout/inference_rollout/test_sample_completion_backfill.py diff --git a/examples/fully_async/fully_async_rollout.py b/examples/fully_async/fully_async_rollout.py index e4c23cc120..8874daa78a 100644 --- a/examples/fully_async/fully_async_rollout.py +++ b/examples/fully_async/fully_async_rollout.py @@ -96,6 +96,54 @@ async def continuous_worker_loop(self): active_tasks = set() max_concurrent_tasks = self.args.rollout_batch_size group_id_counter = 0 + sample_completion_backfill = getattr(self.args, "rollout_sample_completion_backfill", False) + samples_per_group = max(1, int(getattr(self.args, "n_samples_per_prompt", 1) or 1)) + completed_sample_credits = 0 + sample_backfill_initialized = False + + if sample_completion_backfill: + print( + "Sample-completion backfill enabled: " + f"initial_groups={max_concurrent_tasks}, samples_per_group={samples_per_group}" + ) + + def on_sample_done(): + nonlocal completed_sample_credits + completed_sample_credits += 1 + + def submit_one_group() -> bool: + nonlocal group_id_counter + samples = self.data_buffer.get_samples(1) + + for group in samples: + group_id = group_id_counter + group_id_counter += 1 + + generate_kwargs = dict( + args=self.args, + group=group, + sampling_params=self.state.sampling_params.copy(), + evaluation=False, + ) + if sample_completion_backfill: + generate_kwargs["sample_done_callback"] = on_sample_done + + # Create new async task + task = asyncio.create_task(generate_and_rm_group(**generate_kwargs)) + + # Add completion callback + def make_callback(gid): + def task_done_callback(done_task): + result = done_task.result() + self.output_queue.put((gid, result)) + + return task_done_callback + + task.add_done_callback(make_callback(group_id)) + active_tasks.add(task) + return True + + return False while self.running: try: @@ -109,35 +157,29 @@ async def continuous_worker_loop(self): print(f"Task failed with exception: {e}") active_tasks -= done_tasks - # If active task count hasn't reached limit, try to get new data and start tasks - while len(active_tasks) < max_concurrent_tasks and self.running: - samples = self.data_buffer.get_samples(1) - - for group in samples: - group_id = group_id_counter - group_id_counter += 1 - - # Create new async task - task = asyncio.create_task( - generate_and_rm_group( - self.args, - group, - sampling_params=self.state.sampling_params.copy(), - evaluation=False, - ) - ) - - # Add completion callback - def make_callback(gid): - def task_done_callback(done_task): - result = done_task.result() - self.output_queue.put((gid, result)) - - return task_done_callback - - task.add_done_callback(make_callback(group_id)) - active_tasks.add(task) - break + if sample_completion_backfill: + if not sample_backfill_initialized: + while len(active_tasks) < max_concurrent_tasks and self.running: + if not submit_one_group(): + break + sample_backfill_initialized = len(active_tasks) >= max_concurrent_tasks + + while ( + sample_backfill_initialized + and completed_sample_credits >= samples_per_group + and self.running + ): + if not submit_one_group(): + break + completed_sample_credits -= samples_per_group + + if not active_tasks and self.running: + submit_one_group() + else: + # If active task count hasn't reached limit, try to get new data and start tasks. + while len(active_tasks) < max_concurrent_tasks and self.running: + if not submit_one_group(): + break # Brief sleep to avoid busy waiting await asyncio.sleep(1) diff --git a/miles/rollout/inference_rollout/inference_rollout_common.py b/miles/rollout/inference_rollout/inference_rollout_common.py index 9f1cc603b0..2603693677 100644 --- a/miles/rollout/inference_rollout/inference_rollout_common.py +++ b/miles/rollout/inference_rollout/inference_rollout_common.py @@ -1,6 +1,7 @@ import asyncio import logging from argparse import Namespace +from collections.abc import Callable from copy import deepcopy from typing import Any @@ -118,7 +119,11 @@ async def generate_and_rm( async def generate_and_rm_group( - state: GenerateState, group: list[Sample], sampling_params: dict[str, Any], evaluation: bool = False + state: GenerateState, + group: list[Sample], + sampling_params: dict[str, Any], + evaluation: bool = False, + sample_done_callback: Callable[[], None] | None = None, ) -> list[Sample]: args = state.args @@ -132,9 +137,10 @@ async def generate_and_rm_group( current_sampling_params = sampling_params.copy() if getattr(args, "sglang_enable_deterministic_inference", False): current_sampling_params["sampling_seed"] = args.rollout_seed + idx - tasks.append( - asyncio.create_task(generate_and_rm(state, sample, current_sampling_params, evaluation=evaluation)) - ) + task = asyncio.create_task(generate_and_rm(state, sample, current_sampling_params, evaluation=evaluation)) + if sample_done_callback is not None: + task.add_done_callback(lambda _task: sample_done_callback()) + tasks.append(task) group = await asyncio.gather(*tasks) logger.debug(f"{log_prefix} [group] All {len(group)} samples completed") diff --git a/miles/rollout/inference_rollout/inference_rollout_train.py b/miles/rollout/inference_rollout/inference_rollout_train.py index eed15557bd..23d89f9065 100644 --- a/miles/rollout/inference_rollout/inference_rollout_train.py +++ b/miles/rollout/inference_rollout/inference_rollout_train.py @@ -2,6 +2,7 @@ import logging from argparse import Namespace from collections.abc import Callable +from contextlib import suppress import sglang_router from packaging.version import parse @@ -56,7 +57,11 @@ async def get_worker_urls(args: Namespace): return [worker["url"] for worker in response["workers"]] -def submit_generate_tasks(state: GenerateState, samples: list[list[Sample]]): +def submit_generate_tasks( + state: GenerateState, + samples: list[list[Sample]], + sample_done_callback: Callable[[], None] | None = None, +): return [ asyncio.create_task( # submit a group of samples as a single task. @@ -65,6 +70,7 @@ def submit_generate_tasks(state: GenerateState, samples: list[list[Sample]]): group, sampling_params=state.sampling_params.copy(), evaluation=False, + sample_done_callback=sample_done_callback, ) ) for group in samples @@ -84,6 +90,57 @@ async def generate_rollout_async( metric_gatherer = MetricGatherer() + if getattr(args, "rollout_sample_completion_backfill", False): + data, all_data, aborted_samples = await _generate_rollout_sample_completion_backfill_async( + state, + rollout_id, + data_source, + dynamic_filter, + metric_gatherer, + ) + else: + data, all_data, aborted_samples = await _generate_rollout_group_level_async( + state, + rollout_id, + data_source, + dynamic_filter, + metric_gatherer, + ) + + assert len(data) == args.rollout_batch_size, f"Got {len(data)} samples, expected {args.rollout_batch_size}" + data = sorted(data, key=lambda group: group[0][0].index if isinstance(group[0], list) else group[0].index) + all_samples = sorted( + all_data, key=lambda group: group[0][0].index if isinstance(group[0], list) else group[0].index + ) + + # reset the global state to prevent effects on the next rollout or eval. + state.reset() + + if f := load_function(args.rollout_sample_filter_path): + f(args, data) + # There can be circumstances where users want to process all samples including filtered ones. + if f := load_function(args.rollout_all_samples_process_path): + f(args, all_samples, data_source) + + await recompute_samples_rollout_logprobs_via_prefill( + args, + [sample for group in data for sample in group], + url=f"http://{args.sglang_router_ip}:{args.sglang_router_port}/generate", + sampling_params=state.sampling_params, + ) + + return RolloutFnTrainOutput(samples=data, metrics=metric_gatherer.collect()), aborted_samples + + +async def _generate_rollout_group_level_async( + state: GenerateState, + rollout_id: int, + data_source: Callable[[int], list[list[Sample]]], + dynamic_filter, + metric_gatherer: MetricGatherer, +) -> tuple[list[list[Sample]], list[list[Sample]], list[list[Sample]]]: + args = state.args + # target_data_size is the total number of valid samples to get target_data_size = args.rollout_batch_size @@ -138,26 +195,114 @@ async def generate_rollout_async( # there are still some unfinished requests, abort them aborted_samples = await abort(state, pendings, rollout_id) - assert len(data) == args.rollout_batch_size, f"Got {len(data)} samples, expected {args.rollout_batch_size}" - data = sorted(data, key=lambda group: group[0][0].index if isinstance(group[0], list) else group[0].index) - all_samples = sorted( - all_data, key=lambda group: group[0][0].index if isinstance(group[0], list) else group[0].index + return data, all_data, aborted_samples + + +async def _generate_rollout_sample_completion_backfill_async( + state: GenerateState, + rollout_id: int, + data_source: Callable[[int], list[list[Sample]]], + dynamic_filter, + metric_gatherer: MetricGatherer, +) -> tuple[list[list[Sample]], list[list[Sample]], list[list[Sample]]]: + args = state.args + target_data_size = args.rollout_batch_size + group_size = args.n_samples_per_prompt + sample_done_queue: asyncio.Queue[int] = asyncio.Queue() + accept_sample_done = True + + def on_sample_done() -> None: + if accept_sample_done and not state.aborted: + sample_done_queue.put_nowait(1) + + pendings = set() + data = [] + all_data = [] + do_print = True + sample_done_credit = 0 + pbar = tqdm(total=target_data_size * group_size, desc="Rollout generation") + + def submit_groups(num_groups: int) -> int: + if num_groups <= 0: + return 0 + samples = data_source(num_groups) + new_tasks = submit_generate_tasks(state, samples, sample_done_callback=on_sample_done) + pendings.update(new_tasks) + return len(new_tasks) + + logger.info( + "[rollout] sample-completion backfill enabled: target_groups=%s group_size=%s", + target_data_size, + group_size, ) + submit_groups(target_data_size) - # reset the global state to prevent effects on the next rollout or eval. - state.reset() + while len(data) < target_data_size: + if not pendings: + # Defensive fallback for group-level task exceptions. Normal flow keeps + # pending sample slots replenished from sample completion credits. + submit_groups(max(1, target_data_size - len(data))) + + sample_done_task = asyncio.create_task(sample_done_queue.get()) + done, _ = await asyncio.wait(pendings | {sample_done_task}, return_when=asyncio.FIRST_COMPLETED) + + sample_done_count = 0 + if sample_done_task in done: + sample_done_count += sample_done_task.result() + else: + sample_done_task.cancel() + with suppress(asyncio.CancelledError): + await sample_done_task + + while True: + try: + sample_done_count += sample_done_queue.get_nowait() + except asyncio.QueueEmpty: + break + sample_done_credit += sample_done_count - if f := load_function(args.rollout_sample_filter_path): - f(args, data) - # There can be circumstances where users want to process all samples including filtered ones. - if f := load_function(args.rollout_all_samples_process_path): - f(args, all_samples, data_source) + group_done = done & pendings + if group_done: + pendings.difference_update(group_done) - await recompute_samples_rollout_logprobs_via_prefill( - args, - [sample for group in data for sample in group], - url=f"http://{args.sglang_router_ip}:{args.sglang_router_port}/generate", - sampling_params=state.sampling_params, + for task in group_done: + try: + group: list[Sample] = task.result() + except Exception as e: + logger.error(f"[rollout] Task raised exception: {e!r}", exc_info=True) + continue + + if do_print: + sample = group[0][0] if isinstance(group[0], list) else group[0] + logger.info( + f"First rollout sample: {[str(sample.prompt) + sample.response]}, label: {sample.label}, reward: {sample.reward}", + ) + do_print = False + + assert len(group) == group_size + all_data.append(group) + dynamic_filter_output = call_dynamic_filter(dynamic_filter, args, group) + if not dynamic_filter_output.keep: + metric_gatherer.on_dynamic_filter_drop(reason=dynamic_filter_output.reason) + continue + + if len(data) < target_data_size: + data.append(group) + pbar.update(group_size) + + while sample_done_credit >= group_size and len(data) < target_data_size: + submitted = submit_groups(1) + if submitted <= 0: + break + sample_done_credit -= group_size + + pbar.close() + sample = data[-1][0][0] if isinstance(data[-1][0], list) else data[-1][0] + logger.info( + f"Finish rollout: {[str(sample.prompt) + sample.response]}, label: {sample.label}, reward: {sample.reward}", ) - return RolloutFnTrainOutput(samples=data, metrics=metric_gatherer.collect()), aborted_samples + accept_sample_done = False + aborted_samples = await abort(state, pendings, rollout_id) + + return data, all_data, aborted_samples diff --git a/miles/rollout/sglang_rollout.py b/miles/rollout/sglang_rollout.py index 64662ed522..ebf1b1275c 100644 --- a/miles/rollout/sglang_rollout.py +++ b/miles/rollout/sglang_rollout.py @@ -306,7 +306,11 @@ async def generate_and_rm( async def generate_and_rm_group( - args: Namespace, group: list[Sample], sampling_params: dict[str, Any], evaluation: bool = False + args: Namespace, + group: list[Sample], + sampling_params: dict[str, Any], + evaluation: bool = False, + sample_done_callback: Callable[[], None] | None = None, ) -> list[Sample]: state = GenerateState(args) @@ -325,9 +329,10 @@ async def generate_and_rm_group( if getattr(args, "sglang_enable_deterministic_inference", False): seed = state.group_sampling_seeds[idx] current_sampling_params["sampling_seed"] = seed - tasks.append( - asyncio.create_task(generate_and_rm(args, sample, current_sampling_params, evaluation=evaluation)) - ) + task = asyncio.create_task(generate_and_rm(args, sample, current_sampling_params, evaluation=evaluation)) + if sample_done_callback is not None: + task.add_done_callback(lambda _task: sample_done_callback()) + tasks.append(task) group = await asyncio.gather(*tasks) diff --git a/miles/utils/arguments.py b/miles/utils/arguments.py index 53dcc34f8e..ea66a1ef5b 100644 --- a/miles/utils/arguments.py +++ b/miles/utils/arguments.py @@ -458,6 +458,17 @@ def add_rollout_arguments(parser): "You could use `miles.rollout.filter_hub.dynamic_sampling_filters.check_reward_nonzero_std` as an example." ), ) + parser.add_argument( + "--rollout-sample-completion-backfill", + action="store_true", + default=False, + help=( + "If set, training rollout replenishes one new prompt group after " + "n_samples_per_prompt individual samples complete, instead of waiting " + "for a full group task to return. Disabled by default to preserve the " + "legacy group-level scheduling behavior." + ), + ) # partial rollout parser.add_argument( diff --git a/tests/fast/rollout/inference_rollout/test_sample_completion_backfill.py b/tests/fast/rollout/inference_rollout/test_sample_completion_backfill.py new file mode 100644 index 0000000000..15e8ee9304 --- /dev/null +++ b/tests/fast/rollout/inference_rollout/test_sample_completion_backfill.py @@ -0,0 +1,155 @@ +from tests.ci.ci_register import register_cpu_ci + +register_cpu_ci(est_time=60, suite="stage-a-cpu", labels=[]) + +import asyncio +import sys +from argparse import Namespace +from types import ModuleType + +import pytest + +if "sglang_router" not in sys.modules: + sglang_router_stub = ModuleType("sglang_router") + sglang_router_stub.__version__ = "0.0.0" + sys.modules["sglang_router"] = sglang_router_stub + +import miles.rollout.inference_rollout.inference_rollout_train as train +from miles.rollout.filter_hub.base_types import MetricGatherer +from miles.utils.types import Sample + + +class FakeGenerateState: + def __init__(self, args: Namespace): + self.args = args + self.sampling_params = {} + self.aborted = False + self.reset_count = 0 + + def reset(self) -> None: + self.aborted = False + self.reset_count += 1 + + +def make_group(group_index: int, group_size: int) -> list[Sample]: + return [ + Sample( + group_index=group_index, + index=group_index * 100 + sample_index, + prompt=f"prompt {group_index}", + response="ok", + response_length=1, + label="ok", + reward=1, + status=Sample.Status.COMPLETED, + ) + for sample_index in range(group_size) + ] + + +@pytest.mark.asyncio +async def test_generate_rollout_without_backfill_flag_uses_legacy_group_scheduler(monkeypatch): + group = make_group(group_index=1, group_size=2) + args = Namespace( + rollout_global_dataset=True, + rollout_batch_size=1, + rollout_sample_filter_path=None, + rollout_all_samples_process_path=None, + dynamic_sampling_filter_path=None, + sglang_router_ip="127.0.0.1", + sglang_router_port=30000, + ) + state = FakeGenerateState(args) + called = [] + + async def noop_configure_sglang(_args): + return None + + async def noop_recompute(*_args, **_kwargs): + return None + + async def fake_group_level(*_args, **_kwargs): + called.append("group_level") + return [group], [group], [] + + async def unexpected_sample_completion_backfill(*_args, **_kwargs): + raise AssertionError("sample-completion backfill should be disabled by default") + + monkeypatch.setattr(train.dumper_utils, "configure_sglang", noop_configure_sglang) + monkeypatch.setattr(train, "recompute_samples_rollout_logprobs_via_prefill", noop_recompute) + monkeypatch.setattr(train, "load_function", lambda _path: None) + monkeypatch.setattr(train, "_generate_rollout_group_level_async", fake_group_level) + monkeypatch.setattr( + train, + "_generate_rollout_sample_completion_backfill_async", + unexpected_sample_completion_backfill, + ) + + output, aborted_samples = await train.generate_rollout_async(state, rollout_id=0, data_source=lambda _n: []) + + assert called == ["group_level"] + assert output.samples == [group] + assert aborted_samples == [] + assert state.reset_count == 1 + + +@pytest.mark.asyncio +async def test_sample_completion_backfill_submits_group_after_enough_samples_finish(monkeypatch): + args = Namespace(rollout_batch_size=1, n_samples_per_prompt=2) + state = FakeGenerateState(args) + data_source_calls = [] + submitted_group_indices = [] + next_group_index = 0 + + def data_source(num_groups: int) -> list[list[Sample]]: + nonlocal next_group_index + data_source_calls.append(num_groups) + groups = [] + for _ in range(num_groups): + next_group_index += 1 + groups.append(make_group(group_index=next_group_index, group_size=args.n_samples_per_prompt)) + return groups + + async def never_complete(): + await asyncio.Future() + + async def complete_group(group: list[Sample]) -> list[Sample]: + await asyncio.sleep(0) + return group + + def fake_submit_generate_tasks(_state, samples, sample_done_callback=None): + tasks = [] + for group in samples: + submitted_group_indices.append(group[0].group_index) + if len(submitted_group_indices) == 1: + assert sample_done_callback is not None + for _ in group: + sample_done_callback() + tasks.append(asyncio.create_task(never_complete())) + else: + tasks.append(asyncio.create_task(complete_group(group))) + return tasks + + async def fake_abort(_state, pendings, _rollout_id): + _state.aborted = True + for task in pendings: + task.cancel() + await asyncio.gather(*pendings, return_exceptions=True) + return [] + + monkeypatch.setattr(train, "submit_generate_tasks", fake_submit_generate_tasks) + monkeypatch.setattr(train, "abort", fake_abort) + + data, all_data, aborted_samples = await train._generate_rollout_sample_completion_backfill_async( + state, + rollout_id=0, + data_source=data_source, + dynamic_filter=None, + metric_gatherer=MetricGatherer(), + ) + + assert data_source_calls == [1, 1] + assert submitted_group_indices == [1, 2] + assert data == [make_group(group_index=2, group_size=args.n_samples_per_prompt)] + assert all_data == data + assert aborted_samples == [] From f814b44ba0b37fe1723acdb7cb6da75206468850 Mon Sep 17 00:00:00 2001 From: Shang Yang Date: Tue, 14 Jul 2026 21:00:02 +0000 Subject: [PATCH 2/3] Stop backfill loop when data source is exhausted If the data source returns no groups while nothing is in flight, break out of the collection loop instead of blocking forever on the sample-completion queue (review feedback). Also document the credit-conservation invariant at the sample_done_callback call sites: the callback deliberately fires for every sample task regardless of outcome, so completion credits exactly conserve in-flight sample concurrency -- conditioning on success would leak scheduling credits. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_013vyY58s2TFpCzqEP4KaSxC --- .../rollout/inference_rollout/inference_rollout_common.py | 5 +++++ miles/rollout/inference_rollout/inference_rollout_train.py | 7 +++++-- miles/rollout/sglang_rollout.py | 3 +++ 3 files changed, 13 insertions(+), 2 deletions(-) diff --git a/miles/rollout/inference_rollout/inference_rollout_common.py b/miles/rollout/inference_rollout/inference_rollout_common.py index 2603693677..9e957168d6 100644 --- a/miles/rollout/inference_rollout/inference_rollout_common.py +++ b/miles/rollout/inference_rollout/inference_rollout_common.py @@ -139,6 +139,11 @@ async def generate_and_rm_group( current_sampling_params["sampling_seed"] = args.rollout_seed + idx task = asyncio.create_task(generate_and_rm(state, sample, current_sampling_params, evaluation=evaluation)) if sample_done_callback is not None: + # Fires for every sample task regardless of outcome (success, exception, + # or cancellation): each submitted sample yields exactly one completion + # credit, so in-flight sample concurrency is conserved exactly. Do not + # condition this on task success -- skipping failed samples would leak + # scheduling credits and decay concurrency over time. task.add_done_callback(lambda _task: sample_done_callback()) tasks.append(task) diff --git a/miles/rollout/inference_rollout/inference_rollout_train.py b/miles/rollout/inference_rollout/inference_rollout_train.py index 23d89f9065..95e2139326 100644 --- a/miles/rollout/inference_rollout/inference_rollout_train.py +++ b/miles/rollout/inference_rollout/inference_rollout_train.py @@ -240,8 +240,11 @@ def submit_groups(num_groups: int) -> int: while len(data) < target_data_size: if not pendings: # Defensive fallback for group-level task exceptions. Normal flow keeps - # pending sample slots replenished from sample completion credits. - submit_groups(max(1, target_data_size - len(data))) + # pending sample slots replenished from sample completion credits. If the + # data source is exhausted and nothing is in flight, stop instead of + # blocking forever on the sample-completion queue. + if submit_groups(max(1, target_data_size - len(data))) == 0: + break sample_done_task = asyncio.create_task(sample_done_queue.get()) done, _ = await asyncio.wait(pendings | {sample_done_task}, return_when=asyncio.FIRST_COMPLETED) diff --git a/miles/rollout/sglang_rollout.py b/miles/rollout/sglang_rollout.py index ebf1b1275c..e55161a65e 100644 --- a/miles/rollout/sglang_rollout.py +++ b/miles/rollout/sglang_rollout.py @@ -331,6 +331,9 @@ async def generate_and_rm_group( current_sampling_params["sampling_seed"] = seed task = asyncio.create_task(generate_and_rm(args, sample, current_sampling_params, evaluation=evaluation)) if sample_done_callback is not None: + # Fires for every sample task regardless of outcome; see the note in + # inference_rollout_common.generate_and_rm_group -- credits must conserve + # in-flight sample concurrency exactly. task.add_done_callback(lambda _task: sample_done_callback()) tasks.append(task) From c41b93c1edb369c060c16bef3724de12a1546bfe Mon Sep 17 00:00:00 2001 From: Shang Yang Date: Tue, 14 Jul 2026 22:40:02 +0000 Subject: [PATCH 3/3] Apply pre-commit formatting Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_013vyY58s2TFpCzqEP4KaSxC --- examples/fully_async/fully_async_rollout.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/examples/fully_async/fully_async_rollout.py b/examples/fully_async/fully_async_rollout.py index 8874daa78a..d515fc6183 100644 --- a/examples/fully_async/fully_async_rollout.py +++ b/examples/fully_async/fully_async_rollout.py @@ -165,9 +165,7 @@ def task_done_callback(done_task): sample_backfill_initialized = len(active_tasks) >= max_concurrent_tasks while ( - sample_backfill_initialized - and completed_sample_credits >= samples_per_group - and self.running + sample_backfill_initialized and completed_sample_credits >= samples_per_group and self.running ): if not submit_one_group(): break