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