FfiQueue.put holds the subscriber lock across call_soon_threadsafe, so subscribe/unsubscribe on other event loops wait for the whole delivery #888

Summary

FfiQueue.put runs on the native FFI callback thread and holds self._lock while it runs every subscriber's filter_fn and calls loop.call_soon_threadsafe(...) for each match:

def put(self, item: T) -> None:
    with self._lock:
        for queue, loop, filter_fn in self._subscribers:
            ...
            loop.call_soon_threadsafe(queue.put_nowait, item)

call_soon_threadsafe writes to the target loop's self-pipe, which releases the GIL. When other threads are busy, the callback thread has to win the GIL back once per subscriber, and the lock stays held the whole time.

subscribe and unsubscribe take the same lock. Every FFI request (perform_rpc, publish_data, set_attributes, Room.disconnect, ...) calls subscribe from its event loop thread. So that loop waits, off CPU, until the callback thread finishes delivering one event to all subscribers.

Production impact

We run many jobs as threads in one process (JobExecutorType.THREAD). Each job has its own event loop and its own Room, so _subscribers holds a few entries per job. Our loop stall monitor samples the loop thread's stack during each stall. In 7 days it recorded 6 stalls of 100 to 222 ms with the loop thread blocked here:

File ".../livekit/rtc/participant.py", line 468, in perform_rpc
File ".../livekit/rtc/participant.py", line 481, in _perform_rpc_ffi
File ".../livekit/rtc/_ffi_client.py", line 150, in subscribe

In the sampled one, the loop thread used 3 ms of CPU during a 101 ms stall, while over the surrounding 714 ms the FFI callback thread (Dummy-105) used 220 ms.

Suggested fix

Make the subscriber list copy-on-write. subscribe and unsubscribe build a new list under the lock and assign it, and put iterates the list it read without taking the lock:

def put(self, item: T) -> None:
    for queue, loop, filter_fn in self._subscribers:
        ...  # unchanged body

def subscribe(self, loop=None, filter_fn=None) -> Queue[T]:
    with self._lock:
        queue = Queue[T]()
        loop = loop or asyncio.get_event_loop()
        self._subscribers = [*self._subscribers, (queue, loop, filter_fn)]
        return queue

def unsubscribe(self, queue: Queue[T]) -> None:
    with self._lock:
        for i, (q, _, _) in enumerate(self._subscribers):
            if q == queue:
                self._subscribers = self._subscribers[:i] + self._subscribers[i + 1 :]
                break

The attribute read is atomic under the GIL. A put that runs at the same time as a change delivers as if it ran just before the change, which the current lock also allows. The lock only serializes writers.

We carry this as a monkeypatch, with tests that hold a put mid-delivery (a filter_fn that blocks on an event) and check that subscribe and unsubscribe on another thread still return. Happy to open a PR with those tests.

Related

#563 and #564 reduce how many events reach each subscriber. This issue is separate: the lock is held across delivery even when every subscriber filters.

Environment

• livekit 1.1.20 (the code on main is the same)
• livekit-agents 1.8.8 with JobExecutorType.THREAD
• CPython 3.12.12, Linux aarch64