diff --git a/docs/sdk-conformance.md b/docs/sdk-conformance.md index 09583dc..ffd2d41 100644 --- a/docs/sdk-conformance.md +++ b/docs/sdk-conformance.md @@ -142,6 +142,11 @@ Concurrent calls share one fetch. and not to the full response. Only an address lookup can continue after the call stops waiting, until the system resolver gives up. Both SDKs measure the TTL and Retry-After in elapsed time, so a change of the system clock does not move them. +- Fork (Python): a process that `os.fork()` makes gets the copies of the parent, but not the fetch + threads of the parent. A call that starts in the child does not wait for a fetch of the parent. + If the call must fetch, it starts a new fetch. If a signal handler forks while a call waits for a + fetch, that call continues in the child. The call gets `timeout` 10 seconds after the start of + the fetch, because no thread of the child completes the fetch. - Failure: when a fetch fails, the SDKs keep the last copy for another TTL and send the error to `onError` / `on_error`. With no copy, the call gets the fallback. With no fallback, the call gets a `PromptError`. With no copy, the calls of the next 10 seconds (or the TTL, if it is shorter) diff --git a/sdks/python/README.md b/sdks/python/README.md index b785d10..bb9b109 100644 --- a/sdks/python/README.md +++ b/sdks/python/README.md @@ -219,6 +219,13 @@ pinned = telemetry_dev.get_prompt("support/reply", version=13) call returns it at once and fetches a new copy in a background thread. Only a call with no copy waits for a fetch, for 10 seconds at most, also when the network stops. Calls for the same prompt share one fetch. `get_prompt_async` runs the same function in a worker thread. +- A process that `os.fork()` makes keeps the copies of its parent. This includes a process of + `multiprocessing` with the `fork` start method. A call that starts in the child does not wait + for a fetch of the parent. If the call must fetch, it starts a new fetch. +- If a signal handler forks while a call waits for a fetch, that call continues in the child. No + thread of the child completes that fetch. Thus the call waits until 10 seconds after the start + of the fetch. Then the call returns the `fallback`, or raises `PromptError` with the code + `timeout`. - If a fetch fails, `get_prompt` returns the last copy that it fetched and reports the error through `on_error`. With no copy, it returns the `fallback` (`version` is `None` and `is_fallback` is `True`). With no fallback, it raises `PromptError` whose `code` tells why, for diff --git a/sdks/python/src/telemetry_dev/_prompts.py b/sdks/python/src/telemetry_dev/_prompts.py index 9673740..0ba6b9d 100644 --- a/sdks/python/src/telemetry_dev/_prompts.py +++ b/sdks/python/src/telemetry_dev/_prompts.py @@ -5,6 +5,7 @@ import copy import hashlib import json +import os import random import re import socket @@ -455,12 +456,57 @@ class _Entry: retry_at: float = 0.0 # Set when the running fetch ends. pending: threading.Event | None = None + # The process whose thread runs the fetch. A forked process gets the entry, but not the thread. + pending_pid: int = 0 # When calls stop waiting for the running fetch. deadline: float = 0.0 _cache: OrderedDict[tuple[str, str, str, str], _Entry] = OrderedDict() -_cache_lock = threading.Lock() +# Reentrant, so that a fork in a thread that holds the lock, for example in a signal handler, does +# not wait for itself in _lock_for_fork. _reset_in_child can replace it, so read it from the module +# each time. +_cache_lock = threading.RLock() +# Whether _lock_for_fork got the lock in this thread. os.fork() runs the after hooks also when a +# before hook raises, for example in a signal handler. +_forking = threading.local() + + +def _lock_for_fork() -> None: + _cache_lock.acquire() + _forking.locked = True + + +def _unlock_after_fork() -> None: + if getattr(_forking, "locked", False): + _forking.locked = False + _cache_lock.release() + + +def _reset_in_child() -> None: + """Unlock the cache, and drop the fetches of the parent: the child has none of its threads.""" + global _cache_lock + if getattr(_forking, "locked", False): + _forking.locked = False + _cache_lock.release() + else: + # The before hook did not get the lock, because a signal stopped its wait, or did not run, + # as in uWSGI with py-call-osafterfork. A thread of the parent can hold the lock, and only + # this thread exists here. + _cache_lock = threading.RLock() + for entry in _cache.values(): + entry.pending = None + + +# A fork that runs these hooks waits for a thread that holds the lock, so that the child gets no +# entry that a thread changed only in part. os.register_at_fork cannot remove a hook, and +# importlib.reload runs this module again with the same globals. Thus the hooks of the first run +# stay, and they find the lock and the cache when they run. Windows has no fork. +if hasattr(os, "register_at_fork") and "_fork_hooks" not in globals(): + _fork_hooks = True + os.register_at_fork( + before=_lock_for_fork, after_in_parent=_unlock_after_fork, after_in_child=_reset_in_child + ) def _load( @@ -481,15 +527,23 @@ def _load( entry = _cache[key] = _Entry() now = monotonic() ttl = ttl_seconds if entry.response is not None else min(ttl_seconds, _ERROR_TTL_S) - if now >= max(entry.checked_at + ttl, entry.retry_at) and entry.pending is None: - entry.pending = threading.Event() + if entry.pending is not None and entry.pending_pid != os.getpid(): + # A fork in C code that runs no hooks copied the fetch but not its thread. This check + # does not replace the lock. If a thread of the parent held the lock at that fork, a + # call in the child waits for the lock with no limit. + entry.pending = None + # A local, because a fork in this thread, as in a signal handler, can clear entry.pending. + pending = entry.pending + if now >= max(entry.checked_at + ttl, entry.retry_at) and pending is None: + pending = entry.pending = threading.Event() + entry.pending_pid = os.getpid() entry.deadline = now + _FETCH_TIMEOUT_S threading.Thread( - target=_refresh, args=(entry, entry.pending, fetch, on_error), daemon=True + target=_refresh, args=(entry, pending, fetch, on_error), daemon=True ).start() if entry.response is not None: return entry.response - pending, deadline = entry.pending, entry.deadline + deadline = entry.deadline if pending is not None and not pending.wait(deadline - monotonic()): raise _timeout() with _cache_lock: diff --git a/sdks/python/tests/otlp_capture.py b/sdks/python/tests/otlp_capture.py index 0649f72..35cfce9 100644 --- a/sdks/python/tests/otlp_capture.py +++ b/sdks/python/tests/otlp_capture.py @@ -33,7 +33,8 @@ class CaptureServer: status: int = 200 # The (status, JSON body) replies to GET requests, in order; the last one repeats. replies: list[tuple[int, object]] = field(default_factory=lambda: []) - # Seconds that the server waits before it answers a GET request. + # Seconds that the server waits before it answers a GET request. The server reads it before the + # request shows in requests, so that a test that sees the request can change it for the next. delay: float = 0 # The location header of GET replies, for a redirect status. location: str | None = None @@ -66,6 +67,7 @@ def do_POST(self) -> None: self.end_headers() def do_GET(self) -> None: + delay = capture.delay captured.append( CapturedRequest( path=self.requestline.split(" ")[1], @@ -73,7 +75,7 @@ def do_GET(self) -> None: body=b"", ) ) - time.sleep(capture.delay) + time.sleep(delay) status, reply = ( capture.replies[0] if len(capture.replies) == 1 else capture.replies.pop(0) ) diff --git a/sdks/python/tests/test_prompts.py b/sdks/python/tests/test_prompts.py index 892d676..764c4ef 100644 --- a/sdks/python/tests/test_prompts.py +++ b/sdks/python/tests/test_prompts.py @@ -2,17 +2,22 @@ import asyncio import contextvars +import importlib import itertools +import os +import signal import socket +import sys import threading import time +import traceback import urllib.error import urllib.request from collections.abc import Callable, Generator from datetime import datetime, timedelta, timezone from email.utils import format_datetime from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer -from types import SimpleNamespace +from types import FrameType, SimpleNamespace from typing import Any, NoReturn from urllib.parse import parse_qsl, urlsplit @@ -34,6 +39,12 @@ from .conftest import MakeClient from .otlp_capture import CaptureServer, otlp_capture_server +# Python 3.12+ gives a warning for fork() in a process with threads, and the capture server has one. +pytestmark = pytest.mark.filterwarnings( + "ignore:This process .* is multi-threaded:DeprecationWarning" +) +needs_fork = pytest.mark.skipif(not hasattr(os, "fork"), reason="needs os.fork") + def reply(**overrides: Any) -> dict[str, Any]: body: dict[str, Any] = { @@ -93,6 +104,57 @@ def wait_for(check: Callable[[], bool]) -> None: time.sleep(0.01) +def in_fork(child: Callable[[], str], fork: Callable[[], int] | None = None) -> str: + """Run ``child`` in a process that ``fork`` (by default ``os.fork``) makes, and return its + result, or the repr of its exception or of an exception in a fork hook of the prompt module. + The parent kills a child that does not end in 10 s, and the test fails.""" + # Python ignores an exception in a fork hook and only gives it to sys.unraisablehook. The + # hooks of OpenTelemetry raise after an earlier test discards its batch processor. + hook_errors: list[BaseException | None] = [] + + def record(unraisable: sys.UnraisableHookArgs) -> None: + frames = traceback.extract_tb(unraisable.exc_traceback) + if any(frame.filename == prompts_module.__file__ for frame in frames): + hook_errors.append(unraisable.exc_value) + + previous_hook = sys.unraisablehook + parent = os.getpid() + read_end, write_end = os.pipe() + sys.unraisablehook = record + try: + pid = (fork or os.fork)() + finally: + sys.unraisablehook = previous_hook + # Not pid == 0: a fork function that forks more than one time can return a pid in a child. + if os.getpid() != parent: + try: + os.close(read_end) + try: + result = repr(hook_errors[0]) if hook_errors else child() + except BaseException as exc: + result = repr(exc) + os.write(write_end, result.encode()) + finally: + # The child must not return into pytest. + os._exit(0) + os.close(write_end) + with os.fdopen(read_end, "rb") as reader: + deadline = time.monotonic() + 10 + while os.waitpid(pid, os.WNOHANG) == (0, 0): + if time.monotonic() > deadline: + os.kill(pid, signal.SIGKILL) + os.waitpid(pid, 0) + pytest.fail("The forked process did not end in 10 s") + time.sleep(0.01) + result = reader.read().decode() + assert hook_errors == [] + return result + + +def version_of(server: CaptureServer) -> int | None: + return get_prompt("support/reply", api_key="td_live_a", base_url=server.url).version + + @pytest.mark.parametrize( ("options", "query"), [ @@ -525,6 +587,245 @@ def close(self) -> None: assert timeouts[0] is not None and timeouts[0] <= 0.2 +@needs_fork +@pytest.mark.parametrize("reused_pid", [False, True]) +def test_a_forked_process_fetches_the_prompt_that_the_parent_was_fetching( + server: CaptureServer, monkeypatch: pytest.MonkeyPatch, reused_pid: bool +) -> None: + # This puts a limit on the wait of a child that waits for the fetch of the parent. + monkeypatch.setattr(prompts_module, "_FETCH_TIMEOUT_S", 3) + server.delay = 1 + versions: list[int | None] = [] + parent = threading.Thread(target=lambda: versions.append(version_of(server))) + parent.start() + wait_for(lambda: len(server.requests) == 1) + server.delay = 0 + parent_pid = os.getpid() + + def child() -> str: + if reused_pid: + # A reused pid: the child gets the id of the process that started the fetch. + monkeypatch.setattr(os, "getpid", lambda: parent_pid) + return str(version_of(server)) + + assert in_fork(child) == "12" + parent.join() + assert versions == [12] + # The fetch of the parent, and the fetch of the child. + assert len(server.requests) == 2 + + +def test_a_call_does_not_wait_for_a_fetch_that_another_process_started( + server: CaptureServer, +) -> None: + server.delay = 1 + first = threading.Thread(target=version_of, args=(server,)) + first.start() + wait_for(lambda: len(server.requests) == 1) + server.delay = 0 + # A fork that runs no fork hooks, as a fork in C code, copies the fetch of the parent but not + # the thread that runs it. Here this process is the child: the fetch gets the id of its parent. + (entry,) = prompts_module._cache.values() # pyright: ignore[reportPrivateUsage] + entry.pending_pid = os.getppid() + + assert version_of(server) == 12 + assert len(server.requests) == 2 + first.join() + + +@needs_fork +def test_a_forked_process_refreshes_the_copy_that_the_parent_was_refreshing( + server: CaptureServer, clock: list[float] +) -> None: + assert version_of(server) == 12 + clock[0] += 60 + server.replies = [(200, reply(version=13))] + server.delay = 1 + assert version_of(server) == 12 + wait_for(lambda: len(server.requests) == 2) + server.delay = 0 + + def child() -> str: + kept = version_of(server) + wait_for(lambda: version_of(server) == 13) + return str(kept) + + # The child returns the copy of the parent. Then its refresh gets version 13. + assert in_fork(child) == "12" + + +@needs_fork +@pytest.mark.filterwarnings("error::pytest.PytestUnhandledThreadExceptionWarning") +def test_a_fork_waits_for_a_thread_that_holds_the_cache_lock(server: CaptureServer) -> None: + held = threading.Event() + released_at: list[float] = [] + forked_at: list[float] = [] + errors: list[BaseException] = [] + + def hold() -> None: + try: + with prompts_module._cache_lock: # pyright: ignore[reportPrivateUsage] + held.set() + time.sleep(0.2) + released_at.append(time.monotonic()) + except BaseException as exc: + errors.append(exc) + + def fork() -> int: + pid = os.fork() + if pid != 0: + forked_at.append(time.monotonic()) + return pid + + holder = threading.Thread(target=hold) + holder.start() + held.wait() + # The child does not get a lock that a thread of the parent holds. + assert in_fork(lambda: str(version_of(server)), fork) == "12" + holder.join() + assert errors == [] + assert forked_at[0] > released_at[0] + + +@needs_fork +def test_a_fork_in_the_thread_that_holds_the_cache_lock_returns( + server: CaptureServer, monkeypatch: pytest.MonkeyPatch +) -> None: + # A process that still holds the lock after the fork waits for its fetch until the timeout. + monkeypatch.setattr(prompts_module, "_FETCH_TIMEOUT_S", 3) + + def fork_in_the_lock() -> int: + with prompts_module._cache_lock: # pyright: ignore[reportPrivateUsage] + return os.fork() + + def fork_and_fetch() -> str: + child = in_fork(lambda: str(version_of(server)), fork_in_the_lock) + return f"{child} {version_of(server)}" + + # The fork runs in a child, so that a fork that waits for itself stops only that child. + assert in_fork(fork_and_fetch) == "12 12" + + +@needs_fork +def test_a_fork_in_a_call_that_starts_a_fetch_returns_the_prompt( + server: CaptureServer, monkeypatch: pytest.MonkeyPatch +) -> None: + # A process that waits for a fetch that it does not have waits until the timeout. + monkeypatch.setattr(prompts_module, "_FETCH_TIMEOUT_S", 3) + pids: list[int] = [] + results: list[str] = [] + + class ForkingThread(threading.Thread): + def start(self) -> None: + # A fork in the call, as in a signal handler, after the call made its fetch and before + # the fetch thread starts. The child hook drops the fetch of the entry. + if not pids: + pids.append(os.fork()) + super().start() + + def fork() -> int: + with monkeypatch.context() as patch: + # Only the threads of the prompt module: a fork hook of OpenTelemetry starts a thread. + patch.setattr( + prompts_module, + "threading", + SimpleNamespace(**{**vars(threading), "Thread": ForkingThread}), + ) + try: + results.append(str(version_of(server))) + except Exception as exc: + # The child must return the pid to in_fork, and not raise into pytest. + results.append(repr(exc)) + return pids[0] + + # The parent and the child each start the fetch thread, and each call returns its result. + assert in_fork(lambda: results[0], fork) == "12" + assert results == ["12"] + + +@needs_fork +def test_a_fork_after_a_reload_of_the_prompt_module_returns( + server: CaptureServer, monkeypatch: pytest.MonkeyPatch +) -> None: + saved = dict(prompts_module.__dict__) + importlib.reload(prompts_module) + try: + # The reload sets the timeout again. A process that still holds the lock after the fork + # waits for its fetch until the timeout. + monkeypatch.setattr(prompts_module, "_FETCH_TIMEOUT_S", 3) + assert in_fork(lambda: str(version_of(server))) == "12" + assert version_of(server) == 12 + finally: + # The reload gave the module new classes and a new cache. + prompts_module.__dict__.update(saved) + + +@needs_fork +@pytest.mark.skipif(not hasattr(signal, "setitimer"), reason="needs signal.setitimer") +@pytest.mark.parametrize("before_hook", ["stopped by a signal", "not run"]) +def test_the_fork_hooks_handle_a_lock_that_the_before_hook_did_not_get( + server: CaptureServer, monkeypatch: pytest.MonkeyPatch, before_hook: str +) -> None: + class Interrupted(Exception): + pass + + lock = prompts_module._cache_lock # pyright: ignore[reportPrivateUsage] + # The child hook replaces the lock. monkeypatch puts it back. + monkeypatch.setattr(prompts_module, "_cache_lock", lock) + held, done = threading.Event(), threading.Event() + errors: list[BaseException] = [] + + def hold() -> None: + try: + with lock: + held.set() + # The fork must end while this thread holds the lock. If no signal stops the + # before hook, the hook waits for this thread, and this assert fails after 5 s. + assert done.wait(5) + except BaseException as exc: + errors.append(exc) + + def interrupt(signum: int, frame: FrameType | None) -> NoReturn: + raise Interrupted + + def fork_in_a_signal() -> int: + # SIGALRM raises while the before hook waits for the lock. os.fork() gives the error to + # sys.unraisablehook, forks, and then runs the after hooks. + in_fork_hook = sys.unraisablehook + + def record(unraisable: sys.UnraisableHookArgs) -> None: + if not isinstance(unraisable.exc_value, Interrupted): + in_fork_hook(unraisable) + + previous = signal.signal(signal.SIGALRM, interrupt) + sys.unraisablehook = record + signal.setitimer(signal.ITIMER_REAL, 0.1) + try: + return os.fork() + finally: + signal.setitimer(signal.ITIMER_REAL, 0) + sys.unraisablehook = in_fork_hook + signal.signal(signal.SIGALRM, previous) + + holder = threading.Thread(target=hold) + holder.start() + held.wait() + try: + if before_hook == "stopped by a signal": + # In the child, the holder is a thread of the parent that does not exist there. + assert in_fork(lambda: str(version_of(server)), fork_in_a_signal) == "12" + else: + # uWSGI with py-call-osafterfork forks in C and runs only the child hook. + prompts_module._reset_in_child() # pyright: ignore[reportPrivateUsage] + child_lock = prompts_module._cache_lock # pyright: ignore[reportPrivateUsage] + assert child_lock.acquire(blocking=False) + child_lock.release() + finally: + done.set() + holder.join() + assert errors == [] + + def test_get_prompt_rejects_a_prompt_of_the_other_type_and_bad_options( server: CaptureServer, ) -> None: