From 02b9039f485b2cd72faaa73f70eaf5016bcb2398 Mon Sep 17 00:00:00 2001 From: Cosmin Staicu Date: Thu, 8 Oct 2026 23:44:30 +0300 Subject: [PATCH] perf(rtc-node): route audio stream events and capture callbacks by handle Every FFI event was emitted to every listener in the process. Each audio stream and each room checked every other stream's frames, and every frame an AudioSource captured added a waitFor listener that all other waiters tested. Dispatch cost grew with the square of the concurrent streams; with a few dozen calls in one process it was about a third of the JS thread. Audio stream events now go to the listener registered for their stream handle, and captureAudioFrame callbacks to the waiter for their async id. A callback nothing waits for is still emitted, so waitFor callers keep working. AudioStreamSource keeps its handle id as a plain bigint, because reading the native FfiHandle.handle getter crosses into native code on every event. Room.onFfiEvent returns before taking its event lock for anything that is not a room event or an RPC invocation, the only cases processFfiEvent acts on. Signed-off-by: Cosmin Staicu --- .changeset/route-ffi-events-by-handle.md | 5 + packages/livekit-rtc/src/audio_source.ts | 6 +- packages/livekit-rtc/src/audio_stream.ts | 14 ++- packages/livekit-rtc/src/ffi_client.test.ts | 100 ++++++++++++++++++++ packages/livekit-rtc/src/ffi_client.ts | 42 ++++++++ packages/livekit-rtc/src/room.ts | 5 + 6 files changed, 164 insertions(+), 8 deletions(-) create mode 100644 .changeset/route-ffi-events-by-handle.md create mode 100644 packages/livekit-rtc/src/ffi_client.test.ts diff --git a/.changeset/route-ffi-events-by-handle.md b/.changeset/route-ffi-events-by-handle.md new file mode 100644 index 00000000..17225405 --- /dev/null +++ b/.changeset/route-ffi-events-by-handle.md @@ -0,0 +1,5 @@ +--- +'@livekit/rtc-node': patch +--- + +Audio stream events and capture callbacks go straight to the stream or waiter they belong to instead of to every FFI listener, and rooms skip events that are neither room events nor RPC invocations before taking their event lock. Dispatch cost no longer grows with the number of audio streams and rooms in the process. diff --git a/packages/livekit-rtc/src/audio_source.ts b/packages/livekit-rtc/src/audio_source.ts index 42d3d279..17a6eb03 100644 --- a/packages/livekit-rtc/src/audio_source.ts +++ b/packages/livekit-rtc/src/audio_source.ts @@ -159,9 +159,9 @@ export class AudioSource { message: { case: 'captureAudioFrame', value: req }, }); - const cb = await FfiClient.instance.waitFor((ev) => { - return ev.message.case == 'captureAudioFrame' && ev.message.value.asyncId == res.asyncId; - }); + const cb: CaptureAudioFrameCallback = await FfiClient.instance.waitForCaptureAudioFrame( + res.asyncId!, + ); if (cb.error) { throw new Error(cb.error); diff --git a/packages/livekit-rtc/src/audio_stream.ts b/packages/livekit-rtc/src/audio_stream.ts index ae28a655..49f01a95 100644 --- a/packages/livekit-rtc/src/audio_stream.ts +++ b/packages/livekit-rtc/src/audio_stream.ts @@ -6,7 +6,7 @@ import { AudioStreamType, NewAudioStreamRequest } from '@livekit/rtc-ffi-binding import type { UnderlyingSource } from 'node:stream/web'; import { AudioFrame } from './audio_frame.js'; import type { FfiEvent } from './ffi_client.js'; -import { FfiClient, FfiClientEvent, FfiHandle } from './ffi_client.js'; +import { FfiClient, FfiHandle } from './ffi_client.js'; import { type FrameProcessor, isFrameProcessor } from './frame_processor.js'; import { log } from './log.js'; import type { Track } from './track.js'; @@ -35,6 +35,9 @@ export interface NoiseCancellationOptions { export class AudioStreamSource implements UnderlyingSource { private controller?: ReadableStreamDefaultController; private ffiHandle: FfiHandle; + // The plain value events carry, which routes them here; the native `handle` getter costs a call + // across into native code on every read. + private ffiHandleId: bigint; private disposed = false; private sampleRate: number; private numChannels: number; @@ -86,9 +89,10 @@ export class AudioStreamSource implements UnderlyingSource { }, }); - this.ffiHandle = new FfiHandle(res.stream!.handle!.id!); + this.ffiHandleId = res.stream!.handle!.id!; + this.ffiHandle = new FfiHandle(this.ffiHandleId); - FfiClient.instance.on(FfiClientEvent.FfiEvent, this.onEvent); + FfiClient.instance.onAudioStreamEvent(this.ffiHandleId, this.onEvent); track.registerAudioStream(this); } @@ -104,7 +108,7 @@ export class AudioStreamSource implements UnderlyingSource { if ( ev.message.case != 'audioStreamEvent' || - ev.message.value.streamHandle != this.ffiHandle.handle + ev.message.value.streamHandle != this.ffiHandleId ) { return; } @@ -148,7 +152,7 @@ export class AudioStreamSource implements UnderlyingSource { * @internal */ teardown() { - FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent); + FfiClient.instance.offAudioStreamEvent(this.ffiHandleId); if (this.disposed) { return; } diff --git a/packages/livekit-rtc/src/ffi_client.test.ts b/packages/livekit-rtc/src/ffi_client.test.ts new file mode 100644 index 00000000..80d6b675 --- /dev/null +++ b/packages/livekit-rtc/src/ffi_client.test.ts @@ -0,0 +1,100 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { FfiEvent } from '@livekit/rtc-ffi-bindings'; +import type { PartialMessage } from '@livekit/rtc-ffi-bindings'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { FfiClient, FfiClientEvent } from './ffi_client.js'; + +const native = vi.hoisted(() => ({ + onEvent: undefined as ((data: Uint8Array) => void) | undefined, +})); + +vi.mock('@livekit/rtc-ffi-bindings', async () => { + const actual = await vi.importActual( + '@livekit/rtc-ffi-bindings', + ); + return { + ...actual, + livekitInitialize: (onEvent: (data: Uint8Array) => void) => { + native.onEvent = onEvent; + }, + }; +}); + +function deliver(event: PartialMessage) { + native.onEvent!(new FfiEvent(event).toBinary()); +} + +function audioStreamEos(streamHandle: bigint): PartialMessage { + return { + message: { + case: 'audioStreamEvent', + value: { streamHandle, message: { case: 'eos', value: {} } }, + }, + }; +} + +function captureAudioFrameCallback(asyncId: bigint): PartialMessage { + return { message: { case: 'captureAudioFrame', value: { asyncId } } }; +} + +describe('FfiClient event routing', () => { + let client: FfiClient; + let emitted: FfiEvent[]; + + beforeEach(() => { + globalThis._ffiClientInstance = undefined; + client = FfiClient.instance; + emitted = []; + client.on(FfiClientEvent.FfiEvent, (ev) => emitted.push(ev)); + }); + + it('hands an audio stream event to its own stream only, without emitting it', () => { + const first = vi.fn(); + const second = vi.fn(); + client.onAudioStreamEvent(1n, first); + client.onAudioStreamEvent(2n, second); + + deliver(audioStreamEos(2n)); + + expect(first).not.toHaveBeenCalled(); + expect(second).toHaveBeenCalledOnce(); + expect(second.mock.calls[0]![0].message.value.streamHandle).toBe(2n); + expect(emitted).toEqual([]); + }); + + it('drops an audio stream event once its stream is removed', () => { + const listener = vi.fn(); + client.onAudioStreamEvent(1n, listener); + client.offAudioStreamEvent(1n); + + deliver(audioStreamEos(1n)); + + expect(listener).not.toHaveBeenCalled(); + expect(emitted).toEqual([]); + }); + + it('resolves the waiter for a capture callback, without emitting it', async () => { + const callback = client.waitForCaptureAudioFrame(7n); + + deliver(captureAudioFrameCallback(7n)); + + await expect(callback).resolves.toMatchObject({ asyncId: 7n }); + expect(emitted).toEqual([]); + }); + + it('still emits a capture callback that nothing waits for', () => { + deliver(captureAudioFrameCallback(8n)); + + expect(emitted).toHaveLength(1); + expect(emitted[0]!.message.case).toBe('captureAudioFrame'); + }); + + it('emits every other event as before', () => { + deliver({ message: { case: 'disconnect', value: { asyncId: 9n } } }); + + expect(emitted).toHaveLength(1); + expect(emitted[0]!.message.case).toBe('disconnect'); + }); +}); diff --git a/packages/livekit-rtc/src/ffi_client.ts b/packages/livekit-rtc/src/ffi_client.ts index f81ac8d8..45b31072 100644 --- a/packages/livekit-rtc/src/ffi_client.ts +++ b/packages/livekit-rtc/src/ffi_client.ts @@ -2,6 +2,7 @@ // // SPDX-License-Identifier: Apache-2.0 import { + type CaptureAudioFrameCallback, FfiEvent, FfiHandle, FfiRequest, @@ -43,6 +44,17 @@ export class FfiClient extends (EventEmitter as new () => TypedEmitter void>(); + + // Every audio frame an agent speaks gets its own callback. Waiting on the emitter would add and + // remove a listener per frame and make every other waiter test it. + private captureAudioFrameWaiters = new Map< + bigint, + (callback: CaptureAudioFrameCallback) => void + >(); + constructor() { super(); this.setMaxListeners(0); @@ -50,6 +62,19 @@ export class FfiClient extends (EventEmitter as new () => TypedEmitter { const event = FfiEvent.fromBinary(event_data); + if (event.message.case === 'audioStreamEvent') { + this.audioStreamListeners.get(event.message.value.streamHandle!)?.(event); + return; + } + if (event.message.case === 'captureAudioFrame') { + const asyncId = event.message.value.asyncId!; + const waiter = this.captureAudioFrameWaiters.get(asyncId); + if (waiter) { + this.captureAudioFrameWaiters.delete(asyncId); + waiter(event.message.value); + return; + } + } this.emit(FfiClientEvent.FfiEvent, event); }, true, @@ -57,6 +82,23 @@ export class FfiClient extends (EventEmitter as new () => TypedEmitter void) { + this.audioStreamListeners.set(streamHandle, listener); + } + + /** @internal */ + offAudioStreamEvent(streamHandle: bigint) { + this.audioStreamListeners.delete(streamHandle); + } + + /** @internal Call in the same tick as the request, so its callback cannot arrive first. */ + waitForCaptureAudioFrame(asyncId: bigint): Promise { + return new Promise((resolve) => { + this.captureAudioFrameWaiters.set(asyncId, resolve); + }); + } + request(req: PartialMessage): T { const request = new FfiRequest(req); const req_data = request.toBinary(); diff --git a/packages/livekit-rtc/src/room.ts b/packages/livekit-rtc/src/room.ts index 4ed47938..c43554e5 100644 --- a/packages/livekit-rtc/src/room.ts +++ b/packages/livekit-rtc/src/room.ts @@ -549,6 +549,11 @@ export class Room extends (EventEmitter as new () => TypedEmitter } private onFfiEvent = async (ffiEvent: FfiEvent) => { + // Every room in the process sees every FFI event, most of them other streams' audio frames. + // processFfiEvent acts on these two cases only, so the rest return before waiting on the lock. + if (ffiEvent.message.case != 'roomEvent' && ffiEvent.message.case != 'rpcMethodInvocation') { + return; + } const unlock = await this.ffiEventLock.lock(); try { if (!this.localParticipant || !this.ffiHandle || !this.info) {