Fix writer drain deadlock

scripts/test_writer_deterministic_md5.sh

@@ -0,0 +1,60 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+cd "$(dirname "$0")/.."
+
+FASTP="${FASTP:-./fastp}"
+if [ ! -x "$FASTP" ]; then
+    echo "ERROR: fastp binary not found at '$FASTP' (run 'make' first)" >&2
+    exit 2
+fi
+
+R1=testdata/R1.fq
+R2=testdata/R2.fq
+for f in "$R1" "$R2"; do
+    [ -r "$f" ] || { echo "ERROR: missing fixture $f" >&2; exit 2; }
+done
+
+TMP="$(mktemp -d 2>/dev/null || mktemp -d -t fastp-md5)"
+trap 'rm -rf "$TMP"' EXIT
+
+run_fastp() {
+    local run_id="$1"
+    perl -e 'alarm shift @ARGV; exec @ARGV or die "exec: $!"' 30 \
+        "$FASTP" \
+        -i "$R1" -I "$R2" \
+        -o "$TMP/o1.$run_id.fq" -O "$TMP/o2.$run_id.fq" \
+        -w 16 \
+        --disable_adapter_trimming --disable_quality_filtering --disable_length_filtering \
+        --json "$TMP/r.$run_id.json" --html "$TMP/r.$run_id.html" \
+        >"$TMP/run.$run_id.log" 2>&1
+}
+
+md5_file() {
+    if command -v md5sum >/dev/null 2>&1; then
+        set -- $(md5sum "$1")
+        printf '%s\n' "$1"
+    else
+        md5 -q "$1"
+    fi
+}
+
+printf '[case] %-50s ' "PE w=16 repeat decompressed MD5"
+run_fastp 1
+base1="$(md5_file "$TMP/o1.1.fq")"
+base2="$(md5_file "$TMP/o2.1.fq")"
+for i in 2 3 4 5; do
+    run_fastp "$i"
+    md51="$(md5_file "$TMP/o1.$i.fq")"
+    md52="$(md5_file "$TMP/o2.$i.fq")"
+    if [ "$md51" != "$base1" ] || [ "$md52" != "$base2" ]; then
+        echo "FAIL"
+        echo "R1 baseline=$base1 run$i=$md51" >&2
+        echo "R2 baseline=$base2 run$i=$md52" >&2
+        exit 1
+    fi
+done
+echo "OK"
+
+echo
+echo "ALL PASSED"

scripts/test_writer_no_hang.sh

@@ -0,0 +1,72 @@
+#!/usr/bin/env bash
+# Regression test for WriterThread round-robin SPSC deadlock.
+# Background: src/writerthread.cpp:output() rotates per-tid SPSC lists. When a
+# slot is permanently empty (producerFinished && isEmpty) the rotation must
+# skip the slot; otherwise the writer loops on usleep forever while other slots
+# still hold items. See: project_master_writer_skip_drained.md.
+#
+# Each case must finish within the per-case timeout. SIGALRM (perl alarm) kills
+# the binary on hang, producing a non-zero exit that this script reports.
+
+set -euo pipefail
+
+cd "$(dirname "$0")/.."
+
+FASTP="${FASTP:-./fastp}"
+if [ ! -x "$FASTP" ]; then
+    echo "ERROR: fastp binary not found at '$FASTP' (run 'make' first)" >&2
+    exit 2
+fi
+
+R1=testdata/R1.fq
+R2=testdata/R2.fq
+for f in "$R1" "$R2"; do
+    [ -r "$f" ] || { echo "ERROR: missing fixture $f" >&2; exit 2; }
+done
+
+TMP="$(mktemp -d 2>/dev/null || mktemp -d -t fastp-spsc)"
+trap 'rm -rf "$TMP"' EXIT
+
+run_case() {
+    local label="$1"; shift
+    local timeout_s="$1"; shift
+    printf '[case] %-50s ' "$label"
+    if perl -e 'alarm shift @ARGV; exec @ARGV or die "exec: $!"' "$timeout_s" \
+            "$FASTP" "$@" >"$TMP/out.log" 2>&1; then
+        echo "OK"
+    else
+        local rc=$?
+        echo "FAIL (exit $rc, possibly SIGALRM=hang)"
+        echo "----- tail of $TMP/out.log -----" >&2
+        tail -30 "$TMP/out.log" >&2 || true
+        exit 1
+    fi
+}
+
+# Trailing-empty slot: 8 reads -> 1 pack, w=16 workers => 15 lists never produce.
+run_case "PE w=16 trailing-empty"  15 \
+  -i "$R1" -I "$R2" -o "$TMP/o1.fq" -O "$TMP/o2.fq" -w 16 \
+  --disable_adapter_trimming --disable_quality_filtering --disable_length_filtering \
+  --json "$TMP/r.json" --html "$TMP/r.html"
+
+# Optional writers (--failed_out, --unpaired*) take the conditional-input path
+# that historically had the highest deadlock blast radius.
+run_case "PE w=16 + failed_out + unpaired" 15 \
+  -i "$R1" -I "$R2" -o "$TMP/o1.fq" -O "$TMP/o2.fq" -w 16 \
+  --failed_out "$TMP/failed.fq" --unpaired1 "$TMP/u1.fq" --unpaired2 "$TMP/u2.fq" \
+  --json "$TMP/r.json" --html "$TMP/r.html"
+
+run_case "SE w=16 trailing-empty"  15 \
+  -i "$R1" -o "$TMP/o.fq" -w 16 \
+  --disable_adapter_trimming --disable_quality_filtering --disable_length_filtering \
+  --json "$TMP/r.json" --html "$TMP/r.html"
+
+# Race-prone repeat to catch timing-dependent hangs.
+for i in 1 2 3 4 5; do
+    run_case "PE w=64 trailing-empty repeat $i" 20 \
+      -i "$R1" -I "$R2" -o "$TMP/o1.fq" -O "$TMP/o2.fq" -w 64 \
+      --json "$TMP/r.json" --html "$TMP/r.html"
+done
+
+echo
+echo "ALL PASSED"

src/writerthread.cpp

@@ -22,6 +22,8 @@ WriterThread::WriterThread(Options* opt, string filename, bool isSTDOUT){
     mCompBufs = NULL;
     mCompBufSizes = NULL;
     mBufferLists = NULL;
+    mNextOutputSeq = 0;
+    mOrderedOutput = true;
 
     if (mPwriteMode) {
         mFd = open(mFilename.c_str(), O_WRONLY | O_CREAT | O_TRUNC, 0644);
@@ -58,19 +60,29 @@ WriterThread::~WriterThread() {
 bool WriterThread::isCompleted()
 {
     if (mPwriteMode) return true;  // no writer thread needed
-    return mInputCompleted && (mBufferLength==0);
+    return mInputCompleted.load(std::memory_order_acquire) && allBufferListsDrained();
+}
+
+bool WriterThread::allBufferListsDrained() {
+    if(mBufferLists == NULL)
+        return true;
+    for(int t=0; t<mOptions->thread; t++) {
+        if(!mBufferLists[t]->isProducerFinished() || !mBufferLists[t]->isEmpty())
+            return false;
+    }
+    return true;
 }
 
 bool WriterThread::setInputCompleted() {
     if (mPwriteMode) {
         setInputCompletedPwrite();
-        mInputCompleted = true;
+        mInputCompleted.store(true, std::memory_order_release);
         return true;
     }
-    mInputCompleted = true;
     for(int t=0; t<mOptions->thread; t++) {
         mBufferLists[t]->setProducerFinished();
     }
+    mInputCompleted.store(true, std::memory_order_release);
     return true;
 }
 
@@ -94,16 +106,40 @@ void WriterThread::setInputCompletedPwrite() {
 
 void WriterThread::output(){
     if (mPwriteMode) return;  // no-op
-    SingleProducerSingleConsumerList<string*>* list = mBufferLists[mWorkingBufferList];
-    if(!list->canBeConsumed()) {
-        usleep(100);
-    } else {
+    if(mOrderedOutput)
+        outputOrdered();
+    else
+        outputReady();
+}
+
+void WriterThread::outputOrdered() {
+    int tid = mNextOutputSeq % mOptions->thread;
+    SingleProducerSingleConsumerList<string*>* list = mBufferLists[tid];
+    if(list->canBeConsumed()) {
         string* str = list->consume();
         mWriter1->write(str->data(), str->length());
         delete str;
         mBufferLength--;
+        mNextOutputSeq++;
+        return;
+    }
+    usleep(100);
+}
+
+void WriterThread::outputReady() {
+    for(int i=0; i<mOptions->thread; i++) {
+        SingleProducerSingleConsumerList<string*>* list = mBufferLists[mWorkingBufferList];
+        if(list->canBeConsumed()) {
+            string* str = list->consume();
+            mWriter1->write(str->data(), str->length());
+            delete str;
+            mBufferLength--;
+            mWorkingBufferList = (mWorkingBufferList+1)%mOptions->thread;
+            return;
+        }
         mWorkingBufferList = (mWorkingBufferList+1)%mOptions->thread;
     }
+    usleep(100);
 }
 
 void WriterThread::input(int tid, string* data) {

src/writerthread.h

@@ -35,6 +35,7 @@ class WriterThread{
     void output();
     void input(int tid, string* data);
     bool setInputCompleted();
+    bool allBufferListsDrained();
 
     long bufferLength() {return mBufferLength;};
     string getFilename() {return mFilename;}
@@ -43,17 +44,21 @@ class WriterThread{
 private:
     void deleteWriter();
     void inputPwrite(int tid, string* data);
+    void outputOrdered();
+    void outputReady();
     void setInputCompletedPwrite();
 
 private:
     Writer* mWriter1;
     Options* mOptions;
     string mFilename;
 
-    bool mInputCompleted;
+    atomic_bool mInputCompleted;
     atomic_long mBufferLength;
     SingleProducerSingleConsumerList<string*>** mBufferLists;
     int mWorkingBufferList;
+    size_t mNextOutputSeq;
+    bool mOrderedOutput;
 
     // pwrite mode: parallel libdeflate gz compression + direct file write
     bool mPwriteMode;