media/libmediatranscoding/TranscodingSessionController.cpp

/*
 * Copyright (C) 2020 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_NDEBUG 0
#define LOG_TAG "TranscodingSessionController"

#define VALIDATE_STATE 1

#include <inttypes.h>
#include <media/TranscodingSessionController.h>
#include <media/TranscodingUidPolicy.h>
#include <utils/Log.h>

#include <thread>
#include <utility>

namespace android {

static_assert((SessionIdType)-1 < 0, "SessionIdType should be signed");

constexpr static uid_t OFFLINE_UID = -1;
constexpr static size_t kSessionHistoryMax = 100;

//static
String8 TranscodingSessionController::sessionToString(const SessionKeyType& sessionKey) {
    return String8::format("{client:%lld, session:%d}", (long long)sessionKey.first,
                           sessionKey.second);
}

//static
const char* TranscodingSessionController::sessionStateToString(const Session::State sessionState) {
    switch (sessionState) {
    case Session::State::NOT_STARTED:
        return "NOT_STARTED";
    case Session::State::RUNNING:
        return "RUNNING";
    case Session::State::PAUSED:
        return "PAUSED";
    case Session::State::FINISHED:
        return "FINISHED";
    case Session::State::CANCELED:
        return "CANCELED";
    case Session::State::ERROR:
        return "ERROR";
    default:
        break;
    }
    return "(unknown)";
}

///////////////////////////////////////////////////////////////////////////////
struct TranscodingSessionController::Watchdog {
    Watchdog(TranscodingSessionController* owner, int64_t timeoutUs);
    ~Watchdog();

    // Starts monitoring the session.
    void start(const SessionKeyType& key);
    // Stops monitoring the session.
    void stop();
    // Signals that the session is still alive. Must be sent at least every mTimeoutUs.
    // (Timeout will happen if no ping in mTimeoutUs since the last ping.)
    void keepAlive();

private:
    void threadLoop();
    void updateTimer_l();

    TranscodingSessionController* mOwner;
    const int64_t mTimeoutUs;
    mutable std::mutex mLock;
    std::condition_variable mCondition GUARDED_BY(mLock);
    // Whether watchdog is monitoring a session for timeout.
    bool mActive GUARDED_BY(mLock);
    // Whether watchdog is aborted and the monitoring thread should exit.
    bool mAbort GUARDED_BY(mLock);
    // When watchdog is active, the next timeout time point.
    std::chrono::steady_clock::time_point mNextTimeoutTime GUARDED_BY(mLock);
    // When watchdog is active, the session being watched.
    SessionKeyType mSessionToWatch GUARDED_BY(mLock);
    std::thread mThread;
};

TranscodingSessionController::Watchdog::Watchdog(TranscodingSessionController* owner,
                                                 int64_t timeoutUs)
      : mOwner(owner),
        mTimeoutUs(timeoutUs),
        mActive(false),
        mAbort(false),
        mThread(&Watchdog::threadLoop, this) {
    ALOGV("Watchdog CTOR: %p", this);
}

TranscodingSessionController::Watchdog::~Watchdog() {
    ALOGV("Watchdog DTOR: %p", this);

    {
        // Exit the looper thread.
        std::scoped_lock lock{mLock};

        mAbort = true;
        mCondition.notify_one();
    }

    mThread.join();
    ALOGV("Watchdog DTOR: %p, done.", this);
}

void TranscodingSessionController::Watchdog::start(const SessionKeyType& key) {
    std::scoped_lock lock{mLock};

    if (!mActive) {
        ALOGI("Watchdog start: %s", sessionToString(key).c_str());

        mActive = true;
        mSessionToWatch = key;
        updateTimer_l();
        mCondition.notify_one();
    }
}

void TranscodingSessionController::Watchdog::stop() {
    std::scoped_lock lock{mLock};

    if (mActive) {
        ALOGI("Watchdog stop: %s", sessionToString(mSessionToWatch).c_str());

        mActive = false;
        mCondition.notify_one();
    }
}

void TranscodingSessionController::Watchdog::keepAlive() {
    std::scoped_lock lock{mLock};

    if (mActive) {
        ALOGI("Watchdog keepAlive: %s", sessionToString(mSessionToWatch).c_str());

        updateTimer_l();
        mCondition.notify_one();
    }
}

// updateTimer_l() is only called with lock held.
void TranscodingSessionController::Watchdog::updateTimer_l() NO_THREAD_SAFETY_ANALYSIS {
    std::chrono::microseconds timeout(mTimeoutUs);
    mNextTimeoutTime = std::chrono::steady_clock::now() + timeout;
}

// Unfortunately std::unique_lock is incompatible with -Wthread-safety.
void TranscodingSessionController::Watchdog::threadLoop() NO_THREAD_SAFETY_ANALYSIS {
    std::unique_lock<std::mutex> lock{mLock};

    while (!mAbort) {
        if (!mActive) {
            mCondition.wait(lock);
            continue;
        }
        // Watchdog active, wait till next timeout time.
        if (mCondition.wait_until(lock, mNextTimeoutTime) == std::cv_status::timeout) {
            // If timeout happens, report timeout and deactivate watchdog.
            mActive = false;
            // Make a copy of session key, as once we unlock, it could be unprotected.
            SessionKeyType sessionKey = mSessionToWatch;

            ALOGE("Watchdog timeout: %s", sessionToString(sessionKey).c_str());

            lock.unlock();
            mOwner->onError(sessionKey.first, sessionKey.second,
                            TranscodingErrorCode::kWatchdogTimeout);
            lock.lock();
        }
    }
}
///////////////////////////////////////////////////////////////////////////////
struct TranscodingSessionController::Pacer {
    Pacer(const ControllerConfig& config)
          : mBurstThresholdMs(config.pacerBurstThresholdMs),
            mBurstCountQuota(config.pacerBurstCountQuota),
            mBurstTimeQuotaSec(config.pacerBurstTimeQuotaSeconds) {}

    ~Pacer() = default;

    bool onSessionStarted(uid_t uid);
    void onSessionCompleted(uid_t uid, std::chrono::microseconds runningTime);

private:
    // Threshold of time between finish/start below which a back-to-back start is counted.
    int32_t mBurstThresholdMs;
    // Maximum allowed back-to-back start count.
    int32_t mBurstCountQuota;
    // Maximum allowed back-to-back running time.
    int32_t mBurstTimeQuotaSec;

    struct UidHistoryEntry {
        bool sessionActive = false;
        int32_t burstCount = 0;
        std::chrono::steady_clock::duration burstDuration{0};
        std::chrono::steady_clock::time_point lastCompletedTime;
    };
    std::map<uid_t, UidHistoryEntry> mUidHistoryMap;
};

bool TranscodingSessionController::Pacer::onSessionStarted(uid_t uid) {
    // If uid doesn't exist, only insert the entry and mark session active. Skip quota checking.
    if (mUidHistoryMap.find(uid) == mUidHistoryMap.end()) {
        mUidHistoryMap.emplace(uid, UidHistoryEntry{});
        mUidHistoryMap[uid].sessionActive = true;
        ALOGV("Pacer::onSessionStarted: uid %d: new", uid);
        return true;
    }

    // TODO: if Thermal throttling or resoure lost happened to occurr between this start
    // and the previous completion, we should deduct the paused time from the elapsed time.
    // (Individual session's pause time, on the other hand, doesn't need to be deducted
    // because it doesn't affect the gap between last completion and the start.
    auto timeSinceLastComplete =
            std::chrono::steady_clock::now() - mUidHistoryMap[uid].lastCompletedTime;
    if (mUidHistoryMap[uid].burstCount >= mBurstCountQuota &&
        mUidHistoryMap[uid].burstDuration >= std::chrono::seconds(mBurstTimeQuotaSec)) {
        ALOGW("Pacer::onSessionStarted: uid %d: over quota, burst count %d, time %lldms", uid,
              mUidHistoryMap[uid].burstCount,
              (long long)mUidHistoryMap[uid].burstDuration.count() / 1000000);
        return false;
    }

    // If not over quota, allow the session, and reset as long as this is not too close
    // to previous completion.
    if (timeSinceLastComplete > std::chrono::milliseconds(mBurstThresholdMs)) {
        ALOGV("Pacer::onSessionStarted: uid %d: reset quota", uid);
        mUidHistoryMap[uid].burstCount = 0;
        mUidHistoryMap[uid].burstDuration = std::chrono::milliseconds(0);
    } else {
        ALOGV("Pacer::onSessionStarted: uid %d: burst count %d, time %lldms", uid,
              mUidHistoryMap[uid].burstCount,
              (long long)mUidHistoryMap[uid].burstDuration.count() / 1000000);
    }

    mUidHistoryMap[uid].sessionActive = true;
    return true;
}

void TranscodingSessionController::Pacer::onSessionCompleted(
        uid_t uid, std::chrono::microseconds runningTime) {
    // Skip quota update if this uid missed the start. (Could happen if the uid is added via
    // addClientUid() after the session start.)
    if (mUidHistoryMap.find(uid) == mUidHistoryMap.end() || !mUidHistoryMap[uid].sessionActive) {
        ALOGV("Pacer::onSessionCompleted: uid %d: not started", uid);
        return;
    }
    ALOGV("Pacer::onSessionCompleted: uid %d: runningTime %lld", uid, runningTime.count() / 1000);
    mUidHistoryMap[uid].sessionActive = false;
    mUidHistoryMap[uid].burstCount++;
    mUidHistoryMap[uid].burstDuration += runningTime;
    mUidHistoryMap[uid].lastCompletedTime = std::chrono::steady_clock::now();
}

///////////////////////////////////////////////////////////////////////////////

TranscodingSessionController::TranscodingSessionController(
        const TranscoderFactoryType& transcoderFactory,
        const std::shared_ptr<UidPolicyInterface>& uidPolicy,
        const std::shared_ptr<ResourcePolicyInterface>& resourcePolicy,
        const std::shared_ptr<ThermalPolicyInterface>& thermalPolicy,
        const ControllerConfig* config)
      : mTranscoderFactory(transcoderFactory),
        mUidPolicy(uidPolicy),
        mResourcePolicy(resourcePolicy),
        mThermalPolicy(thermalPolicy),
        mCurrentSession(nullptr),
        mResourceLost(false) {
    // Only push empty offline queue initially. Realtime queues are added when requests come in.
    mUidSortedList.push_back(OFFLINE_UID);
    mOfflineUidIterator = mUidSortedList.begin();
    mSessionQueues.emplace(OFFLINE_UID, SessionQueueType());
    mUidPackageNames[OFFLINE_UID] = "(offline)";
    mThermalThrottling = thermalPolicy->getThrottlingStatus();
    if (config != nullptr) {
        mConfig = *config;
    }
    mPacer.reset(new Pacer(mConfig));
    ALOGD("@@@ watchdog %lld, burst count %d, burst time %d, burst threshold %d",
          (long long)mConfig.watchdogTimeoutUs, mConfig.pacerBurstCountQuota,
          mConfig.pacerBurstTimeQuotaSeconds, mConfig.pacerBurstThresholdMs);
}

TranscodingSessionController::~TranscodingSessionController() {}

void TranscodingSessionController::dumpSession_l(const Session& session, String8& result,
                                                 bool closedSession) {
    const size_t SIZE = 256;
    char buffer[SIZE];
    const TranscodingRequestParcel& request = session.request;
    snprintf(buffer, SIZE, "      Session: %s, %s, %d%%\n", sessionToString(session.key).c_str(),
             sessionStateToString(session.getState()), session.lastProgress);
    result.append(buffer);
    snprintf(buffer, SIZE, "        pkg: %s\n", request.clientPackageName.c_str());
    result.append(buffer);
    snprintf(buffer, SIZE, "        src: %s\n", request.sourceFilePath.c_str());
    result.append(buffer);
    snprintf(buffer, SIZE, "        dst: %s\n", request.destinationFilePath.c_str());
    result.append(buffer);

    if (closedSession) {
        snprintf(buffer, SIZE,
                 "        waiting: %.1fs, running: %.1fs, paused: %.1fs, paused count: %d\n",
                 session.waitingTime.count() / 1000000.0f, session.runningTime.count() / 1000000.0f,
                 session.pausedTime.count() / 1000000.0f, session.pauseCount);
        result.append(buffer);
    }
}

void TranscodingSessionController::dumpAllSessions(int fd, const Vector<String16>& args __unused) {
    String8 result;

    const size_t SIZE = 256;
    char buffer[SIZE];
    std::scoped_lock lock{mLock};

    snprintf(buffer, SIZE, "\n========== Dumping live sessions queues =========\n");
    result.append(buffer);
    snprintf(buffer, SIZE, "  Total num of Sessions: %zu\n", mSessionMap.size());
    result.append(buffer);

    std::vector<int32_t> uids(mUidSortedList.begin(), mUidSortedList.end());

    for (int32_t i = 0; i < uids.size(); i++) {
        const uid_t uid = uids[i];

        if (mSessionQueues[uid].empty()) {
            continue;
        }
        snprintf(buffer, SIZE, "    uid: %d, pkg: %s\n", uid,
                 mUidPackageNames.count(uid) > 0 ? mUidPackageNames[uid].c_str() : "(unknown)");
        result.append(buffer);
        snprintf(buffer, SIZE, "      Num of sessions: %zu\n", mSessionQueues[uid].size());
        result.append(buffer);
        for (auto& sessionKey : mSessionQueues[uid]) {
            auto sessionIt = mSessionMap.find(sessionKey);
            if (sessionIt == mSessionMap.end()) {
                snprintf(buffer, SIZE, "Failed to look up Session %s  \n",
                         sessionToString(sessionKey).c_str());
                result.append(buffer);
                continue;
            }
            dumpSession_l(sessionIt->second, result);
        }
    }

    snprintf(buffer, SIZE, "\n========== Dumping past sessions =========\n");
    result.append(buffer);
    for (auto& session : mSessionHistory) {
        dumpSession_l(session, result, true /*closedSession*/);
    }

    write(fd, result.string(), result.size());
}

