[davidberenstein1957]

Description

PeriodicScheduler used to chain a fresh threading.Timer for each tick and armed the next timer before running the payload. This PR replaces that with a single daemon thread that waits on an Event until an absolute time.monotonic() deadline.

• Cadence: ticks don't drift. A call that overruns its slot skips the missed slots and waits for the next one. Ticks never pile up and never run back-to-back.
• Errors: exceptions are logged inside the loop, so a failing measurement does not kill the thread.
• stop(): sets the event, then joins the worker for up to min(interval, 5 s). If the join times out, it keeps the thread reference and logs a warning.
• start(): a no-op while the scheduler is running. After a stop it starts a new loop right away and passes it the previous thread. If that thread is still finishing a call, the new loop waits for it before its first tick. The function therefore never runs concurrently, and the restart is never lost. This also holds when stop() is called from inside the callback.
• Locking: start() and stop() are serialised by a lock.
• Clock: time is read through a clock attribute (MonotonicClock). Tests replace it with a fake clock to drive the loop without sleeping.
• Tracker: _measure_power_and_energy is serialised by an RLock. The final measurement in tracker.stop() therefore waits for an in-flight scheduled one, even when that tick takes longer than the join timeout.

Related Issue

Part of #1338

Motivation and Context

With chained Timers, a measurement slower than the tick interval re-entered the payload. Two concurrent calls then mutated _total_energy / _last_measured_time at the same time and corrupted the reported numbers. This happens routinely for _scheduler_monitor_power (1 s, hard-coded) when powermetrics or RAPL is slow. There was also a stop() race in which measurements and API pushes could continue after the tracker had written its final row.

Behaviour changes worth flagging:

• stop() can block. PeriodicScheduler.stop() waits up to min(measure_power_secs, 5 s) for the call in progress; in the normal case it takes about 0.1 ms. tracker.stop() can wait longer: its final measurement now waits for an in-flight tick to finish, for example a slow Electricity Maps lookup. start_task() calls _scheduler.stop(), so it can block too.
• Missing measurements instead of overlapping ones. A slow callback now produces missing measurements rather than overlapping ones. The existing "Background scheduler didn't run for a long period" warning can fire where master used to produce corrupted concurrent measurements.
• from_run removed. The from_run parameter of start() was removed; only the removed _run used it.

How Has This Been Tested?

• tests/test_scheduler.py: 11 tests, driven by a fake clock (tests/testutils.py::FakeClock), so they assert exact tick times instead of sleeping. They cover:
  • a single worker thread
  • no re-entry
  • overrun skipping with a pause before the next call
  • no drift
  • prompt and final stop()
  • idempotent start() / restart
  • restart after a timed-out stop, with no overlap and no lost restart
  • stop() from inside the callback
  • concurrent start() calls
  • exceptions do not kill the loop
• tests/test_emissions_tracker_constant.py::TestFinalMeasurementWaitsForInFlightTick: tracker.stop() does not overlap a scheduled measurement that outlasts the join timeout.
• Every test added during review fails when its fix is reverted (checked by reverting each fix individually).
• The scheduler tests passed 20 runs in a row.
• Full suite with an empty HOME, as in CI (uv run task test-coverage): 818 passed, 3 skipped, 0 failed.
• Patch coverage: 58/58 changed lines in codecarbon/ (100 %).
• pre-commit passes.

Screenshots (if appropriate):

N/A

Types of changes

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to change)

AI Usage Disclosure

• 🟥 AI-vibecoded
• 🟠 AI-generated
• ⭐ AI-assisted
• ♻️ No AI used

Checklist:

• My code follows the code style of this project.
• My change requires a change to the documentation.
• I have updated the documentation accordingly.
• I have read the docs/how-to/contributing.md document.
• I have added tests to cover my changes.
• All new and existing tests passed.

This should merge before #1340, which builds on this scheduler. #1324 fixes the same defect with a narrower change; close it if this PR is merged.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 93.66%. Comparing base (7bd15e2) to head (b2254b6).

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #1339      +/-   ##
==========================================
+ Coverage   93.62%   93.66%   +0.03%     
==========================================
  Files          50       50              
  Lines        5556     5591      +35     
==========================================
+ Hits         5202     5237      +35     
  Misses        354      354              

☔ View full report in Codecov by Harness.
📢 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.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[davidberenstein1957]

Note for whoever reviews this: #1324 fixes the same scheduler defect by a different route. It is a narrow fix to the existing chained-Timer design, while this PR replaces the design with a single daemon thread. Worth picking one before reviewing either, otherwise they conflict on merge.

[benoit-cty]

🤖 This review comment was written and posted by Claude Opus 5.5 (AI assistant), at the request of @benoit-cty. Findings were checked by reading the code and running tests locally (merged with current master where relevant), but please double-check before acting on them.

Verdict: ✅ Approve with nits

Replacing the chained Timers with one daemon thread on a monotonic deadline is a real improvement:

