diff --git a/docs/guide/media/media-files.md b/docs/guide/media/media-files.md index bc382eec..13e22f5b 100644 --- a/docs/guide/media/media-files.md +++ b/docs/guide/media/media-files.md @@ -175,7 +175,7 @@ source.setListener(new MediaPlayerListener() { ::: warning Every call arrives on the player's own thread, and that thread is the one decoding the media. A listener must return promptly, and must not wait on the player. -Closing the player, or the `MediaFileSource`, from a listener is fine: it takes effect at once, and the native player is released on another thread once the listener has returned. +Closing the player, or the `MediaFileSource`, from a listener is fine: it takes effect at once, and the native player is released on another thread once the listener has returned. The same holds for a sink of a track made from the player's sources, which the player calls on its own thread as well. ::: ## Feeding Your Own Media Sources diff --git a/webrtc-media/src/main/cpp/include/media/MediaPacer.h b/webrtc-media/src/main/cpp/include/media/MediaPacer.h index 5449b456..e3f44384 100644 --- a/webrtc-media/src/main/cpp/include/media/MediaPacer.h +++ b/webrtc-media/src/main/cpp/include/media/MediaPacer.h @@ -116,6 +116,10 @@ namespace ffmpeg // playback has got to. int64_t GetPositionUs() const; + // True on the pacing thread, which is the one that calls the sinks + // of the tracks fed by the sources. + bool IsOwnThread() const; + private: struct VideoItem { diff --git a/webrtc-media/src/main/cpp/include/media/MediaPlayer.h b/webrtc-media/src/main/cpp/include/media/MediaPlayer.h index 92c1fc9c..14d4c8b6 100644 --- a/webrtc-media/src/main/cpp/include/media/MediaPlayer.h +++ b/webrtc-media/src/main/cpp/include/media/MediaPlayer.h @@ -106,6 +106,13 @@ namespace ffmpeg int GetState() const; + // True on any thread of the player, which are the ones its + // observer and the sinks of the tracks it feeds are called on. + // Such a thread cannot destroy the player: that waits for the + // thread, and what is below the call on its stack still uses + // the player. + bool IsOwnThread() const; + private: void Run(); diff --git a/webrtc-media/src/main/cpp/src/JNI_MediaPlayer.cpp b/webrtc-media/src/main/cpp/src/JNI_MediaPlayer.cpp index 9cc9d78b..731204e3 100644 --- a/webrtc-media/src/main/cpp/src/JNI_MediaPlayer.cpp +++ b/webrtc-media/src/main/cpp/src/JNI_MediaPlayer.cpp @@ -24,6 +24,7 @@ #include #include +#include extern "C" { #include @@ -155,7 +156,22 @@ JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_media_player_MediaPlayer_dispose { // Closing twice is allowed, so a handle that is already zero is simply // nothing to do. - if (handle != 0) { - delete PlayerOf(handle); + if (handle == 0) { + return; + } + + ffmpeg::MediaPlayer * player = PlayerOf(handle); + + if (player->IsOwnThread()) { + // Called from something the player is calling, such as a sink of a + // track it feeds. Destroying it here would wait for this very thread, + // and the delivery that made the call still uses the player once the + // call returns. Another thread destroys it, and its wait ends as soon + // as this thread is out of the player. + std::thread([player] { delete player; }).detach(); + + return; } + + delete player; } diff --git a/webrtc-media/src/main/cpp/src/media/MediaPacer.cpp b/webrtc-media/src/main/cpp/src/media/MediaPacer.cpp index d1793690..1cb9f9b4 100644 --- a/webrtc-media/src/main/cpp/src/media/MediaPacer.cpp +++ b/webrtc-media/src/main/cpp/src/media/MediaPacer.cpp @@ -252,6 +252,13 @@ namespace ffmpeg return position_us_; } + bool MediaPacer::IsOwnThread() const + { + std::lock_guard lock(mutex_); + + return thread_.get_id() == std::this_thread::get_id(); + } + void MediaPacer::Run() { std::unique_lock lock(mutex_); diff --git a/webrtc-media/src/main/cpp/src/media/MediaPlayer.cpp b/webrtc-media/src/main/cpp/src/media/MediaPlayer.cpp index 75fb7a9b..744d1d45 100644 --- a/webrtc-media/src/main/cpp/src/media/MediaPlayer.cpp +++ b/webrtc-media/src/main/cpp/src/media/MediaPlayer.cpp @@ -244,6 +244,19 @@ namespace ffmpeg return state_; } + bool MediaPlayer::IsOwnThread() const + { + { + std::lock_guard lock(mutex_); + + if (thread_.get_id() == std::this_thread::get_id()) { + return true; + } + } + + return pacer_->IsOwnThread(); + } + void MediaPlayer::Run() { AVPacket * packet = av_packet_alloc(); diff --git a/webrtc-media/src/main/java/dev/onvoid/webrtc/media/player/MediaFileSource.java b/webrtc-media/src/main/java/dev/onvoid/webrtc/media/player/MediaFileSource.java index c7ec2df5..6c5b5a4d 100644 --- a/webrtc-media/src/main/java/dev/onvoid/webrtc/media/player/MediaFileSource.java +++ b/webrtc-media/src/main/java/dev/onvoid/webrtc/media/player/MediaFileSource.java @@ -19,6 +19,7 @@ import java.io.IOException; import java.nio.file.Path; import java.time.Duration; +import java.util.concurrent.atomic.AtomicBoolean; import dev.onvoid.webrtc.media.audio.CustomAudioSource; import dev.onvoid.webrtc.media.video.CustomVideoSource; @@ -66,7 +67,12 @@ public class MediaFileSource implements AutoCloseable { private final MediaPlayer player; - private boolean closed; + /** + * Not guarded by a lock: a close holding one while it waits for the + * player's threads would never end if one of them, in a listener or a + * sink, closed this source as well. + */ + private final AtomicBoolean closed = new AtomicBoolean(); /** @@ -275,13 +281,11 @@ public MediaPlayerState getState() { * a source that is already closed does nothing. */ @Override - public synchronized void close() { - if (closed) { + public void close() { + if (!closed.compareAndSet(false, true)) { return; } - closed = true; - player.close(); disposeSources(); diff --git a/webrtc-media/src/main/java/dev/onvoid/webrtc/media/player/MediaPlayer.java b/webrtc-media/src/main/java/dev/onvoid/webrtc/media/player/MediaPlayer.java index 4a8c7930..ce9d9954 100644 --- a/webrtc-media/src/main/java/dev/onvoid/webrtc/media/player/MediaPlayer.java +++ b/webrtc-media/src/main/java/dev/onvoid/webrtc/media/player/MediaPlayer.java @@ -252,11 +252,12 @@ public MediaPlayerState getState() { * and waits for the player's thread to finish. Closing a player that is * already closed does nothing. *

- * Closing from within a {@link MediaPlayerListener} call is allowed, but - * cannot wait: the player's thread is the one making that call. The player - * is closed at once, in that every later command does nothing and its - * state reads as closed, and it is released on another thread as soon as - * the listener call has returned. + * Closing from within a {@link MediaPlayerListener} call, or from a sink + * of a track made from one of the sources this player feeds, is allowed, + * but cannot wait: the player's thread is the one making that call. The + * player is closed at once, in that every later command does nothing and + * its state reads as closed, and it is released on another thread as soon + * as that call has returned. */ @Override public void close() { @@ -274,6 +275,9 @@ public void close() { return; } + // A sink of a track this player feeds is called on one of its threads + // too, but not through the listener. The native side tells those + // apart itself, and hands the release to another thread. if (callbackDepth.get()[0] == 0) { dispose(closing); diff --git a/webrtc-media/src/test/java/dev/onvoid/webrtc/media/player/MediaFileSourceTest.java b/webrtc-media/src/test/java/dev/onvoid/webrtc/media/player/MediaFileSourceTest.java index 4ac77d80..bf4eb72c 100644 --- a/webrtc-media/src/test/java/dev/onvoid/webrtc/media/player/MediaFileSourceTest.java +++ b/webrtc-media/src/test/java/dev/onvoid/webrtc/media/player/MediaFileSourceTest.java @@ -19,12 +19,14 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; 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.io.IOException; import java.net.URL; import java.nio.file.Path; import java.nio.file.Paths; +import java.time.Duration; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -162,6 +164,54 @@ public void onEndOfStream() { assertEquals(MediaPlayerState.CLOSED, source.getState()); } + @Test + void closesFromSinkWhileClosing() throws Exception { + MediaFileSource source = new MediaFileSource(asset()); + VideoTrack track = factory.createVideoTrack("video", source.getVideoSource()); + CountDownLatch delivering = new CountDownLatch(1); + AtomicInteger sinkCloses = new AtomicInteger(); + + VideoTrackSink sink = frame -> { + try { + if (delivering.getCount() == 0) { + return; + } + + delivering.countDown(); + + // Once the player reads as closed, the close on the test + // thread is waiting for this very thread to return. + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + + while (source.getState() != MediaPlayerState.CLOSED + && System.nanoTime() < deadline) { + Thread.yield(); + } + + // Must not wait for the close already under way. + source.close(); + sinkCloses.incrementAndGet(); + } + finally { + frame.release(); + } + }; + + track.addSink(sink); + source.play(); + + assertTrue(delivering.await(10, TimeUnit.SECONDS), "nothing delivered"); + + // Would never return if the two closes waited for each other. + assertTimeoutPreemptively(Duration.ofSeconds(10), source::close); + + assertEquals(1, sinkCloses.get()); + assertEquals(MediaPlayerState.CLOSED, source.getState()); + + track.removeSink(sink); + track.dispose(); + } + private static Path asset() throws Exception { URL url = MediaFileSourceTest.class.getResource(ASSET); diff --git a/webrtc-media/src/test/java/dev/onvoid/webrtc/media/player/MediaPlayerTest.java b/webrtc-media/src/test/java/dev/onvoid/webrtc/media/player/MediaPlayerTest.java index 642a2616..d02d5353 100644 --- a/webrtc-media/src/test/java/dev/onvoid/webrtc/media/player/MediaPlayerTest.java +++ b/webrtc-media/src/test/java/dev/onvoid/webrtc/media/player/MediaPlayerTest.java @@ -431,6 +431,65 @@ public void onStateChanged(MediaPlayerState state) { } } + @Test + void closesFromVideoSink() throws Exception { + try (Sources sources = new Sources()) { + VideoTrack track = factory.createVideoTrack("video", sources.video); + MediaPlayer player = new MediaPlayer(new MediaReader(asset()), + sources.video, sources.audio); + CountDownLatch closed = new CountDownLatch(1); + + // Called on the player's pacing thread, which is the one closing + // it has to wait for, and not through the listener. + VideoTrackSink sink = frame -> { + try { + player.close(); + closed.countDown(); + } + finally { + frame.release(); + } + }; + + track.addSink(sink); + player.play(); + + assertTrue(closed.await(10, TimeUnit.SECONDS), "not closed"); + assertEquals(MediaPlayerState.CLOSED, player.getState()); + + player.close(); + + track.removeSink(sink); + track.dispose(); + } + } + + @Test + void closesFromAudioSink() throws Exception { + try (Sources sources = new Sources()) { + AudioTrack track = factory.createAudioTrack("audio", sources.audio); + MediaPlayer player = new MediaPlayer(new MediaReader(asset()), + sources.video, sources.audio); + CountDownLatch closed = new CountDownLatch(1); + + AudioTrackSink sink = (data, bits, rate, ch, count) -> { + player.close(); + closed.countDown(); + }; + + track.addSink(sink); + player.play(); + + assertTrue(closed.await(10, TimeUnit.SECONDS), "not closed"); + assertEquals(MediaPlayerState.CLOSED, player.getState()); + + player.close(); + + track.removeSink(sink); + track.dispose(); + } + } + @Test void outlivesDisposedSources() throws Exception { CustomVideoSource videoSource = new CustomVideoSource();