/*
 * Returns nullptr if there is no session, or we're paused globally (due to resource lost,
 * thermal throttling, etc.). Otherwise, return the session that should be run next.
 */
TranscodingSessionController::Session* TranscodingSessionController::getTopSession_l() {
    if (mSessionMap.empty()) {
        return nullptr;
    }

    // Return nullptr if we're paused globally due to resource lost or thermal throttling.
    if (((mResourcePolicy != nullptr && mResourceLost) ||
         (mThermalPolicy != nullptr && mThermalThrottling))) {
        return nullptr;
    }

    uid_t topUid = *mUidSortedList.begin();
    // If the current session is running, and it's in the topUid's queue, let it continue
    // to run even if it's not the earliest in that uid's queue.
    // For example, uid(B) is added to a session while it's pending in uid(A)'s queue, then
    // B is brought to front which caused the session to run, then user switches back to A.
    if (mCurrentSession != nullptr && mCurrentSession->getState() == Session::RUNNING &&
        mCurrentSession->allClientUids.count(topUid) > 0) {
        return mCurrentSession;
    }
    SessionKeyType topSessionKey = *mSessionQueues[topUid].begin();
    return &mSessionMap[topSessionKey];
}

void TranscodingSessionController::setSessionState_l(Session* session, Session::State state) {
    bool wasRunning = (session->getState() == Session::RUNNING);
    session->setState(state);
    bool isRunning = (session->getState() == Session::RUNNING);

    if (wasRunning == isRunning) {
        return;
    }

    // Currently we only have 1 running session, and we always put the previous
    // session in non-running state before we run the new session, so it's okay
    // to start/stop the watchdog here. If this assumption changes, we need to
    // track the number of running sessions and start/stop watchdog based on that.
    if (isRunning) {
        mWatchdog->start(session->key);
    } else {
        mWatchdog->stop();
    }
}

