chore(internal): raise instead of hanging on awake() after stop()

PeriodicThread.awake() cleared _served and waited on it. When called
after stop() had completed, the worker had already set _served one last
time during its cleanup and exited — no future set() would satisfy the
wait, so the call blocked forever. Since the wait ran with the GIL
released, even Python-level signal handlers could not interrupt it; the
only escape was SIGKILL.

Serialize stop() and awake() via the existing _awake_mutex, and in
awake() re-check under the mutex: if _stopping is set and we are not in
the _before_fork()/_after_fork() window (tracked by _skip_shutdown),
raise RuntimeError rather than risk blocking on a _served event nothing
will set.

Internal API; no release note.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: r1viollet

ddtrace/internal/_threads.cpp

@@ -698,14 +698,33 @@ PeriodicThread_awake(PeriodicThread* self, PyObject* Py_UNUSED(args))
         return NULL;
     }
 
+    bool was_stopped = false;
     {
         AllowThreads _(self->_state);
         std::lock_guard<std::mutex> lock(*self->_awake_mutex);
 
-        self->_served->clear();
-        self->_request->set(REQUEST_REASON_AWAKE);
+        // If stop() has fired and we are not in the fork-paused window, the
+        // worker has either already exited or is about to exit and will not
+        // be restarted. In that case a future _served->set() is not
+        // guaranteed, and clearing _served first would strand us in wait()
+        // forever. _skip_shutdown is set only by _before_fork() and cleared
+        // by _after_fork(), so (_stopping && !_skip_shutdown) captures the
+        // permanently-stopped case without blocking the legitimate
+        // _before_fork()/awake()/_after_fork() sequence (the new worker
+        // started by _after_fork() will consume the queued request).
+        if (self->_stopping && !self->_skip_shutdown) {
+            was_stopped = true;
+        } else {
+            self->_served->clear();
+            self->_request->set(REQUEST_REASON_AWAKE);
+
+            self->_served->wait();
+        }
+    }
 
-        self->_served->wait();
+    if (was_stopped) {
+        PyErr_SetString(PyExc_RuntimeError, "Periodic thread is stopped");
+        return NULL;
     }
 
     Py_RETURN_NONE;
@@ -720,8 +739,17 @@ PeriodicThread_stop(PeriodicThread* self, PyObject* Py_UNUSED(args))
         return NULL;
     }
 
-    self->_stopping = true;
-    self->_request->set(REQUEST_REASON_STOP);
+    // Synchronize with awake() via _awake_mutex: without this, stop() can
+    // flip _stopping (and the worker can observe it, exit, and set _served
+    // one last time) in between awake()'s recheck and its _served->clear(),
+    // leaving awake() blocked on a _served that nothing will set again.
+    {
+        AllowThreads _(self->_state);
+        std::lock_guard<std::mutex> lock(*self->_awake_mutex);
+
+        self->_stopping = true;
+        self->_request->set(REQUEST_REASON_STOP);
+    }
 
     Py_RETURN_NONE;
 }

tests/internal/test_periodic.py

@@ -48,6 +48,28 @@ def _run_periodic():
     t.join()
 
 
+def test_periodic_awake_after_stop_raises_not_hangs():
+    """Regression: awake() after a completed stop() used to block forever.
+
+    The native awake() cleared _served and then waited on it, but the worker
+    thread had already set _served one last time during its cleanup and was
+    gone — nothing would set _served again. The wait happened with the GIL
+    released, so not even a Python-level SIGALRM handler could interrupt it;
+    the only escape was SIGKILL.
+
+    After the fix, awake() rechecks _stopping under the awake mutex (with
+    stop() acquiring the same mutex) and raises RuntimeError instead of
+    blocking.
+    """
+    t = periodic.PeriodicThread(60.0, lambda: None)
+    t.start()
+    t.stop()
+    t.join()  # fully drained
+
+    with pytest.raises(RuntimeError):
+        t.awake()
+
+
 def test_periodic_error():
     x = {"OK": False}
 
