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