From 3d892a21c07463664a96f4ff73ebdd2712783046 Mon Sep 17 00:00:00 2001 From: Alex Andres Date: Thu, 22 Oct 2026 00:37:07 +0200 Subject: [PATCH] fix: close races in MediaReader.getInfo and WebRtc.createPeer MediaReader.getInfo now holds the same monitor as close, which may run on any thread; before, it could use a reader close had just freed (SIGSEGV). createPeer registers the peer before checking the runtime again, so a closing runtime either closes it before freeing the factory or is seen. open skips creating the connection once the peer is closed, and ends its wait when the runtime has stopped its threads and dropped the task. --- .../webrtc/media/player/MediaReader.java | 4 +- .../webrtc/media/player/MediaReaderTest.java | 30 +++++++++ .../dev/onvoid/webrtc/client/PeerImpl.java | 12 ++++ .../onvoid/webrtc/client/SerialExecutor.java | 3 + .../java/dev/onvoid/webrtc/client/WebRtc.java | 27 ++++++-- .../dev/onvoid/webrtc/client/PeerTests.java | 65 +++++++++++++++++++ 6 files changed, 133 insertions(+), 8 deletions(-) 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 d420eb4a5..49e9fcde4 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 4bc0477e6..398550fc0 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 947b343c7..e946f8a1f 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 6b36b4ced..d7d62499e 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 341237b70..296d2d134 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 2c46a4782..6543025bc 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();