void TranscodingSessionController::Session::setState(Session::State newState) {
    if (state == newState) {
        return;
    }
    auto nowTime = std::chrono::steady_clock::now();
    if (state != INVALID) {
        std::chrono::microseconds elapsedTime =
                std::chrono::duration_cast<std::chrono::microseconds>(nowTime - stateEnterTime);
        switch (state) {
        case PAUSED:
            pausedTime = pausedTime + elapsedTime;
            break;
        case RUNNING:
            runningTime = runningTime + elapsedTime;
            break;
        case NOT_STARTED:
            waitingTime = waitingTime + elapsedTime;
            break;
        default:
            break;
        }
    }
    if (newState == PAUSED) {
        pauseCount++;
    }
    stateEnterTime = nowTime;
    state = newState;
}

void TranscodingSessionController::updateCurrentSession_l() {
    Session* curSession = mCurrentSession;
    Session* topSession = nullptr;

    // Delayed init of transcoder and watchdog.
    if (mTranscoder == nullptr) {
        mTranscoder = mTranscoderFactory(shared_from_this());
        mWatchdog = std::make_shared<Watchdog>(this, mConfig.watchdogTimeoutUs);
    }

    // If we found a different top session, or the top session's running state is not
    // correct. Take some actions to ensure it's correct.
    while ((topSession = getTopSession_l()) != curSession ||
           (topSession != nullptr && !topSession->isRunning())) {
        ALOGV("updateCurrentSession_l: topSession is %s, curSession is %s",
              topSession == nullptr ? "null" : sessionToString(topSession->key).c_str(),
              curSession == nullptr ? "null" : sessionToString(curSession->key).c_str());

        // If current session is running, pause it first. Note this is needed for either
        // cases: 1) Top session is changing to another session, or 2) Top session is
        // changing to null (which means we should be globally paused).
        if (curSession != nullptr && curSession->getState() == Session::RUNNING) {
            mTranscoder->pause(curSession->key.first, curSession->key.second);
            setSessionState_l(curSession, Session::PAUSED);
        }

        if (topSession == nullptr) {
            // Nothing more to run (either no session or globally paused).
            break;
        }

        // Otherwise, ensure topSession is running.
        if (topSession->getState() == Session::NOT_STARTED) {
            // Check if at least one client has quota to start the session.
            bool keepForClient = false;
            for (uid_t uid : topSession->allClientUids) {
                if (mPacer->onSessionStarted(uid)) {
                    keepForClient = true;
                    // DO NOT break here, because book-keeping still needs to happen
                    // for the other uids.
                }
            }
            if (!keepForClient) {
                // Unfortunately all uids requesting this session are out of quota.
                // Drop this session and try the next one.
                {
                    auto clientCallback = mSessionMap[topSession->key].callback.lock();
                    if (clientCallback != nullptr) {
                        clientCallback->onTranscodingFailed(
                                topSession->key.second, TranscodingErrorCode::kDroppedByService);
                    }
                }
                removeSession_l(topSession->key, Session::DROPPED_BY_PACER);
                continue;
            }
            mTranscoder->start(topSession->key.first, topSession->key.second, topSession->request,
                               topSession->callingUid, topSession->callback.lock());
            setSessionState_l(topSession, Session::RUNNING);
        } else if (topSession->getState() == Session::PAUSED) {
            mTranscoder->resume(topSession->key.first, topSession->key.second, topSession->request,
                                topSession->callingUid, topSession->callback.lock());
            setSessionState_l(topSession, Session::RUNNING);
        }
        break;
    }
    mCurrentSession = topSession;
}

