Revert the changes to the span batcher - this needs to be a separate class from batcher but that will not be done in this set of changes. Update batcher deadlock fix to use weakmethod, to be consistent with the approach taken in monitor.py

Author: ericapisani

sentry_sdk/_batcher.py

@@ -38,24 +38,23 @@ def __init__(
 
         self._flusher: "Optional[threading.Thread]" = None
         self._flusher_pid: "Optional[int]" = None
+        self._reset_thread_state()
 
-        self_ref = weakref.ref(self)
+        # See https://github.com/getsentry/sentry-python/blob/051cc01640a29bfd64b1f1e2e3414c02f027dd1b/sentry_sdk/monitor.py#L41-L50
+        if hasattr(os, "register_at_fork"):
+            weak_reset = weakref.WeakMethod(self._reset_thread_state)
 
-        def _reset_thread_state() -> None:
-            batcher = self_ref()
+            def _reset_in_child() -> None:
+                method = weak_reset()
+                if method is not None:
+                    method()
 
-            if batcher is not None:
-                batcher._flusher = None
-                batcher._lock = threading.Lock()
-                batcher._flusher_pid = None
+            os.register_at_fork(after_in_child=_reset_in_child)
 
-        # Same as https://github.com/getsentry/sentry-python/issues/6148.
-        # If os.fork() runs while another thread holds self._lock,
-        # the child inherits the lock locked but the holding thread does
-        # not exist in the child, so the lock can never be released and
-        # _ensure_thread deadlocks forever.
-        if hasattr(os, "register_at_fork"):
-            os.register_at_fork(after_in_child=_reset_thread_state)
+    def _reset_thread_state(self) -> None:
+        self._flusher = None
+        self._lock = threading.Lock()
+        self._flusher_pid = None
 
     def _ensure_thread(self) -> bool:
         """For forking processes we might need to restart this thread.

sentry_sdk/_span_batcher.py

@@ -1,6 +1,9 @@
 from collections import defaultdict
 from datetime import datetime, timezone
-from typing import TYPE_CHECKING
+import os
+import threading
+from typing import TYPE_CHECKING, Optional
+import weakref
 
 from sentry_sdk._batcher import Batcher
 from sentry_sdk.envelope import Envelope, Item, PayloadRef
@@ -31,14 +34,41 @@ def __init__(
         capture_func: "Callable[[Envelope], None]",
         record_lost_func: "Callable[..., None]",
     ) -> None:
-        super().__init__(capture_func, record_lost_func)
         # Spans from different traces cannot be emitted in the same envelope
         # since the envelope contains a shared trace header. That's why we bucket
         # by trace_id, so that we can then send the buckets each in its own
         # envelope.
         # trace_id -> span buffer
         self._span_buffer: dict[str, list["StreamedSpan"]] = defaultdict(list)
         self._running_size: dict[str, int] = defaultdict(lambda: 0)
+        self._capture_func = capture_func
+        self._record_lost_func = record_lost_func
+        self._running = True
+        self._lock = threading.Lock()
+        self._active: "threading.local" = threading.local()
+
+        self._flush_event: "threading.Event" = threading.Event()
+
+        self._flusher: "Optional[threading.Thread]" = None
+        self._flusher_pid: "Optional[int]" = None
+
+        self._reset_thread_state()
+
+        # See https://github.com/getsentry/sentry-python/blob/051cc01640a29bfd64b1f1e2e3414c02f027dd1b/sentry_sdk/monitor.py#L41-L50
+        if hasattr(os, "register_at_fork"):
+            weak_reset = weakref.WeakMethod(self._reset_thread_state)
+
+            def _reset_in_child() -> None:
+                method = weak_reset()
+                if method is not None:
+                    method()
+
+            os.register_at_fork(after_in_child=_reset_in_child)
+
+    def _reset_thread_state(self) -> None:
+        self._flusher = None
+        self._lock = threading.Lock()
+        self._flusher_pid = None
 
     def add(self, span: "StreamedSpan") -> None:
         # Bail out if the current thread is already executing batcher code.