• no drift
• overrunning ticks are skipped rather than overlapping
• exceptions are logged inside the loop
• the final measure and flush always still run

The 7 new scheduler tests pass, and the PR merges cleanly with master.

Issues:

1. The "wedged-thread handling" in the description isn't implemented (medium).
   • The description says a timed-out stop() keeps its thread reference, logs a warning, and start() then refuses to run a second loop.
   • The code (scheduler.py:69) always sets _thread = None, logs nothing, and uses one Event per run.
   • So after a stop() that timed out (e.g. via start_task(), emissions_tracker.py:754), start() launches a new loop while the old callback is still running. That gives two concurrent _measure_power_and_energy calls, which is exactly the overlap this PR wants to remove.
   • test_start_after_a_timed_out_stop_leaves_only_the_new_loop explicitly accepts that overlap.
   • Fix: if join() times out, keep the reference and log a warning. In start(), if the previous thread is still alive, wait for it (or refuse) before starting a new loop. Update the test.
2. The final flush can still overlap an in-flight tick, silently (medium).
   • stop() waits at most min(interval, 5 s), but a tick can make network calls: Electricity Maps (30 s timeout), and the API (up to about 26 s with feat(api): reuse connections, add timeouts, and stamp emissions with measurement time #1340).
   • If one is in flight, tracker.stop() (emissions_tracker.py:909-930) runs the final measurement and _persist_data concurrently with it.
   • At minimum, log a warning when the join times out. Better, make the final measurement wait for the in-flight tick, e.g. with a lock around the tick body.

Nits:

• Pre-existing: start_task never restarts _scheduler.
• The tests rely on real sleeps and may flake on shared CI runners. Consider an injectable clock/event.

[davidberenstein1957]

Made the changes in 61bc10d: timed-out stop() now warns and blocks a second loop.

Not done: making the final flush wait on the in-flight tick (needs a lock shared between scheduler and tracker, bigger change); the pre-existing start_task restart and the real-sleep tests are left as is.

[benoit-cty]

Commit	Problem it fixes
fix(scheduler): resume after a timed-out stop instead of refusing to start	After a stop() whose join timed out, start() blocked for up to min(interval, 5 s). If the old callback was still running, start() then logged a warning and returned, and nothing ever retried it. A start_task() / stop_task() pair during a slow measurement therefore turned periodic measurement off for the rest of the run, because _scheduler_paused_by_task had already been cleared. Now start() always arms the new loop and passes it the previous thread. The new loop joins that thread before its first tick, so start() never blocks, two loops still never run at the same time, and the restart always happens.
fix(scheduler): keep the thread reference when stop() is called from the callback	When stop() was called from inside the scheduled function, it set _thread = None while that function was still running. A later start() then ran a second loop alongside it.
fix(scheduler): skip overrun slots instead of calling again immediately	next_call = max(next_call + interval, now) set the next deadline to now after an overrun. A function slower than its interval, such as a 1.2 s GPU read in the 1 s _monitor_power scheduler, then ran back-to-back at 100 % duty cycle. The class docstring says those ticks are skipped. The deadline now advances by whole intervals, so the cadence stays on its fixed grid.
fix(scheduler): guard start() and stop() with a lock again	The old scheduler serialised start() and stop() with a Lock, and the refactor removed it. Two threads calling start() together could each arm a loop. The first loop's Event was then overwritten, so stop() could never end that loop. The lock only covers the state changes; stop() still joins outside it.
fix(tracker): make the final measurement wait for an in-flight scheduled one	This is the "known limit" from the description. A threading.RLock around _measure_power_and_energy and the stale check means tracker.stop() runs its final measurement and _persist_data only after the in-flight tick has finished. Previously both could update _total_energy / _last_measured_time concurrently when a tick took longer than the join timeout, for example an Electricity Maps call (30 s timeout) or an API push. It is about ten lines in the tracker and the scheduler needs no changes.
refactor(scheduler): replace the stop() timeout parameter with a module constant	Nothing passed stop(timeout=...), and _join_timeout(None) only computed min(interval, 5.0). The cap is now STOP_JOIN_TIMEOUT_SECONDS, so it has a name and tests can lower it.
test(scheduler): drive the scheduler tests with an injectable fake clock	The tests slept on real time with a 50 ms interval and asserted tick counts, so a loaded CI runner could fail them. PeriodicScheduler now reads time and waits through a clock attribute (MonotonicClock by default). Tests replace it with FakeClock from tests/testutils.py, which only moves when the test advances it, so they assert exact tick times. A new test checks that the cadence does not drift. The tracker test now synchronises on the final measurement instead of a 1.5 s Timer. The 12 tests take 1.6 s instead of about 8 s.

About the clock: **kwargs are passed through to the scheduled function, so a clock= constructor argument would take that name away from it. That's why the clock is an attribute you set before start(). The only real-time waits left in the tests are 100 ms upper bounds on things that must not happen, such as the new loop ticking while the old call is still running. A correct scheduler passes them however slow the machine is.