From 413b96fded6f119f3385f76faf5be5fe965b9393 Mon Sep 17 00:00:00 2001 From: Alex Andres Date: Sat, 24 Oct 2026 15:57:53 +0200 Subject: [PATCH] feat: limit and measure the client API's handler threads Event and frame pools can be capped with eventThreads and frameThreads; peers and subscriptions take turns of at most 64 tasks or 10 ms, and a limited pool rejects nothing while it runs. WebRtc.loadStats() reports queued, oldest waiting, active and dropped work. The audio buffer of onAudio is set by audioBuffer or per subscription. --- docs/guide/client-api.md | 53 +++- .../onvoid/webrtc/client/AudioDispatcher.java | 89 ++----- .../dev/onvoid/webrtc/client/Dispatcher.java | 214 ++++++++++++++++ .../onvoid/webrtc/client/FrameDispatcher.java | 87 +------ .../dev/onvoid/webrtc/client/LoadStats.java | 186 ++++++++++++++ .../onvoid/webrtc/client/LocalMediaImpl.java | 2 +- .../onvoid/webrtc/client/LocalVideoImpl.java | 3 +- .../dev/onvoid/webrtc/client/PeerImpl.java | 4 +- .../dev/onvoid/webrtc/client/RemoteAudio.java | 18 ++ .../onvoid/webrtc/client/RemoteAudioImpl.java | 14 +- .../onvoid/webrtc/client/RemoteVideoImpl.java | 3 +- .../onvoid/webrtc/client/SerialExecutor.java | 85 +++++-- .../java/dev/onvoid/webrtc/client/WebRtc.java | 172 ++++++++++++- .../dev/onvoid/webrtc/client/Workload.java | 126 ++++++++++ .../onvoid/webrtc/client/DispatcherTests.java | 232 ++++++++++++++++++ .../dev/onvoid/webrtc/client/LoadTests.java | 178 ++++++++++++++ .../webrtc/client/SerialExecutorTests.java | 71 +++++- 17 files changed, 1363 insertions(+), 174 deletions(-) create mode 100644 webrtc/src/main/java/dev/onvoid/webrtc/client/Dispatcher.java create mode 100644 webrtc/src/main/java/dev/onvoid/webrtc/client/LoadStats.java create mode 100644 webrtc/src/main/java/dev/onvoid/webrtc/client/Workload.java create mode 100644 webrtc/src/test/java/dev/onvoid/webrtc/client/DispatcherTests.java create mode 100644 webrtc/src/test/java/dev/onvoid/webrtc/client/LoadTests.java diff --git a/docs/guide/client-api.md b/docs/guide/client-api.md index 1460d4df..d583fb49 100644 --- a/docs/guide/client-api.md +++ b/docs/guide/client-api.md @@ -60,6 +60,9 @@ WebRtc rtc = WebRtc.builder() | `hardwareVideoCodecs(boolean)` | `false` | Encodes and decodes video on the GPU where it can, as the [Video Codecs](/guide/advanced/video-codecs) guide describes. | | `fieldTrials(Map)` | none | [Field trials](/guide/advanced/field-trials). | | `eventExecutor(Executor)` | threads of the runtime | Where handlers run, see [Threads](#threads). | +| `eventThreads(int)` | unlimited | The most threads the runtime runs handlers on, see [Limiting Threads](#limiting-threads). | +| `frameThreads(int)` | unlimited | The most threads that run `onFrame` and `onAudio` handlers. | +| `audioBuffer(Duration)` | 5 seconds | How much audio an `onAudio` subscription keeps for a handler that falls behind, see [Audio](#audio). | | `errorHandler(Consumer)` | the thread's uncaught exception handler | Receives what handlers throw, and failures nobody else hears of. | | `operationTimeout(Duration)` | 30 seconds | How long an operation such as applying a description may take before it fails with a `TimeoutException`. | | `closeTimeout(Duration)` | 10 seconds | How long `close()` waits. | @@ -198,7 +201,7 @@ Every method of the client API may be called from any thread, including from ins |---|---|---| | Your calls into the API | your thread | They don't block. The exceptions are `WebRtc.createPeer`, which waits briefly for the connection, and `WebRtc.close()`, which waits up to the close timeout. | | Peer, track and channel handlers, future callbacks, `Signaling.send` | the peer's event thread | Yes, but that delays only the events of this peer. They must not wait for a future of this API, see below. | -| `onFrame` and `onAudio` handlers | a thread of each subscription | Better not. A slow video handler gets the newest frame next, and the frames in between are dropped. | +| `onFrame` and `onAudio` handlers | a thread of each subscription | Better not. A slow video handler gets the newest frame next, and the frames in between are dropped; a slow audio handler loses the oldest audio beyond its buffer. | | `LocalTrack.onEnded` and `Devices.onChange` handlers | a background thread of the runtime | Briefly. | | WebRTC's own callbacks | WebRTC's threads | Not your code. The library only copies what they report and queues it. | @@ -206,6 +209,46 @@ The handlers of one peer never run at the same time, and they run in the order o An executor of your own that rejects work, such as a bounded pool that is full, loses the events it rejects, and the rejection goes to the error handler. The futures those events would have completed complete all the same, on the thread that found them rejected. Once the runtime is closed, every future of this API that is not done fails with `PeerClosedException`, including those of calls made after the close. +### Limiting Threads + +By default the runtime starts a thread for each peer whose events wait and for each subscription whose frames wait, so a thousand peers with slow handlers take a thousand threads. `eventThreads(n)` and `frameThreads(n)` cap them: + +```java +WebRtc rtc = WebRtc.builder() + .eventThreads(8) // handlers of all peers, future callbacks, Signaling.send + .frameThreads(4) // onFrame and onAudio handlers, local previews included + .build(); +``` + +With a limit, the peers take turns, and so do the subscriptions: one that has work runs up to 64 handlers, for at most 10 ms, then goes to the back of the line. A slow handler thus delays the others but never starves them, and a peer's events still run one at a time and in order. The limit rejects nothing: events wait for their turn and are never dropped, so no future stays open, and once the runtime closes, the futures still open fail after the events queued before them. Media that waits is bounded by its subscription instead, by the `FrameDelivery` of a video handler and the audio buffer of an audio handler, which drop the oldest frames and audio when they are full. + +A limit has one cost: a handler that waits for what another peer's handler does, say on a latch, can wait forever once all threads are taken by such waits. Handlers that wait for something outside the API should hand the work to a thread of their own. + +The control threads of the library, which call into WebRTC, are not limited. Their tasks wait for one another, such as a camera switch for every peer to take the new track. + +`rtc.loadStats()` tells how much work waits and runs, at this moment and cheaply enough to be read every second: + +```java +LoadStats load = rtc.loadStats(); + +LoadStats.Work events = load.events(); +metrics.gauge("webrtc.events.queued", events.queued()); // events waiting for a thread +metrics.gauge("webrtc.events.oldest_ms", events.oldestWaiting().toMillis()); // how long the oldest has waited +metrics.gauge("webrtc.events.active", events.active()); // handlers running now +metrics.gauge("webrtc.video.dropped", load.video().dropped()); // frames dropped since start +metrics.gauge("webrtc.audio.dropped", load.audio().dropped()); // audio chunks dropped since start +metrics.gauge("webrtc.event_threads", load.eventThreads()); +``` + +| Part | What waits | What `dropped()` counts | +|---|---|---| +| `control()` | tasks of the library that call into WebRTC | tasks the control pool rejected while running, which it never does | +| `events()` | events of all peers, for handlers, future callbacks and `Signaling.send` | events an `eventExecutor` of your own rejected | +| `video()` | frames for `onFrame` handlers | frames a handler had no time for | +| `audio()` | chunks for `onAudio` handlers | chunks beyond the audio buffer | + +A queue that keeps growing, or an oldest wait that keeps getting longer, tells that handlers do not keep up. A higher limit helps handlers that wait for I/O; handlers that compute only spread the delay over more threads. + ### Handlers on the UI Thread Pass the UI toolkit's executor, and every handler may touch UI components directly: @@ -442,7 +485,7 @@ A track ends when the remote peer stops sending it or the peer closes. ### Video Frames -Each `onFrame` subscription has a thread of its own, so a slow renderer holds up neither WebRTC's decoder nor the peer's other events. By default the handler always gets the newest frame, and the frames it had no time for are dropped. `FrameDelivery.queue(n)` keeps up to `n` frames instead: +Each `onFrame` subscription has a thread of its own, so a slow renderer holds up neither WebRTC's decoder nor the peer's other events. With `frameThreads(n)` the subscriptions share `n` threads and take turns, see [Limiting Threads](#limiting-threads). By default the handler always gets the newest frame, and the frames it had no time for are dropped. `FrameDelivery.queue(n)` keeps up to `n` frames instead: ```java video.onFrame(FrameDelivery.queue(10), frame -> encoder.encode(frame)); @@ -454,6 +497,12 @@ A frame is valid while the handler runs. Call `frame.retain()` to keep it longer `onAudio` hands decoded audio to a handler in order, usually 10 ms of 16-bit PCM per `AudioChunk`. Audio reaches these handlers only while the runtime plays out, which a `HEADLESS` runtime always does and a `DEVICES` or `CUSTOM` runtime does to its speakers. `setVolume(double)` changes the playout gain of a track, from 0 to 10, where 1 plays it as received. +A handler that falls behind gets the audio of the last 5 seconds; the oldest audio beyond that is dropped and counted in `rtc.loadStats().audio().dropped()`. `WebRtc.Builder.audioBuffer(Duration)` changes the buffer of every subscription, and `onAudio(Duration, handler)` that of one. A speech recognizer that must stay live keeps a short buffer; a recorder that can catch up after a stall keeps a long one, at about 190 KiB per second of 48 kHz stereo: + +```java +track.asAudio().onAudio(Duration.ofMillis(200), chunk -> recognizer.feed(chunk.samples())); +``` + ### Receiving Without Sending A viewer that sends nothing asks for what it wants to receive, so that the offer includes it: diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/AudioDispatcher.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/AudioDispatcher.java index 2a4b26e1..883e8e1b 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/AudioDispatcher.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/AudioDispatcher.java @@ -16,39 +16,24 @@ package dev.onvoid.webrtc.client; -import java.util.ArrayDeque; +import java.time.Duration; import java.util.function.Consumer; import dev.onvoid.webrtc.media.audio.AudioTrackSink; /** * Hands the audio of a track to a handler on a thread of its own, in order. - * When the handler falls more than {@link #CAPACITY} chunks behind, the oldest - * waiting chunk is dropped. + * When the chunks waiting for the handler last longer than the buffer, the + * oldest are dropped. */ -final class AudioDispatcher implements AudioTrackSink { - - /** Five seconds of 10 ms chunks. */ - private static final int CAPACITY = 500; - - private final Consumer handler; - - private final SerialExecutor executor; - - private final Consumer errorHandler; - - private final ArrayDeque queue = new ArrayDeque<>(); - - private boolean scheduled; - - private boolean closed; - - - AudioDispatcher(Consumer handler, SerialExecutor executor, - Consumer errorHandler) { - this.handler = handler; - this.executor = executor; - this.errorHandler = errorHandler; +final class AudioDispatcher extends Dispatcher implements AudioTrackSink { + + /** + * @param buffer how much audio may wait for the handler. + */ + AudioDispatcher(Consumer handler, Duration buffer, SerialExecutor executor, + Workload workload, Consumer errorHandler) { + super(handler, micros(buffer), executor, workload, errorHandler); } @Override @@ -57,51 +42,21 @@ public void onData(byte[] data, int bitsPerSample, int sampleRate, int channels, return; } - AudioChunk chunk = new AudioChunk(data.clone(), sampleRate, channels, frames); - - synchronized (queue) { - if (closed) { - return; - } - - queue.add(chunk); - - if (queue.size() > CAPACITY) { - queue.poll(); - } - if (!scheduled) { - scheduled = true; - executor.execute(this::drain); - } - } + offer(new AudioChunk(data.clone(), sampleRate, channels, frames)); } - void close() { - synchronized (queue) { - closed = true; - queue.clear(); - } + /** + * @return the length of the chunk in microseconds. + */ + @Override + long weight(AudioChunk chunk) { + return chunk.sampleRate() > 0 ? chunk.frames() * 1_000_000L / chunk.sampleRate() : 0; } - private void drain() { - while (true) { - AudioChunk chunk; - - synchronized (queue) { - chunk = queue.poll(); + private static long micros(Duration duration) { + long seconds = duration.getSeconds(); - if (chunk == null) { - scheduled = false; - return; - } - } - - try { - handler.accept(chunk); - } - catch (Throwable e) { - errorHandler.accept(e); - } - } + return seconds >= Long.MAX_VALUE / 1_000_000 ? Long.MAX_VALUE + : seconds * 1_000_000 + duration.getNano() / 1_000; } } diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/Dispatcher.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/Dispatcher.java new file mode 100644 index 00000000..9e4bae33 --- /dev/null +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/Dispatcher.java @@ -0,0 +1,214 @@ +/* + * Copyright 2026 Alex Andres + * + * 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. + */ + +package dev.onvoid.webrtc.client; + +import java.util.ArrayDeque; +import java.util.function.Consumer; + +/** + * Hands the media of one subscription to its handler, in order, so that a + * slow handler holds up neither WebRTC nor any other subscription. + *

