From 6459139533f55dea764eb9a8cd036d01aa11683f Mon Sep 17 00:00:00 2001
From: David Berenstein <david.m.berenstein@gmail.com>
Date: Wed, 19 Aug 2026 16:19:35 +0200
Subject: [PATCH 1/9] refactor(scheduler): run ticks on one daemon thread
 instead of chained Timers

Every tick spawned a fresh threading.Timer and the next one was armed
before the callback ran, so a callback slower than the interval overlapped
with itself and the thread count grew with the run.

Run the loop on a single daemon thread waiting on an Event, with an
absolute deadline so the cadence does not drift and an overrun skips ahead
instead of firing catch-up ticks. Each run owns its Event, so a thread left
behind by a timed-out stop() keeps its own set event and exits after its
callback returns, while start() takes effect immediately on a new thread.

Note: stop() now blocks up to min(interval, 5.0) waiting for the in-flight
call. start_task() calls _scheduler.stop() (emissions_tracker.py:755), so
start_task() becomes potentially multi-second where it used to be instant.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 codecarbon/external/scheduler.py |  88 ++++++++++++++--------
 tests/test_scheduler.py          | 125 +++++++++++++++++++++++++++++++
 2 files changed, 180 insertions(+), 33 deletions(-)
 create mode 100644 tests/test_scheduler.py

diff --git a/codecarbon/external/scheduler.py b/codecarbon/external/scheduler.py
index b60aaad4d..86b5305e4 100644
--- a/codecarbon/external/scheduler.py
+++ b/codecarbon/external/scheduler.py
@@ -1,52 +1,74 @@
-from threading import Lock, Timer
+import time
+from threading import Event, Thread, current_thread
+
+from codecarbon.external.logger import logger
 
 
 class PeriodicScheduler:
     """
-    A periodic task running in threading.Timers
-    From https://stackoverflow.com/a/18906292/14541668
+    Run ``function`` every ``interval`` seconds on a single daemon thread.
+
+    The deadline is absolute, so the cadence does not drift with the time the
+    function itself takes. A tick that overruns its slot is skipped rather than
+    queued, so the function is never re-entered.
     """
 
     def __init__(self, interval, function, *args, **kwargs):
         """
-        Init the scheduler. You have to call start() after initialization.
-        ::interval:: interval in seconds to run the function.
-        ::function:: function to run.
+        ::interval:: in seconds, the delay between two calls to function.
+        ::function:: the function to call.
         ::args:: args to pass to the function.
         ::kwargs:: kwargs to pass to the function.
         """
-        self._lock = Lock()
-        self._timer = None
-        self.function = function
         self.interval = interval
+        self.function = function
         self.args = args
         self.kwargs = kwargs
-        self._stopped = True
+        self._thread = None
+        self._stop_event = None
 
-    def start(self, from_run=False):
+    @property
+    def _stopped(self):
+        return self._thread is None
+
+    def start(self):
         """
-        Start the scheduler.
-        ::from_run:: For internal purposes to allow re-scheduling
-        Please do not use from_run=True until you know what you do !
+        Start the scheduler. Calling it on a running scheduler is a no-op.
         """
-        self._lock.acquire()
-        if from_run or self._stopped:
-            self._stopped = False
-            self._timer = Timer(self.interval, self._run)
-            self._timer.daemon = True
-            self._timer.start()
-        self._lock.release()
-
-    def _run(self):
-        self.start(from_run=True)
-        self.function(*self.args, **self.kwargs)
-
-    def stop(self):
+        if self._thread is not None:
+            return
+        # One event per run: a thread left behind by a timed-out stop() keeps
+        # its own, still-set event and exits as soon as its callback returns.
+        self._stop_event = Event()
+        self._thread = Thread(
+            target=self._loop,
+            args=(self._stop_event,),
+            daemon=True,
+            name=f"codecarbon-{getattr(self.function, '__name__', 'scheduler')}",
+        )
+        self._thread.start()
+
+    def _loop(self, stop_event):
+        next_call = time.monotonic() + self.interval
+        while not stop_event.wait(max(0.0, next_call - time.monotonic())):
+            try:
+                self.function(*self.args, **self.kwargs)
+            except Exception:  # noqa: BLE001 - must not kill the only thread
+                logger.error("Scheduled measurement failed", exc_info=True)
+            # Absolute deadline, but never a burst of catch-up ticks.
+            next_call = max(next_call + self.interval, time.monotonic())
+
+    def stop(self, timeout=None):
         """
-        Stop the scheduler.
+        Stop the scheduler and wait for the in-flight call to return.
+        ::timeout:: seconds to wait for the running function, bounded by
+        default so a wedged measurement cannot hang the caller for long.
         """
-        if not self._stopped:
-            self._lock.acquire()
-            self._stopped = True
-            self._timer.cancel()
-            self._lock.release()
+        if self._stop_event is not None:
+            self._stop_event.set()
+        thread, self._thread = self._thread, None
+        if thread is not None and thread is not current_thread():
+            # ponytail: 5s cap is a guess at "a measurement should never take
+            # longer than this"; make it configurable if a slow hardware
+            # backend ever needs more.
+            thread.join(min(self.interval, 5.0) if timeout is None else timeout)
diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py
new file mode 100644
index 000000000..ae2fbf11a
--- /dev/null
+++ b/tests/test_scheduler.py
@@ -0,0 +1,125 @@
+import threading
+import time
+import unittest
+
+from codecarbon.external.scheduler import PeriodicScheduler
+
+INTERVAL = 0.05
+
+
+class TestPeriodicScheduler(unittest.TestCase):
+    def test_ticks_run_on_a_single_thread(self):
+        """Regression guard: one long-lived thread, not one thread per tick."""
+        names = set()
+        scheduler = PeriodicScheduler(
+            INTERVAL, lambda: names.add(threading.current_thread().name)
+        )
+        scheduler.start()
+        time.sleep(INTERVAL * 10)
+        scheduler.stop()
+        # A one-element set also proves at least one tick happened.
+        self.assertEqual(len(names), 1, f"expected one worker thread, got {names}")
+
+    def test_slow_function_is_never_re_entered(self):
+        """A function slower than the interval must not overlap with itself."""
+        state = {"in_flight": 0, "overlaps": 0, "calls": 0}
+
+        def slow():
+            state["calls"] += 1
+            state["in_flight"] += 1
+            if state["in_flight"] > 1:
+                state["overlaps"] += 1
+            time.sleep(INTERVAL * 2.4)
+            state["in_flight"] -= 1
+
+        scheduler = PeriodicScheduler(INTERVAL, slow)
+        scheduler.start()
+        time.sleep(INTERVAL * 20)
+        scheduler.stop()
+        self.assertGreater(state["calls"], 1)
+        self.assertEqual(state["overlaps"], 0)
+
+    def test_stop_is_prompt_and_final(self):
+        calls = []
+        scheduler = PeriodicScheduler(10.0, lambda: calls.append(1))
+        scheduler.start()
+        before = time.monotonic()
+        scheduler.stop()
+        self.assertLess(time.monotonic() - before, 1.0)
+        self.assertTrue(scheduler._stopped)
+        time.sleep(0.1)
+        self.assertEqual(calls, [])
+
+    def test_stop_before_start_and_double_stop_do_not_raise(self):
+        scheduler = PeriodicScheduler(INTERVAL, lambda: None)
+        scheduler.stop()
+        scheduler.start()
+        scheduler.stop()
+        scheduler.stop()
+        self.assertTrue(scheduler._stopped)
+
+    def test_start_is_idempotent_and_restartable(self):
+        calls = []
+        scheduler = PeriodicScheduler(INTERVAL, lambda: calls.append(1))
+        scheduler.start()
+        first = scheduler._thread
+        scheduler.start()
+        self.assertIs(scheduler._thread, first, "start() armed a second thread")
+        scheduler.stop()
+        self.assertFalse(first.is_alive())
+        stopped_at = len(calls)
+
+        scheduler.start()
+        second = scheduler._thread
+        time.sleep(INTERVAL * 4)
+        scheduler.stop()
+        self.assertGreater(len(calls), stopped_at, "restart did not resume ticking")
+        self.assertFalse(second.is_alive())
+
+    def test_start_after_a_timed_out_stop_leaves_only_the_new_loop(self):
+        """A wedged callback exits when released instead of ticking again."""
+        release = threading.Event()
+        idents = []
+        blocked_once = threading.Event()
+
+        def wedged():
+            idents.append(threading.get_ident())
+            if not blocked_once.is_set():
+                blocked_once.set()
+                release.wait(5)
+
+        scheduler = PeriodicScheduler(INTERVAL, wedged)
+        scheduler.start()
+        first = scheduler._thread
+        self.assertTrue(blocked_once.wait(2), "callback never ran")
+
+        scheduler.stop()  # join times out, callback still blocked
+        self.assertTrue(first.is_alive())
+
+        scheduler.start()  # a new run, immediately, on its own event
+        second = scheduler._thread
+        self.assertIsNot(second, first)
+
+        release.set()
+        first.join(2)
+        self.assertFalse(first.is_alive(), "the wedged thread never exited")
+
+        time.sleep(INTERVAL * 6)
+        scheduler.stop()
+        self.assertEqual(
+            set(idents[1:]), {second.ident}, f"the old loop kept ticking: {idents}"
+        )
+
+    def test_exception_does_not_kill_the_loop(self):
+        calls = []
+
+        def flaky():
+            calls.append(1)
+            if len(calls) == 1:
+                raise ValueError("boom")
+
+        scheduler = PeriodicScheduler(INTERVAL, flaky)
+        scheduler.start()
+        time.sleep(INTERVAL * 6)
+        scheduler.stop()
+        self.assertGreater(len(calls), 1)

