fix: scale reader backpressure with worker thread count (#721)

src/common.h

@@ -36,6 +36,14 @@ static const int PACK_SIZE = 1000;
 // similar peak memory: 32 × 1000 reads ≈ 32K reads in flight ≈ old 128 × 256.
 static const int PACK_IN_MEM_LIMIT = 32;
 
+// Each worker thread owns one input list, and a list's newest item only becomes
+// consumable once another item is produced behind it (or the producer finishes).
+// Reader backpressure must therefore allow at least one in-flight pack per worker;
+// otherwise, with more workers than PACK_IN_MEM_LIMIT, readers stop before any
+// worker has a consumable pack and all threads wait forever (#721).
+inline long packInMemLimit(int threads) {
+    return threads * 2L > PACK_IN_MEM_LIMIT ? threads * 2L : PACK_IN_MEM_LIMIT;
+}
 
 // different filtering results, bigger number means worse
 // if r1 and r2 are both failed, then the bigger one of the two results will be recorded

src/peprocessor.cpp

@@ -15,6 +15,7 @@
 
 PairEndProcessor::PairEndProcessor(Options* opt){
     mOptions = opt;
+    mPackInMemLimit = packInMemLimit(mOptions->thread);
     mLeftReaderFinished = false;
     mRightReaderFinished = false;
     mFinishedThreads = 0;
@@ -820,12 +821,12 @@ void PairEndProcessor::readerTask(bool isLeft)
             {
                 std::unique_lock<std::mutex> lk(mBackpressureMtx);
                 if(isLeft) {
-                    while(mLeftPackReadCounter - mPackProcessedCounter.load(std::memory_order_acquire) > PACK_IN_MEM_LIMIT){
+                    while(mLeftPackReadCounter - mPackProcessedCounter.load(std::memory_order_acquire) > mPackInMemLimit){
                         slept++;
                         mBackpressureCV.wait_for(lk, std::chrono::milliseconds(1));
                     }
                 } else {
-                    while(mRightPackReadCounter - mPackProcessedCounter.load(std::memory_order_acquire) > PACK_IN_MEM_LIMIT){
+                    while(mRightPackReadCounter - mPackProcessedCounter.load(std::memory_order_acquire) > mPackInMemLimit){
                         slept++;
                         mBackpressureCV.wait_for(lk, std::chrono::milliseconds(1));
                     }
@@ -834,9 +835,9 @@ void PairEndProcessor::readerTask(bool isLeft)
             readNum += count;
             // if the writer threads are far behind this producer, sleep and wait
             // check this only when necessary
-            if(readNum % (PACK_SIZE * PACK_IN_MEM_LIMIT) == 0 && mLeftWriter) {
+            if(readNum % (PACK_SIZE * mPackInMemLimit) == 0 && mLeftWriter) {
                 std::unique_lock<std::mutex> lk(mBackpressureMtx);
-                while( (mLeftWriter && mLeftWriter->bufferLength() > PACK_IN_MEM_LIMIT) || (mRightWriter && mRightWriter->bufferLength() > PACK_IN_MEM_LIMIT) ){
+                while( (mLeftWriter && mLeftWriter->bufferLength() > mPackInMemLimit) || (mRightWriter && mRightWriter->bufferLength() > mPackInMemLimit) ){
                     slept++;
                     mBackpressureCV.wait_for(lk, std::chrono::milliseconds(1));
                 }
@@ -962,17 +963,17 @@ void PairEndProcessor::interleavedReaderTask()
             // if the consumer is far behind this producer, sleep and wait to limit memory usage
             {
                 std::unique_lock<std::mutex> lk(mBackpressureMtx);
-                while(mLeftPackReadCounter - mPackProcessedCounter.load(std::memory_order_acquire) > PACK_IN_MEM_LIMIT){
+                while(mLeftPackReadCounter - mPackProcessedCounter.load(std::memory_order_acquire) > mPackInMemLimit){
                     slept++;
                     mBackpressureCV.wait_for(lk, std::chrono::milliseconds(1));
                 }
             }
             readNum += count;
             // if the writer threads are far behind this producer, sleep and wait
             // check this only when necessary
-            if(readNum % (PACK_SIZE * PACK_IN_MEM_LIMIT) == 0 && mLeftWriter) {
+            if(readNum % (PACK_SIZE * mPackInMemLimit) == 0 && mLeftWriter) {
                 std::unique_lock<std::mutex> lk(mBackpressureMtx);
-                while( (mLeftWriter && mLeftWriter->bufferLength() > PACK_IN_MEM_LIMIT) || (mRightWriter && mRightWriter->bufferLength() > PACK_IN_MEM_LIMIT) ){
+                while( (mLeftWriter && mLeftWriter->bufferLength() > mPackInMemLimit) || (mRightWriter && mRightWriter->bufferLength() > mPackInMemLimit) ){
                     slept++;
                     mBackpressureCV.wait_for(lk, std::chrono::milliseconds(1));
                 }

src/peprocessor.h

@@ -64,6 +64,7 @@ class PairEndProcessor{
     size_t mLeftPackReadCounter;
     size_t mRightPackReadCounter;
     alignas(128) atomic_long mPackProcessedCounter;
+    long mPackInMemLimit;
     ReadPool* mLeftReadPool;
     ReadPool* mRightReadPool;
     atomic_bool shouldStopReading;

src/seprocessor.cpp

@@ -14,6 +14,7 @@
 
 SingleEndProcessor::SingleEndProcessor(Options* opt){
     mOptions = opt;
+    mPackInMemLimit = packInMemLimit(mOptions->thread);
     mReaderFinished = false;
     mFinishedThreads = 0;
     mFilter = new Filter(opt);
@@ -382,17 +383,17 @@ void SingleEndProcessor::readerTask()
             // if the processor is far behind this reader, sleep and wait to limit memory usage
             {
                 std::unique_lock<std::mutex> lk(mBackpressureMtx);
-                while( mPackReadCounter - mPackProcessedCounter.load(std::memory_order_acquire) > PACK_IN_MEM_LIMIT){
+                while( mPackReadCounter - mPackProcessedCounter.load(std::memory_order_acquire) > mPackInMemLimit){
                     slept++;
                     mBackpressureCV.wait_for(lk, std::chrono::milliseconds(1));
                 }
             }
             readNum += count;
             // if the writer threads are far behind this reader, sleep and wait
             // check this only when necessary
-            if(readNum % (PACK_SIZE * PACK_IN_MEM_LIMIT) == 0 && mLeftWriter) {
+            if(readNum % (PACK_SIZE * mPackInMemLimit) == 0 && mLeftWriter) {
                 std::unique_lock<std::mutex> lk(mBackpressureMtx);
-                while(mLeftWriter->bufferLength() > PACK_IN_MEM_LIMIT) {
+                while(mLeftWriter->bufferLength() > mPackInMemLimit) {
                     slept++;
                     mBackpressureCV.wait_for(lk, std::chrono::milliseconds(1));
                 }

src/seprocessor.h

@@ -50,6 +50,7 @@ class SingleEndProcessor{
     SingleProducerSingleConsumerList<ReadPack*>** mInputLists;
     size_t mPackReadCounter;
     alignas(128) atomic_long mPackProcessedCounter;
+    long mPackInMemLimit;
     ReadPool* mReadPool;
     std::mutex mBackpressureMtx;
     std::condition_variable mBackpressureCV;