void TranscodingSessionController::addUidToSession_l(uid_t clientUid,
                                                     const SessionKeyType& sessionKey) {
    // If it's an offline session, the queue was already added in constructor.
    // If it's a real-time sessions, check if a queue is already present for the uid,
    // and add a new queue if needed.
    if (clientUid != OFFLINE_UID) {
        if (mSessionQueues.count(clientUid) == 0) {
            mUidPolicy->registerMonitorUid(clientUid);
            if (mUidPolicy->isUidOnTop(clientUid)) {
                mUidSortedList.push_front(clientUid);
            } else {
                // Shouldn't be submitting real-time requests from non-top app,
                // put it in front of the offline queue.
                mUidSortedList.insert(mOfflineUidIterator, clientUid);
            }
        } else if (clientUid != *mUidSortedList.begin()) {
            if (mUidPolicy->isUidOnTop(clientUid)) {
                mUidSortedList.remove(clientUid);
                mUidSortedList.push_front(clientUid);
            }
        }
    }
    // Append this session to the uid's queue.
    mSessionQueues[clientUid].push_back(sessionKey);
}

void TranscodingSessionController::removeSession_l(const SessionKeyType& sessionKey,
                                                   Session::State finalState, bool keepForOffline) {
    ALOGV("%s: session %s", __FUNCTION__, sessionToString(sessionKey).c_str());

    if (mSessionMap.count(sessionKey) == 0) {
        ALOGE("session %s doesn't exist", sessionToString(sessionKey).c_str());
        return;
    }

    // Remove session from uid's queue.
    bool uidQueueRemoved = false;
    for (uid_t uid : mSessionMap[sessionKey].allClientUids) {
        if (keepForOffline && uid == OFFLINE_UID) {
            continue;
        }
        SessionQueueType& sessionQueue = mSessionQueues[uid];
        auto it = std::find(sessionQueue.begin(), sessionQueue.end(), sessionKey);
        if (it == sessionQueue.end()) {
            ALOGW("couldn't find session %s in queue for uid %d",
                  sessionToString(sessionKey).c_str(), uid);
            continue;
        }
        sessionQueue.erase(it);

        // If this is the last session in a real-time queue, remove this uid's queue.
        if (uid != OFFLINE_UID && sessionQueue.empty()) {
            mUidSortedList.remove(uid);
            mSessionQueues.erase(uid);
            mUidPolicy->unregisterMonitorUid(uid);

            uidQueueRemoved = true;
        }
    }

    if (uidQueueRemoved) {
        std::unordered_set<uid_t> topUids = mUidPolicy->getTopUids();
        moveUidsToTop_l(topUids, false /*preserveTopUid*/);
    }

    if (keepForOffline) {
        mSessionMap[sessionKey].allClientUids = {OFFLINE_UID};
        return;
    }

    // Clear current session.
    if (mCurrentSession == &mSessionMap[sessionKey]) {
        mCurrentSession = nullptr;
    }

    setSessionState_l(&mSessionMap[sessionKey], finalState);

    if (finalState == Session::FINISHED || finalState == Session::ERROR) {
        for (uid_t uid : mSessionMap[sessionKey].allClientUids) {
            mPacer->onSessionCompleted(uid, mSessionMap[sessionKey].runningTime);
        }
    }

    mSessionHistory.push_back(mSessionMap[sessionKey]);
    if (mSessionHistory.size() > kSessionHistoryMax) {
        mSessionHistory.erase(mSessionHistory.begin());
    }

    // Remove session from session map.
    mSessionMap.erase(sessionKey);
}

