NuCachedSource2.cpp revision a53d87c7b1428fe02f535c31dafd64cb1362fde9 
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    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
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    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
342void 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
396void 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
425void 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
456ssize_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
499size_t NuCachedSource2::cachedSize() {
500    Mutex::Autolock autoLock(mLock);
501    return mCacheOffset + mCache->totalSize();
502}
503
504size_t NuCachedSource2::approxDataRemaining(status_t *finalStatus) const {
505    Mutex::Autolock autoLock(mLock);
506    return approxDataRemaining_l(finalStatus);
507}
508
509size_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
524ssize_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
580status_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
601void NuCachedSource2::resumeFetchingIfNecessary() {
602    Mutex::Autolock autoLock(mLock);
603
604    restartPrefetcherIfNecessary_l(true /* ignore low water threshold */);
605}
606
607sp<DecryptHandle> NuCachedSource2::DrmInitialization(const char* mime) {
608    return mSource->DrmInitialization(mime);
609}
610
611void NuCachedSource2::getDrmInfo(sp<DecryptHandle> &handle, DrmManagerClient **client) {
612    mSource->getDrmInfo(handle, client);
613}
614
615String8 NuCachedSource2::getUri() {
616    return mSource->getUri();
617}
618
619String8 NuCachedSource2::getMIMEType() const {
620    return mSource->getMIMEType();
621}
622
623void 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
632void 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
674void 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