Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 19 additions & 7 deletions livekit-rtc/livekit/rtc/_ffi_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import annotations

import signal
import asyncio
import sys
Expand All @@ -24,6 +26,7 @@
import platform
import atexit
import threading
import weakref
from typing import Callable, Generic, List, Optional, TypeVar

from ._proto import ffi_pb2 as proto_ffi
Expand Down Expand Up @@ -68,10 +71,14 @@ def get_ffi_lib() -> ctypes.CDLL:
INVALID_HANDLE = 0


_live_handles: weakref.WeakSet[FfiHandle] = weakref.WeakSet()


class FfiHandle:
def __init__(self, handle: int) -> None:
self.handle = handle
self._disposed = False
_live_handles.add(self)

def __del__(self) -> None:
self.dispose()
Expand All @@ -93,7 +100,7 @@ def dispose(self) -> None:
if self.handle != INVALID_HANDLE and not self._disposed:
self._disposed = True
ffi = FfiClient._instance
if ffi is None or ffi._pid != os.getpid():
if ffi is None or ffi._pid != os.getpid() or getattr(FfiClient, "_ffi_disposed", False):
return
dropped = ffi._ffi_lib.livekit_ffi_drop_handle(ctypes.c_uint64(self.handle))
assert dropped
Expand Down Expand Up @@ -223,6 +230,7 @@ def to_python_level(level: proto_ffi.LogLevel.ValueType) -> Optional[int]:

class FfiClient:
_instance: Optional["FfiClient"] = None
_ffi_disposed: bool = False

@classproperty
def instance(cls) -> "FfiClient":
Expand All @@ -231,6 +239,7 @@ def instance(cls) -> "FfiClient":
return cls._instance

def __init__(self) -> None:
FfiClient._ffi_disposed = False
self._pid = os.getpid()
self._lock = threading.RLock()
self._queue = FfiQueue[proto_ffi.FfiEvent]()
Expand Down Expand Up @@ -266,14 +275,17 @@ def __init__(self) -> None:
ffi_event_callback, True, b"python", __version__.encode("ascii")
)

ffi_lib = self._ffi_lib
init_pid = self._pid

@atexit.register
def _dispose_lk_ffi() -> None:
if os.getpid() != init_pid:
return
ffi_lib.livekit_ffi_dispose()
self._dispose()

def _dispose(self) -> None:
if self._pid != os.getpid() or FfiClient._ffi_disposed:
return
FfiClient._ffi_disposed = True
for handle in list(_live_handles):
handle.mark_consumed()
self._ffi_lib.livekit_ffi_dispose()

@property
def queue(self) -> FfiQueue[proto_ffi.FfiEvent]:
Expand Down
91 changes: 91 additions & 0 deletions livekit-rtc/tests/test_ffi_handle_lifecycle.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,91 @@
# Copyright 2026 LiveKit, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import annotations

import gc
import subprocess
import sys

import pytest

from livekit.rtc._ffi_client import FfiClient, FfiHandle
from livekit.rtc._proto import audio_frame_pb2, ffi_pb2


def _allocate_real_handle() -> int:
req = ffi_pb2.FfiRequest()
req.new_audio_source.type = audio_frame_pb2.AUDIO_SOURCE_NATIVE
req.new_audio_source.sample_rate = 48000
req.new_audio_source.num_channels = 1
req.new_audio_source.queue_size_ms = 100
resp = FfiClient.instance.request(req)
handle_id = resp.new_audio_source.source.handle.id
assert handle_id != 0
return int(handle_id)


def test_active_runtime_invalid_handle_asserts() -> None:
FfiClient.instance
with pytest.raises(AssertionError):
FfiHandle(999999).dispose()


def test_live_handles_marked_consumed_and_no_assert_after_dispose() -> None:
handle_id = _allocate_real_handle()
handle = FfiHandle(handle_id)
assert not handle.disposed

# Global dispose marks all live handles consumed and sets _ffi_disposed
FfiClient.instance._dispose()
assert handle.disposed

# Explicit dispose or __del__ after global dispose should be a no-op and not assert
handle.dispose()
handle.__del__()

# Newly instantiated handle after global dispose also safely no-ops without assert
late_handle = FfiHandle(12345)
late_handle.dispose()

del handle
del late_handle
gc.collect()


def test_subprocess_exit_with_undisposed_handle_has_no_assertion_error() -> None:
code = """
import sys
from livekit.rtc._proto import audio_frame_pb2, ffi_pb2
from livekit.rtc._ffi_client import FfiClient, FfiHandle

req = ffi_pb2.FfiRequest()
req.new_audio_source.type = audio_frame_pb2.AUDIO_SOURCE_NATIVE
req.new_audio_source.sample_rate = 48000
req.new_audio_source.num_channels = 1
req.new_audio_source.queue_size_ms = 100
resp = FfiClient.instance.request(req)
handle_id = resp.new_audio_source.source.handle.id
# Leave handle alive at process shutdown
global_handle = FfiHandle(handle_id)
sys.exit(0)
"""
result = subprocess.run(
[sys.executable, "-c", code],
capture_output=True,
text=True,
)
assert result.returncode == 0
assert "AssertionError" not in result.stderr
assert "Exception ignored" not in result.stderr
Loading