/**
 * Moves the set of uids to the front of mUidSortedList (which is used to pick
 * the next session to run).
 *
 * This is called when 1) we received a onTopUidsChanged() callback from UidPolicy,
 * or 2) we removed the session queue for a uid because it becomes empty.
 *
 * In case of 1), if there are multiple uids in the set, and the current front
 * uid in mUidSortedList is still in the set, we try to keep that uid at front
 * so that current session run is not interrupted. (This is not a concern for case 2)
 * because the queue for a uid was just removed entirely.)
 */
void TranscodingSessionController::moveUidsToTop_l(const std::unordered_set<uid_t>& uids,
                                                   bool preserveTopUid) {
    // If uid set is empty, nothing to do. Do not change the queue status.
    if (uids.empty()) {
        return;
    }

    // Save the current top uid.
    uid_t curTopUid = *mUidSortedList.begin();
    bool pushCurTopToFront = false;
    int32_t numUidsMoved = 0;

    // Go through the sorted uid list once, and move the ones in top set to front.
    for (auto it = mUidSortedList.begin(); it != mUidSortedList.end();) {
        uid_t uid = *it;

        if (uid != OFFLINE_UID && uids.count(uid) > 0) {
            it = mUidSortedList.erase(it);

            // If this is the top we're preserving, don't push it here, push
            // it after the for-loop.
            if (uid == curTopUid && preserveTopUid) {
                pushCurTopToFront = true;
            } else {
                mUidSortedList.push_front(uid);
            }

            // If we found all uids in the set, break out.
            if (++numUidsMoved == uids.size()) {
                break;
            }
        } else {
            ++it;
        }
    }

    if (pushCurTopToFront) {
        mUidSortedList.push_front(curTopUid);
    }
}