From e3267f5d82e6f51f5e349be8b9515b271e416198 Mon Sep 17 00:00:00 2001
From: David Berenstein <david.m.berenstein@gmail.com>
Date: Wed, 23 Sep 2026 16:39:43 +0900
Subject: [PATCH 2/9] fix(scheduler): never start a second loop while a
 timed-out tick is running

stop() keeps the thread reference and logs a warning when the join times
out; start() waits for it and refuses to start a second loop if it is
still busy.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 codecarbon/external/scheduler.py | 42 ++++++++++++++++++++++++--------
 tests/test_scheduler.py          | 27 +++++++++++---------
 2 files changed, 48 insertions(+), 21 deletions(-)

diff --git a/codecarbon/external/scheduler.py b/codecarbon/external/scheduler.py
index 86b5305e4..06e04e37b 100644
--- a/codecarbon/external/scheduler.py
+++ b/codecarbon/external/scheduler.py
@@ -29,16 +29,31 @@ def __init__(self, interval, function, *args, **kwargs):
 
     @property
     def _stopped(self):
-        return self._thread is None
+        return self._stop_event is None or self._stop_event.is_set()
+
+    def _join_timeout(self, timeout):
+        # ponytail: 5s cap is a guess at "a measurement should never take
+        # longer than this"; make it configurable if a slow hardware
+        # backend ever needs more.
+        return min(self.interval, 5.0) if timeout is None else timeout
 
     def start(self):
         """
         Start the scheduler. Calling it on a running scheduler is a no-op.
+        If a previous run is still finishing its callback (a timed-out
+        stop()), wait for it; if it is still busy, refuse to start a second
+        loop so the function is never run concurrently.
         """
         if self._thread is not None:
