fix: close a media player from a sink of a track it feeds

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

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
 			{

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();
 

webrtc-media/src/main/cpp/src/JNI_MediaPlayer.cpp

@@ -24,6 +24,7 @@
 
 #include <memory>
 #include <string>
+#include <thread>
 
 extern "C" {
 #include <libavutil/error.h>
@@ -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;
 }

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<std::mutex> lock(mutex_);
+
+		return thread_.get_id() == std::this_thread::get_id();
+	}
+
 	void MediaPacer::Run()
 	{
 		std::unique_lock<std::mutex> lock(mutex_);

webrtc-media/src/main/cpp/src/media/MediaPlayer.cpp

@@ -244,6 +244,19 @@ namespace ffmpeg
 		return state_;
 	}
 
+	bool MediaPlayer::IsOwnThread() const
+	{
+		{
+			std::lock_guard<std::mutex> lock(mutex_);
+
+			if (thread_.get_id() == std::this_thread::get_id()) {
+				return true;
+			}
+		}
+
+		return pacer_->IsOwnThread();
+	}
+
 	void MediaPlayer::Run()
 	{
 		AVPacket * packet = av_packet_alloc();

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();

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.
 	 * <p>
-	 * 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);
 

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);
 

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();