bool TranscodingSessionController::submit(
        ClientIdType clientId, SessionIdType sessionId, uid_t callingUid, uid_t clientUid,
        const TranscodingRequestParcel& request,
        const std::weak_ptr<ITranscodingClientCallback>& callback) {
    SessionKeyType sessionKey = std::make_pair(clientId, sessionId);

    ALOGV("%s: session %s, uid %d, prioirty %d", __FUNCTION__, sessionToString(sessionKey).c_str(),
          clientUid, (int32_t)request.priority);

    std::scoped_lock lock{mLock};

    if (mSessionMap.count(sessionKey) > 0) {
        ALOGE("session %s already exists", sessionToString(sessionKey).c_str());
        return false;
    }

    // Add the uid package name to the store of package names we already know.
    if (mUidPackageNames.count(clientUid) == 0) {
        mUidPackageNames.emplace(clientUid, request.clientPackageName);
    }

    // TODO(chz): only support offline vs real-time for now. All kUnspecified sessions
    // go to offline queue.
    if (request.priority == TranscodingSessionPriority::kUnspecified) {
        clientUid = OFFLINE_UID;
    }

    // Add session to session map.
    mSessionMap[sessionKey].key = sessionKey;
    mSessionMap[sessionKey].callingUid = callingUid;
    mSessionMap[sessionKey].allClientUids.insert(clientUid);
    mSessionMap[sessionKey].request = request;
    mSessionMap[sessionKey].callback = callback;
    setSessionState_l(&mSessionMap[sessionKey], Session::NOT_STARTED);

    addUidToSession_l(clientUid, sessionKey);

    updateCurrentSession_l();

    validateState_l();
    return true;
}

bool TranscodingSessionController::cancel(ClientIdType clientId, SessionIdType sessionId) {
    SessionKeyType sessionKey = std::make_pair(clientId, sessionId);

    ALOGV("%s: session %s", __FUNCTION__, sessionToString(sessionKey).c_str());

    std::list<SessionKeyType> sessionsToRemove, sessionsForOffline;

    std::scoped_lock lock{mLock};

    if (sessionId < 0) {
        for (auto it = mSessionMap.begin(); it != mSessionMap.end(); ++it) {
            if (it->first.first == clientId) {
                // If there is offline request, only keep the offline client;
                // otherwise remove the session.
                if (it->second.allClientUids.count(OFFLINE_UID) > 0) {
                    sessionsForOffline.push_back(it->first);
                } else {
                    sessionsToRemove.push_back(it->first);
                }
            }
        }
    } else {
        if (mSessionMap.count(sessionKey) == 0) {
            ALOGE("session %s doesn't exist", sessionToString(sessionKey).c_str());
            return false;
        }
        sessionsToRemove.push_back(sessionKey);
    }

    for (auto it = sessionsToRemove.begin(); it != sessionsToRemove.end(); ++it) {
        // If the session has ever been started, stop it now.
        // Note that stop() is needed even if the session is currently paused. This instructs
        // the transcoder to discard any states for the session, otherwise the states may
        // never be discarded.
        if (mSessionMap[*it].getState() != Session::NOT_STARTED) {
            mTranscoder->stop(it->first, it->second);
        }

        // Remove the session.
        removeSession_l(*it, Session::CANCELED);
    }

    for (auto it = sessionsForOffline.begin(); it != sessionsForOffline.end(); ++it) {
        removeSession_l(*it, Session::CANCELED, true /*keepForOffline*/);
    }

    // Start next session.
    updateCurrentSession_l();

    validateState_l();
    return true;
}

bool TranscodingSessionController::addClientUid(ClientIdType clientId, SessionIdType sessionId,
                                                uid_t clientUid) {
    SessionKeyType sessionKey = std::make_pair(clientId, sessionId);

    std::scoped_lock lock{mLock};

    if (mSessionMap.count(sessionKey) == 0) {
        ALOGE("session %s doesn't exist", sessionToString(sessionKey).c_str());
        return false;
    }

    if (mSessionMap[sessionKey].allClientUids.count(clientUid) > 0) {
        ALOGE("session %s already has uid %d", sessionToString(sessionKey).c_str(), clientUid);
        return false;
    }

    mSessionMap[sessionKey].allClientUids.insert(clientUid);
    addUidToSession_l(clientUid, sessionKey);

    updateCurrentSession_l();

    validateState_l();
    return true;
}

bool TranscodingSessionController::getClientUids(ClientIdType clientId, SessionIdType sessionId,
                                                 std::vector<int32_t>* out_clientUids) {
    SessionKeyType sessionKey = std::make_pair(clientId, sessionId);

    std::scoped_lock lock{mLock};

    if (mSessionMap.count(sessionKey) == 0) {
        ALOGE("session %s doesn't exist", sessionToString(sessionKey).c_str());
        return false;
    }

    out_clientUids->clear();
    for (uid_t uid : mSessionMap[sessionKey].allClientUids) {
        if (uid != OFFLINE_UID) {
            out_clientUids->push_back(uid);
        }
    }
    return true;
}