-            return
-        # One event per run: a thread left behind by a timed-out stop() keeps
-        # its own, still-set event and exits as soon as its callback returns.
+            if not self._stopped:
+                return
+            self._thread.join(self._join_timeout(None))
+            if self._thread.is_alive():
+                logger.warning(
+                    "Previous scheduled measurement is still running;"
+                    " not starting a second loop"
+                )
+                return
         self._stop_event = Event()
         self._thread = Thread(
             target=self._loop,
@@ -66,9 +81,16 @@ def stop(self, timeout=None):
         """
         if self._stop_event is not None:
             self._stop_event.set()
-        thread, self._thread = self._thread, None
-        if thread is not None and thread is not current_thread():
-            # ponytail: 5s cap is a guess at "a measurement should never take
-            # longer than this"; make it configurable if a slow hardware
-            # backend ever needs more.
-            thread.join(min(self.interval, 5.0) if timeout is None else timeout)
+        thread = self._thread
+        if thread is None or thread is current_thread():
+            self._thread = None
+            return
+        thread.join(self._join_timeout(timeout))
+        if thread.is_alive():
+            # Keep the reference so start() will not launch a second loop.
+            logger.warning(
+                "Scheduled measurement still running after stop(); it will"
+                " finish in the background and may overlap the next one"
+            )
+        else:
+            self._thread = None
diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py
index ae2fbf11a..55fe46081 100644
--- a/tests/test_scheduler.py
+++ b/tests/test_scheduler.py
@@ -76,39 +76,44 @@ def test_start_is_idempotent_and_restartable(self):
         self.assertGreater(len(calls), stopped_at, "restart did not resume ticking")
         self.assertFalse(second.is_alive())
 
-    def test_start_after_a_timed_out_stop_leaves_only_the_new_loop(self):
-        """A wedged callback exits when released instead of ticking again."""
+    def test_start_after_a_timed_out_stop_never_runs_two_loops(self):
+        """A wedged callback blocks a restart until it returns."""
         release = threading.Event()
-        idents = []
+        in_flight = {"now": 0, "max": 0}
         blocked_once = threading.Event()
 
         def wedged():
-            idents.append(threading.get_ident())
+            in_flight["now"] += 1
+            in_flight["max"] = max(in_flight["max"], in_flight["now"])
             if not blocked_once.is_set():
                 blocked_once.set()
                 release.wait(5)
+            in_flight["now"] -= 1
 
         scheduler = PeriodicScheduler(INTERVAL, wedged)
         scheduler.start()
         first = scheduler._thread
         self.assertTrue(blocked_once.wait(2), "callback never ran")
 
-        scheduler.stop()  # join times out, callback still blocked
+        with self.assertLogs("codecarbon", level="WARNING"):
+            scheduler.stop()  # join times out, callback still blocked
         self.assertTrue(first.is_alive())
+        self.assertIs(scheduler._thread, first, "stop() dropped the wedged thread")
 
-        scheduler.start()  # a new run, immediately, on its own event
-        second = scheduler._thread
-        self.assertIsNot(second, first)
+        with self.assertLogs("codecarbon", level="WARNING"):
+            scheduler.start()  # refused: the old callback is still running
+        self.assertIs(scheduler._thread, first, "start() launched a second loop")
 
         release.set()
         first.join(2)
         self.assertFalse(first.is_alive(), "the wedged thread never exited")
 
+        scheduler.start()  # now the restart goes through
+        second = scheduler._thread
+        self.assertIsNot(second, first)
         time.sleep(INTERVAL * 6)
         scheduler.stop()
-        self.assertEqual(
-            set(idents[1:]), {second.ident}, f"the old loop kept ticking: {idents}"
-        )
+        self.assertEqual(in_flight["max"], 1, "two loops ran the callback at once")
 
     def test_exception_does_not_kill_the_loop(self):
         calls = []

From 04a8dce4eaf23ff2869dbab30f8948c1e68c90ff Mon Sep 17 00:00:00 2001
From: benoit-cty <anne@onyme.fr>
Date: Mon, 26 Oct 2026 18:39:24 +0100
Subject: [PATCH 3/9] fix(scheduler): resume after a timed-out stop instead of
 refusing to start

After a stop() whose join timed out, start() blocked the caller for up to
min(interval, 5s) and then, if the old callback was still running, only
logged a warning and returned. Nothing retried it, so a start_task() /
stop_task() pair during a slow measurement left the periodic scheduler off
for the rest of the run.

start() now always arms the new loop and hands it the previous thread; the
new loop joins it before its first tick, so the function still never runs
concurrently, start() never blocks, and the restart is not lost.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 codecarbon/external/scheduler.py | 26 ++++++++++----------------
 tests/test_scheduler.py          | 23 ++++++++++++++---------
 2 files changed, 24 insertions(+), 25 deletions(-)

diff --git a/codecarbon/external/scheduler.py b/codecarbon/external/scheduler.py
index 06e04e37b..dfb0f0fdf 100644
--- a/codecarbon/external/scheduler.py
+++ b/codecarbon/external/scheduler.py
@@ -41,29 +41,23 @@ def start(self):
         """
         Start the scheduler. Calling it on a running scheduler is a no-op.
         If a previous run is still finishing its callback (a timed-out
-        stop()), wait for it; if it is still busy, refuse to start a second
-        loop so the function is never run concurrently.
+        stop()), the new loop waits for it before its first tick, so the
+        function is never run concurrently and the restart is not lost.
         """
-        if self._thread is not None:
-            if not self._stopped:
-                return
-            self._thread.join(self._join_timeout(None))
-            if self._thread.is_alive():
-                logger.warning(
-                    "Previous scheduled measurement is still running;"
-                    " not starting a second loop"
-                )
-                return
+        if self._thread is not None and not self._stopped:
+            return
         self._stop_event = Event()
         self._thread = Thread(
             target=self._loop,
-            args=(self._stop_event,),
+            args=(self._stop_event, self._thread),
             daemon=True,
             name=f"codecarbon-{getattr(self.function, '__name__', 'scheduler')}",
         )
         self._thread.start()
 
-    def _loop(self, stop_event):
+    def _loop(self, stop_event, previous=None):
+        if previous is not None:
+            previous.join()
         next_call = time.monotonic() + self.interval
         while not stop_event.wait(max(0.0, next_call - time.monotonic())):
             try:
@@ -87,10 +81,10 @@ def stop(self, timeout=None):
             return
         thread.join(self._join_timeout(timeout))
         if thread.is_alive():
-            # Keep the reference so start() will not launch a second loop.
+            # Keep the reference so the next start() waits for it.
             logger.warning(
                 "Scheduled measurement still running after stop(); it will"
-                " finish in the background and may overlap the next one"
+                " finish in the background"
             )
         else:
             self._thread = None
diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py
index 55fe46081..0906f339e 100644
--- a/tests/test_scheduler.py
+++ b/tests/test_scheduler.py
@@ -76,15 +76,17 @@ def test_start_is_idempotent_and_restartable(self):
         self.assertGreater(len(calls), stopped_at, "restart did not resume ticking")
         self.assertFalse(second.is_alive())
 
-    def test_start_after_a_timed_out_stop_never_runs_two_loops(self):
-        """A wedged callback blocks a restart until it returns."""
+    def test_start_after_a_timed_out_stop_resumes_without_overlap(self):
+        """A restart during a wedged callback waits for it, then ticks again."""
         release = threading.Event()
         in_flight = {"now": 0, "max": 0}
+        threads = []
         blocked_once = threading.Event()
 
         def wedged():
             in_flight["now"] += 1
             in_flight["max"] = max(in_flight["max"], in_flight["now"])
+            threads.append(threading.current_thread())
             if not blocked_once.is_set():
                 blocked_once.set()
                 release.wait(5)
@@ -100,19 +102,22 @@ def wedged():
         self.assertTrue(first.is_alive())
         self.assertIs(scheduler._thread, first, "stop() dropped the wedged thread")
 
-        with self.assertLogs("codecarbon", level="WARNING"):
-            scheduler.start()  # refused: the old callback is still running
-        self.assertIs(scheduler._thread, first, "start() launched a second loop")
+        before = time.monotonic()
+        scheduler.start()  # must neither block nor be refused
+        self.assertLess(time.monotonic() - before, INTERVAL)
+        second = scheduler._thread
+        self.assertIsNot(second, first)
+        self.assertFalse(scheduler._stopped)
+
+        time.sleep(INTERVAL * 3)
+        self.assertEqual(threads, [first], "the new loop ran during the old call")
 
         release.set()
         first.join(2)
         self.assertFalse(first.is_alive(), "the wedged thread never exited")
-
-        scheduler.start()  # now the restart goes through
-        second = scheduler._thread
-        self.assertIsNot(second, first)
         time.sleep(INTERVAL * 6)
         scheduler.stop()
+        self.assertIn(second, threads, "the restarted loop never ticked")
         self.assertEqual(in_flight["max"], 1, "two loops ran the callback at once")
 
     def test_exception_does_not_kill_the_loop(self):

From 1927fb0ba03c79f96da02d18e979b2f119364195 Mon Sep 17 00:00:00 2001
From: benoit-cty <anne@onyme.fr>
Date: Mon, 26 Oct 2026 18:39:44 +0100
Subject: [PATCH 4/9] fix(scheduler): keep the thread reference when stop() is
 called from the callback

stop() called from inside the scheduled function cleared _thread although
that function was still running. A following start() then saw no previous
thread and launched a loop that ran concurrently with it, the overlap this
refactor is meant to rule out.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 codecarbon/external/scheduler.py |  3 ++-
 tests/test_scheduler.py          | 32 ++++++++++++++++++++++++++++++++
 2 files changed, 34 insertions(+), 1 deletion(-)

diff --git a/codecarbon/external/scheduler.py b/codecarbon/external/scheduler.py
index dfb0f0fdf..b84c427aa 100644
--- a/codecarbon/external/scheduler.py
+++ b/codecarbon/external/scheduler.py
@@ -77,7 +77,8 @@ def stop(self, timeout=None):
             self._stop_event.set()
         thread = self._thread
         if thread is None or thread is current_thread():
-            self._thread = None
+            # Called from the callback itself: it is still running, so keep
+            # the reference for the next start() to wait on.
             return
         thread.join(self._join_timeout(timeout))
         if thread.is_alive():
diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py
index 0906f339e..69e7da37f 100644
--- a/tests/test_scheduler.py
+++ b/tests/test_scheduler.py
@@ -120,6 +120,38 @@ def wedged():
         self.assertIn(second, threads, "the restarted loop never ticked")
         self.assertEqual(in_flight["max"], 1, "two loops ran the callback at once")
 
+    def test_stop_from_the_callback_then_start_does_not_overlap(self):
+        """stop() called by the callback itself must not let start() overlap it."""
+        release = threading.Event()
+        stopped_itself = threading.Event()
+        in_flight = {"now": 0, "max": 0}
+        threads = []
+
+        def stops_itself():
+            in_flight["now"] += 1
+            in_flight["max"] = max(in_flight["max"], in_flight["now"])
+            threads.append(threading.current_thread())
+            if not stopped_itself.is_set():
+                scheduler.stop()
+                stopped_itself.set()
+                release.wait(5)
+            in_flight["now"] -= 1
+
+        scheduler = PeriodicScheduler(INTERVAL, stops_itself)
+        scheduler.start()
+        first = scheduler._thread
+        self.assertTrue(stopped_itself.wait(2), "callback never ran")
+
+        scheduler.start()
+        time.sleep(INTERVAL * 3)
+        self.assertEqual(threads, [first], "the new loop ran during the old call")
+
+        release.set()
+        time.sleep(INTERVAL * 6)
+        scheduler.stop()
+        self.assertGreater(len(threads), 1, "the restarted loop never ticked")
+        self.assertEqual(in_flight["max"], 1, "two loops ran the callback at once")
+
     def test_exception_does_not_kill_the_loop(self):
         calls = []
 

From 2bc5629822daa8685ed33f89a55305be74682c89 Mon Sep 17 00:00:00 2001
From: benoit-cty <anne@onyme.fr>
Date: Mon, 26 Oct 2026 18:40:13 +0100
Subject: [PATCH 5/9] fix(scheduler): skip overrun slots instead of calling
 again immediately

When a call outlasted the interval, next_call = max(next_call + interval,
now) set the deadline to now, so the next call started at once. A function
slower than its interval, e.g. a 1.2 s GPU read in the 1 s _monitor_power
scheduler, then ran back-to-back at 100 % duty cycle instead of skipping
the missed ticks as the class docstring says.

The deadline now advances by whole intervals past the current time, which
keeps the grid and leaves a pause before the next call.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 codecarbon/external/scheduler.py |  9 +++++++--
 tests/test_scheduler.py          | 19 +++++++++++++++++++
 2 files changed, 26 insertions(+), 2 deletions(-)

diff --git a/codecarbon/external/scheduler.py b/codecarbon/external/scheduler.py
index b84c427aa..80c03d5ab 100644
--- a/codecarbon/external/scheduler.py
+++ b/codecarbon/external/scheduler.py
@@ -1,3 +1,4 @@
+import math
 import time
 from threading import Event, Thread, current_thread
 
@@ -64,8 +65,12 @@ def _loop(self, stop_event, previous=None):
                 self.function(*self.args, **self.kwargs)
             except Exception:  # noqa: BLE001 - must not kill the only thread
                 logger.error("Scheduled measurement failed", exc_info=True)
-            # Absolute deadline, but never a burst of catch-up ticks.
-            next_call = max(next_call + self.interval, time.monotonic())
+            next_call += self.interval
+            overrun = time.monotonic() - next_call
+            if overrun > 0 and self.interval > 0:
+                # Skip the slots the call overran: no burst of catch-up ticks,
+                # and no back-to-back call either.
+                next_call += math.ceil(overrun / self.interval) * self.interval
 
     def stop(self, timeout=None):
         """
diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py
index 69e7da37f..ac5d091da 100644
--- a/tests/test_scheduler.py
+++ b/tests/test_scheduler.py
@@ -39,6 +39,25 @@ def slow():
         self.assertGreater(state["calls"], 1)
         self.assertEqual(state["overlaps"], 0)
 
+    def test_overrunning_call_waits_for_the_next_slot(self):
+        """A call longer than the interval skips the missed slots, not the wait."""
+        interval = 0.1
+        spans = []
+
+        def slow():
+            start = time.monotonic()
+            time.sleep(interval * 2.5)
+            spans.append((start, time.monotonic()))
+
+        scheduler = PeriodicScheduler(interval, slow)
+        scheduler.start()
+        time.sleep(interval * 12)
+        scheduler.stop()
+        self.assertGreater(len(spans), 2)
+        gaps = [nxt[0] - prev[1] for prev, nxt in zip(spans, spans[1:])]
+        # The next slot is 0.5 interval after each call ends.
+        self.assertGreater(min(gaps), interval * 0.2, f"back-to-back calls: {gaps}")
+
     def test_stop_is_prompt_and_final(self):
         calls = []
         scheduler = PeriodicScheduler(10.0, lambda: calls.append(1))

From c0bf25db1e50fdbd40e45d58bf448b43fd4be156 Mon Sep 17 00:00:00 2001
From: benoit-cty <anne@onyme.fr>
Date: Mon, 26 Oct 2026 18:40:54 +0100
Subject: [PATCH 6/9] fix(scheduler): guard start() and stop() with a lock
 again

The Timer-based scheduler serialised start()/stop() with a Lock; the
refactor dropped it. Both now check then replace _thread and _stop_event,
so two threads calling start() together could each arm a loop. The first
loop's Event was overwritten and could never be set, leaving a loop that
stop() cannot end.

The lock covers the state changes only; stop() still joins outside it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 codecarbon/external/scheduler.py | 35 ++++++++++++++++++--------------
 tests/test_scheduler.py          | 35 ++++++++++++++++++++++++++++++++
 2 files changed, 55 insertions(+), 15 deletions(-)

diff --git a/codecarbon/external/scheduler.py b/codecarbon/external/scheduler.py
index 80c03d5ab..eb117d721 100644
--- a/codecarbon/external/scheduler.py
+++ b/codecarbon/external/scheduler.py
@@ -1,6 +1,6 @@
 import math
 import time
-from threading import Event, Thread, current_thread
+from threading import Event, Lock, Thread, current_thread
 
 from codecarbon.external.logger import logger
 
@@ -25,6 +25,7 @@ def __init__(self, interval, function, *args, **kwargs):
         self.function = function
         self.args = args
         self.kwargs = kwargs
+        self._lock = Lock()
         self._thread = None
         self._stop_event = None
 
@@ -45,16 +46,17 @@ def start(self):
         stop()), the new loop waits for it before its first tick, so the
         function is never run concurrently and the restart is not lost.
         """
-        if self._thread is not None and not self._stopped:
-            return
-        self._stop_event = Event()
-        self._thread = Thread(
-            target=self._loop,
-            args=(self._stop_event, self._thread),
-            daemon=True,
-            name=f"codecarbon-{getattr(self.function, '__name__', 'scheduler')}",
-        )
-        self._thread.start()
+        with self._lock:
+            if self._thread is not None and not self._stopped:
+                return
+            self._stop_event = Event()
+            self._thread = Thread(
+                target=self._loop,
+                args=(self._stop_event, self._thread),
+                daemon=True,
+                name=f"codecarbon-{getattr(self.function, '__name__', 'scheduler')}",
+            )
+            self._thread.start()
 
     def _loop(self, stop_event, previous=None):
         if previous is not None:
@@ -78,9 +80,10 @@ def stop(self, timeout=None):
         ::timeout:: seconds to wait for the running function, bounded by
         default so a wedged measurement cannot hang the caller for long.
         """
-        if self._stop_event is not None:
-            self._stop_event.set()
-        thread = self._thread
+        with self._lock:
+            if self._stop_event is not None:
+                self._stop_event.set()
+            thread = self._thread
         if thread is None or thread is current_thread():
             # Called from the callback itself: it is still running, so keep
             # the reference for the next start() to wait on.
