Home | History | Annotate | Download | only in libstagefright 
      1 /*
      2  * Copyright (C) 2010 The Android Open Source Project
      3  *
      4  * Licensed under the Apache License, Version 2.0 (the "License");
      5  * you may not use this file except in compliance with the License.
      6  * You may obtain a copy of the License at
      7  *
      8  *      http://www.apache.org/licenses/LICENSE-2.0
      9  *
     10  * Unless required by applicable law or agreed to in writing, software
     11  * distributed under the License is distributed on an "AS IS" BASIS,
     12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     13  * See the License for the specific language governing permissions and
     14  * limitations under the License.
     15  */
     16 
     17 //#define LOG_NDEBUG 0
     18 #define LOG_TAG "NuCachedSource2"
     19 #include <utils/Log.h>
     20 
     21 #include "include/NuCachedSource2.h"
     22 #include "include/HTTPBase.h"
     23 
     24 #include <cutils/properties.h>
     25 #include <media/stagefright/foundation/ADebug.h>
     26 #include <media/stagefright/foundation/AMessage.h>
     27 #include <media/stagefright/MediaErrors.h>
     28 
     29 namespace android {
     30 
     31 struct PageCache {
     32     PageCache(size_t pageSize);
     33     ~PageCache();
     34 
     35     struct Page {
     36         void *mData;
     37         size_t mSize;
     38     };
     39 
     40     Page *acquirePage();
     41     void releasePage(Page *page);
     42 
     43     void appendPage(Page *page);
     44     size_t releaseFromStart(size_t maxBytes);
     45 
     46     size_t totalSize() const {
     47         return mTotalSize;
     48     }
     49 
     50     void copy(size_t from, void *data, size_t size);
     51 
     52 private:
     53     size_t mPageSize;
     54     size_t mTotalSize;
     55 
     56     List<Page *> mActivePages;
     57     List<Page *> mFreePages;
     58 
     59     void freePages(List<Page *> *list);
     60 
     61     DISALLOW_EVIL_CONSTRUCTORS(PageCache);
     62 };
     63 
     64 PageCache::PageCache(size_t pageSize)
     65     : mPageSize(pageSize),
     66       mTotalSize(0) {
     67 }
     68 
     69 PageCache::~PageCache() {
     70     freePages(&mActivePages);
     71     freePages(&mFreePages);
     72 }
     73 
     74 void PageCache::freePages(List<Page *> *list) {
     75     List<Page *>::iterator it = list->begin();
     76     while (it != list->end()) {
     77         Page *page = *it;
     78 
     79         free(page->mData);
     80         delete page;
     81         page = NULL;
     82 
     83         ++it;
     84     }
     85 }
     86 
     87 PageCache::Page *PageCache::acquirePage() {
     88     if (!mFreePages.empty()) {
     89         List<Page *>::iterator it = mFreePages.begin();
     90         Page *page = *it;
     91         mFreePages.erase(it);
     92 
     93         return page;
     94     }
     95 
     96     Page *page = new Page;
     97     page->mData = malloc(mPageSize);
     98     page->mSize = 0;
     99 
    100     return page;
    101 }
    102 
    103 void PageCache::releasePage(Page *page) {
    104     page->mSize = 0;
    105     mFreePages.push_back(page);
    106 }
    107 
    108 void PageCache::appendPage(Page *page) {
    109     mTotalSize += page->mSize;
    110     mActivePages.push_back(page);
    111 }
    112 
    113 size_t PageCache::releaseFromStart(size_t maxBytes) {
    114     size_t bytesReleased = 0;
    115 
    116     while (maxBytes > 0 && !mActivePages.empty()) {
    117         List<Page *>::iterator it = mActivePages.begin();
    118 
    119         Page *page = *it;
    120 
    121         if (maxBytes < page->mSize) {
    122             break;
    123         }
    124 
    125         mActivePages.erase(it);
    126 
    127         maxBytes -= page->mSize;
    128         bytesReleased += page->mSize;
    129 
    130         releasePage(page);
    131     }
    132 
    133     mTotalSize -= bytesReleased;
    134     return bytesReleased;
    135 }
    136 
    137 void PageCache::copy(size_t from, void *data, size_t size) {
    138     LOGV("copy from %d size %d", from, size);
    139 
    140     if (size == 0) {
    141         return;
    142     }
    143 
    144     CHECK_LE(from + size, mTotalSize);
    145 
    146     size_t offset = 0;
    147     List<Page *>::iterator it = mActivePages.begin();
    148     while (from >= offset + (*it)->mSize) {
    149         offset += (*it)->mSize;
    150         ++it;
    151     }
    152 
    153     size_t delta = from - offset;
    154     size_t avail = (*it)->mSize - delta;
    155 
    156     if (avail >= size) {
    157         memcpy(data, (const uint8_t *)(*it)->mData + delta, size);
    158         return;
    159     }
    160 
    161     memcpy(data, (const uint8_t *)(*it)->mData + delta, avail);
    162     ++it;
    163     data = (uint8_t *)data + avail;
    164     size -= avail;
    165 
    166     while (size > 0) {
    167         size_t copy = (*it)->mSize;
    168         if (copy > size) {
    169             copy = size;
    170         }
    171         memcpy(data, (*it)->mData, copy);
    172         data = (uint8_t *)data + copy;
    173         size -= copy;
    174         ++it;
    175     }
    176 }
    177 
    178 ////////////////////////////////////////////////////////////////////////////////
    179 
    180 NuCachedSource2::NuCachedSource2(
    181         const sp<DataSource> &source,
    182         const char *cacheConfig,
    183         bool disconnectAtHighwatermark)
    184     : mSource(source),
    185       mReflector(new AHandlerReflector<NuCachedSource2>(this)),
    186       mLooper(new ALooper),
    187       mCache(new PageCache(kPageSize)),
    188       mCacheOffset(0),
    189       mFinalStatus(OK),
    190       mLastAccessPos(0),
    191       mFetching(true),
    192       mLastFetchTimeUs(-1),
    193       mNumRetriesLeft(kMaxNumRetries),
    194       mHighwaterThresholdBytes(kDefaultHighWaterThreshold),
    195       mLowwaterThresholdBytes(kDefaultLowWaterThreshold),
    196       mKeepAliveIntervalUs(kDefaultKeepAliveIntervalUs),
    197       mDisconnectAtHighwatermark(disconnectAtHighwatermark) {
    198     // We are NOT going to support disconnect-at-highwatermark indefinitely
    199     // and we are not guaranteeing support for client-specified cache
    200     // parameters. Both of these are temporary measures to solve a specific
    201     // problem that will be solved in a better way going forward.
    202 
    203     updateCacheParamsFromSystemProperty();
    204 
    205     if (cacheConfig != NULL) {
    206         updateCacheParamsFromString(cacheConfig);
    207     }
    208 
    209     if (mDisconnectAtHighwatermark) {
    210         // Makes no sense to disconnect and do keep-alives...
    211         mKeepAliveIntervalUs = 0;
    212     }
    213 
    214     mLooper->setName("NuCachedSource2");
    215     mLooper->registerHandler(mReflector);
    216     mLooper->start();
    217 
    218     Mutex::Autolock autoLock(mLock);
    219     (new AMessage(kWhatFetchMore, mReflector->id()))->post();
    220 }
    221 
    222 NuCachedSource2::~NuCachedSource2() {
    223     mLooper->stop();
    224     mLooper->unregisterHandler(mReflector->id());
    225 
    226     delete mCache;
    227     mCache = NULL;
    228 }
    229 
    230 status_t NuCachedSource2::getEstimatedBandwidthKbps(int32_t *kbps) {
    231     if (mSource->flags() & kIsHTTPBasedSource) {
    232         HTTPBase* source = static_cast<HTTPBase *>(mSource.get());
    233         return source->getEstimatedBandwidthKbps(kbps);
    234     }
    235     return ERROR_UNSUPPORTED;
    236 }
    237 
    238 status_t NuCachedSource2::setCacheStatCollectFreq(int32_t freqMs) {
    239     if (mSource->flags() & kIsHTTPBasedSource) {
    240         HTTPBase *source = static_cast<HTTPBase *>(mSource.get());
    241         return source->setBandwidthStatCollectFreq(freqMs);
    242     }
    243     return ERROR_UNSUPPORTED;
    244 }
    245 
    246 status_t NuCachedSource2::initCheck() const {
    247     return mSource->initCheck();
    248 }
    249 
    250 status_t NuCachedSource2::getSize(off64_t *size) {
    251     return mSource->getSize(size);
    252 }
    253 
    254 uint32_t NuCachedSource2::flags() {
    255     // Remove HTTP related flags since NuCachedSource2 is not HTTP-based.
    256     uint32_t flags = mSource->flags() & ~(kWantsPrefetching | kIsHTTPBasedSource);
    257     return (flags | kIsCachingDataSource);
    258 }
    259 
    260 void NuCachedSource2::onMessageReceived(const sp<AMessage> &msg) {
    261     switch (msg->what()) {
    262         case kWhatFetchMore:
    263         {
    264             onFetch();
    265             break;
    266         }
    267 
    268         case kWhatRead:
    269         {
    270             onRead(msg);
    271             break;
    272         }
    273 
    274         default:
    275             TRESPASS();
    276     }
    277 }
    278 
    279 void NuCachedSource2::fetchInternal() {
    280     LOGV("fetchInternal");
    281 
    282     bool reconnect = false;
    283 
    284     {
    285         Mutex::Autolock autoLock(mLock);
    286         CHECK(mFinalStatus == OK || mNumRetriesLeft > 0);
    287 
    288         if (mFinalStatus != OK) {
    289             --mNumRetriesLeft;
    290 
    291             reconnect = true;
    292         }
    293     }
    294 
    295     if (reconnect) {
    296         status_t err =
    297             mSource->reconnectAtOffset(mCacheOffset + mCache->totalSize());
    298 
    299         Mutex::Autolock autoLock(mLock);
    300 
    301         if (err == ERROR_UNSUPPORTED) {
    302             mNumRetriesLeft = 0;
    303             return;
    304         } else if (err != OK) {
    305             LOGI("The attempt to reconnect failed, %d retries remaining",
    306                  mNumRetriesLeft);
    307 
    308             return;
    309         }
    310     }
    311 
    312     PageCache::Page *page = mCache->acquirePage();
    313 
    314     ssize_t n = mSource->readAt(
    315             mCacheOffset + mCache->totalSize(), page->mData, kPageSize);
    316 
    317     Mutex::Autolock autoLock(mLock);
    318 
    319     if (n < 0) {
    320         LOGE("source returned error %ld, %d retries left", n, mNumRetriesLeft);
    321         mFinalStatus = n;
    322         mCache->releasePage(page);
    323     } else if (n == 0) {
    324         LOGI("ERROR_END_OF_STREAM");
    325 
    326         mNumRetriesLeft = 0;
    327         mFinalStatus = ERROR_END_OF_STREAM;
    328 
    329         mCache->releasePage(page);
    330     } else {
    331         if (mFinalStatus != OK) {
    332             LOGI("retrying a previously failed read succeeded.");
    333         }
    334         mNumRetriesLeft = kMaxNumRetries;
    335         mFinalStatus = OK;
    336 
    337         page->mSize = n;
    338         mCache->appendPage(page);
    339     }
    340 }
    341 
    342 void NuCachedSource2::onFetch() {
    343     LOGV("onFetch");
    344 
    345     if (mFinalStatus != OK && mNumRetriesLeft == 0) {
    346         LOGV("EOS reached, done prefetching for now");
    347         mFetching = false;
    348     }
    349 
    350     bool keepAlive =
    351         !mFetching
    352             && mFinalStatus == OK
    353             && mKeepAliveIntervalUs > 0
    354             && ALooper::GetNowUs() >= mLastFetchTimeUs + mKeepAliveIntervalUs;
    355 
    356     if (mFetching || keepAlive) {
    357         if (keepAlive) {
    358             LOGI("Keep alive");
    359         }
    360 
    361         fetchInternal();
    362 
    363         mLastFetchTimeUs = ALooper::GetNowUs();
    364 
    365         if (mFetching && mCache->totalSize() >= mHighwaterThresholdBytes) {
    366             LOGI("Cache full, done prefetching for now");
    367             mFetching = false;
    368 
    369             if (mDisconnectAtHighwatermark
    370                     && (mSource->flags() & DataSource::kIsHTTPBasedSource)) {
    371                 LOGV("Disconnecting at high watermark");
    372                 static_cast<HTTPBase *>(mSource.get())->disconnect();
    373             }
    374         }
    375     } else {
    376         Mutex::Autolock autoLock(mLock);
    377         restartPrefetcherIfNecessary_l();
    378     }
    379 
    380     int64_t delayUs;
    381     if (mFetching) {
    382         if (mFinalStatus != OK && mNumRetriesLeft > 0) {
    383             // We failed this time and will try again in 3 seconds.
    384             delayUs = 3000000ll;
    385         } else {
    386             delayUs = 0;
    387         }
    388     } else {
    389         delayUs = 100000ll;
    390     }
    391 
    392     (new AMessage(kWhatFetchMore, mReflector->id()))->post(delayUs);
    393 }
    394 
    395 void NuCachedSource2::onRead(const sp<AMessage> &msg) {
    396     LOGV("onRead");
    397 
    398     int64_t offset;
    399     CHECK(msg->findInt64("offset", &offset));
    400 
    401     void *data;
    402     CHECK(msg->findPointer("data", &data));
    403 
    404     size_t size;
    405     CHECK(msg->findSize("size", &size));
    406 
    407     ssize_t result = readInternal(offset, data, size);
    408 
    409     if (result == -EAGAIN) {
    410         msg->post(50000);
    411         return;
    412     }
    413 
    414     Mutex::Autolock autoLock(mLock);
    415 
    416     CHECK(mAsyncResult == NULL);
    417 
    418     mAsyncResult = new AMessage;
    419     mAsyncResult->setInt32("result", result);
    420 
    421     mCondition.signal();
    422 }
    423 
    424 void NuCachedSource2::restartPrefetcherIfNecessary_l(
    425         bool ignoreLowWaterThreshold, bool force) {
    426     static const size_t kGrayArea = 1024 * 1024;
    427 
    428     if (mFetching || (mFinalStatus != OK && mNumRetriesLeft == 0)) {
    429         return;
    430     }
    431 
    432     if (!ignoreLowWaterThreshold && !force
    433             && mCacheOffset + mCache->totalSize() - mLastAccessPos
    434                 >= mLowwaterThresholdBytes) {
    435         return;
    436     }
    437 
    438     size_t maxBytes = mLastAccessPos - mCacheOffset;
    439 
    440     if (!force) {
    441         if (maxBytes < kGrayArea) {
    442             return;
    443         }
    444 
    445         maxBytes -= kGrayArea;
    446     }
    447 
    448     size_t actualBytes = mCache->releaseFromStart(maxBytes);
    449     mCacheOffset += actualBytes;
    450 
    451     LOGI("restarting prefetcher, totalSize = %d", mCache->totalSize());
    452     mFetching = true;
    453 }
    454 
    455 ssize_t NuCachedSource2::readAt(off64_t offset, void *data, size_t size) {
    456     Mutex::Autolock autoSerializer(mSerializer);
    457 
    458     LOGV("readAt offset %lld, size %d", offset, size);
    459 
    460     Mutex::Autolock autoLock(mLock);
    461 
    462     // If the request can be completely satisfied from the cache, do so.
    463 
    464     if (offset >= mCacheOffset
    465             && offset + size <= mCacheOffset + mCache->totalSize()) {
    466         size_t delta = offset - mCacheOffset;
    467         mCache->copy(delta, data, size);
    468 
    469         mLastAccessPos = offset + size;
    470 
    471         return size;
    472     }
    473 
    474     sp<AMessage> msg = new AMessage(kWhatRead, mReflector->id());
    475     msg->setInt64("offset", offset);
    476     msg->setPointer("data", data);
    477     msg->setSize("size", size);
    478 
    479     CHECK(mAsyncResult == NULL);
    480     msg->post();
    481 
    482     while (mAsyncResult == NULL) {
    483         mCondition.wait(mLock);
    484     }
    485 
    486     int32_t result;
    487     CHECK(mAsyncResult->findInt32("result", &result));
    488 
    489     mAsyncResult.clear();
    490 
    491     if (result > 0) {
    492         mLastAccessPos = offset + result;
    493     }
    494 
    495     return (ssize_t)result;
    496 }
    497 
    498 size_t NuCachedSource2::cachedSize() {
    499     Mutex::Autolock autoLock(mLock);
    500     return mCacheOffset + mCache->totalSize();
    501 }
    502 
    503 size_t NuCachedSource2::approxDataRemaining(status_t *finalStatus) {
    504     Mutex::Autolock autoLock(mLock);
    505     return approxDataRemaining_l(finalStatus);
    506 }
    507 
    508 size_t NuCachedSource2::approxDataRemaining_l(status_t *finalStatus) {
    509     *finalStatus = mFinalStatus;
    510 
    511     if (mFinalStatus != OK && mNumRetriesLeft > 0) {
    512         // Pretend that everything is fine until we're out of retries.
    513         *finalStatus = OK;
    514     }
    515 
    516     off64_t lastBytePosCached = mCacheOffset + mCache->totalSize();
    517     if (mLastAccessPos < lastBytePosCached) {
    518         return lastBytePosCached - mLastAccessPos;
    519     }
    520     return 0;
    521 }
    522 
    523 ssize_t NuCachedSource2::readInternal(off64_t offset, void *data, size_t size) {
    524     CHECK_LE(size, (size_t)mHighwaterThresholdBytes);
    525 
    526     LOGV("readInternal offset %lld size %d", offset, size);
    527 
    528     Mutex::Autolock autoLock(mLock);
    529 
    530     if (!mFetching) {
    531         mLastAccessPos = offset;
    532         restartPrefetcherIfNecessary_l(
    533                 false, // ignoreLowWaterThreshold
    534                 true); // force
    535     }
    536 
    537     if (offset < mCacheOffset
    538             || offset >= (off64_t)(mCacheOffset + mCache->totalSize())) {
    539         static const off64_t kPadding = 256 * 1024;
    540 
    541         // In the presence of multiple decoded streams, once of them will
    542         // trigger this seek request, the other one will request data "nearby"
    543         // soon, adjust the seek position so that that subsequent request
    544         // does not trigger another seek.
    545         off64_t seekOffset = (offset > kPadding) ? offset - kPadding : 0;
    546 
    547         seekInternal_l(seekOffset);
    548     }
    549 
    550     size_t delta = offset - mCacheOffset;
    551 
    552     if (mFinalStatus != OK) {
    553         if (delta >= mCache->totalSize()) {
    554             return mFinalStatus;
    555         }
    556 
    557         size_t avail = mCache->totalSize() - delta;
    558 
    559         if (avail > size) {
    560             avail = size;
    561         }
    562 
    563         mCache->copy(delta, data, avail);
    564 
    565         return avail;
    566     }
    567 
    568     if (offset + size <= mCacheOffset + mCache->totalSize()) {
    569         mCache->copy(delta, data, size);
    570 
    571         return size;
    572     }
    573 
    574     LOGV("deferring read");
    575 
    576     return -EAGAIN;
    577 }
    578 
    579 status_t NuCachedSource2::seekInternal_l(off64_t offset) {
    580     mLastAccessPos = offset;
    581 
    582     if (offset >= mCacheOffset
    583             && offset <= (off64_t)(mCacheOffset + mCache->totalSize())) {
    584         return OK;
    585     }
    586 
    587     LOGI("new range: offset= %lld", offset);
    588 
    589     mCacheOffset = offset;
    590 
    591     size_t totalSize = mCache->totalSize();
    592     CHECK_EQ(mCache->releaseFromStart(totalSize), totalSize);
    593 
    594     mFinalStatus = OK;
    595     mFetching = true;
    596 
    597     return OK;
    598 }
    599 
    600 void NuCachedSource2::resumeFetchingIfNecessary() {
    601     Mutex::Autolock autoLock(mLock);
    602 
    603     restartPrefetcherIfNecessary_l(true /* ignore low water threshold */);
    604 }
    605 
    606 sp<DecryptHandle> NuCachedSource2::DrmInitialization() {
    607     return mSource->DrmInitialization();
    608 }
    609 
    610 void NuCachedSource2::getDrmInfo(sp<DecryptHandle> &handle, DrmManagerClient **client) {
    611     mSource->getDrmInfo(handle, client);
    612 }
    613 
    614 String8 NuCachedSource2::getUri() {
    615     return mSource->getUri();
    616 }
    617 
    618 String8 NuCachedSource2::getMIMEType() const {
    619     return mSource->getMIMEType();
    620 }
    621 
    622 void NuCachedSource2::updateCacheParamsFromSystemProperty() {
    623     char value[PROPERTY_VALUE_MAX];
    624     if (!property_get("media.stagefright.cache-params", value, NULL)) {
    625         return;
    626     }
    627 
    628     updateCacheParamsFromString(value);
    629 }
    630 
    631 void NuCachedSource2::updateCacheParamsFromString(const char *s) {
    632     ssize_t lowwaterMarkKb, highwaterMarkKb;
    633     int keepAliveSecs;
    634 
    635     if (sscanf(s, "%ld/%ld/%d",
    636                &lowwaterMarkKb, &highwaterMarkKb, &keepAliveSecs) != 3) {
    637         LOGE("Failed to parse cache parameters from '%s'.", s);
    638         return;
    639     }
    640 
    641     if (lowwaterMarkKb >= 0) {
    642         mLowwaterThresholdBytes = lowwaterMarkKb * 1024;
    643     } else {
    644         mLowwaterThresholdBytes = kDefaultLowWaterThreshold;
    645     }
    646 
    647     if (highwaterMarkKb >= 0) {
    648         mHighwaterThresholdBytes = highwaterMarkKb * 1024;
    649     } else {
    650         mHighwaterThresholdBytes = kDefaultHighWaterThreshold;
    651     }
    652 
    653     if (mLowwaterThresholdBytes >= mHighwaterThresholdBytes) {
    654         LOGE("Illegal low/highwater marks specified, reverting to defaults.");
    655 
    656         mLowwaterThresholdBytes = kDefaultLowWaterThreshold;
    657         mHighwaterThresholdBytes = kDefaultHighWaterThreshold;
    658     }
    659 
    660     if (keepAliveSecs >= 0) {
    661         mKeepAliveIntervalUs = keepAliveSecs * 1000000ll;
    662     } else {
    663         mKeepAliveIntervalUs = kDefaultKeepAliveIntervalUs;
    664     }
    665 
    666     LOGV("lowwater = %d bytes, highwater = %d bytes, keepalive = %lld us",
    667          mLowwaterThresholdBytes,
    668          mHighwaterThresholdBytes,
    669          mKeepAliveIntervalUs);
    670 }
    671 
    672 // static
    673 void NuCachedSource2::RemoveCacheSpecificHeaders(
    674         KeyedVector<String8, String8> *headers,
    675         String8 *cacheConfig,
    676         bool *disconnectAtHighwatermark) {
    677     *cacheConfig = String8();
    678     *disconnectAtHighwatermark = false;
    679 
    680     if (headers == NULL) {
    681         return;
    682     }
    683 
    684     ssize_t index;
    685     if ((index = headers->indexOfKey(String8("x-cache-config"))) >= 0) {
    686         *cacheConfig = headers->valueAt(index);
    687 
    688         headers->removeItemsAt(index);
    689 
    690         LOGV("Using special cache config '%s'", cacheConfig->string());
    691     }
    692 
    693     if ((index = headers->indexOfKey(
    694                     String8("x-disconnect-at-highwatermark"))) >= 0) {
    695         *disconnectAtHighwatermark = true;
    696         headers->removeItemsAt(index);
    697 
    698         LOGV("Client requested disconnection at highwater mark");
    699     }
    700 }
    701 
    702 }  // namespace android
    703