@@ -351,7 +373,11 @@ def _get_native_thread_name():
             # Use pthread_getname_np
             pthread = ctypes.CDLL("/usr/lib/libpthread.dylib")
             pthread_getname_np = pthread.pthread_getname_np
-            pthread_getname_np.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_size_t]
+            pthread_getname_np.argtypes = [
+                ctypes.c_void_p,
+                ctypes.c_char_p,
+                ctypes.c_size_t,
+            ]
             pthread_getname_np.restype = ctypes.c_int
 
             # Get current thread handle
@@ -464,14 +490,20 @@ def _capture_native_name():
     t1.join()
 
     if native_name[0] is not None:
-        assert native_name[0] == "ShortName", f"Expected 'ShortName', got '{native_name[0]}'"
+        assert native_name[0] == "ShortName", (
+            f"Expected 'ShortName', got '{native_name[0]}'"
+        )
 
     # Test 2: Long name with module:class format
     # On Linux (15 char limit), should keep "StackCollectorThread" -> truncated to "StackCollectorT"
     # On macOS (63 char limit), should keep the full class name "StackCollectorThread"
     thread_started.clear()
     native_name[0] = None
-    t2 = periodic.PeriodicThread(0.1, _capture_native_name, name="ddtrace.profiling.collector:StackCollectorThread")
+    t2 = periodic.PeriodicThread(
+        0.1,
+        _capture_native_name,
+        name="ddtrace.profiling.collector:StackCollectorThread",
+    )
     t2.start()
     thread_started.wait()
     t2.stop()
@@ -480,7 +512,9 @@ def _capture_native_name():
     if native_name[0] is not None:
         if system == "Linux":
             # Linux truncates to 15 characters
-            assert native_name[0] == "StackCollectorT", f"Expected 'StackCollectorT' on Linux, got '{native_name[0]}'"
+            assert native_name[0] == "StackCollectorT", (
+                f"Expected 'StackCollectorT' on Linux, got '{native_name[0]}'"
+            )
         elif system == "Darwin":
             # macOS can fit the full class name
             assert native_name[0].endswith("StackCollectorThread"), (
@@ -490,7 +524,9 @@ def _capture_native_name():
     # Test 3: Long name without colon (should truncate from start)
     thread_started.clear()
     native_name[0] = None
-    t3 = periodic.PeriodicThread(0.1, _capture_native_name, name="VeryLongThreadNameWithoutColonSeparator")
+    t3 = periodic.PeriodicThread(
+        0.1, _capture_native_name, name="VeryLongThreadNameWithoutColonSeparator"
+    )
     t3.start()
     thread_started.wait()
     t3.stop()
@@ -499,7 +535,9 @@ def _capture_native_name():
     if native_name[0] is not None:
         if system == "Linux":
             # Should truncate from the start to 15 characters
-            assert native_name[0] == "VeryLongThreadN", f"Expected 'VeryLongThreadN' on Linux, got '{native_name[0]}'"
+            assert native_name[0] == "VeryLongThreadN", (
+                f"Expected 'VeryLongThreadN' on Linux, got '{native_name[0]}'"
+            )
         elif system == "Darwin":
             # macOS limit is 63, name is 41 chars, should fit fully
             assert native_name[0] == "VeryLongThreadNameWithoutColonSeparator", (
@@ -510,7 +548,9 @@ def _capture_native_name():
     # "module:ClassNameFit" on Linux should become "ClassNameFit" (12 chars, fits)
     thread_started.clear()
     native_name[0] = None
-    t4 = periodic.PeriodicThread(0.1, _capture_native_name, name="some.module:ClassNameFit")
+    t4 = periodic.PeriodicThread(
+        0.1, _capture_native_name, name="some.module:ClassNameFit"
+    )
     t4.start()
     thread_started.wait()
     t4.stop()