+ * Each item is a task of its own on the subscription's executor, so that a + * handler that falls behind takes turns with the other subscriptions of a + * shared pool instead of keeping a thread. When the waiting items weigh more + * than the budget, the oldest are dropped; the newest one is always kept. + * + * @param the media item. + */ +abstract class Dispatcher { + + private final Consumer handler; + + private final long budget; + + private final SerialExecutor executor; + + private final Workload workload; + + private final Consumer errorHandler; + + private final ArrayDeque> queue = new ArrayDeque<>(); + + private final Runnable next = this::handleNext; + + private final Workload.Backlog backlog = this::count; + + private long queuedWeight; + + private boolean scheduled; + + private boolean closed; + + + /** + * @param budget the weight of the items that may wait. + * @param workload counts the items that wait, run and are dropped. + */ + Dispatcher(Consumer handler, long budget, SerialExecutor executor, + Workload workload, Consumer errorHandler) { + this.handler = handler; + this.budget = budget; + this.executor = executor; + this.workload = workload; + this.errorHandler = errorHandler; + } + + /** + * @return how much of the budget the item takes. + */ + abstract long weight(T item); + + /** + * Lets go of an item once it was handled, dropped or left over at close. + */ + void release(T item) { + } + + /** + * Queues an item for the handler, and drops the oldest ones over the + * budget. + */ + final void offer(T item) { + boolean schedule = false; + + synchronized (queue) { + if (closed) { + release(item); + return; + } + if (queue.isEmpty()) { + workload.waiting(backlog); + } + + Entry entry = new Entry<>(item, weight(item), workload.now()); + + queue.add(entry); + queuedWeight += entry.weight; + + while (queuedWeight > budget && queue.size() > 1) { + Entry oldest = queue.poll(); + + queuedWeight -= oldest.weight; + workload.dropped(1); + release(oldest.item); + } + + if (!scheduled) { + scheduled = true; + schedule = true; + } + } + + if (schedule) { + executor.execute(next); + } + } + + /** + * Drops what waits, and what comes later. + */ + final void close() { + synchronized (queue) { + closed = true; + + for (Entry entry : queue) { + release(entry.item); + } + + queue.clear(); + queuedWeight = 0; + workload.idle(backlog); + } + } + + private void handleNext() { + Entry entry; + + synchronized (queue) { + entry = queue.poll(); + + if (entry == null) { + scheduled = false; + return; + } + + queuedWeight -= entry.weight; + + if (queue.isEmpty()) { + workload.idle(backlog); + } + } + + workload.started(); + + try { + handler.accept(entry.item); + } + catch (Throwable e) { + errorHandler.accept(e); + } + finally { + workload.finished(); + release(entry.item); + } + + boolean more; + + synchronized (queue) { + more = !queue.isEmpty(); + + if (!more) { + scheduled = false; + } + } + + if (more) { + // Behind what else the executor has, which ends its turn on the + // pool once the turn has run long enough. + executor.execute(next); + } + } + + private void count(Workload.Tally tally) { + synchronized (queue) { + Entry oldest = queue.peek(); + + if (oldest != null) { + tally.add(queue.size(), oldest.queuedAt); + } + } + } + + + + private static final class Entry { + + final T item; + + final long weight; + + final long queuedAt; + + + Entry(T item, long weight, long queuedAt) { + this.item = item; + this.weight = weight; + this.queuedAt = queuedAt; + } + } +} diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/FrameDispatcher.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/FrameDispatcher.java index 457f77c7..1d4a888e 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/FrameDispatcher.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/FrameDispatcher.java @@ -16,7 +16,6 @@ package dev.onvoid.webrtc.client; -import java.util.ArrayDeque; import java.util.function.Consumer; import dev.onvoid.webrtc.media.video.VideoFrame; @@ -30,91 +29,25 @@ * owns: it releases a frame once the handler returned, when the queue is full * and the frame is dropped, or when the subscription closed. */ -final class FrameDispatcher implements VideoTrackSink { - - private final Consumer handler; - - private final int capacity; - - private final SerialExecutor executor; - - private final Consumer errorHandler; - - private final ArrayDeque queue = new ArrayDeque<>(); - - private boolean scheduled; - - private boolean closed; - +final class FrameDispatcher extends Dispatcher implements VideoTrackSink { FrameDispatcher(Consumer handler, FrameDelivery delivery, - SerialExecutor executor, Consumer errorHandler) { - this.handler = handler; - this.capacity = delivery.capacity; - this.executor = executor; - this.errorHandler = errorHandler; + SerialExecutor executor, Workload workload, Consumer errorHandler) { + super(handler, delivery.capacity, executor, workload, errorHandler); } @Override public void onVideoFrame(VideoFrame frame) { - VideoFrame dropped = null; - - synchronized (queue) { - if (closed) { - dropped = frame; - } - else { - queue.add(frame); - - if (queue.size() > capacity) { - dropped = queue.poll(); - } - if (!scheduled) { - scheduled = true; - executor.execute(this::drain); - } - } - } - - if (dropped != null) { - dropped.release(); - } + offer(frame); } - void close() { - synchronized (queue) { - closed = true; - - for (VideoFrame frame : queue) { - frame.release(); - } - - queue.clear(); - } + @Override + long weight(VideoFrame frame) { + return 1; } - private void drain() { - while (true) { - VideoFrame frame; - - synchronized (queue) { - frame = queue.poll(); - - if (frame == null) { - scheduled = false; - return; - } - } - - try { - handler.accept(frame); - } - catch (Throwable e) { - errorHandler.accept(e); - } - finally { - frame.release(); - } - } + @Override + void release(VideoFrame frame) { + frame.release(); } } diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/LoadStats.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/LoadStats.java new file mode 100644 index 00000000..20da70e7 --- /dev/null +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/LoadStats.java @@ -0,0 +1,186 @@ +/* + * Copyright 2026 Alex Andres + * + * 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. + */ + +package dev.onvoid.webrtc.client; + +import java.time.Duration; + +/** + * How much work the threads of a runtime have, at one moment. Read it with + * {@link WebRtc#loadStats()}, as often as needed; taking it is cheap and + * waits for nothing. + *

