[Feature] Add refresh rate for ClusterExecutors

[jan-janssen]

Summary by CodeRabbit

• New Features
  • Added configurable refresh rate parameter across all executor backends including Flux, single-node, Slurm, and file-based executors. Users can now control how frequently the system checks for task status updates. Default refresh interval is 0.01 seconds. This provides fine-grained control over resource utilization and responsiveness during task execution.

[coderabbitai]

📝 Walkthrough

Walkthrough

This pull request adds a refresh_rate parameter throughout the executor system to control the frequency of queue and memory refresh operations. The parameter is forwarded from high-level executors (Flux, SingleNode, Slurm) through to low-level file-based execution functions, with a default value of 0.01 seconds and a throttling mechanism at the memory refresh layer.

Changes

Cohort / File(s)	Summary
High-level Executors src/executorlib/executor/flux.py, src/executorlib/executor/single.py, src/executorlib/executor/slurm.py	Forward refresh_rate parameter to underlying file executor creation functions. Each executor now passes this parameter through the non-blocking/file-based execution path.
File Task Scheduler src/executorlib/task_scheduler/file/task_scheduler.py	Added refresh_rate parameter (default 0.01) to FileTaskScheduler.__init__ and create_file_executor function. Parameter is stored and forwarded to execution processes.
Low-level Execution Logic src/executorlib/task_scheduler/file/shared.py	Extended execute_tasks_h5 and _refresh_memory_dict with refresh_rate parameter. Implements sleep-based throttling in memory refresh operations to respect the specified refresh cadence.

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~20 minutes

Possibly related PRs

• Fix: Implement one _refresh_memory_dict() function #902: Modifies _refresh_memory_dict and memory refresh call sites, directly related to the refresh mechanism enhancements.
• Upgrade terminate function #715: Modifies FileTaskScheduler constructor and create_file_executor API signatures, affecting the same public interfaces.

Poem

🐰 A whisker-twitch brings refresh rates near,
Each executor passes parameters clear,
With sleep and throttle, smooth as a hop,
The queues refresh without a stop!
✨

🚥 Pre-merge checks | ✅ 4

✅ Passed checks (4 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title accurately describes the main feature being added: a refresh rate parameter for ClusterExecutors, which is consistently implemented across flux.py, single.py, slurm.py and the underlying task scheduler modules.
Docstring Coverage	✅ Passed	Docstring coverage is 88.89% which is sufficient. The required threshold is 80.00%.
Merge Conflict Detection	✅ Passed	✅ No merge conflicts detected when merging into main

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing touches

• 📝 Generate docstrings

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment
• Commit unit tests in branch file_refresh_rate

---

No actionable comments were generated in the recent review. 🎉

🧹 Recent nitpick comments

src/executorlib/task_scheduler/file/shared.py (1)

333-344: Throttling only kicks in when no tasks complete, which introduces a one-cycle delay.

The not value.done() filter in the comprehension runs before _check_task_output sets the result on the future. So a task that completes during this refresh cycle is still included in memory_updated_dict, making len(memory_updated_dict) == len(memory_dict) true, and triggering a sleep(refresh_rate) even though progress was made. The completed future is only pruned in the next call.

With the default 0.01s this is negligible, but worth being aware of if users set a higher refresh rate.

♻️ Optional fix: check for newly-done futures after the comprehension

     memory_updated_dict = {
         key: _check_task_output(
             task_key=key,
             future_obj=value,
             cache_directory=cache_dir_dict[key],
         )
         for key, value in memory_dict.items()
         if not value.done()
     }
-    if len(memory_updated_dict) == len(memory_dict):
+    memory_updated_dict = {
+        key: value
+        for key, value in memory_updated_dict.items()
+        if not value.done()
+    }
+    if len(memory_updated_dict) == len(memory_dict):
         sleep(refresh_rate)
     return memory_updated_dict

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 93.79%. Comparing base (3fd4fb3) to head (6468774).
⚠️ Report is 2 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #919      +/-   ##
==========================================
+ Coverage   93.78%   93.79%   +0.01%     
==========================================
  Files          38       38              
  Lines        1947     1951       +4     
==========================================
+ Hits         1826     1830       +4     
  Misses        121      121              

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.