Skip to content

Commit 5481356

Browse files
Kim YannKim Yann
Kim Yann
authored and
Kim Yann
committed
perf: replace yield() with hwy::BlockUntilDifferent/WakeAll futex
Replace all std::this_thread::yield() busy-spin loops with hwy::BlockUntilDifferent/WakeAll from Highway's futex.h polyfill. This provides cross-platform kernel-level thread blocking (Linux futex, macOS __ulock, FreeBSD NanoSleep fallback) instead of CPU-burning spins. Changes: - writerthread: output() waits on mBufferLength via BlockUntilDifferent, input() wakes writer via WakeAll after produce - writerthread.h: add waitForBufferBelow() using BlockUntilDifferent loop - peprocessor: replace 6 yield() sites with atomic wait/notify on mPackProducedCounter and mPackProcessedCounter - seprocessor: same pattern as peprocessor for SE pipeline - Change counter types from atomic_long to atomic<uint32_t> for Highway futex compatibility (uint32_t required by BlockUntilDifferent) Benchmark (5M PE reads, gz→gz, -w 3): master: 56.5s wall, 8.0s sys, 2680K page-faults yield (before): 79.4s wall, 26.8s sys, 3278K page-faults futex (after): 47.8s wall, 1.4s sys, 120K page-faults wall -15%, sys -82%, page-faults -95% vs master Output md5 matches master (correctness verified)
1 parent 536e911 commit 5481356

7 files changed

Lines changed: 60 additions & 34 deletions

File tree

‎Makefile‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -23,9 +23,9 @@ TARGET := fastp
2323

2424
BIN_TARGET := ${TARGET}
2525

26-
CXX := /home/kimy/build-env/bin/x86_64-conda-linux-gnu-g++
27-
INCLUDE_DIRS ?= /home/kimy/build-env/include
28-
LIBRARY_DIRS ?= /home/kimy/build-env/lib
26+
CXX := /var/tmp/kimy/workspace/build-env/bin/x86_64-conda-linux-gnu-g++
27+
INCLUDE_DIRS ?= /var/tmp/kimy/workspace/build-env/include
28+
LIBRARY_DIRS ?= /var/tmp/kimy/workspace/build-env/lib
2929
CXXFLAGS := -std=c++23 -pthread -g -O3 -MD -MP -I. -I${DIR_INC} $(foreach includedir,$(INCLUDE_DIRS),-I$(includedir)) $(HWY_CFLAGS) $(ISAL_CFLAGS) $(DEFLATE_CFLAGS) ${CXXFLAGS}
3030
LIBS := -lisal -ldeflate -lhwy -lpthread
3131

‎src/peprocessor.cpp‎

Lines changed: 25 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
#include "peprocessor.h"
2+
#include "hwy/contrib/thread_pool/futex.h"
23
#include "fastqreader.h"
34
#include <iostream>
45
#include <print>
@@ -42,6 +43,7 @@ PairEndProcessor::PairEndProcessor(Options* opt)
4243
mLeftPackReadCounter = 0;
4344
mRightPackReadCounter = 0;
4445
mPackProcessedCounter = 0;
46+
mPackProducedCounter = 0;
4547

4648
mLeftReadPool = new ReadPool(mOptions);
4749
mRightReadPool = new ReadPool(mOptions);
@@ -686,7 +688,7 @@ bool PairEndProcessor::processPairEnd(ReadPack* leftPack, ReadPack* rightPack, T
686688
delete rightPack;
687689

688690
mPackProcessedCounter.fetch_add(1, std::memory_order_release);
689-
mPackProcessedCounter.notify_all();
691+
hwy::WakeAll(mPackProcessedCounter);
690692

691693
return true;
692694
}
@@ -758,6 +760,8 @@ void PairEndProcessor::readerTask(bool isLeft)
758760
mRightInputLists[mRightPackReadCounter % mOptions->thread]->produce(pack);
759761
mRightPackReadCounter++;
760762
}
763+
mPackProducedCounter.fetch_add(1, std::memory_order_release);
764+
hwy::WakeAll(mPackProducedCounter);
761765
data = NULL;
762766
if(read) {
763767
delete read;
@@ -794,32 +798,32 @@ void PairEndProcessor::readerTask(bool isLeft)
794798
mRightInputLists[mRightPackReadCounter % mOptions->thread]->produce(pack);
795799
mRightPackReadCounter++;
796800
}
801+
mPackProducedCounter.fetch_add(1, std::memory_order_release);
802+
hwy::WakeAll(mPackProducedCounter);
797803