+ * A queue that keeps growing, or an oldest waiting item that keeps getting + * older, tells that handlers do not keep up. Raising a {@link + * WebRtc.Builder#eventThreads(int) thread limit} helps when the handlers + * wait for something; when they compute, it only spreads the delay. + */ +public final class LoadStats { + + private final Work control; + + private final Work events; + + private final Work video; + + private final Work audio; + + private final int controlThreads; + + private final int eventThreads; + + private final int frameThreads; + + + LoadStats(Work control, Work events, Work video, Work audio, + int controlThreads, int eventThreads, int frameThreads) { + this.control = control; + this.events = events; + this.video = video; + this.audio = audio; + this.controlThreads = controlThreads; + this.eventThreads = eventThreads; + this.frameThreads = frameThreads; + } + + /** + * @return the tasks of the library that call into WebRTC. They wait + * there only while the control threads are busy, so a backlog here points + * at slow native calls, not at handlers. + */ + public Work control() { + return control; + } + + /** + * @return the events of all peers waiting for their handlers, future + * callbacks and {@code Signaling.send} calls. Events are never dropped; + * {@link Work#dropped()} counts those an {@link + * WebRtc.Builder#eventExecutor(java.util.concurrent.Executor) executor of + * the application} rejected. + */ + public Work events() { + return events; + } + + /** + * @return the video frames waiting for {@code onFrame} handlers, local + * previews included. {@link Work#dropped()} counts the frames a handler + * had no time for. + */ + public Work video() { + return video; + } + + /** + * @return the audio chunks waiting for {@code onAudio} handlers. {@link + * Work#dropped()} counts the chunks that no longer fit into the audio + * buffer of their subscription. + */ + public Work audio() { + return audio; + } + + /** + * @return the threads of the control pool right now. + */ + public int controlThreads() { + return controlThreads; + } + + /** + * @return the threads of the runtime's event pool right now; 0 when the + * application's executor runs the events. + */ + public int eventThreads() { + return eventThreads; + } + + /** + * @return the threads that run {@code onFrame} and {@code onAudio} + * handlers right now. + */ + public int frameThreads() { + return frameThreads; + } + + @Override + public String toString() { + return "LoadStats[control " + control + " on " + controlThreads + " threads, events " + events + + " on " + eventThreads + " threads, video " + video + ", audio " + audio + + " on " + frameThreads + " threads]"; + } + + + + /** The work of one kind. */ + public static final class Work { + + private final int queued; + + private final Duration oldestWaiting; + + private final int active; + + private final long dropped; + + + Work(int queued, Duration oldestWaiting, int active, long dropped) { + this.queued = queued; + this.oldestWaiting = oldestWaiting; + this.active = active; + this.dropped = dropped; + } + + /** + * @return the items waiting for a thread, of all peers and + * subscriptions together. + */ + public int queued() { + return queued; + } + + /** + * @return how long the item that waits longest has waited; zero when + * none waits. + */ + public Duration oldestWaiting() { + return oldestWaiting; + } + + /** + * @return the items being handled right now, which is the number of + * handlers running. + */ + public int active() { + return active; + } + + /** + * @return the items dropped since the runtime was built, without + * those left over when a subscription, peer or runtime closed. + */ + public long dropped() { + return dropped; + } + + @Override + public String toString() { + return queued + " queued (oldest " + oldestWaiting.toMillis() + " ms), " + + active + " active, " + dropped + " dropped"; + } + } +} diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalMediaImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalMediaImpl.java index e55802fe..dff148a5 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalMediaImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalMediaImpl.java @@ -75,7 +75,7 @@ final class LocalMediaImpl implements LocalMedia { LocalMediaImpl(WebRtc rtc, AudioDeviceModuleBase audioModule) { this.rtc = rtc; this.audioModule = audioModule; - this.executor = new SerialExecutor(rtc.controlPool, rtc::reportError, false, null); + this.executor = new SerialExecutor(rtc.controlPool, rtc.controlWork, rtc::reportError, false, null); } @Override diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalVideoImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalVideoImpl.java index 56835996..941d3517 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalVideoImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalVideoImpl.java @@ -44,7 +44,8 @@ public Subscription onFrame(Consumer handler) { requireNonNull(handler, "handler"); SerialExecutor executor = new SerialExecutor(rtc.framePool, rtc::reportError, false, null); - FrameDispatcher dispatcher = new FrameDispatcher(handler, FrameDelivery.LATEST, executor, rtc::reportError); + FrameDispatcher dispatcher = new FrameDispatcher(handler, FrameDelivery.LATEST, executor, rtc.videoWork, + rtc::reportError); synchronized (this) { if (!isLive()) { diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java index c829d873..9dbbbb3f 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java @@ -172,8 +172,8 @@ final class PeerImpl implements Peer { this.options = options; this.name = name; this.signaling = options.signaling; - this.control = new SerialExecutor(rtc.controlPool, rtc::reportError, false, null); - this.events = new SerialExecutor(rtc.eventExecutor, rtc::reportError, true, + this.control = new SerialExecutor(rtc.controlPool, rtc.controlWork, rtc::reportError, false, null); + this.events = new SerialExecutor(rtc.eventExecutor, rtc.eventWork, rtc::reportError, true, rtc.ownsEventExecutor ? "webrtc-client-events-" + name : null); this.stateHandlers = new Handlers<>(rtc::reportError); this.trackHandlers = new Handlers<>(rtc::reportError); diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteAudio.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteAudio.java index d64aca72..16122f54 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteAudio.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteAudio.java @@ -16,6 +16,7 @@ package dev.onvoid.webrtc.client; +import java.time.Duration; import java.util.function.Consumer; /** @@ -27,6 +28,10 @@ public interface RemoteAudio extends RemoteTrack { * Calls the handler with the decoded audio, in order. Audio reaches * handlers only while the runtime plays out, which a {@link * AudioMode#HEADLESS} runtime always does. + *

+ * A handler that falls behind gets the audio of the last {@link + * WebRtc.Builder#audioBuffer(Duration) audio buffer} of the runtime; the + * oldest audio beyond that is dropped. * * @param handler called on a thread of this subscription. * @@ -34,6 +39,19 @@ public interface RemoteAudio extends RemoteTrack { */ Subscription onAudio(Consumer handler); + /** + * Like {@link #onAudio(Consumer)}, with a buffer of its own: a handler + * that falls behind gets the audio of the last {@code buffer}, and the + * oldest audio beyond that is dropped. A buffer shorter than one chunk + * keeps only the newest chunk. + * + * @param buffer the most audio that waits for the handler. + * @param handler called on a thread of this subscription. + * + * @return the subscription that stops the audio. + */ + Subscription onAudio(Duration buffer, Consumer handler); + /** * Sets the playout gain of this track. * diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteAudioImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteAudioImpl.java index aca25b8d..01df2d9e 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteAudioImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteAudioImpl.java @@ -18,6 +18,7 @@ import static java.util.Objects.requireNonNull; +import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.function.Consumer; @@ -52,10 +53,21 @@ public RemoteAudio asAudio() { @Override public Subscription onAudio(Consumer handler) { + return onAudio(peer.rtc.audioBuffer, handler); + } + + @Override + public Subscription onAudio(Duration buffer, Consumer handler) { + requireNonNull(buffer, "buffer"); requireNonNull(handler, "handler"); + if (buffer.isNegative() || buffer.isZero()) { + throw new IllegalArgumentException("The buffer must be positive, got " + buffer); + } + SerialExecutor executor = new SerialExecutor(peer.rtc.framePool, peer.rtc::reportError, false, null); - AudioDispatcher dispatcher = new AudioDispatcher(handler, executor, peer.rtc::reportError); + AudioDispatcher dispatcher = new AudioDispatcher(handler, buffer, executor, peer.rtc.audioWork, + peer.rtc::reportError); onControl(() -> { dispatchers.add(dispatcher); diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteVideoImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteVideoImpl.java index 60999bc9..67a23124 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteVideoImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteVideoImpl.java @@ -62,7 +62,8 @@ public Subscription onFrame(FrameDelivery delivery, Consumer handler requireNonNull(handler, "handler"); SerialExecutor executor = new SerialExecutor(peer.rtc.framePool, peer.rtc::reportError, false, null); - FrameDispatcher dispatcher = new FrameDispatcher(handler, delivery, executor, peer.rtc::reportError); + FrameDispatcher dispatcher = new FrameDispatcher(handler, delivery, executor, peer.rtc.videoWork, + peer.rtc::reportError); onControl(() -> { dispatchers.add(dispatcher); diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/SerialExecutor.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/SerialExecutor.java index 4d21c8eb..dc3d2931 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/SerialExecutor.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/SerialExecutor.java @@ -32,6 +32,12 @@ * A task that throws does not stop the ones after it; its exception goes to * the error handler. *

+ * Each turn on the executor runs at most {@link #BATCH_SIZE} tasks and ends + * once it has run for {@link #TURN_NANOS}, after which the executor queues + * its next turn behind those of the others. A pool with a fixed number of + * threads and a FIFO queue thus serves every executor in turn, however slow + * the tasks of some of them are, and it holds at most one turn of each. + *

* When the executor rejects the tasks, they are dropped, and those submitted * with a fallback have it run instead, on the thread that found them * rejected. A rejection by an executor that was not shut down goes to the @@ -40,7 +46,10 @@ final class SerialExecutor implements Executor { /** Tasks run per turn, so one busy executor cannot starve the others of a shared pool. */ - private static final int BATCH_SIZE = 64; + static final int BATCH_SIZE = 64; + + /** How long a turn runs tasks before it gives the others of a shared pool a turn. */ + static final long TURN_NANOS = 10_000_000; /** Marks the threads that run event tasks, to catch a future waited on there. */ private static final ThreadLocal EVENT_DEPTH = new ThreadLocal() { @@ -59,7 +68,11 @@ protected int[] initialValue() { private final String threadName; - private final ArrayDeque tasks = new ArrayDeque<>(); + private final Workload workload; + + private final ArrayDeque tasks = new ArrayDeque<>(); + + private final Workload.Backlog backlog = this::count; private boolean scheduled; @@ -76,7 +89,17 @@ protected int[] initialValue() { */ SerialExecutor(Executor executor, Consumer errorHandler, boolean eventThread, String threadName) { + this(executor, new Workload(System::nanoTime), errorHandler, eventThread, threadName); + } + + /** + * @param workload counts the tasks that wait and run, and those dropped + * when the executor rejects them while it runs. + */ + SerialExecutor(Executor executor, Workload workload, Consumer errorHandler, + boolean eventThread, String threadName) { this.executor = executor; + this.workload = workload; this.errorHandler = errorHandler; this.eventThread = eventThread; this.threadName = threadName; @@ -95,12 +118,20 @@ static boolean onEventThread() { * keeps a future the task would complete from staying open forever. */ void execute(Runnable task, Runnable dropped) { - execute(new Droppable(task, dropped)); + submit(new Task(task, dropped, workload.now())); } @Override public void execute(Runnable task) { + submit(new Task(task, null, workload.now())); + } + + private void submit(Task task) { synchronized (tasks) { + if (tasks.isEmpty()) { + workload.waiting(backlog); + } + tasks.add(task); if (scheduled) { @@ -125,26 +156,28 @@ private void schedule() { executor.execute(this::drain); } catch (RejectedExecutionException e) { - List dropped; + List dropped; synchronized (tasks) { scheduled = false; dropped = new ArrayList<>(tasks); tasks.clear(); + workload.idle(backlog); } // The runtime's own pools reject only once it has stopped them, // and then nothing is lost that it does not fail itself. if (!(executor instanceof ExecutorService && ((ExecutorService) executor).isShutdown())) { + workload.dropped(dropped.size()); errorHandler.accept(new WebRtcException("The executor rejected " + dropped.size() + " tasks, which were dropped", e)); } // In order, as they would have run. - for (Runnable task : dropped) { - if (task instanceof Droppable) { + for (Task task : dropped) { + if (task.dropped != null) { try { - ((Droppable) task).dropped.run(); + task.dropped.run(); } catch (Throwable error) { errorHandler.accept(error); @@ -169,10 +202,11 @@ private void drain() { runner = thread; boolean more = true; + long start = workload.now(); try { - for (int i = 0; i < BATCH_SIZE; i++) { - Runnable task; + for (int i = 0; i < BATCH_SIZE && (i == 0 || workload.now() - start < TURN_NANOS); i++) { + Task task; synchronized (tasks) { task = tasks.poll(); @@ -182,14 +216,22 @@ private void drain() { more = false; break; } + if (tasks.isEmpty()) { + workload.idle(backlog); + } } + workload.started(); + try { - task.run(); + task.task.run(); } catch (Throwable e) { errorHandler.accept(e); } + finally { + workload.finished(); + } } } finally { @@ -209,23 +251,32 @@ private void drain() { } } + private void count(Workload.Tally tally) { + synchronized (tasks) { + Task oldest = tasks.peek(); + if (oldest != null) { + tally.add(tasks.size(), oldest.queuedAt); + } + } + } - private static final class Droppable implements Runnable { + + + private static final class Task { final Runnable task; + /** Runs instead of the task should the executor reject it, or null. */ final Runnable dropped; + final long queuedAt; + - Droppable(Runnable task, Runnable dropped) { + Task(Runnable task, Runnable dropped, long queuedAt) { this.task = task; this.dropped = dropped; - } - - @Override - public void run() { - task.run(); + this.queuedAt = queuedAt; } } } diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java index 3524311d..8c2f49bd 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java @@ -30,13 +30,16 @@ import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; +import java.util.function.LongSupplier; import java.util.function.Supplier; import dev.onvoid.webrtc.PeerConnectionFactory; @@ -74,6 +77,11 @@ * one peer run one at a time and in order; those of different peers may run * at the same time, unless the {@link Builder#eventExecutor(Executor) event * executor} is serial, such as the Swing event dispatch thread. + *

+ * By default the runtime starts a thread for each peer and each subscription + * whose handlers have work. {@link Builder#eventThreads(int)} and {@link + * Builder#frameThreads(int)} limit them; what finds no thread then waits its + * turn, and {@link #loadStats()} tells how much waits. */ public final class WebRtc implements AutoCloseable { @@ -95,6 +103,21 @@ public final class WebRtc implements AutoCloseable { final Duration operationTimeout; + /** How much audio an {@code onAudio} subscription buffers by default. */ + final Duration audioBuffer; + + /** The tasks of the control executors. */ + final Workload controlWork; + + /** The tasks of the peers' event executors. */ + final Workload eventWork; + + /** The frames of the video subscriptions. */ + final Workload videoWork; + + /** The chunks of the audio subscriptions. */ + final Workload audioWork; + private final AudioDeviceModuleBase audioModule; private final AudioProcessing audioProcessing; @@ -126,11 +149,19 @@ private WebRtc(Builder builder) { this.audioMode = builder.audioMode; this.operationTimeout = builder.operationTimeout; this.closeTimeout = builder.closeTimeout; + this.audioBuffer = builder.audioBuffer; this.errorHandler = builder.errorHandler != null ? builder.errorHandler : WebRtc::uncaught; + this.controlWork = new Workload(builder.clock); + this.eventWork = new Workload(builder.clock); + this.videoWork = new Workload(builder.clock); + this.audioWork = new Workload(builder.clock); + + // Never limited: control tasks wait for other peers' control tasks, + // such as a camera switch for every sender to take the new track. this.controlPool = builder.controlPool != null ? builder.controlPool : Executors.newCachedThreadPool(threads("webrtc-client-control-" + runtime + "-")); - this.framePool = Executors.newCachedThreadPool(threads("webrtc-client-frames-" + runtime + "-")); + this.framePool = pool(builder.frameThreads, "webrtc-client-frames-" + runtime + "-"); this.timer = builder.timer != null ? builder.timer : Executors.newSingleThreadScheduledExecutor(threads("webrtc-client-timer-" + runtime + "-")); @@ -139,7 +170,7 @@ private WebRtc(Builder builder) { this.ownsEventExecutor = false; } else { - this.eventExecutor = Executors.newCachedThreadPool(threads("webrtc-client-events-" + runtime + "-")); + this.eventExecutor = pool(builder.eventThreads, "webrtc-client-events-" + runtime + "-"); this.ownsEventExecutor = true; } @@ -263,6 +294,20 @@ public AudioMode audioMode() { return audioMode; } + /** + * Tells how much work waits for the threads of this runtime and how much + * runs, at this moment. Cheap enough to be read every second, from any + * thread. + * + * @return the load of the runtime. + */ + public LoadStats loadStats() { + return new LoadStats(controlWork.snapshot(), eventWork.snapshot(), + videoWork.snapshot(), audioWork.snapshot(), + poolSize(controlPool), ownsEventExecutor ? poolSize(eventExecutor) : 0, + poolSize(framePool)); + } + /** * Closes every peer, then every track, then the factory, and stops the * threads of this API. @@ -520,6 +565,30 @@ private void shutdownPools() { } } + /** + * Creates a pool of at most the given number of threads, or an unlimited + * one for 0. A limited pool queues what finds no thread, in order, and + * never rejects it while it runs; since each serial executor has at most + * one turn queued at a time, the queue holds no more than one turn per + * peer or subscription. + */ + private static ExecutorService pool(int limit, String prefix) { + if (limit == 0) { + return Executors.newCachedThreadPool(threads(prefix)); + } + + ThreadPoolExecutor pool = new ThreadPoolExecutor(limit, limit, 60, TimeUnit.SECONDS, + new LinkedBlockingQueue<>(), threads(prefix)); + // Idle threads end, as those of an unlimited pool do. + pool.allowCoreThreadTimeOut(true); + + return pool; + } + + private static int poolSize(Executor executor) { + return executor instanceof ThreadPoolExecutor ? ((ThreadPoolExecutor) executor).getPoolSize() : 0; + } + private static ThreadFactory threads(String prefix) { AtomicInteger count = new AtomicInteger(); @@ -556,12 +625,20 @@ public static final class Builder { private Duration closeTimeout = Duration.ofSeconds(10); + private Duration audioBuffer = Duration.ofSeconds(5); + + private int eventThreads; + + private int frameThreads; + private Consumer factoryCustomizer; private ExecutorService controlPool; private ScheduledExecutorService timer; + private LongSupplier clock = System::nanoTime; + private Builder() { } @@ -628,6 +705,70 @@ public Builder eventExecutor(Executor executor) { return this; } + /** + * Limits the threads the runtime runs handlers on, those of peers, + * tracks and channels, future callbacks and {@code Signaling.send}. + * Unlimited by default, so that each peer whose events wait gets a + * thread. + *

+ * With a limit, the peers take turns: a peer that has events runs a + * few of them, for at most a few milliseconds, then lets the next + * peer in line run, so that slow handlers delay the others but never + * starve them. Nothing is rejected or dropped for the limit; events + * wait, and {@link LoadStats#events()} tells how many and for how + * long. A handler that waits for what another peer's handler does + * may then wait forever, once every thread is taken by such waits. + *

+ * Applies only to the runtime's own threads, not to an {@link + * #eventExecutor(Executor) executor of the application}. + * + * @param limit the most threads, 1 or more. + * + * @return this builder. + */ + public Builder eventThreads(int limit) { + this.eventThreads = atLeastOne(limit); + return this; + } + + /** + * Limits the threads that run {@code onFrame} and {@code onAudio} + * handlers, of all subscriptions together, local previews included. + * Unlimited by default, so that each subscription with media waiting + * gets a thread. + *

+ * With a limit, the subscriptions take turns as the peers do with + * {@link #eventThreads(int)}. A subscription waiting for a thread + * keeps what its {@link FrameDelivery} or audio buffer holds and drops + * the oldest beyond that, which {@link LoadStats#video()} and {@link + * LoadStats#audio()} count. + * + * @param limit the most threads, 1 or more. + * + * @return this builder. + */ + public Builder frameThreads(int limit) { + this.frameThreads = atLeastOne(limit); + return this; + } + + /** + * Sets how much decoded audio an {@link RemoteAudio#onAudio(Consumer) + * onAudio} subscription keeps for a handler that falls behind. The + * oldest audio beyond that is dropped. A short buffer keeps a live + * stream close to real time; a long one rides out a handler that + * stalls now and then, at the cost of memory, about 190 KiB per second + * of 48 kHz stereo. + * + * @param buffer the most audio that waits; 5 seconds by default. + * + * @return this builder. + */ + public Builder audioBuffer(Duration buffer) { + this.audioBuffer = positive(buffer); + return this; + } + /** * Sets what receives the exceptions handlers throw and the failures * nobody else hears of. By default they go to the uncaught exception @@ -697,15 +838,42 @@ Builder timer(ScheduledExecutorService timer) { return this; } + /** + * Replaces the clock of the load statistics and of the turns of the + * executors, so that tests decide how much time passes. + * + * @param nanos the time in nanoseconds, as {@link System#nanoTime()}. + */ + Builder clock(LongSupplier nanos) { + this.clock = requireNonNull(nanos, "nanos"); + return this; + } + /** * @return the runtime. * * @throws ConfigurationException if the factory cannot be created. + * @throws IllegalStateException if both {@link #eventThreads(int)} + * and {@link #eventExecutor(Executor)} + * were set. */ public WebRtc build() { + if (eventExecutor != null && eventThreads != 0) { + throw new IllegalStateException("eventThreads limits the runtime's own event threads, " + + "which an eventExecutor replaces; set only one of them"); + } + return new WebRtc(this); } + private static int atLeastOne(int limit) { + if (limit < 1) { + throw new IllegalArgumentException("The limit must be 1 or more, got " + limit); + } + + return limit; + } + private static Duration positive(Duration duration) { requireNonNull(duration, "duration"); diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/Workload.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/Workload.java new file mode 100644 index 00000000..673301d9 --- /dev/null +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/Workload.java @@ -0,0 +1,126 @@ +/* + * Copyright 2026 Alex Andres + * + * 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. + */ + +package dev.onvoid.webrtc.client; + +import java.time.Duration; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.LongAdder; +import java.util.function.LongSupplier; + +/** + * Counts the work of one kind that waits and runs, for {@link LoadStats}. + *

+ * Each queue that holds work registers itself while it is not empty, so that + * only the busy ones are looked at, and an idle one is not held on to. + */ +final class Workload { + + /** A queue of work. */ + interface Backlog { + + /** + * Adds what waits in the queue to the tally, under the queue's lock, + * so that its count and its oldest item fit together. + */ + void count(Tally tally); + + } + + private final LongSupplier clock; + + private final Set backlogs = ConcurrentHashMap.newKeySet(); + + private final AtomicInteger active = new AtomicInteger(); + + private final LongAdder dropped = new LongAdder(); + + + /** + * @param clock the time in nanoseconds, as {@link System#nanoTime()}. + */ + Workload(LongSupplier clock) { + this.clock = clock; + } + + long now() { + return clock.getAsLong(); + } + + /** Called, under the queue's lock, when the queue was empty and got an item. */ + void waiting(Backlog backlog) { + backlogs.add(backlog); + } + + /** Called, under the queue's lock, when the queue became empty. */ + void idle(Backlog backlog) { + backlogs.remove(backlog); + } + + void started() { + active.incrementAndGet(); + } + + void finished() { + active.decrementAndGet(); + } + + void dropped(long count) { + dropped.add(count); + } + + LoadStats.Work snapshot() { + Tally tally = new Tally(); + + for (Backlog backlog : backlogs) { + backlog.count(tally); + } + + Duration oldestWaiting = tally.queued > 0 + ? Duration.ofNanos(Math.max(0, now() - tally.oldest)) + : Duration.ZERO; + + return new LoadStats.Work(tally.queued, oldestWaiting, active.get(), dropped.sum()); + } + + + + /** What the busy queues hold, summed up. */ + static final class Tally { + + private int queued; + + private long oldest; + + + /** + * @param waiting the number of items a queue holds; one that went + * idle meanwhile adds none. + * @param queuedAt the clock time its oldest item was queued at. + */ + void add(int waiting, long queuedAt) { + if (waiting <= 0) { + return; + } + + // Compared as a difference, as nanoTime values must be. + oldest = queued == 0 || queuedAt - oldest < 0 ? queuedAt : oldest; + queued += waiting; + } + } +} diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/DispatcherTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/DispatcherTests.java new file mode 100644 index 00000000..83f66670 --- /dev/null +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/DispatcherTests.java @@ -0,0 +1,232 @@ +/* + * Copyright 2026 Alex Andres + * + * 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. + */ + +package dev.onvoid.webrtc.client; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.function.Consumer; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import dev.onvoid.webrtc.media.video.I420Buffer; +import dev.onvoid.webrtc.media.video.VideoFrame; +import dev.onvoid.webrtc.media.video.VideoFrameBuffer; + +import org.junit.jupiter.api.Test; + +/** + * Feeds audio and video dispatchers by hand. Their executors share a {@link + * ManualExecutor} as the subscriptions of a runtime share its frame pool, and + * a clock that moves only when the test moves it. + */ +class DispatcherTests { + + /** 10 ms of 48 kHz mono. */ + private static final int CHUNK_FRAMES = 480; + + private final List errors = new ArrayList<>(); + + private final ManualExecutor pool = new ManualExecutor(); + + private long now; + + private final Workload work = new Workload(() -> now); + + + @Test + void audioBufferDropsOldest() { + List heard = new ArrayList<>(); + AudioDispatcher audio = audio(Duration.ofMillis(30), chunk -> heard.add(number(chunk))); + + for (int i = 0; i < 5; i++) { + push(audio, i); + } + + LoadStats.Work load = work.snapshot(); + + assertEquals(3, load.queued()); + assertEquals(2, load.dropped()); + + pool.runAll(); + + assertEquals(Arrays.asList(2, 3, 4), heard); + assertEquals(0, work.snapshot().queued()); + assertTrue(errors.isEmpty(), () -> "Errors: " + errors); + } + + @Test + void shortBufferKeepsNewest() { + List heard = new ArrayList<>(); + AudioDispatcher audio = audio(Duration.ofMillis(1), chunk -> heard.add(number(chunk))); + + for (int i = 0; i < 3; i++) { + push(audio, i); + } + + pool.runAll(); + + assertEquals(Arrays.asList(2), heard); + assertEquals(2, work.snapshot().dropped()); + } + + @Test + void handlerCountedActive() { + List seen = new ArrayList<>(); + AudioDispatcher audio = audio(Duration.ofSeconds(1), chunk -> seen.add(work.snapshot())); + + push(audio, 0); + now += 4_000_000; + push(audio, 1); + now += 6_000_000; + + LoadStats.Work before = work.snapshot(); + + assertEquals(2, before.queued()); + assertEquals(Duration.ofMillis(10), before.oldestWaiting()); + + pool.runAll(); + + // The first chunk is in the handler, the second waits since 4 ms. + assertEquals(1, seen.get(0).active()); + assertEquals(1, seen.get(0).queued()); + assertEquals(Duration.ofMillis(6), seen.get(0).oldestWaiting()); + assertEquals(0, work.snapshot().active()); + } + + @Test + void subscriptionsTakeTurns() { + // A handler as slow as the audio it gets keeps a thread of a shared + // pool only for its turn, after which the other subscription runs. + List heard = new ArrayList<>(); + long slow = SerialExecutor.TURN_NANOS / 4; + AudioDispatcher busy = audio(Duration.ofSeconds(10), chunk -> { + heard.add("busy-" + number(chunk)); + now += slow; + }); + AudioDispatcher quiet = audio(Duration.ofSeconds(10), chunk -> heard.add("quiet-" + number(chunk))); + + for (int i = 0; i < 20; i++) { + push(busy, i); + } + push(quiet, 0); + + pool.runAll(); + + List expected = IntStream.range(0, 4).mapToObj(i -> "busy-" + i).collect(Collectors.toList()); + expected.add("quiet-0"); + + assertEquals(expected, heard.subList(0, 5)); + assertEquals(21, heard.size()); + } + + @Test + void framesReleased() { + List shown = new ArrayList<>(); + List buffers = new ArrayList<>(); + FrameDispatcher video = new FrameDispatcher(frame -> shown.add((int) frame.timestampNs), + FrameDelivery.queue(2), executor(), work, errors::add); + + for (int i = 0; i < 5; i++) { + buffers.add(new CountedBuffer()); + } + for (int i = 0; i < 4; i++) { + video.onVideoFrame(new VideoFrame(buffers.get(i), i)); + } + + assertEquals(2, work.snapshot().dropped()); + + pool.runAll(); + video.onVideoFrame(new VideoFrame(buffers.get(4), 4)); + video.close(); + pool.runAll(); + + // The first two were dropped, the next two shown, the last left over. + assertEquals(Arrays.asList(2, 3), shown); + buffers.forEach(buffer -> assertEquals(1, buffer.released)); + assertEquals(2, work.snapshot().dropped()); + assertEquals(0, work.snapshot().queued()); + } + + private AudioDispatcher audio(Duration buffer, Consumer handler) { + return new AudioDispatcher(handler, buffer, executor(), work, errors::add); + } + + /** + * As a runtime does, the executor of a subscription counts nothing + * itself; its dispatcher counts the media. + */ + private SerialExecutor executor() { + return new SerialExecutor(pool, new Workload(() -> now), errors::add, false, null); + } + + /** + * Pushes a chunk whose first sample tells which it is. + */ + private static void push(AudioDispatcher audio, int number) { + byte[] data = new byte[CHUNK_FRAMES * 2]; + data[0] = (byte) number; + + audio.onData(data, 16, 48000, 1, CHUNK_FRAMES); + } + + private static int number(AudioChunk chunk) { + return chunk.samples()[0]; + } + + + + private static final class CountedBuffer implements VideoFrameBuffer { + + int released; + + + @Override + public int getWidth() { + return 2; + } + + @Override + public int getHeight() { + return 2; + } + + @Override + public I420Buffer toI420() { + throw new UnsupportedOperationException(); + } + + @Override + public VideoFrameBuffer cropAndScale(int cropX, int cropY, int cropWidth, int cropHeight, + int scaleWidth, int scaleHeight) { + throw new UnsupportedOperationException(); + } + + @Override + public void retain() { + } + + @Override + public void release() { + released++; + } + } +} diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/LoadTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/LoadTests.java new file mode 100644 index 00000000..4b466e2c --- /dev/null +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/LoadTests.java @@ -0,0 +1,178 @@ +/* + * Copyright 2026 Alex Andres + * + * 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. + */ + +package dev.onvoid.webrtc.client; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +import org.junit.jupiter.api.Test; + +/** + * Runs runtimes with limited pools. A handler that holds the only thread is + * a latch the test opens, and the clock of the load statistics moves only + * when the test moves it. + */ +class LoadTests extends ClientTestBase { + + private final AtomicLong now = new AtomicLong(); + + + @Test + void limitsChecked() { + WebRtc.Builder builder = WebRtc.builder(); + + assertThrows(IllegalArgumentException.class, () -> builder.eventThreads(0)); + assertThrows(IllegalArgumentException.class, () -> builder.frameThreads(-1)); + assertThrows(IllegalArgumentException.class, () -> builder.audioBuffer(Duration.ZERO)); + + // A limit of the runtime's own threads means nothing for the + // application's executor. + IllegalStateException e = assertThrows(IllegalStateException.class, + () -> builder.eventThreads(2).eventExecutor(Runnable::run).build()); + assertTrue(e.getMessage().contains("eventExecutor"), e::getMessage); + } + + @Test + void eventsWaitTheirTurn() throws Exception { + replaceRuntime(WebRtc.builder().eventThreads(1)); + + CountDownLatch release = new CountDownLatch(1); + CountDownLatch blocking = new CountDownLatch(1); + CompletableFuture second = new CompletableFuture<>(); + SerialExecutor first = events("first"); + SerialExecutor other = events("other"); + + first.execute(() -> { + blocking.countDown(); + await(release); + }); + assertTrue(blocking.await(TIMEOUT_SECONDS, TimeUnit.SECONDS)); + + other.execute(() -> second.complete(Thread.currentThread().getName())); + now.addAndGet(TimeUnit.MILLISECONDS.toNanos(250)); + + LoadStats load = rtc.loadStats(); + + assertEquals(1, load.eventThreads()); + assertEquals(1, load.events().active()); + assertEquals(1, load.events().queued()); + assertEquals(Duration.ofMillis(250), load.events().oldestWaiting()); + assertFalse(second.isDone()); + + release.countDown(); + + // On the same, only thread, under the name of its own peer. + assertEquals("other", await(second)); + assertEquals(0, rtc.loadStats().events().queued()); + assertEquals(1, rtc.loadStats().eventThreads()); + } + + @Test + void closeFailsWaitingFutures() throws Exception { + // The only event thread is held while the runtime closes. The peer's + // future fails once the thread is free, after the events before it. + replaceRuntime(WebRtc.builder().eventThreads(1)); + + CountDownLatch release = new CountDownLatch(1); + CountDownLatch blocking = new CountDownLatch(1); + Peer peer = rtc.createPeer(PeerOptions.builder().negotiation(Negotiation.MANUAL).build()); + + events("blocker").execute(() -> { + blocking.countDown(); + await(release); + }); + assertTrue(blocking.await(TIMEOUT_SECONDS, TimeUnit.SECONDS)); + + // Closing the peers needs no event thread; telling of it does. + rtc.close(); + + assertEquals(PeerState.CLOSED, peer.state()); + assertFalse(peer.whenConnected().isDone()); + + release.countDown(); + + ExecutionException e = assertThrows(ExecutionException.class, () -> await(peer.whenConnected())); + assertInstanceOf(PeerClosedException.class, e.getCause()); + await(peer.whenClosed()); + await(rtc.closeAsync()); + } + + @Test + void limitedPoolsCarryCall() throws Exception { + replaceRuntime(WebRtc.builder().eventThreads(1).frameThreads(1).audioBuffer(Duration.ofMillis(200))); + + Peer[] peers = pair(); + CustomAudio voice = rtc.media().customAudio("voice"); + CompletableFuture heard = new CompletableFuture<>(); + CompletableFuture heardAgain = new CompletableFuture<>(); + + peers[1].onTrack(track -> { + track.asAudio().onAudio(heard::complete); + track.asAudio().onAudio(Duration.ofMillis(20), heardAgain::complete); + }); + peers[0].addTrack(voice); + + // One second of silence in one go. + voice.push(new byte[48000 * 2], PcmFormat.of(48000, 1)); + + assertTrue(await(heard).frames() > 0); + assertTrue(await(heardAgain).frames() > 0); + + LoadStats load = rtc.loadStats(); + + assertTrue(load.eventThreads() <= 1, load::toString); + assertTrue(load.frameThreads() <= 1, load::toString); + } + + /** + * Replaces the runtime of the base class with one built here, which the + * base class then closes and checks. + */ + private void replaceRuntime(WebRtc.Builder builder) { + rtc.close(); + rtc = builder.audio(AudioMode.HEADLESS) + .errorHandler(errors::add) + .clock(now::get) + .build(); + } + + /** + * @return an executor on the runtime's event pool, as a peer has. + */ + private SerialExecutor events(String name) { + return new SerialExecutor(rtc.eventExecutor, rtc.eventWork, rtc::reportError, true, name); + } + + private static void await(CountDownLatch latch) { + try { + latch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } +} diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/SerialExecutorTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/SerialExecutorTests.java index 298e67f0..ef8a890a 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/SerialExecutorTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/SerialExecutorTests.java @@ -22,6 +22,7 @@ import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -34,7 +35,8 @@ /** * Drives a {@link SerialExecutor} through a {@link ManualExecutor}, so that - * each test decides when the pool runs a task and when it rejects one. + * each test decides when the pool runs a task and when it rejects one, and + * through a clock that moves only when the test moves it. */ class SerialExecutorTests { @@ -42,7 +44,11 @@ class SerialExecutorTests { private final ManualExecutor pool = new ManualExecutor(); - private final SerialExecutor serial = new SerialExecutor(pool, errors::add, false, null); + private long now; + + private final Workload work = new Workload(() -> now); + + private final SerialExecutor serial = new SerialExecutor(pool, work, errors::add, false, null); @Test @@ -72,6 +78,8 @@ void shutdownRejectsQuietly() { // The pool's owner shut it down and fails what it loses itself. assertEquals(Arrays.asList("fallback"), ran); assertTrue(errors.isEmpty(), () -> "Errors: " + errors); + assertEquals(0, work.snapshot().dropped()); + assertEquals(0, work.snapshot().queued()); } @Test @@ -102,6 +110,8 @@ void rejectedTurnDropsRest() { assertEquals(Arrays.asList(64, 65, 67, 68, 69), dropped); assertEquals(1, errors.size(), () -> "Errors: " + errors); assertTrue(errors.get(0).getMessage().contains("rejected 6 tasks"), errors.get(0).getMessage()); + assertEquals(6, work.snapshot().dropped()); + assertEquals(0, work.snapshot().queued()); // Taken again once the pool accepts work. pool.rejecting(false); @@ -136,7 +146,7 @@ void throwingFallbackReported() { @Test void batchesShareThePool() { List ran = new ArrayList<>(); - SerialExecutor other = new SerialExecutor(pool, errors::add, false, null); + SerialExecutor other = new SerialExecutor(pool, work, errors::add, false, null); for (int i = 0; i < 100; i++) { int n = i; @@ -156,6 +166,61 @@ void batchesShareThePool() { assertEquals(0, pool.queued()); } + @Test + void turnEndsAfterTimeSlice() { + List ran = new ArrayList<>(); + SerialExecutor other = new SerialExecutor(pool, work, errors::add, false, null); + long slow = SerialExecutor.TURN_NANOS * 6 / 10; + + for (int i = 0; i < 3; i++) { + int n = i; + serial.execute(() -> { + ran.add("slow-" + n); + now += slow; + }); + } + other.execute(() -> ran.add("other")); + + pool.runAll(); + + // Two slow tasks fill a turn, far below the batch size. + assertEquals(Arrays.asList("slow-0", "slow-1", "other", "slow-2"), ran); + } + + @Test + void loadCounted() { + List seen = new ArrayList<>(); + SerialExecutor other = new SerialExecutor(pool, work, errors::add, false, null); + + serial.execute(() -> seen.add(work.snapshot())); + now += 5_000_000; + serial.execute(() -> { }); + other.execute(() -> { }); + now += 3_000_000; + + LoadStats.Work before = work.snapshot(); + + assertEquals(3, before.queued()); + assertEquals(Duration.ofMillis(8), before.oldestWaiting()); + assertEquals(0, before.active()); + + pool.runAll(); + + // While the first task ran: the other two waited, the older since 5 ms. + LoadStats.Work during = seen.get(0); + + assertEquals(2, during.queued()); + assertEquals(Duration.ofMillis(3), during.oldestWaiting()); + assertEquals(1, during.active()); + + LoadStats.Work after = work.snapshot(); + + assertEquals(0, after.queued()); + assertEquals(Duration.ZERO, after.oldestWaiting()); + assertEquals(0, after.active()); + assertEquals(0, after.dropped()); + } + @Test void throwingTaskReported() { List ran = new ArrayList<>();