bool TranscodingSessionController::getSession(ClientIdType clientId, SessionIdType sessionId,
                                              TranscodingRequestParcel* request) {
    SessionKeyType sessionKey = std::make_pair(clientId, sessionId);

    std::scoped_lock lock{mLock};

    if (mSessionMap.count(sessionKey) == 0) {
        ALOGE("session %s doesn't exist", sessionToString(sessionKey).c_str());
        return false;
    }

    *(TranscodingRequest*)request = mSessionMap[sessionKey].request;
    return true;
}

void TranscodingSessionController::notifyClient(ClientIdType clientId, SessionIdType sessionId,
                                                const char* reason,
                                                std::function<void(const SessionKeyType&)> func) {
    SessionKeyType sessionKey = std::make_pair(clientId, sessionId);

    std::scoped_lock lock{mLock};

    if (mSessionMap.count(sessionKey) == 0) {
        ALOGW("%s: ignoring %s for session %s that doesn't exist", __FUNCTION__, reason,
              sessionToString(sessionKey).c_str());
        return;
    }

    // Only ignore if session was never started. In particular, propagate the status
    // to client if the session is paused. Transcoder could have posted finish when
    // we're pausing it, and the finish arrived after we changed current session.
    if (mSessionMap[sessionKey].getState() == Session::NOT_STARTED) {
        ALOGW("%s: ignoring %s for session %s that was never started", __FUNCTION__, reason,
              sessionToString(sessionKey).c_str());
        return;
    }

    ALOGV("%s: session %s %s", __FUNCTION__, sessionToString(sessionKey).c_str(), reason);
    func(sessionKey);
}

void TranscodingSessionController::onStarted(ClientIdType clientId, SessionIdType sessionId) {
    notifyClient(clientId, sessionId, "started", [=](const SessionKeyType& sessionKey) {
        auto callback = mSessionMap[sessionKey].callback.lock();
        if (callback != nullptr) {
            callback->onTranscodingStarted(sessionId);
        }
    });
}

void TranscodingSessionController::onPaused(ClientIdType clientId, SessionIdType sessionId) {
    notifyClient(clientId, sessionId, "paused", [=](const SessionKeyType& sessionKey) {
        auto callback = mSessionMap[sessionKey].callback.lock();
        if (callback != nullptr) {
            callback->onTranscodingPaused(sessionId);
        }
    });
}

void TranscodingSessionController::onResumed(ClientIdType clientId, SessionIdType sessionId) {
    notifyClient(clientId, sessionId, "resumed", [=](const SessionKeyType& sessionKey) {
        auto callback = mSessionMap[sessionKey].callback.lock();
        if (callback != nullptr) {
            callback->onTranscodingResumed(sessionId);
        }
    });
}

void TranscodingSessionController::onFinish(ClientIdType clientId, SessionIdType sessionId) {
    notifyClient(clientId, sessionId, "finish", [=](const SessionKeyType& sessionKey) {
        {
            auto clientCallback = mSessionMap[sessionKey].callback.lock();
            if (clientCallback != nullptr) {
                clientCallback->onTranscodingFinished(
                        sessionId, TranscodingResultParcel({sessionId, -1 /*actualBitrateBps*/,
                                                            std::nullopt /*sessionStats*/}));
            }
        }

        // Remove the session.
        removeSession_l(sessionKey, Session::FINISHED);

        // Start next session.
        updateCurrentSession_l();

        validateState_l();
    });
}

void TranscodingSessionController::onError(ClientIdType clientId, SessionIdType sessionId,
                                           TranscodingErrorCode err) {
    notifyClient(clientId, sessionId, "error", [=](const SessionKeyType& sessionKey) {
        if (err == TranscodingErrorCode::kWatchdogTimeout) {
            // Abandon the transcoder, as its handler thread might be stuck in some call to
            // MediaTranscoder altogether, and may not be able to handle any new tasks.
            mTranscoder->stop(clientId, sessionId, true /*abandon*/);
            // Clear the last ref count before we create new transcoder.
            mTranscoder = nullptr;
            mTranscoder = mTranscoderFactory(shared_from_this());
        }

        {
            auto clientCallback = mSessionMap[sessionKey].callback.lock();
            if (clientCallback != nullptr) {
                clientCallback->onTranscodingFailed(sessionId, err);
            }
        }

        // Remove the session.
        removeSession_l(sessionKey, Session::ERROR);

        // Start next session.
        updateCurrentSession_l();

        validateState_l();
    });
}

void TranscodingSessionController::onProgressUpdate(ClientIdType clientId, SessionIdType sessionId,
                                                    int32_t progress) {
    notifyClient(clientId, sessionId, "progress", [=](const SessionKeyType& sessionKey) {
        auto callback = mSessionMap[sessionKey].callback.lock();
        if (callback != nullptr) {
            callback->onProgressUpdate(sessionId, progress);
        }
        mSessionMap[sessionKey].lastProgress = progress;
    });
}

