Method processorTask

src/peprocessor.cpp:1021–1073

}

void PairEndProcessor::processorTask(ThreadConfig* config)
{
    SingleProducerSingleConsumerList<ReadPack*>* inputLeft = config->getLeftInput();
    SingleProducerSingleConsumerList<ReadPack*>* inputRight = config->getRightInput();
    while(true) {
        if(config->canBeStopped()){
            break;
        }
        while(inputLeft->canBeConsumed() && inputRight->canBeConsumed()) {
            ReadPack* dataLeft = inputLeft->consume();
            ReadPack* dataRight = inputRight->consume();
            processPairEnd(dataLeft, dataRight, config);
        }
        if(inputLeft->isProducerFinished() && !inputLeft->canBeConsumed()) {
            break;
        } else if(inputRight->isProducerFinished() && !inputRight->canBeConsumed()) {
            break;
        } else {
            std::unique_lock<std::mutex> lk(mBackpressureMtx);
            mBackpressureCV.wait_for(lk, std::chrono::milliseconds(1));
        }
    }
    inputLeft->setConsumerFinished();
    inputRight->setConsumerFinished();

    int finishedCount = mFinishedThreads.fetch_add(1, std::memory_order_release) + 1;
    if(mOptions->verbose) {
        string msg = "thread " + to_string(config->getThreadId() + 1) + " data processing completed";
        loginfo(msg);
    }

    if(finishedCount == mOptions->thread) {
        if(mLeftWriter)
            mLeftWriter->setInputCompleted();
        if(mRightWriter)
            mRightWriter->setInputCompleted();
        if(mUnpairedLeftWriter)
            mUnpairedLeftWriter->setInputCompleted();
        if(mUnpairedRightWriter)
            mUnpairedRightWriter->setInputCompleted();
        if(mMergedWriter)
            mMergedWriter->setInputCompleted();
        if(mFailedWriter)
            mFailedWriter->setInputCompleted();
        if(mOverlappedWriter)
            mOverlappedWriter->setInputCompleted();
    }
    
    if(mOptions->verbose) {
        string msg = "thread " + to_string(config->getThreadId() + 1) + " finished";
        loginfo(msg);
    }
}

void PairEndProcessor::writerTask(WriterThread* config)
{