From c43a15db8e6142cbeb3f696efe89facced166c49 Mon Sep 17 00:00:00 2001 From: Sangyoon Park Date: Mon, 17 Aug 2026 14:07:01 -0400 Subject: [PATCH] Stage new files after freeing capacity in tracking cycle Files that finished processing released their capacity only after `_stage_new_files` had already run, so a waiting file sat idle for a full cycle before taking the freed slot. - Move `_stage_new_files` after the failure and completion handlers, and keep `_check_inactivity` last so a cycle that just staged work is not counted as idle - Move the idle-clock announcement into `_check_inactivity`, where the settled state is now final, and guard it so it logs once per settling --- src/tigerflow/pipeline.py | 18 ++++++---- tests/integration/pipeline/conftest.py | 20 +++++++++++ .../integration/pipeline/test_file_staging.py | 33 ++++++++++++++++++- tests/integration/pipeline/test_lifecycle.py | 28 ++++++++++++++++ 4 files changed, 92 insertions(+), 7 deletions(-) diff --git a/src/tigerflow/pipeline.py b/src/tigerflow/pipeline.py index 5a6a554..5cd4693 100644 --- a/src/tigerflow/pipeline.py +++ b/src/tigerflow/pipeline.py @@ -60,6 +60,7 @@ def __init__( self._idle_timeout = timedelta(minutes=idle_timeout) self._last_active = datetime.now() + self._idle_announced = False self._delete_input = delete_input @@ -216,14 +217,17 @@ def _run_tracking_cycle(self): """Run one iteration of the pipeline tracking loop. `_handle_task_timeout` reads the task status that `_check_task_status` - refreshes, so it must not run before it; a stale status resubmits a - Slurm job that has already been replaced. + refreshes, so a stale status would resubmit an already replaced Slurm job. + `_stage_new_files` follows `_report_failed_files` and `_handle_processed_files` + so that slots those free up can be filled in the same cycle. `_check_inactivity` + runs last because staging adds to `_filenames`, and a cycle that just started + work must not be counted as idle. """ self._check_task_status() self._handle_task_timeout() - self._stage_new_files() self._report_failed_files() self._handle_processed_files() + self._stage_new_files() self._check_inactivity() def _start_tasks(self): @@ -473,12 +477,14 @@ def _handle_processed_files(self): # Log progress if completed_file_ids: logger.info("Completed processing {} files", len(completed_file_ids)) - if self._all_tracked_files_settled(): - logger.info("No more files to process, starting idle time count") def _check_inactivity(self): - if not self._all_tracked_files_settled(): + settled = self._all_tracked_files_settled() + if not settled: self._last_active = datetime.now() + elif not self._idle_announced: + logger.info("No more files to process, starting idle time count") + self._idle_announced = settled inactivity = datetime.now() - self._last_active if inactivity > self._idle_timeout: diff --git a/tests/integration/pipeline/conftest.py b/tests/integration/pipeline/conftest.py index 0432e38..7f098de 100644 --- a/tests/integration/pipeline/conftest.py +++ b/tests/integration/pipeline/conftest.py @@ -137,6 +137,26 @@ def error_logs() -> Iterator[list[str]]: logger.remove(sink_id) +@pytest.fixture +def idle_start_logs() -> Iterator[list[str]]: + """Yield a list collecting the message of every idle-clock-start record logged. + + Filtered by message content as well as level: the tracking cycle logs other + INFO records that a level-only sink would mix in and throw off exact counts. + """ + records: list[str] = [] + + sink_id = logger.add( + lambda message: records.append(message.record["message"]), + level="INFO", + filter=lambda record: "starting idle time count" in record["message"], + ) + try: + yield records + finally: + logger.remove(sink_id) + + @pytest.fixture def pending_worker_logs() -> Iterator[list[str]]: """Yield a list collecting the message of every "Workers pending" record logged. diff --git a/tests/integration/pipeline/test_file_staging.py b/tests/integration/pipeline/test_file_staging.py index 4ce1406..3273fcb 100644 --- a/tests/integration/pipeline/test_file_staging.py +++ b/tests/integration/pipeline/test_file_staging.py @@ -13,7 +13,7 @@ from tigerflow.pipeline import Pipeline from tigerflow.staging import StagingContext -from .helpers import PipelineFactory, task_spec +from .helpers import PipelineFactory, start_fake_tasks, task_spec, write_output DUPLICATING_STEP = { "kind": "callable", @@ -216,6 +216,37 @@ def test_max_staged_respects_capacity( "Capacity is already full, so no further file should be staged" ) + def test_completion_frees_capacity_within_the_same_cycle( + self, pipeline_factory: PipelineFactory, input_dir: Path + ): + """A file completing lets a waiting file take its slot in that same cycle. + + Which two files stage first is left to directory order, so the assertions + pin the count and the swap rather than specific names. + """ + pipeline = pipeline_factory( + staging={"steps": [{"kind": "max_staged", "count": 2}]} + ) + start_fake_tasks(pipeline) + for i in range(3): + (input_dir / f"f{i}.txt").write_text("x") + + pipeline._run_tracking_cycle() + first_batch = {f.name for f in pipeline._symlinks_dir.iterdir()} + assert len(first_batch) == 2 + + completed = sorted(first_batch)[0] + write_output(pipeline, Path(completed).stem) + pipeline._run_tracking_cycle() + + staged = {f.name for f in pipeline._symlinks_dir.iterdir()} + assert len(staged) == 2, ( + "The completed file frees a slot that the waiting file should take " + "in the same cycle" + ) + assert completed not in staged + assert staged - first_batch, "The waiting file should now be staged" + def test_counts_stay_consistent_in_mixed_state( self, pipeline_factory: PipelineFactory, input_dir: Path ): diff --git a/tests/integration/pipeline/test_lifecycle.py b/tests/integration/pipeline/test_lifecycle.py index 82b615e..bb6b304 100644 --- a/tests/integration/pipeline/test_lifecycle.py +++ b/tests/integration/pipeline/test_lifecycle.py @@ -51,6 +51,34 @@ def test_stages_then_completes_across_cycles( assert (pipeline._finished_dir / "a.txt").exists() assert not (pipeline._symlinks_dir / "a.txt").exists() + def test_idle_start_logged_once_per_settling( + self, + pipeline_factory: PipelineFactory, + input_dir: Path, + idle_start_logs: list[str], + ): + """The idle clock is announced when work runs out, then stays quiet.""" + pipeline = pipeline_factory() + start_fake_tasks(pipeline) + + (input_dir / "a.txt").write_text("payload") + pipeline._run_tracking_cycle() + assert idle_start_logs == [] + + write_output(pipeline, "a") + pipeline._run_tracking_cycle() + assert len(idle_start_logs) == 1 + + pipeline._run_tracking_cycle() + assert len(idle_start_logs) == 1 + + # A new file makes the pipeline active again, so settling announces afresh + (input_dir / "b.txt").write_text("payload") + pipeline._run_tracking_cycle() + write_output(pipeline, "b") + pipeline._run_tracking_cycle() + assert len(idle_start_logs) == 2 + def test_file_staged_only_once( self, pipeline_factory: PipelineFactory, input_dir: Path ):