798804
//re-initialize data for next pack
799805
data = new Read*[PACK_SIZE];
800806
memset(data, 0, sizeof(Read*)*PACK_SIZE);
801807
// if the processor is far behind this reader, wait to limit memory usage
802808
if(isLeft) {
803809
while(mLeftPackReadCounter - mPackProcessedCounter.load(std::memory_order_acquire) > PACK_IN_MEM_LIMIT){
804-
long cur = mPackProcessedCounter.load(std::memory_order_acquire);
805-
mPackProcessedCounter.wait(cur, std::memory_order_acquire);
810+
uint32_t cur = mPackProcessedCounter.load(std::memory_order_acquire);
811+
hwy::BlockUntilDifferent(cur, mPackProcessedCounter);
806812
slept++;
807813
}
808814
} else {
809815
while(mRightPackReadCounter - mPackProcessedCounter.load(std::memory_order_acquire) > PACK_IN_MEM_LIMIT){
810-
long cur = mPackProcessedCounter.load(std::memory_order_acquire);
811-
mPackProcessedCounter.wait(cur, std::memory_order_acquire);
816+
uint32_t cur = mPackProcessedCounter.load(std::memory_order_acquire);
817+
hwy::BlockUntilDifferent(cur, mPackProcessedCounter);
812818
slept++;
813819
}
814820
}
815821
readNum += count;
816822
// if the writer threads are far behind this producer, sleep and wait
817823
// check this only when necessary
818824
if(readNum % (PACK_SIZE * PACK_IN_MEM_LIMIT) == 0 && mLeftWriter) {
819-
while( (mLeftWriter && mLeftWriter->bufferLength() > PACK_IN_MEM_LIMIT) || (mRightWriter && mRightWriter->bufferLength() > PACK_IN_MEM_LIMIT) ){
820-
std::this_thread::yield();
821-
slept++;
822-
}
825+
if(mLeftWriter) mLeftWriter->waitForBufferBelow(PACK_IN_MEM_LIMIT);
826+
if(mRightWriter) mRightWriter->waitForBufferBelow(PACK_IN_MEM_LIMIT);
823827
}
824828
// reset count to 0
825829
count = 0;
@@ -900,6 +904,9 @@ void PairEndProcessor::interleavedReaderTask()
900904
mRightInputLists[mRightPackReadCounter % mOptions->thread]->produce(packRight);
901905
mRightPackReadCounter++;
902906

907+
mPackProducedCounter.fetch_add(1, std::memory_order_release);
908+
hwy::WakeAll(mPackProducedCounter);
909+
903910
dataLeft = NULL;
904911
dataRight = NULL;
905912
break;
@@ -931,25 +938,26 @@ void PairEndProcessor::interleavedReaderTask()
931938
mRightInputLists[mRightPackReadCounter % mOptions->thread]->produce(packRight);
932939
mRightPackReadCounter++;
933940

