← Back
Editing: worker.py
import asyncio import os import threading from abc import ABC, abstractmethod from time import sleep, time from typing import TYPE_CHECKING from sentry_sdk._queue import FullError, Queue from sentry_sdk.consts import DEFAULT_QUEUE_SIZE from sentry_sdk.utils import logger, mark_sentry_task_internal if TYPE_CHECKING: from typing import Any, Callable, Optional _TERMINATOR = object() class Worker(ABC): """Base class for all workers.""" @property @abstractmethod def is_alive(self) -> bool: """Whether the worker is alive and running.""" pass @abstractmethod def kill(self) -> None: """Kill the worker. It will not process any more events.""" pass def flush( self, timeout: float, callback: "Optional[Callable[[int, float], Any]]" = None ) -> None: """Flush the worker, blocking until done or timeout is reached.""" return None @abstractmethod def full(self) -> bool: """Whether the worker's queue is full.""" pass @abstractmethod def submit(self, callback: "Callable[[], Any]") -> bool: """Schedule a callback. Returns True if queued, False if full.""" pass class BackgroundWorker(Worker): def __init__(self, queue_size: int = DEFAULT_QUEUE_SIZE) -> None: self._queue: "Queue" = Queue(queue_size) self._lock = threading.Lock() self._thread: "Optional[threading.Thread]" = None self._thread_for_pid: "Optional[int]" = None @property def is_alive(self) -> bool: if self._thread_for_pid != os.getpid(): return False if not self._thread: return False return self._thread.is_alive() def _ensure_thread(self) -> None: if not self.is_alive: self.start() def _timed_queue_join(self, timeout: float) -> bool: deadline = time() + timeout queue = self._queue queue.all_tasks_done.acquire() try: while queue.unfinished_tasks: delay = deadline - time() if delay <= 0: return False queue.all_tasks_done.wait(timeout=delay) return True finally: queue.all_tasks_done.release() def start(self) -> None: with self._lock: if not self.is_alive: self._thread = threading.Thread( target=self._target, name="sentry-sdk.BackgroundWorker" ) self._thread.daemon = True try: self._thread.start() self._thread_for_pid = os.getpid() except RuntimeError: # At this point we can no longer start because the interpreter # is already shutting down. Sadly at this point we can no longer # send out events. self._thread = None def kill(self) -> None: """ Kill worker thread. Returns immediately. Not useful for waiting on shutdown for events, use `flush` for that. """ logger.debug("background worker got kill request") with self._lock: if self._thread: try: self._queue.put_nowait(_TERMINATOR) except FullError: logger.debug("background worker queue full, kill failed") self._thread = None self._thread_for_pid = None def flush(self, timeout: float, callback: "Optional[Any]" = None) -> None: logger.debug("background worker got flush request") with self._lock: if self.is_alive and timeout > 0.0: self._wait_flush(timeout, callback) logger.debug("background worker flushed") def full(self) -> bool: return self._queue.full() def _wait_flush(self, timeout: float, callback: "Optional[Any]") -> None: initial_timeout = min(0.1, timeout) if not self._timed_queue_join(initial_timeout): pending = self._queue.qsize() + 1 logger.debug("%d event(s) pending on flush", pending) if callback is not None: callback(pending, timeout) if not self._timed_queue_join(timeout - initial_timeout): pending = self._queue.qsize() + 1 logger.error("flush timed out, dropped %s events", pending) def submit(self, callback: "Callable[[], Any]") -> bool: self._ensure_thread() try: self._queue.put_nowait(callback) return True except FullError: return False def _target(self) -> None: while True: callback = self._queue.get() try: if callback is _TERMINATOR: break try: callback() except Exception: logger.error("Failed processing job", exc_info=True) finally: self._queue.task_done() sleep(0) class AsyncWorker(Worker): def __init__(self, queue_size: int = DEFAULT_QUEUE_SIZE) -> None: self._queue: "Optional[asyncio.Queue[Any]]" = None self._queue_size = queue_size self._task: "Optional[asyncio.Task[None]]" = None # Event loop needs to remain in the same process self._task_for_pid: "Optional[int]" = None self._loop: "Optional[asyncio.AbstractEventLoop]" = None # Track active callback tasks so they have a strong reference and can be cancelled on kill self._active_tasks: "set[asyncio.Task[None]]" = set() @property def is_alive(self) -> bool: if self._task_for_pid != os.getpid(): return False if not self._task or not self._loop: return False return self._loop.is_running() and not self._task.done() def kill(self) -> None: if self._task: # Cancel the main consumer task to prevent duplicate consumers self._task.cancel() # Also cancel any active callback tasks # Avoid modifying the set while cancelling tasks tasks_to_cancel = set(self._active_tasks) for task in tasks_to_cancel: task.cancel() self._active_tasks.clear() self._loop = None self._task = None self._task_for_pid = None def start(self) -> None: if not self.is_alive: try: self._loop = asyncio.get_running_loop() # Always create a fresh queue on start to avoid stale items self._queue = asyncio.Queue(maxsize=self._queue_size) with mark_sentry_task_internal(): self._task = self._loop.create_task(self._target()) self._task_for_pid = os.getpid() except RuntimeError: # There is no event loop running logger.warning("No event loop running, async worker not started") self._loop = None self._task = None self._task_for_pid = None def full(self) -> bool: if self._queue is None: return True return self._queue.full() def _ensure_task(self) -> None: if not self.is_alive: self.start() async def _wait_flush( self, timeout: float, callback: "Optional[Any]" = None ) -> None: if not self._loop or not self._loop.is_running() or self._queue is None: return initial_timeout = min(0.1, timeout) # Timeout on the join try: await asyncio.wait_for(self._queue.join(), timeout=initial_timeout) except asyncio.TimeoutError: pending = self._queue.qsize() + len(self._active_tasks) logger.debug("%d event(s) pending on flush", pending) if callback is not None: callback(pending, timeout) try: remaining_timeout = timeout - initial_timeout await asyncio.wait_for(self._queue.join(), timeout=remaining_timeout) except asyncio.TimeoutError: pending = self._queue.qsize() + len(self._active_tasks) logger.error("flush timed out, dropped %s events", pending) def flush( # type: ignore[override] self, timeout: float, callback: "Optional[Any]" = None ) -> "Optional[asyncio.Task[None]]": if self.is_alive and timeout > 0.0 and self._loop and self._loop.is_running(): with mark_sentry_task_internal(): return self._loop.create_task(self._wait_flush(timeout, callback)) return None def submit(self, callback: "Callable[[], Any]") -> bool: self._ensure_task() if self._queue is None: return False try: self._queue.put_nowait(callback) return True except asyncio.QueueFull: return False async def _target(self) -> None: if self._queue is None: return try: while True: callback = await self._queue.get() if callback is _TERMINATOR: self._queue.task_done() break # Firing tasks instead of awaiting them allows for concurrent requests with mark_sentry_task_internal(): task = asyncio.create_task(self._process_callback(callback)) # Create a strong reference to the task so it can be cancelled on kill # and does not get garbage collected while running self._active_tasks.add(task) # Capture queue ref at dispatch time so done callbacks use the # correct queue even if kill()/start() replace self._queue. queue_ref = self._queue task.add_done_callback(lambda t: self._on_task_complete(t, queue_ref)) # Yield to let the event loop run other tasks await asyncio.sleep(0) except asyncio.CancelledError: pass # Expected during kill() async def _process_callback(self, callback: "Callable[[], Any]") -> None: # Callback is an async coroutine, need to await it await callback() def _on_task_complete( self, task: "asyncio.Task[None]", queue: "Optional[asyncio.Queue[Any]]" = None, ) -> None: try: task.result() except asyncio.CancelledError: pass # Task was cancelled, expected during shutdown except Exception: logger.error("Failed processing job", exc_info=True) finally: # Mark the task as done and remove it from the active tasks set # Use the queue reference captured at dispatch time, not self._queue, # to avoid calling task_done() on a different queue after kill()/start(). if queue is not None: queue.task_done() self._active_tasks.discard(task)
Save File
Cancel