From 00c182b897ff49b302df7ae33bb7ea707afb0fbc Mon Sep 17 00:00:00 2001 From: Alex Andres Date: Sun, 25 Oct 2026 15:17:30 +0100 Subject: [PATCH] fix: keep local track releases independent of handlers and cancelled futures Ended and device handlers run on an event executor of local media, after the release completed internally. Lifecycle futures are handed out as copies, so cancelling one no longer fails the runtime's close. --- docs/guide/client-api.md | 4 +- .../onvoid/webrtc/client/ClientFuture.java | 15 ++++ .../onvoid/webrtc/client/DataChannelImpl.java | 2 +- .../dev/onvoid/webrtc/client/Devices.java | 3 +- .../dev/onvoid/webrtc/client/DevicesImpl.java | 8 +- .../dev/onvoid/webrtc/client/HttpSession.java | 2 +- .../onvoid/webrtc/client/LocalCameraImpl.java | 4 +- .../onvoid/webrtc/client/LocalMediaImpl.java | 34 ++++++++- .../dev/onvoid/webrtc/client/LocalTrack.java | 11 ++- .../onvoid/webrtc/client/LocalTrackImpl.java | 48 ++++++++---- .../dev/onvoid/webrtc/client/PeerImpl.java | 6 +- .../java/dev/onvoid/webrtc/client/WebRtc.java | 28 ++++--- .../webrtc/client/RuntimeCloseTests.java | 75 +++++++++++++++++++ 13 files changed, 201 insertions(+), 39 deletions(-) diff --git a/docs/guide/client-api.md b/docs/guide/client-api.md index 45f4dd46..865dedbe 100644 --- a/docs/guide/client-api.md +++ b/docs/guide/client-api.md @@ -204,7 +204,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; a slow audio handler loses the oldest audio beyond its buffer. | -| `LocalTrack.onEnded` and `Devices.onChange` handlers | a background thread of the runtime | Briefly. | +| `LocalTrack.onEnded` and `Devices.onChange` handlers, callbacks of the futures of local tracks and devices | the event thread of the runtime's local media | Yes, but that delays the other events of local media. They must not wait for a future of this API either. A track is released before its `onEnded` handlers run, so they never hold up a release. | | WebRTC's own callbacks | WebRTC's threads | Not your code. The library only copies what they report and queues it. | The handlers of one peer never run at the same time, and they run in the order of the events. The handlers of different peers may run at the same time. With the default executor, the event thread is named after its peer, such as `webrtc-client-events-alice`, which shows up in thread dumps. @@ -329,6 +329,8 @@ peer.onChannel(channel -> { A future that is already done may be read anywhere, and any thread that is not an event thread, such as `main`, may wait. +Each call returns a future of its own, also of `closeAsync()`, `whenClosed()`, `whenConnected()` and `whenOpen()`. Cancelling one only stops waiting: the close goes on, and the futures of other callers, and of the runtime's own close, are not affected. + ::: warning On Java 8, only the futures the API returns refuse to be waited on. Futures derived from them with `thenApply` and the like do not, so waiting for one of those in a handler hangs that peer. From Java 9 on, the derived futures refuse too. diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/ClientFuture.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/ClientFuture.java index 95a4cce0..4ab921d3 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/ClientFuture.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/ClientFuture.java @@ -46,6 +46,21 @@ static ClientFuture completed(T value) { return future; } + /** + * Hands out a future of its own for a future shared by many callers: a + * caller that cancels or completes its copy changes nothing for the + * others, nor for what waits for the shared one. + * + * @param shared the future the copy completes with, on the same thread. + * + * @return a new future. + */ + static ClientFuture copyOf(CompletableFuture shared) { + ClientFuture copy = new ClientFuture<>(); + shared.whenComplete((value, error) -> PeerImpl.complete(copy, value, error)); + return copy; + } + @Override public T get() throws InterruptedException, ExecutionException { checkBlocking(); diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/DataChannelImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/DataChannelImpl.java index 84357572..d39cd5bf 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/DataChannelImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/DataChannelImpl.java @@ -157,7 +157,7 @@ public ChannelState state() { @Override public CompletableFuture whenOpen() { - return opened; + return ClientFuture.copyOf(opened); } @Override diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/Devices.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/Devices.java index 36806f88..214d5aaa 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/Devices.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/Devices.java @@ -85,7 +85,8 @@ public interface Devices { /** * Calls the handler after a device was connected or disconnected. * - * @param handler called on a background thread with the new snapshot. + * @param handler called on an event thread of the runtime with the new + * snapshot. * * @return the subscription that removes the handler. */ diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/DevicesImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/DevicesImpl.java index 1d876ac5..0a697bbb 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/DevicesImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/DevicesImpl.java @@ -164,7 +164,7 @@ void dispose() { private void changed() { media.executor.execute(() -> { refresh(); - handlers.accept(this); + media.events.execute(() -> handlers.accept(this)); }); } @@ -247,7 +247,9 @@ private static List desktopSources(DesktopCapturer capturer) { * Runs a switch, which uses the audio module, on the media executor. It * is registered like the creation of a track, so that a runtime that * begins to close either waits for it before it frees the module, or is - * seen closing here. + * seen closing here. The future handed out completes on the event + * executor, so that the application's callbacks never hold up the media + * executor. */ private CompletableFuture onMediaExecutor(Supplier action) { CompletableFuture result = new CompletableFuture<>(); @@ -279,6 +281,6 @@ private CompletableFuture onMediaExecutor(Supplier action) { result.completeExceptionally(new PeerClosedException("The runtime is closed")); }); - return result; + return media.toPublic(result); } } diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpSession.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpSession.java index b3a2aa02..c1c55695 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpSession.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpSession.java @@ -113,7 +113,7 @@ public CompletableFuture closeAsync() { .whenComplete((ignored, error) -> closed.complete(null)); } - return closed; + return ClientFuture.copyOf(closed); } /** diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalCameraImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalCameraImpl.java index e5586867..cbf185ce 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalCameraImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalCameraImpl.java @@ -75,7 +75,9 @@ public CompletableFuture switchCamera(VideoDevice device) { } }); - return switched; + // Completed on the event executor, so that the application's + // callbacks never hold up the media executor. + return media.toPublic(switched); } private void switchTo(VideoDevice device) { 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 dff148a5..7e5560f7 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalMediaImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalMediaImpl.java @@ -50,9 +50,16 @@ final class LocalMediaImpl implements LocalMedia { final WebRtc rtc; - /** Releases tracks and runs their ended handlers, off WebRTC's and the application's threads. */ + /** Releases tracks and switches devices, off WebRTC's and the application's threads. */ final SerialExecutor executor; + /** + * Runs the ended and device handlers and completes the futures handed + * out, as a peer's event executor does, so that no application code holds + * up a release. + */ + final SerialExecutor events; + private final AudioDeviceModuleBase audioModule; private final Set tracks = Collections.newSetFromMap(new ConcurrentHashMap<>()); @@ -76,6 +83,8 @@ final class LocalMediaImpl implements LocalMedia { this.rtc = rtc; this.audioModule = audioModule; this.executor = 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-media" : null); } @Override @@ -310,6 +319,29 @@ List> closeAll() { return closing; } + /** + * Hands a future out to the application: it completes on the event + * executor, after the handlers queued before, and cancelling it leaves the + * internal one alone. + */ + CompletableFuture toPublic(CompletableFuture internal) { + ClientFuture result = new ClientFuture<>(); + + internal.whenComplete((value, error) -> { + Runnable deliver = () -> PeerImpl.complete(result, value, error); + + // Completed in any case: should the event executor reject the + // task, on the thread that found it rejected. + events.execute(deliver, deliver); + }); + + // Tracked after, so that a track released before the runtime stopped + // its threads still reports that, rather than the runtime's close. + rtc.track(result, events); + + return result; + } + synchronized void dispose() { if (devices != null) { devices.dispose(); diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalTrack.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalTrack.java index 292e0582..14d40ea5 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalTrack.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalTrack.java @@ -48,10 +48,12 @@ public interface LocalTrack extends AutoCloseable { void setEnabled(boolean enabled); /** - * Calls the handler once the track ended. A handler added after that is - * called at once. + * Calls the handler once the track ended and its native track is + * released. A handler added after that is called at once. * - * @param handler called on a background thread. + * @param handler called on an event thread of the runtime, where waiting + * for a future of this API that is not done throws a + * {@link BlockingCallException}. * * @return the subscription that removes the handler. */ @@ -71,6 +73,9 @@ public interface LocalTrack extends AutoCloseable { * LocalMedia#videoTrack} or {@link LocalMedia#audioTrack}, may be closed; * or fails with a {@link WebRtcException} when the track or its source * could not be released, which then fails the runtime's close as well. + * It completes on an event thread, after the ended handlers ran. Each + * call returns a future of its own: cancelling it stops waiting, not the + * close. */ CompletableFuture closeAsync(); diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalTrackImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalTrackImpl.java index 30059781..2dbfeb5a 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalTrackImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalTrackImpl.java @@ -32,7 +32,9 @@ * Calls on the native track and source happen under this object's lock and * only while the track has not ended, so that closing never races with them. * Closing first makes every sender let go of the track, each on its peer's - * control executor, and only then releases the native track and source. + * control executor, and only then releases the native track and source. The + * ended handlers and the futures handed out follow on the media's event + * executor, so that neither holds up a release nor the runtime's close. */ abstract class LocalTrackImpl implements LocalTrack { @@ -50,7 +52,12 @@ abstract class LocalTrackImpl implements LocalTrack { /** Guarded by this. */ private Runnable releaseSource; - private final CompletableFuture closed = new CompletableFuture<>(); + /** + * Completed on the media executor once the native track is released, + * before the handlers hear of it. Never handed out, so that no caller can + * cancel what the runtime waits for. + */ + private final CompletableFuture closedInternal = new CompletableFuture<>(); private volatile boolean enabled = true; @@ -117,7 +124,7 @@ public Subscription onEnded(Runnable handler) { } } - media.executor.execute(handler); + media.events.execute(handler); return () -> { }; } @@ -129,7 +136,7 @@ public synchronized boolean isEnded() { @Override public CompletableFuture closeAsync() { - return closeInternal(); + return media.toPublic(closeInternal()); } @Override @@ -205,12 +212,18 @@ boolean isLive() { return !ended; } + /** + * Starts closing, once. + * + * @return a future that completes on the media executor once the native + * track is released, before the handlers hear of it. + */ CompletableFuture closeInternal() { List using; synchronized (this) { if (ended) { - return closed; + return closedInternal; } ended = true; @@ -227,7 +240,7 @@ CompletableFuture closeInternal() { CompletableFuture.allOf(detached.toArray(new CompletableFuture[0])) .whenComplete((ignored, error) -> media.executor.execute(this::release)); - return closed; + return closedInternal; } /** @@ -270,20 +283,25 @@ private void release() { endedHandlers.clear(); } - for (Runnable handler : handlers) { - try { - handler.run(); - } - catch (Throwable e) { - rtc.reportError(e); + // Queued before the release completes, so that the futures handed out + // complete after the handlers ran, and the runtime cannot stop the + // event executor in between. + media.events.execute(() -> { + for (Runnable handler : handlers) { + try { + handler.run(); + } + catch (Throwable e) { + rtc.reportError(e); + } } - } + }); if (failure == null) { - closed.complete(null); + closedInternal.complete(null); } else { - closed.completeExceptionally(failure); + closedInternal.completeExceptionally(failure); } } 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 3888a9eb..21f75d03 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java @@ -401,12 +401,12 @@ public Subscription onError(Consumer handler) { @Override public CompletableFuture whenConnected() { - return connected; + return ClientFuture.copyOf(connected); } @Override public CompletableFuture whenClosed() { - return closedFuture; + return ClientFuture.copyOf(closedFuture); } @Override @@ -522,7 +522,7 @@ public CompletableFuture withNative(Function action public CompletableFuture closeAsync() { closeInternal(); - return closedFuture; + return ClientFuture.copyOf(closedFuture); } @Override 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 cebca96b..1fb3e468 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java @@ -135,6 +135,7 @@ public final class WebRtc implements AutoCloseable { private final AtomicBoolean closing = new AtomicBoolean(); + /** Never handed out, so that no caller can cancel what the close waits for. */ private final CompletableFuture closed = new CompletableFuture<>(); /** The futures tracked that are not done, with what fails each. */ @@ -319,16 +320,17 @@ public LoadStats loadStats() { * that names the cause, as it does when disposing fails. * * @return a future that completes once all is closed, or fails when not - * everything could be released. + * everything could be released. Each call returns a future of its own: + * cancelling it stops waiting, not the close. */ public CompletableFuture closeAsync() { - if (closing.compareAndSet(false, true)) { - Thread thread = new Thread(this::shutdown, "webrtc-client-close"); - thread.setDaemon(true); - thread.start(); - } + startClose(); + + // Not on an event thread, so that one may wait for it. + CompletableFuture result = new CompletableFuture<>(); + closed.whenComplete((ignored, error) -> PeerImpl.complete(result, null, error)); - return closed; + return result; } /** @@ -343,10 +345,10 @@ public CompletableFuture closeAsync() { */ @Override public void close() { - CompletableFuture future = closeAsync(); + startClose(); try { - future.get(Timeouts.toMillis(closeTimeout), TimeUnit.MILLISECONDS); + closed.get(Timeouts.toMillis(closeTimeout), TimeUnit.MILLISECONDS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -359,6 +361,14 @@ public void close() { } } + private void startClose() { + if (closing.compareAndSet(false, true)) { + Thread thread = new Thread(this::shutdown, "webrtc-client-close"); + thread.setDaemon(true); + thread.start(); + } + } + /** * Registers a peer before the runtime is checked again: a runtime that * begins to close either finds the peer among those it closes and waits diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java index b6ec4932..df8bda8a 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java @@ -290,6 +290,68 @@ void closeWhileCreatingTrack() throws Exception { await(track.closeAsync()); } + @Test + void endedHandlerWaitsForClose() throws Exception { + // Handlers of a track wait for its own close and for that of another + // track. Both releases complete without the handlers, which run on an + // event thread, where the waits are refused rather than hanging. + LocalVideo first = rtc.media().customVideo("first"); + LocalVideo second = rtc.media().customVideo("second"); + CompletableFuture ownWait = new CompletableFuture<>(); + CompletableFuture otherWait = new CompletableFuture<>(); + + first.onEnded(() -> ownWait.complete(waitFor(first.closeAsync()))); + first.onEnded(() -> otherWait.complete(waitFor(second.closeAsync()))); + first.close(); + + assertInstanceOf(BlockingCallException.class, await(ownWait)); + assertInstanceOf(BlockingCallException.class, await(otherWait)); + + await(first.closeAsync()); + await(second.closeAsync()); + assertTrue(second.isEnded()); + } + + @Test + void cancelledTrackClose() throws Exception { + // The application stops waiting for a track's close before the release + // ran. The release goes on, and the runtime, which waits for it, does + // not take the cancellation for a failure. + ManualExecutor control = new ManualExecutor(); + WebRtc runtime = runtime(control); + LocalVideo track = runtime.media().customVideo("cancelled"); + + assertTrue(track.closeAsync().cancel(false)); + control.awaitQueued(TIMEOUT_SECONDS); + + CompletableFuture closed = runtime.closeAsync(); + control.runUntil(closed::isDone, TIMEOUT_SECONDS); + + await(closed); + await(track.closeAsync()); + } + + @Test + void cancelledRuntimeClose() throws Exception { + // Each caller gets a future of its own, whose cancellation neither + // stops the close nor reaches the others. + assertTrue(rtc.closeAsync().cancel(false)); + + rtc.close(); + await(rtc.closeAsync()); + } + + @Test + void cancelledPeerClose() throws Exception { + Peer peer = rtc.createPeer(manualOptions()); + + assertTrue(peer.whenClosed().cancel(false)); + assertTrue(peer.closeAsync().cancel(false)); + + await(peer.whenClosed()); + await(peer.closeAsync()); + } + @Test void releaseFailureFailsClose() throws Exception { // A step of the release fails. The steps after it run all the same, @@ -449,6 +511,19 @@ private WebRtc runtime(ManualExecutor control) { .build(); } + /** + * @return what waiting for the future threw, or null. + */ + private static Throwable waitFor(CompletableFuture future) { + try { + future.join(); + return null; + } + catch (Throwable e) { + return e; + } + } + private static AudioDevice speaker() { return new AudioDevice("Test speaker", "test-speaker") { }; }