Merge pull request #1754 from colinkho:loader-plumbing

PiperOrigin-RevId: 684781854

libraries/exoplayer/src/main/java/androidx/media3/exoplayer/source/ProgressiveMediaPeriod.java

@@ -54,6 +54,7 @@ import androidx.media3.exoplayer.upstream.LoadErrorHandlingPolicy.LoadErrorInfo;
 import androidx.media3.exoplayer.upstream.Loader;
 import androidx.media3.exoplayer.upstream.Loader.LoadErrorAction;
 import androidx.media3.exoplayer.upstream.Loader.Loadable;
+import androidx.media3.exoplayer.util.ReleasableExecutor;
 import androidx.media3.extractor.DiscardingTrackOutput;
 import androidx.media3.extractor.Extractor;
 import androidx.media3.extractor.ExtractorOutput;
@@ -172,6 +173,8 @@ import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
    * @param continueLoadingCheckIntervalBytes The number of bytes that should be loaded between each
    *     invocation of {@link Callback#onContinueLoadingRequested(SequenceableLoader)}.
    * @param singleSampleDurationUs The duration of media with a single sample in microseconds.
+   * @param downloadExecutor An optional externally provided {@link ReleasableExecutor} for loading
+   *     and extracting media.
    */
   // maybeFinishPrepare is not posted to the handler until initialization completes.
   @SuppressWarnings({"nullness:argument", "nullness:methodref.receiver.bound"})
@@ -187,7 +190,8 @@ import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
       Allocator allocator,
       @Nullable String customCacheKey,
       int continueLoadingCheckIntervalBytes,
-      long singleSampleDurationUs) {
+      long singleSampleDurationUs,
+      @Nullable ReleasableExecutor downloadExecutor) {
     this.uri = uri;
     this.dataSource = dataSource;
     this.drmSessionManager = drmSessionManager;
@@ -198,7 +202,10 @@ import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
     this.allocator = allocator;
     this.customCacheKey = customCacheKey;
     this.continueLoadingCheckIntervalBytes = continueLoadingCheckIntervalBytes;
-    loader = new Loader("ProgressiveMediaPeriod");
+    loader =
+        downloadExecutor != null
+            ? new Loader(downloadExecutor)
+            : new Loader("ProgressiveMediaPeriod");
     this.progressiveMediaExtractor = progressiveMediaExtractor;
     this.singleSampleDurationUs = singleSampleDurationUs;
     loadCondition = new ConditionVariable();

libraries/exoplayer/src/main/java/androidx/media3/exoplayer/source/ProgressiveMediaSource.java

@@ -24,6 +24,7 @@ import androidx.annotation.Nullable;
 import androidx.media3.common.C;
 import androidx.media3.common.MediaItem;
 import androidx.media3.common.Timeline;
+import androidx.media3.common.util.Consumer;
 import androidx.media3.common.util.UnstableApi;
 import androidx.media3.common.util.Util;
 import androidx.media3.datasource.DataSource;
@@ -34,10 +35,13 @@ import androidx.media3.exoplayer.drm.DrmSessionManagerProvider;
 import androidx.media3.exoplayer.upstream.Allocator;
 import androidx.media3.exoplayer.upstream.DefaultLoadErrorHandlingPolicy;
 import androidx.media3.exoplayer.upstream.LoadErrorHandlingPolicy;
+import androidx.media3.exoplayer.util.ReleasableExecutor;
 import androidx.media3.extractor.DefaultExtractorsFactory;
 import androidx.media3.extractor.Extractor;
 import androidx.media3.extractor.ExtractorsFactory;
+import com.google.common.base.Supplier;
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
+import java.util.concurrent.Executor;
 
 /**
  * Provides one period that loads data from a {@link Uri} and extracted using an {@link Extractor}.
@@ -64,6 +68,7 @@ public final class ProgressiveMediaSource extends BaseMediaSource
     private DrmSessionManagerProvider drmSessionManagerProvider;
     private LoadErrorHandlingPolicy loadErrorHandlingPolicy;
     private int continueLoadingCheckIntervalBytes;
+    @Nullable private Supplier<ReleasableExecutor> downloadExecutorSupplier;
 
     /**
      * Creates a new factory for {@link ProgressiveMediaSource}s.
@@ -197,6 +202,23 @@ public final class ProgressiveMediaSource extends BaseMediaSource
       return this;
     }
 
+    /**
+     * Sets a supplier for an {@link Executor} that is used for loading the media.
+     *
+     * @param downloadExecutor A {@link Supplier} that provides an externally managed {@link
+     *     Executor} for downloading and extraction.
+     * @param downloadExecutorReleaser A callback triggered once a load task is finished and a
+     *     supplied executor is no longer required.
+     * @return This factory, for convenience.
+     */
+    @CanIgnoreReturnValue
+    public <T extends Executor> Factory setDownloadExecutor(
+        Supplier<T> downloadExecutor, Consumer<T> downloadExecutorReleaser) {
+      this.downloadExecutorSupplier =
+          () -> ReleasableExecutor.from(downloadExecutor.get(), downloadExecutorReleaser);
+      return this;
+    }
+
     /**
      * Returns a new {@link ProgressiveMediaSource} using the current parameters.
      *
@@ -213,7 +235,8 @@ public final class ProgressiveMediaSource extends BaseMediaSource
           progressiveMediaExtractorFactory,
           drmSessionManagerProvider.get(mediaItem),
           loadErrorHandlingPolicy,
-          continueLoadingCheckIntervalBytes);
+          continueLoadingCheckIntervalBytes,
+          downloadExecutorSupplier);
     }
 
     @Override
@@ -233,6 +256,9 @@ public final class ProgressiveMediaSource extends BaseMediaSource
   private final DrmSessionManager drmSessionManager;
   private final LoadErrorHandlingPolicy loadableLoadErrorHandlingPolicy;
   private final int continueLoadingCheckIntervalBytes;
+
+  @Nullable private final Supplier<ReleasableExecutor> downloadExecutorSupplier;
+
   private boolean timelineIsPlaceholder;
   private long timelineDurationUs;
   private boolean timelineIsSeekable;
@@ -248,7 +274,8 @@ public final class ProgressiveMediaSource extends BaseMediaSource
       ProgressiveMediaExtractor.Factory progressiveMediaExtractorFactory,
       DrmSessionManager drmSessionManager,
       LoadErrorHandlingPolicy loadableLoadErrorHandlingPolicy,
-      int continueLoadingCheckIntervalBytes) {
+      int continueLoadingCheckIntervalBytes,
+      @Nullable Supplier<ReleasableExecutor> downloadExecutorSupplier) {
     this.mediaItem = mediaItem;
     this.dataSourceFactory = dataSourceFactory;
     this.progressiveMediaExtractorFactory = progressiveMediaExtractorFactory;
@@ -257,6 +284,7 @@ public final class ProgressiveMediaSource extends BaseMediaSource
     this.continueLoadingCheckIntervalBytes = continueLoadingCheckIntervalBytes;
     this.timelineIsPlaceholder = true;
     this.timelineDurationUs = C.TIME_UNSET;
+    this.downloadExecutorSupplier = downloadExecutorSupplier;
   }
 
   @Override
@@ -312,7 +340,8 @@ public final class ProgressiveMediaSource extends BaseMediaSource
         allocator,
         localConfiguration.customCacheKey,
         continueLoadingCheckIntervalBytes,
-        Util.msToUs(localConfiguration.imageDurationMs));
+        Util.msToUs(localConfiguration.imageDurationMs),
+        downloadExecutorSupplier != null ? downloadExecutorSupplier.get() : null);
   }
 
   @Override

libraries/exoplayer/src/main/java/androidx/media3/exoplayer/source/chunk/ChunkSampleStream.java

@@ -45,6 +45,7 @@ import androidx.media3.exoplayer.upstream.LoadErrorHandlingPolicy;
 import androidx.media3.exoplayer.upstream.LoadErrorHandlingPolicy.LoadErrorInfo;
 import androidx.media3.exoplayer.upstream.Loader;
 import androidx.media3.exoplayer.upstream.Loader.LoadErrorAction;
+import androidx.media3.exoplayer.util.ReleasableExecutor;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -119,6 +120,8 @@ public class ChunkSampleStream<T extends ChunkSource>
    *     events.
    * @param canReportInitialDiscontinuity Whether the stream can report an initial discontinuity if
    *     the first chunk can't start at the beginning and needs to preroll data.
+   * @param downloadExecutor An optional externally provided {@link ReleasableExecutor} for loading
+   *     and extracting media.
    */
   public ChunkSampleStream(
       @C.TrackType int primaryTrackType,
@@ -132,7 +135,8 @@ public class ChunkSampleStream<T extends ChunkSource>
       DrmSessionEventListener.EventDispatcher drmEventDispatcher,
       LoadErrorHandlingPolicy loadErrorHandlingPolicy,
       MediaSourceEventListener.EventDispatcher mediaSourceEventDispatcher,
-      boolean canReportInitialDiscontinuity) {
+      boolean canReportInitialDiscontinuity,
+      @Nullable ReleasableExecutor downloadExecutor) {
     this.primaryTrackType = primaryTrackType;
     this.embeddedTrackTypes = embeddedTrackTypes == null ? new int[0] : embeddedTrackTypes;
     this.embeddedTrackFormats = embeddedTrackFormats == null ? new Format[0] : embeddedTrackFormats;
@@ -141,7 +145,8 @@ public class ChunkSampleStream<T extends ChunkSource>
     this.mediaSourceEventDispatcher = mediaSourceEventDispatcher;
     this.loadErrorHandlingPolicy = loadErrorHandlingPolicy;
     this.canReportInitialDiscontinuity = canReportInitialDiscontinuity;
-    loader = new Loader("ChunkSampleStream");
+    loader =
+        downloadExecutor != null ? new Loader(downloadExecutor) : new Loader("ChunkSampleStream");
     nextChunkHolder = new ChunkHolder();
     mediaChunks = new ArrayList<>();
     readOnlyMediaChunks = Collections.unmodifiableList(mediaChunks);

libraries/exoplayer/src/main/java/androidx/media3/exoplayer/upstream/Loader.java

@@ -31,12 +31,12 @@ import androidx.media3.common.util.Log;
 import androidx.media3.common.util.TraceUtil;
 import androidx.media3.common.util.UnstableApi;
 import androidx.media3.common.util.Util;
+import androidx.media3.exoplayer.util.ReleasableExecutor;
 import java.io.IOException;
 import java.lang.annotation.Documented;
 import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
-import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -218,8 +218,7 @@ public final class Loader implements LoaderErrorThrower {
     }
   }
 
-  private final Executor downloadExecutor;
+  private final ReleasableExecutor downloadExecutor;
-  private final Runnable downloadExecutorReleaser;
 
   @Nullable private LoadTask<? extends Loadable> currentTask;
   @Nullable private IOException fatalError;
@@ -231,20 +230,21 @@ public final class Loader implements LoaderErrorThrower {
    *     component using the loader.
    */
   public Loader(String threadNameSuffix) {
-    ExecutorService executorService =
+    this(
-        Util.newSingleThreadExecutor(THREAD_NAME_PREFIX + threadNameSuffix);
+        /* downloadExecutor= */ ReleasableExecutor.from(
-    this.downloadExecutor = executorService;
+            Util.newSingleThreadExecutor(THREAD_NAME_PREFIX + threadNameSuffix),
-    this.downloadExecutorReleaser = executorService::shutdown;
+            ExecutorService::shutdown));
   }
 
   /**
    * Constructs an instance.
    *
-   * @param downloadExecutor An {@link Executor} for supplying the loader's thread.
+   * @param downloadExecutor A {@link ReleasableExecutor} to run the load task. The {@link
+   *     ReleasableExecutor} will be {@linkplain ReleasableExecutor#release() released} once the
+   *     loader no longer requires it for new load tasks.
    */
-  public Loader(Executor downloadExecutor) {
+  public Loader(ReleasableExecutor downloadExecutor) {
     this.downloadExecutor = downloadExecutor;
-    this.downloadExecutorReleaser = () -> {};
   }
 
   /**
@@ -328,7 +328,7 @@ public final class Loader implements LoaderErrorThrower {
     if (callback != null) {
       downloadExecutor.execute(new ReleaseTask(callback));
     }
-    downloadExecutorReleaser.run();
+    downloadExecutor.release();
   }
 
   // LoaderErrorThrower implementation.

libraries/exoplayer/src/main/java/androidx/media3/exoplayer/util/ReleasableExecutor.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2024 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package androidx.media3.exoplayer.util;
+
+import androidx.media3.common.util.Consumer;
+import androidx.media3.common.util.UnstableApi;
+import java.util.concurrent.Executor;
+
+/**
+ * An {@link Executor} with a dedicated {@link #release} method to signal when it is not longer
+ * needed.
+ */
+@UnstableApi
+public interface ReleasableExecutor extends Executor {
+
+  /**
+   * Releases the {@link Executor}, indicating that the caller no longer requires it for executing
+   * new commands.
+   *
+   * <p>When calling this method, there may still be pending commands that are currently executed.
+   */
+  void release();
+
+  /**
+   * Creates a {@link ReleasableExecutor} from an {@link Executor} and a release callback.
+   *
+   * @param executor The {@link Executor}
+   * @param releaseCallback The release callback, accepting the {@code executor} as an argument.
+   * @return The releasable executor.
+   * @param <T> The type of {@link Executor}.
+   */
+  static <T extends Executor> ReleasableExecutor from(T executor, Consumer<T> releaseCallback) {
+    return new ReleasableExecutor() {
+      @Override
+      public void execute(Runnable command) {
+        executor.execute(command);
+      }
+
+      @Override
+      public void release() {
+        releaseCallback.accept(executor);
+      }
+    };
+  }
+}

libraries/exoplayer/src/test/java/androidx/media3/exoplayer/source/ProgressiveMediaPeriodTest.java

@@ -19,7 +19,9 @@ import static androidx.media3.test.utils.robolectric.RobolectricUtil.runMainLoop
 import static com.google.common.truth.Truth.assertThat;
 
 import android.net.Uri;
+import androidx.annotation.Nullable;
 import androidx.media3.common.C;
+import androidx.media3.common.util.Consumer;
 import androidx.media3.datasource.AssetDataSource;
 import androidx.media3.exoplayer.LoadingInfo;
 import androidx.media3.exoplayer.analytics.PlayerId;
@@ -28,12 +30,16 @@ import androidx.media3.exoplayer.drm.DrmSessionManager;
 import androidx.media3.exoplayer.source.MediaSource.MediaPeriodId;
 import androidx.media3.exoplayer.upstream.DefaultAllocator;
 import androidx.media3.exoplayer.upstream.DefaultLoadErrorHandlingPolicy;
+import androidx.media3.exoplayer.util.ReleasableExecutor;
 import androidx.media3.extractor.Extractor;
 import androidx.media3.extractor.ExtractorsFactory;
 import androidx.media3.extractor.mp4.Mp4Extractor;
 import androidx.media3.extractor.png.PngExtractor;
+import androidx.media3.extractor.text.SubtitleParser;
 import androidx.test.core.app.ApplicationProvider;
 import androidx.test.ext.junit.runners.AndroidJUnit4;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.junit.Test;
@@ -67,8 +73,35 @@ public final class ProgressiveMediaPeriodTest {
     testExtractorsUpdatesSourceInfoBeforeOnPreparedCallback(extractor, C.TIME_UNSET);
   }
 
+  @Test
+  public void supplyingCustomDownloadExecutor_downloadsOnCustomThread() throws TimeoutException {
+    AtomicBoolean hasThreadRun = new AtomicBoolean(false);
+    AtomicBoolean hasReleaseCallbackRun = new AtomicBoolean(false);
+    Executor executor =
+        Executors.newSingleThreadExecutor(r -> new ExecutionTrackingThread(r, hasThreadRun));
+
+    testExtractorsUpdatesSourceInfoBeforeOnPreparedCallback(
+        new BundledExtractorsAdapter(Mp4Extractor.newFactory(SubtitleParser.Factory.UNSUPPORTED)),
+        C.TIME_UNSET,
+        executor,
+        e -> hasReleaseCallbackRun.set(true));
+
+    assertThat(hasThreadRun.get()).isTrue();
+    assertThat(hasReleaseCallbackRun.get()).isTrue();
+  }
+
   private static void testExtractorsUpdatesSourceInfoBeforeOnPreparedCallback(
       ProgressiveMediaExtractor extractor, long imageDurationUs) throws TimeoutException {
+    testExtractorsUpdatesSourceInfoBeforeOnPreparedCallback(
+        extractor, imageDurationUs, /* executor= */ null, /* executorReleased= */ null);
+  }
+
+  private static void testExtractorsUpdatesSourceInfoBeforeOnPreparedCallback(
+      ProgressiveMediaExtractor extractor,
+      long imageDurationUs,
+      @Nullable Executor executor,
+      @Nullable Consumer<Executor> executorReleased)
+      throws TimeoutException {
     AtomicBoolean sourceInfoRefreshCalled = new AtomicBoolean(false);
     ProgressiveMediaPeriod.Listener sourceInfoRefreshListener =
         (durationUs, isSeekable, isLive) -> sourceInfoRefreshCalled.set(true);
@@ -88,7 +121,8 @@ public final class ProgressiveMediaPeriodTest {
             new DefaultAllocator(/* trimOnReset= */ true, C.DEFAULT_BUFFER_SEGMENT_SIZE),
             /* customCacheKey= */ null,
             ProgressiveMediaSource.DEFAULT_LOADING_CHECK_INTERVAL_BYTES,
-            imageDurationUs);
+            imageDurationUs,
+            executor != null ? ReleasableExecutor.from(executor, executorReleased) : null);
 
     AtomicBoolean prepareCallbackCalled = new AtomicBoolean(false);
     AtomicBoolean sourceInfoRefreshCalledBeforeOnPrepared = new AtomicBoolean(false);
@@ -111,4 +145,19 @@ public final class ProgressiveMediaPeriodTest {
 
     assertThat(sourceInfoRefreshCalledBeforeOnPrepared.get()).isTrue();
   }
+
+  private static final class ExecutionTrackingThread extends Thread {
+    private final AtomicBoolean hasRun;
+
+    public ExecutionTrackingThread(Runnable runnable, AtomicBoolean hasRun) {
+      super(runnable, "TestExecutionTrackingThread");
+      this.hasRun = hasRun;
+    }
+
+    @Override
+    public void run() {
+      hasRun.set(true);
+      super.run();
+    }
+  }
 }

libraries/exoplayer_dash/src/main/java/androidx/media3/exoplayer/dash/DashMediaPeriod.java

@@ -837,7 +837,8 @@ import java.util.regex.Pattern;
             drmEventDispatcher,
             loadErrorHandlingPolicy,
             mediaSourceEventDispatcher,
-            canReportInitialDiscontinuity);
+            canReportInitialDiscontinuity,
+            /* downloadExecutor= */ null);
     synchronized (this) {
       // The map is also accessed on the loading thread so synchronize access.
       trackEmsgHandlerBySampleStream.put(stream, trackPlayerEmsgHandler);

libraries/exoplayer_smoothstreaming/src/main/java/androidx/media3/exoplayer/smoothstreaming/SsMediaPeriod.java

@@ -265,7 +265,8 @@ import java.util.List;
         drmEventDispatcher,
         loadErrorHandlingPolicy,
         mediaSourceEventDispatcher,
-        /* canReportInitialDiscontinuity= */ false);
+        /* canReportInitialDiscontinuity= */ false,
+        /* downloadExecutor= */ null);
   }
 
   private static TrackGroupArray buildTrackGroups(

libraries/test_utils/src/main/java/androidx/media3/test/utils/FakeAdaptiveMediaPeriod.java

@@ -188,7 +188,8 @@ public class FakeAdaptiveMediaPeriod
                 new DrmSessionEventListener.EventDispatcher(),
                 new DefaultLoadErrorHandlingPolicy(/* minimumLoadableRetryCount= */ 3),
                 mediaSourceEventDispatcher,
-                /* canReportInitialDiscontinuity= */ false);
+                /* canReportInitialDiscontinuity= */ false,
+                /* downloadExecutor= */ null);
         streams[i] = sampleStream;
         sampleStreams.add(sampleStream);
         streamResetFlags[i] = true;