@@ -93,4 +96,6 @@ def stop(self, timeout=None):
                 " finish in the background"
             )
         else:
-            self._thread = None
+            with self._lock:
+                if self._thread is thread:
+                    self._thread = None
diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py
index ac5d091da..1ab3c4f36 100644
--- a/tests/test_scheduler.py
+++ b/tests/test_scheduler.py
@@ -1,7 +1,9 @@
 import threading
 import time
 import unittest
+from unittest import mock
 
+from codecarbon.external import scheduler as scheduler_module
 from codecarbon.external.scheduler import PeriodicScheduler
 
 INTERVAL = 0.05
@@ -171,6 +173,39 @@ def stops_itself():
         self.assertGreater(len(threads), 1, "the restarted loop never ticked")
         self.assertEqual(in_flight["max"], 1, "two loops ran the callback at once")
 
+    def test_concurrent_starts_arm_a_single_loop(self):
+        """Two threads calling start() at once must not create two loops."""
+
+        class SlowEvent(threading.Event):
+            def __init__(self):
+                time.sleep(INTERVAL)  # widen the check-then-act window
+                super().__init__()
+
+        def concurrent_start_target():
+            pass
+
+        scheduler = PeriodicScheduler(INTERVAL, concurrent_start_target)
+        barrier = threading.Barrier(2)
+
+        def start():
+            barrier.wait()
+            scheduler.start()
+
+        with mock.patch.object(scheduler_module, "Event", SlowEvent):
+            starters = [threading.Thread(target=start) for _ in range(2)]
+            for starter in starters:
+                starter.start()
+            for starter in starters:
+                starter.join()
+        scheduler.stop()
+        time.sleep(INTERVAL * 2)
+        alive = [
+            t.name
+            for t in threading.enumerate()
+            if t.name == "codecarbon-concurrent_start_target"
+        ]
+        self.assertEqual(alive, [], "a loop survived stop(): two loops were armed")
+
     def test_exception_does_not_kill_the_loop(self):
         calls = []
 

From bd1c387de1e9b54e2329654987b6e51faf80b2db Mon Sep 17 00:00:00 2001
From: benoit-cty <anne@onyme.fr>
Date: Mon, 26 Oct 2026 18:41:51 +0100
Subject: [PATCH 7/9] fix(tracker): make the final measurement wait for an
 in-flight scheduled one

PeriodicScheduler.stop() gives up after min(measure_power_secs, 5 s), but a
scheduled _measure_power_and_energy can take longer: an Electricity Maps
lookup has a 30 s timeout and live_out() may push to the API. tracker.stop()
then ran its final measurement and _persist_data on the main thread while
the tick was still updating _total_energy, _last_measured_time and
_measure_occurrence, corrupting the last row.

A re-entrant lock now serialises _measure_power_and_energy and the stale
check in _measure_power_and_energy_if_stale, so the final measurement runs
after the tick has finished.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 codecarbon/emissions_tracker.py          | 12 +++++++--
 tests/test_emissions_tracker_constant.py | 32 ++++++++++++++++++++++++
 2 files changed, 42 insertions(+), 2 deletions(-)

diff --git a/codecarbon/emissions_tracker.py b/codecarbon/emissions_tracker.py
index 127d368b3..6737e83ab 100644
--- a/codecarbon/emissions_tracker.py
+++ b/codecarbon/emissions_tracker.py
@@ -283,6 +283,9 @@ def _initialize_runtime_state(self) -> None:
         self._ram_used_history: List[float] = []
         # The monitor thread appends while the measurement thread clears/averages
         self._utilization_history_lock = threading.Lock()
+        # A scheduled measurement still running when stop() gave up joining it
+        # must not overlap the final one
+        self._measurement_lock = threading.RLock()
         self._total_cpu_energy: Energy = Energy.from_energy(kWh=0)
         self._total_gpu_energy: Energy = Energy.from_energy(kWh=0)
         self._total_ram_energy: Energy = Energy.from_energy(kWh=0)
@@ -1357,8 +1360,9 @@ def _do_measurements(self) -> None:
 
     def _measure_power_and_energy_if_stale(self, min_interval_s: float = 0.05) -> None:
         """Measure only if the last sample is older than ``min_interval_s``."""
-        if time.perf_counter() - self._last_measured_time >= min_interval_s:
-            self._measure_power_and_energy()
+        with self._measurement_lock:
+            if time.perf_counter() - self._last_measured_time >= min_interval_s:
+                self._measure_power_and_energy()
 
     def _measure_power_and_energy(self) -> None:
         """
@@ -1366,6 +1370,10 @@ def _measure_power_and_energy(self) -> None:
         every `self._measure_power_secs` seconds.
         :return: None
         """
+        with self._measurement_lock:
+            self._measure_power_and_energy_locked()
+
+    def _measure_power_and_energy_locked(self) -> None:
         try:
             last_duration = time.perf_counter() - self._last_measured_time
         except AttributeError as e:
diff --git a/tests/test_emissions_tracker_constant.py b/tests/test_emissions_tracker_constant.py
index ba3f4407e..33305eb54 100644
--- a/tests/test_emissions_tracker_constant.py
+++ b/tests/test_emissions_tracker_constant.py
@@ -1,5 +1,6 @@
 import os
 import tempfile
+import threading
 import time
 import unittest
 from unittest import mock
@@ -202,3 +203,34 @@ def test_carbon_tracker_offline_region_error(self):
 
         finally:
             tracker.stop()
+
+
+class TestFinalMeasurementWaitsForInFlightTick(unittest.TestCase):
+    def test_stop_does_not_overlap_a_slow_scheduled_measurement(self):
+        tracker = OfflineEmissionsTracker(
+            country_iso_code="USA",
+            measure_power_secs=1,
+            save_to_file=False,
+            allow_multiple_runs=True,
+        )
+        tracker.start()
+        do_measurements = tracker._do_measurements
+        in_flight = {"now": 0, "max": 0}
+        entered = threading.Event()
+        release = threading.Event()
+
+        def slow_do_measurements():
+            in_flight["now"] += 1
+            in_flight["max"] = max(in_flight["max"], in_flight["now"])
+            if threading.current_thread().name.startswith("codecarbon-_measure"):
+                entered.set()
+                # Outlasts stop()'s 1 s join of the scheduler thread.
+                release.wait(5)
+            do_measurements()
+            in_flight["now"] -= 1
+
+        tracker._do_measurements = slow_do_measurements
+        self.assertTrue(entered.wait(5), "the scheduled measurement never ran")
+        threading.Timer(1.5, release.set).start()
+        tracker.stop()
+        self.assertEqual(in_flight["max"], 1, "final measurement overlapped a tick")

From 864b3d14f17c8031cda1f3311b68558858ca38d9 Mon Sep 17 00:00:00 2001
From: benoit-cty <anne@onyme.fr>
Date: Mon, 26 Oct 2026 18:55:23 +0100
Subject: [PATCH 8/9] refactor(scheduler): replace the stop() timeout parameter
 with a module constant

No caller passed stop(timeout=...), and _join_timeout(None) only computed
min(interval, 5.0). A named module constant says what the cap is and lets a
test lower it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 codecarbon/external/scheduler.py | 19 ++++++++-----------
 1 file changed, 8 insertions(+), 11 deletions(-)

