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     ALOGV("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     ALOGV("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             ALOGI("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         ALOGE("source returned error %ld, %d retries left", n, mNumRetriesLeft);
    321         mFinalStatus = n;
    322         mCache->releasePage(page);
    323     } else if (n == 0) {
    324         ALOGI("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             ALOGI("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     ALOGV("onFetch");
    344 
    345     if (mFinalStatus != OK && mNumRetriesLeft == 0) {
    346         ALOGV("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             ALOGI("Keep alive");
    359         }
    360 
    361         fetchInternal();
    362 
    363         mLastFetchTimeUs = ALooper::GetNowUs();
    364 
    365         if (mFetching && mCache->totalSize() >= mHighwaterThresholdBytes) {
    366             ALOGI("Cache full, done prefetching for now");
    367             mFetching = false;
    368 
    369             if (mDisconnectAtHighwatermark
    370                     && (mSource->flags() & DataSource::kIsHTTPBasedSource)) {
    371                 ALOGV("Disconnecting at high watermark");
    372                 static_cast<HTTPBase *>(mSource.get())->disconnect();
    373                 mFinalStatus = -EAGAIN;
    374             }
    375         }
    376     } else {
    377         Mutex::Autolock autoLock(mLock);
    378         restartPrefetcherIfNecessary_l();
    379     }
    380 
    381     int64_t delayUs;
    382     if (mFetching) {
    383         if (mFinalStatus != OK && mNumRetriesLeft > 0) {
    384             // We failed this time and will try again in 3 seconds.
    385             delayUs = 3000000ll;
    386         } else {
    387             delayUs = 0;
    388         }
    389     } else {
    390         delayUs = 100000ll;
    391     }
    392 
    393     (new AMessage(kWhatFetchMore, mReflector->id()))->post(delayUs);
    394 }
    395 
    396 void NuCachedSource2::onRead(const sp<AMessage> &msg) {
    397     ALOGV("onRead");
    398 
    399     int64_t offset;
    400     CHECK(msg->findInt64("offset", &offset));
    401 
    402     void *data;
    403     CHECK(msg->findPointer("data", &data));
    404 
    405     size_t size;
    406     CHECK(msg->findSize("size", &size));
    407 
    408     ssize_t result = readInternal(offset, data, size);
    409 
    410     if (result == -EAGAIN) {
    411         msg->post(50000);
    412         return;
    413     }
    414 
    415     Mutex::Autolock autoLock(mLock);
    416 
    417     CHECK(mAsyncResult == NULL);
    418 
    419     mAsyncResult = new AMessage;
    420     mAsyncResult->setInt32("result", result);
    421 
    422     mCondition.signal();
    423 }
    424 
    425 void NuCachedSource2::restartPrefetcherIfNecessary_l(
    426         bool ignoreLowWaterThreshold, bool force) {
    427     static const size_t kGrayArea = 1024 * 1024;
    428 
    429     if (mFetching || (mFinalStatus != OK && mNumRetriesLeft == 0)) {
    430         return;
    431     }
    432 
    433     if (!ignoreLowWaterThreshold && !force
    434             && mCacheOffset + mCache->totalSize() - mLastAccessPos
    435                 >= mLowwaterThresholdBytes) {
    436         return;
    437     }
    438 
    439     size_t maxBytes = mLastAccessPos - mCacheOffset;
    440 
    441     if (!force) {
    442         if (maxBytes < kGrayArea) {
    443             return;
    444         }
    445 
    446         maxBytes -= kGrayArea;
    447     }
    448 
    449     size_t actualBytes = mCache->releaseFromStart(maxBytes);
    450     mCacheOffset += actualBytes;
    451 
    452     ALOGI("restarting prefetcher, totalSize = %d", mCache->totalSize());
    453     mFetching = true;
    454 }
    455 
    456 ssize_t NuCachedSource2::readAt(off64_t offset, void *data, size_t size) {
    457     Mutex::Autolock autoSerializer(mSerializer);
    458 
    459     ALOGV("readAt offset %lld, size %d", offset, size);
    460 
    461     Mutex::Autolock autoLock(mLock);
    462 
    463     // If the request can be completely satisfied from the cache, do so.
    464 
    465     if (offset >= mCacheOffset
    466             && offset + size <= mCacheOffset + mCache->totalSize()) {
    467         size_t delta = offset - mCacheOffset;
    468         mCache->copy(delta, data, size);
    469 
    470         mLastAccessPos = offset + size;
    471 
    472         return size;
    473     }
    474 
    475     sp<AMessage> msg = new AMessage(kWhatRead, mReflector->id());
    476     msg->setInt64("offset", offset);
    477     msg->setPointer("data", data);
    478     msg->setSize("size", size);
    479 
    480     CHECK(mAsyncResult == NULL);
    481     msg->post();
    482 
    483     while (mAsyncResult == NULL) {
    484         mCondition.wait(mLock);
    485     }
    486 
    487     int32_t result;
    488     CHECK(mAsyncResult->findInt32("result", &result));
    489 
    490     mAsyncResult.clear();
    491 
    492     if (result > 0) {
    493         mLastAccessPos = offset + result;
    494     }
    495 
    496     return (ssize_t)result;
    497 }
    498 
    499 size_t NuCachedSource2::cachedSize() {
    500     Mutex::Autolock autoLock(mLock);
    501     return mCacheOffset + mCache->totalSize();
    502 }
    503 
    504 size_t NuCachedSource2::approxDataRemaining(status_t *finalStatus) const {
    505     Mutex::Autolock autoLock(mLock);
    506     return approxDataRemaining_l(finalStatus);
    507 }
    508 
    509 size_t NuCachedSource2::approxDataRemaining_l(status_t *finalStatus) const {
    510     *finalStatus = mFinalStatus;
    511 
    512     if (mFinalStatus != OK && mNumRetriesLeft > 0) {
    513         // Pretend that everything is fine until we're out of retries.
    514         *finalStatus = OK;
    515     }
    516 
    517     off64_t lastBytePosCached = mCacheOffset + mCache->totalSize();
    518     if (mLastAccessPos < lastBytePosCached) {
    519         return lastBytePosCached - mLastAccessPos;
    520     }
    521     return 0;
    522 }
    523 
    524 ssize_t NuCachedSource2::readInternal(off64_t offset, void *data, size_t size) {
    525     CHECK_LE(size, (size_t)mHighwaterThresholdBytes);
    526 
    527     ALOGV("readInternal offset %lld size %d", offset, size);
    528 
    529     Mutex::Autolock autoLock(mLock);
    530 
    531     if (!mFetching) {
    532         mLastAccessPos = offset;
    533         restartPrefetcherIfNecessary_l(
    534                 false, // ignoreLowWaterThreshold
    535                 true); // force
    536     }
    537 
    538     if (offset < mCacheOffset
    539             || offset >= (off64_t)(mCacheOffset + mCache->totalSize())) {
    540         static const off64_t kPadding = 256 * 1024;
    541 
    542         // In the presence of multiple decoded streams, once of them will
    543         // trigger this seek request, the other one will request data "nearby"
    544         // soon, adjust the seek position so that that subsequent request
    545         // does not trigger another seek.
    546         off64_t seekOffset = (offset > kPadding) ? offset - kPadding : 0;
    547 
    548         seekInternal_l(seekOffset);
    549     }
    550 
    551     size_t delta = offset - mCacheOffset;
    552 
    553     if (mFinalStatus != OK && mNumRetriesLeft == 0) {
    554         if (delta >= mCache->totalSize()) {
    555             return mFinalStatus;
    556         }
    557 
    558         size_t avail = mCache->totalSize() - delta;
    559 
    560         if (avail > size) {
    561             avail = size;
    562         }
    563 
    564         mCache->copy(delta, data, avail);
    565 
    566         return avail;
    567     }
    568 
    569     if (offset + size <= mCacheOffset + mCache->totalSize()) {
    570         mCache->copy(delta, data, size);
    571 
    572         return size;
    573     }
    574 
    575     ALOGV("deferring read");
    576 
    577     return -EAGAIN;
    578 }
    579 
    580 status_t NuCachedSource2::seekInternal_l(off64_t offset) {
    581     mLastAccessPos = offset;
    582 
    583     if (offset >= mCacheOffset
    584             && offset <= (off64_t)(mCacheOffset + mCache->totalSize())) {
    585         return OK;
    586     }
    587 
    588     ALOGI("new range: offset= %lld", offset);
    589 
    590     mCacheOffset = offset;
    591 
    592     size_t totalSize = mCache->totalSize();
    593     CHECK_EQ(mCache->releaseFromStart(totalSize), totalSize);
    594 
    595     mNumRetriesLeft = kMaxNumRetries;
    596     mFetching = true;
    597 
    598     return OK;
    599 }
    600 
    601 void NuCachedSource2::resumeFetchingIfNecessary() {
    602     Mutex::Autolock autoLock(mLock);
    603 
    604     restartPrefetcherIfNecessary_l(true /* ignore low water threshold */);
    605 }
    606 
    607 sp<DecryptHandle> NuCachedSource2::DrmInitialization(const char* mime) {
    608     return mSource->DrmInitialization(mime);
    609 }
    610 
    611 void NuCachedSource2::getDrmInfo(sp<DecryptHandle> &handle, DrmManagerClient **client) {
    612     mSource->getDrmInfo(handle, client);
    613 }
    614 
    615 String8 NuCachedSource2::getUri() {
    616     return mSource->getUri();
    617 }
    618 
    619 String8 NuCachedSource2::getMIMEType() const {
    620     return mSource->getMIMEType();
    621 }
    622 
    623 void NuCachedSource2::updateCacheParamsFromSystemProperty() {
    624     char value[PROPERTY_VALUE_MAX];
    625     if (!property_get("media.stagefright.cache-params", value, NULL)) {
    626         return;
    627     }
    628 
    629     updateCacheParamsFromString(value);
    630 }
    631 
    632 void NuCachedSource2::updateCacheParamsFromString(const char *s) {
    633     ssize_t lowwaterMarkKb, highwaterMarkKb;
    634     int keepAliveSecs;
    635 
    636     if (sscanf(s, "%ld/%ld/%d",
    637                &lowwaterMarkKb, &highwaterMarkKb, &keepAliveSecs) != 3) {
    638         ALOGE("Failed to parse cache parameters from '%s'.", s);
    639         return;
    640     }
    641 
    642     if (lowwaterMarkKb >= 0) {
    643         mLowwaterThresholdBytes = lowwaterMarkKb * 1024;
    644     } else {
    645         mLowwaterThresholdBytes = kDefaultLowWaterThreshold;
    646     }
    647 
    648     if (highwaterMarkKb >= 0) {
    649         mHighwaterThresholdBytes = highwaterMarkKb * 1024;
    650     } else {
    651         mHighwaterThresholdBytes = kDefaultHighWaterThreshold;
    652     }
    653 
    654     if (mLowwaterThresholdBytes >= mHighwaterThresholdBytes) {
    655         ALOGE("Illegal low/highwater marks specified, reverting to defaults.");
    656 
    657         mLowwaterThresholdBytes = kDefaultLowWaterThreshold;
    658         mHighwaterThresholdBytes = kDefaultHighWaterThreshold;
    659     }
    660 
    661     if (keepAliveSecs >= 0) {
    662         mKeepAliveIntervalUs = keepAliveSecs * 1000000ll;
    663     } else {
    664         mKeepAliveIntervalUs = kDefaultKeepAliveIntervalUs;
    665     }
    666 
    667     ALOGV("lowwater = %d bytes, highwater = %d bytes, keepalive = %lld us",
    668          mLowwaterThresholdBytes,
    669          mHighwaterThresholdBytes,
    670          mKeepAliveIntervalUs);
    671 }
    672 
    673 // static
    674 void NuCachedSource2::RemoveCacheSpecificHeaders(
    675         KeyedVector<String8, String8> *headers,
    676         String8 *cacheConfig,
    677         bool *disconnectAtHighwatermark) {
    678     *cacheConfig = String8();
    679     *disconnectAtHighwatermark = false;
    680 
    681     if (headers == NULL) {
    682         return;
    683     }
    684 
    685     ssize_t index;
    686     if ((index = headers->indexOfKey(String8("x-cache-config"))) >= 0) {
    687         *cacheConfig = headers->valueAt(index);
    688 
    689         headers->removeItemsAt(index);
    690 
    691         ALOGV("Using special cache config '%s'", cacheConfig->string());
    692     }
    693 
    694     if ((index = headers->indexOfKey(
    695                     String8("x-disconnect-at-highwatermark"))) >= 0) {
    696         *disconnectAtHighwatermark = true;
    697         headers->removeItemsAt(index);
    698 
    699         ALOGV("Client requested disconnection at highwater mark");
    700     }
    701 }
    702 
    703 }  // namespace android
    704