Commit 8bd01a7b by olly Committed by Oliver Woodman

Discard samples on the write-side of SampleQueue

PiperOrigin-RevId: 319205008
parent f8217140
...@@ -480,8 +480,7 @@ import org.checkerframework.checker.nullness.qual.MonotonicNonNull; ...@@ -480,8 +480,7 @@ import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
} }
maybeNotifyDownstreamFormat(sampleQueueIndex); maybeNotifyDownstreamFormat(sampleQueueIndex);
int result = int result =
sampleQueues[sampleQueueIndex].read( sampleQueues[sampleQueueIndex].read(formatHolder, buffer, formatRequired, loadingFinished);
formatHolder, buffer, formatRequired, loadingFinished, lastSeekPositionUs);
if (result == C.RESULT_NOTHING_READ) { if (result == C.RESULT_NOTHING_READ) {
maybeStartDeferredRetry(sampleQueueIndex); maybeStartDeferredRetry(sampleQueueIndex);
} }
...@@ -815,6 +814,9 @@ import org.checkerframework.checker.nullness.qual.MonotonicNonNull; ...@@ -815,6 +814,9 @@ import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
loadable.setLoadPosition( loadable.setLoadPosition(
Assertions.checkNotNull(seekMap).getSeekPoints(pendingResetPositionUs).first.position, Assertions.checkNotNull(seekMap).getSeekPoints(pendingResetPositionUs).first.position,
pendingResetPositionUs); pendingResetPositionUs);
for (SampleQueue sampleQueue : sampleQueues) {
sampleQueue.setStartTimeUs(pendingResetPositionUs);
}
pendingResetPositionUs = C.TIME_UNSET; pendingResetPositionUs = C.TIME_UNSET;
} }
extractedSamplesCountAtStartOfLoad = getExtractedSamplesCount(); extractedSamplesCountAtStartOfLoad = getExtractedSamplesCount();
......
...@@ -16,6 +16,7 @@ ...@@ -16,6 +16,7 @@
package com.google.android.exoplayer2.source; package com.google.android.exoplayer2.source;
import android.os.Looper; import android.os.Looper;
import android.util.Log;
import androidx.annotation.CallSuper; import androidx.annotation.CallSuper;
import androidx.annotation.Nullable; import androidx.annotation.Nullable;
import androidx.annotation.VisibleForTesting; import androidx.annotation.VisibleForTesting;
...@@ -52,6 +53,7 @@ public class SampleQueue implements TrackOutput { ...@@ -52,6 +53,7 @@ public class SampleQueue implements TrackOutput {
} }
@VisibleForTesting /* package */ static final int SAMPLE_CAPACITY_INCREMENT = 1000; @VisibleForTesting /* package */ static final int SAMPLE_CAPACITY_INCREMENT = 1000;
private static final String TAG = "SampleQueue";
private final SampleDataQueue sampleDataQueue; private final SampleDataQueue sampleDataQueue;
private final SampleExtrasHolder extrasHolder; private final SampleExtrasHolder extrasHolder;
...@@ -77,6 +79,7 @@ public class SampleQueue implements TrackOutput { ...@@ -77,6 +79,7 @@ public class SampleQueue implements TrackOutput {
private int relativeFirstIndex; private int relativeFirstIndex;
private int readPosition; private int readPosition;
private long startTimeUs;
private long largestDiscardedTimestampUs; private long largestDiscardedTimestampUs;
private long largestQueuedTimestampUs; private long largestQueuedTimestampUs;
private boolean isLastSampleQueued; private boolean isLastSampleQueued;
...@@ -87,6 +90,8 @@ public class SampleQueue implements TrackOutput { ...@@ -87,6 +90,8 @@ public class SampleQueue implements TrackOutput {
@Nullable private Format upstreamFormat; @Nullable private Format upstreamFormat;
@Nullable private Format upstreamCommittedFormat; @Nullable private Format upstreamCommittedFormat;
private int upstreamSourceId; private int upstreamSourceId;
private boolean upstreamAllSamplesAreSyncSamples;
private boolean loggedUnexpectedNonSyncSample;
private long sampleOffsetUs; private long sampleOffsetUs;
private boolean pendingSplice; private boolean pendingSplice;
...@@ -119,6 +124,7 @@ public class SampleQueue implements TrackOutput { ...@@ -119,6 +124,7 @@ public class SampleQueue implements TrackOutput {
sizes = new int[capacity]; sizes = new int[capacity];
cryptoDatas = new CryptoData[capacity]; cryptoDatas = new CryptoData[capacity];
formats = new Format[capacity]; formats = new Format[capacity];
startTimeUs = Long.MIN_VALUE;
largestDiscardedTimestampUs = Long.MIN_VALUE; largestDiscardedTimestampUs = Long.MIN_VALUE;
largestQueuedTimestampUs = Long.MIN_VALUE; largestQueuedTimestampUs = Long.MIN_VALUE;
upstreamFormatRequired = true; upstreamFormatRequired = true;
...@@ -155,6 +161,7 @@ public class SampleQueue implements TrackOutput { ...@@ -155,6 +161,7 @@ public class SampleQueue implements TrackOutput {
relativeFirstIndex = 0; relativeFirstIndex = 0;
readPosition = 0; readPosition = 0;
upstreamKeyframeRequired = true; upstreamKeyframeRequired = true;
startTimeUs = Long.MIN_VALUE;
largestDiscardedTimestampUs = Long.MIN_VALUE; largestDiscardedTimestampUs = Long.MIN_VALUE;
largestQueuedTimestampUs = Long.MIN_VALUE; largestQueuedTimestampUs = Long.MIN_VALUE;
isLastSampleQueued = false; isLastSampleQueued = false;
...@@ -167,6 +174,16 @@ public class SampleQueue implements TrackOutput { ...@@ -167,6 +174,16 @@ public class SampleQueue implements TrackOutput {
} }
/** /**
* Sets the start time for the queue. Samples with earlier timestamps will be discarded or have
* the {@link C#BUFFER_FLAG_DECODE_ONLY} flag set when read.
*
* @param startTimeUs The start time, in microseconds.
*/
public final void setStartTimeUs(long startTimeUs) {
this.startTimeUs = startTimeUs;
}
/**
* Sets a source identifier for subsequent samples. * Sets a source identifier for subsequent samples.
* *
* @param sourceId The source identifier. * @param sourceId The source identifier.
...@@ -325,8 +342,6 @@ public class SampleQueue implements TrackOutput { ...@@ -325,8 +342,6 @@ public class SampleQueue implements TrackOutput {
* it's not changing. A sample will never be read if set to true, however it is still possible * it's not changing. A sample will never be read if set to true, however it is still possible
* for the end of stream or nothing to be read. * for the end of stream or nothing to be read.
* @param loadingFinished True if an empty queue should be considered the end of the stream. * @param loadingFinished True if an empty queue should be considered the end of the stream.
* @param decodeOnlyUntilUs If a buffer is read, the {@link C#BUFFER_FLAG_DECODE_ONLY} flag will
* be set if the buffer's timestamp is less than this value.
* @return The result, which can be {@link C#RESULT_NOTHING_READ}, {@link C#RESULT_FORMAT_READ} or * @return The result, which can be {@link C#RESULT_NOTHING_READ}, {@link C#RESULT_FORMAT_READ} or
* {@link C#RESULT_BUFFER_READ}. * {@link C#RESULT_BUFFER_READ}.
*/ */
...@@ -335,11 +350,9 @@ public class SampleQueue implements TrackOutput { ...@@ -335,11 +350,9 @@ public class SampleQueue implements TrackOutput {
FormatHolder formatHolder, FormatHolder formatHolder,
DecoderInputBuffer buffer, DecoderInputBuffer buffer,
boolean formatRequired, boolean formatRequired,
boolean loadingFinished, boolean loadingFinished) {
long decodeOnlyUntilUs) {
int result = int result =
readSampleMetadata( readSampleMetadata(formatHolder, buffer, formatRequired, loadingFinished, extrasHolder);
formatHolder, buffer, formatRequired, loadingFinished, decodeOnlyUntilUs, extrasHolder);
if (result == C.RESULT_BUFFER_READ && !buffer.isEndOfStream() && !buffer.isFlagsOnly()) { if (result == C.RESULT_BUFFER_READ && !buffer.isEndOfStream() && !buffer.isFlagsOnly()) {
sampleDataQueue.readToBuffer(buffer, extrasHolder); sampleDataQueue.readToBuffer(buffer, extrasHolder);
} }
...@@ -357,6 +370,7 @@ public class SampleQueue implements TrackOutput { ...@@ -357,6 +370,7 @@ public class SampleQueue implements TrackOutput {
if (sampleIndex < absoluteFirstIndex || sampleIndex > absoluteFirstIndex + length) { if (sampleIndex < absoluteFirstIndex || sampleIndex > absoluteFirstIndex + length) {
return false; return false;
} }
startTimeUs = Long.MIN_VALUE;
readPosition = sampleIndex - absoluteFirstIndex; readPosition = sampleIndex - absoluteFirstIndex;
return true; return true;
} }
...@@ -382,6 +396,7 @@ public class SampleQueue implements TrackOutput { ...@@ -382,6 +396,7 @@ public class SampleQueue implements TrackOutput {
if (offset == -1) { if (offset == -1) {
return false; return false;
} }
startTimeUs = timeUs;
readPosition += offset; readPosition += offset;
return true; return true;
} }
...@@ -513,6 +528,22 @@ public class SampleQueue implements TrackOutput { ...@@ -513,6 +528,22 @@ public class SampleQueue implements TrackOutput {
} }
timeUs += sampleOffsetUs; timeUs += sampleOffsetUs;
if (upstreamAllSamplesAreSyncSamples) {
if (timeUs < startTimeUs) {
// If we know that all samples are sync samples, we can discard those that come before the
// start time on the write side of the queue.
return;
}
if ((flags & C.BUFFER_FLAG_KEY_FRAME) == 0) {
// The flag should always be set unless the source content has incorrect sample metadata.
// Log a warning (once per format change, to avoid log spam) and override the flag.
if (!loggedUnexpectedNonSyncSample) {
Log.w(TAG, "Overriding unexpected non-sync sample for format: " + upstreamFormat);
loggedUnexpectedNonSyncSample = true;
}
flags |= C.BUFFER_FLAG_KEY_FRAME;
}
}
if (pendingSplice) { if (pendingSplice) {
if (!isKeyframe || !attemptSplice(timeUs)) { if (!isKeyframe || !attemptSplice(timeUs)) {
return; return;
...@@ -568,25 +599,9 @@ public class SampleQueue implements TrackOutput { ...@@ -568,25 +599,9 @@ public class SampleQueue implements TrackOutput {
DecoderInputBuffer buffer, DecoderInputBuffer buffer,
boolean formatRequired, boolean formatRequired,
boolean loadingFinished, boolean loadingFinished,
long decodeOnlyUntilUs,
SampleExtrasHolder extrasHolder) { SampleExtrasHolder extrasHolder) {
buffer.waitingForKeys = false; buffer.waitingForKeys = false;
// This is a temporary fix for https://github.com/google/ExoPlayer/issues/6155. if (!hasNextSample()) {
// TODO: Remove it and replace it with a fix that discards samples when writing to the queue.
boolean hasNextSample;
int relativeReadIndex = C.INDEX_UNSET;
while ((hasNextSample = hasNextSample())) {
relativeReadIndex = getRelativeIndex(readPosition);
long timeUs = timesUs[relativeReadIndex];
if (timeUs < decodeOnlyUntilUs
&& MimeTypes.allSamplesAreSyncSamples(formats[relativeReadIndex].sampleMimeType)) {
readPosition++;
} else {
break;
}
}
if (!hasNextSample) {
if (loadingFinished || isLastSampleQueued) { if (loadingFinished || isLastSampleQueued) {
buffer.setFlags(C.BUFFER_FLAG_END_OF_STREAM); buffer.setFlags(C.BUFFER_FLAG_END_OF_STREAM);
return C.RESULT_BUFFER_READ; return C.RESULT_BUFFER_READ;
...@@ -598,6 +613,7 @@ public class SampleQueue implements TrackOutput { ...@@ -598,6 +613,7 @@ public class SampleQueue implements TrackOutput {
} }
} }
int relativeReadIndex = getRelativeIndex(readPosition);
if (formatRequired || formats[relativeReadIndex] != downstreamFormat) { if (formatRequired || formats[relativeReadIndex] != downstreamFormat) {
onFormatResult(formats[relativeReadIndex], formatHolder); onFormatResult(formats[relativeReadIndex], formatHolder);
return C.RESULT_FORMAT_READ; return C.RESULT_FORMAT_READ;
...@@ -610,7 +626,7 @@ public class SampleQueue implements TrackOutput { ...@@ -610,7 +626,7 @@ public class SampleQueue implements TrackOutput {
buffer.setFlags(flags[relativeReadIndex]); buffer.setFlags(flags[relativeReadIndex]);
buffer.timeUs = timesUs[relativeReadIndex]; buffer.timeUs = timesUs[relativeReadIndex];
if (buffer.timeUs < decodeOnlyUntilUs) { if (buffer.timeUs < startTimeUs) {
buffer.addFlag(C.BUFFER_FLAG_DECODE_ONLY); buffer.addFlag(C.BUFFER_FLAG_DECODE_ONLY);
} }
if (buffer.isFlagsOnly()) { if (buffer.isFlagsOnly()) {
...@@ -631,16 +647,19 @@ public class SampleQueue implements TrackOutput { ...@@ -631,16 +647,19 @@ public class SampleQueue implements TrackOutput {
// current upstreamFormat so we can detect format changes on the read side using cheap // current upstreamFormat so we can detect format changes on the read side using cheap
// referential quality. // referential quality.
return false; return false;
} else if (Util.areEqual(format, upstreamCommittedFormat)) { }
if (Util.areEqual(format, upstreamCommittedFormat)) {
// The format has changed back to the format of the last committed sample. If they are // The format has changed back to the format of the last committed sample. If they are
// different objects, we revert back to using upstreamCommittedFormat as the upstreamFormat // different objects, we revert back to using upstreamCommittedFormat as the upstreamFormat
// so we can detect format changes on the read side using cheap referential equality. // so we can detect format changes on the read side using cheap referential equality.
upstreamFormat = upstreamCommittedFormat; upstreamFormat = upstreamCommittedFormat;
return true;
} else { } else {
upstreamFormat = format; upstreamFormat = format;
return true;
} }
upstreamAllSamplesAreSyncSamples =
MimeTypes.allSamplesAreSyncSamples(upstreamFormat.sampleMimeType);
loggedUnexpectedNonSyncSample = false;
return true;
} }
private synchronized long discardSampleMetadataTo( private synchronized long discardSampleMetadataTo(
......
...@@ -87,7 +87,6 @@ public class ChunkSampleStream<T extends ChunkSource> implements SampleStream, S ...@@ -87,7 +87,6 @@ public class ChunkSampleStream<T extends ChunkSource> implements SampleStream, S
private long lastSeekPositionUs; private long lastSeekPositionUs;
private int nextNotifyPrimaryFormatMediaChunkIndex; private int nextNotifyPrimaryFormatMediaChunkIndex;
/* package */ long decodeOnlyUntilPositionUs;
/* package */ boolean loadingFinished; /* package */ boolean loadingFinished;
/** /**
...@@ -282,14 +281,12 @@ public class ChunkSampleStream<T extends ChunkSource> implements SampleStream, S ...@@ -282,14 +281,12 @@ public class ChunkSampleStream<T extends ChunkSource> implements SampleStream, S
if (seekToMediaChunk != null) { if (seekToMediaChunk != null) {
// When seeking to the start of a chunk we use the index of the first sample in the chunk // When seeking to the start of a chunk we use the index of the first sample in the chunk
// rather than the seek position. This ensures we seek to the keyframe at the start of the // rather than the seek position. This ensures we seek to the keyframe at the start of the
// chunk even if the sample timestamps are slightly offset from the chunk start times. // chunk even if its timestamp is slightly earlier than the advertised chunk start time.
seekInsideBuffer = primarySampleQueue.seekTo(seekToMediaChunk.getFirstSampleIndex(0)); seekInsideBuffer = primarySampleQueue.seekTo(seekToMediaChunk.getFirstSampleIndex(0));
decodeOnlyUntilPositionUs = 0;
} else { } else {
seekInsideBuffer = seekInsideBuffer =
primarySampleQueue.seekTo( primarySampleQueue.seekTo(
positionUs, /* allowTimeBeyondBuffer= */ positionUs < getNextLoadPositionUs()); positionUs, /* allowTimeBeyondBuffer= */ positionUs < getNextLoadPositionUs());
decodeOnlyUntilPositionUs = lastSeekPositionUs;
} }
if (seekInsideBuffer) { if (seekInsideBuffer) {
...@@ -383,8 +380,7 @@ public class ChunkSampleStream<T extends ChunkSource> implements SampleStream, S ...@@ -383,8 +380,7 @@ public class ChunkSampleStream<T extends ChunkSource> implements SampleStream, S
} }
maybeNotifyPrimaryTrackFormatChanged(); maybeNotifyPrimaryTrackFormatChanged();
return primarySampleQueue.read( return primarySampleQueue.read(formatHolder, buffer, formatRequired, loadingFinished);
formatHolder, buffer, formatRequired, loadingFinished, decodeOnlyUntilPositionUs);
} }
@Override @Override
...@@ -577,9 +573,16 @@ public class ChunkSampleStream<T extends ChunkSource> implements SampleStream, S ...@@ -577,9 +573,16 @@ public class ChunkSampleStream<T extends ChunkSource> implements SampleStream, S
if (isMediaChunk(loadable)) { if (isMediaChunk(loadable)) {
BaseMediaChunk mediaChunk = (BaseMediaChunk) loadable; BaseMediaChunk mediaChunk = (BaseMediaChunk) loadable;
if (pendingReset) { if (pendingReset) {
boolean resetToMediaChunk = mediaChunk.startTimeUs == pendingResetPositionUs; // Only set the queue start times if we're not seeking to a chunk boundary. If we are
// Only enable setting of the decode only flag if we're not resetting to a chunk boundary. // seeking to a chunk boundary then we want the queue to pass through all of the samples in
decodeOnlyUntilPositionUs = resetToMediaChunk ? 0 : pendingResetPositionUs; // the chunk. Doing this ensures we'll always output the keyframe at the start of the chunk,
// even if its timestamp is slightly earlier than the advertised chunk start time.
if (mediaChunk.startTimeUs != pendingResetPositionUs) {
primarySampleQueue.setStartTimeUs(pendingResetPositionUs);
for (SampleQueue embeddedSampleQueue : embeddedSampleQueues) {
embeddedSampleQueue.setStartTimeUs(pendingResetPositionUs);
}
}
pendingResetPositionUs = C.TIME_UNSET; pendingResetPositionUs = C.TIME_UNSET;
} }
mediaChunk.init(chunkOutput); mediaChunk.init(chunkOutput);
...@@ -799,12 +802,7 @@ public class ChunkSampleStream<T extends ChunkSource> implements SampleStream, S ...@@ -799,12 +802,7 @@ public class ChunkSampleStream<T extends ChunkSource> implements SampleStream, S
return C.RESULT_NOTHING_READ; return C.RESULT_NOTHING_READ;
} }
maybeNotifyDownstreamFormat(); maybeNotifyDownstreamFormat();
return sampleQueue.read( return sampleQueue.read(formatHolder, buffer, formatRequired, loadingFinished);
formatHolder,
buffer,
formatRequired,
loadingFinished,
decodeOnlyUntilPositionUs);
} }
public void release() { public void release() {
......
...@@ -380,11 +380,7 @@ public final class PlayerEmsgHandler implements Handler.Callback { ...@@ -380,11 +380,7 @@ public final class PlayerEmsgHandler implements Handler.Callback {
buffer.clear(); buffer.clear();
int result = int result =
sampleQueue.read( sampleQueue.read(
formatHolder, formatHolder, buffer, /* formatRequired= */ false, /* loadingFinished= */ false);
buffer,
/* formatRequired= */ false,
/* loadingFinished= */ false,
/* decodeOnlyUntilUs= */ 0);
if (result == C.RESULT_BUFFER_READ) { if (result == C.RESULT_BUFFER_READ) {
buffer.flip(); buffer.flip();
return buffer; return buffer;
......
...@@ -560,8 +560,7 @@ import org.checkerframework.checker.nullness.qual.RequiresNonNull; ...@@ -560,8 +560,7 @@ import org.checkerframework.checker.nullness.qual.RequiresNonNull;
} }
int result = int result =
sampleQueues[sampleQueueIndex].read( sampleQueues[sampleQueueIndex].read(formatHolder, buffer, requireFormat, loadingFinished);
formatHolder, buffer, requireFormat, loadingFinished, lastSeekPositionUs);
if (result == C.RESULT_FORMAT_READ) { if (result == C.RESULT_FORMAT_READ) {
Format format = Assertions.checkNotNull(formatHolder.format); Format format = Assertions.checkNotNull(formatHolder.format);
if (sampleQueueIndex == primarySampleQueueIndex) { if (sampleQueueIndex == primarySampleQueueIndex) {
...@@ -641,6 +640,9 @@ import org.checkerframework.checker.nullness.qual.RequiresNonNull; ...@@ -641,6 +640,9 @@ import org.checkerframework.checker.nullness.qual.RequiresNonNull;
if (isPendingReset()) { if (isPendingReset()) {
chunkQueue = Collections.emptyList(); chunkQueue = Collections.emptyList();
loadPositionUs = pendingResetPositionUs; loadPositionUs = pendingResetPositionUs;
for (SampleQueue sampleQueue : sampleQueues) {
sampleQueue.setStartTimeUs(pendingResetPositionUs);
}
} else { } else {
chunkQueue = readOnlyMediaChunks; chunkQueue = readOnlyMediaChunks;
HlsMediaChunk lastMediaChunk = getLastMediaChunk(); HlsMediaChunk lastMediaChunk = getLastMediaChunk();
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment