Skip to content

Data packets create a ctypes array type for each new payload length #911

Closed

Description

@zek

Summary

Each data packet with a new payload length creates a new ctypes array type, on the send side and on the receive side:

  • LocalParticipant.publish_data (participant.py#L339): cdata = (ctypes.c_byte * data_len)(*payload)
  • Room data packet handler (room.py#L1118-L1121): ctypes.cast(buffer_info.data_ptr, ctypes.POINTER(ctypes.c_byte * buffer_info.data_len))

The effect depends on the Python version:

  • Python 3.12 and 3.13: ctypes.POINTER() keeps each pointer type in a cache that is never cleared, and the pointer type holds the array type. The receive side keeps one array type and one pointer type alive for each new length, for the life of the process.
  • Python 3.14: the types are freed, but each type is a reference cycle. Only the cyclic GC can free them, and a busy data channel gives it a lot of work.

The (*payload) unpack in publish_data also converts each byte to a Python int, the same per-byte copy that #848 reported for FfiClient.request.

Reproduction

The script connects two participants to a LiveKit server, sends 100 packets of different lengths with the GC disabled, waits until all 100 arrive, then runs gc.collect().

LIVEKIT_URL=ws://localhost:7880 LIVEKIT_API_KEY=devkey LIVEKIT_API_SECRET=secret \
  uv run --no-project --python 3.12 --with livekit==1.2.0 --with livekit-api repro.py
repro.py
import asyncio, ctypes, gc, os, sys
from livekit import api, rtc

N = 100

def token(identity):
    return (api.AccessToken(os.environ["LIVEKIT_API_KEY"], os.environ["LIVEKIT_API_SECRET"])
            .with_identity(identity)
            .with_grants(api.VideoGrants(room_join=True, room="ctypes-repro")).to_jwt())

async def main():
    url = os.environ["LIVEKIT_URL"]
    sender, receiver = rtc.Room(), rtc.Room()
    got = []
    receiver.on("data_received", lambda p: got.append(p.data))
    await receiver.connect(url, token("receiver"))
    await sender.connect(url, token("sender"))
    while not sender.remote_participants:
        await asyncio.sleep(0.1)
    await sender.local_participant.publish_data(b"warmup")
    while not got:
        await asyncio.sleep(0.1)
    got.clear()
    gc.collect()
    gc.disable()
    for i in range(N):
        await sender.local_participant.publish_data(b"x" * (1000 + i))
    while len(got) < N:
        await asyncio.sleep(0.1)
    freed = gc.collect()
    gc.enable()
    lengths = range(1000, 1000 + N)
    kept = [c for c in ctypes.Array.__subclasses__() if getattr(c, "_length_", None) in lengths]
    print(f"livekit {rtc.__version__}, Python {sys.version.split()[0]}: "
          f"freed by gc.collect(): {freed}, array types still alive: {len(kept)}")
    await sender.disconnect()
    await receiver.disconnect()

asyncio.run(main())
Python livekit 1.2.0
3.12.12 freed by gc.collect(): 0, array types still alive: 100
3.14.3 freed by gc.collect(): 1300, array types still alive: 0

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions