media/bufferpool/aidl/default/Accessor.cpp

/*
 * Copyright (C) 2022 The Android Open Source Project
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#define LOG_TAG "AidlBufferPoolAcc"
//#define LOG_NDEBUG 0

#include <sys/types.h>
#include <stdint.h>
#include <time.h>
#include <unistd.h>
#include <utils/Log.h>
#include <thread>

#include "Accessor.h"
#include "Connection.h"
#include "DataHelper.h"

namespace aidl::android::hardware::media::bufferpool2::implementation {

namespace {
    static constexpr nsecs_t kEvictGranularityNs = 1000000000; // 1 sec
    static constexpr nsecs_t kEvictDurationNs = 5000000000; // 5 secs
}

#ifdef __ANDROID_VNDK__
static constexpr uint32_t kSeqIdVndkBit = 1U << 31;
#else
static constexpr uint32_t kSeqIdVndkBit = 0;
#endif

static constexpr uint32_t kSeqIdMax = 0x7fffffff;
uint32_t Accessor::sSeqId = time(nullptr) & kSeqIdMax;

namespace {
// anonymous namespace
static std::shared_ptr<ConnectionDeathRecipient> sConnectionDeathRecipient =
    std::make_shared<ConnectionDeathRecipient>();

void serviceDied(void *cookie) {
    if (sConnectionDeathRecipient) {
        sConnectionDeathRecipient->onDead(cookie);
    }
}
}

std::shared_ptr<ConnectionDeathRecipient> Accessor::getConnectionDeathRecipient() {
    return sConnectionDeathRecipient;
}

ConnectionDeathRecipient::ConnectionDeathRecipient() {
    mDeathRecipient = ndk::ScopedAIBinder_DeathRecipient(
            AIBinder_DeathRecipient_new(serviceDied));
}

void ConnectionDeathRecipient::add(
        int64_t connectionId,
        const std::shared_ptr<Accessor> &accessor) {
    std::lock_guard<std::mutex> lock(mLock);
    if (mAccessors.find(connectionId) == mAccessors.end()) {
        mAccessors.insert(std::make_pair(connectionId, accessor));
    }
}

void ConnectionDeathRecipient::remove(int64_t connectionId) {
    std::lock_guard<std::mutex> lock(mLock);
    mAccessors.erase(connectionId);
    auto it = mConnectionToCookie.find(connectionId);
    if (it != mConnectionToCookie.end()) {
        void * cookie = it->second;
        mConnectionToCookie.erase(it);
        auto cit = mCookieToConnections.find(cookie);
        if (cit != mCookieToConnections.end()) {
            cit->second.erase(connectionId);
            if (cit->second.size() == 0) {
                mCookieToConnections.erase(cit);
            }
        }
    }
}

void ConnectionDeathRecipient::addCookieToConnection(
        void *cookie,
        int64_t connectionId) {
    std::lock_guard<std::mutex> lock(mLock);
    if (mAccessors.find(connectionId) == mAccessors.end()) {
        return;
    }
    mConnectionToCookie.insert(std::make_pair(connectionId, cookie));
    auto it = mCookieToConnections.find(cookie);
    if (it != mCookieToConnections.end()) {
        it->second.insert(connectionId);
    } else {
        mCookieToConnections.insert(std::make_pair(
                cookie, std::set<int64_t>{connectionId}));
    }
}

void ConnectionDeathRecipient::onDead(void *cookie) {
    std::map<int64_t, const std::weak_ptr<Accessor>> connectionsToClose;
    {
        std::lock_guard<std::mutex> lock(mLock);

        auto it = mCookieToConnections.find(cookie);
        if (it != mCookieToConnections.end()) {
            for (auto conIt = it->second.begin(); conIt != it->second.end(); ++conIt) {
                auto accessorIt = mAccessors.find(*conIt);
                if (accessorIt != mAccessors.end()) {
                    connectionsToClose.insert(std::make_pair(*conIt, accessorIt->second));
                    mAccessors.erase(accessorIt);
                }
                mConnectionToCookie.erase(*conIt);
            }
            mCookieToConnections.erase(it);
        }
    }

    if (connectionsToClose.size() > 0) {
        std::shared_ptr<Accessor> accessor;
        for (auto it = connectionsToClose.begin(); it != connectionsToClose.end(); ++it) {
            accessor = it->second.lock();

            if (accessor) {
                accessor->close(it->first);
                ALOGD("connection %lld closed on death", (long long)it->first);
            }
        }
    }
}

AIBinder_DeathRecipient *ConnectionDeathRecipient::getRecipient() {
    return mDeathRecipient.get();
}

::ndk::ScopedAStatus Accessor::connect(const std::shared_ptr<::aidl::android::hardware::media::bufferpool2::IObserver>& in_observer, ::aidl::android::hardware::media::bufferpool2::IAccessor::ConnectionInfo* _aidl_return) {
    std::shared_ptr<Connection> connection;
    ConnectionId connectionId;
    uint32_t msgId;
    StatusDescriptor statusDesc;
    InvalidationDescriptor invDesc;
    BufferPoolStatus status = connect(
            in_observer, false, &connection, &connectionId, &msgId, &statusDesc, &invDesc);
    if (status == ResultStatus::OK) {
        _aidl_return->connection = connection;
        _aidl_return->connectionId = connectionId;
        _aidl_return->msgId = msgId;
        _aidl_return->toFmqDesc = std::move(statusDesc);
        _aidl_return->fromFmqDesc = std::move(invDesc);
        return ::ndk::ScopedAStatus::ok();
    }
    return ::ndk::ScopedAStatus::fromServiceSpecificError(status);
}

Accessor::Accessor(const std::shared_ptr<BufferPoolAllocator> &allocator)
    : mAllocator(allocator), mScheduleEvictTs(0) {}

Accessor::~Accessor() {
}

bool Accessor::isValid() {
    return mBufferPool.isValid();
}

BufferPoolStatus Accessor::flush() {
    std::lock_guard<std::mutex> lock(mBufferPool.mMutex);
    mBufferPool.processStatusMessages();
    mBufferPool.flush(ref<Accessor>());
    return ResultStatus::OK;
}

BufferPoolStatus Accessor::allocate(
        ConnectionId connectionId,
        const std::vector<uint8_t> &params,
        BufferId *bufferId, const native_handle_t** handle) {
    std::unique_lock<std::mutex> lock(mBufferPool.mMutex);
    mBufferPool.processStatusMessages();
    BufferPoolStatus status = ResultStatus::OK;
    if (!mBufferPool.getFreeBuffer(mAllocator, params, bufferId, handle)) {
        lock.unlock();
        std::shared_ptr<BufferPoolAllocation> alloc;
        size_t allocSize;
        status = mAllocator->allocate(params, &alloc, &allocSize);
        lock.lock();
        if (status == ResultStatus::OK) {
            status = mBufferPool.addNewBuffer(alloc, allocSize, params, bufferId, handle);
        }
        ALOGV("create a buffer %d : %u %p",
              status == ResultStatus::OK, *bufferId, *handle);
    }
    if (status == ResultStatus::OK) {
        // TODO: handle ownBuffer failure
        mBufferPool.handleOwnBuffer(connectionId, *bufferId);
    }
    mBufferPool.cleanUp();
    scheduleEvictIfNeeded();
    return status;
}

BufferPoolStatus Accessor::fetch(
        ConnectionId connectionId, TransactionId transactionId,
        BufferId bufferId, const native_handle_t** handle) {
    std::lock_guard<std::mutex> lock(mBufferPool.mMutex);
    mBufferPool.processStatusMessages();
    auto found = mBufferPool.mTransactions.find(transactionId);
    if (found != mBufferPool.mTransactions.end() &&
            contains(&mBufferPool.mPendingTransactions,
                     connectionId, transactionId)) {
        if (found->second->mSenderValidated &&
                found->second->mStatus == BufferStatus::TRANSFER_FROM &&
                found->second->mBufferId == bufferId) {
            found->second->mStatus = BufferStatus::TRANSFER_FETCH;
            auto bufferIt = mBufferPool.mBuffers.find(bufferId);
            if (bufferIt != mBufferPool.mBuffers.end()) {
                mBufferPool.mStats.onBufferFetched();
                *handle = bufferIt->second->handle();
                return ResultStatus::OK;
            }
        }
    }
    mBufferPool.cleanUp();
    scheduleEvictIfNeeded();
    return ResultStatus::CRITICAL_ERROR;
}

BufferPoolStatus Accessor::connect(
        const std::shared_ptr<IObserver> &observer, bool local,
        std::shared_ptr<Connection> *connection, ConnectionId *pConnectionId,
        uint32_t *pMsgId,
        StatusDescriptor* statusDescPtr,
        InvalidationDescriptor* invDescPtr) {
    std::shared_ptr<Connection> newConnection = ::ndk::SharedRefBase::make<Connection>();
    BufferPoolStatus status = ResultStatus::CRITICAL_ERROR;
    {
        std::lock_guard<std::mutex> lock(mBufferPool.mMutex);
        if (newConnection) {
            int32_t pid = getpid();
            ConnectionId id = (int64_t)pid << 32 | sSeqId | kSeqIdVndkBit;
            status = mBufferPool.mObserver.open(id, statusDescPtr);
            if (status == ResultStatus::OK) {
                newConnection->initialize(ref<Accessor>(), id);
                *connection = newConnection;
                *pConnectionId = id;
                *pMsgId = mBufferPool.mInvalidation.mInvalidationId;
                mBufferPool.mConnectionIds.insert(id);
                mBufferPool.mInvalidationChannel.getDesc(invDescPtr);
                mBufferPool.mInvalidation.onConnect(id, observer);
                if (sSeqId == kSeqIdMax) {
                   sSeqId = 0;
                } else {
                    ++sSeqId;
                }
            }

        }
        mBufferPool.processStatusMessages();
        mBufferPool.cleanUp();
        scheduleEvictIfNeeded();
    }
    if (!local && status == ResultStatus::OK) {
        std::shared_ptr<Accessor> accessor(ref<Accessor>());
        sConnectionDeathRecipient->add(*pConnectionId, accessor);
    }
    return status;
}

BufferPoolStatus Accessor::close(ConnectionId connectionId) {
    {
        std::lock_guard<std::mutex> lock(mBufferPool.mMutex);
        ALOGV("connection close %lld: %u", (long long)connectionId, mBufferPool.mInvalidation.mId);
        mBufferPool.processStatusMessages();
        mBufferPool.handleClose(connectionId);
        mBufferPool.mObserver.close(connectionId);
        mBufferPool.mInvalidation.onClose(connectionId);
        // Since close# will be called after all works are finished, it is OK to
        // evict unused buffers.
        mBufferPool.cleanUp(true);
        scheduleEvictIfNeeded();
    }
    sConnectionDeathRecipient->remove(connectionId);
    return ResultStatus::OK;
}

void Accessor::cleanUp(bool clearCache) {
    // transaction timeout, buffer caching TTL handling
    std::lock_guard<std::mutex> lock(mBufferPool.mMutex);
    mBufferPool.processStatusMessages();
    mBufferPool.cleanUp(clearCache);
}

void Accessor::handleInvalidateAck() {
    std::map<ConnectionId, const std::shared_ptr<IObserver>> observers;
    uint32_t invalidationId;
    {
        std::lock_guard<std::mutex> lock(mBufferPool.mMutex);
        mBufferPool.processStatusMessages();
        mBufferPool.mInvalidation.onHandleAck(&observers, &invalidationId);
    }
    // Do not hold lock for send invalidations
    size_t deadClients = 0;
    for (auto it = observers.begin(); it != observers.end(); ++it) {
        const std::shared_ptr<IObserver> observer = it->second;
        if (observer) {
            ::ndk::ScopedAStatus status = observer->onMessage(it->first, invalidationId);
            if (!status.isOk()) {
                ++deadClients;
            }
        }
    }
    if (deadClients > 0) {
        ALOGD("During invalidation found %zu dead clients", deadClients);
    }
}

void Accessor::invalidatorThread(
            std::map<uint32_t, const std::weak_ptr<Accessor>> &accessors,
            std::mutex &mutex,
            std::condition_variable &cv,
            bool &ready) {
    constexpr uint32_t NUM_SPIN_TO_INCREASE_SLEEP = 1024;
    constexpr uint32_t NUM_SPIN_TO_LOG = 1024*8;
    constexpr useconds_t MAX_SLEEP_US = 10000;
    uint32_t numSpin = 0;
    useconds_t sleepUs = 1;

    while(true) {
        std::map<uint32_t, const std::weak_ptr<Accessor>> copied;
        {
            std::unique_lock<std::mutex> lock(mutex);
            while (!ready) {
                numSpin = 0;
                sleepUs = 1;
                cv.wait(lock);
            }
            copied.insert(accessors.begin(), accessors.end());
        }
        std::list<ConnectionId> erased;
        for (auto it = copied.begin(); it != copied.end(); ++it) {
            const std::shared_ptr<Accessor> acc = it->second.lock();
            if (!acc) {
                erased.push_back(it->first);
            } else {
                acc->handleInvalidateAck();
            }
        }
        {
            std::unique_lock<std::mutex> lock(mutex);
            for (auto it = erased.begin(); it != erased.end(); ++it) {
                accessors.erase(*it);
            }
            if (accessors.size() == 0) {
                ready = false;
            } else {
                // N.B. Since there is not a efficient way to wait over FMQ,
                // polling over the FMQ is the current way to prevent draining
                // CPU.
                lock.unlock();
                ++numSpin;
                if (numSpin % NUM_SPIN_TO_INCREASE_SLEEP == 0 &&
                    sleepUs < MAX_SLEEP_US) {
                    sleepUs *= 10;
                }
                if (numSpin % NUM_SPIN_TO_LOG == 0) {
                    ALOGW("invalidator thread spinning");
                }
                ::usleep(sleepUs);
            }
        }
    }
}

Accessor::AccessorInvalidator::AccessorInvalidator() : mReady(false) {
    std::thread invalidator(
            invalidatorThread,
            std::ref(mAccessors),
            std::ref(mMutex),
            std::ref(mCv),
            std::ref(mReady));
    invalidator.detach();
}

void Accessor::AccessorInvalidator::addAccessor(
        uint32_t accessorId, const std::weak_ptr<Accessor> &accessor) {
    bool notify = false;
    std::unique_lock<std::mutex> lock(mMutex);
    if (mAccessors.find(accessorId) == mAccessors.end()) {
        if (!mReady) {
            mReady = true;
            notify = true;
        }
        mAccessors.emplace(accessorId, accessor);
        ALOGV("buffer invalidation added bp:%u %d", accessorId, notify);
    }
    lock.unlock();
    if (notify) {
        mCv.notify_one();
    }
}

void Accessor::AccessorInvalidator::delAccessor(uint32_t accessorId) {
    std::lock_guard<std::mutex> lock(mMutex);
    mAccessors.erase(accessorId);
    ALOGV("buffer invalidation deleted bp:%u", accessorId);
    if (mAccessors.size() == 0) {
        mReady = false;
    }
}

std::unique_ptr<Accessor::AccessorInvalidator> Accessor::sInvalidator;

void Accessor::createInvalidator() {
    if (!sInvalidator) {
        sInvalidator = std::make_unique<Accessor::AccessorInvalidator>();
    }
}

void Accessor::evictorThread(
        std::map<const std::weak_ptr<Accessor>, nsecs_t, std::owner_less<>> &accessors,
        std::mutex &mutex,
        std::condition_variable &cv) {
    std::list<const std::weak_ptr<Accessor>> evictList;
    while (true) {
        int expired = 0;
        int evicted = 0;
        {
            nsecs_t now = systemTime();
            std::unique_lock<std::mutex> lock(mutex);
            while (accessors.size() == 0) {
                cv.wait(lock);
            }
            auto it = accessors.begin();
            while (it != accessors.end()) {
                if (now > (it->second + kEvictDurationNs)) {
                    ++expired;
                    evictList.push_back(it->first);
                    it = accessors.erase(it);
                } else {
                    ++it;
                }
            }
        }
        // evict idle accessors;
        for (auto it = evictList.begin(); it != evictList.end(); ++it) {
            const std::shared_ptr<Accessor> accessor = it->lock();
            if (accessor) {
                accessor->cleanUp(true);
                ++evicted;
            }
        }
        if (expired > 0) {
            ALOGD("evictor expired: %d, evicted: %d", expired, evicted);
        }
        evictList.clear();
        ::usleep(kEvictGranularityNs / 1000);
    }
}

Accessor::AccessorEvictor::AccessorEvictor() {
    std::thread evictor(
            evictorThread,
            std::ref(mAccessors),
            std::ref(mMutex),
            std::ref(mCv));
    evictor.detach();
}

void Accessor::AccessorEvictor::addAccessor(
        const std::weak_ptr<Accessor> &accessor, nsecs_t ts) {
    std::lock_guard<std::mutex> lock(mMutex);
    bool notify = mAccessors.empty();
    auto it = mAccessors.find(accessor);
    if (it == mAccessors.end()) {
        mAccessors.emplace(accessor, ts);
    } else {
        it->second = ts;
    }
    if (notify) {
        mCv.notify_one();
    }
}

std::unique_ptr<Accessor::AccessorEvictor> Accessor::sEvictor;

void Accessor::createEvictor() {
    if (!sEvictor) {
        sEvictor = std::make_unique<Accessor::AccessorEvictor>();
    }
}

void Accessor::scheduleEvictIfNeeded() {
    nsecs_t now = systemTime();

    if (now > (mScheduleEvictTs + kEvictGranularityNs)) {
        mScheduleEvictTs = now;
        sEvictor->addAccessor(ref<Accessor>(), now);
    }
}

}  // namespace aidl::android::hardware::media::bufferpool2::implemntation {