Use node stream for AudioStream and VideoStream#458
Conversation
|
| /** | ||
| * Push an item into the buffer. If the buffer is full, the newest item will | ||
| * be removed. | ||
| */ |
There was a problem hiding this comment.
This is what the python implementation does. Wouldn't it make sense if the oldest time is evicted?
There was a problem hiding this comment.
When thinking about a generic ring buffer I think you're right that it should shift the values to the left by one and insert the newest at the end.
I'm not sure if there are additional considerations in the python implementation though.
| * Get an item from the buffer. If the buffer is empty, this will return a promise | ||
| * that resolves when a new item is pushed. | ||
| */ | ||
| async get(): Promise<T> { |
There was a problem hiding this comment.
I think it's best to just move the locking inside here. Are there any downsides to this?
There was a problem hiding this comment.
The way it's implemented now doesn't guarantee call order (I believe, not verified), so locking would definitely be better.
Shifting the lock exclusively into here will make it harder to raise errors from within the AudioStream's next handler (we're currently not doing that anyways, but we should keep that in mind in case we ever want to)
There was a problem hiding this comment.
I've talked to @nbsp in the past about moving more into the direction of utilising built in node (web) streams for our queues.
See https://github.com/livekit/agents-js/pull/194/files for a first step into that direction.
I'm not entirely sure it's the right call for all our queuing logic, but in general, if there's a way to make this work consistently across the repo with built-in primitives that's what I'd prefer.
Having custom implementations of both RingQueue and AsyncIterableQueue sounds a bit redundant though, esp. when the RingQueue in this PR acts as an async queue and we don't really make use of the capacity parameter.
If we stick to a custom implementation, let's think about merging these two into one.
We've had some bugs in the past around async queuing and having custom implementations adds one more place where things can potentially go wrong.
| /** | ||
| * Push an item into the buffer. If the buffer is full, the newest item will | ||
| * be removed. | ||
| */ |
There was a problem hiding this comment.
When thinking about a generic ring buffer I think you're right that it should shift the values to the left by one and insert the newest at the end.
I'm not sure if there are additional considerations in the python implementation though.
| * Get an item from the buffer. If the buffer is empty, this will return a promise | ||
| * that resolves when a new item is pushed. | ||
| */ | ||
| async get(): Promise<T> { |
There was a problem hiding this comment.
The way it's implemented now doesn't guarantee call order (I believe, not verified), so locking would definitely be better.
Shifting the lock exclusively into here will make it harder to raise errors from within the AudioStream's next handler (we're currently not doing that anyways, but we should keep that in mind in case we ever want to)
Thanks super helpful context.
This makes total sense let me see if I can get this working with |
Shubhrakanti
changed the title
| if (controller.desiredSize && controller.desiredSize > 0) { | ||
| controller.enqueue(ev); | ||
| } else { | ||
| console.warn('Audio stream buffer is full, dropping frame'); | ||
| } |
There was a problem hiding this comment.
This is what the python implementation does. Wouldn't it make sense if the oldest time is evicted?
@theomonnom if this needs to be the behavior here as well then maybe node streams are not a good choice because we don't have explicit control over the buffer eviction policy.
| }, | ||
| ); | ||
|
|
||
| this.reader = source.pipeThrough(transformStream).getReader(); |
There was a problem hiding this comment.
we could acquire the reader lazily in next() and also make sure to call releaseLock when done
| }, | ||
| }); | ||
|
|
||
| const transformStream = new TransformStream<FfiEvent, AudioFrame>( |
There was a problem hiding this comment.
is the transform actually needed? wondering if we can handle this directly in the ReadableStream?
There was a problem hiding this comment.
TLDR
We can't have Video/AudioStream be a ReadableStream without introducing breaking changes to the current api. Right now we don't automatically close the stream when an iterator finishes. If we extend the ReadableStream interface, this behavior is unavoidable.
Explanation
AsyncIterators in JS have a return method that is invoked when iterator stops or exits due to a break; statement. When this happens, the ReadableStream under the hood will invoke the cancel() method on the source.
For example if we had code like this
const stream = new VideoStream(track);
let i = 0;
for await (const frame of stream) {
i++;
console.log('loop 1 timestamp', frame.timestampUs);
if (i > 5) {
break;
}
}
console.log('finished loop 1');
for await (const frame of stream) {
i++;
console.log('loop 2 timestamp', frame.timestampUs);
if (i > 10) {
break;
}
}
console.log('finished loop 2');
}Output from existing implementation.
loop 1 timestamp 352996437000n
loop 1 timestamp 352996470000n
loop 1 timestamp 352996527000n
...
finished loop 1 -> no resource clean up
loop 2 timestamp 352996537000n
loop 2 timestamp 352996660000n
loop 2 timestamp 352996957000n
...
finished loop 2
Loop 2 pulls of the stream because there is no resource clean up.
Output from new implementation with ReadableStream
loop 1 timestamp 352996437000n
loop 1 timestamp 352996470000n
loop 1 timestamp 352996527000n
...
finished loop 1 -> under the hood source.close() is called
finished loop 2
Loop 2 is empty because the stream is closed.
I tried to hack around this by doing something like
class VideoStreamSource implements UnderlyingSource<VideoFrameEvent> {
...
cancel() {
// do nothing on the inherited cancel method
}
close() {
this.closed = true;
FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
this.ffiHandle.dispose();
console.log('+++ cancelled');
}
}
export class VideoStream extends ReadableStream<VideoFrameEvent> {
private source: VideoStreamSource;
constructor(track: Track) {
const source = new VideoStreamSource(track);
super(source);
this.source = source;
}
close() {
this.source.cancel();
}
}but then you get an error throw new ERR_INVALID_STATE.TypeError('Controller is already closed');
Proposals
1. Make the breaking change and update the API with a new version.
Ideally we get all the benefits of ReadableStream (like locking, tee(), pipeTo/Through).
I also think the current implementation allows user code to have subtle bugs. For example above, if we do something like
for await (const frame of stream) {
...
}
await someReallyLongRunningFunction()
for await (const frame of stream) {
...
}The FFiHandle in the VideoStream will continue to push frames into the buffer even though we're not consuming from the stream. Might cause OOM.
2. Just use ReadableStream inside VideoStream
This makes the current implementation logic cleaner and there are no breaking changes, but we loose the benefits of ReadableStream.
| if (!this.closed) { | ||
| this.controller.enqueue(videoFrameEvent); |
There was a problem hiding this comment.
This is okay to do without checking desiredSize first. The official standard has an example of a push stream with no back pressure support. We should just make a comment making it explicit. If the request arises we should limit the queue size based on memory size in bytes not number of frames.
| private controller?: ReadableStreamDefaultController<VideoFrameEvent>; | ||
| private ffiHandle: FfiHandle; | ||
| private closed = false; |
There was a problem hiding this comment.
technically making these private and moving them here would not be backward compatible. Any reason they shouldn't be?
There was a problem hiding this comment.
Discussed offline. It's okay
| [Symbol.asyncIterator](): AudioStream { | ||
| return this; | ||
| export class AudioStream extends ReadableStream<AudioFrame> { | ||
| constructor( |
There was a problem hiding this comment.
we would want to add the more specific overload constructors here that @typester had added for the AudioStream
2 participants
Use node streams to buffer incoming audio and video frames