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) {