diff --git a/codecarbon/external/scheduler.py b/codecarbon/external/scheduler.py
index eb117d721..38039d615 100644
--- a/codecarbon/external/scheduler.py
+++ b/codecarbon/external/scheduler.py
@@ -4,6 +4,10 @@
 
 from codecarbon.external.logger import logger
 
+# Longest stop() waits for the in-flight call, a guess at "a measurement should
+# never take longer than this", so a wedged one cannot hang the caller for long.
+STOP_JOIN_TIMEOUT_SECONDS = 5.0
+
 
 class PeriodicScheduler:
     """
@@ -33,12 +37,6 @@ def __init__(self, interval, function, *args, **kwargs):
     def _stopped(self):
         return self._stop_event is None or self._stop_event.is_set()
 
-    def _join_timeout(self, timeout):
-        # ponytail: 5s cap is a guess at "a measurement should never take
-        # longer than this"; make it configurable if a slow hardware
-        # backend ever needs more.
-        return min(self.interval, 5.0) if timeout is None else timeout
-
     def start(self):
         """
         Start the scheduler. Calling it on a running scheduler is a no-op.
@@ -74,11 +72,10 @@ def _loop(self, stop_event, previous=None):
                 # and no back-to-back call either.
                 next_call += math.ceil(overrun / self.interval) * self.interval
 
-    def stop(self, timeout=None):
+    def stop(self):
         """
-        Stop the scheduler and wait for the in-flight call to return.
-        ::timeout:: seconds to wait for the running function, bounded by
-        default so a wedged measurement cannot hang the caller for long.
+        Stop the scheduler and wait for the in-flight call to return, at most
+        min(interval, STOP_JOIN_TIMEOUT_SECONDS) seconds.
         """
         with self._lock:
             if self._stop_event is not None:
@@ -88,7 +85,7 @@ def stop(self, timeout=None):
             # Called from the callback itself: it is still running, so keep
             # the reference for the next start() to wait on.
             return
