Donate to e Foundation | Murena handsets with /e/OS | Own a part of Murena! Learn more

Commit 87d199cf authored by Chong Zhang's avatar Chong Zhang
Browse files

transcoding: drop transcoding request for bad actors 

Drop requests for a uid if it submits more than X back-to-back requests
that lasted longer than Y seconds.

bug: 177631807
test: new and existing unit tests; CTS MediaTranscodeManagerTest;
Manual testing Yelp app (with shortened thresholds)

Change-Id: I88287be62190ee5bc815dedb2a62c54ed3c41824
parent b2d4b7b0

media/libmediatranscoding/TranscodingSessionController.cpp

+145 −42
Original line number Diff line number Diff line
@@ -87,15 +87,12 @@ private: 
    // 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::system_clock::time_point mNextTimeoutTime GUARDED_BY(mLock);

    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;

};



static constexpr int64_t kWatchdogTimeoutUs = 3000000LL;

static constexpr int64_t kTranscoderHeartBeatIntervalUs = 1000000LL;



TranscodingSessionController::Watchdog::Watchdog(TranscodingSessionController* owner,

                                                 int64_t timeoutUs)

      : mOwner(owner),
@@ -159,7 +156,7 @@ void TranscodingSessionController::Watchdog::keepAlive() { 
// 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::system_clock::now() + timeout;

    mNextTimeoutTime = std::chrono::steady_clock::now() + timeout;

}



// Unfortunately std::unique_lock is incompatible with -Wthread-safety.
@@ -187,13 +184,85 @@ void TranscodingSessionController::Watchdog::threadLoop() NO_THREAD_SAFETY_ANALY 
        }

    }

}

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

struct TranscodingSessionController::Pacer {

    Pacer(const ControllerConfig& config)

          : mBurstThresholdMs(config.pacerBurstThresholdMs),

            mBurstCountQuota(config.pacerBurstCountQuota),

            mBurstTimeQuotaSec(config.pacerBurstTimeQuotaSeconds) {}



    ~Pacer() = default;



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

    bool onSessionStarted(uid_t uid);



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 {

        std::chrono::steady_clock::time_point lastCompletedTime;

        int32_t burstCount = 0;

        std::chrono::steady_clock::duration burstDuration{0};

    };

    std::map<uid_t, UidHistoryEntry> mUidHistoryMap;

};



void TranscodingSessionController::Pacer::onSessionCompleted(

        uid_t uid, std::chrono::microseconds runningTime) {

    if (mUidHistoryMap.find(uid) == mUidHistoryMap.end()) {

        mUidHistoryMap.emplace(uid, UidHistoryEntry{});

    }

    mUidHistoryMap[uid].lastCompletedTime = std::chrono::steady_clock::now();

    mUidHistoryMap[uid].burstCount++;

    mUidHistoryMap[uid].burstDuration += runningTime;

}



bool TranscodingSessionController::Pacer::onSessionStarted(uid_t uid) {

    // If uid doesn't exist, this uid has no completed sessions. Skip.

    if (mUidHistoryMap.find(uid) == mUidHistoryMap.end()) {

        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: uid %d: over quota, burst count %d, time %lldms", uid,

              mUidHistoryMap[uid].burstCount, (long long)mUidHistoryMap[uid].burstDuration.count());

        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: uid %d: reset quota", uid);

        mUidHistoryMap[uid].burstCount = 0;

        mUidHistoryMap[uid].burstDuration = std::chrono::milliseconds(0);

    } else {

        ALOGV("Pacer: uid %d: burst count %d, time %lldms", uid, mUidHistoryMap[uid].burstCount,

              (long long)mUidHistoryMap[uid].burstDuration.count());

    }



    return true;

}



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



TranscodingSessionController::TranscodingSessionController(

        const TranscoderFactoryType& transcoderFactory,

        const std::shared_ptr<UidPolicyInterface>& uidPolicy,

        const std::shared_ptr<ResourcePolicyInterface>& resourcePolicy,

        const std::shared_ptr<ThermalPolicyInterface>& thermalPolicy)

        const std::shared_ptr<ThermalPolicyInterface>& thermalPolicy,

        const ControllerConfig* config)

      : mTranscoderFactory(transcoderFactory),

        mUidPolicy(uidPolicy),

        mResourcePolicy(resourcePolicy),