941+
mPackProducedCounter.fetch_add(1, std::memory_order_release);
942+
hwy::WakeAll(mPackProducedCounter);
943+
934944
//re-initialize data for next pack
935945
dataLeft = new Read*[PACK_SIZE];
936946
dataRight = new Read*[PACK_SIZE];
937947
memset(dataLeft, 0, sizeof(Read*)*PACK_SIZE);
938948
memset(dataRight, 0, sizeof(Read*)*PACK_SIZE);
939949
// if the consumer is far behind this producer, wait to limit memory usage
940950
while(mLeftPackReadCounter - mPackProcessedCounter.load(std::memory_order_acquire) > PACK_IN_MEM_LIMIT){
941-
long cur = mPackProcessedCounter.load(std::memory_order_acquire);
942-
mPackProcessedCounter.wait(cur, std::memory_order_acquire);
951+
uint32_t cur = mPackProcessedCounter.load(std::memory_order_acquire);
952+
hwy::BlockUntilDifferent(cur, mPackProcessedCounter);
943953
slept++;
944954
}
945955
readNum += count;
946956
// if the writer threads are far behind this producer, sleep and wait
947957
// check this only when necessary
948958
if(readNum % (PACK_SIZE * PACK_IN_MEM_LIMIT) == 0 && mLeftWriter) {
949-
while( (mLeftWriter && mLeftWriter->bufferLength() > PACK_IN_MEM_LIMIT) || (mRightWriter && mRightWriter->bufferLength() > PACK_IN_MEM_LIMIT) ){
950-
std::this_thread::yield();
951-
slept++;
952-
}
959+
if(mLeftWriter) mLeftWriter->waitForBufferBelow(PACK_IN_MEM_LIMIT);
960+
if(mRightWriter) mRightWriter->waitForBufferBelow(PACK_IN_MEM_LIMIT);
953961
}
954962
// reset count to 0
955963
count = 0;
@@ -1009,7 +1017,8 @@ void PairEndProcessor::processorTask(ThreadConfig* config)
10091017
} else if(inputRight->isProducerFinished() && !inputRight->canBeConsumed()) {
10101018
break;
10111019
} else {
1012-
std::this_thread::yield();
1020+
uint32_t cur = mPackProducedCounter.load(std::memory_order_acquire);
1021+
hwy::BlockUntilDifferent(cur, mPackProducedCounter);
10131022
}
10141023
}
10151024
inputLeft->setConsumerFinished();

‎src/peprocessor.h‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,8 @@ class PairEndProcessor{
6464
SingleProducerSingleConsumerList<ReadPack*>** mRightInputLists;
6565
size_t mLeftPackReadCounter;
6666
size_t mRightPackReadCounter;
67-
alignas(128) atomic_long mPackProcessedCounter;
67+
alignas(128) std::atomic<uint32_t> mPackProcessedCounter;
68+
alignas(128) std::atomic<uint32_t> mPackProducedCounter;
6869
ReadPool* mLeftReadPool;
6970
ReadPool* mRightReadPool;
7071
atomic_bool shouldStopReading;

‎src/seprocessor.cpp‎

Lines changed: 12 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
#include "seprocessor.h"
2+
#include "hwy/contrib/thread_pool/futex.h"
23
#include <print>
34
#include "fastqreader.h"
45
#include <iostream>
@@ -29,6 +30,7 @@ SingleEndProcessor::SingleEndProcessor(Options* opt)
2930

3031
mPackReadCounter = 0;
3132
mPackProcessedCounter = 0;
33+
mPackProducedCounter = 0;
3234

3335
mReadPool = new ReadPool(mOptions);
3436
}
@@ -317,7 +319,7 @@ bool SingleEndProcessor::processSingleEnd(ReadPack* pack, ThreadConfig* config){
317319
delete pack;
318320

319321
mPackProcessedCounter.fetch_add(1, std::memory_order_release);
320-
mPackProcessedCounter.notify_all();
322+
hwy::WakeAll(mPackProcessedCounter);
321323

322324
return true;
323325
}
@@ -347,6 +349,8 @@ void SingleEndProcessor::readerTask()
347349
pack->count = count;
348350
mInputLists[mPackReadCounter % mOptions->thread]->produce(pack);
349351
mPackReadCounter++;
352+
mPackProducedCounter.fetch_add(1, std::memory_order_release);
353+
hwy::WakeAll(mPackProducedCounter);
350354
data = NULL;
351355
if(read) {
352356
delete read;
@@ -372,23 +376,22 @@ void SingleEndProcessor::readerTask()
372376
pack->count = count;
373377
mInputLists[mPackReadCounter % mOptions->thread]->produce(pack);
374378
mPackReadCounter++;
379+
mPackProducedCounter.fetch_add(1, std::memory_order_release);
380+
hwy::WakeAll(mPackProducedCounter);
375381
//re-initialize data for next pack
376382
data = new Read*[PACK_SIZE];
377383
memset(data, 0, sizeof(Read*)*PACK_SIZE);
378384
// if the processor is far behind this reader, wait to limit memory usage
379385
while(mPackReadCounter - mPackProcessedCounter.load(std::memory_order_acquire) > PACK_IN_MEM_LIMIT){
380-
long cur = mPackProcessedCounter.load(std::memory_order_acquire);
381-
mPackProcessedCounter.wait(cur, std::memory_order_acquire);
386+
uint32_t cur = mPackProcessedCounter.load(std::memory_order_acquire);
387+
hwy::BlockUntilDifferent(cur, mPackProcessedCounter);
382388
slept++;
383389
}
384390
readNum += count;
385391
// if the writer threads are far behind this reader, sleep and wait
386392
// check this only when necessary
387393
if(readNum % (PACK_SIZE * PACK_IN_MEM_LIMIT) == 0 && mLeftWriter) {
388-
while(mLeftWriter->bufferLength() > PACK_IN_MEM_LIMIT) {
389-
std::this_thread::yield();
390-
slept++;
391-
}
394+
mLeftWriter->waitForBufferBelow(PACK_IN_MEM_LIMIT);
392395
}
393396
// reset count to 0
394397
count = 0;
@@ -444,7 +447,8 @@ void SingleEndProcessor::processorTask(ThreadConfig* config)
444447
break;
445448
}
446449
} else {
447-
std::this_thread::yield();
450+
uint32_t cur = mPackProducedCounter.load(std::memory_order_acquire);
451+
hwy::BlockUntilDifferent(cur, mPackProducedCounter);
448452
}
449453
}
450454
input->setConsumerFinished();