void TranscodingSessionController::onHeartBeat(ClientIdType clientId, SessionIdType sessionId) {
    notifyClient(clientId, sessionId, "heart-beat",
                 [=](const SessionKeyType& /*sessionKey*/) { mWatchdog->keepAlive(); });
}

void TranscodingSessionController::onResourceLost(ClientIdType clientId, SessionIdType sessionId) {
    ALOGI("%s", __FUNCTION__);

    notifyClient(clientId, sessionId, "resource_lost", [=](const SessionKeyType& sessionKey) {
        if (mResourceLost) {
            return;
        }

        Session* resourceLostSession = &mSessionMap[sessionKey];
        if (resourceLostSession->getState() != Session::RUNNING) {
            ALOGW("session %s lost resource but is no longer running",
                  sessionToString(sessionKey).c_str());
            return;
        }
        // If we receive a resource loss event, the transcoder already paused the transcoding,
        // so we don't need to call onPaused() to pause it. However, we still need to notify
        // the client and update the session state here.
        setSessionState_l(resourceLostSession, Session::PAUSED);
        // Notify the client as a paused event.
        auto clientCallback = resourceLostSession->callback.lock();
        if (clientCallback != nullptr) {
            clientCallback->onTranscodingPaused(sessionKey.second);
        }
        if (mResourcePolicy != nullptr) {
            mResourcePolicy->setPidResourceLost(resourceLostSession->request.clientPid);
        }
        mResourceLost = true;

        validateState_l();
    });
}

void TranscodingSessionController::onTopUidsChanged(const std::unordered_set<uid_t>& uids) {
    if (uids.empty()) {
        ALOGW("%s: ignoring empty uids", __FUNCTION__);
        return;
    }

    std::string uidStr;
    for (auto it = uids.begin(); it != uids.end(); it++) {
        if (!uidStr.empty()) {
            uidStr += ", ";
        }
        uidStr += std::to_string(*it);
    }

    ALOGD("%s: topUids: size %zu, uids: %s", __FUNCTION__, uids.size(), uidStr.c_str());

    std::scoped_lock lock{mLock};

    moveUidsToTop_l(uids, true /*preserveTopUid*/);

    updateCurrentSession_l();

    validateState_l();
}

void TranscodingSessionController::onResourceAvailable() {
    std::scoped_lock lock{mLock};

    if (!mResourceLost) {
        return;
    }

    ALOGI("%s", __FUNCTION__);

    mResourceLost = false;
    updateCurrentSession_l();

    validateState_l();
}

void TranscodingSessionController::onThrottlingStarted() {
    std::scoped_lock lock{mLock};

    if (mThermalThrottling) {
        return;
    }

    ALOGI("%s", __FUNCTION__);

    mThermalThrottling = true;
    updateCurrentSession_l();

    validateState_l();
}

void TranscodingSessionController::onThrottlingStopped() {
    std::scoped_lock lock{mLock};

    if (!mThermalThrottling) {
        return;
    }

    ALOGI("%s", __FUNCTION__);

    mThermalThrottling = false;
    updateCurrentSession_l();

    validateState_l();
}

void TranscodingSessionController::validateState_l() {
#ifdef VALIDATE_STATE
    LOG_ALWAYS_FATAL_IF(mSessionQueues.count(OFFLINE_UID) != 1,
                        "mSessionQueues offline queue number is not 1");
    LOG_ALWAYS_FATAL_IF(*mOfflineUidIterator != OFFLINE_UID,
                        "mOfflineUidIterator not pointing to offline uid");
    LOG_ALWAYS_FATAL_IF(mUidSortedList.size() != mSessionQueues.size(),
                        "mUidSortedList and mSessionQueues size mismatch, %zu vs %zu",
                        mUidSortedList.size(), mSessionQueues.size());

    int32_t totalSessions = 0;
    for (auto uid : mUidSortedList) {
        LOG_ALWAYS_FATAL_IF(mSessionQueues.count(uid) != 1,
                            "mSessionQueues count for uid %d is not 1", uid);
        for (auto& sessionKey : mSessionQueues[uid]) {
            LOG_ALWAYS_FATAL_IF(mSessionMap.count(sessionKey) != 1,
                                "mSessions count for session %s is not 1",
                                sessionToString(sessionKey).c_str());
        }

        totalSessions += mSessionQueues[uid].size();
    }
    int32_t totalSessionsAlternative = 0;
    for (auto const& s : mSessionMap) {
        totalSessionsAlternative += s.second.allClientUids.size();
    }
    LOG_ALWAYS_FATAL_IF(totalSessions != totalSessionsAlternative,
                        "session count (including dup) from mSessionQueues doesn't match that from "
                        "mSessionMap, %d vs %d",
                        totalSessions, totalSessionsAlternative);
#endif  // VALIDATE_STATE
}

}  // namespace android