-        thread.join(self._join_timeout(timeout))
+        thread.join(min(self.interval, STOP_JOIN_TIMEOUT_SECONDS))
         if thread.is_alive():
             # Keep the reference so the next start() waits for it.
             logger.warning(

From b2254b617a3c66b6fd7cdead8b415581d64bae57 Mon Sep 17 00:00:00 2001
From: benoit-cty <anne@onyme.fr>
Date: Mon, 26 Oct 2026 19:00:15 +0100
Subject: [PATCH 9/9] test(scheduler): drive the scheduler tests with an
 injectable fake clock

The scheduler tests slept on real time with a 50 ms interval and asserted
how many ticks had happened, so a loaded CI runner could make them fail.

PeriodicScheduler now reads time and waits through a clock attribute,
MonotonicClock by default, that a test replaces before start(). The
FakeClock in tests/testutils.py only moves when the test advances it and
returns once the loop is waiting again, so the tests assert exact tick
times (e.g. [1, 4, 7, 10] for an overrun). A new test checks that the
cadence does not drift. The only remaining real-time waits are upper
bounds on things that must not happen, which a correct scheduler passes
however slow the machine is. The tracker overlap test synchronises on the
final measurement instead of a 1.5 s Timer.

The scheduler and tracker tests now take 1.6 s instead of about 8 s.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 codecarbon/external/scheduler.py         |  19 ++-
 tests/test_emissions_tracker_constant.py |  37 ++++--
 tests/test_scheduler.py                  | 149 +++++++++++++----------
 tests/testutils.py                       |  53 ++++++++
 4 files changed, 186 insertions(+), 72 deletions(-)

diff --git a/codecarbon/external/scheduler.py b/codecarbon/external/scheduler.py
index 38039d615..4256e67fa 100644
--- a/codecarbon/external/scheduler.py
+++ b/codecarbon/external/scheduler.py
@@ -9,6 +9,16 @@
 STOP_JOIN_TIMEOUT_SECONDS = 5.0
 
 
+class MonotonicClock:
+    """The real clock. Tests swap in a fake one to drive the loop."""
+
+    monotonic = staticmethod(time.monotonic)
+
+    @staticmethod
+    def wait(event, timeout):
+        return event.wait(timeout)
+
+
 class PeriodicScheduler:
     """
     Run ``function`` every ``interval`` seconds on a single daemon thread.
@@ -29,6 +39,8 @@ def __init__(self, interval, function, *args, **kwargs):
         self.function = function
         self.args = args
         self.kwargs = kwargs
+        # Replace before start() to control time, e.g. in tests.
+        self.clock = MonotonicClock()
         self._lock = Lock()
         self._thread = None
         self._stop_event = None
@@ -59,14 +71,15 @@ def start(self):
     def _loop(self, stop_event, previous=None):
         if previous is not None:
             previous.join()
-        next_call = time.monotonic() + self.interval
-        while not stop_event.wait(max(0.0, next_call - time.monotonic())):
+        clock = self.clock
+        next_call = clock.monotonic() + self.interval
+        while not clock.wait(stop_event, max(0.0, next_call - clock.monotonic())):
             try:
                 self.function(*self.args, **self.kwargs)
             except Exception:  # noqa: BLE001 - must not kill the only thread
                 logger.error("Scheduled measurement failed", exc_info=True)
             next_call += self.interval
-            overrun = time.monotonic() - next_call
+            overrun = clock.monotonic() - next_call
             if overrun > 0 and self.interval > 0:
                 # Skip the slots the call overran: no burst of catch-up ticks,
                 # and no back-to-back call either.
diff --git a/tests/test_emissions_tracker_constant.py b/tests/test_emissions_tracker_constant.py
index 33305eb54..6cfe85b17 100644
--- a/tests/test_emissions_tracker_constant.py
+++ b/tests/test_emissions_tracker_constant.py
@@ -13,6 +13,8 @@
     OfflineEmissionsTracker,
     track_emissions,
 )
+from codecarbon.external import scheduler as scheduler_module
+from tests.testutils import FakeClock
 
 
 def heavy_computation(run_time_secs: int = 3):
@@ -206,31 +208,52 @@ def test_carbon_tracker_offline_region_error(self):
 
 
 class TestFinalMeasurementWaitsForInFlightTick(unittest.TestCase):
+    @mock.patch.object(scheduler_module, "STOP_JOIN_TIMEOUT_SECONDS", 0.01)
     def test_stop_does_not_overlap_a_slow_scheduled_measurement(self):
         tracker = OfflineEmissionsTracker(
             country_iso_code="USA",
             measure_power_secs=1,
-            save_to_file=False,
+            output_methods=[],
             allow_multiple_runs=True,
         )
+        clock = tracker._scheduler.clock = FakeClock()
         tracker.start()
+        main = threading.current_thread()
         do_measurements = tracker._do_measurements
         in_flight = {"now": 0, "max": 0}
         entered = threading.Event()
-        release = threading.Event()
+        # Set once stop() reaches its final measurement, whether it waits for
+        # the lock (fixed) or measures right away (the overlap).
+        final_measurement_reached = threading.Event()
 
         def slow_do_measurements():
             in_flight["now"] += 1
             in_flight["max"] = max(in_flight["max"], in_flight["now"])
-            if threading.current_thread().name.startswith("codecarbon-_measure"):
+            if threading.current_thread() is main:
+                final_measurement_reached.set()
+            else:
                 entered.set()
-                # Outlasts stop()'s 1 s join of the scheduler thread.
-                release.wait(5)
+                final_measurement_reached.wait(5)
             do_measurements()
             in_flight["now"] -= 1
 
+        class LockSpy:
+            def __init__(self, lock):
+                self.lock = lock
+
+            def __enter__(self):
+                if threading.current_thread() is main:
+                    final_measurement_reached.set()
+                return self.lock.__enter__()
+
+            def __exit__(self, *exc):
+                return self.lock.__exit__(*exc)
+
         tracker._do_measurements = slow_do_measurements
+        tracker._measurement_lock = LockSpy(tracker._measurement_lock)
+        clock.advance(1, settle=False)
         self.assertTrue(entered.wait(5), "the scheduled measurement never ran")
-        threading.Timer(1.5, release.set).start()
-        tracker.stop()
+        # Make the last sample stale so stop() takes its final measurement.
+        tracker._last_measured_time -= 1
+        tracker.stop()  # the join times out while the tick is still running
         self.assertEqual(in_flight["max"], 1, "final measurement overlapped a tick")
diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py
index 1ab3c4f36..b190f1be8 100644
--- a/tests/test_scheduler.py
+++ b/tests/test_scheduler.py
@@ -5,22 +5,34 @@
 
 from codecarbon.external import scheduler as scheduler_module
 from codecarbon.external.scheduler import PeriodicScheduler
+from tests.testutils import FakeClock
 
-INTERVAL = 0.05
+# Fake seconds: time only moves when a test calls clock.advance().
+INTERVAL = 1.0
+# Real seconds a negative check waits for something that must not happen. It
+# only bounds the wait: a correct scheduler passes however slow the machine.
+NEVER = 0.1
 
 
 class TestPeriodicScheduler(unittest.TestCase):
+    def make_scheduler(self, function, interval=INTERVAL):
+        scheduler = PeriodicScheduler(interval, function)
+        self.clock = scheduler.clock = FakeClock()
+        self.addCleanup(scheduler.stop)
+        return scheduler
+
     def test_ticks_run_on_a_single_thread(self):
         """Regression guard: one long-lived thread, not one thread per tick."""
-        names = set()
-        scheduler = PeriodicScheduler(
-            INTERVAL, lambda: names.add(threading.current_thread().name)
+        names = []
+        scheduler = self.make_scheduler(
+            lambda: names.append(threading.current_thread().name)
         )
         scheduler.start()
-        time.sleep(INTERVAL * 10)
+        for _ in range(5):
+            self.clock.advance(INTERVAL)
         scheduler.stop()
-        # A one-element set also proves at least one tick happened.
-        self.assertEqual(len(names), 1, f"expected one worker thread, got {names}")
+        self.assertEqual(len(names), 5)
+        self.assertEqual(len(set(names)), 1, f"expected one worker, got {names}")
 
     def test_slow_function_is_never_re_entered(self):
         """A function slower than the interval must not overlap with itself."""
@@ -31,48 +43,64 @@ def slow():
             state["in_flight"] += 1
             if state["in_flight"] > 1:
                 state["overlaps"] += 1
-            time.sleep(INTERVAL * 2.4)
+            self.clock.sleep(INTERVAL * 2.4)
             state["in_flight"] -= 1
 
-        scheduler = PeriodicScheduler(INTERVAL, slow)
+        scheduler = self.make_scheduler(slow)
         scheduler.start()
-        time.sleep(INTERVAL * 20)
+        for _ in range(20):
+            self.clock.advance(INTERVAL / 4)
         scheduler.stop()
         self.assertGreater(state["calls"], 1)
         self.assertEqual(state["overlaps"], 0)
 
     def test_overrunning_call_waits_for_the_next_slot(self):
         """A call longer than the interval skips the missed slots, not the wait."""
-        interval = 0.1
-        spans = []
+        starts = []
 
         def slow():
-            start = time.monotonic()
-            time.sleep(interval * 2.5)
-            spans.append((start, time.monotonic()))
+            starts.append(self.clock.monotonic())
+            self.clock.sleep(INTERVAL * 2.5)
 
-        scheduler = PeriodicScheduler(interval, slow)
+        scheduler = self.make_scheduler(slow)
         scheduler.start()
-        time.sleep(interval * 12)
+        # The first slot is 4 steps away, then each 2.5 s call leaves half an
+        # interval, 2 steps, before the next free slot.
+        for _ in range(10):
+            self.clock.advance(INTERVAL / 4)
         scheduler.stop()
-        self.assertGreater(len(spans), 2)
-        gaps = [nxt[0] - prev[1] for prev, nxt in zip(spans, spans[1:])]
-        # The next slot is 0.5 interval after each call ends.
-        self.assertGreater(min(gaps), interval * 0.2, f"back-to-back calls: {gaps}")
+        self.assertEqual(starts, [1.0, 4.0, 7.0, 10.0])
+
+    def test_cadence_does_not_drift(self):
+        """Deadlines are absolute: the time a call takes is not added to them."""
+        starts = []
 
-    def test_stop_is_prompt_and_final(self):
+        def takes_a_while():
+            starts.append(self.clock.monotonic())
+            self.clock.sleep(INTERVAL * 0.5)
+
+        scheduler = self.make_scheduler(takes_a_while)
+        scheduler.start()
+        for _ in range(10):
+            self.clock.advance(INTERVAL / 4)
+        scheduler.stop()
+        self.assertEqual(starts, [1.0, 2.0, 3.0, 4.0])
+
+    def test_stop_is_final(self):
         calls = []
-        scheduler = PeriodicScheduler(10.0, lambda: calls.append(1))
+        scheduler = self.make_scheduler(lambda: calls.append(1))
         scheduler.start()
-        before = time.monotonic()
+        thread = scheduler._thread
+        self.assertTrue(self.clock.wait_parked())
         scheduler.stop()
-        self.assertLess(time.monotonic() - before, 1.0)
         self.assertTrue(scheduler._stopped)
-        time.sleep(0.1)
+        self.assertFalse(thread.is_alive(), "stop() returned before the loop ended")
+        self.assertIsNone(scheduler._thread)
+        self.clock.sleep(INTERVAL * 10)
         self.assertEqual(calls, [])
 
     def test_stop_before_start_and_double_stop_do_not_raise(self):
-        scheduler = PeriodicScheduler(INTERVAL, lambda: None)
+        scheduler = self.make_scheduler(lambda: None)
         scheduler.stop()
         scheduler.start()
         scheduler.stop()
@@ -81,41 +109,40 @@ def test_stop_before_start_and_double_stop_do_not_raise(self):
 
     def test_start_is_idempotent_and_restartable(self):
         calls = []
-        scheduler = PeriodicScheduler(INTERVAL, lambda: calls.append(1))
+        scheduler = self.make_scheduler(lambda: calls.append(1))
         scheduler.start()
         first = scheduler._thread
         scheduler.start()
         self.assertIs(scheduler._thread, first, "start() armed a second thread")
+        self.clock.advance(INTERVAL)
         scheduler.stop()
         self.assertFalse(first.is_alive())
-        stopped_at = len(calls)
+        self.assertEqual(len(calls), 1)
 
         scheduler.start()
         second = scheduler._thread
-        time.sleep(INTERVAL * 4)
+        self.clock.advance(INTERVAL)
         scheduler.stop()
-        self.assertGreater(len(calls), stopped_at, "restart did not resume ticking")
+        self.assertEqual(len(calls), 2, "restart did not resume ticking")
         self.assertFalse(second.is_alive())
 
+    @mock.patch.object(scheduler_module, "STOP_JOIN_TIMEOUT_SECONDS", 0.01)
     def test_start_after_a_timed_out_stop_resumes_without_overlap(self):
         """A restart during a wedged callback waits for it, then ticks again."""
         release = threading.Event()
-        in_flight = {"now": 0, "max": 0}
-        threads = []
         blocked_once = threading.Event()
+        threads = []
 
         def wedged():
-            in_flight["now"] += 1
-            in_flight["max"] = max(in_flight["max"], in_flight["now"])
             threads.append(threading.current_thread())
             if not blocked_once.is_set():
                 blocked_once.set()
                 release.wait(5)
-            in_flight["now"] -= 1
 
-        scheduler = PeriodicScheduler(INTERVAL, wedged)
+        scheduler = self.make_scheduler(wedged)
         scheduler.start()
         first = scheduler._thread
+        self.clock.advance(INTERVAL, settle=False)
         self.assertTrue(blocked_once.wait(2), "callback never ran")
 
         with self.assertLogs("codecarbon", level="WARNING"):
@@ -123,68 +150,65 @@ def wedged():
         self.assertTrue(first.is_alive())
         self.assertIs(scheduler._thread, first, "stop() dropped the wedged thread")
 
-        before = time.monotonic()
-        scheduler.start()  # must neither block nor be refused
-        self.assertLess(time.monotonic() - before, INTERVAL)
+        scheduler.start()  # must not be refused
         second = scheduler._thread
         self.assertIsNot(second, first)
         self.assertFalse(scheduler._stopped)
-
-        time.sleep(INTERVAL * 3)
-        self.assertEqual(threads, [first], "the new loop ran during the old call")
+        self.assertFalse(
+            self.clock.wait_parked(NEVER), "the new loop runs during the old call"
+        )
 
         release.set()
         first.join(2)
         self.assertFalse(first.is_alive(), "the wedged thread never exited")
-        time.sleep(INTERVAL * 6)
+        self.clock.advance(INTERVAL)
         scheduler.stop()
-        self.assertIn(second, threads, "the restarted loop never ticked")
-        self.assertEqual(in_flight["max"], 1, "two loops ran the callback at once")
+        self.assertEqual(threads, [first, second])
 
+    @mock.patch.object(scheduler_module, "STOP_JOIN_TIMEOUT_SECONDS", 0.01)
     def test_stop_from_the_callback_then_start_does_not_overlap(self):
         """stop() called by the callback itself must not let start() overlap it."""
         release = threading.Event()
         stopped_itself = threading.Event()
-        in_flight = {"now": 0, "max": 0}
         threads = []
 
         def stops_itself():
-            in_flight["now"] += 1
-            in_flight["max"] = max(in_flight["max"], in_flight["now"])
             threads.append(threading.current_thread())
             if not stopped_itself.is_set():
                 scheduler.stop()
                 stopped_itself.set()
                 release.wait(5)
-            in_flight["now"] -= 1
 
-        scheduler = PeriodicScheduler(INTERVAL, stops_itself)
+        scheduler = self.make_scheduler(stops_itself)
         scheduler.start()
         first = scheduler._thread
+        self.clock.advance(INTERVAL, settle=False)
         self.assertTrue(stopped_itself.wait(2), "callback never ran")
 
         scheduler.start()
-        time.sleep(INTERVAL * 3)
-        self.assertEqual(threads, [first], "the new loop ran during the old call")
+        second = scheduler._thread
+        self.assertFalse(
+            self.clock.wait_parked(NEVER), "the new loop runs during the old call"
+        )
 
         release.set()
-        time.sleep(INTERVAL * 6)
+        first.join(2)
+        self.clock.advance(INTERVAL)
         scheduler.stop()
-        self.assertGreater(len(threads), 1, "the restarted loop never ticked")
-        self.assertEqual(in_flight["max"], 1, "two loops ran the callback at once")
+        self.assertEqual(threads, [first, second])
 
     def test_concurrent_starts_arm_a_single_loop(self):
         """Two threads calling start() at once must not create two loops."""
 
         class SlowEvent(threading.Event):
             def __init__(self):
-                time.sleep(INTERVAL)  # widen the check-then-act window
+                time.sleep(NEVER)  # widen the check-then-act window
                 super().__init__()
 
         def concurrent_start_target():
             pass
 
-        scheduler = PeriodicScheduler(INTERVAL, concurrent_start_target)
+        scheduler = self.make_scheduler(concurrent_start_target)
         barrier = threading.Barrier(2)
 
         def start():
@@ -198,7 +222,6 @@ def start():
             for starter in starters:
                 starter.join()
         scheduler.stop()
-        time.sleep(INTERVAL * 2)
         alive = [
             t.name
             for t in threading.enumerate()
@@ -214,8 +237,10 @@ def flaky():
             if len(calls) == 1:
                 raise ValueError("boom")
 
-        scheduler = PeriodicScheduler(INTERVAL, flaky)
+        scheduler = self.make_scheduler(flaky)
         scheduler.start()
-        time.sleep(INTERVAL * 6)
+        with self.assertLogs("codecarbon", level="ERROR"):
+            self.clock.advance(INTERVAL)
+        self.clock.advance(INTERVAL)
         scheduler.stop()
-        self.assertGreater(len(calls), 1)
+        self.assertEqual(len(calls), 2)
diff --git a/tests/testutils.py b/tests/testutils.py
index e3d1dc2d1..732ce20e1 100644
--- a/tests/testutils.py
+++ b/tests/testutils.py
@@ -1,4 +1,5 @@
 import builtins
+import threading
 import unittest
 from pathlib import Path
 
@@ -33,3 +34,55 @@ def conditional_open_func(path, *args, **kwargs):
         return conditional_open_func
 
     return mocked_open
+
+
+class FakeClock:
+    """
+    Fake time for ``PeriodicScheduler.clock``: the loop only moves when the
+    test calls advance(), so tests assert on exact ticks instead of sleeping.
+    """
+
+    def __init__(self):
+        self.now = 0.0
+        self._cond = threading.Condition()
+        self._parked = set()
+
+    def monotonic(self):
+        with self._cond:
+            return self.now
+
+    def sleep(self, seconds):
+        """Let a scheduled function take ``seconds`` of fake time."""
+        with self._cond:
+            self.now += seconds
+
+    def wait(self, event, timeout):
+        me = threading.current_thread()
+        with self._cond:
+            deadline = self.now + timeout
+            while not event.is_set() and self.now < deadline:
+                self._parked.add(me)
+                self._cond.notify_all()
+                # stop() sets the event without notifying us, so poll it.
+                self._cond.wait(0.001)
+            self._parked.discard(me)
+            self._cond.notify_all()
+            return event.is_set()
+
+    def wait_parked(self, timeout=2.0):
+        """True once a loop is waiting for its next deadline."""
+        with self._cond:
+            return self._cond.wait_for(lambda: self._parked, timeout)
+
+    def advance(self, seconds, settle=True):
+        """
+        Move time forward once a loop is waiting. With ``settle``, return only
+        when the loop has run what was due and is waiting again.
+        """
+        assert self.wait_parked(), "no scheduler loop is waiting"
+        with self._cond:
+            self.now += seconds
+            self._parked.clear()
+            self._cond.notify_all()
+        if settle:
+            assert self.wait_parked(), "the scheduler loop did not settle"