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
29namespace android {
30
31struct 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
52private:
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
64PageCache::PageCache(size_t pageSize)
65    : mPageSize(pageSize),
66      mTotalSize(0) {
67}
68
69PageCache::~PageCache() {
70    freePages(&mActivePages);
71    freePages(&mFreePages);
72}
73
74void 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
87PageCache::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
103void PageCache::releasePage(Page *page) {
104    page->mSize = 0;
105    mFreePages.push_back(page);
106}
107
108void PageCache::appendPage(Page *page) {
109    mTotalSize += page->mSize;
110    mActivePages.push_back(page);
111}
112
113size_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
137void 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
180NuCachedSource2::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
222NuCachedSource2::~NuCachedSource2() {
223    mLooper->stop();
224    mLooper->unregisterHandler(mReflector->id());
225
226    delete mCache;
227    mCache = NULL;
228}
229
230status_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
238status_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
246status_t NuCachedSource2::initCheck() const {
247    return mSource->initCheck();
248}
249
250status_t NuCachedSource2::getSize(off64_t *size) {
251    return mSource->getSize(size);
252}
253
254uint32_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
260void 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
279void 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
342void 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
395void 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
424void 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
455ssize_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
498size_t NuCachedSource2::cachedSize() {
499    Mutex::Autolock autoLock(mLock);
500    return mCacheOffset + mCache->totalSize();
501}
502
503size_t NuCachedSource2::approxDataRemaining(status_t *finalStatus) {
504    Mutex::Autolock autoLock(mLock);
505    return approxDataRemaining_l(finalStatus);
506}
507
508size_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
523ssize_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
579status_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
600void NuCachedSource2::resumeFetchingIfNecessary() {
601    Mutex::Autolock autoLock(mLock);
602
603    restartPrefetcherIfNecessary_l(true /* ignore low water threshold */);
604}
605
606sp<DecryptHandle> NuCachedSource2::DrmInitialization() {
607    return mSource->DrmInitialization();
608}
609
610void NuCachedSource2::getDrmInfo(sp<DecryptHandle> &handle, DrmManagerClient **client) {
611    mSource->getDrmInfo(handle, client);
612}
613
614String8 NuCachedSource2::getUri() {
615    return mSource->getUri();
616}
617
618String8 NuCachedSource2::getMIMEType() const {
619    return mSource->getMIMEType();
620}
621
622void 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
631void 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
673void 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