diff --git a/docs/guide/client-api.md b/docs/guide/client-api.md index 26995757..b42af6b8 100644 --- a/docs/guide/client-api.md +++ b/docs/guide/client-api.md @@ -202,6 +202,8 @@ Every method of the client API may be called from any thread, including from ins 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. +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. + ### Handlers on the UI Thread Pass the UI toolkit's executor, and every handler may touch UI components directly: 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 74de5401..6856aac3 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/DataChannelImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/DataChannelImpl.java @@ -95,6 +95,8 @@ final class DataChannelImpl implements DataChannel { this.label = label; this.messageHandlers = new Handlers<>(peer.rtc::reportError); this.stateHandlers = new Handlers<>(peer.rtc::reportError); + + peer.rtc.track(opened, peer.events); } /** @@ -255,7 +257,8 @@ void closeFromPeer(WebRtcException error) { state = ChannelState.CLOSED; peer.channelClosed(this); - peer.events.execute(() -> deliverState(ChannelState.CLOSED, error)); + peer.events.execute(() -> deliverState(ChannelState.CLOSED, error), + () -> settleOpened(ChannelState.CLOSED, error)); } private CompletableFuture enqueue(byte[] data, boolean binary) { @@ -351,7 +354,7 @@ else if (newState == ChannelState.CLOSING) { failPending(new PeerClosedException("The data channel " + label + " is closing")); } - peer.events.execute(() -> deliverState(newState, null)); + peer.events.execute(() -> deliverState(newState, null), () -> settleOpened(newState, null)); } private void bufferChanged() { @@ -362,6 +365,20 @@ private void bufferChanged() { // On the event executor. + /** + * Completes the opened future for the state, which is all that is left of + * delivering it when the event executor rejects that. + */ + private void settleOpened(ChannelState newState, WebRtcException error) { + if (newState == ChannelState.OPEN) { + opened.complete(this); + } + else if (newState == ChannelState.CLOSED) { + opened.completeExceptionally(error != null ? error + : new PeerClosedException("The data channel " + label + " is closed")); + } + } + private void deliverState(ChannelState newState, WebRtcException error) { if (newState == deliveredState || deliveredState == ChannelState.CLOSED) { return; @@ -369,14 +386,9 @@ private void deliverState(ChannelState newState, WebRtcException error) { deliveredState = newState; stateHandlers.accept(newState); + settleOpened(newState, error); - if (newState == ChannelState.OPEN) { - opened.complete(this); - } - else if (newState == ChannelState.CLOSED) { - opened.completeExceptionally(error != null ? error - : new PeerClosedException("The data channel " + label + " is closed")); - + if (newState == ChannelState.CLOSED) { messageHandlers.clear(); stateHandlers.clear(); early.clear(); 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 3059c796..04de2e14 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/DevicesImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/DevicesImpl.java @@ -227,6 +227,7 @@ private static List desktopSources(DesktopCapturer capturer) { private CompletableFuture onMediaExecutor(Supplier action) { CompletableFuture result = new CompletableFuture<>(); + media.rtc.track(result, null); media.executor.execute(() -> { try { 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 16a5492e..23f22d4f 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalCameraImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalCameraImpl.java @@ -54,6 +54,7 @@ public CompletableFuture switchCamera(VideoDevice device) { requireNonNull(device, "device"); CompletableFuture switched = new CompletableFuture<>(); + media.rtc.track(switched, null); // Opening a camera can take a moment, so it happens off the caller's thread. media.executor.execute(() -> { 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 e946f8a1..0a0a7be2 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java @@ -178,6 +178,9 @@ final class PeerImpl implements Peer { this.trackHandlers = new Handlers<>(rtc::reportError); this.channelHandlers = new Handlers<>(rtc::reportError); this.errorHandlers = new Handlers<>(rtc::reportError); + + rtc.track(connected, events); + rtc.track(closedFuture, events); } /** @@ -224,13 +227,12 @@ void open() { opened.completeExceptionally(e); } + }, () -> { + // A runtime that stopped its threads dropped the task. + opened.completeExceptionally(new PeerClosedException("The runtime closed before " + name + + " was created")); }); - // A runtime that stops its threads drops the task if it has not run, - // and then nothing else would end the wait. - rtc.whenShutDown().thenRun(() -> opened.completeExceptionally( - new PeerClosedException("The runtime closed before " + name + " was created"))); - try { opened.get(); } @@ -556,7 +558,12 @@ boolean isClosing() { CompletableFuture toPublic(CompletableFuture internal) { ClientFuture result = new ClientFuture<>(); - internal.whenComplete((value, error) -> events.execute(() -> complete(result, value, error))); + rtc.track(result, events); + + // Completed in any case: should the event executor reject the task, on + // the thread that found it rejected. + internal.whenComplete((value, error) -> events.execute(() -> complete(result, value, error), + () -> complete(result, value, error))); return result; } @@ -977,7 +984,7 @@ private void connectionChanged(RTCPeerConnectionState nativeState) { PeerState newState = PeerState.of(nativeState); state = newState; - events.execute(() -> deliverState(newState)); + events.execute(() -> deliverState(newState), () -> settleConnected(newState)); } private void gatheringChanged(RTCIceGatheringState gatheringState) { @@ -1137,6 +1144,9 @@ private void teardown() { channelHandlers.clear(); stateHandlers.clear(); errorHandlers.clear(); + }, () -> { + connected.completeExceptionally(error); + closedFuture.complete(null); }); } @@ -1149,7 +1159,14 @@ private void deliverState(PeerState newState) { deliveredState = newState; stateHandlers.accept(newState); + settleConnected(newState); + } + /** + * Completes the connected future for the state, which is all that is left + * of delivering it when the event executor rejects that. + */ + private void settleConnected(PeerState newState) { if (newState == PeerState.CONNECTED) { connected.complete(this); } 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 d7d62499..4d21c8eb 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/SerialExecutor.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/SerialExecutor.java @@ -17,7 +17,10 @@ package dev.onvoid.webrtc.client; import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; import java.util.concurrent.RejectedExecutionException; import java.util.function.Consumer; @@ -28,6 +31,11 @@ *

