From 822a489e595336be447f47f5c2a051e8fdd1cdff Mon Sep 17 00:00:00 2001 From: Robert Shih Date: Thu, 6 Feb 2014 14:01:30 -0800 Subject: LiveSession: Add support for block-by-block fetchFile. Bug: 11854054 Change-Id: I4025ba7fab8fab2e0c720f73894e908fd98a43d8 --- media/libstagefright/httplive/LiveSession.cpp | 97 +++++++++++++++++---------- media/libstagefright/httplive/LiveSession.h | 18 ++++- 2 files changed, 80 insertions(+), 35 deletions(-) (limited to 'media/libstagefright/httplive') diff --git a/media/libstagefright/httplive/LiveSession.cpp b/media/libstagefright/httplive/LiveSession.cpp index bd12ddc..0333a2a 100644 --- a/media/libstagefright/httplive/LiveSession.cpp +++ b/media/libstagefright/httplive/LiveSession.cpp @@ -512,53 +512,80 @@ sp LiveSession::addFetcher(const char *uri) { return info.mFetcher; } +/* + * Illustration of parameters: + * + * 0 `range_offset` + * +------------+-------------------------------------------------------+--+--+ + * | | | next block to fetch | | | + * | | `source` handle => `out` buffer | | | | + * | `url` file |<--------- buffer size --------->|<--- `block_size` -->| | | + * | |<----------- `range_length` / buffer capacity ----------->| | + * |<------------------------------ file_size ------------------------------->| + * + * Special parameter values: + * - range_length == -1 means entire file + * - block_size == 0 means entire range + * + */ status_t LiveSession::fetchFile( const char *url, sp *out, - int64_t range_offset, int64_t range_length) { - *out = NULL; + int64_t range_offset, int64_t range_length, + uint32_t block_size, /* download block size */ + sp *source /* to return and reuse source */) { + off64_t size; + sp temp_source; + if (source == NULL) { + source = &temp_source; + } - sp source; + if (*source == NULL) { + if (!strncasecmp(url, "file://", 7)) { + *source = new FileSource(url + 7); + } else if (strncasecmp(url, "http://", 7) + && strncasecmp(url, "https://", 8)) { + return ERROR_UNSUPPORTED; + } else { + KeyedVector headers = mExtraHeaders; + if (range_offset > 0 || range_length >= 0) { + headers.add( + String8("Range"), + String8( + StringPrintf( + "bytes=%lld-%s", + range_offset, + range_length < 0 + ? "" : StringPrintf("%lld", + range_offset + range_length - 1).c_str()).c_str())); + } + status_t err = mHTTPDataSource->connect(url, &headers); - if (!strncasecmp(url, "file://", 7)) { - source = new FileSource(url + 7); - } else if (strncasecmp(url, "http://", 7) - && strncasecmp(url, "https://", 8)) { - return ERROR_UNSUPPORTED; - } else { - KeyedVector headers = mExtraHeaders; - if (range_offset > 0 || range_length >= 0) { - headers.add( - String8("Range"), - String8( - StringPrintf( - "bytes=%lld-%s", - range_offset, - range_length < 0 - ? "" : StringPrintf("%lld", range_offset + range_length - 1).c_str()).c_str())); - } - status_t err = mHTTPDataSource->connect(url, &headers); + if (err != OK) { + return err; + } - if (err != OK) { - return err; + *source = mHTTPDataSource; } - - source = mHTTPDataSource; } - off64_t size; - status_t err = source->getSize(&size); - - if (err != OK) { + status_t getSizeErr = (*source)->getSize(&size); + if (getSizeErr != OK) { size = 65536; } - sp buffer = new ABuffer(size); - buffer->setRange(0, 0); + sp buffer = *out != NULL ? *out : new ABuffer(size); + if (*out == NULL) { + buffer->setRange(0, 0); + } + // adjust range_length if only reading partial block + if (block_size > 0 && (range_length == -1 || buffer->size() + block_size < range_length)) { + range_length = buffer->size() + block_size; + } for (;;) { + // Only resize when we don't know the size. size_t bufferRemaining = buffer->capacity() - buffer->size(); - - if (bufferRemaining == 0) { + if (bufferRemaining == 0 && getSizeErr != OK) { bufferRemaining = 32768; ALOGV("increasing download buffer to %d bytes", @@ -583,7 +610,9 @@ status_t LiveSession::fetchFile( } } - ssize_t n = source->readAt( + // The DataSource is responsible for informing us of error (n < 0) or eof (n == 0) + // to help us break out of the loop. + ssize_t n = (*source)->readAt( buffer->size(), buffer->data() + buffer->size(), maxBytesToRead); diff --git a/media/libstagefright/httplive/LiveSession.h b/media/libstagefright/httplive/LiveSession.h index 99b480a8..f42d7ae 100644 --- a/media/libstagefright/httplive/LiveSession.h +++ b/media/libstagefright/httplive/LiveSession.h @@ -145,9 +145,25 @@ private: status_t onSeek(const sp &msg); void onFinishDisconnect2(); + // If given a non-zero block_size (default 0), it is used to cap the number of + // bytes read in from the DataSource. If given a non-NULL buffer, new content + // is read into the end. + // + // The DataSource we read from is responsible for signaling error or EOF to help us + // break out of the read loop. The DataSource can be returned to the caller, so + // that the caller can reuse it for subsequent fetches (within the initially + // requested range). + // + // For reused HTTP sources, the caller must download a file sequentially without + // any overlaps or gaps to prevent reconnection. status_t fetchFile( const char *url, sp *out, - int64_t range_offset = 0, int64_t range_length = -1); + /* request/open a file starting at range_offset for range_length bytes */ + int64_t range_offset = 0, int64_t range_length = -1, + /* download block size */ + uint32_t block_size = 0, + /* reuse DataSource if doing partial fetch */ + sp *source = NULL); sp fetchPlaylist( const char *url, uint8_t *curPlaylistHash, bool *unchanged); -- cgit v1.1 From 2cd94583868b775a548233a4f7cd1d988fc6344f Mon Sep 17 00:00:00 2001 From: Robert Shih Date: Thu, 6 Feb 2014 14:25:25 -0800 Subject: PlaylistFetcher: Add support for block-by-block decryption. Bug: 11854054 Change-Id: Ifd3f3369275889e716b360087b5b60d01635b578 --- media/libstagefright/httplive/PlaylistFetcher.cpp | 95 ++++++++++++++--------- media/libstagefright/httplive/PlaylistFetcher.h | 17 +++- 2 files changed, 75 insertions(+), 37 deletions(-) (limited to 'media/libstagefright/httplive') diff --git a/media/libstagefright/httplive/PlaylistFetcher.cpp b/media/libstagefright/httplive/PlaylistFetcher.cpp index f095987..9cb16fe 100644 --- a/media/libstagefright/httplive/PlaylistFetcher.cpp +++ b/media/libstagefright/httplive/PlaylistFetcher.cpp @@ -170,7 +170,8 @@ int64_t PlaylistFetcher::delayUsToRefreshPlaylist() const { } status_t PlaylistFetcher::decryptBuffer( - size_t playlistIndex, const sp &buffer) { + size_t playlistIndex, const sp &buffer, + bool first) { sp itemMeta; bool found = false; AString method; @@ -188,6 +189,7 @@ status_t PlaylistFetcher::decryptBuffer( if (!found) { method = "NONE"; } + buffer->meta()->setString("cipher-method", method.c_str()); if (method == "NONE") { return OK; @@ -227,59 +229,77 @@ status_t PlaylistFetcher::decryptBuffer( return UNKNOWN_ERROR; } - unsigned char aes_ivec[16]; + size_t n = buffer->size(); + if (!n) { + return OK; + } + CHECK(n % 16 == 0); - AString iv; - if (itemMeta->findString("cipher-iv", &iv)) { - if ((!iv.startsWith("0x") && !iv.startsWith("0X")) - || iv.size() != 16 * 2 + 2) { - ALOGE("malformed cipher IV '%s'.", iv.c_str()); - return ERROR_MALFORMED; - } + if (first) { + // If decrypting the first block in a file, read the iv from the manifest + // or derive the iv from the file's sequence number. - memset(aes_ivec, 0, sizeof(aes_ivec)); - for (size_t i = 0; i < 16; ++i) { - char c1 = tolower(iv.c_str()[2 + 2 * i]); - char c2 = tolower(iv.c_str()[3 + 2 * i]); - if (!isxdigit(c1) || !isxdigit(c2)) { + AString iv; + if (itemMeta->findString("cipher-iv", &iv)) { + if ((!iv.startsWith("0x") && !iv.startsWith("0X")) + || iv.size() != 16 * 2 + 2) { ALOGE("malformed cipher IV '%s'.", iv.c_str()); return ERROR_MALFORMED; } - uint8_t nibble1 = isdigit(c1) ? c1 - '0' : c1 - 'a' + 10; - uint8_t nibble2 = isdigit(c2) ? c2 - '0' : c2 - 'a' + 10; - aes_ivec[i] = nibble1 << 4 | nibble2; + memset(mAESInitVec, 0, sizeof(mAESInitVec)); + for (size_t i = 0; i < 16; ++i) { + char c1 = tolower(iv.c_str()[2 + 2 * i]); + char c2 = tolower(iv.c_str()[3 + 2 * i]); + if (!isxdigit(c1) || !isxdigit(c2)) { + ALOGE("malformed cipher IV '%s'.", iv.c_str()); + return ERROR_MALFORMED; + } + uint8_t nibble1 = isdigit(c1) ? c1 - '0' : c1 - 'a' + 10; + uint8_t nibble2 = isdigit(c2) ? c2 - '0' : c2 - 'a' + 10; + + mAESInitVec[i] = nibble1 << 4 | nibble2; + } + } else { + memset(mAESInitVec, 0, sizeof(mAESInitVec)); + mAESInitVec[15] = mSeqNumber & 0xff; + mAESInitVec[14] = (mSeqNumber >> 8) & 0xff; + mAESInitVec[13] = (mSeqNumber >> 16) & 0xff; + mAESInitVec[12] = (mSeqNumber >> 24) & 0xff; } - } else { - memset(aes_ivec, 0, sizeof(aes_ivec)); - aes_ivec[15] = mSeqNumber & 0xff; - aes_ivec[14] = (mSeqNumber >> 8) & 0xff; - aes_ivec[13] = (mSeqNumber >> 16) & 0xff; - aes_ivec[12] = (mSeqNumber >> 24) & 0xff; } AES_cbc_encrypt( buffer->data(), buffer->data(), buffer->size(), - &aes_key, aes_ivec, AES_DECRYPT); - - // hexdump(buffer->data(), buffer->size()); + &aes_key, mAESInitVec, AES_DECRYPT); - size_t n = buffer->size(); - CHECK_GT(n, 0u); + return OK; +} - size_t pad = buffer->data()[n - 1]; +status_t PlaylistFetcher::checkDecryptPadding(const sp &buffer) { + status_t err; + AString method; + CHECK(buffer->meta()->findString("cipher-method", &method)); + if (method == "NONE") { + return OK; + } - CHECK_GT(pad, 0u); - CHECK_LE(pad, 16u); - CHECK_GE((size_t)n, pad); - for (size_t i = 0; i < pad; ++i) { - CHECK_EQ((unsigned)buffer->data()[n - 1 - i], pad); + uint8_t padding = 0; + if (buffer->size() > 0) { + padding = buffer->data()[buffer->size() - 1]; } - n -= pad; + if (padding > 16) { + return ERROR_MALFORMED; + } - buffer->setRange(buffer->offset(), n); + for (size_t i = buffer->size() - padding; i < padding; i++) { + if (buffer->data()[i] != padding) { + return ERROR_MALFORMED; + } + } + buffer->setRange(buffer->offset(), buffer->size() - padding); return OK; } @@ -706,6 +726,9 @@ void PlaylistFetcher::onDownloadNext() { CHECK(buffer != NULL); err = decryptBuffer(mSeqNumber - firstSeqNumberInPlaylist, buffer); + if (err == OK) { + err = checkDecryptPadding(buffer); + } if (err != OK) { ALOGE("decryptBuffer failed w/ error %d", err); diff --git a/media/libstagefright/httplive/PlaylistFetcher.h b/media/libstagefright/httplive/PlaylistFetcher.h index 78dea20..ac04a77 100644 --- a/media/libstagefright/httplive/PlaylistFetcher.h +++ b/media/libstagefright/httplive/PlaylistFetcher.h @@ -119,8 +119,23 @@ private: uint64_t mFirstPTS; int64_t mAbsoluteTimeAnchorUs; + // Stores the initialization vector to decrypt the next block of cipher text, which can + // either be derived from the sequence number, read from the manifest, or copied from + // the last block of cipher text (cipher-block chaining). + unsigned char mAESInitVec[16]; + + // Set first to true if decrypting the first segment of a playlist segment. When + // first is true, reset the initialization vector based on the available + // information in the manifest; otherwise, use the initialization vector as + // updated by the last call to AES_cbc_encrypt. + // + // For the input to decrypt correctly, decryptBuffer must be called on + // consecutive byte ranges on block boundaries, e.g. 0..15, 16..47, 48..63, + // and so on. status_t decryptBuffer( - size_t playlistIndex, const sp &buffer); + size_t playlistIndex, const sp &buffer, + bool first = true); + status_t checkDecryptPadding(const sp &buffer); void postMonitorQueue(int64_t delayUs = 0, int64_t minDelayUs = 0); void cancelMonitorQueue(); -- cgit v1.1 From 6c8495c8f1ccc35db972ee7ac0dbb8baf5843548 Mon Sep 17 00:00:00 2001 From: Robert Shih Date: Fri, 14 Feb 2014 12:21:50 -0800 Subject: LiveSession refactor Bug: 11854054 Change-Id: Ia91dce109835e042f72934376d4838b4cc72cb10 --- media/libstagefright/httplive/LiveSession.cpp | 186 ++++++++++---------------- media/libstagefright/httplive/LiveSession.h | 28 +++- media/libstagefright/httplive/M3UParser.cpp | 12 -- media/libstagefright/httplive/M3UParser.h | 6 +- 4 files changed, 92 insertions(+), 140 deletions(-) (limited to 'media/libstagefright/httplive') diff --git a/media/libstagefright/httplive/LiveSession.cpp b/media/libstagefright/httplive/LiveSession.cpp index 0333a2a..79ad100 100644 --- a/media/libstagefright/httplive/LiveSession.cpp +++ b/media/libstagefright/httplive/LiveSession.cpp @@ -66,14 +66,13 @@ LiveSession::LiveSession( mHTTPDataSource->setUID(mUID); } - mPacketSources.add( - STREAMTYPE_AUDIO, new AnotherPacketSource(NULL /* meta */)); + mStreams[kAudioIndex] = StreamItem("audio"); + mStreams[kVideoIndex] = StreamItem("video"); + mStreams[kSubtitleIndex] = StreamItem("subtitle"); - mPacketSources.add( - STREAMTYPE_VIDEO, new AnotherPacketSource(NULL /* meta */)); - - mPacketSources.add( - STREAMTYPE_SUBTITLES, new AnotherPacketSource(NULL /* meta */)); + for (size_t i = 0; i < kMaxStreams; ++i) { + mPacketSources.add(indexToType(i), new AnotherPacketSource(NULL /* meta */)); + } } LiveSession::~LiveSession() { @@ -374,6 +373,12 @@ int LiveSession::SortByBandwidth(const BandwidthItem *a, const BandwidthItem *b) return 1; } +// static +LiveSession::StreamType LiveSession::indexToType(int idx) { + CHECK(idx >= 0 && idx < kMaxStreams); + return (StreamType)(1 << idx); +} + void LiveSession::onConnect(const sp &msg) { AString url; CHECK(msg->findString("url", &url)); @@ -858,19 +863,11 @@ void LiveSession::changeConfiguration( uint32_t streamMask = 0; - AString audioURI; - if (mPlaylist->getAudioURI(item.mPlaylistIndex, &audioURI)) { - streamMask |= STREAMTYPE_AUDIO; - } - - AString videoURI; - if (mPlaylist->getVideoURI(item.mPlaylistIndex, &videoURI)) { - streamMask |= STREAMTYPE_VIDEO; - } - - AString subtitleURI; - if (mPlaylist->getSubtitleURI(item.mPlaylistIndex, &subtitleURI)) { - streamMask |= STREAMTYPE_SUBTITLES; + AString URIs[kMaxStreams]; + for (size_t i = 0; i < kMaxStreams; ++i) { + if (mPlaylist->getTypeURI(item.mPlaylistIndex, mStreams[i].mType, &URIs[i])) { + streamMask |= indexToType(i); + } } // Step 1, stop and discard fetchers that are no longer needed. @@ -882,10 +879,10 @@ void LiveSession::changeConfiguration( // If we're seeking all current fetchers are discarded. if (timeUs < 0ll) { - if (((streamMask & STREAMTYPE_AUDIO) && uri == audioURI) - || ((streamMask & STREAMTYPE_VIDEO) && uri == videoURI) - || ((streamMask & STREAMTYPE_SUBTITLES) && uri == subtitleURI)) { - discardFetcher = false; + for (size_t j = 0; j < kMaxStreams; ++j) { + if ((streamMask & indexToType(j)) && uri == URIs[j]) { + discardFetcher = false; + } } } @@ -899,14 +896,10 @@ void LiveSession::changeConfiguration( sp msg = new AMessage(kWhatChangeConfiguration2, id()); msg->setInt32("streamMask", streamMask); msg->setInt64("timeUs", timeUs); - if (streamMask & STREAMTYPE_AUDIO) { - msg->setString("audioURI", audioURI.c_str()); - } - if (streamMask & STREAMTYPE_VIDEO) { - msg->setString("videoURI", videoURI.c_str()); - } - if (streamMask & STREAMTYPE_SUBTITLES) { - msg->setString("subtitleURI", subtitleURI.c_str()); + for (size_t i = 0; i < kMaxStreams; ++i) { + if (streamMask & indexToType(i)) { + msg->setString(mStreams[i].uriKey().c_str(), URIs[i].c_str()); + } } // Every time a fetcher acknowledges the stopAsync or pauseAsync request @@ -937,18 +930,13 @@ void LiveSession::onChangeConfiguration2(const sp &msg) { uint32_t streamMask; CHECK(msg->findInt32("streamMask", (int32_t *)&streamMask)); - AString audioURI, videoURI, subtitleURI; - if (streamMask & STREAMTYPE_AUDIO) { - CHECK(msg->findString("audioURI", &audioURI)); - ALOGV("audioURI = '%s'", audioURI.c_str()); - } - if (streamMask & STREAMTYPE_VIDEO) { - CHECK(msg->findString("videoURI", &videoURI)); - ALOGV("videoURI = '%s'", videoURI.c_str()); - } - if (streamMask & STREAMTYPE_SUBTITLES) { - CHECK(msg->findString("subtitleURI", &subtitleURI)); - ALOGV("subtitleURI = '%s'", subtitleURI.c_str()); + AString URIs[kMaxStreams]; + for (size_t i = 0; i < kMaxStreams; ++i) { + if (streamMask & indexToType(i)) { + const AString &uriKey = mStreams[i].uriKey(); + CHECK(msg->findString(uriKey.c_str(), &URIs[i])); + ALOGV("%s = '%s'", uriKey.c_str(), URIs[i].c_str()); + } } // Determine which decoders to shutdown on the player side, @@ -958,15 +946,12 @@ void LiveSession::onChangeConfiguration2(const sp &msg) { // 2) its streamtype was already active and still is but the URI // has changed. uint32_t changedMask = 0; - if (((mStreamMask & streamMask & STREAMTYPE_AUDIO) - && !(audioURI == mAudioURI)) - || (mStreamMask & ~streamMask & STREAMTYPE_AUDIO)) { - changedMask |= STREAMTYPE_AUDIO; - } - if (((mStreamMask & streamMask & STREAMTYPE_VIDEO) - && !(videoURI == mVideoURI)) - || (mStreamMask & ~streamMask & STREAMTYPE_VIDEO)) { - changedMask |= STREAMTYPE_VIDEO; + for (size_t i = 0; i < kMaxStreams && i != kSubtitleIndex; ++i) { + if (((mStreamMask & streamMask & indexToType(i)) + && !(URIs[i] == mStreams[i].mUri)) + || (mStreamMask & ~streamMask & indexToType(i))) { + changedMask |= indexToType(i); + } } if (changedMask == 0) { @@ -998,15 +983,10 @@ void LiveSession::onChangeConfiguration3(const sp &msg) { uint32_t streamMask; CHECK(msg->findInt32("streamMask", (int32_t *)&streamMask)); - AString audioURI, videoURI, subtitleURI; - if (streamMask & STREAMTYPE_AUDIO) { - CHECK(msg->findString("audioURI", &audioURI)); - } - if (streamMask & STREAMTYPE_VIDEO) { - CHECK(msg->findString("videoURI", &videoURI)); - } - if (streamMask & STREAMTYPE_SUBTITLES) { - CHECK(msg->findString("subtitleURI", &subtitleURI)); + for (size_t i = 0; i < kMaxStreams; ++i) { + if (streamMask & indexToType(i)) { + CHECK(msg->findString(mStreams[i].uriKey().c_str(), &mStreams[i].mUri)); + } } int64_t timeUs; @@ -1018,9 +998,6 @@ void LiveSession::onChangeConfiguration3(const sp &msg) { mRealTimeBaseUs = ALooper::GetNowUs() - timeUs; mStreamMask = streamMask; - mAudioURI = audioURI; - mVideoURI = videoURI; - mSubtitleURI = subtitleURI; // Resume all existing fetchers and assign them packet sources. for (size_t i = 0; i < mFetcherInfos.size(); ++i) { @@ -1028,22 +1005,12 @@ void LiveSession::onChangeConfiguration3(const sp &msg) { uint32_t resumeMask = 0; - sp audioSource; - if ((streamMask & STREAMTYPE_AUDIO) && uri == audioURI) { - audioSource = mPacketSources.valueFor(STREAMTYPE_AUDIO); - resumeMask |= STREAMTYPE_AUDIO; - } - - sp videoSource; - if ((streamMask & STREAMTYPE_VIDEO) && uri == videoURI) { - videoSource = mPacketSources.valueFor(STREAMTYPE_VIDEO); - resumeMask |= STREAMTYPE_VIDEO; - } - - sp subtitleSource; - if ((streamMask & STREAMTYPE_SUBTITLES) && uri == subtitleURI) { - subtitleSource = mPacketSources.valueFor(STREAMTYPE_SUBTITLES); - resumeMask |= STREAMTYPE_SUBTITLES; + sp sources[kMaxStreams]; + for (size_t j = 0; j < kMaxStreams; ++j) { + if ((streamMask & indexToType(j)) && uri == mStreams[j].mUri) { + sources[j] = mPacketSources.valueFor(indexToType(j)); + resumeMask |= indexToType(j); + } } CHECK_NE(resumeMask, 0u); @@ -1053,7 +1020,7 @@ void LiveSession::onChangeConfiguration3(const sp &msg) { streamMask &= ~resumeMask; mFetcherInfos.valueAt(i).mFetcher->startAsync( - audioSource, videoSource, subtitleSource); + sources[kAudioIndex], sources[kVideoIndex], sources[kSubtitleIndex]); } // streamMask now only contains the types that need a new fetcher created. @@ -1062,52 +1029,33 @@ void LiveSession::onChangeConfiguration3(const sp &msg) { ALOGV("creating new fetchers for mask 0x%08x", streamMask); } - while (streamMask != 0) { - StreamType streamType = (StreamType)(streamMask & ~(streamMask - 1)); + for (size_t i = 0; i < kMaxStreams; i++) { + if (!(indexToType(i) & streamMask)) { + continue; + } AString uri; - switch (streamType) { - case STREAMTYPE_AUDIO: - uri = audioURI; - break; - case STREAMTYPE_VIDEO: - uri = videoURI; - break; - case STREAMTYPE_SUBTITLES: - uri = subtitleURI; - break; - default: - TRESPASS(); - } + uri = mStreams[i].mUri; sp fetcher = addFetcher(uri.c_str()); CHECK(fetcher != NULL); - sp audioSource; - if ((streamMask & STREAMTYPE_AUDIO) && uri == audioURI) { - audioSource = mPacketSources.valueFor(STREAMTYPE_AUDIO); - audioSource->clear(); - - streamMask &= ~STREAMTYPE_AUDIO; - } - - sp videoSource; - if ((streamMask & STREAMTYPE_VIDEO) && uri == videoURI) { - videoSource = mPacketSources.valueFor(STREAMTYPE_VIDEO); - videoSource->clear(); + sp sources[kMaxStreams]; + // TRICKY: looping from i as earlier streams are already removed from streamMask + for (size_t j = i; j < kMaxStreams; ++j) { + if ((streamMask & indexToType(j)) && uri == mStreams[j].mUri) { + sources[j] = mPacketSources.valueFor(indexToType(j)); + sources[j]->clear(); - streamMask &= ~STREAMTYPE_VIDEO; - } - - sp subtitleSource; - if ((streamMask & STREAMTYPE_SUBTITLES) && uri == subtitleURI) { - subtitleSource = mPacketSources.valueFor(STREAMTYPE_SUBTITLES); - subtitleSource->clear(); - - streamMask &= ~STREAMTYPE_SUBTITLES; + streamMask &= ~indexToType(j); + } } - fetcher->startAsync(audioSource, videoSource, subtitleSource, timeUs); + fetcher->startAsync( + sources[kAudioIndex], + sources[kVideoIndex], + sources[kSubtitleIndex], + timeUs); } // All fetchers have now been started, the configuration change diff --git a/media/libstagefright/httplive/LiveSession.h b/media/libstagefright/httplive/LiveSession.h index f42d7ae..b38c518 100644 --- a/media/libstagefright/httplive/LiveSession.h +++ b/media/libstagefright/httplive/LiveSession.h @@ -42,10 +42,17 @@ struct LiveSession : public AHandler { const sp ¬ify, uint32_t flags = 0, bool uidValid = false, uid_t uid = 0); + enum StreamIndex { + kAudioIndex = 0, + kVideoIndex = 1, + kSubtitleIndex = 2, + kMaxStreams = 3, + }; + enum StreamType { - STREAMTYPE_AUDIO = 1, - STREAMTYPE_VIDEO = 2, - STREAMTYPE_SUBTITLES = 4, + STREAMTYPE_AUDIO = 1 << kAudioIndex, + STREAMTYPE_VIDEO = 1 << kVideoIndex, + STREAMTYPE_SUBTITLES = 1 << kSubtitleIndex, }; status_t dequeueAccessUnit(StreamType stream, sp *accessUnit); @@ -105,6 +112,19 @@ private: bool mIsPrepared; }; + struct StreamItem { + const char *mType; + AString mUri; + StreamItem() : mType("") {} + StreamItem(const char *type) : mType(type) {} + AString uriKey() { + AString key(mType); + key.append("URI"); + return key; + } + }; + StreamItem mStreams[kMaxStreams]; + sp mNotify; uint32_t mFlags; bool mUIDValid; @@ -123,7 +143,6 @@ private: sp mPlaylist; KeyedVector mFetcherInfos; - AString mAudioURI, mVideoURI, mSubtitleURI; uint32_t mStreamMask; KeyedVector > mPacketSources; @@ -171,6 +190,7 @@ private: size_t getBandwidthIndex(); static int SortByBandwidth(const BandwidthItem *, const BandwidthItem *); + static StreamType indexToType(int idx); void changeConfiguration( int64_t timeUs, size_t bandwidthIndex, bool pickTrack = false); diff --git a/media/libstagefright/httplive/M3UParser.cpp b/media/libstagefright/httplive/M3UParser.cpp index ae19ffa..a9184b5 100644 --- a/media/libstagefright/httplive/M3UParser.cpp +++ b/media/libstagefright/httplive/M3UParser.cpp @@ -388,18 +388,6 @@ bool M3UParser::getTypeURI(size_t index, const char *key, AString *uri) const { return true; } -bool M3UParser::getAudioURI(size_t index, AString *uri) const { - return getTypeURI(index, "audio", uri); -} - -bool M3UParser::getVideoURI(size_t index, AString *uri) const { - return getTypeURI(index, "video", uri); -} - -bool M3UParser::getSubtitleURI(size_t index, AString *uri) const { - return getTypeURI(index, "subtitles", uri); -} - static bool MakeURL(const char *baseURL, const char *url, AString *out) { out->clear(); diff --git a/media/libstagefright/httplive/M3UParser.h b/media/libstagefright/httplive/M3UParser.h index b93b0e5..ccd6556 100644 --- a/media/libstagefright/httplive/M3UParser.h +++ b/media/libstagefright/httplive/M3UParser.h @@ -45,9 +45,7 @@ struct M3UParser : public RefBase { status_t getTrackInfo(Parcel* reply) const; ssize_t getSelectedIndex() const; - bool getAudioURI(size_t index, AString *uri) const; - bool getVideoURI(size_t index, AString *uri) const; - bool getSubtitleURI(size_t index, AString *uri) const; + bool getTypeURI(size_t index, const char *key, AString *uri) const; protected: virtual ~M3UParser(); @@ -95,8 +93,6 @@ private: status_t parseMedia(const AString &line); - bool getTypeURI(size_t index, const char *key, AString *uri) const; - static status_t ParseInt32(const char *s, int32_t *x); static status_t ParseDouble(const char *s, double *x); -- cgit v1.1 From 777ee5ed736c8f6c3f7d196ea022f7432bfd23e1 Mon Sep 17 00:00:00 2001 From: Robert Shih Date: Thu, 20 Feb 2014 13:07:26 -0800 Subject: Initial HLS seamless switch implementation. Bug: 11854054 Change-Id: I75fc2a258111295039ac13cc37e407df25891dd2 --- media/libstagefright/httplive/LiveSession.cpp | 280 +++++++++++++++++++--- media/libstagefright/httplive/LiveSession.h | 36 +++ media/libstagefright/httplive/PlaylistFetcher.cpp | 251 +++++++++++++++++-- media/libstagefright/httplive/PlaylistFetcher.h | 22 +- 4 files changed, 541 insertions(+), 48 deletions(-) (limited to 'media/libstagefright/httplive') diff --git a/media/libstagefright/httplive/LiveSession.cpp b/media/libstagefright/httplive/LiveSession.cpp index 79ad100..9e30ebd 100644 --- a/media/libstagefright/httplive/LiveSession.cpp +++ b/media/libstagefright/httplive/LiveSession.cpp @@ -37,6 +37,8 @@ #include #include +#include + #include #include #include @@ -57,10 +59,14 @@ LiveSession::LiveSession( : 0)), mPrevBandwidthIndex(-1), mStreamMask(0), + mNewStreamMask(0), + mSwapMask(0), mCheckBandwidthGeneration(0), + mSwitchGeneration(0), mLastDequeuedTimeUs(0ll), mRealTimeBaseUs(0ll), mReconfigurationInProgress(false), + mSwitchInProgress(false), mDisconnectReplyID(0) { if (mUIDValid) { mHTTPDataSource->setUID(mUID); @@ -72,16 +78,37 @@ LiveSession::LiveSession( for (size_t i = 0; i < kMaxStreams; ++i) { mPacketSources.add(indexToType(i), new AnotherPacketSource(NULL /* meta */)); + mPacketSources2.add(indexToType(i), new AnotherPacketSource(NULL /* meta */)); } } LiveSession::~LiveSession() { } +sp LiveSession::createFormatChangeBuffer(bool swap) { + ABuffer *discontinuity = new ABuffer(0); + discontinuity->meta()->setInt32("discontinuity", ATSParser::DISCONTINUITY_FORMATCHANGE); + discontinuity->meta()->setInt32("swapPacketSource", swap); + discontinuity->meta()->setInt32("switchGeneration", mSwitchGeneration); + discontinuity->meta()->setInt64("timeUs", -1); + return discontinuity; +} + +void LiveSession::swapPacketSource(StreamType stream) { + sp &aps = mPacketSources.editValueFor(stream); + sp &aps2 = mPacketSources2.editValueFor(stream); + sp tmp = aps; + aps = aps2; + aps2 = tmp; + aps2->clear(); +} + status_t LiveSession::dequeueAccessUnit( StreamType stream, sp *accessUnit) { if (!(mStreamMask & stream)) { - return UNKNOWN_ERROR; + // return -EWOULDBLOCK to avoid halting the decoder + // when switching between audio/video and audio only. + return -EWOULDBLOCK; } sp packetSource = mPacketSources.valueFor(stream); @@ -121,6 +148,25 @@ status_t LiveSession::dequeueAccessUnit( streamStr, type, extra == NULL ? "NULL" : extra->debugString().c_str()); + + int32_t swap; + if (type == ATSParser::DISCONTINUITY_FORMATCHANGE + && (*accessUnit)->meta()->findInt32("swapPacketSource", &swap) + && swap) { + + int32_t switchGeneration; + CHECK((*accessUnit)->meta()->findInt32("switchGeneration", &switchGeneration)); + { + Mutex::Autolock lock(mSwapMutex); + if (switchGeneration == mSwitchGeneration) { + swapPacketSource(stream); + sp msg = new AMessage(kWhatSwapped, id()); + msg->setInt32("stream", stream); + msg->setInt32("switchGeneration", switchGeneration); + msg->post(); + } + } + } } else if (err == OK) { if (stream == STREAMTYPE_AUDIO || stream == STREAMTYPE_VIDEO) { int64_t timeUs; @@ -142,6 +188,7 @@ status_t LiveSession::dequeueAccessUnit( } status_t LiveSession::getStreamFormat(StreamType stream, sp *format) { + // No swapPacketSource race condition; called from the same thread as dequeueAccessUnit. if (!(mStreamMask & stream)) { return UNKNOWN_ERROR; } @@ -238,7 +285,12 @@ void LiveSession::onMessageReceived(const sp &msg) { if (what == PlaylistFetcher::kWhatStopped) { AString uri; CHECK(msg->findString("uri", &uri)); - mFetcherInfos.removeItem(uri); + if (mFetcherInfos.removeItem(uri) < 0) { + // ignore duplicated kWhatStopped messages. + break; + } + + tryToFinishBandwidthSwitch(); } if (mContinuation != NULL) { @@ -274,6 +326,8 @@ void LiveSession::onMessageReceived(const sp &msg) { postPrepared(err); } + cancelBandwidthSwitch(); + mPacketSources.valueFor(STREAMTYPE_AUDIO)->signalEOS(err); mPacketSources.valueFor(STREAMTYPE_VIDEO)->signalEOS(err); @@ -312,6 +366,27 @@ void LiveSession::onMessageReceived(const sp &msg) { break; } + case PlaylistFetcher::kWhatStartedAt: + { + int32_t switchGeneration; + CHECK(msg->findInt32("switchGeneration", &switchGeneration)); + + if (switchGeneration != mSwitchGeneration) { + break; + } + + // Resume fetcher for the original variant; the resumed fetcher should + // continue until the timestamps found in msg, which is stored by the + // new fetcher to indicate where the new variant has started buffering. + for (size_t i = 0; i < mFetcherInfos.size(); i++) { + const FetcherInfo info = mFetcherInfos.valueAt(i); + if (info.mToBeRemoved) { + info.mFetcher->resumeUntilAsync(msg); + } + } + break; + } + default: TRESPASS(); } @@ -356,6 +431,11 @@ void LiveSession::onMessageReceived(const sp &msg) { break; } + case kWhatSwapped: + { + onSwapped(msg); + break; + } default: TRESPASS(); break; @@ -466,6 +546,10 @@ void LiveSession::finishDisconnect() { // during disconnection either. cancelCheckBandwidthEvent(); + // Protect mPacketSources from a swapPacketSource race condition through disconnect. + // (finishDisconnect, onFinishDisconnect2) + cancelBandwidthSwitch(); + for (size_t i = 0; i < mFetcherInfos.size(); ++i) { mFetcherInfos.valueAt(i).mFetcher->stopAsync(); } @@ -505,11 +589,13 @@ sp LiveSession::addFetcher(const char *uri) { sp notify = new AMessage(kWhatFetcherNotify, id()); notify->setString("uri", uri); + notify->setInt32("switchGeneration", mSwitchGeneration); FetcherInfo info; info.mFetcher = new PlaylistFetcher(notify, this, uri); info.mDurationUs = -1ll; info.mIsPrepared = false; + info.mToBeRemoved = false; looper()->registerHandler(info.mFetcher); mFetcherInfos.add(uri, info); @@ -844,8 +930,25 @@ status_t LiveSession::selectTrack(size_t index, bool select) { return err; } +bool LiveSession::canSwitchUp() { + // Allow upwards bandwidth switch when a stream has buffered at least 10 seconds. + status_t err = OK; + for (size_t i = 0; i < mPacketSources.size(); ++i) { + sp source = mPacketSources.valueAt(i); + int64_t dur = source->getBufferedDurationUs(&err); + if (err == OK && dur > 10000000) { + return true; + } + } + return false; +} + void LiveSession::changeConfiguration( int64_t timeUs, size_t bandwidthIndex, bool pickTrack) { + // Protect mPacketSources from a swapPacketSource race condition through reconfiguration. + // (changeConfiguration, onChangeConfiguration2, onChangeConfiguration3). + cancelBandwidthSwitch(); + CHECK(!mReconfigurationInProgress); mReconfigurationInProgress = true; @@ -861,7 +964,8 @@ void LiveSession::changeConfiguration( CHECK_LT(bandwidthIndex, mBandwidthItems.size()); const BandwidthItem &item = mBandwidthItems.itemAt(bandwidthIndex); - uint32_t streamMask = 0; + uint32_t streamMask = 0; // streams that should be fetched by the new fetcher + uint32_t resumeMask = 0; // streams that should be fetched by the original fetcher AString URIs[kMaxStreams]; for (size_t i = 0; i < kMaxStreams; ++i) { @@ -879,9 +983,14 @@ void LiveSession::changeConfiguration( // If we're seeking all current fetchers are discarded. if (timeUs < 0ll) { + // delay fetcher removal + discardFetcher = false; + for (size_t j = 0; j < kMaxStreams; ++j) { - if ((streamMask & indexToType(j)) && uri == URIs[j]) { - discardFetcher = false; + StreamType type = indexToType(j); + if ((streamMask & type) && uri == URIs[j]) { + resumeMask |= type; + streamMask &= ~type; } } } @@ -893,8 +1002,15 @@ void LiveSession::changeConfiguration( } } - sp msg = new AMessage(kWhatChangeConfiguration2, id()); + sp msg; + if (timeUs < 0ll) { + // skip onChangeConfiguration2 (decoder destruction) if switching. + msg = new AMessage(kWhatChangeConfiguration3, id()); + } else { + msg = new AMessage(kWhatChangeConfiguration2, id()); + } msg->setInt32("streamMask", streamMask); + msg->setInt32("resumeMask", resumeMask); msg->setInt64("timeUs", timeUs); for (size_t i = 0; i < kMaxStreams; ++i) { if (streamMask & indexToType(i)) { @@ -977,11 +1093,13 @@ void LiveSession::onChangeConfiguration2(const sp &msg) { } void LiveSession::onChangeConfiguration3(const sp &msg) { + mContinuation.clear(); // All remaining fetchers are still suspended, the player has shutdown // any decoders that needed it. - uint32_t streamMask; + uint32_t streamMask, resumeMask; CHECK(msg->findInt32("streamMask", (int32_t *)&streamMask)); + CHECK(msg->findInt32("resumeMask", (int32_t *)&resumeMask)); for (size_t i = 0; i < kMaxStreams; ++i) { if (streamMask & indexToType(i)) { @@ -990,37 +1108,39 @@ void LiveSession::onChangeConfiguration3(const sp &msg) { } int64_t timeUs; + bool switching = false; CHECK(msg->findInt64("timeUs", &timeUs)); if (timeUs < 0ll) { timeUs = mLastDequeuedTimeUs; + switching = true; } mRealTimeBaseUs = ALooper::GetNowUs() - timeUs; - mStreamMask = streamMask; + mNewStreamMask = streamMask; - // Resume all existing fetchers and assign them packet sources. + // Of all existing fetchers: + // * Resume fetchers that are still needed and assign them original packet sources. + // * Mark otherwise unneeded fetchers for removal. + ALOGV("resuming fetchers for mask 0x%08x", resumeMask); for (size_t i = 0; i < mFetcherInfos.size(); ++i) { const AString &uri = mFetcherInfos.keyAt(i); - uint32_t resumeMask = 0; - sp sources[kMaxStreams]; for (size_t j = 0; j < kMaxStreams; ++j) { - if ((streamMask & indexToType(j)) && uri == mStreams[j].mUri) { + if ((resumeMask & indexToType(j)) && uri == mStreams[j].mUri) { sources[j] = mPacketSources.valueFor(indexToType(j)); - resumeMask |= indexToType(j); } } - CHECK_NE(resumeMask, 0u); - - ALOGV("resuming fetchers for mask 0x%08x", resumeMask); - - streamMask &= ~resumeMask; - - mFetcherInfos.valueAt(i).mFetcher->startAsync( - sources[kAudioIndex], sources[kVideoIndex], sources[kSubtitleIndex]); + FetcherInfo &info = mFetcherInfos.editValueAt(i); + if (sources[kAudioIndex] != NULL || sources[kVideoIndex] != NULL + || sources[kSubtitleIndex] != NULL) { + info.mFetcher->startAsync( + sources[kAudioIndex], sources[kVideoIndex], sources[kSubtitleIndex]); + } else { + info.mToBeRemoved = true; + } } // streamMask now only contains the types that need a new fetcher created. @@ -1029,6 +1149,8 @@ void LiveSession::onChangeConfiguration3(const sp &msg) { ALOGV("creating new fetchers for mask 0x%08x", streamMask); } + // Find out when the original fetchers have buffered up to and start the new fetchers + // at a later timestamp. for (size_t i = 0; i < kMaxStreams; i++) { if (!(indexToType(i) & streamMask)) { continue; @@ -1040,12 +1162,40 @@ void LiveSession::onChangeConfiguration3(const sp &msg) { sp fetcher = addFetcher(uri.c_str()); CHECK(fetcher != NULL); + int32_t latestSeq = -1; + int64_t latestTimeUs = 0ll; sp sources[kMaxStreams]; + // TRICKY: looping from i as earlier streams are already removed from streamMask for (size_t j = i; j < kMaxStreams; ++j) { if ((streamMask & indexToType(j)) && uri == mStreams[j].mUri) { sources[j] = mPacketSources.valueFor(indexToType(j)); - sources[j]->clear(); + + if (!switching) { + sources[j]->clear(); + } else { + int32_t type, seq; + int64_t srcTimeUs; + sp meta = sources[j]->getLatestMeta(); + + if (meta != NULL && !meta->findInt32("discontinuity", &type)) { + CHECK(meta->findInt32("seq", &seq)); + if (seq > latestSeq) { + latestSeq = seq; + } + CHECK(meta->findInt64("timeUs", &srcTimeUs)); + if (srcTimeUs > latestTimeUs) { + latestTimeUs = srcTimeUs; + } + } + + sources[j] = mPacketSources2.valueFor(indexToType(j)); + sources[j]->clear(); + uint32_t extraStreams = mNewStreamMask & (~mStreamMask); + if (extraStreams & indexToType(j)) { + sources[j]->queueAccessUnit(createFormatChangeBuffer(/* swap = */ false)); + } + } streamMask &= ~indexToType(j); } @@ -1055,7 +1205,9 @@ void LiveSession::onChangeConfiguration3(const sp &msg) { sources[kAudioIndex], sources[kVideoIndex], sources[kSubtitleIndex], - timeUs); + timeUs, + latestTimeUs /* min start time(us) */, + latestSeq >= 0 ? latestSeq + 1 : -1 /* starting sequence number hint */ ); } // All fetchers have now been started, the configuration change @@ -1064,14 +1216,61 @@ void LiveSession::onChangeConfiguration3(const sp &msg) { scheduleCheckBandwidthEvent(); ALOGV("XXX configuration change completed."); - mReconfigurationInProgress = false; + if (switching) { + mSwitchInProgress = true; + } else { + mStreamMask = mNewStreamMask; + } if (mDisconnectReplyID != 0) { finishDisconnect(); } } +void LiveSession::onSwapped(const sp &msg) { + int32_t switchGeneration; + CHECK(msg->findInt32("switchGeneration", &switchGeneration)); + if (switchGeneration != mSwitchGeneration) { + return; + } + + int32_t stream; + CHECK(msg->findInt32("stream", &stream)); + mSwapMask |= stream; + if (mSwapMask != mStreamMask) { + return; + } + + // Check if new variant contains extra streams. + uint32_t extraStreams = mNewStreamMask & (~mStreamMask); + while (extraStreams) { + StreamType extraStream = (StreamType) (extraStreams & ~(extraStreams - 1)); + swapPacketSource(extraStream); + extraStreams &= ~extraStream; + } + + tryToFinishBandwidthSwitch(); +} + +// Mark switch done when: +// 1. all old buffers are swapped out, AND +// 2. all old fetchers are removed. +void LiveSession::tryToFinishBandwidthSwitch() { + bool needToRemoveFetchers = false; + for (size_t i = 0; i < mFetcherInfos.size(); ++i) { + if (mFetcherInfos.valueAt(i).mToBeRemoved) { + needToRemoveFetchers = true; + break; + } + } + if (!needToRemoveFetchers && mSwapMask == mStreamMask) { + mStreamMask = mNewStreamMask; + mSwitchInProgress = false; + mSwapMask = 0; + } +} + void LiveSession::scheduleCheckBandwidthEvent() { sp msg = new AMessage(kWhatCheckBandwidth, id()); msg->setInt32("generation", mCheckBandwidthGeneration); @@ -1082,16 +1281,37 @@ void LiveSession::cancelCheckBandwidthEvent() { ++mCheckBandwidthGeneration; } -void LiveSession::onCheckBandwidth() { - if (mReconfigurationInProgress) { - scheduleCheckBandwidthEvent(); - return; +void LiveSession::cancelBandwidthSwitch() { + Mutex::Autolock lock(mSwapMutex); + mSwitchGeneration++; + mSwitchInProgress = false; + mSwapMask = 0; +} + +bool LiveSession::canSwitchBandwidthTo(size_t bandwidthIndex) { + if (mReconfigurationInProgress || mSwitchInProgress) { + return false; + } + + if (mPrevBandwidthIndex < 0) { + return true; } + if (bandwidthIndex == (size_t)mPrevBandwidthIndex) { + return false; + } else if (bandwidthIndex > (size_t)mPrevBandwidthIndex) { + return canSwitchUp(); + } else { + return true; + } +} + +void LiveSession::onCheckBandwidth() { size_t bandwidthIndex = getBandwidthIndex(); - if (mPrevBandwidthIndex < 0 - || bandwidthIndex != (size_t)mPrevBandwidthIndex) { + if (canSwitchBandwidthTo(bandwidthIndex)) { changeConfiguration(-1ll /* timeUs */, bandwidthIndex); + } else { + scheduleCheckBandwidthEvent(); } // Handling the kWhatCheckBandwidth even here does _not_ automatically diff --git a/media/libstagefright/httplive/LiveSession.h b/media/libstagefright/httplive/LiveSession.h index b38c518..e26d024 100644 --- a/media/libstagefright/httplive/LiveSession.h +++ b/media/libstagefright/httplive/LiveSession.h @@ -81,6 +81,11 @@ struct LiveSession : public AHandler { kWhatPreparationFailed, }; + // create a format-change discontinuity + // + // swap: + // whether is format-change discontinuity should trigger a buffer swap + sp createFormatChangeBuffer(bool swap = true); protected: virtual ~LiveSession(); @@ -99,6 +104,7 @@ private: kWhatChangeConfiguration2 = 'chC2', kWhatChangeConfiguration3 = 'chC3', kWhatFinishDisconnect2 = 'fin2', + kWhatSwapped = 'swap', }; struct BandwidthItem { @@ -110,6 +116,7 @@ private: sp mFetcher; int64_t mDurationUs; bool mIsPrepared; + bool mToBeRemoved; }; struct StreamItem { @@ -145,9 +152,26 @@ private: KeyedVector mFetcherInfos; uint32_t mStreamMask; + // Masks used during reconfiguration: + // mNewStreamMask: streams in the variant playlist we're switching to; + // we don't want to immediately overwrite the original value. + uint32_t mNewStreamMask; + + // mSwapMask: streams that have started to playback content in the new variant playlist; + // we use this to track reconfiguration progress. + uint32_t mSwapMask; + KeyedVector > mPacketSources; + // A second set of packet sources that buffer content for the variant we're switching to. + KeyedVector > mPacketSources2; + + // A mutex used to serialize two sets of events: + // * the swapping of packet sources in dequeueAccessUnit on the player thread, AND + // * a forced bandwidth switch termination in cancelSwitch on the live looper. + Mutex mSwapMutex; int32_t mCheckBandwidthGeneration; + int32_t mSwitchGeneration; size_t mContinuationCounter; sp mContinuation; @@ -156,6 +180,7 @@ private: int64_t mRealTimeBaseUs; bool mReconfigurationInProgress; + bool mSwitchInProgress; uint32_t mDisconnectReplyID; sp addFetcher(const char *uri); @@ -197,16 +222,27 @@ private: void onChangeConfiguration(const sp &msg); void onChangeConfiguration2(const sp &msg); void onChangeConfiguration3(const sp &msg); + void onSwapped(const sp &msg); + void tryToFinishBandwidthSwitch(); void scheduleCheckBandwidthEvent(); void cancelCheckBandwidthEvent(); + // cancelBandwidthSwitch is atomic wrt swapPacketSource; call it to prevent packet sources + // from being swapped out on stale discontinuities while manipulating + // mPacketSources/mPacketSources2. + void cancelBandwidthSwitch(); + + bool canSwitchBandwidthTo(size_t bandwidthIndex); void onCheckBandwidth(); void finishDisconnect(); void postPrepared(status_t err); + void swapPacketSource(StreamType stream); + bool canSwitchUp(); + DISALLOW_EVIL_CONSTRUCTORS(LiveSession); }; diff --git a/media/libstagefright/httplive/PlaylistFetcher.cpp b/media/libstagefright/httplive/PlaylistFetcher.cpp index 9cb16fe..0eac8b3 100644 --- a/media/libstagefright/httplive/PlaylistFetcher.cpp +++ b/media/libstagefright/httplive/PlaylistFetcher.cpp @@ -48,16 +48,20 @@ namespace android { // static const int64_t PlaylistFetcher::kMinBufferedDurationUs = 10000000ll; const int64_t PlaylistFetcher::kMaxMonitorDelayUs = 3000000ll; +const int32_t PlaylistFetcher::kNumSkipFrames = 10; PlaylistFetcher::PlaylistFetcher( const sp ¬ify, const sp &session, const char *uri) : mNotify(notify), + mStartTimeUsNotify(notify->dup()), mSession(session), mURI(uri), mStreamTypeMask(0), mStartTimeUs(-1ll), + mMinStartTimeUs(0ll), + mStopParams(NULL), mLastPlaylistFetchTimeUs(-1ll), mSeqNumber(-1), mNumRetries(0), @@ -69,6 +73,8 @@ PlaylistFetcher::PlaylistFetcher( mFirstPTSValid(false), mAbsoluteTimeAnchorUs(0ll) { memset(mPlaylistHash, 0, sizeof(mPlaylistHash)); + mStartTimeUsNotify->setInt32("what", kWhatStartedAt); + mStartTimeUsNotify->setInt32("streamMask", 0); } PlaylistFetcher::~PlaylistFetcher() { @@ -325,7 +331,9 @@ void PlaylistFetcher::startAsync( const sp &audioSource, const sp &videoSource, const sp &subtitleSource, - int64_t startTimeUs) { + int64_t startTimeUs, + int64_t minStartTimeUs, + int32_t startSeqNumberHint) { sp msg = new AMessage(kWhatStart, id()); uint32_t streamTypeMask = 0ul; @@ -347,6 +355,8 @@ void PlaylistFetcher::startAsync( msg->setInt32("streamTypeMask", streamTypeMask); msg->setInt64("startTimeUs", startTimeUs); + msg->setInt64("minStartTimeUs", minStartTimeUs); + msg->setInt32("startSeqNumberHint", startSeqNumberHint); msg->post(); } @@ -358,6 +368,12 @@ void PlaylistFetcher::stopAsync() { (new AMessage(kWhatStop, id()))->post(); } +void PlaylistFetcher::resumeUntilAsync(const sp ¶ms) { + AMessage* msg = new AMessage(kWhatResumeUntil, id()); + msg->setMessage("params", params); + msg->post(); +} + void PlaylistFetcher::onMessageReceived(const sp &msg) { switch (msg->what()) { case kWhatStart: @@ -392,6 +408,7 @@ void PlaylistFetcher::onMessageReceived(const sp &msg) { } case kWhatMonitorQueue: + case kWhatDownloadNext: { int32_t generation; CHECK(msg->findInt32("generation", &generation)); @@ -401,7 +418,17 @@ void PlaylistFetcher::onMessageReceived(const sp &msg) { break; } - onMonitorQueue(); + if (msg->what() == kWhatMonitorQueue) { + onMonitorQueue(); + } else { + onDownloadNext(); + } + break; + } + + case kWhatResumeUntil: + { + onResumeUntil(msg); break; } @@ -417,7 +444,10 @@ status_t PlaylistFetcher::onStart(const sp &msg) { CHECK(msg->findInt32("streamTypeMask", (int32_t *)&streamTypeMask)); int64_t startTimeUs; + int32_t startSeqNumberHint; CHECK(msg->findInt64("startTimeUs", &startTimeUs)); + CHECK(msg->findInt64("minStartTimeUs", (int64_t *) &mMinStartTimeUs)); + CHECK(msg->findInt32("startSeqNumberHint", &startSeqNumberHint)); if (streamTypeMask & LiveSession::STREAMTYPE_AUDIO) { void *ptr; @@ -455,6 +485,10 @@ status_t PlaylistFetcher::onStart(const sp &msg) { mPrepared = false; } + if (startSeqNumberHint >= 0) { + mSeqNumber = startSeqNumberHint; + } + postMonitorQueue(); return OK; @@ -462,20 +496,70 @@ status_t PlaylistFetcher::onStart(const sp &msg) { void PlaylistFetcher::onPause() { cancelMonitorQueue(); +} + +void PlaylistFetcher::onStop() { + cancelMonitorQueue(); mPacketSources.clear(); mStreamTypeMask = 0; } -void PlaylistFetcher::onStop() { - cancelMonitorQueue(); +// Resume until we have reached the boundary timestamps listed in `msg`; when +// the remaining time is too short (within a resume threshold) stop immediately +// instead. +status_t PlaylistFetcher::onResumeUntil(const sp &msg) { + sp params; + CHECK(msg->findMessage("params", ¶ms)); + + bool stop = false; + for (size_t i = 0; i < mPacketSources.size(); i++) { + sp packetSource = mPacketSources.valueAt(i); + + const char *stopKey; + int streamType = mPacketSources.keyAt(i); + switch (streamType) { + case LiveSession::STREAMTYPE_VIDEO: + stopKey = "timeUsVideo"; + break; - for (size_t i = 0; i < mPacketSources.size(); ++i) { - mPacketSources.valueAt(i)->clear(); + case LiveSession::STREAMTYPE_AUDIO: + stopKey = "timeUsAudio"; + break; + + case LiveSession::STREAMTYPE_SUBTITLES: + stopKey = "timeUsSubtitle"; + break; + + default: + TRESPASS(); + } + + // Don't resume if we would stop within a resume threshold. + int64_t latestTimeUs = 0, stopTimeUs = 0; + sp latestMeta = packetSource->getLatestMeta(); + if (latestMeta != NULL + && (latestMeta->findInt64("timeUs", &latestTimeUs) + && params->findInt64(stopKey, &stopTimeUs))) { + int64_t diffUs = stopTimeUs - latestTimeUs; + if (diffUs < resumeThreshold(latestMeta)) { + stop = true; + } + } } - mPacketSources.clear(); - mStreamTypeMask = 0; + if (stop) { + for (size_t i = 0; i < mPacketSources.size(); i++) { + mPacketSources.valueAt(i)->queueAccessUnit(mSession->createFormatChangeBuffer()); + } + stopAsync(); + return OK; + } + + mStopParams = params; + postMonitorQueue(); + + return OK; } void PlaylistFetcher::notifyError(status_t err) { @@ -519,8 +603,9 @@ void PlaylistFetcher::onMonitorQueue() { packetSource->getBufferedDurationUs(&finalResult); finalResult = OK; } else { - bool first = true; - + // 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. for (size_t i = 0; i < mPacketSources.size(); ++i) { if ((mStreamTypeMask & mPacketSources.keyAt(i)) == 0) { continue; @@ -528,9 +613,10 @@ void PlaylistFetcher::onMonitorQueue() { int64_t bufferedStreamDurationUs = mPacketSources.valueAt(i)->getBufferedDurationUs(&finalResult); - if (first || bufferedStreamDurationUs < bufferedDurationUs) { + ALOGV("buffered %lld for stream %d", + bufferedStreamDurationUs, mPacketSources.keyAt(i)); + if (bufferedStreamDurationUs > bufferedDurationUs) { bufferedDurationUs = bufferedStreamDurationUs; - first = false; } } } @@ -550,7 +636,12 @@ void PlaylistFetcher::onMonitorQueue() { if (finalResult == OK && downloadMore) { ALOGV("monitoring, buffered=%lld < %lld", bufferedDurationUs, durationToBufferUs); - onDownloadNext(); + // delay the next download slightly; hopefully this gives other concurrent fetchers + // a better chance to run. + // onDownloadNext(); + sp msg = new AMessage(kWhatDownloadNext, id()); + msg->setInt32("generation", mMonitorQueueGeneration); + msg->post(1000l); } else { // Nothing to do yet, try again in a second. @@ -617,6 +708,12 @@ void PlaylistFetcher::onDownloadNext() { const int32_t lastSeqNumberInPlaylist = firstSeqNumberInPlaylist + (int32_t)mPlaylist->size() - 1; + if (mStartup && mSeqNumber >= 0 + && (mSeqNumber < firstSeqNumberInPlaylist || mSeqNumber > lastSeqNumberInPlaylist)) { + // in case we guessed wrong during reconfiguration, try fetching the latest content. + mSeqNumber = lastSeqNumberInPlaylist; + } + if (mSeqNumber < 0) { CHECK_GE(mStartTimeUs, 0ll); @@ -761,6 +858,18 @@ void PlaylistFetcher::onDownloadNext() { err = extractAndQueueAccessUnits(buffer, itemMeta); + if (err == -EAGAIN) { + // bad starting sequence number hint + postMonitorQueue(); + return; + } + + if (err == ERROR_OUT_OF_RANGE) { + // reached stopping point + stopAsync(); + return; + } + if (err != OK) { notifyError(err); return; @@ -817,12 +926,15 @@ status_t PlaylistFetcher::extractAndQueueAccessUnits( } if (mTSParser == NULL) { - mTSParser = new ATSParser; + // Use TS_TIMESTAMPS_ARE_ABSOLUTE so pts carry over between fetchers. + mTSParser = new ATSParser(ATSParser::TS_TIMESTAMPS_ARE_ABSOLUTE); } if (mNextPTSTimeUs >= 0ll) { sp extra = new AMessage; - extra->setInt64(IStreamListener::kKeyMediaTimeUs, mNextPTSTimeUs); + // Since we are using absolute timestamps, signal an offset of 0 to prevent + // ATSParser from skewing the timestamps of access units. + extra->setInt64(IStreamListener::kKeyMediaTimeUs, 0); mTSParser->signalDiscontinuity( ATSParser::DISCONTINUITY_SEEK, extra); @@ -841,17 +953,23 @@ status_t PlaylistFetcher::extractAndQueueAccessUnits( offset += 188; } + status_t err = OK; for (size_t i = mPacketSources.size(); i-- > 0;) { sp packetSource = mPacketSources.valueAt(i); + const char *key; ATSParser::SourceType type; - switch (mPacketSources.keyAt(i)) { + 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: @@ -878,19 +996,87 @@ status_t PlaylistFetcher::extractAndQueueAccessUnits( continue; } + int64_t timeUs; sp accessUnit; status_t finalResult; while (source->hasBufferAvailable(&finalResult) && source->dequeueAccessUnit(&accessUnit) == OK) { - // Note that we do NOT dequeue any discontinuities. + + CHECK(accessUnit->meta()->findInt64("timeUs", &timeUs)); + if (mMinStartTimeUs > 0) { + if (timeUs < mMinStartTimeUs) { + // TODO untested path + // try a later ts + int32_t targetDuration; + mPlaylist->meta()->findInt32("target-duration", &targetDuration); + int32_t incr = (mMinStartTimeUs - timeUs) / 1000000 / targetDuration; + if (incr == 0) { + // increment mSeqNumber by at least one + incr = 1; + } + mSeqNumber += incr; + err = -EAGAIN; + break; + } else { + int64_t startTimeUs; + if (mStartTimeUsNotify != NULL + && !mStartTimeUsNotify->findInt64(key, &startTimeUs)) { + mStartTimeUsNotify->setInt64(key, timeUs); + + uint32_t streamMask = 0; + mStartTimeUsNotify->findInt32("streamMask", (int32_t *) &streamMask); + streamMask |= mPacketSources.keyAt(i); + mStartTimeUsNotify->setInt32("streamMask", streamMask); + + if (streamMask == mStreamTypeMask) { + mStartTimeUsNotify->post(); + mStartTimeUsNotify.clear(); + } + } + } + } + + if (mStopParams != NULL) { + // Queue discontinuity in original stream. + int64_t stopTimeUs; + if (!mStopParams->findInt64(key, &stopTimeUs) || timeUs >= stopTimeUs) { + packetSource->queueAccessUnit(mSession->createFormatChangeBuffer()); + mStreamTypeMask &= ~stream; + mPacketSources.removeItemsAt(i); + break; + } + } + + // Note that we do NOT dequeue any discontinuities except for format change. // for simplicity, store a reference to the format in each unit sp format = source->getFormat(); if (format != NULL) { accessUnit->meta()->setObject("format", format); } + + // Stash the sequence number so we can hint future fetchers where to start at. + accessUnit->meta()->setInt32("seq", mSeqNumber); packetSource->queueAccessUnit(accessUnit); } + + if (err != OK) { + break; + } + } + + if (err != OK) { + for (size_t i = mPacketSources.size(); i-- > 0;) { + sp packetSource = mPacketSources.valueAt(i); + packetSource->clear(); + } + return err; + } + + if (!mStreamTypeMask) { + // Signal gap is filled between original and new stream. + ALOGV("ERROR OUT OF RANGE"); + return ERROR_OUT_OF_RANGE; } return OK; @@ -907,6 +1093,7 @@ status_t PlaylistFetcher::extractAndQueueAccessUnits( CHECK(itemMeta->findInt64("durationUs", &durationUs)); buffer->meta()->setInt64("timeUs", getSegmentStartTimeUs(mSeqNumber)); buffer->meta()->setInt64("durationUs", durationUs); + buffer->meta()->setInt32("seq", mSeqNumber); packetSource->queueAccessUnit(buffer); return OK; @@ -1043,6 +1230,7 @@ status_t PlaylistFetcher::extractAndQueueAccessUnits( // Each AAC frame encodes 1024 samples. numSamples += 1024; + unit->meta()->setInt32("seq", mSeqNumber); packetSource->queueAccessUnit(unit); offset += aac_frame_length; @@ -1070,4 +1258,33 @@ void PlaylistFetcher::updateDuration() { msg->post(); } +int64_t PlaylistFetcher::resumeThreshold(const sp &msg) { + int64_t durationUs, threshold; + if (msg->findInt64("durationUs", &durationUs)) { + return kNumSkipFrames * durationUs; + } + + sp obj; + msg->findObject("format", &obj); + MetaData *format = static_cast(obj.get()); + + const char *mime; + CHECK(format->findCString(kKeyMIMEType, &mime)); + bool audio = !strncasecmp(mime, "audio/", 6); + if (audio) { + // Assumes 1000 samples per frame. + int32_t sampleRate; + CHECK(format->findInt32(kKeySampleRate, &sampleRate)); + return kNumSkipFrames /* frames */ * 1000 /* samples */ + * (1000000 / sampleRate) /* sample duration (us) */; + } else { + int32_t frameRate; + if (format->findInt32(kKeyFrameRate, &frameRate) && frameRate > 0) { + return kNumSkipFrames * (1000000 / frameRate); + } + } + + return 500000ll; +} + } // namespace android diff --git a/media/libstagefright/httplive/PlaylistFetcher.h b/media/libstagefright/httplive/PlaylistFetcher.h index ac04a77..2e0349f 100644 --- a/media/libstagefright/httplive/PlaylistFetcher.h +++ b/media/libstagefright/httplive/PlaylistFetcher.h @@ -43,6 +43,7 @@ struct PlaylistFetcher : public AHandler { kWhatTemporarilyDoneFetching, kWhatPrepared, kWhatPreparationFailed, + kWhatStartedAt, }; PlaylistFetcher( @@ -56,12 +57,16 @@ struct PlaylistFetcher : public AHandler { const sp &audioSource, const sp &videoSource, const sp &subtitleSource, - int64_t startTimeUs = -1ll); + int64_t startTimeUs = -1ll, + int64_t minStartTimeUs = 0ll /* start after this timestamp */, + int32_t startSeqNumberHint = -1 /* try starting at this sequence number */); void pauseAsync(); void stopAsync(); + void resumeUntilAsync(const sp ¶ms); + protected: virtual ~PlaylistFetcher(); virtual void onMessageReceived(const sp &msg); @@ -76,17 +81,25 @@ private: kWhatPause = 'paus', kWhatStop = 'stop', kWhatMonitorQueue = 'moni', + kWhatResumeUntil = 'rsme', + kWhatDownloadNext = 'dlnx', }; static const int64_t kMinBufferedDurationUs; static const int64_t kMaxMonitorDelayUs; + static const int32_t kNumSkipFrames; + // notifications to mSession sp mNotify; + sp mStartTimeUsNotify; + sp mSession; AString mURI; uint32_t mStreamTypeMask; int64_t mStartTimeUs; + int64_t mMinStartTimeUs; // start fetching no earlier than this value + sp mStopParams; // message containing the latest timestamps we should fetch. KeyedVector > mPacketSources; @@ -153,6 +166,9 @@ private: void onMonitorQueue(); void onDownloadNext(); + // Resume a fetcher to continue until the stopping point stored in msg. + status_t onResumeUntil(const sp &msg); + status_t extractAndQueueAccessUnits( const sp &buffer, const sp &itemMeta); @@ -165,6 +181,10 @@ private: void updateDuration(); + // Before resuming a fetcher in onResume, check the remaining duration is longer than that + // returned by resumeThreshold. + int64_t resumeThreshold(const sp &msg); + DISALLOW_EVIL_CONSTRUCTORS(PlaylistFetcher); }; -- cgit v1.1 From 75d03185ac7be95c700f3f375080989e5de03ef6 Mon Sep 17 00:00:00 2001 From: Martin Storsjo Date: Fri, 22 Nov 2013 17:05:05 +0200 Subject: M3UParser: Skip query strings when looking for the last slash in a URL Bug: 13174301 Change-Id: I72d3a5e11fef9bbd75b291bc490c9cab1dce58da --- media/libstagefright/httplive/M3UParser.cpp | 34 +++++++++++++++++++---------- 1 file changed, 22 insertions(+), 12 deletions(-) (limited to 'media/libstagefright/httplive') diff --git a/media/libstagefright/httplive/M3UParser.cpp b/media/libstagefright/httplive/M3UParser.cpp index a9184b5..8f530ee 100644 --- a/media/libstagefright/httplive/M3UParser.cpp +++ b/media/libstagefright/httplive/M3UParser.cpp @@ -423,22 +423,32 @@ static bool MakeURL(const char *baseURL, const char *url, AString *out) { } else { // URL is a relative path - size_t n = strlen(baseURL); - if (baseURL[n - 1] == '/') { - out->setTo(baseURL); - out->append(url); + // Check for a possible query string + const char *qsPos = strchr(baseURL, '?'); + size_t end; + if (qsPos != NULL) { + end = qsPos - baseURL; } else { - const char *slashPos = strrchr(baseURL, '/'); - - if (slashPos > &baseURL[6]) { - out->setTo(baseURL, slashPos - baseURL); - } else { - out->setTo(baseURL); + end = strlen(baseURL); + } + // Check for the last slash before a potential query string + for (ssize_t pos = end - 1; pos >= 0; pos--) { + if (baseURL[pos] == '/') { + end = pos; + break; } + } - out->append("/"); - out->append(url); + // Check whether the found slash actually is part of the path + // and not part of the "http://". + if (end > 6) { + out->setTo(baseURL, end); + } else { + out->setTo(baseURL); } + + out->append("/"); + out->append(url); } ALOGV("base:'%s', url:'%s' => '%s'", baseURL, url, out->c_str()); -- cgit v1.1 From a1076eb135b74a32e9bdc1aed17aee4374eb41af Mon Sep 17 00:00:00 2001 From: Martin Storsjo Date: Fri, 22 Nov 2013 17:57:31 +0200 Subject: LiveSession: Use the actual, possibly redirected url as base in the M3U This fixes playback of HLS streams where the M3U playlists use relative paths and the main playlist is opened via a redirect. Bug: 13174301 Change-Id: I787e1c050daddc566be4c8e3f84803d3f91eec46 --- media/libstagefright/httplive/LiveSession.cpp | 14 +++++++++++--- media/libstagefright/httplive/LiveSession.h | 3 ++- 2 files changed, 13 insertions(+), 4 deletions(-) (limited to 'media/libstagefright/httplive') diff --git a/media/libstagefright/httplive/LiveSession.cpp b/media/libstagefright/httplive/LiveSession.cpp index 9e30ebd..687e871 100644 --- a/media/libstagefright/httplive/LiveSession.cpp +++ b/media/libstagefright/httplive/LiveSession.cpp @@ -623,7 +623,8 @@ status_t LiveSession::fetchFile( const char *url, sp *out, int64_t range_offset, int64_t range_length, uint32_t block_size, /* download block size */ - sp *source /* to return and reuse source */) { + sp *source, /* to return and reuse source */ + String8 *actualUrl) { off64_t size; sp temp_source; if (source == NULL) { @@ -719,6 +720,12 @@ status_t LiveSession::fetchFile( } *out = buffer; + if (actualUrl != NULL) { + *actualUrl = (*source)->getUri(); + if (actualUrl->isEmpty()) { + *actualUrl = url; + } + } return OK; } @@ -730,7 +737,8 @@ sp LiveSession::fetchPlaylist( *unchanged = false; sp buffer; - status_t err = fetchFile(url, &buffer); + String8 actualUrl; + status_t err = fetchFile(url, &buffer, 0, -1, 0, NULL, &actualUrl); if (err != OK) { return NULL; @@ -764,7 +772,7 @@ sp LiveSession::fetchPlaylist( #endif sp playlist = - new M3UParser(url, buffer->data(), buffer->size()); + new M3UParser(actualUrl.string(), buffer->data(), buffer->size()); if (playlist->initCheck() != OK) { ALOGE("failed to parse .m3u8 playlist"); diff --git a/media/libstagefright/httplive/LiveSession.h b/media/libstagefright/httplive/LiveSession.h index e26d024..376d451 100644 --- a/media/libstagefright/httplive/LiveSession.h +++ b/media/libstagefright/httplive/LiveSession.h @@ -207,7 +207,8 @@ private: /* download block size */ uint32_t block_size = 0, /* reuse DataSource if doing partial fetch */ - sp *source = NULL); + sp *source = NULL, + String8 *actualUrl = NULL); sp fetchPlaylist( const char *url, uint8_t *curPlaylistHash, bool *unchanged); -- cgit v1.1