‎src/seprocessor.h‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,8 @@ class SingleEndProcessor{
5050
Duplicate* mDuplicate;
5151
SingleProducerSingleConsumerList<ReadPack*>** mInputLists;
5252
size_t mPackReadCounter;
53-
alignas(128) atomic_long mPackProcessedCounter;
53+
alignas(128) std::atomic<uint32_t> mPackProcessedCounter;
54+
alignas(128) std::atomic<uint32_t> mPackProducedCounter;
5455
ReadPool* mReadPool;
5556
};
5657

‎src/writerthread.cpp‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -99,12 +99,14 @@ void WriterThread::output(){
9999
if (mPwriteMode) return; // no-op
100100
SingleProducerSingleConsumerList<string*>* list = mBufferLists[mWorkingBufferList];
101101
if(!list->canBeConsumed()) {
102-
std::this_thread::yield();
102+
uint32_t cur = mBufferLength.load(std::memory_order_acquire);
103+
if(cur == 0) hwy::BlockUntilDifferent(cur, mBufferLength);
103104
} else {
104105
string* str = list->consume();
105106
mWriter1->write(str->data(), str->length());
106107
delete str;
107-
mBufferLength--;
108+
mBufferLength.fetch_sub(1, std::memory_order_release);
109+
hwy::WakeAll(mBufferLength);
108110
mWorkingBufferList = (mWorkingBufferList+1)%mOptions->thread;
109111
}
110112
}
@@ -115,7 +117,8 @@ void WriterThread::input(int tid, string* data) {
115117
return;
116118
}
117119
mBufferLists[tid]->produce(data);
118-
mBufferLength++;
120+
mBufferLength.fetch_add(1, std::memory_order_release);
121+
hwy::WakeAll(mBufferLength);
119122
}
120123

121124
void WriterThread::inputPwrite(int tid, string* data) {

‎src/writerthread.h‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
#include "writer.h"
99
#include "options.h"
1010
#include <atomic>
11+
#include "hwy/contrib/thread_pool/futex.h"
1112
#include <mutex>
1213
#include <libdeflate.h>
1314
#include "singleproducersingleconsumerlist.h"
@@ -36,7 +37,14 @@ class WriterThread{
3637
void input(int tid, string* data);
3738
bool setInputCompleted();
3839

39-
long bufferLength() {return mBufferLength;};
40+
uint32_t bufferLength() {return mBufferLength;};
41+
void waitForBufferBelow(uint32_t limit) {
42+
for(;;) {
43+
uint32_t cur = mBufferLength.load(std::memory_order_acquire);
44+
if(cur <= limit) break;
45+
hwy::BlockUntilDifferent(cur, mBufferLength);
46+
}
47+
}
4048
string getFilename() {return mFilename;}
4149
bool isPwriteMode() {return mPwriteMode;}
4250

@@ -51,7 +59,7 @@ class WriterThread{
5159
string mFilename;
5260

5361
bool mInputCompleted;
54-
atomic_long mBufferLength;
62+
std::atomic<uint32_t> mBufferLength;
5563
SingleProducerSingleConsumerList<string*>** mBufferLists;
5664
int mWorkingBufferList;
5765

0 commit comments

Comments
 (0)