* A task that throws does not stop the ones after it; its exception goes to * the error handler. + *

+ * 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 + * error handler as well, since it loses work the application expects. */ final class SerialExecutor implements Executor { @@ -55,8 +63,6 @@ protected int[] initialValue() { private boolean scheduled; - private boolean shutdown; - private volatile Thread runner; @@ -84,13 +90,17 @@ static boolean onEventThread() { return EVENT_DEPTH.get()[0] > 0; } + /** + * Runs the task, or the fallback should the executor reject it, which + * keeps a future the task would complete from staying open forever. + */ + void execute(Runnable task, Runnable dropped) { + execute(new Droppable(task, dropped)); + } + @Override public void execute(Runnable task) { synchronized (tasks) { - if (shutdown) { - return; - } - tasks.add(task); if (scheduled) { @@ -110,29 +120,37 @@ boolean inExecutor() { return runner == Thread.currentThread(); } - /** - * Drops the tasks not yet run and refuses new ones. A task running now - * finishes. - */ - void shutdown() { - synchronized (tasks) { - shutdown = true; - tasks.clear(); - } - } - private void schedule() { try { executor.execute(this::drain); } catch (RejectedExecutionException e) { - // The pool rejects only once the runtime has stopped it. A caller - // that waits on a dropped task ends its wait with the runtime's - // end, as PeerImpl.open does. + List dropped; + synchronized (tasks) { scheduled = false; + dropped = new ArrayList<>(tasks); tasks.clear(); } + + // 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())) { + 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) { + try { + ((Droppable) task).dropped.run(); + } + catch (Throwable error) { + errorHandler.accept(error); + } + } + } } } @@ -190,4 +208,24 @@ private void drain() { schedule(); } } + + + + private static final class Droppable implements Runnable { + + final Runnable task; + + final Runnable dropped; + + + Droppable(Runnable task, Runnable dropped) { + this.task = task; + this.dropped = dropped; + } + + @Override + public void run() { + task.run(); + } + } } 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 296d2d13..0b6fccd1 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java @@ -111,6 +111,12 @@ public final class WebRtc implements AutoCloseable { private final CompletableFuture closed = new CompletableFuture<>(); + /** The futures tracked that are not done, with what fails each. */ + private final Map, Runnable> pending = new ConcurrentHashMap<>(); + + /** Set once the control tasks have run, after which none ever runs. */ + private volatile boolean stopped; + private WebRtc(Builder builder) { int runtime = RUNTIME_COUNT.incrementAndGet(); @@ -342,11 +348,25 @@ boolean isClosing() { } /** - * @return a future that completes once the runtime has closed and stopped - * its threads, after which no task given to them runs. + * Fails the future, unless it is done by then, once the runtime has + * stopped the threads that could complete it. A task that would have + * completed it is dropped then, and the future would stay open forever. + * + * @param future the future handed out or waited on. + * @param events the event executor the future completes on, so that the + * failure comes after what is queued there; or null. */ - CompletableFuture whenShutDown() { - return closed; + void track(CompletableFuture future, SerialExecutor events) { + Runnable fail = () -> future.completeExceptionally( + new PeerClosedException("The runtime closed before this completed")); + Runnable orphan = events == null ? fail : () -> events.execute(fail, fail); + + pending.put(future, orphan); + future.whenComplete((value, error) -> pending.remove(future)); + + if (stopped) { + orphan.run(); + } } private void shutdown() { @@ -393,10 +413,32 @@ private void shutdown() { } finally { shutdownPools(); + failPending(); closed.complete(null); } } + /** + * Fails the futures that nothing completes any more, once the control + * tasks still queued have run. + */ + private void failPending() { + try { + controlPool.awaitTermination(closeTimeout.toMillis(), TimeUnit.MILLISECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + + // Set before the futures are listed: one tracked meanwhile either is + // listed or sees it set. + stopped = true; + + for (Runnable orphan : new ArrayList<>(pending.values())) { + orphan.run(); + } + } + /** * Closes what the supplier names, again and again, until it names nothing. * @@ -526,6 +568,11 @@ public Builder fieldTrials(Map fieldTrials) { * SwingUtilities::invokeLater} or {@code Platform::runLater}. The * order of each peer's events is kept whatever the executor. By * default the runtime runs them on threads of its own. + *

+ * An executor that rejects work loses the events it rejects, which + * goes to the error handler. The futures those events would have + * completed complete all the same, on the thread that found them + * rejected. * * @param executor runs handlers. * diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java index 6543025b..c947b384 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java @@ -31,7 +31,12 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import dev.onvoid.webrtc.RTCSessionDescription; @@ -495,6 +500,54 @@ void failedAttachClosesPeer() { assertEquals(1, rtc.peers().size()); } + @Test + void operationAfterRuntimeCloseFails() throws Exception { + // The runtime's threads are gone, so the task that would fail the + // operation never runs. Its future has to fail all the same. + Peer peer = rtc.createPeer(manualOptions()); + + rtc.close(); + + assertInstanceOf(PeerClosedException.class, cause(peer.negotiator().createOffer())); + assertInstanceOf(PeerClosedException.class, cause(peer.withNative(connection -> connection))); + } + + @Test + void rejectedEventsReported() throws Exception { + // An event executor that refuses work, as a bounded pool does when full. + AtomicBoolean refusing = new AtomicBoolean(); + ExecutorService pool = Executors.newSingleThreadExecutor(); + Executor executor = task -> { + if (refusing.get()) { + throw new RejectedExecutionException("Full"); + } + pool.execute(task); + }; + WebRtc runtime = WebRtc.builder() + .audio(AudioMode.HEADLESS) + .eventExecutor(executor) + .errorHandler(errors::add) + .build(); + + try { + Peer peer = runtime.createPeer(manualOptions()); + refusing.set(true); + + // Delivered all the same, though not on the event executor. + assertTrue(await(peer.negotiator().createOffer()).sdp.startsWith("v=0")); + assertTrue(errors.stream().anyMatch(error -> error.getCause() instanceof RejectedExecutionException), + () -> "Errors: " + errors); + + refusing.set(false); + } + finally { + runtime.close(); + pool.shutdown(); + } + + errors.clear(); + } + private static Throwable cause(CompletableFuture future) throws Exception { try { future.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);