refactor: native fork-safe threads (#14163)

We refactor the native periodic thread implementation to be fork-safe
whereby all such threads that are running at the time of a fork are
automatically stopped before the fork, then restarted after it. We also
take care to avoid stopping and restarting threads in the parent process
if we detect an immediate call to fork again.

Some of the implications of this change are that there is no longer the
need for fork-safe synchronisation objects, such as Locks and Events.
This is because all (ddtrace) threads are guaranteed to be stopped
before a fork, and restarted afterwards. There is also no longer the
need to manually recreate threads, as currently done in many places. The
only thing that needs to be taken care of is to ensure that the state of
the periodic services is as expected after a fork. For example, many
products have the need to avoid sending duplicate values from different
processes. This can be achieved by either subclassing from
`ForksafeAwakeablePeriodicService` and implementing the `reset` method
to let any periodic service to automatically trigger the reset logic on
forks (the preferred way), or by the current approach of registering a
fork-safe hook.

## Performance Analysis

Stopping and restarting threads can be expensive but gives good
fork-safety guarantees. Because threads have to be joined before the
fork can continue, a forked process might get delayed by a thread
currently busy on I/O. Because our periodic threads run on periods that
are O(1s), we can expect these unfortunate event to be quite rare.
Besides, it is quite likely that a process forks at the very beginning
of its execution, where it is unlikely that the periodic threads have
had a chance to trigger their first periodic action.

The timer set around close `fork` invocation should reduce delays even
more by preventing the parent process from stopping and restarting
threads in between `fork` calls. To give some number, the invocation of
1000 forks in a tight loop with the tracer and the profiler enabled
require ~1.2s to complete on an M1. In comparison, the same loop without
`ddtrace` takes 0.3s. The same loop with stop/restart in between forks
takes ~1.5s.

The wall-time view through a profiler gives the following picture:

<img width="1490" height="350" alt="Screenshot 2025-07-29 at 11 27 49"
src="https://github.com/user-attachments/assets/8e336a9e-96b0-4c96-be62-f76cac8cd8e0"
/>

A good portion of the overhead comes from the tracing of `fork` itself,
which could be improved by moving it to the native layer in follow-up
work.

## Checklist
- [ ] PR author has checked that all the criteria below are met
- The PR description includes an overview of the change
- The PR description articulates the motivation for the change
- The change includes tests OR the PR description describes a testing
strategy
- The PR description notes risks associated with the change, if any
- Newly-added code is easy to change
- The change follows the [library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
- The change includes or references documentation updates if necessary
- Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist
- [ ] Reviewer has checked that all the criteria below are met 
- Title is accurate
- All changes are related to the pull request's stated goal
- Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- Testing strategy adequately addresses listed risks
- Newly-added code is easy to change
- Release note makes sense to a user of the library
- If necessary, author has acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment
- Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)

---------

Co-authored-by: Alberto Vara <alberto.vara@datadoghq.com>
Co-authored-by: Munir Abdinur <munir.abdinur@datadoghq.com>
Co-authored-by: vianney <vianney.ruhlmann@datadoghq.com>
Co-authored-by: Brett Langdon <brett.langdon@datadoghq.com>
Co-authored-by: Gyuheon Oh <gyuheon.oh@datadoghq.com>
Co-authored-by: Gyuheon Oh <102937919+gyuheon0h@users.noreply.github.com>
Co-authored-by: Thomas Kowalski <thomas.kowalski@datadoghq.com>
Co-authored-by: Juanjo Alvarez Martinez <juanjo.alvarezmartinez@datadoghq.com>

Author: 8 people

ddtrace/_trace/context.py

@@ -1,6 +1,5 @@
 import base64
 import re
-import threading
 from typing import Any
 from typing import Optional
 from typing import Text
@@ -14,6 +13,7 @@
 from ddtrace.internal.constants import W3C_TRACEPARENT_KEY
 from ddtrace.internal.constants import W3C_TRACESTATE_KEY
 from ddtrace.internal.logger import get_logger
+from ddtrace.internal.threads import RLock
 from ddtrace.internal.utils.http import w3c_get_dd_list_member as _w3c_get_dd_list_member
 
 
@@ -60,7 +60,7 @@ def __init__(
         sampling_priority: Optional[float] = None,
         meta: Optional[dict[str, str]] = None,
         metrics: Optional[dict[str, NumericType]] = None,
-        lock: Optional[threading.RLock] = None,
+        lock: Optional[RLock] = None,
         span_links: Optional[list[SpanLink]] = None,
         baggage: Optional[dict[str, Any]] = None,
         is_remote: bool = True,
@@ -89,7 +89,7 @@ def __init__(
             # DEV: A `forksafe.RLock` is not necessary here since Contexts
             # are recreated by the tracer after fork
             # https://github.com/DataDog/dd-trace-py/blob/a1932e8ddb704d259ea8a3188d30bf542f59fd8d/ddtrace/tracer.py#L489-L508
-            self._lock = threading.RLock()
+            self._lock = RLock()
 
     def __getstate__(self) -> _ContextState:
         return (
@@ -116,7 +116,7 @@ def __setstate__(self, state: _ContextState) -> None:
             self._reactivate,
         ) = state
         # We cannot serialize and lock, so we must recreate it unless we already have one
-        self._lock = threading.RLock()
+        self._lock = RLock()
 
     def __enter__(self) -> "Context":
         self._lock.acquire()

ddtrace/contrib/internal/ray/span_manager.py

@@ -16,8 +16,8 @@
 from ddtrace.constants import ERROR_MSG
 from ddtrace.constants import SPAN_KIND
 from ddtrace.ext import SpanKind
-from ddtrace.internal.forksafe import Lock
 from ddtrace.internal.logger import get_logger
+from ddtrace.internal.threads import Lock
 
 from .constants import DD_PARTIAL_VERSION
 from .constants import DD_WAS_LONG_RUNNING

ddtrace/contrib/internal/subprocess/patch.py

@@ -15,10 +15,10 @@
 from ddtrace.contrib.internal.subprocess.constants import COMMANDS
 from ddtrace.ext import SpanTypes
 from ddtrace.internal import core
-from ddtrace.internal.forksafe import RLock
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.settings._config import config
 from ddtrace.internal.settings.asm import config as asm_config
+from ddtrace.internal.threads import RLock
 from ddtrace.trace import tracer
 
 

ddtrace/debugging/_encoding.py

@@ -16,10 +16,10 @@
 from ddtrace.debugging._config import di_config
 from ddtrace.debugging._signal.log import LogSignal
 from ddtrace.debugging._signal.snapshot import Snapshot
-from ddtrace.internal import forksafe
 from ddtrace.internal import process_tags
 from ddtrace.internal._encoding import BufferFull
 from ddtrace.internal.logger import get_logger
+from ddtrace.internal.threads import RLock
 from ddtrace.internal.utils.formats import format_trace_id
 
 
@@ -322,7 +322,7 @@ def __init__(
     ) -> None:
         self._encoder = encoder
         self._buffer = JsonBuffer(buffer_size)
-        self._lock = forksafe.RLock()
+        self._lock = RLock()
         self._on_full = on_full
         self.count = 0
         self.max_size = buffer_size - self._buffer.size

ddtrace/debugging/_origin/span.py

@@ -17,9 +17,9 @@
 from ddtrace.debugging._signal.snapshot import Snapshot
 from ddtrace.debugging._uploader import SignalUploader
 from ddtrace.debugging._uploader import UploaderProduct
-from ddtrace.internal.forksafe import Lock
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.safety import _isinstance
+from ddtrace.internal.threads import Lock
 from ddtrace.internal.wrapping.context import LazyWrappingContext
 
 

ddtrace/debugging/_probe/registry.py

@@ -6,8 +6,8 @@
 from ddtrace.debugging._probe.model import Probe
 from ddtrace.debugging._probe.model import ProbeLocationMixin
 from ddtrace.debugging._probe.status import ProbeStatusLogger
-from ddtrace.internal import forksafe
 from ddtrace.internal.logger import get_logger
+from ddtrace.internal.threads import RLock
 
 
 logger = get_logger(__name__)
@@ -66,7 +66,7 @@ def __init__(self, status_logger: ProbeStatusLogger, *args: Any, **kwargs: Any)
         # Used to keep track of probes pending installation
         self._pending: dict[str, list[Probe]] = defaultdict(list)
 
-        self._lock = forksafe.RLock()
+        self._lock = RLock()
 
     def register(self, *probes: Probe) -> None:
         """Register a probe."""

ddtrace/internal/_threads.cpp

@@ -244,6 +244,12 @@ class Event
         return _cond.wait_for(lock, timeout, [this]() { return _set; });
     }
 
+    bool wait(std::chrono::time_point<std::chrono::steady_clock> until)
+    {
+        std::unique_lock<std::mutex> lock(_mutex);
+        return _cond.wait_until(lock, until, [this]() { return _set; });
+    }
+
     void clear()
     {
         std::lock_guard<std::mutex> lock(_mutex);
@@ -273,7 +279,9 @@ typedef struct periodic_thread
 
     bool _stopping;
     bool _atexit;
-    bool _after_fork;
+    bool _skip_shutdown;
+
+    std::chrono::time_point<std::chrono::steady_clock> _next_call_time;
 
     std::unique_ptr<Event> _started;
     std::unique_ptr<Event> _stopped;
@@ -304,8 +312,6 @@ static PyMemberDef PeriodicThread_members[] = {
       0,
       "whether to ignore the thread for profiling" },
 
-    { "_is_after_fork", T_BOOL, offsetof(PeriodicThread, _after_fork), READONLY, "whether the thread is after fork" },
-
     { NULL } /* Sentinel */
 };
 
@@ -341,7 +347,7 @@ PeriodicThread_init(PeriodicThread* self, PyObject* args, PyObject* kwargs)
 
     self->_stopping = false;
     self->_atexit = false;
-    self->_after_fork = false;
+    self->_skip_shutdown = false;
 
     self->_started = std::make_unique<Event>();
     self->_stopped = std::make_unique<Event>();
@@ -383,16 +389,8 @@ PeriodicThread__on_shutdown(PeriodicThread* self)
 
 // ----------------------------------------------------------------------------
 static PyObject*
-PeriodicThread_start(PeriodicThread* self, PyObject* args)
+PeriodicThread_start(PeriodicThread* self, PyObject* Py_UNUSED(args))
 {
-    // After fork, the child process should not restart threads that were running in the parent
-    // until properly reinitialized through forksafe handlers. This prevents crashes when
-    // pthread_create is called before threading state is safe (e.g., in uvloop's _after_fork).
-    // Check this first because after fork, self->_thread is non-null but the thread doesn't exist.
-    if (self->_after_fork) {
-        Py_RETURN_NONE;
-    }
-
     if (self->_thread != nullptr) {
         PyErr_SetString(PyExc_RuntimeError, "Thread already started");
         return NULL;
@@ -401,6 +399,11 @@ PeriodicThread_start(PeriodicThread* self, PyObject* args)
     if (self->_stopping)
         Py_RETURN_NONE;
 
+    // Initialize the next call time to the current time plus the interval.
+    // This ensures that the first call happens after the specified interval.
+    self->_next_call_time =
+      std::chrono::steady_clock::now() + std::chrono::milliseconds((long long)(self->interval * 1000));
+
     // Start the thread
     self->_thread = std::make_unique<std::thread>([self]() {
         GILGuard _gil;
@@ -423,15 +426,14 @@ PeriodicThread_start(PeriodicThread* self, PyObject* args)
         self->_started->set();
 
         bool error = false;
-        auto interval = std::chrono::milliseconds((long long)(self->interval * 1000));
         if (self->_no_wait_at_start)
             self->_request->set();
 
         while (!self->_stopping) {
             {
                 AllowThreads _;
 
-                if (self->_request->wait(interval)) {
+                if (self->_request->wait(self->_next_call_time)) {
                     if (self->_stopping)
                         break;
 
@@ -454,6 +456,9 @@ PeriodicThread_start(PeriodicThread* self, PyObject* args)
                 break;
             }
 
+            self->_next_call_time =
+              std::chrono::steady_clock::now() + std::chrono::milliseconds((long long)(self->interval * 1000));
+
             // If this came from a request mark it as served
             self->_served->set();
         }
@@ -464,7 +469,7 @@ PeriodicThread_start(PeriodicThread* self, PyObject* args)
 
         // Run the shutdown callback if there was no error and we are not
         // at Python shutdown.
-        if (!self->_atexit && !error && self->_on_shutdown != Py_None) {
+        if (!self->_atexit && !error && self->_on_shutdown != Py_None && !self->_skip_shutdown) {
 #if PY_VERSION_HEX >= 0x30d0000
             if (!Py_IsFinalizing()) {
 #else
@@ -474,6 +479,8 @@ PeriodicThread_start(PeriodicThread* self, PyObject* args)
             }
         }
 
+        PyDict_DelItem(_periodic_threads, self->ident);
+
         // Notify the join method that the thread has stopped
         self->_stopped->set();
     });
@@ -493,14 +500,14 @@ PeriodicThread_start(PeriodicThread* self, PyObject* args)
 
 // ----------------------------------------------------------------------------
 static PyObject*
-PeriodicThread_awake(PeriodicThread* self, PyObject* args)
+PeriodicThread_awake(PeriodicThread* self, PyObject* Py_UNUSED(args))
 {
     if (self->_thread == nullptr) {
         PyErr_SetString(PyExc_RuntimeError, "Thread not started");
         return NULL;
     }
 
-    if (!self->_after_fork) {
+    {
         AllowThreads _;
         std::lock_guard<std::mutex> lock(*self->_awake_mutex);
 
@@ -514,19 +521,15 @@ PeriodicThread_awake(PeriodicThread* self, PyObject* args)
 
 // ----------------------------------------------------------------------------
 static PyObject*
-PeriodicThread_stop(PeriodicThread* self, PyObject* args)
+PeriodicThread_stop(PeriodicThread* self, PyObject* Py_UNUSED(args))
 {
     if (self->_thread == nullptr) {
         PyErr_SetString(PyExc_RuntimeError, "Thread not started");
         return NULL;
     }
 
-    if (!self->_after_fork) {
-        // The thread is no longer running so it makes no sense to stop it.
-        // Attempting to acquire the Event lock could deadlock.
-        self->_stopping = true;
-        self->_request->set();
-    }
+    self->_stopping = true;
+    self->_request->set();
 
     Py_RETURN_NONE;
 }
@@ -545,11 +548,6 @@ PeriodicThread_join(PeriodicThread* self, PyObject* args, PyObject* kwargs)
         return NULL;
     }
 
-    if (self->_after_fork) {
-        // The thread is no longer running so it makes no sense to join it.
-        Py_RETURN_NONE;
-    }
-
     PyObject* timeout = Py_None;
 
     if (args != NULL && kwargs != NULL) {
@@ -586,7 +584,7 @@ PeriodicThread_join(PeriodicThread* self, PyObject* args, PyObject* kwargs)
 
 // ----------------------------------------------------------------------------
 static PyObject*
-PeriodicThread__atexit(PeriodicThread* self, PyObject* args)
+PeriodicThread__atexit(PeriodicThread* self, PyObject* Py_UNUSED(args))
 {
     self->_atexit = true;
 
@@ -601,9 +599,32 @@ PeriodicThread__atexit(PeriodicThread* self, PyObject* args)
 
 // ----------------------------------------------------------------------------
 static PyObject*
-PeriodicThread__after_fork(PeriodicThread* self, PyObject* args)
+PeriodicThread__after_fork(PeriodicThread* self, PyObject* Py_UNUSED(args))
 {
-    self->_after_fork = true;
+    self->_thread = nullptr;
+
+    self->_stopping = false;
+    self->_atexit = false;
+    self->_skip_shutdown = false;
+
+    // We don't clear the request event because we might have pending awake
+    // requests.
+    self->_started->clear();
+    self->_stopped->clear();
+    self->_served->clear();
+
+    PeriodicThread_start(self, NULL);
+
+    Py_RETURN_NONE;
+}
+
+// ----------------------------------------------------------------------------
+static PyObject*
+PeriodicThread__before_fork(PeriodicThread* self, PyObject* Py_UNUSED(args))
+{
+    self->_skip_shutdown = true;
+
+    PeriodicThread_stop(self, NULL);
 
     Py_RETURN_NONE;
 }
@@ -661,8 +682,9 @@ static PyMethodDef PeriodicThread_methods[] = {
     { "join", (PyCFunction)PeriodicThread_join, METH_VARARGS | METH_KEYWORDS, "Join the thread" },
     /* Private */
     { "_atexit", (PyCFunction)PeriodicThread__atexit, METH_NOARGS, "Stop the thread at exit" },
-    { "_after_fork", (PyCFunction)PeriodicThread__after_fork, METH_NOARGS, "Mark the thread as after fork" },
-    { NULL } /* Sentinel */
+    { "_after_fork", (PyCFunction)PeriodicThread__after_fork, METH_NOARGS, "Refresh the thread after fork" },
+    { "_before_fork", (PyCFunction)PeriodicThread__before_fork, METH_NOARGS, "Prepare the thread for fork" },
+    { NULL, NULL, 0, NULL } /* Sentinel */
 };
 
 // ----------------------------------------------------------------------------

ddtrace/internal/_threads.pyi

@@ -1,7 +1,6 @@
 import typing as t
 
 class PeriodicThread:
-    _is_after_fork: bool
     name: str
     ident: int
     interval: float
@@ -20,5 +19,6 @@ class PeriodicThread:
     def awake(self) -> None: ...
     def _atexit(self) -> None: ...
     def _after_fork(self) -> None: ...
+    def _before_fork(self) -> None: ...
 
 periodic_threads: dict[int, PeriodicThread]

ddtrace/internal/datadog/profiling/stack/include/sampler.hpp

@@ -80,6 +80,9 @@ class Sampler
 
     // Delegates to the StackRenderer to clear its caches after fork
     void postfork_child();
+
+    // Restart the sampler after fork if it was running
+    void restart_after_fork();
 };
 
 } // namespace Datadog