Fork safe ConcurrentMultiSpanProcessor

[gregoiredx]

Description

ConcurrentMultiSpanProcessor is not "fork safe". This MR tries to solve this problem.

Forking a threaded process is not a good idea in the first place, citing the Python doc, see https://docs.python.org/3/library/multiprocessing.html#contexts-and-start-methods

Note that safely forking a multithreaded process is problematic.

Also, the OTel Python SDK suggest to only initialize the SDK post fork, see https://opentelemetry-python.readthedocs.io/en/latest/examples/fork-process-model/README.html

Nevertheless, on an existing code base, it's not always easy to avoid all forks. So it looks like a good idea to try and make the SDK as fork safe as possible.

Note: BatchProcessor is already relying on register_at_fork used here, see

opentelemetry-python/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py

Line 125 in 102fec2

os.register_at_fork(after_in_child=lambda: weak_reinit()()) # pyright: ignore[reportOptionalCall] pylint: disable=unnecessary-lambda

Fixes #2349

Type of change

Please delete options that are not relevant.

• Bug fix (non-breaking change which fixes an issue)

How Has This Been Tested?

See unit test and discussions below.

An equivalent patch is currently running in our stack.

Does This PR Require a Contrib Repo Change?

?

Checklist:

• Followed the style guidelines of this project
• Changelogs have been updated
• Unit tests have been added
• Documentation has been updated

[linux-foundation-easycla]

The committers listed above are authorized under a signed CLA.

• ✅ login: gregoiredx / name: Grégoire Deveaux (808fe91)

[DylanRussell]

Is there something that says explicitly THreadPoolExecutor needs to be reinitialized in forked processes ? I see the note about not forking threaded processes, but it's still a bit unclear to me..

[gregoiredx]

Is there something that says explicitly THreadPoolExecutor needs to be reinitialized in forked processes ?

I don't think it's stated as such anywhere.

What is sure is that on POSIX environments, os.fork() keeps only the thread which called fork in a running state in the child process, cf https://pubs.opengroup.org/onlinepubs/9799919799/functions/fork.html:

A process shall be created with a single thread. If a multi-threaded process calls fork(), the new process shall contain a replica of the calling thread and its entire address space

Since I don't see an easy way to "restart" the threads from the ThreadPoolExecutor, restarting a fresh new thread pool seems like a good solution.

[gregoiredx]

More on this, specifically on Linux, https://man7.org/linux/man-pages/man2/fork.2.html

The child process is created with a single thread—the one that called fork(). The entire virtual address space of the parent is replicated in the child, including the states of mutexes, condition variables, and other pthreads objects; the use of pthread_atfork(3) may be helpful for dealing with problems that this can cause.

[gregoiredx]

I still don't see how to add fork to the actual test suite but here is at least an example I tried to put up to help test what's going on here:

import os
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=2)

def reinit_executor():
    global executor
    executor = ThreadPoolExecutor(max_workers=2)

# Remove this line to see the child process hanging forever:
os.register_at_fork(after_in_child=reinit_executor)

print(f"pid:{os.getpid()} in main thread before fork")
print(executor.submit(lambda : f"pid:{os.getpid()} in sub thread before fork").result())

child_pid = os.fork()
if child_pid:
    print(f"pid:{os.getpid()} forked parent process to child process", child_pid)
else:
    print(f"pid:{os.getpid()} forked process, in child")

print(f"pid:{os.getpid()} in main thread after fork")
print(executor.submit(lambda: f"pid:{os.getpid()} in sub thread after fork").result())

if child_pid:
    print(f"pid:{os.getpid()} waiting for child process")
    os.waitpid(child_pid, 0)
else:
    print(f"pid:{os.getpid()} child process end")

[gregoiredx]

And a version with the ConcurrentMultiSpanProcessor.
The code below will hang forever on main and work as expected on the current branch:

import dataclasses
import os

from opentelemetry.sdk.trace import ConcurrentMultiSpanProcessor, SpanProcessor


@dataclasses.dataclass
class DummySpan:
    name: str

class PrintSpanProcessor(SpanProcessor):
    def on_start(
        self,
        span,
        parent_context = None,
    ) :
        print(f"pid:{os.getpid()} on_start", span.name)

processor = ConcurrentMultiSpanProcessor()
processor.add_span_processor(PrintSpanProcessor())

processor.on_start(DummySpan(name="before fork"))

child_pid = os.fork()
if child_pid:
    print(f"pid:{os.getpid()} forked parent process to child process", child_pid)
else:
    print(f"pid:{os.getpid()} forked process, in child")

print(f"pid:{os.getpid()} in main thread after fork")
processor.on_start(DummySpan(name="a span"))

if child_pid:
    print(f"pid:{os.getpid()} waiting for child process")
    os.waitpid(child_pid, 0)
else:
    print(f"pid:{os.getpid()} child process end")

[gregoiredx]

Thanks to the tests in TestBatchProcessor, I've been able to write a unit test in the end, reproducing the problem 🎉