diff options
Diffstat (limited to 'media/libstagefright/httplive/PlaylistFetcher.cpp')
| -rw-r--r-- | media/libstagefright/httplive/PlaylistFetcher.cpp | 764 |
1 files changed, 502 insertions, 262 deletions
diff --git a/media/libstagefright/httplive/PlaylistFetcher.cpp b/media/libstagefright/httplive/PlaylistFetcher.cpp index 7f818a8..368612d 100644 --- a/media/libstagefright/httplive/PlaylistFetcher.cpp +++ b/media/libstagefright/httplive/PlaylistFetcher.cpp @@ -33,6 +33,7 @@ #include <media/stagefright/foundation/ABitReader.h> #include <media/stagefright/foundation/ABuffer.h> #include <media/stagefright/foundation/ADebug.h> +#include <media/stagefright/foundation/AUtils.h> #include <media/stagefright/foundation/hexdump.h> #include <media/stagefright/FileSource.h> #include <media/stagefright/MediaDefs.h> @@ -47,12 +48,96 @@ namespace android { // static -const int64_t PlaylistFetcher::kMinBufferedDurationUs = 10000000ll; +const int64_t PlaylistFetcher::kMinBufferedDurationUs = 30000000ll; const int64_t PlaylistFetcher::kMaxMonitorDelayUs = 3000000ll; -const int64_t PlaylistFetcher::kFetcherResumeThreshold = 100000ll; // LCM of 188 (size of a TS packet) & 1k works well const int32_t PlaylistFetcher::kDownloadBlockSize = 47 * 1024; -const int32_t PlaylistFetcher::kNumSkipFrames = 5; + +struct PlaylistFetcher::DownloadState : public RefBase { + DownloadState(); + void resetState(); + bool hasSavedState() const; + void restoreState( + AString &uri, + sp<AMessage> &itemMeta, + sp<ABuffer> &buffer, + sp<ABuffer> &tsBuffer, + int32_t &firstSeqNumberInPlaylist, + int32_t &lastSeqNumberInPlaylist); + void saveState( + AString &uri, + sp<AMessage> &itemMeta, + sp<ABuffer> &buffer, + sp<ABuffer> &tsBuffer, + int32_t &firstSeqNumberInPlaylist, + int32_t &lastSeqNumberInPlaylist); + +private: + bool mHasSavedState; + AString mUri; + sp<AMessage> mItemMeta; + sp<ABuffer> mBuffer; + sp<ABuffer> mTsBuffer; + int32_t mFirstSeqNumberInPlaylist; + int32_t mLastSeqNumberInPlaylist; +}; + +PlaylistFetcher::DownloadState::DownloadState() { + resetState(); +} + +bool PlaylistFetcher::DownloadState::hasSavedState() const { + return mHasSavedState; +} + +void PlaylistFetcher::DownloadState::resetState() { + mHasSavedState = false; + + mUri.clear(); + mItemMeta = NULL; + mBuffer = NULL; + mTsBuffer = NULL; + mFirstSeqNumberInPlaylist = 0; + mLastSeqNumberInPlaylist = 0; +} + +void PlaylistFetcher::DownloadState::restoreState( + AString &uri, + sp<AMessage> &itemMeta, + sp<ABuffer> &buffer, + sp<ABuffer> &tsBuffer, + int32_t &firstSeqNumberInPlaylist, + int32_t &lastSeqNumberInPlaylist) { + if (!mHasSavedState) { + return; + } + + uri = mUri; + itemMeta = mItemMeta; + buffer = mBuffer; + tsBuffer = mTsBuffer; + firstSeqNumberInPlaylist = mFirstSeqNumberInPlaylist; + lastSeqNumberInPlaylist = mLastSeqNumberInPlaylist; + + resetState(); +} + +void PlaylistFetcher::DownloadState::saveState( + AString &uri, + sp<AMessage> &itemMeta, + sp<ABuffer> &buffer, + sp<ABuffer> &tsBuffer, + int32_t &firstSeqNumberInPlaylist, + int32_t &lastSeqNumberInPlaylist) { + mHasSavedState = true; + + mUri = uri; + mItemMeta = itemMeta; + mBuffer = buffer; + mTsBuffer = tsBuffer; + mFirstSeqNumberInPlaylist = firstSeqNumberInPlaylist; + mLastSeqNumberInPlaylist = lastSeqNumberInPlaylist; +} PlaylistFetcher::PlaylistFetcher( const sp<AMessage> ¬ify, @@ -71,18 +156,21 @@ PlaylistFetcher::PlaylistFetcher( mSeqNumber(-1), mNumRetries(0), mStartup(true), - mAdaptive(false), - mPrepared(false), + mIDRFound(false), + mSeekMode(LiveSession::kSeekModeExactPosition), mTimeChangeSignaled(false), mNextPTSTimeUs(-1ll), mMonitorQueueGeneration(0), mSubtitleGeneration(subtitleGeneration), mLastDiscontinuitySeq(-1ll), - mStopping(false), mRefreshState(INITIAL_MINIMUM_RELOAD_DELAY), mFirstPTSValid(false), - mVideoBuffer(new AnotherPacketSource(NULL)) { + mFirstTimeUs(-1ll), + mVideoBuffer(new AnotherPacketSource(NULL)), + mThresholdRatio(-1.0f), + mDownloadState(new DownloadState()) { memset(mPlaylistHash, 0, sizeof(mPlaylistHash)); + mHTTPDataSource = mSession->getHTTPDataSource(); } PlaylistFetcher::~PlaylistFetcher() { @@ -119,6 +207,32 @@ int64_t PlaylistFetcher::getSegmentStartTimeUs(int32_t seqNumber) const { return segmentStartUs; } +int64_t PlaylistFetcher::getSegmentDurationUs(int32_t seqNumber) const { + CHECK(mPlaylist != NULL); + + int32_t firstSeqNumberInPlaylist; + if (mPlaylist->meta() == NULL || !mPlaylist->meta()->findInt32( + "media-sequence", &firstSeqNumberInPlaylist)) { + firstSeqNumberInPlaylist = 0; + } + + int32_t lastSeqNumberInPlaylist = + firstSeqNumberInPlaylist + (int32_t)mPlaylist->size() - 1; + + CHECK_GE(seqNumber, firstSeqNumberInPlaylist); + CHECK_LE(seqNumber, lastSeqNumberInPlaylist); + + int32_t index = seqNumber - firstSeqNumberInPlaylist; + sp<AMessage> itemMeta; + CHECK(mPlaylist->itemAt( + index, NULL /* uri */, &itemMeta)); + + int64_t itemDurationUs; + CHECK(itemMeta->findInt64("durationUs", &itemDurationUs)); + + return itemDurationUs; +} + int64_t PlaylistFetcher::delayUsToRefreshPlaylist() const { int64_t nowUs = ALooper::GetNowUs(); @@ -334,9 +448,12 @@ void PlaylistFetcher::cancelMonitorQueue() { ++mMonitorQueueGeneration; } -void PlaylistFetcher::setStopping(bool stopping) { - AutoMutex _l(mStoppingLock); - mStopping = stopping; +void PlaylistFetcher::setStoppingThreshold(float thresholdRatio) { + AutoMutex _l(mThresholdLock); + if (mStreamTypeMask == LiveSession::STREAMTYPE_SUBTITLES) { + return; + } + mThresholdRatio = thresholdRatio; } void PlaylistFetcher::startAsync( @@ -346,7 +463,7 @@ void PlaylistFetcher::startAsync( int64_t startTimeUs, int64_t segmentStartTimeUs, int32_t startDiscontinuitySeq, - bool adaptive) { + LiveSession::SeekMode seekMode) { sp<AMessage> msg = new AMessage(kWhatStart, this); uint32_t streamTypeMask = 0ul; @@ -370,19 +487,19 @@ void PlaylistFetcher::startAsync( msg->setInt64("startTimeUs", startTimeUs); msg->setInt64("segmentStartTimeUs", segmentStartTimeUs); msg->setInt32("startDiscontinuitySeq", startDiscontinuitySeq); - msg->setInt32("adaptive", adaptive); + msg->setInt32("seekMode", seekMode); msg->post(); } -void PlaylistFetcher::pauseAsync(bool immediate) { - if (immediate) { - setStopping(true); +void PlaylistFetcher::pauseAsync(float thresholdRatio) { + if (thresholdRatio >= 0.0f) { + setStoppingThreshold(thresholdRatio); } (new AMessage(kWhatPause, this))->post(); } void PlaylistFetcher::stopAsync(bool clear) { - setStopping(true); + setStoppingThreshold(0.0f); sp<AMessage> msg = new AMessage(kWhatStop, this); msg->setInt32("clear", clear); @@ -414,6 +531,10 @@ void PlaylistFetcher::onMessageReceived(const sp<AMessage> &msg) { sp<AMessage> notify = mNotify->dup(); notify->setInt32("what", kWhatPaused); + notify->setInt32("seekMode", + mDownloadState->hasSavedState() + ? LiveSession::kSeekModeNextSample + : LiveSession::kSeekModeNextSegment); notify->post(); break; } @@ -463,7 +584,7 @@ status_t PlaylistFetcher::onStart(const sp<AMessage> &msg) { mStopParams.clear(); mStartTimeUsNotify = mNotify->dup(); mStartTimeUsNotify->setInt32("what", kWhatStartedAt); - mStartTimeUsNotify->setInt32("streamMask", 0); + mStartTimeUsNotify->setString("uri", mURI); uint32_t streamTypeMask; CHECK(msg->findInt32("streamTypeMask", (int32_t *)&streamTypeMask)); @@ -471,11 +592,11 @@ status_t PlaylistFetcher::onStart(const sp<AMessage> &msg) { int64_t startTimeUs; int64_t segmentStartTimeUs; int32_t startDiscontinuitySeq; - int32_t adaptive; + int32_t seekMode; CHECK(msg->findInt64("startTimeUs", &startTimeUs)); CHECK(msg->findInt64("segmentStartTimeUs", &segmentStartTimeUs)); CHECK(msg->findInt32("startDiscontinuitySeq", &startDiscontinuitySeq)); - CHECK(msg->findInt32("adaptive", &adaptive)); + CHECK(msg->findInt32("seekMode", &seekMode)); if (streamTypeMask & LiveSession::STREAMTYPE_AUDIO) { void *ptr; @@ -507,20 +628,26 @@ status_t PlaylistFetcher::onStart(const sp<AMessage> &msg) { mStreamTypeMask = streamTypeMask; mSegmentStartTimeUs = segmentStartTimeUs; - mDiscontinuitySeq = startDiscontinuitySeq; + + if (startDiscontinuitySeq >= 0) { + mDiscontinuitySeq = startDiscontinuitySeq; + } mRefreshState = INITIAL_MINIMUM_RELOAD_DELAY; + mSeekMode = (LiveSession::SeekMode) seekMode; + + if (startTimeUs >= 0 || mSeekMode == LiveSession::kSeekModeNextSample) { + mStartup = true; + mIDRFound = false; + mVideoBuffer->clear(); + } if (startTimeUs >= 0) { mStartTimeUs = startTimeUs; mFirstPTSValid = false; mSeqNumber = -1; - mStartup = true; - mPrepared = false; - mIDRFound = false; mTimeChangeSignaled = false; - mAdaptive = adaptive; - mVideoBuffer->clear(); + mDownloadState->resetState(); } postMonitorQueue(); @@ -532,7 +659,7 @@ void PlaylistFetcher::onPause() { cancelMonitorQueue(); mLastDiscontinuitySeq = mDiscontinuitySeq; - setStopping(false); + setStoppingThreshold(-1.0f); } void PlaylistFetcher::onStop(const sp<AMessage> &msg) { @@ -547,10 +674,14 @@ void PlaylistFetcher::onStop(const sp<AMessage> &msg) { } } + // close off the connection after use + mHTTPDataSource->disconnect(); + + mDownloadState->resetState(); mPacketSources.clear(); mStreamTypeMask = 0; - setStopping(false); + setStoppingThreshold(-1.0f); } // Resume until we have reached the boundary timestamps listed in `msg`; when @@ -560,61 +691,18 @@ status_t PlaylistFetcher::onResumeUntil(const sp<AMessage> &msg) { sp<AMessage> params; CHECK(msg->findMessage("params", ¶ms)); - size_t stopCount = 0; - for (size_t i = 0; i < mPacketSources.size(); i++) { - sp<AnotherPacketSource> packetSource = mPacketSources.valueAt(i); - - const char *stopKey; - int streamType = mPacketSources.keyAt(i); - - if (streamType == LiveSession::STREAMTYPE_SUBTITLES) { - // the subtitle track can always be stopped - ++stopCount; - continue; - } - - switch (streamType) { - case LiveSession::STREAMTYPE_VIDEO: - stopKey = "timeUsVideo"; - break; - - case LiveSession::STREAMTYPE_AUDIO: - stopKey = "timeUsAudio"; - break; - - default: - TRESPASS(); - } - - // check if this stream has too little data left to be resumed - int32_t discontinuitySeq; - int64_t latestTimeUs = 0, stopTimeUs = 0; - sp<AMessage> latestMeta = packetSource->getLatestEnqueuedMeta(); - if (latestMeta != NULL - && latestMeta->findInt32("discontinuitySeq", &discontinuitySeq) - && discontinuitySeq == mDiscontinuitySeq - && latestMeta->findInt64("timeUs", &latestTimeUs) - && params->findInt64(stopKey, &stopTimeUs) - && stopTimeUs - latestTimeUs < kFetcherResumeThreshold) { - ++stopCount; - } - } - - // Don't resume if all streams are within a resume threshold - if (stopCount == mPacketSources.size()) { - for (size_t i = 0; i < mPacketSources.size(); i++) { - mPacketSources.valueAt(i)->queueAccessUnit(mSession->createFormatChangeBuffer()); - } - stopAsync(/* clear = */ false); - return OK; - } - mStopParams = params; onDownloadNext(); return OK; } +void PlaylistFetcher::notifyStopReached() { + sp<AMessage> notify = mNotify->dup(); + notify->setInt32("what", kWhatStopReached); + notify->post(); +} + void PlaylistFetcher::notifyError(status_t err) { sp<AMessage> notify = mNotify->dup(); notify->setInt32("what", kWhatError); @@ -634,7 +722,12 @@ void PlaylistFetcher::queueDiscontinuity( void PlaylistFetcher::onMonitorQueue() { bool downloadMore = false; - refreshPlaylist(); + + // in the middle of an unfinished download, delay + // playlist refresh as it'll change seq numbers + if (!mDownloadState->hasSavedState()) { + refreshPlaylist(); + } int32_t targetDurationSecs; int64_t targetDurationUs = kMinBufferedDurationUs; @@ -648,23 +741,23 @@ void PlaylistFetcher::onMonitorQueue() { targetDurationUs = targetDurationSecs * 1000000ll; } - int64_t durationToBufferUs = kMinBufferedDurationUs; - int64_t bufferedDurationUs = 0ll; - status_t finalResult = NOT_ENOUGH_DATA; + status_t finalResult = OK; if (mStreamTypeMask == LiveSession::STREAMTYPE_SUBTITLES) { sp<AnotherPacketSource> packetSource = mPacketSources.valueFor(LiveSession::STREAMTYPE_SUBTITLES); bufferedDurationUs = packetSource->getBufferedDurationUs(&finalResult); - finalResult = OK; } else { - // Use max stream duration to prevent us from waiting on a non-existent stream; - // when we cannot make out from the manifest what streams are included in a playlist - // we might assume extra streams. + // Use min stream duration, but ignore streams that never have any packet + // enqueued to prevent us from waiting on a non-existent stream; + // when we cannot make out from the manifest what streams are included in + // a playlist we might assume extra streams. + bufferedDurationUs = -1ll; for (size_t i = 0; i < mPacketSources.size(); ++i) { - if ((mStreamTypeMask & mPacketSources.keyAt(i)) == 0) { + if ((mStreamTypeMask & mPacketSources.keyAt(i)) == 0 + || mPacketSources[i]->getLatestEnqueuedMeta() == NULL) { continue; } @@ -672,24 +765,19 @@ void PlaylistFetcher::onMonitorQueue() { mPacketSources.valueAt(i)->getBufferedDurationUs(&finalResult); ALOGV("buffered %" PRId64 " for stream %d", bufferedStreamDurationUs, mPacketSources.keyAt(i)); - if (bufferedStreamDurationUs > bufferedDurationUs) { + if (bufferedDurationUs == -1ll + || bufferedStreamDurationUs < bufferedDurationUs) { bufferedDurationUs = bufferedStreamDurationUs; } } - } - downloadMore = (bufferedDurationUs < durationToBufferUs); - - // signal start if buffered up at least the target size - if (!mPrepared && bufferedDurationUs > targetDurationUs && downloadMore) { - mPrepared = true; - - ALOGV("prepared, buffered=%" PRId64 " > %" PRId64 "", - bufferedDurationUs, targetDurationUs); + if (bufferedDurationUs == -1ll) { + bufferedDurationUs = 0ll; + } } - if (finalResult == OK && downloadMore) { + if (finalResult == OK && bufferedDurationUs < kMinBufferedDurationUs) { ALOGV("monitoring, buffered=%" PRId64 " < %" PRId64 "", - bufferedDurationUs, durationToBufferUs); + bufferedDurationUs, kMinBufferedDurationUs); // delay the next download slightly; hopefully this gives other concurrent fetchers // a better chance to run. // onDownloadNext(); @@ -697,13 +785,16 @@ void PlaylistFetcher::onMonitorQueue() { msg->setInt32("generation", mMonitorQueueGeneration); msg->post(1000l); } else { - // Nothing to do yet, try again in a second. - int64_t delayUs = mPrepared ? kMaxMonitorDelayUs : targetDurationUs / 2; + // We'd like to maintain buffering above durationToBufferUs, so try + // again when buffer just about to go below durationToBufferUs + // (or after targetDurationUs / 2, whichever is smaller). + int64_t delayUs = bufferedDurationUs - kMinBufferedDurationUs + 1000000ll; + if (delayUs > targetDurationUs / 2) { + delayUs = targetDurationUs / 2; + } ALOGV("pausing for %" PRId64 ", buffered=%" PRId64 " > %" PRId64 "", - delayUs, bufferedDurationUs, durationToBufferUs); - // :TRICKY: need to enforce minimum delay because the delay to - // refresh the playlist will become 0 - postMonitorQueue(delayUs, mPrepared ? targetDurationUs * 2 : 0); + delayUs, bufferedDurationUs, kMinBufferedDurationUs); + postMonitorQueue(delayUs); } } @@ -732,6 +823,13 @@ status_t PlaylistFetcher::refreshPlaylist() { if (mPlaylist->isComplete() || mPlaylist->isEvent()) { updateDuration(); } + // Notify LiveSession to use target-duration based buffering level + // for up/down switch. Default LiveSession::kUpSwitchMark may not + // be reachable for live streams, as our max buffering amount is + // limited to 3 segments. + if (!mPlaylist->isComplete()) { + updateTargetDuration(); + } } mLastPlaylistFetchTimeUs = ALooper::GetNowUs(); @@ -744,10 +842,69 @@ bool PlaylistFetcher::bufferStartsWithTsSyncByte(const sp<ABuffer>& buffer) { return buffer->size() > 0 && buffer->data()[0] == 0x47; } -void PlaylistFetcher::onDownloadNext() { +bool PlaylistFetcher::shouldPauseDownload() { + if (mStreamTypeMask == LiveSession::STREAMTYPE_SUBTITLES) { + // doesn't apply to subtitles + return false; + } + + // Calculate threshold to abort current download + int32_t targetDurationSecs; + CHECK(mPlaylist->meta()->findInt32("target-duration", &targetDurationSecs)); + int64_t targetDurationUs = targetDurationSecs * 1000000ll; + int64_t thresholdUs = -1; + { + AutoMutex _l(mThresholdLock); + thresholdUs = (mThresholdRatio < 0.0f) ? + -1ll : mThresholdRatio * targetDurationUs; + } + + if (thresholdUs < 0) { + // never abort + return false; + } else if (thresholdUs == 0) { + // immediately abort + return true; + } + + // now we have a positive thresholdUs, abort if remaining + // portion to download is over that threshold. + if (mSegmentFirstPTS < 0) { + // this means we haven't even find the first access unit, + // abort now as we must be very far away from the end. + return true; + } + int64_t lastEnqueueUs = mSegmentFirstPTS; + for (size_t i = 0; i < mPacketSources.size(); ++i) { + if ((mStreamTypeMask & mPacketSources.keyAt(i)) == 0) { + continue; + } + sp<AMessage> meta = mPacketSources[i]->getLatestEnqueuedMeta(); + int32_t type; + if (meta == NULL || meta->findInt32("discontinuity", &type)) { + continue; + } + int64_t tmpUs; + CHECK(meta->findInt64("timeUs", &tmpUs)); + if (tmpUs > lastEnqueueUs) { + lastEnqueueUs = tmpUs; + } + } + lastEnqueueUs -= mSegmentFirstPTS; + if (targetDurationUs - lastEnqueueUs > thresholdUs) { + return true; + } + return false; +} + +bool PlaylistFetcher::initDownloadState( + AString &uri, + sp<AMessage> &itemMeta, + int32_t &firstSeqNumberInPlaylist, + int32_t &lastSeqNumberInPlaylist) { status_t err = refreshPlaylist(); - int32_t firstSeqNumberInPlaylist = 0; - int32_t lastSeqNumberInPlaylist = 0; + firstSeqNumberInPlaylist = 0; + lastSeqNumberInPlaylist = 0; bool discontinuity = false; if (mPlaylist != NULL) { @@ -763,6 +920,8 @@ void PlaylistFetcher::onDownloadNext() { } } + mSegmentFirstPTS = -1ll; + if (mPlaylist != NULL && mSeqNumber < 0) { CHECK_GE(mStartTimeUs, 0ll); @@ -790,7 +949,8 @@ void PlaylistFetcher::onDownloadNext() { // timestamps coming from the media container) is used to determine the position // inside a segments. mSeqNumber = getSeqNumberForTime(mSegmentStartTimeUs); - if (mAdaptive) { + if (mStreamTypeMask != LiveSession::STREAMTYPE_SUBTITLES + && mSeekMode != LiveSession::kSeekModeNextSample) { // avoid double fetch/decode mSeqNumber += 1; } @@ -840,12 +1000,12 @@ void PlaylistFetcher::onDownloadNext() { mSeqNumber, firstSeqNumberInPlaylist, lastSeqNumberInPlaylist, delayUs, mNumRetries); postMonitorQueue(delayUs); - return; + return false; } if (err != OK) { notifyError(err); - return; + return false; } // we've missed the boat, let's start 3 segments prior to the latest sequence @@ -860,12 +1020,8 @@ void PlaylistFetcher::onDownloadNext() { // but since the segments we are supposed to fetch have already rolled off // the playlist, i.e. we have already missed the boat, we inevitably have to // skip. - for (size_t i = 0; i < mPacketSources.size(); i++) { - sp<ABuffer> formatChange = mSession->createFormatChangeBuffer(); - mPacketSources.valueAt(i)->queueAccessUnit(formatChange); - } - stopAsync(/* clear = */ false); - return; + notifyStopReached(); + return false; } mSeqNumber = lastSeqNumberInPlaylist - 3; if (mSeqNumber < firstSeqNumberInPlaylist) { @@ -875,20 +1031,27 @@ void PlaylistFetcher::onDownloadNext() { // fall through } else { - ALOGE("Cannot find sequence number %d in playlist " - "(contains %d - %d)", - mSeqNumber, firstSeqNumberInPlaylist, - firstSeqNumberInPlaylist + (int32_t)mPlaylist->size() - 1); + if (mPlaylist != NULL) { + ALOGE("Cannot find sequence number %d in playlist " + "(contains %d - %d)", + mSeqNumber, firstSeqNumberInPlaylist, + firstSeqNumberInPlaylist + (int32_t)mPlaylist->size() - 1); - notifyError(ERROR_END_OF_STREAM); - return; + notifyError(ERROR_END_OF_STREAM); + } else { + // It's possible that we were never able to download the playlist. + // In this case we should notify error, instead of EOS, as EOS during + // prepare means we succeeded in downloading everything. + ALOGE("Failed to download playlist!"); + notifyError(ERROR_IO); + } + + return false; } } mNumRetries = 0; - AString uri; - sp<AMessage> itemMeta; CHECK(mPlaylist->itemAt( mSeqNumber - firstSeqNumberInPlaylist, &uri, @@ -911,20 +1074,6 @@ void PlaylistFetcher::onDownloadNext() { } mLastDiscontinuitySeq = -1; - int64_t range_offset, range_length; - if (!itemMeta->findInt64("range-offset", &range_offset) - || !itemMeta->findInt64("range-length", &range_length)) { - range_offset = 0; - range_length = -1; - } - - ALOGV("fetching segment %d from (%d .. %d)", - mSeqNumber, firstSeqNumberInPlaylist, lastSeqNumberInPlaylist); - - ALOGV("fetching '%s'", uri.c_str()); - - sp<DataSource> source; - sp<ABuffer> buffer, tsBuffer; // decrypt a junk buffer to prefetch key; since a session uses only one http connection, // this avoids interleaved connections to the key and segment file. { @@ -934,7 +1083,7 @@ void PlaylistFetcher::onDownloadNext() { true /* first */); if (err != OK) { notifyError(err); - return; + return false; } } @@ -962,8 +1111,10 @@ void PlaylistFetcher::onDownloadNext() { // Signal a format discontinuity to ATSParser to clear partial data // from previous streams. Not doing this causes bitstream corruption. - mTSParser->signalDiscontinuity( - ATSParser::DISCONTINUITY_FORMATCHANGE, NULL /* extra */); + if (mTSParser != NULL) { + mTSParser->signalDiscontinuity( + ATSParser::DISCONTINUITY_FORMATCHANGE, NULL /* extra */); + } queueDiscontinuity( ATSParser::DISCONTINUITY_FORMATCHANGE, @@ -983,11 +1134,71 @@ void PlaylistFetcher::onDownloadNext() { } } + ALOGV("fetching segment %d from (%d .. %d)", + mSeqNumber, firstSeqNumberInPlaylist, lastSeqNumberInPlaylist); + return true; +} + +void PlaylistFetcher::onDownloadNext() { + AString uri; + sp<AMessage> itemMeta; + sp<ABuffer> buffer; + sp<ABuffer> tsBuffer; + int32_t firstSeqNumberInPlaylist = 0; + int32_t lastSeqNumberInPlaylist = 0; + bool connectHTTP = true; + + if (mDownloadState->hasSavedState()) { + mDownloadState->restoreState( + uri, + itemMeta, + buffer, + tsBuffer, + firstSeqNumberInPlaylist, + lastSeqNumberInPlaylist); + connectHTTP = false; + ALOGV("resuming: '%s'", uri.c_str()); + } else { + if (!initDownloadState( + uri, + itemMeta, + firstSeqNumberInPlaylist, + lastSeqNumberInPlaylist)) { + return; + } + ALOGV("fetching: '%s'", uri.c_str()); + } + + int64_t range_offset, range_length; + if (!itemMeta->findInt64("range-offset", &range_offset) + || !itemMeta->findInt64("range-length", &range_length)) { + range_offset = 0; + range_length = -1; + } + // block-wise download + bool shouldPause = false; ssize_t bytesRead; do { + sp<DataSource> source = mHTTPDataSource; + + int64_t startUs = ALooper::GetNowUs(); bytesRead = mSession->fetchFile( - uri.c_str(), &buffer, range_offset, range_length, kDownloadBlockSize, &source); + uri.c_str(), &buffer, range_offset, range_length, kDownloadBlockSize, + &source, NULL, connectHTTP); + + // add sample for bandwidth estimation, excluding samples from subtitles (as + // its too small), or during startup/resumeUntil (when we could have more than + // one connection open which affects bandwidth) + if (!mStartup && mStopParams == NULL && bytesRead > 0 + && (mStreamTypeMask + & (LiveSession::STREAMTYPE_AUDIO + | LiveSession::STREAMTYPE_VIDEO))) { + int64_t delayUs = ALooper::GetNowUs() - startUs; + mSession->addBandwidthMeasurement(bytesRead, delayUs); + } + + connectHTTP = false; if (bytesRead < 0) { status_t err = bytesRead; @@ -1013,6 +1224,8 @@ void PlaylistFetcher::onDownloadNext() { return; } + bool startUp = mStartup; // save current start up state + err = OK; if (bufferStartsWithTsSyncByte(buffer)) { // Incremental extraction is only supported for MPEG2 transport streams. @@ -1025,7 +1238,6 @@ void PlaylistFetcher::onDownloadNext() { tsBuffer->setRange(tsOff, tsSize); } tsBuffer->setRange(tsBuffer->offset(), tsBuffer->size() + bytesRead); - err = extractAndQueueAccessUnitsFromTs(tsBuffer); } @@ -1040,14 +1252,36 @@ void PlaylistFetcher::onDownloadNext() { return; } else if (err == ERROR_OUT_OF_RANGE) { // reached stopping point - stopAsync(/* clear = */ false); + notifyStopReached(); return; } else if (err != OK) { notifyError(err); return; } - - } while (bytesRead != 0 && !mStopping); + // If we're switching, post start notification + // this should only be posted when the last chunk is full processed by TSParser + if (mSeekMode != LiveSession::kSeekModeExactPosition && startUp != mStartup) { + CHECK(mStartTimeUsNotify != NULL); + mStartTimeUsNotify->post(); + mStartTimeUsNotify.clear(); + shouldPause = true; + } + if (shouldPause || shouldPauseDownload()) { + // save state and return if this is not the last chunk, + // leaving the fetcher in paused state. + if (bytesRead != 0) { + mDownloadState->saveState( + uri, + itemMeta, + buffer, + tsBuffer, + firstSeqNumberInPlaylist, + lastSeqNumberInPlaylist); + return; + } + shouldPause = true; + } + } while (bytesRead != 0); if (bufferStartsWithTsSyncByte(buffer)) { // If we don't see a stream in the program table after fetching a full ts segment @@ -1083,7 +1317,6 @@ void PlaylistFetcher::onDownloadNext() { return; } - err = OK; if (tsBuffer != NULL) { AString method; CHECK(buffer->meta()->findString("cipher-method", &method)); @@ -1097,30 +1330,40 @@ void PlaylistFetcher::onDownloadNext() { } // bulk extract non-ts files + bool startUp = mStartup; if (tsBuffer == NULL) { - err = extractAndQueueAccessUnits(buffer, itemMeta); + status_t err = extractAndQueueAccessUnits(buffer, itemMeta); if (err == -EAGAIN) { // starting sequence number too low/high postMonitorQueue(); return; } else if (err == ERROR_OUT_OF_RANGE) { // reached stopping point - stopAsync(/* clear = */false); + notifyStopReached(); + return; + } else if (err != OK) { + notifyError(err); return; } } - if (err != OK) { - notifyError(err); - return; - } - ++mSeqNumber; - postMonitorQueue(); + // if adapting, pause after found the next starting point + if (mSeekMode != LiveSession::kSeekModeExactPosition && startUp != mStartup) { + CHECK(mStartTimeUsNotify != NULL); + mStartTimeUsNotify->post(); + mStartTimeUsNotify.clear(); + shouldPause = true; + } + + if (!shouldPause) { + postMonitorQueue(); + } } -int32_t PlaylistFetcher::getSeqNumberWithAnchorTime(int64_t anchorTimeUs) const { +int32_t PlaylistFetcher::getSeqNumberWithAnchorTime( + int64_t anchorTimeUs, int64_t targetDiffUs) const { int32_t firstSeqNumberInPlaylist, lastSeqNumberInPlaylist; if (mPlaylist->meta() == NULL || !mPlaylist->meta()->findInt32("media-sequence", &firstSeqNumberInPlaylist)) { @@ -1129,7 +1372,8 @@ int32_t PlaylistFetcher::getSeqNumberWithAnchorTime(int64_t anchorTimeUs) const lastSeqNumberInPlaylist = firstSeqNumberInPlaylist + mPlaylist->size() - 1; int32_t index = mSeqNumber - firstSeqNumberInPlaylist - 1; - while (index >= 0 && anchorTimeUs > mStartTimeUs) { + // adjust anchorTimeUs to within targetDiffUs from mStartTimeUs + while (index >= 0 && anchorTimeUs - mStartTimeUs > targetDiffUs) { sp<AMessage> itemMeta; CHECK(mPlaylist->itemAt(index, NULL /* uri */, &itemMeta)); @@ -1150,28 +1394,22 @@ int32_t PlaylistFetcher::getSeqNumberWithAnchorTime(int64_t anchorTimeUs) const int32_t PlaylistFetcher::getSeqNumberForDiscontinuity(size_t discontinuitySeq) const { int32_t firstSeqNumberInPlaylist; - if (mPlaylist->meta() == NULL - || !mPlaylist->meta()->findInt32("media-sequence", &firstSeqNumberInPlaylist)) { + if (mPlaylist->meta() == NULL || !mPlaylist->meta()->findInt32( + "media-sequence", &firstSeqNumberInPlaylist)) { firstSeqNumberInPlaylist = 0; } - size_t curDiscontinuitySeq = mPlaylist->getDiscontinuitySeq(); - if (discontinuitySeq < curDiscontinuitySeq) { - return firstSeqNumberInPlaylist <= 0 ? 0 : (firstSeqNumberInPlaylist - 1); - } - size_t index = 0; while (index < mPlaylist->size()) { sp<AMessage> itemMeta; CHECK(mPlaylist->itemAt( index, NULL /* uri */, &itemMeta)); - - int64_t discontinuity; - if (itemMeta->findInt64("discontinuity", &discontinuity)) { - curDiscontinuitySeq++; - } - + size_t curDiscontinuitySeq; + CHECK(itemMeta->findInt32("discontinuity-sequence", (int32_t *)&curDiscontinuitySeq)); + int32_t seqNumber = firstSeqNumberInPlaylist + index; if (curDiscontinuitySeq == discontinuitySeq) { - return firstSeqNumberInPlaylist + index; + return seqNumber; + } else if (curDiscontinuitySeq > discontinuitySeq) { + return seqNumber <= 0 ? 0 : seqNumber - 1; } ++index; @@ -1231,6 +1469,7 @@ const sp<ABuffer> &PlaylistFetcher::setAccessUnitProperties( accessUnit->meta()->setInt32("discontinuitySeq", mDiscontinuitySeq); accessUnit->meta()->setInt64("segmentStartTimeUs", getSegmentStartTimeUs(mSeqNumber)); + accessUnit->meta()->setInt64("segmentDurationUs", getSegmentDurationUs(mSeqNumber)); return accessUnit; } @@ -1246,6 +1485,12 @@ status_t PlaylistFetcher::extractAndQueueAccessUnitsFromTs(const sp<ABuffer> &bu // ATSParser from skewing the timestamps of access units. extra->setInt64(IStreamListener::kKeyMediaTimeUs, 0); + // When adapting, signal a recent media time to the parser, + // so that PTS wrap around is handled for the new variant. + if (mStartTimeUs >= 0 && !mStartTimeUsRelative) { + extra->setInt64(IStreamListener::kKeyRecentMediaTimeUs, mStartTimeUs); + } + mTSParser->signalDiscontinuity( ATSParser::DISCONTINUITY_TIME, extra); @@ -1269,30 +1514,15 @@ status_t PlaylistFetcher::extractAndQueueAccessUnitsFromTs(const sp<ABuffer> &bu for (size_t i = mPacketSources.size(); i-- > 0;) { sp<AnotherPacketSource> packetSource = mPacketSources.valueAt(i); - const char *key; - ATSParser::SourceType type; const LiveSession::StreamType stream = mPacketSources.keyAt(i); - switch (stream) { - case LiveSession::STREAMTYPE_VIDEO: - type = ATSParser::VIDEO; - key = "timeUsVideo"; - break; - - case LiveSession::STREAMTYPE_AUDIO: - type = ATSParser::AUDIO; - key = "timeUsAudio"; - break; - - case LiveSession::STREAMTYPE_SUBTITLES: - { - ALOGE("MPEG2 Transport streams do not contain subtitles."); - return ERROR_MALFORMED; - break; - } - - default: - TRESPASS(); + if (stream == LiveSession::STREAMTYPE_SUBTITLES) { + ALOGE("MPEG2 Transport streams do not contain subtitles."); + return ERROR_MALFORMED; } + const char *key = LiveSession::getKeyForStream(stream); + ATSParser::SourceType type = + (stream == LiveSession::STREAMTYPE_AUDIO) ? + ATSParser::AUDIO : ATSParser::VIDEO; sp<AnotherPacketSource> source = static_cast<AnotherPacketSource *>( @@ -1315,97 +1545,99 @@ status_t PlaylistFetcher::extractAndQueueAccessUnitsFromTs(const sp<ABuffer> &bu int64_t timeUs; CHECK(accessUnit->meta()->findInt64("timeUs", &timeUs)); + if (mSegmentFirstPTS < 0ll) { + mSegmentFirstPTS = timeUs; + if (!mStartTimeUsRelative) { + int32_t firstSeqNumberInPlaylist; + if (mPlaylist->meta() == NULL || !mPlaylist->meta()->findInt32( + "media-sequence", &firstSeqNumberInPlaylist)) { + firstSeqNumberInPlaylist = 0; + } + + int32_t targetDurationSecs; + CHECK(mPlaylist->meta()->findInt32("target-duration", &targetDurationSecs)); + int64_t targetDurationUs = targetDurationSecs * 1000000ll; + // mStartup + // mStartup is true until we have queued a packet for all the streams + // we are fetching. We queue packets whose timestamps are greater than + // mStartTimeUs. + // mSegmentStartTimeUs >= 0 + // mSegmentStartTimeUs is non-negative when adapting or switching tracks + // mSeqNumber > firstSeqNumberInPlaylist + // don't decrement mSeqNumber if it already points to the 1st segment + // timeUs - mStartTimeUs > targetDurationUs: + // This and the 2 above conditions should only happen when adapting in a live + // stream; the old fetcher has already fetched to mStartTimeUs; the new fetcher + // would start fetching after timeUs, which should be greater than mStartTimeUs; + // the old fetcher would then continue fetching data until timeUs. We don't want + // timeUs to be too far ahead of mStartTimeUs because we want the old fetcher to + // stop as early as possible. The definition of being "too far ahead" is + // arbitrary; here we use targetDurationUs as threshold. + int64_t targetDiffUs = (mSeekMode == LiveSession::kSeekModeNextSample + ? 0 : targetDurationUs); + if (mStartup && mSegmentStartTimeUs >= 0 + && mSeqNumber > firstSeqNumberInPlaylist + && timeUs - mStartTimeUs > targetDiffUs) { + // we just guessed a starting timestamp that is too high when adapting in a + // live stream; re-adjust based on the actual timestamp extracted from the + // media segment; if we didn't move backward after the re-adjustment + // (newSeqNumber), start at least 1 segment prior. + int32_t newSeqNumber = getSeqNumberWithAnchorTime( + timeUs, targetDiffUs); + if (newSeqNumber >= mSeqNumber) { + --mSeqNumber; + } else { + mSeqNumber = newSeqNumber; + } + mStartTimeUsNotify = mNotify->dup(); + mStartTimeUsNotify->setInt32("what", kWhatStartedAt); + mStartTimeUsNotify->setString("uri", mURI); + mIDRFound = false; + return -EAGAIN; + } + } + } if (mStartup) { if (!mFirstPTSValid) { mFirstTimeUs = timeUs; mFirstPTSValid = true; } + bool startTimeReached = true; if (mStartTimeUsRelative) { timeUs -= mFirstTimeUs; if (timeUs < 0) { timeUs = 0; } + startTimeReached = (timeUs >= mStartTimeUs); } - if (timeUs < mStartTimeUs || (isAvc && !mIDRFound)) { - // buffer up to the closest preceding IDR frame - ALOGV("timeUs %" PRId64 " us < mStartTimeUs %" PRId64 " us", - timeUs, mStartTimeUs); + if (!startTimeReached || (isAvc && !mIDRFound)) { + // buffer up to the closest preceding IDR frame in the next segement, + // or the closest succeeding IDR frame after the exact position if (isAvc) { if (IsIDR(accessUnit)) { mVideoBuffer->clear(); mIDRFound = true; } - if (mIDRFound) { + if (mIDRFound && mStartTimeUsRelative && !startTimeReached) { mVideoBuffer->queueAccessUnit(accessUnit); } } - - continue; + if (!startTimeReached || (isAvc && !mIDRFound)) { + continue; + } } } if (mStartTimeUsNotify != NULL) { - int32_t firstSeqNumberInPlaylist; - if (mPlaylist->meta() == NULL || !mPlaylist->meta()->findInt32( - "media-sequence", &firstSeqNumberInPlaylist)) { - firstSeqNumberInPlaylist = 0; - } - - int32_t targetDurationSecs; - CHECK(mPlaylist->meta()->findInt32("target-duration", &targetDurationSecs)); - int64_t targetDurationUs = targetDurationSecs * 1000000ll; - // mStartup - // mStartup is true until we have queued a packet for all the streams - // we are fetching. We queue packets whose timestamps are greater than - // mStartTimeUs. - // mSegmentStartTimeUs >= 0 - // mSegmentStartTimeUs is non-negative when adapting or switching tracks - // mSeqNumber > firstSeqNumberInPlaylist - // don't decrement mSeqNumber if it already points to the 1st segment - // timeUs - mStartTimeUs > targetDurationUs: - // This and the 2 above conditions should only happen when adapting in a live - // stream; the old fetcher has already fetched to mStartTimeUs; the new fetcher - // would start fetching after timeUs, which should be greater than mStartTimeUs; - // the old fetcher would then continue fetching data until timeUs. We don't want - // timeUs to be too far ahead of mStartTimeUs because we want the old fetcher to - // stop as early as possible. The definition of being "too far ahead" is - // arbitrary; here we use targetDurationUs as threshold. - if (mStartup && mSegmentStartTimeUs >= 0 - && mSeqNumber > firstSeqNumberInPlaylist - && timeUs - mStartTimeUs > targetDurationUs) { - // we just guessed a starting timestamp that is too high when adapting in a - // live stream; re-adjust based on the actual timestamp extracted from the - // media segment; if we didn't move backward after the re-adjustment - // (newSeqNumber), start at least 1 segment prior. - int32_t newSeqNumber = getSeqNumberWithAnchorTime(timeUs); - if (newSeqNumber >= mSeqNumber) { - --mSeqNumber; - } else { - mSeqNumber = newSeqNumber; - } - mStartTimeUsNotify = mNotify->dup(); - mStartTimeUsNotify->setInt32("what", kWhatStartedAt); - return -EAGAIN; - } - - int32_t seq; - if (!mStartTimeUsNotify->findInt32("discontinuitySeq", &seq)) { - mStartTimeUsNotify->setInt32("discontinuitySeq", mDiscontinuitySeq); - } - int64_t startTimeUs; - if (!mStartTimeUsNotify->findInt64(key, &startTimeUs)) { - mStartTimeUsNotify->setInt64(key, timeUs); - - uint32_t streamMask = 0; - mStartTimeUsNotify->findInt32("streamMask", (int32_t *) &streamMask); + uint32_t streamMask = 0; + mStartTimeUsNotify->findInt32("streamMask", (int32_t *) &streamMask); + if (!(streamMask & mPacketSources.keyAt(i))) { streamMask |= mPacketSources.keyAt(i); mStartTimeUsNotify->setInt32("streamMask", streamMask); if (streamMask == mStreamTypeMask) { mStartup = false; - mStartTimeUsNotify->post(); - mStartTimeUsNotify.clear(); } } } @@ -1419,7 +1651,6 @@ status_t PlaylistFetcher::extractAndQueueAccessUnitsFromTs(const sp<ABuffer> &bu || !mStopParams->findInt64(key, &stopTimeUs) || (discontinuitySeq == mDiscontinuitySeq && timeUs >= stopTimeUs)) { - packetSource->queueAccessUnit(mSession->createFormatChangeBuffer()); mStreamTypeMask &= ~stream; mPacketSources.removeItemsAt(i); break; @@ -1666,10 +1897,13 @@ status_t PlaylistFetcher::extractAndQueueAccessUnits( CHECK(mPlaylist->meta()->findInt32("target-duration", &targetDurationSecs)); int64_t targetDurationUs = targetDurationSecs * 1000000ll; + int64_t targetDiffUs =(mSeekMode == LiveSession::kSeekModeNextSample + ? 0 : targetDurationUs); // Duplicated logic from how we handle .ts playlists. if (mStartup && mSegmentStartTimeUs >= 0 - && timeUs - mStartTimeUs > targetDurationUs) { - int32_t newSeqNumber = getSeqNumberWithAnchorTime(timeUs); + && timeUs - mStartTimeUs > targetDiffUs) { + int32_t newSeqNumber = getSeqNumberWithAnchorTime( + timeUs, targetDiffUs); if (newSeqNumber >= mSeqNumber) { --mSeqNumber; } else { @@ -1678,11 +1912,7 @@ status_t PlaylistFetcher::extractAndQueueAccessUnits( return -EAGAIN; } - mStartTimeUsNotify->setInt64("timeUsAudio", timeUs); - mStartTimeUsNotify->setInt32("discontinuitySeq", mDiscontinuitySeq); mStartTimeUsNotify->setInt32("streamMask", LiveSession::STREAMTYPE_AUDIO); - mStartTimeUsNotify->post(); - mStartTimeUsNotify.clear(); mStartup = false; } } @@ -1695,7 +1925,6 @@ status_t PlaylistFetcher::extractAndQueueAccessUnits( || discontinuitySeq > mDiscontinuitySeq || !mStopParams->findInt64("timeUsAudio", &stopTimeUs) || (discontinuitySeq == mDiscontinuitySeq && unitTimeUs >= stopTimeUs)) { - packetSource->queueAccessUnit(mSession->createFormatChangeBuffer()); mStreamTypeMask = 0; mPacketSources.clear(); return ERROR_OUT_OF_RANGE; @@ -1732,4 +1961,15 @@ void PlaylistFetcher::updateDuration() { msg->post(); } +void PlaylistFetcher::updateTargetDuration() { + int32_t targetDurationSecs; + CHECK(mPlaylist->meta()->findInt32("target-duration", &targetDurationSecs)); + int64_t targetDurationUs = targetDurationSecs * 1000000ll; + + sp<AMessage> msg = mNotify->dup(); + msg->setInt32("what", kWhatTargetDurationUpdate); + msg->setInt64("targetDurationUs", targetDurationUs); + msg->post(); +} + } // namespace android |