From a0699b96f7cac6dc4d35359ccc5457d554edd8e6 Mon Sep 17 00:00:00 2001 From: ephraimduncan Date: Thu, 29 Oct 2026 13:54:49 +0000 Subject: [PATCH 1/5] fix(sdk-python): reset prompt fetches in a forked process A process that os.fork() makes kept the pending fetch events of the parent, but not the fetch threads. In the child, a call with no copy waited 10 s and then got a timeout error. A kept copy never refreshed. A cache lock that a different thread held at the fork stayed locked in the child, and each get_prompt call in the child waited for all time. Fork hooks now hold the cache lock during fork(). In the child, the hooks remove the pending fetch events and release the lock. Thus a call that must fetch starts a new fetch. The child keeps the copies of the parent. Closes #98 --- docs/sdk-conformance.md | 4 +- sdks/python/README.md | 3 + sdks/python/src/telemetry_dev/_prompts.py | 25 +++++ sdks/python/tests/test_prompts.py | 116 ++++++++++++++++++++++ 4 files changed, 147 insertions(+), 1 deletion(-) diff --git a/docs/sdk-conformance.md b/docs/sdk-conformance.md index 09583dc..1bc6385 100644 --- a/docs/sdk-conformance.md +++ b/docs/sdk-conformance.md @@ -141,7 +141,9 @@ Concurrent calls share one fetch. its connection down at that time, because the urllib timeout applies to each socket operation 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. + a change of the system clock does not move them. A process that `os.fork()` makes gets the + copies of the parent, but not the fetch threads of the parent. In the child, a fetch that did not + end before the fork counts as not started. Thus a call that must fetch starts a new 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..07f87f1 100644 --- a/sdks/python/README.md +++ b/sdks/python/README.md @@ -219,6 +219,9 @@ 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. The process does not wait for a fetch that did + not end before the fork. A call that must fetch starts a new fetch. - 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..ce84d1c 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 @@ -463,6 +464,30 @@ class _Entry: _cache_lock = threading.Lock() +def _lock_cache() -> None: + _cache_lock.acquire() + + +def _unlock_cache() -> None: + _cache_lock.release() + + +def _reset_fetches() -> None: + # The child has no fetch threads of the parent, thus no thread can set their events. A call + # that must fetch starts a new fetch. The child keeps the copies. + for entry in _cache.values(): + entry.pending = None + _unlock_cache() + + +# fork() waits for each thread that holds the lock. Thus the child gets an unlocked lock, and no +# entry that a thread changed only in part. Windows has no fork. +if hasattr(os, "register_at_fork"): + os.register_at_fork( + before=_lock_cache, after_in_parent=_unlock_cache, after_in_child=_reset_fetches + ) + + def _load( key: tuple[str, str, str, str], ttl_seconds: float, diff --git a/sdks/python/tests/test_prompts.py b/sdks/python/tests/test_prompts.py index 892d676..c70e2e7 100644 --- a/sdks/python/tests/test_prompts.py +++ b/sdks/python/tests/test_prompts.py @@ -3,9 +3,13 @@ import asyncio import contextvars 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 @@ -93,6 +97,55 @@ def wait_for(check: Callable[[], bool]) -> None: time.sleep(0.01) +def in_fork(child: Callable[[], str]) -> str: + """Run ``child`` in a forked process 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 + read_end, write_end = os.pipe() + sys.unraisablehook = record + try: + pid = os.fork() + finally: + sys.unraisablehook = previous_hook + if pid == 0: + 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", api_url=server.url).version + + @pytest.mark.parametrize( ("options", "query"), [ @@ -525,6 +578,69 @@ def close(self) -> None: assert timeouts[0] is not None and timeouts[0] <= 0.2 +@pytest.mark.skipif(not hasattr(os, "fork"), reason="needs os.fork") +# Python 3.12+ gives a warning for fork() in a process with threads, and the capture server has one. +@pytest.mark.filterwarnings("ignore:This process .* is multi-threaded:DeprecationWarning") +def test_a_forked_process_fetches_the_prompt_that_the_parent_was_fetching( + server: CaptureServer, monkeypatch: pytest.MonkeyPatch +) -> 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 + + assert in_fork(lambda: str(version_of(server))) == "12" + parent.join() + assert versions == [12] + # The fetch of the parent, and the fetch of the child. + assert len(server.requests) == 2 + + +@pytest.mark.skipif(not hasattr(os, "fork"), reason="needs os.fork") +@pytest.mark.filterwarnings("ignore:This process .* is multi-threaded:DeprecationWarning") +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" + + +@pytest.mark.skipif(not hasattr(os, "fork"), reason="needs os.fork") +@pytest.mark.filterwarnings("ignore:This process .* is multi-threaded:DeprecationWarning") +def test_a_fork_while_a_thread_holds_the_cache_lock_does_not_deadlock_the_child( + server: CaptureServer, +) -> None: + held = threading.Event() + + def hold() -> None: + with prompts_module._cache_lock: # pyright: ignore[reportPrivateUsage] + held.set() + time.sleep(0.2) + + holder = threading.Thread(target=hold) + holder.start() + held.wait() + assert in_fork(lambda: str(version_of(server))) == "12" + holder.join() + + def test_get_prompt_rejects_a_prompt_of_the_other_type_and_bad_options( server: CaptureServer, ) -> None: From a4cb4938a1cbb4845f02f75c17e50daabd86b464 Mon Sep 17 00:00:00 2001 From: ephraimduncan Date: Thu, 29 Oct 2026 15:27:25 +0000 Subject: [PATCH 2/5] fix(sdk-python): make the prompt cache fork hooks safe A pending fetch keeps the id of the process whose thread runs the fetch. In a different process, a call ignores this fetch and starts a new fetch. Thus a fork without fork hooks also gets a new fetch, for example a fork in C code. The child hook does not reset the entries. The cache lock is reentrant. Thus a fork in the thread that holds the lock does not wait for that thread. The before hook records in a thread local if it got the lock. The after hooks release the lock only in that condition. The module registers the hooks one time in each process, also when importlib.reload runs the module again. The hooks find the lock when they run. The lock test examines the time of the fork and the errors of the holder thread. The fork must return after the holder releases the lock. --- sdks/python/src/telemetry_dev/_prompts.py | 41 +++--- sdks/python/tests/test_prompts.py | 150 +++++++++++++++++++--- 2 files changed, 155 insertions(+), 36 deletions(-) diff --git a/sdks/python/src/telemetry_dev/_prompts.py b/sdks/python/src/telemetry_dev/_prompts.py index ce84d1c..b109baa 100644 --- a/sdks/python/src/telemetry_dev/_prompts.py +++ b/sdks/python/src/telemetry_dev/_prompts.py @@ -456,35 +456,40 @@ 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. +_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_cache() -> None: +def _lock_for_fork() -> None: _cache_lock.acquire() + _forking.locked = True -def _unlock_cache() -> None: - _cache_lock.release() +def _unlock_after_fork() -> None: + if getattr(_forking, "locked", False): + _forking.locked = False + _cache_lock.release() -def _reset_fetches() -> None: - # The child has no fetch threads of the parent, thus no thread can set their events. A call - # that must fetch starts a new fetch. The child keeps the copies. - for entry in _cache.values(): - entry.pending = None - _unlock_cache() - - -# fork() waits for each thread that holds the lock. Thus the child gets an unlocked lock, and no -# entry that a thread changed only in part. Windows has no fork. -if hasattr(os, "register_at_fork"): +# A fork waits for each 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 +# 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_cache, after_in_parent=_unlock_cache, after_in_child=_reset_fetches + before=_lock_for_fork, after_in_parent=_unlock_after_fork, after_in_child=_unlock_after_fork ) @@ -506,8 +511,12 @@ 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 entry.pending is not None and entry.pending_pid != os.getpid(): + # A fork copied the fetch but not its thread, also a fork in C code that runs no hooks. + entry.pending = None if now >= max(entry.checked_at + ttl, entry.retry_at) and entry.pending is None: 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 diff --git a/sdks/python/tests/test_prompts.py b/sdks/python/tests/test_prompts.py index c70e2e7..487bf35 100644 --- a/sdks/python/tests/test_prompts.py +++ b/sdks/python/tests/test_prompts.py @@ -2,6 +2,7 @@ import asyncio import contextvars +import importlib import itertools import os import signal @@ -16,7 +17,7 @@ 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 @@ -38,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] = { @@ -97,10 +104,10 @@ def wait_for(check: Callable[[], bool]) -> None: time.sleep(0.01) -def in_fork(child: Callable[[], str]) -> str: - """Run ``child`` in a forked process 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.""" +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] = [] @@ -114,7 +121,7 @@ def record(unraisable: sys.UnraisableHookArgs) -> None: read_end, write_end = os.pipe() sys.unraisablehook = record try: - pid = os.fork() + pid = (fork or os.fork)() finally: sys.unraisablehook = previous_hook if pid == 0: @@ -578,9 +585,7 @@ def close(self) -> None: assert timeouts[0] is not None and timeouts[0] <= 0.2 -@pytest.mark.skipif(not hasattr(os, "fork"), reason="needs os.fork") -# Python 3.12+ gives a warning for fork() in a process with threads, and the capture server has one. -@pytest.mark.filterwarnings("ignore:This process .* is multi-threaded:DeprecationWarning") +@needs_fork def test_a_forked_process_fetches_the_prompt_that_the_parent_was_fetching( server: CaptureServer, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -600,8 +605,25 @@ def test_a_forked_process_fetches_the_prompt_that_the_parent_was_fetching( assert len(server.requests) == 2 -@pytest.mark.skipif(not hasattr(os, "fork"), reason="needs os.fork") -@pytest.mark.filterwarnings("ignore:This process .* is multi-threaded:DeprecationWarning") +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: @@ -622,23 +644,111 @@ def child() -> str: assert in_fork(child) == "12" -@pytest.mark.skipif(not hasattr(os, "fork"), reason="needs os.fork") -@pytest.mark.filterwarnings("ignore:This process .* is multi-threaded:DeprecationWarning") -def test_a_fork_while_a_thread_holds_the_cache_lock_does_not_deadlock_the_child( - server: CaptureServer, -) -> None: +@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: - with prompts_module._cache_lock: # pyright: ignore[reportPrivateUsage] - held.set() - time.sleep(0.2) + 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() - assert in_fork(lambda: str(version_of(server))) == "12" + # 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_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) + + +@pytest.mark.skipif(not hasattr(signal, "setitimer"), reason="needs signal.setitimer") +def test_the_fork_hooks_do_not_release_a_lock_that_they_did_not_get() -> None: + class Interrupted(Exception): + pass + + lock = prompts_module._cache_lock # pyright: ignore[reportPrivateUsage] + held, done = threading.Event(), threading.Event() + errors: list[BaseException] = [] + + def hold() -> None: + try: + with lock: + held.set() + done.wait(5) + except BaseException as exc: + errors.append(exc) + + def interrupt(signum: int, frame: FrameType | None) -> NoReturn: + raise Interrupted + + holder = threading.Thread(target=hold) + holder.start() + held.wait() + previous = signal.signal(signal.SIGALRM, interrupt) + try: + # A signal handler raises while the before hook waits for the lock. os.fork() reports + # the error, forks, and then runs the after hooks. + signal.setitimer(signal.ITIMER_REAL, 0.1) + with pytest.raises(Interrupted): + prompts_module._lock_for_fork() # pyright: ignore[reportPrivateUsage] + prompts_module._unlock_after_fork() # pyright: ignore[reportPrivateUsage] + finally: + signal.signal(signal.SIGALRM, previous) + done.set() + holder.join() + assert errors == [] def test_get_prompt_rejects_a_prompt_of_the_other_type_and_bad_options( From 072cad1a66928d95197bdc7571e7568106bad37a Mon Sep 17 00:00:00 2001 From: ephraimduncan Date: Thu, 29 Oct 2026 21:07:59 +0000 Subject: [PATCH 3/5] fix(sdk-python): replace a held prompt cache lock in a forked child If the before fork hook did not get the cache lock, a thread of the parent can hold the lock in the child. A signal can stop the wait of the before hook. uWSGI with py-call-osafterfork runs only the child hook. In these two conditions, a get_prompt call in the child waited for all time. The child hook replaces the lock in such a child. The code reads the lock from the module each time. The child hook removes the fetches of the parent again. Before this change, the child found these fetches only through the process ID. If the child got the ID of the process that started a fetch, the child waited for that fetch, and its calls got timeout errors. The ID check stays for a fork that runs no hooks. A call keeps its fetch event in a local variable. Thus a fork in the call, as in a signal handler, does not let the call continue with no event and no result. The capture server reads its delay before it records a request. Thus a test that finds the request and sets the delay to 0 does not cancel the wait of that request. The signal test stops its timer. Thus a failure of the test does not stop pytest with SIGALRM. in_fork finds the child through its process ID. Thus a fork function that forks two times does not send a child back into pytest. --- sdks/python/src/telemetry_dev/_prompts.py | 34 +++++++--- sdks/python/tests/otlp_capture.py | 6 +- sdks/python/tests/test_prompts.py | 80 ++++++++++++++++++++--- 3 files changed, 100 insertions(+), 20 deletions(-) diff --git a/sdks/python/src/telemetry_dev/_prompts.py b/sdks/python/src/telemetry_dev/_prompts.py index b109baa..dce40ae 100644 --- a/sdks/python/src/telemetry_dev/_prompts.py +++ b/sdks/python/src/telemetry_dev/_prompts.py @@ -464,7 +464,8 @@ class _Entry: _cache: OrderedDict[tuple[str, str, str, str], _Entry] = OrderedDict() # 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. +# 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. @@ -482,14 +483,29 @@ def _unlock_after_fork() -> None: _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 waits for each 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 -# when they run. Windows has no fork. +# 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=_unlock_after_fork + before=_lock_for_fork, after_in_parent=_unlock_after_fork, after_in_child=_reset_in_child ) @@ -512,18 +528,20 @@ def _load( now = monotonic() ttl = ttl_seconds if entry.response is not None else min(ttl_seconds, _ERROR_TTL_S) if entry.pending is not None and entry.pending_pid != os.getpid(): - # A fork copied the fetch but not its thread, also a fork in C code that runs no hooks. + # A fork in C code that runs no hooks copied the fetch but not its thread. entry.pending = None - if now >= max(entry.checked_at + ttl, entry.retry_at) and entry.pending is None: - entry.pending = threading.Event() + # 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 487bf35..eaddeca 100644 --- a/sdks/python/tests/test_prompts.py +++ b/sdks/python/tests/test_prompts.py @@ -118,13 +118,15 @@ def record(unraisable: sys.UnraisableHookArgs) -> None: 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 - if pid == 0: + # 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: @@ -586,8 +588,9 @@ def close(self) -> None: @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 + 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) @@ -597,8 +600,15 @@ def test_a_forked_process_fetches_the_prompt_that_the_parent_was_fetching( 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(lambda: str(version_of(server))) == "12" + assert in_fork(child) == "12" parent.join() assert versions == [12] # The fetch of the parent, and the fetch of the child. @@ -696,6 +706,43 @@ def fork_and_fetch() -> str: 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 @@ -714,11 +761,16 @@ def test_a_fork_after_a_reload_of_the_prompt_module_returns( @pytest.mark.skipif(not hasattr(signal, "setitimer"), reason="needs signal.setitimer") -def test_the_fork_hooks_do_not_release_a_lock_that_they_did_not_get() -> None: +@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( + 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] = [] @@ -738,13 +790,21 @@ def interrupt(signum: int, frame: FrameType | None) -> NoReturn: held.wait() previous = signal.signal(signal.SIGALRM, interrupt) try: - # A signal handler raises while the before hook waits for the lock. os.fork() reports - # the error, forks, and then runs the after hooks. - signal.setitimer(signal.ITIMER_REAL, 0.1) - with pytest.raises(Interrupted): - prompts_module._lock_for_fork() # pyright: ignore[reportPrivateUsage] - prompts_module._unlock_after_fork() # pyright: ignore[reportPrivateUsage] + if before_hook == "stopped by a signal": + # A signal handler raises while the before hook waits for the lock. os.fork() + # reports the error, forks, and then runs the after hooks. + signal.setitimer(signal.ITIMER_REAL, 0.1) + with pytest.raises(Interrupted): + prompts_module._lock_for_fork() # pyright: ignore[reportPrivateUsage] + prompts_module._unlock_after_fork() # pyright: ignore[reportPrivateUsage] + # The child hook. uWSGI with py-call-osafterfork forks in C and runs only this hook. In a + # child, the holder is a thread of the parent that does not exist there. + 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: + signal.setitimer(signal.ITIMER_REAL, 0) signal.signal(signal.SIGALRM, previous) done.set() holder.join() From edb1701ca3d5dc1239aac8f61f120c7120ac8b91 Mon Sep 17 00:00:00 2001 From: ephraimduncan Date: Thu, 29 Oct 2026 21:27:55 +0000 Subject: [PATCH 4/5] fix(sdk-python): use base_url in the prompt fork tests The version_of helper of the fork tests gave the api_url option to get_prompt. #101 replaced api_url with base_url. Thus pyright found an error after the rebase on #101. --- sdks/python/tests/test_prompts.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdks/python/tests/test_prompts.py b/sdks/python/tests/test_prompts.py index eaddeca..95adebc 100644 --- a/sdks/python/tests/test_prompts.py +++ b/sdks/python/tests/test_prompts.py @@ -152,7 +152,7 @@ def record(unraisable: sys.UnraisableHookArgs) -> None: def version_of(server: CaptureServer) -> int | None: - return get_prompt("support/reply", api_key="td_live_a", api_url=server.url).version + return get_prompt("support/reply", api_key="td_live_a", base_url=server.url).version @pytest.mark.parametrize( From 0d5994f1cc2405006ea1a45b5e1c5433a22edd79 Mon Sep 17 00:00:00 2001 From: ephraimduncan Date: Thu, 29 Oct 2026 22:01:25 +0000 Subject: [PATCH 5/5] fix(sdk-python): correct the prompt fork claims and fork in the test The README and docs/sdk-conformance.md said that a forked process does not wait for a fetch that did not end before the fork. That is not true for one case. If a signal handler forks while a call waits for a fetch, the 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, and then fails with the timeout code. The two documents now name this case, and the code does not change. The comment at the process ID check in _load, and the message of the commit "replace a held prompt cache lock in a forked child", said that the check handles a fork that runs no hooks. The check removes only the copied fetch. Such a fork can also copy a lock that a thread of the parent holds. Then a call in the child waits for the lock with no limit. The comment now says this, and the comment at the hook registration now says that only a fork that runs the hooks waits for the lock. The signal row of the owner test of the lock replacement called the fork hooks directly. The row now forks with os.fork() through in_fork. A different thread holds the cache lock, and SIGALRM stops the before hook. The child must get the prompt. Thus the row needs os.fork() to run the after hooks when a before hook raises. The row fails when the child hook does not replace the lock. It also fails when an after hook releases a lock that the before hook did not get. If no signal stops the before hook, the holder thread fails the row after 5 s. The fork function of the row stops its timer also when the fork raises. If the child loses the thread local _forking, the child hook replaces the lock and does not release it. The child gets a free lock in the two cases, thus no test can see a difference. --- docs/sdk-conformance.md | 9 +++-- sdks/python/README.md | 8 +++- sdks/python/src/telemetry_dev/_prompts.py | 12 +++--- sdks/python/tests/test_prompts.py | 49 +++++++++++++++-------- 4 files changed, 51 insertions(+), 27 deletions(-) diff --git a/docs/sdk-conformance.md b/docs/sdk-conformance.md index 1bc6385..ffd2d41 100644 --- a/docs/sdk-conformance.md +++ b/docs/sdk-conformance.md @@ -141,9 +141,12 @@ Concurrent calls share one fetch. its connection down at that time, because the urllib timeout applies to each socket operation 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. A process that `os.fork()` makes gets the - copies of the parent, but not the fetch threads of the parent. In the child, a fetch that did not - end before the fork counts as not started. Thus a call that must fetch starts a new fetch. + 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 07f87f1..bb9b109 100644 --- a/sdks/python/README.md +++ b/sdks/python/README.md @@ -220,8 +220,12 @@ pinned = telemetry_dev.get_prompt("support/reply", version=13) 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. The process does not wait for a fetch that did - not end before the fork. A call that must fetch starts a new fetch. + `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 dce40ae..0ba6b9d 100644 --- a/sdks/python/src/telemetry_dev/_prompts.py +++ b/sdks/python/src/telemetry_dev/_prompts.py @@ -498,10 +498,10 @@ def _reset_in_child() -> None: entry.pending = None -# A fork waits for each 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. +# 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( @@ -528,7 +528,9 @@ def _load( now = monotonic() ttl = ttl_seconds if entry.response is not None else min(ttl_seconds, _ERROR_TTL_S) 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. + # 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 diff --git a/sdks/python/tests/test_prompts.py b/sdks/python/tests/test_prompts.py index 95adebc..764c4ef 100644 --- a/sdks/python/tests/test_prompts.py +++ b/sdks/python/tests/test_prompts.py @@ -760,10 +760,11 @@ def test_a_fork_after_a_reload_of_the_prompt_module_returns( 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( - monkeypatch: pytest.MonkeyPatch, before_hook: str + server: CaptureServer, monkeypatch: pytest.MonkeyPatch, before_hook: str ) -> None: class Interrupted(Exception): pass @@ -778,34 +779,48 @@ def hold() -> None: try: with lock: held.set() - done.wait(5) + # 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() - previous = signal.signal(signal.SIGALRM, interrupt) try: if before_hook == "stopped by a signal": - # A signal handler raises while the before hook waits for the lock. os.fork() - # reports the error, forks, and then runs the after hooks. - signal.setitimer(signal.ITIMER_REAL, 0.1) - with pytest.raises(Interrupted): - prompts_module._lock_for_fork() # pyright: ignore[reportPrivateUsage] - prompts_module._unlock_after_fork() # pyright: ignore[reportPrivateUsage] - # The child hook. uWSGI with py-call-osafterfork forks in C and runs only this hook. In a - # child, the holder is a thread of the parent that does not exist there. - 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() + # 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: - signal.setitimer(signal.ITIMER_REAL, 0) - signal.signal(signal.SIGALRM, previous) done.set() holder.join() assert errors == []