| 723 | } |
| 724 | |
| 725 | void PairEndProcessor::readerTask(bool isLeft) |
| 726 | { |
| 727 | if(mOptions->verbose) { |
| 728 | if(isLeft) |
| 729 | loginfo("start to load data of read1"); |
| 730 | else |
| 731 | loginfo("start to load data of read2"); |
| 732 | } |
| 733 | long lastReported = 0; |
| 734 | int slept = 0; |
| 735 | long readNum = 0; |
| 736 | bool splitSizeReEvaluated = false; |
| 737 | Read** data = new Read*[PACK_SIZE]; |
| 738 | memset(data, 0, sizeof(Read*)*PACK_SIZE); |
| 739 | // BGZF decompress thread budget per reader: |
| 740 | // (total CPU cores - worker threads - reader threads - writer threads) / number of gz inputs |
| 741 | int cpus = std::thread::hardware_concurrency(); |
| 742 | int existingThreads = mOptions->thread + 4; // workers + 2 readers + 2 writers |
| 743 | int gzInputs = (ends_with(mOptions->in1, ".gz") ? 1 : 0) + (ends_with(mOptions->in2, ".gz") ? 1 : 0); |
| 744 | int bgzfBudget = 0; |
| 745 | if (gzInputs > 0) |
| 746 | bgzfBudget = std::max(1, ((int)cpus - existingThreads) / gzInputs); |
| 747 | |
| 748 | FastqReader* reader = NULL; |
| 749 | if(isLeft) { |
| 750 | reader = new FastqReader(mOptions->in1, true, mOptions->phred64, bgzfBudget); |
| 751 | reader->setReadPool(mLeftReadPool); |
| 752 | } |
| 753 | else { |
| 754 | reader = new FastqReader(mOptions->in2, true, mOptions->phred64, bgzfBudget); |
| 755 | reader->setReadPool(mRightReadPool); |
| 756 | } |
| 757 | |
| 758 | int count=0; |
| 759 | bool needToBreak = false; |
| 760 | while(true){ |
| 761 | if(shouldStopReading) |
| 762 | break; |
| 763 | Read* read = reader->read(); |
| 764 | if(!read || needToBreak){ |
| 765 | // the last pack |
| 766 | ReadPack* pack = new ReadPack; |
| 767 | pack->data = data; |
| 768 | pack->count = count; |
| 769 | |
| 770 | if(isLeft) { |
| 771 | mLeftInputLists[mLeftPackReadCounter % mOptions->thread]->produce(pack); |
| 772 | mLeftPackReadCounter++; |
| 773 | } else { |
| 774 | mRightInputLists[mRightPackReadCounter % mOptions->thread]->produce(pack); |
| 775 | mRightPackReadCounter++; |
| 776 | } |
| 777 | mBackpressureCV.notify_all(); |
| 778 | data = NULL; |
| 779 | if(read) { |
| 780 | delete read; |
| 781 | read = NULL; |
| 782 | } |
nothing calls this directly
no test coverage detected