diff --git a/webrtc-java-media/src/main/java/dev/onvoid/webrtc/media/player/MediaReader.java b/webrtc-java-media/src/main/java/dev/onvoid/webrtc/media/player/MediaReader.java index d420eb4a..49e9fcde 100644 --- a/webrtc-java-media/src/main/java/dev/onvoid/webrtc/media/player/MediaReader.java +++ b/webrtc-java-media/src/main/java/dev/onvoid/webrtc/media/player/MediaReader.java @@ -111,7 +111,9 @@ public MediaReader(String source, Duration timeout) throws IOException { * * @throws IllegalStateException if this reader has been closed. */ - public MediaInfo getInfo() { + public synchronized MediaInfo getInfo() { + // Synchronized like close, which may run on any thread: otherwise the + // handle read here could be freed while the native side still uses it. return info(handle); } diff --git a/webrtc-java-media/src/test/java/dev/onvoid/webrtc/media/player/MediaReaderTest.java b/webrtc-java-media/src/test/java/dev/onvoid/webrtc/media/player/MediaReaderTest.java index 4bc0477e..398550fc 100644 --- a/webrtc-java-media/src/test/java/dev/onvoid/webrtc/media/player/MediaReaderTest.java +++ b/webrtc-java-media/src/test/java/dev/onvoid/webrtc/media/player/MediaReaderTest.java @@ -25,6 +25,9 @@ import java.net.URL; import java.nio.file.Path; import java.nio.file.Paths; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.Test; @@ -130,6 +133,33 @@ void infoAfterCloseFails() throws Exception { assertThrows(IllegalStateException.class, reader::getInfo); } + @Test + void infoRacingClose() throws Exception { + // Close may run on any thread, so a getInfo on another one either gets + // the info or finds the reader closed, never a reader being freed. + for (int round = 0; round < 200; round++) { + MediaReader reader = new MediaReader(asset()); + CountDownLatch started = new CountDownLatch(1); + + CompletableFuture reading = CompletableFuture.runAsync(() -> { + started.countDown(); + + try { + while (true) { + assertNotNull(reader.getInfo()); + } + } + catch (IllegalStateException e) { + // Closed, which is how the loop ends. + } + }); + + started.await(); + reader.close(); + reading.get(10, TimeUnit.SECONDS); + } + } + private static Path asset() throws Exception { return asset(ASSET); } 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 947b343c..e946f8a1 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java @@ -188,6 +188,13 @@ void open() { CompletableFuture opened = new CompletableFuture<>(); control.execute(() -> { + if (closed) { + // The runtime closed the peer before this ran, and may already + // be freeing the factory. + opened.completeExceptionally(closedError()); + return; + } + try { connection = rtc.factory.createPeerConnection(options.toConfiguration(), new Observer()); @@ -219,6 +226,11 @@ void open() { } }); + // 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(); } 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 6b36b4ce..d7d62499 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/SerialExecutor.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/SerialExecutor.java @@ -126,6 +126,9 @@ private void schedule() { 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. synchronized (tasks) { scheduled = false; tasks.clear(); 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 341237b7..296d2d13 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java @@ -230,24 +230,29 @@ public Peer createPeer(PeerOptions options) { String name = options.name != null ? options.name : "peer-" + number; PeerImpl peer = new PeerImpl(this, options, name); + + // Registered before the runtime is checked again: a runtime that + // begins to close either finds the peer among those it closes and + // waits for it before freeing the factory, or is seen closing here. peers.add(peer); + if (closing.get()) { + peers.remove(peer); + throw new PeerClosedException("The runtime is closed"); + } + try { peer.open(); } catch (RuntimeException e) { // Closed before the error goes out, so that nothing of the failed // peer still runs when the runtime closes and frees WebRTC's threads. - peers.remove(peer); - peer.closeInternal().join(); + // Closing removes the peer. Once the runtime has stopped its threads + // the close cannot run any more, and there is nothing left to wait for. + CompletableFuture.anyOf(peer.closeInternal(), closed).join(); throw e; } - if (closing.get()) { - // The runtime began to close meanwhile and may have missed it. - peer.closeInternal(); - } - return peer; } @@ -336,6 +341,14 @@ boolean isClosing() { return closing.get(); } + /** + * @return a future that completes once the runtime has closed and stopped + * its threads, after which no task given to them runs. + */ + CompletableFuture whenShutDown() { + return closed; + } + private void shutdown() { try { // Peers and tracks created while the runtime closes are closed as 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 2c46a478..6543025b 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java @@ -20,6 +20,7 @@ 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.assertTimeoutPreemptively; import static org.junit.jupiter.api.Assertions.assertTrue; import java.time.Duration; @@ -372,6 +373,70 @@ void runtimeCloseClosesPeers() throws Exception { assertThrows(PeerClosedException.class, this::pair); } + @Test + void openAfterShutdownFails() { + // A createPeer that passed its check just before the runtime closed: + // the runtime's threads are gone and drop the task that would create + // the connection, so opening has to fail rather than wait forever. + PeerImpl peer = new PeerImpl(rtc, manualOptions(), "late"); + + rtc.close(); + + assertThrows(PeerClosedException.class, () -> assertTimeoutPreemptively( + Duration.ofSeconds(TIMEOUT_SECONDS), peer::open)); + } + + @Test + void openAfterCloseFails() throws Exception { + // The closing runtime found the peer before its connection was created. + // It may free the factory once the peer is closed, so the connection + // must not be created after that. + PeerImpl peer = new PeerImpl(rtc, manualOptions(), "closed-first"); + + await(peer.closeInternal()); + + assertThrows(PeerClosedException.class, peer::open); + } + + @Test + void createPeerRacingClose() throws Exception { + // A peer created while its runtime closes is either refused or closed + // with the runtime, and creating it never waits forever. + for (int round = 0; round < 50; round++) { + WebRtc runtime = WebRtc.builder() + .audio(AudioMode.HEADLESS) + .errorHandler(errors::add) + .build(); + CountDownLatch started = new CountDownLatch(1); + + CompletableFuture> created = CompletableFuture.supplyAsync(() -> { + List peers = new ArrayList<>(); + + try { + while (true) { + peers.add(runtime.createPeer(manualOptions())); + started.countDown(); + } + } + catch (PeerClosedException e) { + return peers; + } + }); + + // Closed while peers are being created, not before the first. + started.await(); + runtime.closeAsync().get(TIMEOUT_SECONDS, TimeUnit.SECONDS); + + for (Peer peer : created.get(TIMEOUT_SECONDS, TimeUnit.SECONDS)) { + assertEquals(PeerState.CLOSED, peer.state()); + } + } + } + + private static PeerOptions manualOptions() { + return PeerOptions.builder().negotiation(Negotiation.MANUAL).build(); + } + @Test void statsAfterConnecting() throws Exception { Peer[] peers = pair();