Merge "HLS: bandwidth estimator changes"

3 files changed, 149 insertions, 69 deletions

diff --git a/media/libstagefright/httplive/LiveSession.cpp b/media/libstagefright/httplive/LiveSession.cpp
index a94754b..f5328a6 100644
--- a/media/libstagefright/httplive/LiveSession.cpp
+++ b/media/libstagefright/httplive/LiveSession.cpp
@@ -50,13 +50,75 @@
 namespace android {
 
 // static
-// Number of recently-read bytes to use for bandwidth estimation
-const size_t LiveSession::kBandwidthHistoryBytes = 200 * 1024;
 // High water mark to start up switch or report prepared)
 const int64_t LiveSession::kHighWaterMark = 8000000ll;
 const int64_t LiveSession::kMidWaterMark = 5000000ll;
 const int64_t LiveSession::kLowWaterMark = 3000000ll;
 
+struct LiveSession::BandwidthEstimator : public RefBase {
+    BandwidthEstimator();
+
+    void addBandwidthMeasurement(size_t numBytes, int64_t delayUs);
+    bool estimateBandwidth(int32_t *bandwidth);
+
+private:
+    // Bandwidth estimation parameters
+    static const int32_t kMaxBandwidthHistoryItems = 20;
+    static const int64_t kMaxBandwidthHistoryWindowUs = 3000000ll; // 3 sec
+
+    struct BandwidthEntry {
+        int64_t mDelayUs;
+        size_t mNumBytes;
+    };
+
+    Mutex mLock;
+    List<BandwidthEntry> mBandwidthHistory;
+    int64_t mTotalTransferTimeUs;
+    size_t mTotalTransferBytes;
+
+    DISALLOW_EVIL_CONSTRUCTORS(BandwidthEstimator);
+};
+
+LiveSession::BandwidthEstimator::BandwidthEstimator() :
+    mTotalTransferTimeUs(0),
+    mTotalTransferBytes(0) {
+}
+
+void LiveSession::BandwidthEstimator::addBandwidthMeasurement(
+        size_t numBytes, int64_t delayUs) {
+    AutoMutex autoLock(mLock);
+
+    BandwidthEntry entry;
+    entry.mDelayUs = delayUs;
+    entry.mNumBytes = numBytes;
+    mTotalTransferTimeUs += delayUs;
+    mTotalTransferBytes += numBytes;
+    mBandwidthHistory.push_back(entry);
+
+    // trim old samples, keeping at least kMaxBandwidthHistoryItems samples,
+    // and total transfer time at least kMaxBandwidthHistoryWindowUs.
+    while (mBandwidthHistory.size() > kMaxBandwidthHistoryItems) {
+        List<BandwidthEntry>::iterator it = mBandwidthHistory.begin();
+        if (mTotalTransferTimeUs - it->mDelayUs < kMaxBandwidthHistoryWindowUs) {
+            break;
+        }
+        mTotalTransferTimeUs -= it->mDelayUs;
+        mTotalTransferBytes -= it->mNumBytes;
+        mBandwidthHistory.erase(mBandwidthHistory.begin());
+    }
+}
+
+bool LiveSession::BandwidthEstimator::estimateBandwidth(int32_t *bandwidthBps) {
+    AutoMutex autoLock(mLock);
+
+    if (mBandwidthHistory.size() < 2) {
+        return false;
+    }
+
+    *bandwidthBps = ((double)mTotalTransferBytes * 8E6 / mTotalTransferTimeUs);
+    return true;
+}
+
 LiveSession::LiveSession(
         const sp<AMessage> &notify, uint32_t flags,
         const sp<IMediaHTTPService> &httpService)
@@ -66,10 +128,10 @@ LiveSession::LiveSession(
       mInPreparationPhase(true),
       mHTTPDataSource(new MediaHTTP(mHTTPService->makeHTTPConnection())),
       mCurBandwidthIndex(-1),
+      mBandwidthEstimator(new BandwidthEstimator()),
       mStreamMask(0),
       mNewStreamMask(0),
       mSwapMask(0),
-      mCheckBandwidthGeneration(0),
       mSwitchGeneration(0),
       mSubtitleGeneration(0),
       mLastDequeuedTimeUs(0ll),
@@ -89,13 +151,6 @@ LiveSession::LiveSession(
         mPacketSources.add(indexToType(i), new AnotherPacketSource(NULL /* meta */));
         mPacketSources2.add(indexToType(i), new AnotherPacketSource(NULL /* meta */));
     }
-
-    size_t numHistoryItems = kBandwidthHistoryBytes /
-            PlaylistFetcher::kDownloadBlockSize + 1;
-    if (numHistoryItems < 5) {
-        numHistoryItems = 5;
-    }
-    mHTTPDataSource->setBandwidthHistorySize(numHistoryItems);
 }
 
 LiveSession::~LiveSession() {
@@ -948,8 +1003,15 @@ static double uniformRand() {
 }
 #endif
 
-size_t LiveSession::getBandwidthIndex() {
-    if (mBandwidthItems.size() == 0) {
+void LiveSession::addBandwidthMeasurement(size_t numBytes, int64_t delayUs) {
+    mBandwidthEstimator->addBandwidthMeasurement(numBytes, delayUs);
+}
+
+size_t LiveSession::getBandwidthIndex(int32_t bandwidthBps) {
+    if (mBandwidthItems.size() < 2) {
+        // shouldn't be here if we only have 1 bandwidth, check
+        // logic to get rid of redundant bandwidth polling
+        ALOGW("getBandwidthIndex() called for single bandwidth playlist!");
         return 0;
     }
 
@@ -967,15 +1029,6 @@ size_t LiveSession::getBandwidthIndex() {
     }
 
     if (index < 0) {
-        int32_t bandwidthBps;
-        if (mHTTPDataSource != NULL
-                && mHTTPDataSource->estimateBandwidth(&bandwidthBps)) {
-            ALOGV("bandwidth estimated at %.2f kbps", bandwidthBps / 1024.0f);
-        } else {
-            ALOGV("no bandwidth estimate.");
-            return 0;  // Pick the lowest bandwidth stream by default.
-        }
-
         char value[PROPERTY_VALUE_MAX];
         if (property_get("media.httplive.max-bw", value, NULL)) {
             char *end;
@@ -992,15 +1045,9 @@ size_t LiveSession::getBandwidthIndex() {
 
         index = mBandwidthItems.size() - 1;
         while (index > 0) {
-            // consider only 80% of the available bandwidth, but if we are switching up,
-            // be even more conservative (70%) to avoid overestimating and immediately
-            // switching back.
-            size_t adjustedBandwidthBps = bandwidthBps;
-            if (index > mCurBandwidthIndex) {
-                adjustedBandwidthBps = adjustedBandwidthBps * 7 / 10;
-            } else {
-                adjustedBandwidthBps = adjustedBandwidthBps * 8 / 10;
-            }
+            // be conservative (70%) to avoid overestimating and immediately
+            // switching down again.
+            size_t adjustedBandwidthBps = bandwidthBps * 7 / 10;
             if (mBandwidthItems.itemAt(index).mBandwidth <= adjustedBandwidthBps) {
                 break;
             }
@@ -1577,9 +1624,9 @@ void LiveSession::cancelPollBuffering() {
 
 void LiveSession::onPollBuffering() {
     ALOGV("onPollBuffering: mSwitchInProgress %d, mReconfigurationInProgress %d, "
-            "mInPreparationPhase %d, mStreamMask 0x%x",
+            "mInPreparationPhase %d, mCurBandwidthIndex %d, mStreamMask 0x%x",
         mSwitchInProgress, mReconfigurationInProgress,
-        mInPreparationPhase, mStreamMask);
+        mInPreparationPhase, mCurBandwidthIndex, mStreamMask);
 
     bool low, mid, high;
     if (checkBuffering(low, mid, high)) {
@@ -1588,8 +1635,8 @@ void LiveSession::onPollBuffering() {
         }
 
         // don't switch before we report prepared
-        if (!mInPreparationPhase && (low || high)) {
-            switchBandwidthIfNeeded(high);
+        if (!mInPreparationPhase) {
+            switchBandwidthIfNeeded(high, !mid);
         }
     }
 
@@ -1704,11 +1751,35 @@ bool LiveSession::checkBuffering(bool &low, bool &mid, bool &high) {
     return false;
 }
 
-void LiveSession::switchBandwidthIfNeeded(bool canSwitchUp) {
-    ssize_t bandwidthIndex = getBandwidthIndex();
+void LiveSession::switchBandwidthIfNeeded(bool bufferHigh, bool bufferLow) {
+    // no need to check bandwidth if we only have 1 bandwidth settings
+    if (mBandwidthItems.size() < 2) {
+        return;
+    }
+
+    int32_t bandwidthBps;
+    if (mBandwidthEstimator->estimateBandwidth(&bandwidthBps)) {
+        ALOGV("bandwidth estimated at %.2f kbps", bandwidthBps / 1024.0f);
+    } else {
+        ALOGV("no bandwidth estimate.");
+        return;
+    }
+
+    int32_t curBandwidth = mBandwidthItems.itemAt(mCurBandwidthIndex).mBandwidth;
+    bool bandwidthLow = bandwidthBps < (int32_t)curBandwidth * 8 / 10;
+    bool bandwidthHigh = bandwidthBps > (int32_t)curBandwidth * 12 / 10;
+
+    if ((bufferHigh && bandwidthHigh) || (bufferLow && bandwidthLow)) {
+        ssize_t bandwidthIndex = getBandwidthIndex(bandwidthBps);
+
+        if (bandwidthIndex == mCurBandwidthIndex
+                || (bufferHigh && bandwidthIndex < mCurBandwidthIndex)
+                || (bufferLow && bandwidthIndex > mCurBandwidthIndex)) {
+            return;
+        }
 
-    if ((canSwitchUp && bandwidthIndex > mCurBandwidthIndex)
-            || (!canSwitchUp && bandwidthIndex < mCurBandwidthIndex)) {
+        ALOGI("#### Initiate Bandwidth Switch: %d => %d",
+                mCurBandwidthIndex, bandwidthIndex);
         changeConfiguration(-1, bandwidthIndex, false);
     }
 }
diff --git a/media/libstagefright/httplive/LiveSession.h b/media/libstagefright/httplive/LiveSession.h
index 3b0a9a4..685fefa 100644
--- a/media/libstagefright/httplive/LiveSession.h
+++ b/media/libstagefright/httplive/LiveSession.h
@@ -106,7 +106,6 @@ private:
         kWhatDisconnect                 = 'disc',
         kWhatSeek                       = 'seek',
         kWhatFetcherNotify              = 'notf',
-        kWhatCheckBandwidth             = 'bndw',
         kWhatChangeConfiguration        = 'chC0',
         kWhatChangeConfiguration2       = 'chC2',
         kWhatChangeConfiguration3       = 'chC3',
@@ -115,11 +114,11 @@ private:
         kWhatPollBuffering              = 'poll',
     };
 
-    static const size_t kBandwidthHistoryBytes;
     static const int64_t kHighWaterMark;
     static const int64_t kMidWaterMark;
     static const int64_t kLowWaterMark;
 
+    struct BandwidthEstimator;
     struct BandwidthItem {
         size_t mPlaylistIndex;
         unsigned long mBandwidth;
@@ -170,6 +169,7 @@ private:
 
     Vector<BandwidthItem> mBandwidthItems;
     ssize_t mCurBandwidthIndex;
+    sp<BandwidthEstimator> mBandwidthEstimator;
 
     sp<M3UParser> mPlaylist;
 
@@ -195,7 +195,6 @@ private:
     // * a forced bandwidth switch termination in cancelSwitch on the live looper.
     Mutex mSwapMutex;
 
-    int32_t mCheckBandwidthGeneration;
     int32_t mSwitchGeneration;
     int32_t mSubtitleGeneration;
 
@@ -249,7 +248,8 @@ private:
     sp<M3UParser> fetchPlaylist(
             const char *url, uint8_t *curPlaylistHash, bool *unchanged);
 
-    size_t getBandwidthIndex();
+    void addBandwidthMeasurement(size_t numBytes, int64_t delayUs);
+    size_t getBandwidthIndex(int32_t bandwidthBps);
     int64_t latestMediaSegmentStartTimeUs();
 
     static int SortByBandwidth(const BandwidthItem *, const BandwidthItem *);
@@ -273,7 +273,7 @@ private:
     void cancelPollBuffering();
     void onPollBuffering();
     bool checkBuffering(bool &low, bool &mid, bool &high);
-    void switchBandwidthIfNeeded(bool canSwitchUp);
+    void switchBandwidthIfNeeded(bool bufferHigh, bool bufferLow);
 
     void finishDisconnect();
 
diff --git a/media/libstagefright/httplive/PlaylistFetcher.cpp b/media/libstagefright/httplive/PlaylistFetcher.cpp
index 7f818a8..a447010 100644
--- a/media/libstagefright/httplive/PlaylistFetcher.cpp
+++ b/media/libstagefright/httplive/PlaylistFetcher.cpp
@@ -648,23 +648,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 +672,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 +692,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);
     }
 }
 
@@ -986,9 +984,20 @@ void PlaylistFetcher::onDownloadNext() {
     // block-wise download
     ssize_t bytesRead;
     do {
+        int64_t startUs = ALooper::GetNowUs();
+
         bytesRead = mSession->fetchFile(
                 uri.c_str(), &buffer, range_offset, range_length, kDownloadBlockSize, &source);
 
+        // add sample for bandwidth estimation (excluding subtitles)
+        if (bytesRead > 0
+                && (mStreamTypeMask
+                        & (LiveSession::STREAMTYPE_AUDIO
+                        | LiveSession::STREAMTYPE_VIDEO))) {
+            int64_t delayUs = ALooper::GetNowUs() - startUs;
+            mSession->addBandwidthMeasurement(bytesRead, delayUs);
+        }
+
         if (bytesRead < 0) {
             status_t err = bytesRead;
             ALOGE("failed to fetch .ts segment at url '%s'", uri.c_str());