SpeedChangingAP: synchronize fields accessed from multiple threads

PiperOrigin-RevId: 643046385

libraries/common/src/main/java/androidx/media3/common/audio/SpeedChangingAudioProcessor.java

@@ -35,6 +35,7 @@ import java.util.Queue;
 import java.util.function.LongConsumer;
 import org.checkerframework.checker.initialization.qual.UnknownInitialization;
 import org.checkerframework.checker.nullness.qual.EnsuresNonNull;
+import org.checkerframework.checker.nullness.qual.RequiresNonNull;
 
 /**
  * An {@link AudioProcessor} that changes the speed of audio samples depending on their timestamp.
@@ -44,6 +45,8 @@ import org.checkerframework.checker.nullness.qual.EnsuresNonNull;
 @UnstableApi
 public final class SpeedChangingAudioProcessor extends BaseAudioProcessor {
 
+  private final Object lock;
+
   /** The speed provider that provides the speed for each timestamp. */
   private final SpeedProvider speedProvider;
 
@@ -52,35 +55,47 @@ public final class SpeedChangingAudioProcessor extends BaseAudioProcessor {
    * change required, the input buffer is copied to the output buffer and this processor is not
    * used.
    */
-  private final SonicAudioProcessor sonicAudioProcessor;
+  private final SynchronizedSonicAudioProcessor sonicAudioProcessor;
-
-  private final Object pendingCallbacksLock;
 
   // Elements in the same positions in the queues are associated.
-  @GuardedBy("pendingCallbacksLock")
+
+  @GuardedBy("lock")
   private final LongArrayQueue pendingCallbackInputTimesUs;
 
-  @GuardedBy("pendingCallbacksLock")
+  @GuardedBy("lock")
   private final Queue<TimestampConsumer> pendingCallbacks;
 
   // Elements in the same positions in the arrays are associated.
+
+  @GuardedBy("lock")
   private LongArray inputSegmentStartTimesUs;
+
+  @GuardedBy("lock")
   private LongArray outputSegmentStartTimesUs;
 
-  private float currentSpeed;
+  @GuardedBy("lock")
-  private long bytesRead;
   private long lastProcessedInputTimeUs;
-  private long lastSpeedAdjustedInputTimeUs;
-  private long lastSpeedAdjustedOutputTimeUs;
-  private boolean endOfStreamQueuedToSonic;
 
-  @GuardedBy("pendingCallbacksLock")
+  @GuardedBy("lock")
+  private long lastSpeedAdjustedInputTimeUs;
+
+  @GuardedBy("lock")
+  private long lastSpeedAdjustedOutputTimeUs;
+
+  @GuardedBy("lock")
   private long speedAdjustedTimeAsyncInputTimeUs;
 
+  @GuardedBy("lock")
+  private float currentSpeed;
+
+  private long bytesRead;
+
+  private boolean endOfStreamQueuedToSonic;
+
   public SpeedChangingAudioProcessor(SpeedProvider speedProvider) {
     this.speedProvider = speedProvider;
-    sonicAudioProcessor = new SonicAudioProcessor();
+    lock = new Object();
-    pendingCallbacksLock = new Object();
+    sonicAudioProcessor = new SynchronizedSonicAudioProcessor(lock);
     pendingCallbackInputTimesUs = new LongArrayQueue();
     pendingCallbacks = new ArrayDeque<>();
     speedAdjustedTimeAsyncInputTimeUs = C.TIME_UNSET;
@@ -106,18 +121,8 @@ public final class SpeedChangingAudioProcessor extends BaseAudioProcessor {
             /* multiplier= */ C.MICROS_PER_SECOND,
             /* divisor= */ (long) inputAudioFormat.sampleRate * inputAudioFormat.bytesPerFrame);
     float newSpeed = speedProvider.getSpeed(timeUs);
-    if (newSpeed != currentSpeed) {
+
-      updateSpeedChangeArrays(timeUs);
+    updateSpeed(newSpeed, timeUs);
-      currentSpeed = newSpeed;
-      if (isUsingSonic()) {
-        sonicAudioProcessor.setSpeed(newSpeed);
-        sonicAudioProcessor.setPitch(newSpeed);
-      }
-      // Invalidate any previously created buffers in SonicAudioProcessor and the base class.
-      sonicAudioProcessor.flush();
-      endOfStreamQueuedToSonic = false;
-      super.getOutput();
-    }
 
     int inputBufferLimit = inputBuffer.limit();
     long nextSpeedChangeTimeUs = speedProvider.getNextSpeedChangeTimeUs(timeUs);
@@ -158,7 +163,7 @@ public final class SpeedChangingAudioProcessor extends BaseAudioProcessor {
       buffer.flip();
     }
     bytesRead += inputBuffer.position() - startPosition;
-    lastProcessedInputTimeUs = updateLastProcessedInputTime();
+    updateLastProcessedInputTime();
     inputBuffer.limit(inputBufferLimit);
   }
 
@@ -213,7 +218,7 @@ public final class SpeedChangingAudioProcessor extends BaseAudioProcessor {
    *     from the caller of this method.
    */
   public void getSpeedAdjustedTimeAsync(long inputTimeUs, TimestampConsumer callback) {
-    synchronized (pendingCallbacksLock) {
+    synchronized (lock) {
       checkArgument(speedAdjustedTimeAsyncInputTimeUs < inputTimeUs);
       speedAdjustedTimeAsyncInputTimeUs = inputTimeUs;
       if ((inputTimeUs <= lastProcessedInputTimeUs && pendingCallbackInputTimesUs.isEmpty())
@@ -238,6 +243,7 @@ public final class SpeedChangingAudioProcessor extends BaseAudioProcessor {
    * @return The corresponding input duration in microseconds.
    */
   public long getMediaDurationUs(long playoutDurationUs) {
+    synchronized (lock) {
       int floorIndex = outputSegmentStartTimesUs.size() - 1;
       while (floorIndex > 0 && outputSegmentStartTimesUs.get(floorIndex) > playoutDurationUs) {
         floorIndex--;
@@ -260,11 +266,13 @@ public final class SpeedChangingAudioProcessor extends BaseAudioProcessor {
       }
       return inputSegmentStartTimesUs.get(floorIndex) + lastSegmentInputDurationUs;
     }
+  }
 
   /**
    * Assuming enough audio has been processed, calculates the time at which the {@code inputTimeUs}
    * is outputted at after the speed changes has been applied.
    */
+  @SuppressWarnings("GuardedBy") // All call sites are guarded.
   private long calculateSpeedAdjustedTime(long inputTimeUs) {
     int floorIndex = inputSegmentStartTimesUs.size() - 1;
     while (floorIndex > 0 && inputSegmentStartTimesUs.get(floorIndex) > inputTimeUs) {
@@ -299,7 +307,7 @@ public final class SpeedChangingAudioProcessor extends BaseAudioProcessor {
   }
 
   private void processPendingCallbacks() {
-    synchronized (pendingCallbacksLock) {
+    synchronized (lock) {
       while (!pendingCallbacks.isEmpty()
           && (pendingCallbackInputTimesUs.element() <= lastProcessedInputTimeUs || isEnded())) {
         pendingCallbacks
@@ -309,6 +317,24 @@ public final class SpeedChangingAudioProcessor extends BaseAudioProcessor {
     }
   }
 
+  private void updateSpeed(float newSpeed, long timeUs) {
+    synchronized (lock) {
+      if (newSpeed != currentSpeed) {
+        updateSpeedChangeArrays(timeUs);
+        currentSpeed = newSpeed;
+        if (isUsingSonic()) {
+          sonicAudioProcessor.setSpeed(newSpeed);
+          sonicAudioProcessor.setPitch(newSpeed);
+        }
+        // Invalidate any previously created buffers in SonicAudioProcessor and the base class.
+        sonicAudioProcessor.flush();
+        endOfStreamQueuedToSonic = false;
+        super.getOutput();
+      }
+    }
+  }
+
+  @SuppressWarnings("GuardedBy") // All call sites are guarded.
   private void updateSpeedChangeArrays(long currentSpeedChangeInputTimeUs) {
     long lastSpeedChangeOutputTimeUs =
         outputSegmentStartTimesUs.get(outputSegmentStartTimesUs.size() - 1);
@@ -334,7 +360,8 @@ public final class SpeedChangingAudioProcessor extends BaseAudioProcessor {
         : playoutDurationUs;
   }
 
-  private long updateLastProcessedInputTime() {
+  private void updateLastProcessedInputTime() {
+    synchronized (lock) {
       if (isUsingSonic()) {
         // TODO - b/320242819: Investigate whether bytesRead can be used here rather than
         //  sonicAudioProcessor.getProcessedInputBytes().
@@ -343,23 +370,29 @@ public final class SpeedChangingAudioProcessor extends BaseAudioProcessor {
                 /* timestamp= */ sonicAudioProcessor.getProcessedInputBytes(),
                 /* multiplier= */ C.MICROS_PER_SECOND,
                 /* divisor= */ (long) inputAudioFormat.sampleRate * inputAudioFormat.bytesPerFrame);
-      return inputSegmentStartTimesUs.get(inputSegmentStartTimesUs.size() - 1)
+        lastProcessedInputTimeUs =
+            inputSegmentStartTimesUs.get(inputSegmentStartTimesUs.size() - 1)
                 + currentProcessedInputDurationUs;
-    }
+      } else {
-    return Util.scaleLargeTimestamp(
+        lastProcessedInputTimeUs =
+            Util.scaleLargeTimestamp(
                 /* timestamp= */ bytesRead,
                 /* multiplier= */ C.MICROS_PER_SECOND,
                 /* divisor= */ (long) inputAudioFormat.sampleRate * inputAudioFormat.bytesPerFrame);
       }
+    }
+  }
 
   private boolean isUsingSonic() {
+    synchronized (lock) {
       return currentSpeed != 1f;
     }
+  }
 
   @EnsuresNonNull({"inputSegmentStartTimesUs", "outputSegmentStartTimesUs"})
+  @RequiresNonNull("lock")
   private void resetState(@UnknownInitialization SpeedChangingAudioProcessor this) {
-    currentSpeed = 1f;
+    synchronized (lock) {
-    bytesRead = 0;
       inputSegmentStartTimesUs = new LongArray();
       outputSegmentStartTimesUs = new LongArray();
       inputSegmentStartTimesUs.add(0);
@@ -367,6 +400,10 @@ public final class SpeedChangingAudioProcessor extends BaseAudioProcessor {
       lastProcessedInputTimeUs = 0;
       lastSpeedAdjustedInputTimeUs = 0;
       lastSpeedAdjustedOutputTimeUs = 0;
+      currentSpeed = 1f;
+    }
+
+    bytesRead = 0;
     endOfStreamQueuedToSonic = false;
     // TODO: b/339842724 - This should ideally also reset speedAdjustedTimeAsyncInputTimeUs and
     //  clear pendingCallbacks and pendingCallbacksInputTimes. We can't do this at the moment

libraries/common/src/main/java/androidx/media3/common/audio/SynchronizedSonicAudioProcessor.java

@@ -0,0 +1,131 @@
+/*
+ * Copyright (C) 2017 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.common.audio;
+
+import java.nio.ByteBuffer;
+
+/**
+ * A thread safe version {@link SonicAudioProcessor} that synchronizes calls before forwarding them
+ * to {@link SonicAudioProcessor}.
+ */
+/* package */ class SynchronizedSonicAudioProcessor implements AudioProcessor {
+
+  private final Object lock;
+  private final SonicAudioProcessor sonicAudioProcessor;
+
+  public SynchronizedSonicAudioProcessor(Object lock) {
+    this.lock = lock;
+    sonicAudioProcessor = new SonicAudioProcessor();
+  }
+
+  public final void setSpeed(float speed) {
+    synchronized (lock) {
+      sonicAudioProcessor.setSpeed(speed);
+    }
+  }
+
+  public final void setPitch(float pitch) {
+    synchronized (lock) {
+      sonicAudioProcessor.setPitch(pitch);
+    }
+  }
+
+  public final void setOutputSampleRateHz(int sampleRateHz) {
+    synchronized (lock) {
+      sonicAudioProcessor.setOutputSampleRateHz(sampleRateHz);
+    }
+  }
+
+  public final long getMediaDuration(long playoutDuration) {
+    synchronized (lock) {
+      return sonicAudioProcessor.getMediaDuration(playoutDuration);
+    }
+  }
+
+  public final long getPlayoutDuration(long mediaDuration) {
+    synchronized (lock) {
+      return sonicAudioProcessor.getPlayoutDuration(mediaDuration);
+    }
+  }
+
+  public final long getProcessedInputBytes() {
+    synchronized (lock) {
+      return sonicAudioProcessor.getProcessedInputBytes();
+    }
+  }
+
+  @Override
+  public long getDurationAfterProcessorApplied(long durationUs) {
+    return getPlayoutDuration(durationUs);
+  }
+
+  @Override
+  public final AudioFormat configure(AudioFormat inputAudioFormat)
+      throws UnhandledAudioFormatException {
+    synchronized (lock) {
+      return sonicAudioProcessor.configure(inputAudioFormat);
+    }
+  }
+
+  @Override
+  public final boolean isActive() {
+    synchronized (lock) {
+      return sonicAudioProcessor.isActive();
+    }
+  }
+
+  @Override
+  public final void queueInput(ByteBuffer inputBuffer) {
+    synchronized (lock) {
+      sonicAudioProcessor.queueInput(inputBuffer);
+    }
+  }
+
+  @Override
+  public final void queueEndOfStream() {
+    synchronized (lock) {
+      sonicAudioProcessor.queueEndOfStream();
+    }
+  }
+
+  @Override
+  public final ByteBuffer getOutput() {
+    synchronized (lock) {
+      return sonicAudioProcessor.getOutput();
+    }
+  }
+
+  @Override
+  public final boolean isEnded() {
+    synchronized (lock) {
+      return sonicAudioProcessor.isEnded();
+    }
+  }
+
+  @Override
+  public final void flush() {
+    synchronized (lock) {
+      sonicAudioProcessor.flush();
+    }
+  }
+
+  @Override
+  public final void reset() {
+    synchronized (lock) {
+      sonicAudioProcessor.reset();
+    }
+  }
+}