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

Commit 4f68e81e authored by Amy's avatar Amy Committed by Amy Zhang
Browse files

Adding a DVR Record default implementation in Tuner HAL 1.0 

Test: cuttlefish
Bug: 135709325
Change-Id: I415426d6ec048bdd2ae61a3c5142ad02f1d7f1e4
(cherry picked from commit 5ed13574)
Merged-In: I415426d6ec048bdd2ae61a3c5142ad02f1d7f1e4
parent b442c9bc

tv/tuner/1.0/default/Demux.cpp

+87 −25
Original line number Diff line number Diff line
@@ -51,7 +51,8 @@ Return<Result> Demux::setFrontendDataSource(uint32_t frontendId) { 
    mFrontendSourceFile = mFrontend->getSourceFile();



    mTunerService->setFrontendAsDemuxSource(frontendId, mDemuxId);

    return startBroadcastInputLoop();



    return startFrontendInputLoop();

}



Return<void> Demux::openFilter(const DemuxFilterType& type, uint32_t bufferSize,
@@ -136,14 +137,14 @@ Return<void> Demux::openDvr(DvrType type, uint32_t bufferSize, const sp<IDvrCall 
        return Void();

    }



    sp<Dvr> dvr = new Dvr(type, bufferSize, cb, this);

    mDvr = new Dvr(type, bufferSize, cb, this);



    if (!dvr->createDvrMQ()) {

        _hidl_cb(Result::UNKNOWN_ERROR, dvr);

    if (!mDvr->createDvrMQ()) {

        _hidl_cb(Result::UNKNOWN_ERROR, mDvr);

        return Void();

    }



    _hidl_cb(Result::SUCCESS, dvr);

    _hidl_cb(Result::SUCCESS, mDvr);

    return Void();

}
@@ -166,13 +167,14 @@ Result Demux::removeFilter(uint32_t filterId) { 


    // resetFilterRecords(filterId);

    mUsedFilterIds.erase(filterId);

    mRecordFilterIds.erase(filterId);

    mUnusedFilterIds.insert(filterId);

    mFilters.erase(filterId);



    return Result::SUCCESS;

}



void Demux::startTsFilter(vector<uint8_t> data) {

void Demux::startBroadcastTsFilter(vector<uint8_t> data) {

    set<uint32_t>::iterator it;

    for (it = mUsedFilterIds.begin(); it != mUsedFilterIds.end(); it++) {

        uint16_t pid = ((data[1] & 0x1f) << 8) | ((data[2] & 0xff));
@@ -185,7 +187,17 @@ void Demux::startTsFilter(vector<uint8_t> data) { 
    }

}



bool Demux::startFilterDispatcher() {

void Demux::sendFrontendInputToRecord(vector<uint8_t> data) {

    set<uint32_t>::iterator it;

    for (it = mRecordFilterIds.begin(); it != mRecordFilterIds.end(); it++) {

        if (DEBUG_FILTER) {

            ALOGW("update record filter output");

        }

        mFilters[*it]->updateRecordOutput(data);

    }

}



bool Demux::startBroadcastFilterDispatcher() {

    set<uint32_t>::iterator it;



    // Handle the output data per filter type
@@ -198,6 +210,18 @@ bool Demux::startFilterDispatcher() { 
    return true;

}



bool Demux::startRecordFilterDispatcher() {

    set<uint32_t>::iterator it;



    for (it = mRecordFilterIds.begin(); it != mRecordFilterIds.end(); it++) {

        if (mFilters[*it]->startRecordFilterHandler() != Result::SUCCESS) {

            return false;

        }

    }



    return true;

}



Result Demux::startFilterHandler(uint32_t filterId) {

    return mFilters[filterId]->startFilterHandler();

}
@@ -210,22 +234,22 @@ uint16_t Demux::getFilterTpid(uint32_t filterId) { 
    return mFilters[filterId]->getTpid();

}



Result Demux::startBroadcastInputLoop() {

    pthread_create(&mBroadcastInputThread, NULL, __threadLoopBroadcast, this);

    pthread_setname_np(mBroadcastInputThread, "broadcast_input_thread");

Result Demux::startFrontendInputLoop() {

    pthread_create(&mFrontendInputThread, NULL, __threadLoopFrontend, this);

    pthread_setname_np(mFrontendInputThread, "frontend_input_thread");



    return Result::SUCCESS;

}



void* Demux::__threadLoopBroadcast(void* user) {

void* Demux::__threadLoopFrontend(void* user) {

    Demux* const self = static_cast<Demux*>(user);

    self->broadcastInputThreadLoop();

    self->frontendInputThreadLoop();

    return 0;

}



void Demux::broadcastInputThreadLoop() {

    std::lock_guard<std::mutex> lock(mBroadcastInputThreadLock);

    mBroadcastInputThreadRunning = true;

void Demux::frontendInputThreadLoop() {

    std::lock_guard<std::mutex> lock(mFrontendInputThreadLock);

    mFrontendInputThreadRunning = true;

    mKeepFetchingDataFromFrontend = true;



    // open the stream and get its length
@@ -234,20 +258,20 @@ void Demux::broadcastInputThreadLoop() { 
    int packetSize = 188;

    int writePacketAmount = 6;

    char* buffer = new char[packetSize];

    ALOGW("[Demux] broadcast input thread loop start %s", mFrontendSourceFile.c_str());

    ALOGW("[Demux] Frontend input thread loop start %s", mFrontendSourceFile.c_str());

    if (!inputData.is_open()) {

        mBroadcastInputThreadRunning = false;

        mFrontendInputThreadRunning = false;

        ALOGW("[Demux] Error %s", strerror(errno));

    }



    while (mBroadcastInputThreadRunning) {

    while (mFrontendInputThreadRunning) {

        // move the stream pointer for packet size * 6 every read until the end

        while (mKeepFetchingDataFromFrontend) {

            for (int i = 0; i < writePacketAmount; i++) {

                inputData.read(buffer, packetSize);

                if (!inputData) {

                    mKeepFetchingDataFromFrontend = false;

                    mBroadcastInputThreadRunning = false;

                    mFrontendInputThreadRunning = false;

                    break;

                }

                // filter and dispatch filter output
@@ -256,23 +280,61 @@ void Demux::broadcastInputThreadLoop() { 
                for (int index = 0; index < byteBuffer.size(); index++) {

                    byteBuffer[index] = static_cast<uint8_t>(buffer[index]);

                }

                startTsFilter(byteBuffer);

                if (mIsRecording) {

                    // Feed the data into the Dvr recording input

                    sendFrontendInputToRecord(byteBuffer);

                } else {

                    // Feed the data into the broadcast demux filter

                    startBroadcastTsFilter(byteBuffer);

                }

            }

            if (mIsRecording) {

                // Dispatch the data into the broadcasting filters.

                startRecordFilterDispatcher();

            } else {

                // Dispatch the data into the broadcasting filters.

                startBroadcastFilterDispatcher();

            }

            startFilterDispatcher();

            usleep(100);

        }

    }



    ALOGW("[Demux] Broadcast Input thread end.");

    ALOGW("[Demux] Frontend Input thread end.");

    delete[] buffer;

    inputData.close();

}



void Demux::stopBroadcastInput() {

void Demux::stopFrontendInput() {

    ALOGD("[Demux] stop frontend on demux");

    mKeepFetchingDataFromFrontend = false;

    mBroadcastInputThreadRunning = false;

    std::lock_guard<std::mutex> lock(mBroadcastInputThreadLock);

    mFrontendInputThreadRunning = false;

    std::lock_guard<std::mutex> lock(mFrontendInputThreadLock);

}



void Demux::setIsRecording(bool isRecording) {

    mIsRecording = isRecording;

}



bool Demux::attachRecordFilter(int filterId) {

    if (mFilters[filterId] == nullptr || mDvr == nullptr) {

        return false;

    }



    mRecordFilterIds.insert(filterId);

    mFilters[filterId]->attachFilterToRecord(mDvr);



    return true;

}



bool Demux::detachRecordFilter(int filterId) {

    if (mFilters[filterId] == nullptr || mDvr == nullptr) {

        return false;

    }



    mRecordFilterIds.erase(filterId);

    mFilters[filterId]->detachFilterFromRecord();



    return true;

}



}  // namespace implementation

tv/tuner/1.0/default/Demux.h

+30 −9
Original line number Diff line number Diff line
@@ -81,11 +81,14 @@ class Demux : public IDemux { 
    virtual Return<Result> disconnectCiCam() override;



    // Functions interacts with Tuner Service

    void stopBroadcastInput();

    void stopFrontendInput();

    Result removeFilter(uint32_t filterId);

    bool attachRecordFilter(int filterId);

    bool detachRecordFilter(int filterId);

    Result startFilterHandler(uint32_t filterId);

    void updateFilterOutput(uint16_t filterId, vector<uint8_t> data);

    uint16_t getFilterTpid(uint32_t filterId);

    void setIsRecording(bool isRecording);



  private:

    // Tuner service
@@ -101,9 +104,9 @@ class Demux : public IDemux { 
        uint32_t filterId;

    };



    Result startBroadcastInputLoop();

    static void* __threadLoopBroadcast(void* user);

    void broadcastInputThreadLoop();

    Result startFrontendInputLoop();

    static void* __threadLoopFrontend(void* user);

    void frontendInputThreadLoop();



    /**

     * To create a FilterMQ with the the next available Filter ID.
@@ -117,9 +120,13 @@ class Demux : public IDemux { 
    /**

     * A dispatcher to read and dispatch input data to all the started filters.

     * Each filter handler handles the data filtering/output writing/filterEvent updating.

     * Note that recording filters are not included.

     */

    bool startFilterDispatcher();

    void startTsFilter(vector<uint8_t> data);

    bool startBroadcastFilterDispatcher();

    void startBroadcastTsFilter(vector<uint8_t> data);



    void sendFrontendInputToRecord(vector<uint8_t> data);

    bool startRecordFilterDispatcher();



    uint32_t mDemuxId;

    uint32_t mCiCamId;
@@ -140,19 +147,33 @@ class Demux : public IDemux { 
     * and added into usedFilterIds.

     */

    set<uint32_t> mUnusedFilterIds;

    /**

     * Record all the attached record filter Ids.

     * Any removed filter id should be removed from this set.

     */

    set<uint32_t> mRecordFilterIds;

    /**

     * A list of created FilterMQ ptrs.

     * The array number is the filter ID.

     */

    std::map<uint32_t, sp<Filter>> mFilters;



    /**

     * Local reference to the opened DVR object.

     */

    sp<Dvr> mDvr;



    // Thread handlers

    pthread_t mBroadcastInputThread;

    pthread_t mFrontendInputThread;

    /**

     * If a specific filter's writing loop is still running

     */

    bool mBroadcastInputThreadRunning;

    bool mFrontendInputThreadRunning;

    bool mKeepFetchingDataFromFrontend;

    /**

     * If the dvr recording is running.

     */

    bool mIsRecording = false;

    /**

     * Lock to protect writes to the FMQs

     */
@@ -160,7 +181,7 @@ class Demux : public IDemux { 
    /**

     * Lock to protect writes to the input status

     */

    std::mutex mBroadcastInputThreadLock;

    std::mutex mFrontendInputThreadLock;



    // temp handle single PES filter

    // TODO handle mulptiple Pes filters

tv/tuner/1.0/default/Dvr.cpp

+63 −2
Original line number Diff line number Diff line
@@ -70,7 +70,14 @@ Return<Result> Dvr::attachFilter(const sp<IFilter>& filter) { 
        return status;

    }



    // check if the attached filter is a record filter



    mFilters[filterId] = filter;

    mIsRecordFilterAttached = true;

    if (!mDemux->attachRecordFilter(filterId)) {

        return Result::INVALID_ARGUMENT;

    }

    mDemux->setIsRecording(mIsRecordStarted | mIsRecordFilterAttached);



    return Result::SUCCESS;

}
@@ -95,6 +102,15 @@ Return<Result> Dvr::detachFilter(const sp<IFilter>& filter) { 
    it = mFilters.find(filterId);

    if (it != mFilters.end()) {

        mFilters.erase(filterId);

        if (!mDemux->detachRecordFilter(filterId)) {

            return Result::INVALID_ARGUMENT;

        }

    }



    // If all the filters are detached, record can't be started

    if (mFilters.empty()) {

        mIsRecordFilterAttached = false;

        mDemux->setIsRecording(mIsRecordStarted | mIsRecordFilterAttached);

    }



    return Result::SUCCESS;
@@ -115,8 +131,9 @@ Return<Result> Dvr::start() { 
        pthread_create(&mDvrThread, NULL, __threadLoopPlayback, this);

        pthread_setname_np(mDvrThread, "playback_waiting_loop");

    } else if (mType == DvrType::RECORD) {

        /*pthread_create(&mInputThread, NULL, __threadLoopInput, this);

        pthread_setname_np(mInputThread, "playback_waiting_loop");*/

        mRecordStatus = RecordStatus::DATA_READY;

        mIsRecordStarted = true;

        mDemux->setIsRecording(mIsRecordStarted | mIsRecordFilterAttached);

    }



    // TODO start another thread to send filter status callback to the framework
@@ -131,12 +148,17 @@ Return<Result> Dvr::stop() { 


    std::lock_guard<std::mutex> lock(mDvrThreadLock);



    mIsRecordStarted = false;

    mDemux->setIsRecording(mIsRecordStarted | mIsRecordFilterAttached);



    return Result::SUCCESS;

}



Return<Result> Dvr::flush() {

    ALOGV("%s", __FUNCTION__);



    mRecordStatus = RecordStatus::DATA_READY;



    return Result::SUCCESS;

}
@@ -272,6 +294,45 @@ bool Dvr::startFilterDispatcher() { 
    return true;

}



bool Dvr::writeRecordFMQ(const std::vector<uint8_t>& data) {

    std::lock_guard<std::mutex> lock(mWriteLock);

    ALOGW("[Dvr] write record FMQ");

    if (mDvrMQ->write(data.data(), data.size())) {

        mDvrEventFlag->wake(static_cast<uint32_t>(DemuxQueueNotifyBits::DATA_READY));

        maySendRecordStatusCallback();

        return true;

    }



    maySendRecordStatusCallback();

    return false;

}



void Dvr::maySendRecordStatusCallback() {

    std::lock_guard<std::mutex> lock(mRecordStatusLock);

    int availableToRead = mDvrMQ->availableToRead();

    int availableToWrite = mDvrMQ->availableToWrite();



    RecordStatus newStatus = checkRecordStatusChange(availableToWrite, availableToRead,

                                                     mDvrSettings.record().highThreshold,

                                                     mDvrSettings.record().lowThreshold);

    if (mRecordStatus != newStatus) {

        mCallback->onRecordStatus(newStatus);

        mRecordStatus = newStatus;

    }

}



RecordStatus Dvr::checkRecordStatusChange(uint32_t availableToWrite, uint32_t availableToRead,

                                          uint32_t highThreshold, uint32_t lowThreshold) {

    if (availableToWrite == 0) {

        return DemuxFilterStatus::OVERFLOW;

    } else if (availableToRead > highThreshold) {

        return DemuxFilterStatus::HIGH_WATER;

    } else if (availableToRead < lowThreshold) {

        return DemuxFilterStatus::LOW_WATER;

    }

    return mRecordStatus;

}



}  // namespace implementation

}  // namespace V1_0

}  // namespace tuner

tv/tuner/1.0/default/Dvr.h

+13 −2
Original line number Diff line number Diff line
@@ -79,6 +79,8 @@ class Dvr : public IDvr { 
     * Return false is any of the above processes fails.

     */

    bool createDvrMQ();

    void sendBroadcastInputToDvrRecord(vector<uint8_t> byteBuffer);

    bool writeRecordFMQ(const std::vector<uint8_t>& data);



  private:

    // Demux service
@@ -95,6 +97,8 @@ class Dvr : public IDvr { 
    void maySendRecordStatusCallback();

    PlaybackStatus checkPlaybackStatusChange(uint32_t availableToWrite, uint32_t availableToRead,

                                             uint32_t highThreshold, uint32_t lowThreshold);

    RecordStatus checkRecordStatusChange(uint32_t availableToWrite, uint32_t availableToRead,

                                         uint32_t highThreshold, uint32_t lowThreshold);

    /**

     * A dispatcher to read and dispatch input data to all the started filters.

     * Each filter handler handles the data filtering/output writing/filterEvent updating.
@@ -103,9 +107,9 @@ class Dvr : public IDvr { 
    void startTpidFilter(vector<uint8_t> data);

    bool startFilterDispatcher();

    static void* __threadLoopPlayback(void* user);

    static void* __threadLoopBroadcast(void* user);

    static void* __threadLoopRecord(void* user);

    void playbackThreadLoop();

    void broadcastInputThreadLoop();

    void recordThreadLoop();



    unique_ptr<DvrMQ> mDvrMQ;

    EventFlag* mDvrEventFlag;
@@ -121,6 +125,7 @@ class Dvr : public IDvr { 


    // FMQ status local records

    PlaybackStatus mPlaybackStatus;

    RecordStatus mRecordStatus;

    /**

     * If a specific filter's writing loop is still running

     */
@@ -135,10 +140,16 @@ class Dvr : public IDvr { 
     * Lock to protect writes to the input status

     */

    std::mutex mPlaybackStatusLock;

    std::mutex mRecordStatusLock;

    std::mutex mBroadcastInputThreadLock;

    std::mutex mDvrThreadLock;



    const bool DEBUG_DVR = false;



    // Booleans to check if recording is running.

    // Recording is ready when both of the following are set to true.

    bool mIsRecordStarted = false;

    bool mIsRecordFilterAttached = false;

};



}  // namespace implementation

tv/tuner/1.0/default/Filter.cpp

+41 −10
Original line number Diff line number Diff line
@@ -265,10 +265,16 @@ uint16_t Filter::getTpid() { 


void Filter::updateFilterOutput(vector<uint8_t> data) {

    std::lock_guard<std::mutex> lock(mFilterOutputLock);

    ALOGD("[Filter] handler output updated");

    ALOGD("[Filter] filter output updated");

    mFilterOutput.insert(mFilterOutput.end(), data.begin(), data.end());

}



void Filter::updateRecordOutput(vector<uint8_t> data) {

    std::lock_guard<std::mutex> lock(mRecordFilterOutputLock);

    ALOGD("[Filter] record filter output updated");

    mRecordFilterOutput.insert(mRecordFilterOutput.end(), data.begin(), data.end());

}



Result Filter::startFilterHandler() {

    std::lock_guard<std::mutex> lock(mFilterOutputLock);

    switch (mType.mainType) {
@@ -292,12 +298,11 @@ Result Filter::startFilterHandler() { 
                case DemuxTsFilterType::PCR:

                    startPcrFilterHandler();

                    break;

                case DemuxTsFilterType::RECORD:

                    startRecordFilterHandler();

                    break;

                case DemuxTsFilterType::TEMI:

                    startTemiFilterHandler();

                    break;

                default:

                    break;

            }

            break;

        case DemuxFilterMainType::MMTP:
@@ -342,12 +347,16 @@ Result Filter::startPesFilterHandler() { 
        if (mPesSizeLeft == 0) {

            uint32_t prefix = (mFilterOutput[i + 4] << 16) | (mFilterOutput[i + 5] << 8) |

                              mFilterOutput[i + 6];

            if (DEBUG_FILTER) {

                ALOGD("[Filter] prefix %d", prefix);

            }

            if (prefix == 0x000001) {

                // TODO handle mulptiple Pes filters

                mPesSizeLeft = (mFilterOutput[i + 8] << 8) | mFilterOutput[i + 9];

                mPesSizeLeft += 6;

                if (DEBUG_FILTER) {

                    ALOGD("[Filter] pes data length %d", mPesSizeLeft);

                }

            } else {

                continue;

            }
@@ -360,7 +369,9 @@ Result Filter::startPesFilterHandler() { 
        mPesOutput.insert(mPesOutput.end(), first, last);

        // size does not match then continue

        mPesSizeLeft -= endPoint;

        if (DEBUG_FILTER) {

            ALOGD("[Filter] pes data left %d", mPesSizeLeft);

        }

        if (mPesSizeLeft > 0) {

            continue;

        }
@@ -377,7 +388,9 @@ Result Filter::startPesFilterHandler() { 
                .streamId = mPesOutput[3],

                .dataLength = static_cast<uint16_t>(mPesOutput.size()),

        };

        if (DEBUG_FILTER) {

            ALOGD("[Filter] assembled pes data length %d", pesEvent.dataLength);

        }



        int size = mFilterEvent.events.size();

        mFilterEvent.events.resize(size + 1);
@@ -413,13 +426,23 @@ Result Filter::startMediaFilterHandler() { 
}



Result Filter::startRecordFilterHandler() {

    DemuxFilterTsRecordEvent tsRecordEvent;

    /*DemuxFilterTsRecordEvent tsRecordEvent;

    tsRecordEvent.pid.tPid(0);

    tsRecordEvent.indexMask.tsIndexMask(0x01);

    mFilterEvent.events.resize(1);

    mFilterEvent.events[0].tsRecord(tsRecordEvent);

*/

    std::lock_guard<std::mutex> lock(mRecordFilterOutputLock);

    if (mRecordFilterOutput.empty()) {

        return Result::SUCCESS;

    }



    mFilterOutput.clear();

    if (mDvr == nullptr || !mDvr->writeRecordFMQ(mRecordFilterOutput)) {

        ALOGD("[Filter] dvr fails to write into record FMQ.");

        return Result::UNKNOWN_ERROR;

    }



    mRecordFilterOutput.clear();

    return Result::SUCCESS;

}
@@ -462,6 +485,14 @@ bool Filter::writeDataToFilterMQ(const std::vector<uint8_t>& data) { 
    return false;

}



void Filter::attachFilterToRecord(const sp<Dvr> dvr) {

    mDvr = dvr;

}



void Filter::detachFilterFromRecord() {

    mDvr = nullptr;

}



}  // namespace implementation

}  // namespace V1_0

}  // namespace tuner