rtc: put fork guard in request() instead of instance

Guarding FfiClient.instance meant every __del__/dispose path that reads it
raised the guard in fork children (noisy 'Exception ignored'), which needed
extra cleanup plumbing. request() is the only place that actually hangs on a
dead post-fork runtime, so guard there: active use raises loudly, while
instance and the GC paths revert to their original behavior. atexit dispose
stays pid-guarded (it needs the runtime and would hang at exit).

Author: theomonnom

livekit-rtc/livekit/rtc/_ffi_client.py

@@ -83,10 +83,8 @@ def disposed(self) -> bool:
     def dispose(self) -> None:
         if self.handle != INVALID_HANDLE and not self._disposed:
             self._disposed = True
-            ffi = FfiClient._owned_instance()
-            if ffi is not None:
-                dropped = ffi._ffi_lib.livekit_ffi_drop_handle(ctypes.c_uint64(self.handle))
-                assert dropped
+            dropped = FfiClient.instance._ffi_lib.livekit_ffi_drop_handle(ctypes.c_uint64(self.handle))
+            assert dropped
 
     def __repr__(self) -> str:
         return f"FfiHandle({self.handle})"
@@ -216,27 +214,9 @@ class FfiClient:
 
     @classproperty
     def instance(cls) -> "FfiClient":
-        inst = cls._instance
-        if inst is not None and inst._pid != os.getpid():
-            # The native runtime (tokio worker/IO threads inside liblivekit_ffi)
-            # does not survive fork(), and re-initializing does not rebuild it,
-            # so a child that inherited an initialized FFI hangs on every request.
-            # Fail loudly instead of hanging.
-            raise RuntimeError(
-                "livekit.rtc was used in a parent process before fork(); the native "
-                "runtime cannot be used across fork(). Do not create or connect a Room "
-                "(or use any livekit.rtc object) before forking child processes."
-            )
-        if inst is None:
-            cls._instance = inst = FfiClient()
-        return inst
-
-    @classmethod
-    def _owned_instance(cls) -> Optional["FfiClient"]:
-        # the client only if this process created it; cleanup paths use it to
-        # no-op in fork children instead of touching inherited FFI/locks
-        inst = cls._instance
-        return inst if inst is not None and inst._pid == os.getpid() else None
+        if cls._instance is None:
+            cls._instance = FfiClient()
+        return cls._instance
 
     def __init__(self) -> None:
         self._pid = os.getpid()
@@ -291,6 +271,15 @@ def queue(self) -> FfiQueue[proto_ffi.FfiEvent]:
         return self._queue
 
     def request(self, req: proto_ffi.FfiRequest) -> proto_ffi.FfiResponse:
+        if self._pid != os.getpid():
+            # The native runtime does not survive fork() and cannot be
+            # reinitialized, so requests from a child that inherited the FFI
+            # would hang. Fail loudly instead.
+            raise RuntimeError(
+                "livekit.rtc was used in a parent process before fork(); the native "
+                "runtime cannot be used across fork(). Do not create or connect a Room "
+                "(or use any livekit.rtc object) before forking child processes."
+            )
         proto_data = req.SerializeToString()
         proto_len = len(proto_data)
         data = (ctypes.c_ubyte * proto_len)(*proto_data)

livekit-rtc/livekit/rtc/audio_stream.py

@@ -246,9 +246,7 @@ def from_track(
         )
 
     def __del__(self) -> None:
-        ffi = FfiClient._owned_instance()
-        if ffi is not None:
-            ffi.queue.unsubscribe(self._ffi_queue)
+        FfiClient.instance.queue.unsubscribe(self._ffi_queue)
 
     def _create_owned_stream(self) -> Any:
         assert self._track is not None

livekit-rtc/livekit/rtc/room.py

@@ -190,9 +190,7 @@ def __init__(
 
     def __del__(self) -> None:
         if self._ffi_handle is not None:
-            ffi = FfiClient._owned_instance()
-            if ffi is not None:
-                ffi.queue.unsubscribe(self._ffi_queue)
+            FfiClient.instance.queue.unsubscribe(self._ffi_queue)
 
     @property
     async def sid(self) -> str:

livekit-rtc/livekit/rtc/video_stream.py

@@ -109,9 +109,7 @@ def from_track(
         )
 
     def __del__(self) -> None:
-        ffi = FfiClient._owned_instance()
-        if ffi is not None:
-            ffi.queue.unsubscribe(self._ffi_queue)
+        FfiClient.instance.queue.unsubscribe(self._ffi_queue)
 
     def _create_owned_stream(self) -> Any:
         assert self._track is not None