@@ -206,6 +275,13 @@ TranscodingSessionController::TranscodingSessionController( 
    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() {}
@@ -280,10 +356,21 @@ void TranscodingSessionController::dumpAllSessions(int fd, const Vector<String16 
    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();

    SessionKeyType topSessionKey = *mSessionQueues[topUid].begin();

    return &mSessionMap[topSessionKey];
@@ -313,9 +400,10 @@ void TranscodingSessionController::Session::setState(Session::State newState) { 
    if (state == newState) {

        return;

    }

    auto nowTime = std::chrono::system_clock::now();

    auto nowTime = std::chrono::steady_clock::now();

    if (state != INVALID) {

        std::chrono::microseconds elapsedTime = (nowTime - stateEnterTime);

        std::chrono::microseconds elapsedTime =

                std::chrono::duration_cast<std::chrono::microseconds>(nowTime - stateEnterTime);

        switch (state) {

        case PAUSED:

            pausedTime = pausedTime + elapsedTime;
@@ -338,49 +426,60 @@ void TranscodingSessionController::Session::setState(Session::State newState) { 
}



void TranscodingSessionController::updateCurrentSession_l() {

    Session* topSession = getTopSession_l();

    Session* curSession = mCurrentSession;

    ALOGV("updateCurrentSession: topSession is %s, curSession is %s",

          topSession == nullptr ? "null" : sessionToString(topSession->key).c_str(),

          curSession == nullptr ? "null" : sessionToString(curSession->key).c_str());



    if (topSession == nullptr) {

        mCurrentSession = nullptr;

        return;

    }

    Session* topSession = getTopSession_l();



    bool shouldBeRunning = !((mResourcePolicy != nullptr && mResourceLost) ||

                             (mThermalPolicy != nullptr && mThermalThrottling));

    // If we found a topSession that should be run, and it's not already running,

    // take some actions to ensure it's running.

    if (topSession != curSession ||

        (shouldBeRunning ^ (topSession->getState() == Session::RUNNING))) {

    // Delayed init of transcoder and watchdog.

    if (mTranscoder == nullptr) {

            mTranscoder = mTranscoderFactory(shared_from_this(), kTranscoderHeartBeatIntervalUs);

            mWatchdog = std::make_shared<Watchdog>(this, kWatchdogTimeoutUs);

        mTranscoder = mTranscoderFactory(shared_from_this());

        mWatchdog = std::make_shared<Watchdog>(this, mConfig.watchdogTimeoutUs);

    }



        // If current session is running, pause it first. Note this is true for either

        // cases: 1) If top session is changing, or 2) if top session is not changing but

        // the topSession's state is changing.

    // 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 we are not experiencing resource loss nor thermal throttling, we can start

        // or resume the topSession now.

        if (shouldBeRunning) {



        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) {

                mTranscoder->start(topSession->key.first, topSession->key.second,

                                   topSession->request, topSession->callingUid,

                                   topSession->callback.lock());

            } else if (topSession->getState() == Session::PAUSED) {

                mTranscoder->resume(topSession->key.first, topSession->key.second,

                                    topSession->request, topSession->callingUid,

                                    topSession->callback.lock());

            if (!mPacer->onSessionStarted(topSession->clientUid)) {

                // Unfortunately this uid is out of quota for new sessions.

                // Drop this sesion and try another 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;

}
@@ -421,6 +520,12 @@ void TranscodingSessionController::removeSession_l(const SessionKeyType& session 
    }



    setSessionState_l(&mSessionMap[sessionKey], finalState);



    if (finalState == Session::FINISHED || finalState == Session::ERROR) {

        mPacer->onSessionCompleted(mSessionMap[sessionKey].clientUid,

                                   mSessionMap[sessionKey].runningTime);

    }



    mSessionHistory.push_back(mSessionMap[sessionKey]);

    if (mSessionHistory.size() > kSessionHistoryMax) {

        mSessionHistory.erase(mSessionHistory.begin());
@@ -514,8 +619,6 @@ bool TranscodingSessionController::submit( 
    mSessionMap[sessionKey].key = sessionKey;

    mSessionMap[sessionKey].clientUid = clientUid;

    mSessionMap[sessionKey].callingUid = callingUid;

    mSessionMap[sessionKey].lastProgress = 0;

    mSessionMap[sessionKey].pauseCount = 0;

    mSessionMap[sessionKey].request = request;

    mSessionMap[sessionKey].callback = callback;

    setSessionState_l(&mSessionMap[sessionKey], Session::NOT_STARTED);
@@ -690,7 +793,7 @@ void TranscodingSessionController::onError(ClientIdType clientId, SessionIdType 
            mTranscoder->stop(clientId, sessionId, true /*abandon*/);

            // Clear the last ref count before we create new transcoder.

            mTranscoder = nullptr;

            mTranscoder = mTranscoderFactory(shared_from_this(), kTranscoderHeartBeatIntervalUs);

            mTranscoder = mTranscoderFactory(shared_from_this());

        }



        {

media/libmediatranscoding/aidl/android/media/TranscodingErrorCode.aidl

+14 −8
Original line number Diff line number Diff line
@@ -23,13 +23,19 @@ package android.media; 
 */

@Backing(type = "int")

enum TranscodingErrorCode {

    // Errors exposed to client side.

    kNoError = 0,

    kUnknown = 1,

    kMalformed = 2,

    kUnsupported = 3,

    kInvalidParameter = 4,

    kInvalidOperation = 5,

    kErrorIO = 6,

    kInsufficientResources = 7,

    kWatchdogTimeout = 8,

    kDroppedByService = 1,

    kServiceUnavailable = 2,



    // Other private errors.

    kPrivateErrorFirst     = 1000,

    kUnknown               = kPrivateErrorFirst + 0,

    kMalformed             = kPrivateErrorFirst + 1,

    kUnsupported           = kPrivateErrorFirst + 2,

    kInvalidParameter      = kPrivateErrorFirst + 3,

    kInvalidOperation      = kPrivateErrorFirst + 4,

    kErrorIO               = kPrivateErrorFirst + 5,

    kInsufficientResources = kPrivateErrorFirst + 6,

    kWatchdogTimeout       = kPrivateErrorFirst + 7,

}
 
 No newline at end of file

media/libmediatranscoding/include/media/TranscodingSessionController.h

+26 −8
Original line number Diff line number Diff line
@@ -93,7 +93,18 @@ private: 
    using SessionKeyType = std::pair<ClientIdType, SessionIdType>;

    using SessionQueueType = std::list<SessionKeyType>;

    using TranscoderFactoryType = std::function<std::shared_ptr<TranscoderInterface>(

            const std::shared_ptr<TranscoderCallbackInterface>&, int64_t)>;

            const std::shared_ptr<TranscoderCallbackInterface>&)>;



    struct ControllerConfig {

        // Watchdog timeout.

        int64_t watchdogTimeoutUs = 3000000LL;

        // Threshold of time between finish/start below which a back-to-back start is counted.

        int32_t pacerBurstThresholdMs = 1000;

        // Maximum allowed back-to-back start count.

        int32_t pacerBurstCountQuota = 10;

        // Maximum allowed back-to-back running time.

        int32_t pacerBurstTimeQuotaSeconds = 180;  // 3-min

    };



    struct Session {

        enum State {
@@ -106,16 +117,17 @@ private: 
            FINISHED,

            CANCELED,

            ERROR,

            DROPPED_BY_PACER,

        };

        SessionKeyType key;

        uid_t clientUid;

        uid_t callingUid;

        int32_t lastProgress;

        int32_t pauseCount;

        std::chrono::time_point<std::chrono::system_clock> stateEnterTime;

        std::chrono::microseconds waitingTime;

        std::chrono::microseconds runningTime;

        std::chrono::microseconds pausedTime;

        int32_t lastProgress = 0;

        int32_t pauseCount = 0;

        std::chrono::time_point<std::chrono::steady_clock> stateEnterTime;

        std::chrono::microseconds waitingTime{0};

        std::chrono::microseconds runningTime{0};

        std::chrono::microseconds pausedTime{0};



        TranscodingRequest request;

        std::weak_ptr<ITranscodingClientCallback> callback;
@@ -123,12 +135,16 @@ private: 
        // Must use setState to change state.

        void setState(Session::State state);

        State getState() const { return state; }

        bool isRunning() { return state == RUNNING; }



    private:

        State state = INVALID;

    };



    struct Watchdog;

    struct Pacer;



    ControllerConfig mConfig;



    // TODO(chz): call transcoder without global lock.

    // Use mLock for all entrypoints for now.
@@ -156,12 +172,14 @@ private: 
    bool mThermalThrottling;

    std::list<Session> mSessionHistory;

    std::shared_ptr<Watchdog> mWatchdog;

    std::shared_ptr<Pacer> mPacer;



    // Only allow MediaTranscodingService and unit tests to instantiate.

    TranscodingSessionController(const TranscoderFactoryType& transcoderFactory,

                                 const std::shared_ptr<UidPolicyInterface>& uidPolicy,

                                 const std::shared_ptr<ResourcePolicyInterface>& resourcePolicy,

                                 const std::shared_ptr<ThermalPolicyInterface>& thermalPolicy);

                                 const std::shared_ptr<ThermalPolicyInterface>& thermalPolicy,

                                 const ControllerConfig* config = nullptr);



    void dumpSession_l(const Session& session, String8& result, bool closedSession = false);

    Session* getTopSession_l();

media/libmediatranscoding/tests/TranscodingSessionController_tests.cpp

+94 −23

File changed.

Preview size limit exceeded, changes collapsed.

media/libmediatranscoding/transcoder/MediaSampleWriter.cpp

+2 −2
Original line number Diff line number Diff line
@@ -237,8 +237,8 @@ media_status_t MediaSampleWriter::runWriterLoop(bool* wasStopped) NO_THREAD_SAFE 
    }



    std::chrono::microseconds updateInterval(mHeartBeatIntervalUs);

    std::chrono::system_clock::time_point nextUpdateTime =

            std::chrono::system_clock::now() + updateInterval;

    std::chrono::steady_clock::time_point nextUpdateTime =

            std::chrono::steady_clock::now() + updateInterval;



    while (true) {

        if (trackEosCount >= mTracks.size()) {