1 /* 2 * Copyright (C) 2018 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_TAG "BufferPoolClient" 18 //#define LOG_NDEBUG 0 19 20 #include <thread> 21 #include <utils/Log.h> 22 #include "BufferPoolClient.h" 23 #include "Connection.h" 24 25 namespace android { 26 namespace hardware { 27 namespace media { 28 namespace bufferpool { 29 namespace V1_0 { 30 namespace implementation { 31 32 static constexpr int64_t kReceiveTimeoutUs = 1000000; // 100ms 33 static constexpr int kPostMaxRetry = 3; 34 static constexpr int kCacheTtlUs = 1000000; // TODO: tune 35 36 class BufferPoolClient::Impl 37 : public std::enable_shared_from_this<BufferPoolClient::Impl> { 38 public: 39 explicit Impl(const sp<Accessor> &accessor); 40 41 explicit Impl(const sp<IAccessor> &accessor); 42 43 bool isValid() { 44 return mValid; 45 } 46 47 bool isLocal() { 48 return mValid && mLocal; 49 } 50 51 ConnectionId getConnectionId() { 52 return mConnectionId; 53 } 54 55 sp<IAccessor> &getAccessor() { 56 return mAccessor; 57 } 58 59 bool isActive(int64_t *lastTransactionUs, bool clearCache); 60 61 ResultStatus allocate(const std::vector<uint8_t> ¶ms, 62 native_handle_t **handle, 63 std::shared_ptr<BufferPoolData> *buffer); 64 65 ResultStatus receive( 66 TransactionId transactionId, BufferId bufferId, 67 int64_t timestampUs, 68 native_handle_t **handle, std::shared_ptr<BufferPoolData> *buffer); 69 70 void postBufferRelease(BufferId bufferId); 71 72 bool postSend( 73 BufferId bufferId, ConnectionId receiver, 74 TransactionId *transactionId, int64_t *timestampUs); 75 private: 76 77 bool postReceive( 78 BufferId bufferId, TransactionId transactionId, 79 int64_t timestampUs); 80 81 bool postReceiveResult( 82 BufferId bufferId, TransactionId transactionId, bool result); 83 84 bool syncReleased(); 85 86 void evictCaches(bool clearCache = false); 87 88 ResultStatus allocateBufferHandle( 89 const std::vector<uint8_t>& params, BufferId *bufferId, 90 native_handle_t **handle); 91 92 ResultStatus fetchBufferHandle( 93 TransactionId transactionId, BufferId bufferId, 94 native_handle_t **handle); 95 96 97 struct BlockPoolDataDtor; 98 struct ClientBuffer; 99 100 bool mLocal; 101 bool mValid; 102 sp<IAccessor> mAccessor; 103 sp<Connection> mLocalConnection; 104 sp<IConnection> mRemoteConnection; 105 uint32_t mSeqId; 106 ConnectionId mConnectionId; 107 int64_t mLastEvictCacheUs; 108 109 // CachedBuffers 110 struct BufferCache { 111 std::mutex mLock; 112 bool mCreating; 113 std::condition_variable mCreateCv; 114 std::map<BufferId, std::unique_ptr<ClientBuffer>> mBuffers; 115 int mActive; 116 int64_t mLastChangeUs; 117 118 BufferCache() : mCreating(false), mActive(0), mLastChangeUs(getTimestampNow()) {} 119 120 void incActive_l() { 121 ++mActive; 122 mLastChangeUs = getTimestampNow(); 123 } 124 125 void decActive_l() { 126 --mActive; 127 mLastChangeUs = getTimestampNow(); 128 } 129 } mCache; 130 131 // FMQ - release notifier 132 struct { 133 std::mutex mLock; 134 // TODO: use only one list?(using one list may dealy sending messages?) 135 std::list<BufferId> mReleasingIds; 136 std::list<BufferId> mReleasedIds; 137 std::unique_ptr<BufferStatusChannel> mStatusChannel; 138 } mReleasing; 139 }; 140 141 struct BufferPoolClient::Impl::BlockPoolDataDtor { 142 BlockPoolDataDtor(const std::shared_ptr<BufferPoolClient::Impl> &impl) 143 : mImpl(impl) {} 144 145 void operator()(BufferPoolData *buffer) { 146 BufferId id = buffer->mId; 147 delete buffer; 148 149 auto impl = mImpl.lock(); 150 if (impl && impl->isValid()) { 151 impl->postBufferRelease(id); 152 } 153 } 154 const std::weak_ptr<BufferPoolClient::Impl> mImpl; 155 }; 156 157 struct BufferPoolClient::Impl::ClientBuffer { 158 private: 159 bool mInvalidated; // TODO: implement 160 int64_t mExpireUs; 161 bool mHasCache; 162 ConnectionId mConnectionId; 163 BufferId mId; 164 native_handle_t *mHandle; 165 std::weak_ptr<BufferPoolData> mCache; 166 167 void updateExpire() { 168 mExpireUs = getTimestampNow() + kCacheTtlUs; 169 } 170 171 public: 172 ClientBuffer( 173 ConnectionId connectionId, BufferId id, native_handle_t *handle) 174 : mInvalidated(false), mHasCache(false), 175 mConnectionId(connectionId), mId(id), mHandle(handle) { 176 (void)mInvalidated; 177 mExpireUs = getTimestampNow() + kCacheTtlUs; 178 } 179 180 ~ClientBuffer() { 181 if (mHandle) { 182 native_handle_close(mHandle); 183 native_handle_delete(mHandle); 184 } 185 } 186 187 bool expire() const { 188 int64_t now = getTimestampNow(); 189 return now >= mExpireUs; 190 } 191 192 bool hasCache() const { 193 return mHasCache; 194 } 195 196 std::shared_ptr<BufferPoolData> fetchCache(native_handle_t **pHandle) { 197 if (mHasCache) { 198 std::shared_ptr<BufferPoolData> cache = mCache.lock(); 199 if (cache) { 200 *pHandle = mHandle; 201 } 202 return cache; 203 } 204 return nullptr; 205 } 206 207 std::shared_ptr<BufferPoolData> createCache( 208 const std::shared_ptr<BufferPoolClient::Impl> &impl, 209 native_handle_t **pHandle) { 210 if (!mHasCache) { 211 // Allocates a raw ptr in order to avoid sending #postBufferRelease 212 // from deleter, in case of native_handle_clone failure. 213 BufferPoolData *ptr = new BufferPoolData(mConnectionId, mId); 214 if (ptr) { 215 std::shared_ptr<BufferPoolData> cache(ptr, BlockPoolDataDtor(impl)); 216 if (cache) { 217 mCache = cache; 218 mHasCache = true; 219 *pHandle = mHandle; 220 return cache; 221 } 222 } 223 if (ptr) { 224 delete ptr; 225 } 226 } 227 return nullptr; 228 } 229 230 bool onCacheRelease() { 231 if (mHasCache) { 232 // TODO: verify mCache is not valid; 233 updateExpire(); 234 mHasCache = false; 235 return true; 236 } 237 return false; 238 } 239 }; 240 241 BufferPoolClient::Impl::Impl(const sp<Accessor> &accessor) 242 : mLocal(true), mValid(false), mAccessor(accessor), mSeqId(0), 243 mLastEvictCacheUs(getTimestampNow()) { 244 const QueueDescriptor *fmqDesc; 245 ResultStatus status = accessor->connect( 246 &mLocalConnection, &mConnectionId, &fmqDesc, true); 247 if (status == ResultStatus::OK) { 248 mReleasing.mStatusChannel = 249 std::make_unique<BufferStatusChannel>(*fmqDesc); 250 mValid = mReleasing.mStatusChannel && 251 mReleasing.mStatusChannel->isValid(); 252 } 253 } 254 255 BufferPoolClient::Impl::Impl(const sp<IAccessor> &accessor) 256 : mLocal(false), mValid(false), mAccessor(accessor), mSeqId(0), 257 mLastEvictCacheUs(getTimestampNow()) { 258 bool valid = false; 259 sp<IConnection>& outConnection = mRemoteConnection; 260 ConnectionId& id = mConnectionId; 261 std::unique_ptr<BufferStatusChannel>& outChannel = 262 mReleasing.mStatusChannel; 263 Return<void> transResult = accessor->connect( 264 [&valid, &outConnection, &id, &outChannel] 265 (ResultStatus status, sp<IConnection> connection, 266 ConnectionId connectionId, const QueueDescriptor& desc) { 267 if (status == ResultStatus::OK) { 268 outConnection = connection; 269 id = connectionId; 270 outChannel = std::make_unique<BufferStatusChannel>(desc); 271 if (outChannel && outChannel->isValid()) { 272 valid = true; 273 } 274 } 275 }); 276 mValid = transResult.isOk() && valid; 277 } 278 279 bool BufferPoolClient::Impl::isActive(int64_t *lastTransactionUs, bool clearCache) { 280 { 281 std::lock_guard<std::mutex> lock(mCache.mLock); 282 syncReleased(); 283 evictCaches(clearCache); 284 *lastTransactionUs = mCache.mLastChangeUs; 285 } 286 if (mValid && mLocal && mLocalConnection) { 287 mLocalConnection->cleanUp(clearCache); 288 return true; 289 } 290 return mCache.mActive > 0; 291 } 292 293 ResultStatus BufferPoolClient::Impl::allocate( 294 const std::vector<uint8_t> ¶ms, 295 native_handle_t **pHandle, 296 std::shared_ptr<BufferPoolData> *buffer) { 297 if (!mLocal || !mLocalConnection || !mValid) { 298 return ResultStatus::CRITICAL_ERROR; 299 } 300 BufferId bufferId; 301 native_handle_t *handle = NULL; 302 buffer->reset(); 303 ResultStatus status = allocateBufferHandle(params, &bufferId, &handle); 304 if (status == ResultStatus::OK) { 305 if (handle) { 306 std::unique_lock<std::mutex> lock(mCache.mLock); 307 syncReleased(); 308 evictCaches(); 309 auto cacheIt = mCache.mBuffers.find(bufferId); 310 if (cacheIt != mCache.mBuffers.end()) { 311 // TODO: verify it is recycled. (not having active ref) 312 mCache.mBuffers.erase(cacheIt); 313 } 314 auto clientBuffer = std::make_unique<ClientBuffer>( 315 mConnectionId, bufferId, handle); 316 if (clientBuffer) { 317 auto result = mCache.mBuffers.insert(std::make_pair( 318 bufferId, std::move(clientBuffer))); 319 if (result.second) { 320 *buffer = result.first->second->createCache( 321 shared_from_this(), pHandle); 322 if (*buffer) { 323 mCache.incActive_l(); 324 } 325 } 326 } 327 } 328 if (!*buffer) { 329 ALOGV("client cache creation failure %d: %lld", 330 handle != NULL, (long long)mConnectionId); 331 status = ResultStatus::NO_MEMORY; 332 postBufferRelease(bufferId); 333 } 334 } 335 return status; 336 } 337 338 ResultStatus BufferPoolClient::Impl::receive( 339 TransactionId transactionId, BufferId bufferId, int64_t timestampUs, 340 native_handle_t **pHandle, 341 std::shared_ptr<BufferPoolData> *buffer) { 342 if (!mValid) { 343 return ResultStatus::CRITICAL_ERROR; 344 } 345 if (timestampUs != 0) { 346 timestampUs += kReceiveTimeoutUs; 347 } 348 if (!postReceive(bufferId, transactionId, timestampUs)) { 349 return ResultStatus::CRITICAL_ERROR; 350 } 351 ResultStatus status = ResultStatus::CRITICAL_ERROR; 352 buffer->reset(); 353 while(1) { 354 std::unique_lock<std::mutex> lock(mCache.mLock); 355 syncReleased(); 356 evictCaches(); 357 auto cacheIt = mCache.mBuffers.find(bufferId); 358 if (cacheIt != mCache.mBuffers.end()) { 359 if (cacheIt->second->hasCache()) { 360 *buffer = cacheIt->second->fetchCache(pHandle); 361 if (!*buffer) { 362 // check transfer time_out 363 lock.unlock(); 364 std::this_thread::yield(); 365 continue; 366 } 367 ALOGV("client receive from reference %lld", (long long)mConnectionId); 368 break; 369 } else { 370 *buffer = cacheIt->second->createCache(shared_from_this(), pHandle); 371 if (*buffer) { 372 mCache.incActive_l(); 373 } 374 ALOGV("client receive from cache %lld", (long long)mConnectionId); 375 break; 376 } 377 } else { 378 if (!mCache.mCreating) { 379 mCache.mCreating = true; 380 lock.unlock(); 381 native_handle_t* handle = NULL; 382 status = fetchBufferHandle(transactionId, bufferId, &handle); 383 lock.lock(); 384 if (status == ResultStatus::OK) { 385 if (handle) { 386 auto clientBuffer = std::make_unique<ClientBuffer>( 387 mConnectionId, bufferId, handle); 388 if (clientBuffer) { 389 auto result = mCache.mBuffers.insert( 390 std::make_pair(bufferId, std::move( 391 clientBuffer))); 392 if (result.second) { 393 *buffer = result.first->second->createCache( 394 shared_from_this(), pHandle); 395 if (*buffer) { 396 mCache.incActive_l(); 397 } 398 } 399 } 400 } 401 if (!*buffer) { 402 status = ResultStatus::NO_MEMORY; 403 } 404 } 405 mCache.mCreating = false; 406 lock.unlock(); 407 mCache.mCreateCv.notify_all(); 408 break; 409 } 410 mCache.mCreateCv.wait(lock); 411 } 412 } 413 bool posted = postReceiveResult(bufferId, transactionId, 414 *buffer ? true : false); 415 ALOGV("client receive %lld - %u : %s (%d)", (long long)mConnectionId, bufferId, 416 *buffer ? "ok" : "fail", posted); 417 if (mValid && mLocal && mLocalConnection) { 418 mLocalConnection->cleanUp(false); 419 } 420 if (*buffer) { 421 if (!posted) { 422 buffer->reset(); 423 return ResultStatus::CRITICAL_ERROR; 424 } 425 return ResultStatus::OK; 426 } 427 return status; 428 } 429 430 431 void BufferPoolClient::Impl::postBufferRelease(BufferId bufferId) { 432 std::lock_guard<std::mutex> lock(mReleasing.mLock); 433 mReleasing.mReleasingIds.push_back(bufferId); 434 mReleasing.mStatusChannel->postBufferRelease( 435 mConnectionId, mReleasing.mReleasingIds, mReleasing.mReleasedIds); 436 } 437 438 // TODO: revise ad-hoc posting data structure 439 bool BufferPoolClient::Impl::postSend( 440 BufferId bufferId, ConnectionId receiver, 441 TransactionId *transactionId, int64_t *timestampUs) { 442 bool ret = false; 443 { 444 std::lock_guard<std::mutex> lock(mReleasing.mLock); 445 *timestampUs = getTimestampNow(); 446 *transactionId = (mConnectionId << 32) | mSeqId++; 447 // TODO: retry, add timeout, target? 448 ret = mReleasing.mStatusChannel->postBufferStatusMessage( 449 *transactionId, bufferId, BufferStatus::TRANSFER_TO, mConnectionId, 450 receiver, mReleasing.mReleasingIds, mReleasing.mReleasedIds); 451 } 452 if (mValid && mLocal && mLocalConnection) { 453 mLocalConnection->cleanUp(false); 454 } 455 return ret; 456 } 457 458 bool BufferPoolClient::Impl::postReceive( 459 BufferId bufferId, TransactionId transactionId, int64_t timestampUs) { 460 for (int i = 0; i < kPostMaxRetry; ++i) { 461 std::unique_lock<std::mutex> lock(mReleasing.mLock); 462 int64_t now = getTimestampNow(); 463 if (timestampUs == 0 || now < timestampUs) { 464 bool result = mReleasing.mStatusChannel->postBufferStatusMessage( 465 transactionId, bufferId, BufferStatus::TRANSFER_FROM, 466 mConnectionId, -1, mReleasing.mReleasingIds, 467 mReleasing.mReleasedIds); 468 if (result) { 469 return true; 470 } 471 lock.unlock(); 472 std::this_thread::yield(); 473 } else { 474 mReleasing.mStatusChannel->postBufferStatusMessage( 475 transactionId, bufferId, BufferStatus::TRANSFER_TIMEOUT, 476 mConnectionId, -1, mReleasing.mReleasingIds, 477 mReleasing.mReleasedIds); 478 return false; 479 } 480 } 481 return false; 482 } 483 484 bool BufferPoolClient::Impl::postReceiveResult( 485 BufferId bufferId, TransactionId transactionId, bool result) { 486 std::lock_guard<std::mutex> lock(mReleasing.mLock); 487 // TODO: retry, add timeout 488 return mReleasing.mStatusChannel->postBufferStatusMessage( 489 transactionId, bufferId, 490 result ? BufferStatus::TRANSFER_OK : BufferStatus::TRANSFER_ERROR, 491 mConnectionId, -1, mReleasing.mReleasingIds, 492 mReleasing.mReleasedIds); 493 } 494 495 // should have mCache.mLock 496 bool BufferPoolClient::Impl::syncReleased() { 497 std::lock_guard<std::mutex> lock(mReleasing.mLock); 498 if (mReleasing.mReleasingIds.size() > 0) { 499 mReleasing.mStatusChannel->postBufferRelease( 500 mConnectionId, mReleasing.mReleasingIds, 501 mReleasing.mReleasedIds); 502 } 503 if (mReleasing.mReleasedIds.size() > 0) { 504 for (BufferId& id: mReleasing.mReleasedIds) { 505 ALOGV("client release buffer %lld - %u", (long long)mConnectionId, id); 506 auto found = mCache.mBuffers.find(id); 507 if (found != mCache.mBuffers.end()) { 508 if (found->second->onCacheRelease()) { 509 mCache.decActive_l(); 510 } else { 511 // should not happen! 512 ALOGW("client %lld cache release status inconsitent!", 513 (long long)mConnectionId); 514 } 515 } else { 516 // should not happen! 517 ALOGW("client %lld cache status inconsitent!", (long long)mConnectionId); 518 } 519 } 520 mReleasing.mReleasedIds.clear(); 521 return true; 522 } 523 return false; 524 } 525 526 // should have mCache.mLock 527 void BufferPoolClient::Impl::evictCaches(bool clearCache) { 528 int64_t now = getTimestampNow(); 529 if (now >= mLastEvictCacheUs + kCacheTtlUs || clearCache) { 530 size_t evicted = 0; 531 for (auto it = mCache.mBuffers.begin(); it != mCache.mBuffers.end();) { 532 if (!it->second->hasCache() && (it->second->expire() || clearCache)) { 533 it = mCache.mBuffers.erase(it); 534 ++evicted; 535 } else { 536 ++it; 537 } 538 } 539 ALOGV("cache count %lld : total %zu, active %d, evicted %zu", 540 (long long)mConnectionId, mCache.mBuffers.size(), mCache.mActive, evicted); 541 mLastEvictCacheUs = now; 542 } 543 } 544 545 ResultStatus BufferPoolClient::Impl::allocateBufferHandle( 546 const std::vector<uint8_t>& params, BufferId *bufferId, 547 native_handle_t** handle) { 548 if (mLocalConnection) { 549 const native_handle_t* allocHandle = NULL; 550 ResultStatus status = mLocalConnection->allocate( 551 params, bufferId, &allocHandle); 552 if (status == ResultStatus::OK) { 553 *handle = native_handle_clone(allocHandle); 554 } 555 ALOGV("client allocate result %lld %d : %u clone %p", 556 (long long)mConnectionId, status == ResultStatus::OK, 557 *handle ? *bufferId : 0 , *handle); 558 return status; 559 } 560 return ResultStatus::CRITICAL_ERROR; 561 } 562 563 ResultStatus BufferPoolClient::Impl::fetchBufferHandle( 564 TransactionId transactionId, BufferId bufferId, 565 native_handle_t **handle) { 566 sp<IConnection> connection; 567 if (mLocal) { 568 connection = mLocalConnection; 569 } else { 570 connection = mRemoteConnection; 571 } 572 ResultStatus status; 573 Return<void> transResult = connection->fetch( 574 transactionId, bufferId, 575 [&status, &handle] 576 (ResultStatus outStatus, Buffer outBuffer) { 577 status = outStatus; 578 if (status == ResultStatus::OK) { 579 *handle = native_handle_clone( 580 outBuffer.buffer.getNativeHandle()); 581 } 582 }); 583 return transResult.isOk() ? status : ResultStatus::CRITICAL_ERROR; 584 } 585 586 587 BufferPoolClient::BufferPoolClient(const sp<Accessor> &accessor) { 588 mImpl = std::make_shared<Impl>(accessor); 589 } 590 591 BufferPoolClient::BufferPoolClient(const sp<IAccessor> &accessor) { 592 mImpl = std::make_shared<Impl>(accessor); 593 } 594 595 BufferPoolClient::~BufferPoolClient() { 596 // TODO: how to handle orphaned buffers? 597 } 598 599 bool BufferPoolClient::isValid() { 600 return mImpl && mImpl->isValid(); 601 } 602 603 bool BufferPoolClient::isLocal() { 604 return mImpl && mImpl->isLocal(); 605 } 606 607 bool BufferPoolClient::isActive(int64_t *lastTransactionUs, bool clearCache) { 608 if (!isValid()) { 609 *lastTransactionUs = 0; 610 return false; 611 } 612 return mImpl->isActive(lastTransactionUs, clearCache); 613 } 614 615 ConnectionId BufferPoolClient::getConnectionId() { 616 if (isValid()) { 617 return mImpl->getConnectionId(); 618 } 619 return -1; 620 } 621 622 ResultStatus BufferPoolClient::getAccessor(sp<IAccessor> *accessor) { 623 if (isValid()) { 624 *accessor = mImpl->getAccessor(); 625 return ResultStatus::OK; 626 } 627 return ResultStatus::CRITICAL_ERROR; 628 } 629 630 ResultStatus BufferPoolClient::allocate( 631 const std::vector<uint8_t> ¶ms, 632 native_handle_t **handle, 633 std::shared_ptr<BufferPoolData> *buffer) { 634 if (isValid()) { 635 return mImpl->allocate(params, handle, buffer); 636 } 637 return ResultStatus::CRITICAL_ERROR; 638 } 639 640 ResultStatus BufferPoolClient::receive( 641 TransactionId transactionId, BufferId bufferId, int64_t timestampUs, 642 native_handle_t **handle, std::shared_ptr<BufferPoolData> *buffer) { 643 if (isValid()) { 644 return mImpl->receive(transactionId, bufferId, timestampUs, handle, buffer); 645 } 646 return ResultStatus::CRITICAL_ERROR; 647 } 648 649 ResultStatus BufferPoolClient::postSend( 650 ConnectionId receiverId, 651 const std::shared_ptr<BufferPoolData> &buffer, 652 TransactionId *transactionId, 653 int64_t *timestampUs) { 654 if (isValid()) { 655 bool result = mImpl->postSend( 656 buffer->mId, receiverId, transactionId, timestampUs); 657 return result ? ResultStatus::OK : ResultStatus::CRITICAL_ERROR; 658 } 659 return ResultStatus::CRITICAL_ERROR; 660 } 661 662 } // namespace implementation 663 } // namespace V1_0 664 } // namespace bufferpool 665 } // namespace media 666 } // namespace hardware 667 } // namespace android 668