From 796b7f9b758f3b8b1a11a0dea0c4b7d0da9efc5f Mon Sep 17 00:00:00 2001 From: kautism Date: Sat, 26 Sep 2026 15:42:35 +0800 Subject: [PATCH 1/6] fix-lockfree-correctness --- .github/workflows/ci.yml | 54 ++- Makefile | 120 ++++-- README.md | 253 ++++++------- Visual Stdio/example.vcxproj | 12 +- Visual Stdio/lfqueue.vcxproj | 12 +- Visual Stdio/test.vcxproj | 12 +- cross-platform.h | 219 ++++++----- example.c | 58 +-- lfq.c | 689 ++++++++++++++++++++++------------- lfq.h | 119 +++--- test_aba.c | 262 +++++++------ test_edge_cases.c | 57 +++ test_multithread.c | 356 +++++++++++------- 13 files changed, 1335 insertions(+), 888 deletions(-) create mode 100644 test_edge_cases.c diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6e86fc2..d4c0920 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -6,16 +6,54 @@ on: pull_request: branches: [ "HP" ] +permissions: + contents: read + jobs: - build: + linux: + name: Linux (${{ matrix.cc }}) + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + cc: [gcc, clang] + env: + CC: ${{ matrix.cc }} + steps: + - uses: actions/checkout@v6 + - name: Build and test + run: make clean test + sanitizers: + name: Sanitizers runs-on: ubuntu-latest - permissions: - contents: read + env: + CC: clang + steps: + - uses: actions/checkout@v6 + - name: Address + UndefinedBehavior Sanitizer + run: make clean test-sanitize + - name: ThreadSanitizer + run: make clean test-tsan + windows: + name: Windows MSVC + runs-on: windows-latest steps: - - uses: actions/checkout@v4 - - name: make - run: make - - name: make test - run: make test + - uses: actions/checkout@v6 + - uses: ilammy/msvc-dev-cmd@v1 + with: + arch: x64 + - name: Build + shell: cmd + run: | + if not exist bin mkdir bin + cl /nologo /W4 /WX /O2 /TC lfq.c test_edge_cases.c /Fe:bin\test_edge_cases.exe + cl /nologo /W4 /WX /O2 /TC lfq.c test_multithread.c /DMAX_PRODUCER=4 /DMAX_CONSUMER=4 /DITEMS_PER_PRODUCER=5000 /Fe:bin\test_multithread.exe + cl /nologo /W4 /WX /O2 /TC lfq.c test_aba.c /DABA_THREADS=8 /DABA_ITERATIONS=5000 /Fe:bin\test_aba.exe + - name: Test + shell: cmd + run: | + bin\test_edge_cases.exe + bin\test_multithread.exe + bin\test_aba.exe diff --git a/Makefile b/Makefile index ddef0e7..f40fc5d 100644 --- a/Makefile +++ b/Makefile @@ -1,40 +1,80 @@ -CC=gcc -CFLAGS=-std=gnu99 -O3 -Wall -Wextra -g -LDFLAGS=-g -LOADLIBS=-lpthread - -all : bin/test_p1c1 bin/test_p4c4 bin/test_p100c10 bin/test_p10c100 bin/test_aba bin/example - -bin/example: example.c liblfq.so.1.0.0 - gcc $(CFLAGS) $(LDFLAGS) example.c lfq.c -o bin/example -lpthread - -bin/test_p1c1: liblfq.so.1.0.0 test_multithread.c - gcc $(CFLAGS) $(LDFLAGS) test_multithread.c -o bin/test_p1c1 -L. -Wl,-Bstatic -llfq -Wl,-Bdynamic -lpthread -D MAX_PRODUCER=1 -D MAX_CONSUMER=1 - -bin/test_p4c4: liblfq.so.1.0.0 test_multithread.c - gcc $(CFLAGS) $(LDFLAGS) test_multithread.c -o bin/test_p4c4 -L. -Wl,-Bstatic -llfq -Wl,-Bdynamic -lpthread -D MAX_PRODUCER=4 -D MAX_CONSUMER=4 - -bin/test_p100c10: liblfq.so.1.0.0 test_multithread.c - gcc $(CFLAGS) $(LDFLAGS) test_multithread.c -o bin/test_p100c10 -L. -Wl,-Bstatic -llfq -Wl,-Bdynamic -lpthread -D MAX_PRODUCER=100 -D MAX_CONSUMER=10 - -bin/test_p10c100: liblfq.so.1.0.0 test_multithread.c - gcc $(CFLAGS) $(LDFLAGS) test_multithread.c -o bin/test_p10c100 -L. -Wl,-Bstatic -llfq -Wl,-Bdynamic -lpthread -D MAX_PRODUCER=10 -D MAX_CONSUMER=100 - -bin/test_aba: liblfq.so.1.0.0 test_aba.c - gcc $(CFLAGS) $(LDFLAGS) test_aba.c -o bin/test_aba -L. -Wl,-Bstatic -llfq -Wl,-Bdynamic -lpthread - -liblfq.so.1.0.0: lfq.c lfq.h cross-platform.h - gcc $(CFLAGS) $(CPPFLAGS) -c lfq.c # -fno-pie for static linking? - ar rcs liblfq.a lfq.o - gcc $(CFLAGS) $(CPPFLAGS) -fPIC -c lfq.c - gcc $(LDFLAGS) -shared -o liblfq.so.1.0.0 lfq.o - -test: bin/test_p1c1 bin/test_p4c4 bin/test_p100c10 bin/test_p10c100 bin/test_aba - $(TESTWRAPPER) bin/test_p1c1 - $(TESTWRAPPER) bin/test_p4c4 - $(TESTWRAPPER) bin/test_p100c10 - $(TESTWRAPPER) bin/test_p10c100 - $(TESTWRAPPER) bin/test_aba - -clean: - rm -rf *.o bin/* liblfq.so.1.0.0 liblfq.a +CC ?= gcc +CFLAGS ?= -std=gnu99 -O3 -Wall -Wextra -Wpedantic -g +CPPFLAGS ?= +LDFLAGS ?= +LDLIBS ?= -lpthread + +TEST_BINS = bin/test_edge_cases bin/test_p1c1 bin/test_p4c4 bin/test_p100c10 bin/test_p10c100 bin/test_aba + +.PHONY: all test clean test-sanitize test-tsan + +all: liblfq.a liblfq.so.1.0.0 $(TEST_BINS) bin/example + +bin: + mkdir -p bin + +lfq.o: lfq.c lfq.h cross-platform.h + $(CC) $(CPPFLAGS) $(CFLAGS) -fPIC -c lfq.c -o $@ + +liblfq.a: lfq.o + ar rcs $@ $< + +liblfq.so.1.0.0: lfq.o + $(CC) $(LDFLAGS) -shared -o $@ $< + +bin/example: example.c lfq.c lfq.h cross-platform.h | bin + $(CC) $(CPPFLAGS) $(CFLAGS) $(LDFLAGS) example.c lfq.c -o $@ $(LDLIBS) + +bin/test_edge_cases: test_edge_cases.c lfq.c lfq.h cross-platform.h | bin + $(CC) $(CPPFLAGS) $(CFLAGS) $(LDFLAGS) test_edge_cases.c lfq.c -o $@ $(LDLIBS) + +bin/test_p1c1: test_multithread.c lfq.c lfq.h cross-platform.h | bin + $(CC) $(CPPFLAGS) $(CFLAGS) $(LDFLAGS) test_multithread.c lfq.c -o $@ $(LDLIBS) -D MAX_PRODUCER=1 -D MAX_CONSUMER=1 + +bin/test_p4c4: test_multithread.c lfq.c lfq.h cross-platform.h | bin + $(CC) $(CPPFLAGS) $(CFLAGS) $(LDFLAGS) test_multithread.c lfq.c -o $@ $(LDLIBS) -D MAX_PRODUCER=4 -D MAX_CONSUMER=4 + +bin/test_p100c10: test_multithread.c lfq.c lfq.h cross-platform.h | bin + $(CC) $(CPPFLAGS) $(CFLAGS) $(LDFLAGS) test_multithread.c lfq.c -o $@ $(LDLIBS) -D MAX_PRODUCER=100 -D MAX_CONSUMER=10 -D ITEMS_PER_PRODUCER=20000 + +bin/test_p10c100: test_multithread.c lfq.c lfq.h cross-platform.h | bin + $(CC) $(CPPFLAGS) $(CFLAGS) $(LDFLAGS) test_multithread.c lfq.c -o $@ $(LDLIBS) -D MAX_PRODUCER=10 -D MAX_CONSUMER=100 -D ITEMS_PER_PRODUCER=20000 + +bin/test_aba: test_aba.c lfq.c lfq.h cross-platform.h | bin + $(CC) $(CPPFLAGS) $(CFLAGS) $(LDFLAGS) test_aba.c lfq.c -o $@ $(LDLIBS) + +test: $(TEST_BINS) + ./bin/test_edge_cases + ./bin/test_p1c1 + ./bin/test_p4c4 + ./bin/test_p100c10 + ./bin/test_p10c100 + ./bin/test_aba + +bin/test_edge_asan: test_edge_cases.c lfq.c lfq.h cross-platform.h | bin + $(CC) -std=gnu99 -O1 -g -Wall -Wextra -fno-omit-frame-pointer -fsanitize=address,undefined test_edge_cases.c lfq.c -o $@ $(LDLIBS) + +bin/test_mt_asan: test_multithread.c lfq.c lfq.h cross-platform.h | bin + $(CC) -std=gnu99 -O1 -g -Wall -Wextra -fno-omit-frame-pointer -fsanitize=address,undefined test_multithread.c lfq.c -o $@ $(LDLIBS) -D MAX_PRODUCER=8 -D MAX_CONSUMER=8 -D ITEMS_PER_PRODUCER=5000 + +bin/test_aba_asan: test_aba.c lfq.c lfq.h cross-platform.h | bin + $(CC) -std=gnu99 -O1 -g -Wall -Wextra -fno-omit-frame-pointer -fsanitize=address,undefined test_aba.c lfq.c -o $@ $(LDLIBS) -D ABA_THREADS=8 -D ABA_ITERATIONS=5000 + +test-sanitize: bin/test_edge_asan bin/test_mt_asan bin/test_aba_asan + ASAN_OPTIONS=detect_leaks=1:halt_on_error=1 ./bin/test_edge_asan + ASAN_OPTIONS=detect_leaks=1:halt_on_error=1 ./bin/test_mt_asan + ASAN_OPTIONS=detect_leaks=1:halt_on_error=1 ./bin/test_aba_asan + +bin/test_mt_tsan: test_multithread.c lfq.c lfq.h cross-platform.h | bin + $(CC) -std=gnu99 -O1 -g -Wall -Wextra -fno-omit-frame-pointer -fsanitize=thread test_multithread.c lfq.c -o $@ $(LDLIBS) -D MAX_PRODUCER=8 -D MAX_CONSUMER=8 -D ITEMS_PER_PRODUCER=5000 + +bin/test_aba_tsan: test_aba.c lfq.c lfq.h cross-platform.h | bin + $(CC) -std=gnu99 -O1 -g -Wall -Wextra -fno-omit-frame-pointer -fsanitize=thread test_aba.c lfq.c -o $@ $(LDLIBS) -D ABA_THREADS=8 -D ABA_ITERATIONS=5000 + +test-tsan: bin/test_mt_tsan bin/test_aba_tsan + TSAN_OPTIONS=halt_on_error=1 ./bin/test_mt_tsan + TSAN_OPTIONS=halt_on_error=1 ./bin/test_aba_tsan + +clean: + rm -f *.o liblfq.a liblfq.so.1.0.0 + rm -f bin/test_* bin/example diff --git a/README.md b/README.md index 26aa21c..76091b5 100755 --- a/README.md +++ b/README.md @@ -1,204 +1,159 @@ -# lfqueue [![Build Status](https://travis-ci.org/darkautism/lfqueue.svg?branch=HP)](https://travis-ci.org/darkautism/lfqueue) +# lfqueue -Minimize lock-free queue, it's easy to use and easy to read. It's only 150 line code so it is easy to understand, best code for education ever! +[![C/C++ CI](https://github.com/darkautism/lfqueue/actions/workflows/ci.yml/badge.svg?branch=HP)](https://github.com/darkautism/lfqueue/actions/workflows/ci.yml) -***Do not use these code in production*** +A small multi-producer / multi-consumer lock-free queue written in C. -Support multiple comsumer and multiple producer at sametime. +The default branch is the Hazard Pointer implementation. The code is intentionally +kept compact enough to study, but concurrent memory reclamation is subtle: run the +sanitizer tests before using changes in real systems. -| Arch | Build status | Test | -| ----------------- | ------------ | ------------ | -| Linux | [![Build Status](https://travis-ci.org/darkautism/lfqueue.svg?branch=HP)](https://travis-ci.org/darkautism/lfqueue)| On testing | -| Windows(msbuild) | [![Build status](https://ci.appveyor.com/api/projects/status/yu04l5atf0j259kd?svg=true)](https://ci.appveyor.com/project/darkautism/lfqueue-2puqk) | Not tested [Issue](#free-memory-very-slow-in-visual-studio) | -| Windows(Mingw) | [![Build status](https://ci.appveyor.com/api/projects/status/4nng8ye801ycyvgn/branch/HP?svg=true)](https://ci.appveyor.com/project/darkautism/lfqueue/branch/HP) | Not tested | -| Windows(MinGW64) | [![Build status](https://ci.appveyor.com/api/projects/status/4457r35yh2x4f52d/branch/HP?svg=true)](https://ci.appveyor.com/project/darkautism/lfqueue-4jybw/branch/HP) | Not tested | -| Windows(Cygwin) | [![Build status](https://ci.appveyor.com/api/projects/status/xb9oww8jtbaxa9so/branch/HP?svg=true)](https://ci.appveyor.com/project/darkautism/lfqueue-7hmwx/branch/HP) | Not tested | -| Windows(Cygwin64) | [![Build status](https://ci.appveyor.com/api/projects/status/qjltyv4j963s86xd/branch/HP?svg=true)](https://ci.appveyor.com/project/darkautism/lfqueue-wepul/branch/HP) | Not tested | +## Current design -## Build Guide +The queue uses the Michael-Scott linked-queue protocol: -In any gnu toolchain, just type `make`. +- enqueue linearizes when it CAS-links the new node into `tail->next`; +- lagging `tail` pointers are helped forward by producers and consumers; +- dequeue protects both the current dummy head and its successor with two Hazard + Pointers before dereferencing them; +- removed dummy nodes go to a separate retired list. The retired-list link is + **not** overlaid on the queue's `next` field, so a hazard-protected node remains + immutable until no reader can reference it; +- all shared queue pointers, hazard slots, thread-slot ownership, and lifecycle + state are accessed through full atomic operations on GCC/Clang and MSVC. -In any visual studio build toolchain, just type `msbuild "Visual Stdio\lfqueue.sln" /verbosity:minimal /property:Configuration=Release /property:Platform=x86` or 64bit version `msbuild "Visual Stdio\lfqueue.sln" /verbosity:minimal /property:Configuration=Release /property:Platform=x64`. +`lfq_clean()` uses an atomic lifecycle gate. It succeeds only while no queue +operation is active; once cleanup starts, new operations fail with `-EBUSY` +instead of racing with reclamation. +## Build and test -## How to use +GNU/Clang: -Just copy past everywhere and use it. If you copy these code into your project. - -## Next Milestone - -- Compile on MACOS ( **I do not have MAC, need somebody help!!** ) -- Compile in kernel module. -- Use lock-free memory manager. (free is very slow in windows) - -## Example - -### Sample example +```sh +make +make test +make test-sanitize +make test-tsan +``` -It is an minimize sample code for how to using lfq. +CI runs GCC, Clang, AddressSanitizer + UndefinedBehaviorSanitizer, +ThreadSanitizer, and MSVC tests. -**Even if int or long value is valid input data, but you will hard to distinguish empty queue or other error message.** +Visual Studio projects are also included under `Visual Stdio/`. -**We not suggestion use int or long as queue data type** +## Basic example -``` c +```c +#include #include -#include #include "lfq.h" -int main() { - long ret; - struct lfq_ctx ctx; - lfq_init(&ctx, 0); - lfq_enqueue(&ctx,(void *)1); - lfq_enqueue(&ctx,(void *)3); - lfq_enqueue(&ctx,(void *)5); - lfq_enqueue(&ctx,(void *)8); - lfq_enqueue(&ctx,(void *)4); - lfq_enqueue(&ctx,(void *)6); - - while ( (ret = (long)lfq_dequeue(&ctx)) != 0 ) - printf("lfq_dequeue %ld\n", ret); - - lfq_clean(&ctx); - return 0; -} -``` +int main(void) { + struct lfq_ctx ctx; -### Advance example + if (lfq_init(&ctx, 0) != 0) + return 1; /* 0 => default 16 consumer slots */ -If you want to get best performance with the cost of developing speed, you can control thread_id yourself. + lfq_enqueue(&ctx, (void *)(uintptr_t)1); + lfq_enqueue(&ctx, (void *)(uintptr_t)2); -**This API only impilement on HP branch** + for (;;) { + void *p = lfq_dequeue(&ctx); + if (p == NULL) + break; /* empty */ + if (p == LFQ_ERROR) + return 1; /* errno-style failure */ + printf("%lu\n", (unsigned long)(uintptr_t)p); + } -``` c -#include -#include -#include "lfq.h" - -#define MAX_CONSUMER_THREAD 4 - -int main() { - long ret; - struct lfq_ctx ctx; - lfq_init(&ctx, MAX_CONSUMER_THREAD); - lfq_enqueue(&ctx,(void *)1); - - // The second number is thread id, this thread id should unique between threads. - // And this tid must less than MAX_CONSUMER_THREAD - // In this sample code, this tid must 0, 1, 2, 3. - ret = (long)lfq_dequeue_tid(&ctx, 1); - - lfq_clean(&ctx); - return 0; + return lfq_clean(&ctx) == 0 ? 0 : 1; } ``` ## API -### lfq_init(struct lfq_ctx *ctx, int max_consume_thread) - -Init lock-free queue. - -**Arguments:** -- **ctx** : Lock-free queue handler. -- **max_consume_thread** : Max consume thread numbers. If this value set to zero, use default value (16). - -**Return:** The lfq_init() functions return zero on success. On error, this functions return negative errno. - - -### lfq_clean(struct lfq_ctx *ctx) +### `lfq_init(ctx, max_consume_thread)` -Clean lock-free queue from ctx. +Initializes a fresh context. Passing `0` selects +`LFQ_DEFAULT_MAX_CONSUMERS` (16). A negative value is rejected. -**Arguments:** -- **ctx** : Lock-free queue handler. +The caller must not call `lfq_init()` on a live queue; clean it first. -**Return:** The lfq_clean() functions return zero on success. On error, this functions return -1. +### `lfq_enqueue(ctx, data)` +Enqueues one payload pointer. -### lfq_enqueue(struct lfq_ctx *ctx, void * data) +`NULL` and `LFQ_ERROR` are rejected with `-EINVAL` because the legacy +pointer-returning dequeue API reserves those values for empty/error results. -Push data into queue. +### `lfq_try_dequeue(ctx, &out)` -**Arguments:** -- **ctx** : Lock-free queue handler. -- **data** : User data. +Recommended dequeue API: -**Return:** The lfq_clean() functions return zero on success. On error, this functions return negative errno. +- `1`: item returned in `out` +- `0`: queue empty +- negative value: errno-style error +### `lfq_try_dequeue_tid(ctx, tid, &out)` -### lfq_dequeue(struct lfq_ctx *ctx) +Same operation with an explicit consumer slot. `tid` must be in +`[0, max_consume_thread)`. -Pop data from queue. +The implementation reserves the slot atomically for each call, so manual and +automatic dequeue APIs cannot silently use the same Hazard Pointer slots at the +same time. Concurrent calls using the same explicit `tid` return `-EBUSY`. -**Arguments:** -- **ctx** : Lock-free queue handler. +### `lfq_dequeue()` / `lfq_dequeue_tid()` -**Return:** The lfq_clean() functions return zero if empty queue. Return positive pointer. On error, this functions return negative errno. +Compatibility wrappers: -### lfq_dequeue_tid(struct lfq_ctx *ctx, int tid) +- payload pointer on success +- `NULL` when empty +- `LFQ_ERROR` on error -Pop data from queue. +Prefer the status-returning APIs in new code. -**Arguments:** -- **ctx** : Lock-free queue handler. -- **tid** : Unique thread id. +### `lfq_clean(ctx)` -**Return:** The lfq_dequeue_tid() functions return zero if empty queue. Return positive pointer. On error, this functions return negative errno. - -## Issues - -### ENOMEM - -This lfqueue do not have size limit, so count your struct yourself. - -### Can i iterate lfqueue inner struct? - -No, iterate inner struct is not threadsafe. - -If you do not get segmentation fault because you are iterate lfqueue in single thread. - -We should always iterate lfqueue by `lfq_enqueue` and `lfq_dequeue` method. - -### CPU time slice waste? Other thread blocking and waiting for swap? - -**Enqueue** - -No, CAS operator not lock. Loser not block and wait winner. - -If a thread race win a CAS, other thread will get false and try to retrive next pointer. Because winner already swap tail, so other losers can do next race. - -**Example:** -``` -4 thread race push -1 win Push A, 2 lose, 1 do not have CPU time slice -1 win go out queue, 2 losers race A as tail, 1 race initnode as tail -1 win go out queue, 1 win push B, 1 lose, 1 race A failed because tail not initnode now -``` +Reclaims queue nodes and retired nodes. User payloads are never freed by the +queue. -So lock-free queue have better performance then lock queue. +Cleanup may be attempted concurrently, but it returns `-EBUSY` if an operation +is already active. Once cleanup wins the lifecycle gate, new operations are +blocked until cleanup finishes. -**Dequeue** +## Correctness notes -We choice Hazard Pointers to reaolve ABA problems. So dequeue is fast as enqueue. +Older revisions had several important races: -There has many papers to resolve this problem: +- enqueue published `tail` before linking `old_tail->next`, allowing a + completed enqueue to remain invisible to consumers; +- the retired-list pointer shared a union with the queue `next` pointer, so a + node could be modified while another consumer still hazard-protected it; +- Hazard Pointer slots were plain `volatile` loads/stores rather than atomic + accesses; +- Win64 used an 8-byte CAS on 32-bit `int` fields; +- `lfq_init(ctx, 0)` contradicted the documented default and produced zero + consumer slots; +- cleanup could leave non-empty queues partially unreclaimed. -- Lock-Free Reference Counting -- ABA-Prevention Tags -- [Hazard Pointers](https://github.com/darkautism/lfqueue/tree/HP) **Beta** - - [Algorithm Paper](http://citeseerx.ist.psu.edu/viewdoc/download?doi=10.1.1.395.378&rep=rep1&type=pdf) -- [Fucking stupid head wait](https://github.com/darkautism/lfqueue/tree/FSHW) **Stable** +The current implementation removes those mechanisms instead of adding more +fences around them. -### Free memory very slow in Visual studio +## Tests -Sorry, i have no idea why. Still finding problems in windows. +`test_edge_cases.c` +covers API errors, default initialization, FIFO behavior, slot conflicts, +cleanup of non-empty queues, and post-clean behavior. -## Contributions +`test_multithread.c` +runs MPMC producer/consumer matrices and verifies both item count and a +cross-thread checksum. -[pcordes](https://github.com/pcordes) +`test_aba.c` +stresses rapid node retirement/reuse. It is paired with sanitizer CI; a passing +stress test by itself is not a formal proof of linearizability. ## License diff --git a/Visual Stdio/example.vcxproj b/Visual Stdio/example.vcxproj index 80ee6c2..cfc9fe1 100644 --- a/Visual Stdio/example.vcxproj +++ b/Visual Stdio/example.vcxproj @@ -1,5 +1,5 @@ - + Debug @@ -22,32 +22,32 @@ {F97F305D-2A2A-4610-919C-C91CE4152C67} Win32Proj example - 8.1 + 10.0 Application true - v140 + v143 Unicode Application false - v140 + v143 true Unicode Application true - v140 + v143 Unicode Application false - v140 + v143 true Unicode diff --git a/Visual Stdio/lfqueue.vcxproj b/Visual Stdio/lfqueue.vcxproj index 3f5f8a9..0f6c631 100644 --- a/Visual Stdio/lfqueue.vcxproj +++ b/Visual Stdio/lfqueue.vcxproj @@ -1,5 +1,5 @@ - + Debug @@ -22,33 +22,33 @@ {1E11B533-4CCF-40E5-AD9D-5AAA19ABC776} Win32Proj lfqueue - 8.1 + 10.0 lfqueue StaticLibrary true - v140 + v143 Unicode StaticLibrary false - v140 + v143 true Unicode StaticLibrary true - v140 + v143 Unicode StaticLibrary false - v140 + v143 true Unicode diff --git a/Visual Stdio/test.vcxproj b/Visual Stdio/test.vcxproj index 78ebe52..1e1b953 100644 --- a/Visual Stdio/test.vcxproj +++ b/Visual Stdio/test.vcxproj @@ -1,5 +1,5 @@ - + Debug @@ -22,31 +22,31 @@ {19641A1F-792B-4792-AD1C-005FE5EAD651} Win32Proj test - 8.1 + 10.0 Application true - v140 + v143 Unicode Application false - v140 + v143 true Unicode Application true - v140 + v143 Unicode Application false - v140 + v143 true Unicode diff --git a/cross-platform.h b/cross-platform.h index 2a97882..b1809a7 100644 --- a/cross-platform.h +++ b/cross-platform.h @@ -1,102 +1,145 @@ #ifndef __CROSS_PLATFORM_H__ #define __CROSS_PLATFORM_H__ -// bool define -#ifdef __KERNEL__ - #include -#else - #include -#endif -// malloc free #ifdef __KERNEL__ - #define malloc(x) kmalloc(x, GFP_KERNEL ) - #define free kfree - #define calloc(x,y) kmalloc(x*y, GFP_KERNEL | __GFP_ZERO ) - #include +#include +#include +#define malloc(x) kmalloc((x), GFP_KERNEL) +#define free kfree +#define calloc(x,y) kmalloc((x) * (y), GFP_KERNEL | __GFP_ZERO) +#include +typedef int lfq_atomic_int_t; #else - #include - #include +#include +#include +#include +#include #endif +#if defined(_MSC_VER) -#ifndef asm - #define asm __asm -#endif +#include +typedef volatile LONG lfq_atomic_int_t; -#define cmpxchg( ptr, _old, _new ) { \ - volatile uint32_t *__ptr = (volatile uint32_t *)(ptr); \ - uint32_t __ret; \ - asm volatile( "lock; cmpxchgl %2,%1" \ - : "=a" (__ret), "+m" (*__ptr) \ - : "r" (_new), "0" (_old) \ - : "memory"); \ - ); \ - __ret; \ -} - -//#define CAS cmpxchg -#define ATOMIC_SET __sync_lock_test_and_set -#define ATOMIC_RELEASE __sync_lock_release - -#if defined __GNUC__ - #define ATOMIC_SUB __sync_sub_and_fetch - #define ATOMIC_SUB64 ATOMIC_SUB - #define CAS __sync_bool_compare_and_swap -#define XCHG __sync_lock_test_and_set // yes really. The 2nd arg is limited to 1 on machines with TAS but not XCHG. On x86 it's an arbitrary value - #define ATOMIC_ADD __sync_add_and_fetch - #define ATOMIC_ADD64 ATOMIC_ADD - #define mb __sync_synchronize -#if defined(__x86_64__) || defined(__i386) -// #define lmb() asm volatile( "lfence" ) -// #define smb() asm volatile( "sfence" ) - #define lmb() asm volatile("":::"memory") // compiler barrier only. runtime reordering already impossible on x86 - #define smb() asm volatile("":::"memory") - // "mfence" for lmb and smb makes assertion failures rarer, but doesn't eliminate, so it's just papering over the symptoms -#else - #define lmb() mb() - #define smb() mb() -#endif +static __forceinline void *lfq_atomic_load_ptr(void * volatile *ptr) { + return InterlockedCompareExchangePointer((PVOID volatile *)ptr, NULL, NULL); +} + +static __forceinline void lfq_atomic_store_ptr(void * volatile *ptr, void *value) { + (void)InterlockedExchangePointer((PVOID volatile *)ptr, value); +} + +static __forceinline void *lfq_atomic_exchange_ptr(void * volatile *ptr, void *value) { + return InterlockedExchangePointer((PVOID volatile *)ptr, value); +} + +static __forceinline bool lfq_atomic_cas_ptr(void * volatile *ptr, void **expected, void *desired) { + void *actual = InterlockedCompareExchangePointer((PVOID volatile *)ptr, desired, *expected); + if (actual == *expected) + return true; + *expected = actual; + return false; +} + +static __forceinline LONG lfq_atomic_load_int(lfq_atomic_int_t *ptr) { + return InterlockedCompareExchange((volatile LONG *)ptr, 0, 0); +} + +static __forceinline void lfq_atomic_store_int(lfq_atomic_int_t *ptr, LONG value) { + (void)InterlockedExchange((volatile LONG *)ptr, value); +} + +static __forceinline bool lfq_atomic_cas_int(lfq_atomic_int_t *ptr, LONG *expected, LONG desired) { + LONG actual = InterlockedCompareExchange((volatile LONG *)ptr, desired, *expected); + if (actual == *expected) + return true; + *expected = actual; + return false; +} + +static __forceinline LONG lfq_atomic_fetch_add_int(lfq_atomic_int_t *ptr, LONG value) { + return InterlockedExchangeAdd((volatile LONG *)ptr, value); +} + +#define LFQ_ALIGNAS(n) __declspec(align(n)) +#define mb() MemoryBarrier() +#define lmb() MemoryBarrier() +#define smb() MemoryBarrier() + +#define ATOMIC_ADD(ptr, value) (InterlockedAdd((volatile LONG *)(ptr), (LONG)(value))) +#define ATOMIC_SUB(ptr, value) (InterlockedAdd((volatile LONG *)(ptr), -(LONG)(value))) +#define ATOMIC_ADD64(ptr, value) (InterlockedAdd64((volatile LONG64 *)(ptr), (LONG64)(value))) +#define ATOMIC_SUB64(ptr, value) (InterlockedAdd64((volatile LONG64 *)(ptr), -(LONG64)(value))) + +static __forceinline void lfq_thread_wait(HANDLE handle) { + if (handle) { + WaitForSingleObject(handle, INFINITE); + CloseHandle(handle); + } +} - // thread - #include - #include - #define THREAD_WAIT(x) pthread_join(x, NULL); - #define THREAD_ID pthread_self - #define THREAD_FN void * - #define THREAD_YIELD sched_yield - #define THREAD_TOKEN pthread_t +#define THREAD_WAIT(x) lfq_thread_wait(x) +#define THREAD_ID() GetCurrentThreadId() +#define THREAD_FN DWORD WINAPI +#define THREAD_YIELD() SwitchToThread() +#define THREAD_TOKEN HANDLE #else - #include - #define ATOMIC_SUB(x,y) InterlockedExchangeAddNoFence(x, -y) - #define ATOMIC_SUB64(x,y) InterlockedExchangeAddNoFence64(x, -y) - #define ATOMIC_ADD InterlockedExchangeAddNoFence - #define ATOMIC_ADD64 InterlockedExchangeAddNoFence64 - #ifdef _WIN64 - #define mb() MemoryBarrier() - #define lmb() LoadFence() - #define smb() StoreFence() - inline bool __CAS(LONG64 volatile *x, LONG64 y, LONG64 z) { - return InterlockedCompareExchangeNoFence64(x, z, y) == y; - } - #define CAS(x,y,z) __CAS((LONG64 volatile *)x, (LONG64)y, (LONG64)z) - #else - #define mb() asm mfence - #define lmb() asm lfence - #define smb() asm sfence - inline bool __CAS(LONG volatile *x, LONG y, LONG z) { - return InterlockedCompareExchangeNoFence(x, z, y) == y; - } - #define CAS(x,y,z) __CAS((LONG volatile *)x, (LONG)y, (LONG)z) - #endif - - // thread - #include - #define THREAD_WAIT(x) WaitForSingleObject(x, INFINITE); - #define THREAD_ID GetCurrentThreadId - #define THREAD_FN WORD WINAPI - #define THREAD_YIELD SwitchToThread - #define THREAD_TOKEN HANDLE -#endif + +#include +#include +typedef volatile int lfq_atomic_int_t; + +static inline void *lfq_atomic_load_ptr(void * volatile *ptr) { + return __atomic_load_n(ptr, __ATOMIC_SEQ_CST); +} + +static inline void lfq_atomic_store_ptr(void * volatile *ptr, void *value) { + __atomic_store_n(ptr, value, __ATOMIC_SEQ_CST); +} + +static inline void *lfq_atomic_exchange_ptr(void * volatile *ptr, void *value) { + return __atomic_exchange_n(ptr, value, __ATOMIC_SEQ_CST); +} + +static inline bool lfq_atomic_cas_ptr(void * volatile *ptr, void **expected, void *desired) { + return __atomic_compare_exchange_n(ptr, expected, desired, false, + __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST); +} + +static inline int lfq_atomic_load_int(lfq_atomic_int_t *ptr) { + return __atomic_load_n(ptr, __ATOMIC_SEQ_CST); +} + +static inline void lfq_atomic_store_int(lfq_atomic_int_t *ptr, int value) { + __atomic_store_n(ptr, value, __ATOMIC_SEQ_CST); +} + +static inline bool lfq_atomic_cas_int(lfq_atomic_int_t *ptr, int *expected, int desired) { + return __atomic_compare_exchange_n(ptr, expected, desired, false, + __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST); +} + +static inline int lfq_atomic_fetch_add_int(lfq_atomic_int_t *ptr, int value) { + return __atomic_fetch_add(ptr, value, __ATOMIC_SEQ_CST); +} + +#define LFQ_ALIGNAS(n) __attribute__((aligned(n))) +#define mb() __atomic_thread_fence(__ATOMIC_SEQ_CST) +#define lmb() __atomic_thread_fence(__ATOMIC_ACQUIRE) +#define smb() __atomic_thread_fence(__ATOMIC_RELEASE) + +#define ATOMIC_ADD(ptr, value) __sync_add_and_fetch((ptr), (value)) +#define ATOMIC_SUB(ptr, value) __sync_sub_and_fetch((ptr), (value)) +#define ATOMIC_ADD64(ptr, value) __sync_add_and_fetch((ptr), (value)) +#define ATOMIC_SUB64(ptr, value) __sync_sub_and_fetch((ptr), (value)) + +#define THREAD_WAIT(x) pthread_join((x), NULL) +#define THREAD_ID() ((unsigned long)pthread_self()) +#define THREAD_FN void * +#define THREAD_YIELD() sched_yield() +#define THREAD_TOKEN pthread_t + #endif +#endif diff --git a/example.c b/example.c index f6129e8..56c8e6a 100644 --- a/example.c +++ b/example.c @@ -1,21 +1,37 @@ -#include -#include -#include "lfq.h" - -int main() { - long ret; - struct lfq_ctx ctx; - lfq_init(&ctx, 1); - lfq_enqueue(&ctx,(void *)1); - lfq_enqueue(&ctx,(void *)3); - lfq_enqueue(&ctx,(void *)5); - lfq_enqueue(&ctx,(void *)8); - lfq_enqueue(&ctx,(void *)4); - lfq_enqueue(&ctx,(void *)6); - - while ( (ret = (long)lfq_dequeue(&ctx)) != 0 ) - printf("lfq_dequeue %ld\n", ret); - - lfq_clean(&ctx); - return 0; -} \ No newline at end of file +#include +#include +#include + +#include "lfq.h" + +int main(void) { + struct lfq_ctx ctx; + + if (lfq_init(&ctx, 0) != 0) { + fprintf(stderr, "lfq_init failed\n"); + return 1; + } + + const uintptr_t values[] = {1, 3, 5, 8, 4, 6}; + for (size_t i = 0; i < sizeof(values) / sizeof(values[0]); ++i) { + if (lfq_enqueue(&ctx, (void *)values[i]) != 0) { + fprintf(stderr, "lfq_enqueue failed\n"); + lfq_clean(&ctx); + return 1; + } + } + + for (;;) { + void *item = lfq_dequeue(&ctx); + if (item == NULL) + break; + if (item == LFQ_ERROR) { + fprintf(stderr, "lfq_dequeue failed\n"); + lfq_clean(&ctx); + return 1; + } + printf("lfq_dequeue %" PRIuPTR "\n", (uintptr_t)item); + } + + return lfq_clean(&ctx) == 0 ? 0 : 1; +} diff --git a/lfq.c b/lfq.c index 5a6e728..e52f6df 100755 --- a/lfq.c +++ b/lfq.c @@ -1,244 +1,445 @@ -#include "cross-platform.h" -#include "lfq.h" -#ifdef DEBUG -#include -#endif -#include -#define MAXFREE 150 - -static -int inHP(struct lfq_ctx *ctx, struct lfq_node * lfn) { - for ( int i = 0 ; i < ctx->MAXHPSIZE ; i++ ) { - //lmb(); // not needed, we don't care if loads reorder here, just that we check all the elements - if (ctx->HP[i] == lfn) - return 1; - } - return 0; -} - -static -void enpool(struct lfq_ctx *ctx, struct lfq_node * lfn) { - // add to tail of the free list - lfn->free_next = NULL; - volatile struct lfq_node *old_tail = XCHG(&ctx->fpt, lfn); // seq_cst - old_tail->free_next = lfn; - - // getting nodes out of this will have exactly the same deallocation problem - // as the main queue. - // TODO: a stack might be easier to manage, but would increase contention. - -/* - volatile struct lfq_node * p; - do { - p = ctx->fpt; - } while(!CAS(&ctx->fpt, p, lfn)); // exchange using CAS - p->free_next = lfn; -*/ -} - -static -void free_pool(struct lfq_ctx *ctx, bool freeall ) { - if (!CAS(&ctx->is_freeing, 0, 1)) - return; // this pool free is not support multithreading. - volatile struct lfq_node * p; - - for ( int i = 0 ; i < MAXFREE || freeall ; i++ ) { - p = ctx->fph; - if ( (!p->can_free) || (!p->free_next) || inHP(ctx, (struct lfq_node *)p) ) - goto exit; - ctx->fph = p->free_next; - free((void *)p); - } -exit: - ctx->is_freeing = false; - smb(); -} - -static -void safe_free(struct lfq_ctx *ctx, struct lfq_node * lfn) { - if (lfn->can_free && !inHP(ctx,lfn)) { - // free is not thread-safe - if (CAS(&ctx->is_freeing, 0, 1)) { - lfn->next = (void*)-1; // poison the pointer to detect use-after-free - free(lfn); // we got the lock; actually free - ctx->is_freeing = false; - smb(); - } else // we didn't get the lock; only add to a freelist - enpool(ctx, lfn); - } else - enpool(ctx, lfn); - free_pool(ctx, false); -} - -static -int alloc_tid(struct lfq_ctx *ctx) { - for (int i = 0; i < ctx->MAXHPSIZE; i++) - if (ctx->tid_map[i] == 0) - if (CAS(&ctx->tid_map[i], 0, 1)) - return i; - - return -1; -} - -static -void free_tid(struct lfq_ctx *ctx, int tid) { - ctx->tid_map[tid]=0; -} - -int lfq_init(struct lfq_ctx *ctx, int max_consume_thread) { - struct lfq_node * tmpnode = calloc(1,sizeof(struct lfq_node)); - if (!tmpnode) - return -errno; - - struct lfq_node * free_pool_node = calloc(1,sizeof(struct lfq_node)); - if (!free_pool_node) { - free(tmpnode); - return -errno; - } - - tmpnode->can_free = free_pool_node->can_free = true; - memset(ctx, 0, sizeof(struct lfq_ctx)); - ctx->MAXHPSIZE = max_consume_thread; - ctx->HP = calloc(max_consume_thread,sizeof(struct lfq_node *)); - ctx->tid_map = calloc(max_consume_thread,sizeof(int)); - ctx->head = ctx->tail=tmpnode; - ctx->fph = ctx->fpt=free_pool_node; - - return 0; -} - - -long lfg_count_freelist(const struct lfq_ctx *ctx) { - long count=0; - struct lfq_node *p = (struct lfq_node *)ctx->fph; // non-volatile - while(p) { - count++; - p = p->free_next; - } - - return count; -} - -int lfq_clean(struct lfq_ctx *ctx){ - if ( ctx->tail && ctx->head ) { // if have data in queue - struct lfq_node *tmp; - while ( (struct lfq_node *) ctx->head ) { // while still have node - tmp = (struct lfq_node *) ctx->head->next; - safe_free(ctx, (struct lfq_node *)ctx->head); - ctx->head = tmp; - } - ctx->tail = 0; - } - if ( ctx->fph && ctx->fpt ) { - free_pool(ctx, true); - if ( ctx->fph != ctx->fpt ) - return -1; - free((void *)ctx->fpt); // free the empty node - ctx->fph=ctx->fpt=0; - } - if ( !ctx->fph && !ctx->fpt ) { - free((void *)ctx->HP); - free((void *)ctx->tid_map); - memset(ctx,0,sizeof(struct lfq_ctx)); - } else - return -1; - - return 0; -} - -int lfq_enqueue(struct lfq_ctx *ctx, void * data) { - struct lfq_node * insert_node = calloc(1,sizeof(struct lfq_node)); - if (!insert_node) - return -errno; - insert_node->data=data; -// mb(); // we've only written to "private" memory that other threads can't see. - volatile struct lfq_node *old_tail; -#if 0 - do { - old_tail = (struct lfq_node *) ctx->tail; - } while(!CAS(&ctx->tail,old_tail,insert_node)); -#else - old_tail = XCHG(&ctx->tail, insert_node); -#endif - // We've claimed our spot in the insertion order by modifying tail - // we are the only inserting thread with a pointer to the old tail. - - // now we can make it part of the list by overwriting the NULL pointer in the old tail - // This is safe whether or not other threads have updated ->next in our insert_node -#ifdef DEBUG - assert(!(old_tail->next) && "old tail wasn't NULL"); -#endif - old_tail->next = insert_node; - // TODO: could a consumer thread could have freed the old tail? no because that would leave head=NULL - -// ATOMIC_ADD( &ctx->count, 1); - return 0; -} - -void * lfq_dequeue_tid(struct lfq_ctx *ctx, int tid ) { - //int cn_runtimes = 0; - volatile struct lfq_node *old_head, *new_head; -#if 1 // HP[tid] stuff is necessary for deallocation. (but it's still not safe). - do { - retry: // continue jumps to the bottom of the loop, and would attempt a CAS with uninitialized new_head - old_head = ctx->head; - ctx->HP[tid] = old_head; // seq-cst store. (better: use xchg instead of mov + mfence on x86) - mb(); - - if (old_head != ctx->head) // another thread freed it before seeing our HP[tid] store - goto retry; - new_head = old_head->next; // FIXME: crash with old_head=NULL during deallocation (tid=5)? (main thread=25486, this=25489) - if (new_head==0 /* || new_head != old_head->next*/ ){ // redoing the same load isn't useful - ctx->HP[tid] = 0; - return 0; // never remove the last node - } -#ifdef DEBUG - assert(new_head != (void*)-1 && "read an already-freed node"); -#endif - } while( ! CAS(&ctx->head, old_head, new_head) ); -#else // without HP[] stuff - do { - old_head = ctx->head; - //ctx->HP[tid] = old_head; - new_head = old_head->next; - //if (old_head != ctx->head) continue; - if (!new_head) { - // ctx->HP[tid] = 0; - return 0; // never remove the last node - } -#ifdef DEBUG - assert(new_head != (void*)-1 && "read an already-freed node"); -#endif - } while( !CAS(&ctx->head, old_head, new_head) ); -#endif -// mb(); // CAS is already a memory barrier, at least on x86. - - // we've atomically advanced head, and we're the thread that won the race to claim a node - // We return the data from the *new* head. - // The list starts off with a dummy node, so the current head is always a node that's already been read. - - ctx->HP[tid] = 0; - void *ret = new_head->data; - new_head->can_free = true; -// ATOMIC_SUB( &ctx->count, 1 ); - - //old_head->next = (void*)-1; // done in safe-free in the actual free() path. poison the pointer to detect use-after-free - - // we need to avoid freeing until other readers are definitely not going to load its ->next in the CAS loop - safe_free(ctx, (struct lfq_node *)old_head); - - //free(old_head); - return ret; -} - -void * lfq_dequeue(struct lfq_ctx *ctx ) { - //return lfq_dequeue_tid(ctx, 0); // TODO: let this inline even in the shared library -// old version - int tid = alloc_tid(ctx); - if (tid==-1) - return (void *)-1; // To many thread race - - void * ret = lfq_dequeue_tid(ctx, tid); - free_tid(ctx, tid); - return ret; -} +#include "cross-platform.h" +#include "lfq.h" + +#include +#include + +#define LFQ_RECLAIM_MIN 32 +#define LFQ_OP_CLOSING 0x40000000 + +static struct lfq_node *atomic_load_node(struct lfq_node * volatile *ptr) { + return (struct lfq_node *)lfq_atomic_load_ptr((void * volatile *)ptr); +} + +static void atomic_store_node(struct lfq_node * volatile *ptr, struct lfq_node *value) { + lfq_atomic_store_ptr((void * volatile *)ptr, value); +} + +static struct lfq_node *atomic_exchange_node(struct lfq_node * volatile *ptr, + struct lfq_node *value) { + return (struct lfq_node *)lfq_atomic_exchange_ptr((void * volatile *)ptr, value); +} + +static bool atomic_cas_node(struct lfq_node * volatile *ptr, + struct lfq_node **expected, + struct lfq_node *desired) { + void *raw_expected = *expected; + bool ok = lfq_atomic_cas_ptr((void * volatile *)ptr, &raw_expected, desired); + *expected = (struct lfq_node *)raw_expected; + return ok; +} + +static int begin_op(struct lfq_ctx *ctx) { + if (!ctx) + return -EINVAL; + + for (;;) { + int state = lfq_atomic_load_int(&ctx->op_state); + if (state & LFQ_OP_CLOSING) + return -EBUSY; + if ((state & ~LFQ_OP_CLOSING) == LFQ_OP_CLOSING - 1) + return -EOVERFLOW; + + int expected = state; + if (lfq_atomic_cas_int(&ctx->op_state, &expected, state + 1)) + break; + } + + if (!ctx->HP || !ctx->tid_map || !atomic_load_node(&ctx->head)) { + (void)lfq_atomic_fetch_add_int(&ctx->op_state, -1); + return -EINVAL; + } + return 0; +} + +static void end_op(struct lfq_ctx *ctx) { + (void)lfq_atomic_fetch_add_int(&ctx->op_state, -1); +} + +static size_t hazard_slot_count(const struct lfq_ctx *ctx) { + return (size_t)ctx->MAXHPSIZE * LFQ_HAZARDS_PER_THREAD; +} + +static void hp_store(struct lfq_ctx *ctx, int tid, int slot, struct lfq_node *node) { + size_t index = (size_t)tid * LFQ_HAZARDS_PER_THREAD + (size_t)slot; + lfq_atomic_store_ptr((void * volatile *)&ctx->HP[index], node); +} + +static struct lfq_node *hp_load(const struct lfq_ctx *ctx, size_t index) { + return (struct lfq_node *)lfq_atomic_load_ptr((void * volatile *)&ctx->HP[index]); +} + +static void hp_clear_pair(struct lfq_ctx *ctx, int tid) { + hp_store(ctx, tid, 0, NULL); + hp_store(ctx, tid, 1, NULL); +} + +static bool node_is_hazard(const struct lfq_ctx *ctx, const struct lfq_node *node) { + size_t count = hazard_slot_count(ctx); + for (size_t i = 0; i < count; ++i) { + if (hp_load(ctx, i) == node) + return true; + } + return false; +} + +static void retired_push(struct lfq_ctx *ctx, struct lfq_node *node) { + /* + * Count before publication so a concurrent reclaimer never frees a node + * that has not yet been included in retired_count. + */ + (void)lfq_atomic_fetch_add_int(&ctx->retired_count, 1); + + struct lfq_node *old; + do { + old = atomic_load_node(&ctx->retired_head); + node->retired_next = old; + } while (!atomic_cas_node(&ctx->retired_head, &old, node)); +} + +static void retired_requeue(struct lfq_ctx *ctx, struct lfq_node *node) { + struct lfq_node *old; + do { + old = atomic_load_node(&ctx->retired_head); + node->retired_next = old; + } while (!atomic_cas_node(&ctx->retired_head, &old, node)); +} + +static void try_reclaim(struct lfq_ctx *ctx, bool force) { + int expected = 0; + if (!lfq_atomic_cas_int(&ctx->reclaiming, &expected, 1)) + return; + + /* + * Producers dereference a local tail pointer. A producer that loaded an + * old tail before another thread advanced it does not have a hazard slot, + * so reclamation must wait until all producers have left their queue + * traversal. A producer starting after this check can only load the + * current global tail, which is never on the retired list detached below. + */ + if (!force && lfq_atomic_load_int(&ctx->active_enqueues) != 0) { + lfq_atomic_store_int(&ctx->reclaiming, 0); + return; + } + + if (!force) { + int threshold = ctx->MAXHPSIZE * LFQ_HAZARDS_PER_THREAD * 2; + if (threshold < LFQ_RECLAIM_MIN) + threshold = LFQ_RECLAIM_MIN; + + if (lfq_atomic_load_int(&ctx->retired_count) < threshold) { + lfq_atomic_store_int(&ctx->reclaiming, 0); + return; + } + } + + struct lfq_node *list = atomic_exchange_node(&ctx->retired_head, NULL); + int freed = 0; + + while (list) { + struct lfq_node *next = list->retired_next; + if (node_is_hazard(ctx, list)) { + retired_requeue(ctx, list); + } else { + free(list); + ++freed; + } + list = next; + } + + if (freed) + (void)lfq_atomic_fetch_add_int(&ctx->retired_count, -freed); + + lfq_atomic_store_int(&ctx->reclaiming, 0); +} + +static int acquire_tid(struct lfq_ctx *ctx, int requested_tid) { + if (requested_tid >= 0) { + if (requested_tid >= ctx->MAXHPSIZE) + return -EINVAL; + + int expected = 0; + if (!lfq_atomic_cas_int(&ctx->tid_map[requested_tid], &expected, 1)) + return -EBUSY; + return requested_tid; + } + + for (int i = 0; i < ctx->MAXHPSIZE; ++i) { + int expected = 0; + if (lfq_atomic_cas_int(&ctx->tid_map[i], &expected, 1)) + return i; + } + return -EAGAIN; +} + +static void release_tid(struct lfq_ctx *ctx, int tid) { + hp_clear_pair(ctx, tid); + lfq_atomic_store_int(&ctx->tid_map[tid], 0); +} + +static int dequeue_reserved(struct lfq_ctx *ctx, int tid, void **out) { + for (;;) { + struct lfq_node *head = atomic_load_node(&ctx->head); + if (!head) + return -EINVAL; + + hp_store(ctx, tid, 0, head); + if (head != atomic_load_node(&ctx->head)) + continue; + + struct lfq_node *tail = atomic_load_node(&ctx->tail); + struct lfq_node *next = atomic_load_node(&head->next); + + if (!next) { + hp_clear_pair(ctx, tid); + *out = NULL; + return 0; + } + + hp_store(ctx, tid, 1, next); + + if (head != atomic_load_node(&ctx->head) || + next != atomic_load_node(&head->next)) { + hp_clear_pair(ctx, tid); + continue; + } + + if (head == tail) { + struct lfq_node *expected_tail = tail; + (void)atomic_cas_node(&ctx->tail, &expected_tail, next); + hp_clear_pair(ctx, tid); + continue; + } + + void *value = next->data; + struct lfq_node *expected_head = head; + if (!atomic_cas_node(&ctx->head, &expected_head, next)) { + hp_clear_pair(ctx, tid); + continue; + } + + /* + * Keep our hazards published while putting the removed dummy on the + * retired list. Only after publication is it safe to drop them. + */ + retired_push(ctx, head); + hp_clear_pair(ctx, tid); + try_reclaim(ctx, false); + + *out = value; + return 1; + } +} + +int lfq_init(struct lfq_ctx *ctx, int max_consume_thread) { + if (!ctx) + return -EINVAL; + + if (max_consume_thread == 0) + max_consume_thread = LFQ_DEFAULT_MAX_CONSUMERS; + if (max_consume_thread < 0) + return -EINVAL; + + memset(ctx, 0, sizeof(*ctx)); + + struct lfq_node *dummy = (struct lfq_node *)calloc(1, sizeof(*dummy)); + if (!dummy) + return -ENOMEM; + + size_t hp_count = (size_t)max_consume_thread * LFQ_HAZARDS_PER_THREAD; + if (hp_count / LFQ_HAZARDS_PER_THREAD != (size_t)max_consume_thread) { + free(dummy); + return -EINVAL; + } + + ctx->HP = (struct lfq_node * volatile *)calloc(hp_count, sizeof(*ctx->HP)); + ctx->tid_map = (lfq_atomic_int_t *)calloc((size_t)max_consume_thread, sizeof(*ctx->tid_map)); + if (!ctx->HP || !ctx->tid_map) { + free((void *)ctx->HP); + free((void *)ctx->tid_map); + free(dummy); + memset(ctx, 0, sizeof(*ctx)); + return -ENOMEM; + } + + ctx->MAXHPSIZE = max_consume_thread; + atomic_store_node(&dummy->next, NULL); + atomic_store_node(&ctx->head, dummy); + atomic_store_node(&ctx->tail, dummy); + atomic_store_node(&ctx->retired_head, NULL); + lfq_atomic_store_int(&ctx->retired_count, 0); + lfq_atomic_store_int(&ctx->reclaiming, 0); + lfq_atomic_store_int(&ctx->active_enqueues, 0); + lfq_atomic_store_int(&ctx->op_state, 0); + + return 0; +} + +long lfg_count_freelist(const struct lfq_ctx *ctx) { + if (!ctx) + return 0; + return (long)lfq_atomic_load_int((lfq_atomic_int_t *)&ctx->retired_count); +} + +int lfq_clean(struct lfq_ctx *ctx) { + if (!ctx) + return -EINVAL; + + int expected = 0; + if (!lfq_atomic_cas_int(&ctx->op_state, &expected, LFQ_OP_CLOSING)) + return -EBUSY; + + if (!ctx->HP && !ctx->tid_map && !atomic_load_node(&ctx->head)) { + memset(ctx, 0, sizeof(*ctx)); + return 0; + } + + for (int i = 0; i < ctx->MAXHPSIZE; ++i) { + if (lfq_atomic_load_int(&ctx->tid_map[i]) != 0) { + lfq_atomic_store_int(&ctx->op_state, 0); + return -EBUSY; + } + } + + struct lfq_node *node = atomic_exchange_node(&ctx->head, NULL); + atomic_store_node(&ctx->tail, NULL); + + while (node) { + struct lfq_node *next = atomic_load_node(&node->next); + free(node); + node = next; + } + + node = atomic_exchange_node(&ctx->retired_head, NULL); + while (node) { + struct lfq_node *next = node->retired_next; + free(node); + node = next; + } + + free((void *)ctx->HP); + free((void *)ctx->tid_map); + memset(ctx, 0, sizeof(*ctx)); + return 0; +} + +int lfq_enqueue(struct lfq_ctx *ctx, void *data) { + if (!data || data == LFQ_ERROR) + return -EINVAL; + + int rc = begin_op(ctx); + if (rc < 0) + return rc; + + struct lfq_node *node = (struct lfq_node *)calloc(1, sizeof(*node)); + if (!node) { + end_op(ctx); + return -ENOMEM; + } + + node->data = data; + node->retired_next = NULL; + atomic_store_node(&node->next, NULL); + + /* + * Hold the producer epoch before loading tail. Reclamation may proceed + * concurrently with producers that start later, but it must not free a + * node referenced by an already-running producer's stale local tail. + */ + (void)lfq_atomic_fetch_add_int(&ctx->active_enqueues, 1); + + for (;;) { + struct lfq_node *tail = atomic_load_node(&ctx->tail); + if (!tail) { + (void)lfq_atomic_fetch_add_int(&ctx->active_enqueues, -1); + free(node); + end_op(ctx); + return -EINVAL; + } + + struct lfq_node *next = atomic_load_node(&tail->next); + if (tail != atomic_load_node(&ctx->tail)) + continue; + + if (!next) { + struct lfq_node *expected_next = NULL; + if (atomic_cas_node(&tail->next, &expected_next, node)) { + /* + * The link CAS above is the enqueue linearization point. + * Advancing tail is only an optimization; other threads help + * if this producer is descheduled here. + */ + struct lfq_node *expected_tail = tail; + (void)atomic_cas_node(&ctx->tail, &expected_tail, node); + (void)lfq_atomic_fetch_add_int(&ctx->active_enqueues, -1); + end_op(ctx); + return 0; + } + } else { + struct lfq_node *expected_tail = tail; + (void)atomic_cas_node(&ctx->tail, &expected_tail, next); + } + } +} + +int lfq_try_dequeue_tid(struct lfq_ctx *ctx, int tid, void **out) { + if (!out || tid < 0) + return -EINVAL; + *out = NULL; + + int rc = begin_op(ctx); + if (rc < 0) + return rc; + + int acquired = acquire_tid(ctx, tid); + if (acquired < 0) { + end_op(ctx); + return acquired; + } + + rc = dequeue_reserved(ctx, acquired, out); + release_tid(ctx, acquired); + end_op(ctx); + return rc; +} + +int lfq_try_dequeue(struct lfq_ctx *ctx, void **out) { + if (!out) + return -EINVAL; + *out = NULL; + + int rc = begin_op(ctx); + if (rc < 0) + return rc; + + int tid = acquire_tid(ctx, -1); + if (tid < 0) { + end_op(ctx); + return tid; + } + + rc = dequeue_reserved(ctx, tid, out); + release_tid(ctx, tid); + end_op(ctx); + return rc; +} + +void *lfq_dequeue_tid(struct lfq_ctx *ctx, int tid) { + void *out = NULL; + int rc = lfq_try_dequeue_tid(ctx, tid, &out); + if (rc > 0) + return out; + if (rc == 0) + return NULL; + return LFQ_ERROR; +} + +void *lfq_dequeue(struct lfq_ctx *ctx) { + void *out = NULL; + int rc = lfq_try_dequeue(ctx, &out); + if (rc > 0) + return out; + if (rc == 0) + return NULL; + return LFQ_ERROR; +} diff --git a/lfq.h b/lfq.h index 62d0df2..1277147 100755 --- a/lfq.h +++ b/lfq.h @@ -1,52 +1,67 @@ -#ifndef __LFQ_H__ -#define __LFQ_H__ -#include "cross-platform.h" - -#include // C11 - -struct lfq_node{ - void * data; - union { - struct lfq_node * volatile next; - struct lfq_node * volatile free_next; - }; - volatile int can_free; -}; - -struct lfq_ctx{ - alignas(64) volatile struct lfq_node * volatile head; - int volatile count; - volatile struct lfq_node * * HP; - volatile int * tid_map; - int volatile is_freeing; - volatile struct lfq_node * volatile fph; // free pool head - volatile struct lfq_node * volatile fpt; // free pool tail - int MAXHPSIZE; - - alignas(64) volatile struct lfq_node * volatile tail; // in another cache line to avoid contention -}; - -int lfq_init(struct lfq_ctx *ctx, int max_consume_thread); -int lfq_clean(struct lfq_ctx *ctx); -long lfg_count_freelist(const struct lfq_ctx *ctx); - -int lfq_enqueue(struct lfq_ctx *ctx, void * data); -void * lfq_dequeue_tid(struct lfq_ctx *ctx, int tid ); -void * lfq_dequeue(struct lfq_ctx *ctx ); - -/********************************************************** - * - * This macro will dequeue forever. - * If you do not like high cost cpu, - * use original dequeue function with memory barrier, - * and sleep/thread_yield will be better idea. - * - *********************************************************/ -#define LFQ_MB_DEQUEUE(ctx, ret) ({ \ - do { \ - ret = lfq_dequeue(ctx); \ - mb(); \ - } while(ret == 0); \ -}) - -#endif +#ifndef __LFQ_H__ +#define __LFQ_H__ + +#include "cross-platform.h" + +#define LFQ_DEFAULT_MAX_CONSUMERS 16 +#define LFQ_HAZARDS_PER_THREAD 2 +#define LFQ_ERROR ((void *)-1) + +struct lfq_node { + void *data; + struct lfq_node * volatile next; + struct lfq_node *retired_next; +}; + +struct lfq_ctx { + LFQ_ALIGNAS(64) struct lfq_node * volatile head; + LFQ_ALIGNAS(64) struct lfq_node * volatile tail; + + struct lfq_node * volatile retired_head; + lfq_atomic_int_t retired_count; + lfq_atomic_int_t reclaiming; + /* Prevent reclamation while a producer may hold a stale local tail. */ + lfq_atomic_int_t active_enqueues; + + struct lfq_node * volatile *HP; + lfq_atomic_int_t *tid_map; + int MAXHPSIZE; + + /* One atomic gate: closing bit + active operation count. */ + lfq_atomic_int_t op_state; +}; + +int lfq_init(struct lfq_ctx *ctx, int max_consume_thread); +int lfq_clean(struct lfq_ctx *ctx); + +/* Compatibility name: now returns the number of retired, not-yet-reclaimed nodes. */ +long lfg_count_freelist(const struct lfq_ctx *ctx); + +int lfq_enqueue(struct lfq_ctx *ctx, void *data); + +/* + * Status-returning dequeue API: + * 1 item dequeued and stored in *out + * 0 queue empty + * <0 negative errno-style error + */ +int lfq_try_dequeue(struct lfq_ctx *ctx, void **out); +int lfq_try_dequeue_tid(struct lfq_ctx *ctx, int tid, void **out); + +/* + * Legacy pointer-returning API: + * NULL queue empty + * LFQ_ERROR error + * otherwise dequeued payload + */ +void *lfq_dequeue(struct lfq_ctx *ctx); +void *lfq_dequeue_tid(struct lfq_ctx *ctx, int tid); + +#define LFQ_MB_DEQUEUE(ctx, ret) do { \ + do { \ + (ret) = lfq_dequeue((ctx)); \ + mb(); \ + } while ((ret) == NULL); \ +} while (0) + +#endif diff --git a/test_aba.c b/test_aba.c index 0125295..20ceeb5 100644 --- a/test_aba.c +++ b/test_aba.c @@ -1,166 +1,164 @@ /* - * ABA stress test for lfqueue + * Hazard-pointer / rapid-reuse stress test. * - * The ABA problem: a CAS on a pointer can succeed spuriously when: - * 1. Thread 1 reads head = node_A - * 2. Thread 2 dequeues A, frees it; malloc() returns the same address for - * a new node B; B is enqueued, making head cycle back to address A - * 3. Thread 1's CAS(&head, A, A->next) succeeds even though the node at - * address A is now logically B — A->next is stale/garbage. - * - * To maximize ABA probability this test uses: - * - All threads as both producers AND consumers (tight enq+deq loops) - * - No thread_yield between enqueue and dequeue (maximises racing) - * - Very small queue depth (each thread keeps at most 1 item in-flight) - * - Many threads competing on the same head pointer - * - Magic values in each node to detect data corruption caused by ABA + * This is intentionally not a formal linearizability proof. It maximizes + * rapid queue-node retirement while user payloads are also allocated/freed. + * Sanitizer CI is expected to catch stale accesses and data races. */ #define _GNU_SOURCE +#include +#include #include #include #include -#include -#include + #include "lfq.h" #include "cross-platform.h" -#if defined __GNUC__ || defined __CYGWIN__ || defined __MINGW32__ -#include +#if !defined(_MSC_VER) #include -#else -#include #endif #ifndef ABA_THREADS #define ABA_THREADS 16 #endif - #ifndef ABA_ITERATIONS -#define ABA_ITERATIONS 500000 +#define ABA_ITERATIONS 200000 #endif -/* Two distinct magic values — LIVE (in queue) and DEAD (already freed). - * A read of DEAD_MAGIC on a dequeued item means use-after-free / ABA - * corruption. */ -#define LIVE_MAGIC 0xAB1AB1EFUL -#define DEAD_MAGIC 0xDEADBEEFUL +#define LIVE_MAGIC 0xAB1AB1EFu +#define DEAD_MAGIC 0xDEADBEEFu struct aba_item { - volatile uint32_t magic; + uint32_t magic; }; static volatile uint64_t total_enqueued = 0; static volatile uint64_t total_dequeued = 0; -static volatile int errors = 0; -static volatile int cn_t = 0; /* tid allocator */ - +static lfq_atomic_int_t errors = 0; +static lfq_atomic_int_t cn_t = 0; static struct lfq_ctx ctx; -/* - * Each thread alternates between enqueue and spin-dequeue in a tight loop. - * This keeps the queue depth near zero, so the same pointer addresses cycle - * through head rapidly — exactly the condition that triggers ABA. - */ THREAD_FN aba_thread(void *arg) { - (void)arg; - /* Allocate a dedicated hazard-pointer slot for this thread */ - int tid = ATOMIC_ADD(&cn_t, 1) - 1; - - uint64_t local_enq = 0, local_deq = 0; - - for (int i = 0; i < ABA_ITERATIONS && !errors; i++) { - /* --- Enqueue a fresh item --- */ - struct aba_item *item = malloc(sizeof(struct aba_item)); - if (!item) { - ATOMIC_ADD(&errors, 1); - break; - } - item->magic = LIVE_MAGIC; - - if (lfq_enqueue(&ctx, item) != 0) { - free(item); - ATOMIC_ADD(&errors, 1); - break; - } - local_enq++; - - /* - * --- Spin-dequeue (no yield) --- - * By immediately spinning for a result with no sleep we keep - * maximum pressure on the head CAS, creating the window for ABA: - * our enqueued node may be stolen and its memory recycled before - * we manage to dequeue it ourselves. - */ - struct aba_item *got; - do { - got = lfq_dequeue_tid(&ctx, tid); - } while (got == NULL); - - /* Validate: corruption means ABA or use-after-free occurred */ - if (got->magic != LIVE_MAGIC) { - printf("ABA corruption detected! " - "magic=0x%08X (expected 0x%08X) iter=%d tid=%d\n", - got->magic, (uint32_t)LIVE_MAGIC, i, tid); - ATOMIC_ADD(&errors, 1); - } - /* Poison before free to detect future use-after-free reads */ - got->magic = DEAD_MAGIC; - free(got); - local_deq++; - } - - ATOMIC_ADD64(&total_enqueued, local_enq); - ATOMIC_ADD64(&total_dequeued, local_deq); - return 0; + (void)arg; + int tid = ATOMIC_ADD(&cn_t, 1) - 1; + uint64_t local_enq = 0; + uint64_t local_deq = 0; + + for (int i = 0; i < ABA_ITERATIONS && lfq_atomic_load_int(&errors) == 0; ++i) { + struct aba_item *item = (struct aba_item *)malloc(sizeof(*item)); + if (!item) { + ATOMIC_ADD(&errors, 1); + break; + } + item->magic = LIVE_MAGIC; + + if (lfq_enqueue(&ctx, item) != 0) { + free(item); + ATOMIC_ADD(&errors, 1); + break; + } + ++local_enq; + + for (;;) { + void *raw = NULL; + int rc = lfq_try_dequeue_tid(&ctx, tid, &raw); + if (rc == 0) + continue; + if (rc < 0) { + fprintf(stderr, "dequeue error %d tid=%d\n", rc, tid); + ATOMIC_ADD(&errors, 1); + break; + } + + struct aba_item *got = (struct aba_item *)raw; + if (!got || got->magic != LIVE_MAGIC) { + fprintf(stderr, "payload corruption tid=%d iter=%d\n", tid, i); + ATOMIC_ADD(&errors, 1); + if (got) + free(got); + break; + } + + got->magic = DEAD_MAGIC; + free(got); + ++local_deq; + break; + } + } + + ATOMIC_ADD64(&total_enqueued, local_enq); + ATOMIC_ADD64(&total_dequeued, local_deq); + return 0; } int main(void) { - printf("ABA stress test: %d threads x %d iterations each\n", - ABA_THREADS, ABA_ITERATIONS); - printf("(tight enq+deq loops, no yield — maximises head-pointer reuse)\n"); - - if (lfq_init(&ctx, ABA_THREADS) != 0) { - fprintf(stderr, "lfq_init failed\n"); - return 1; - } - - THREAD_TOKEN threads[ABA_THREADS]; - for (int i = 0; i < ABA_THREADS; i++) { -#if defined __GNUC__ || defined __CYGWIN__ || defined __MINGW32__ - pthread_create(&threads[i], NULL, aba_thread, NULL); + if (lfq_init(&ctx, ABA_THREADS) != 0) { + fprintf(stderr, "lfq_init failed\n"); + return 1; + } + + THREAD_TOKEN threads[ABA_THREADS]; + int created[ABA_THREADS] = {0}; + + for (int i = 0; i < ABA_THREADS; ++i) { +#if defined(_MSC_VER) + threads[i] = CreateThread(NULL, 0, aba_thread, NULL, 0, NULL); + if (threads[i]) { + created[i] = 1; + } else { + ATOMIC_ADD(&errors, 1); + } #else -#pragma warning(disable:4133) - threads[i] = CreateThread(NULL, 0, aba_thread, NULL, 0, 0); + if (pthread_create(&threads[i], NULL, aba_thread, NULL) == 0) { + created[i] = 1; + } else { + ATOMIC_ADD(&errors, 1); + } #endif - } - - for (int i = 0; i < ABA_THREADS; i++) - THREAD_WAIT(threads[i]); - - /* Drain any items left in the queue after all threads exit */ - struct aba_item *item; - int drain_tid = 0; /* safe: all thread tids have been released */ - while ((item = lfq_dequeue_tid(&ctx, drain_tid)) != NULL) { - if (item->magic != LIVE_MAGIC) { - printf("ABA corruption in drain! magic=0x%08X\n", item->magic); - errors++; - } - item->magic = DEAD_MAGIC; - free(item); - total_dequeued++; - } - - long freecount = lfg_count_freelist(&ctx); - int clean = lfq_clean(&ctx); - - printf("Enqueued=%" PRId64 " Dequeued=%" PRId64 - " freelist=%ld clean=%d\n", - total_enqueued, total_dequeued, freecount, clean); - - if (errors) - printf("ABA Test FAILED!! (%d corruption(s) detected)\n", errors); - else - printf("ABA Test PASS!!\n"); - - return errors != 0; + } + + for (int i = 0; i < ABA_THREADS; ++i) { + if (created[i]) + THREAD_WAIT(threads[i]); + } + + for (;;) { + void *raw = NULL; + int rc = lfq_try_dequeue(&ctx, &raw); + if (rc == 0) + break; + if (rc < 0) { + ATOMIC_ADD(&errors, 1); + break; + } + + struct aba_item *item = (struct aba_item *)raw; + if (!item || item->magic != LIVE_MAGIC) { + ATOMIC_ADD(&errors, 1); + if (item) + free(item); + break; + } + + item->magic = DEAD_MAGIC; + free(item); + ++total_dequeued; + } + + long retired = lfg_count_freelist(&ctx); + int clean = lfq_clean(&ctx); + int error_count = lfq_atomic_load_int(&errors); + + printf("ABA stress: enqueued=%" PRIu64 " dequeued=%" PRIu64 + " retired=%ld clean=%d errors=%d\n", + total_enqueued, total_dequeued, retired, clean, error_count); + + if (error_count != 0 || total_enqueued != total_dequeued || clean != 0) { + fprintf(stderr, "ABA stress FAILED\n"); + return 1; + } + + printf("ABA stress PASS\n"); + return 0; } diff --git a/test_edge_cases.c b/test_edge_cases.c new file mode 100644 index 0000000..8ba0265 --- /dev/null +++ b/test_edge_cases.c @@ -0,0 +1,57 @@ +#include +#include +#include +#include "lfq.h" + +#define CHECK(expr) do { \ + if (!(expr)) { \ + fprintf(stderr, "CHECK failed at %s:%d: %s\n", __FILE__, __LINE__, #expr); \ + return 1; \ + } \ +} while (0) + +int main(void) { + struct lfq_ctx ctx; + void *out = NULL; + + CHECK(lfq_init(&ctx, 0) == 0); + CHECK(ctx.MAXHPSIZE == LFQ_DEFAULT_MAX_CONSUMERS); + + CHECK(lfq_enqueue(&ctx, NULL) == -EINVAL); + CHECK(lfq_enqueue(&ctx, LFQ_ERROR) == -EINVAL); + CHECK(lfq_try_dequeue_tid(&ctx, -1, &out) == -EINVAL); + CHECK(lfq_try_dequeue_tid(&ctx, ctx.MAXHPSIZE, &out) == -EINVAL); + + /* Manual and automatic consumers cannot silently share a hazard slot. */ + lfq_atomic_store_int(&ctx.tid_map[1], 1); + CHECK(lfq_try_dequeue_tid(&ctx, 1, &out) == -EBUSY); + CHECK(lfq_clean(&ctx) == -EBUSY); + lfq_atomic_store_int(&ctx.tid_map[1], 0); + + CHECK(lfq_enqueue(&ctx, (void *)(uintptr_t)1) == 0); + CHECK(lfq_enqueue(&ctx, (void *)(uintptr_t)2) == 0); + CHECK(lfq_enqueue(&ctx, (void *)(uintptr_t)3) == 0); + + CHECK(lfq_try_dequeue(&ctx, &out) == 1); + CHECK((uintptr_t)out == 1); + CHECK(lfq_try_dequeue_tid(&ctx, 0, &out) == 1); + CHECK((uintptr_t)out == 2); + CHECK((uintptr_t)lfq_dequeue(&ctx) == 3); + CHECK(lfq_dequeue(&ctx) == NULL); + + CHECK(lfq_clean(&ctx) == 0); + + /* Cleaning a queue with pending items must reclaim queue nodes cleanly. */ + CHECK(lfq_init(&ctx, 2) == 0); + CHECK(lfq_enqueue(&ctx, (void *)(uintptr_t)11) == 0); + CHECK(lfq_enqueue(&ctx, (void *)(uintptr_t)12) == 0); + CHECK(lfq_clean(&ctx) == 0); + + /* Repeated clean is harmless, and operations after clean fail cleanly. */ + CHECK(lfq_clean(&ctx) == 0); + CHECK(lfq_enqueue(&ctx, (void *)(uintptr_t)1) == -EINVAL); + CHECK(lfq_try_dequeue(&ctx, &out) == -EINVAL); + + printf("edge cases PASS\n"); + return 0; +} diff --git a/test_multithread.c b/test_multithread.c index 50658f0..263d1f1 100644 --- a/test_multithread.c +++ b/test_multithread.c @@ -1,136 +1,220 @@ -#define _GNU_SOURCE -#include -#include -#include -#include -#include -#include -#include "lfq.h" -#include "cross-platform.h" - -#if defined __GNUC__ || defined __CYGWIN__ || defined __MINGW32__ -#include -#include -#else -#include -#endif - - -#ifndef MAX_PRODUCER -#define MAX_PRODUCER 100 -#endif -#ifndef MAX_CONSUMER -#define MAX_CONSUMER 10 -#endif - - -#define SOMEID 667814649 - -volatile uint64_t cn_added = 0; -volatile uint64_t cn_deled = 0; - -volatile int cn_t = 0; -volatile int cn_producer = 0; - -struct user_data{ - long data; -}; - -THREAD_FN addq( void * data ) { - struct lfq_ctx * ctx = data; - int ret = 0; - long added; - for (added = 0 ; added < 500000 ; added++) { - struct user_data * p = malloc(sizeof(struct user_data)); - p->data=SOMEID; - if ( ( ret = lfq_enqueue(ctx,p) ) != 0 ) { - printf("lfq_enqueue failed, reason:%s\n", strerror(-ret)); - ATOMIC_ADD64(&cn_added, added); - ATOMIC_SUB(&cn_producer, 1); - return 0; - } - } - ATOMIC_ADD64(&cn_added, added); - ATOMIC_SUB(&cn_producer, 1); - printf("Producer thread [%lu] exited! Still %d running...\n",THREAD_ID(), cn_producer); - return 0; -} - -THREAD_FN delq(void * data) { - struct lfq_ctx * ctx = data; - struct user_data * p; - int tid = ATOMIC_ADD(&cn_t, 1) - 1; - - long deleted = 0; - while(1) { - p = lfq_dequeue_tid(ctx, tid); - if (p) { - if (p->data!=SOMEID){ - printf("data wrong!!\n"); - exit(1); - } - - free(p); - deleted++; - } else { - if (ctx->count || cn_producer) - THREAD_YIELD(); // queue is empty, release CPU slice - else - break; // queue is empty and no more producers - } - } - - ATOMIC_ADD64(&cn_deled, deleted); - - // p = lfq_dequeue_tid(ctx, tid); - printf("Consumer thread [%lu] exited %d\n",THREAD_ID(),cn_producer); - return 0; -} - -int main() { - struct lfq_ctx ctx; - int i = 0; - lfq_init(&ctx, MAX_CONSUMER); - THREAD_TOKEN thread_d[MAX_CONSUMER]; - THREAD_TOKEN thread_a[MAX_PRODUCER]; - - ATOMIC_ADD(&cn_producer, 1); - for ( i = 0 ; i < MAX_CONSUMER ; i++ ) { -#if defined __GNUC__ || defined __CYGWIN__ || defined __MINGW32__ - pthread_create(&thread_d[i], NULL, delq, (void*)&ctx); -#else -#pragma warning(disable:4133) - thread_d[i] = CreateThread(NULL, 0, delq, &ctx, 0, 0); -#endif - } - - for ( i = 0 ; i < MAX_PRODUCER ; i++ ){ - ATOMIC_ADD(&cn_producer, 1); -#if defined __GNUC__ || defined __CYGWIN__ || defined __MINGW32__ - pthread_create(&thread_a[i], NULL, addq, (void*)&ctx); -#else -#pragma warning(disable:4133) - thread_a[i] = CreateThread(NULL, 0, addq, &ctx, 0, 0); -#endif - - } - - ATOMIC_SUB(&cn_producer, 1); - - for (i = 0; i < MAX_PRODUCER; i++) - THREAD_WAIT(thread_a[i]); - - for ( i = 0 ; i < MAX_CONSUMER ; i++ ) - THREAD_WAIT(thread_d[i]); - - long freecount = lfg_count_freelist(&ctx); - int clean = lfq_clean(&ctx); - - printf("Total push %"PRId64" elements, pop %"PRId64" elements. freelist=%ld, clean = %d\n", cn_added, cn_deled, freecount, clean); - if ( cn_added == cn_deled ) - printf("Test PASS!!\n"); - else - printf("Test Failed!!\n"); - - return (cn_added != cn_deled); -} +#define _GNU_SOURCE +#include +#include +#include +#include +#include + +#include "lfq.h" +#include "cross-platform.h" + +#if !defined(_MSC_VER) +#include +#endif + +#ifndef MAX_PRODUCER +#define MAX_PRODUCER 100 +#endif +#ifndef MAX_CONSUMER +#define MAX_CONSUMER 10 +#endif +#ifndef ITEMS_PER_PRODUCER +#define ITEMS_PER_PRODUCER 100000 +#endif + +#define SOMEID 667814649u + +static volatile uint64_t cn_added = 0; +static volatile uint64_t cn_deled = 0; +static volatile uint64_t sum_added = 0; +static volatile uint64_t sum_deled = 0; + +static lfq_atomic_int_t producer_id_next = 0; +static lfq_atomic_int_t consumer_tid_next = 0; +static lfq_atomic_int_t producers_running = 0; +static lfq_atomic_int_t errors = 0; + +struct user_data { + uint32_t magic; + uint32_t producer; + uint32_t sequence; +}; + +THREAD_FN addq(void *data) { + struct lfq_ctx *ctx = (struct lfq_ctx *)data; + uint32_t producer = (uint32_t)(ATOMIC_ADD(&producer_id_next, 1) - 1); + uint64_t local_added = 0; + uint64_t local_sum = 0; + + for (uint32_t i = 0; i < ITEMS_PER_PRODUCER; ++i) { + struct user_data *p = (struct user_data *)malloc(sizeof(*p)); + if (!p) { + ATOMIC_ADD(&errors, 1); + break; + } + + p->magic = SOMEID; + p->producer = producer; + p->sequence = i; + + int rc = lfq_enqueue(ctx, p); + if (rc != 0) { + fprintf(stderr, "lfq_enqueue failed: %d\n", rc); + free(p); + ATOMIC_ADD(&errors, 1); + break; + } + + ++local_added; + local_sum += ((uint64_t)producer << 32) | i; + } + + ATOMIC_ADD64(&cn_added, local_added); + ATOMIC_ADD64(&sum_added, local_sum); + ATOMIC_SUB(&producers_running, 1); + return 0; +} + +THREAD_FN delq(void *data) { + struct lfq_ctx *ctx = (struct lfq_ctx *)data; + int tid = ATOMIC_ADD(&consumer_tid_next, 1) - 1; + uint64_t local_deleted = 0; + uint64_t local_sum = 0; + + for (;;) { + void *raw = NULL; + int rc = lfq_try_dequeue_tid(ctx, tid, &raw); + if (rc == 1) { + struct user_data *p = (struct user_data *)raw; + if (!p || p->magic != SOMEID) { + fprintf(stderr, "data corruption detected\n"); + ATOMIC_ADD(&errors, 1); + if (p) + free(p); + break; + } + + local_sum += ((uint64_t)p->producer << 32) | p->sequence; + ++local_deleted; + free(p); + continue; + } + + if (rc < 0) { + fprintf(stderr, "lfq_try_dequeue_tid failed: %d\n", rc); + ATOMIC_ADD(&errors, 1); + break; + } + + if (lfq_atomic_load_int(&producers_running) == 0) + break; + + THREAD_YIELD(); + } + + ATOMIC_ADD64(&cn_deled, local_deleted); + ATOMIC_ADD64(&sum_deled, local_sum); + return 0; +} + +int main(void) { + struct lfq_ctx ctx; + if (lfq_init(&ctx, MAX_CONSUMER) != 0) { + fprintf(stderr, "lfq_init failed\n"); + return 1; + } + + THREAD_TOKEN consumers[MAX_CONSUMER]; + THREAD_TOKEN producers[MAX_PRODUCER]; + int producer_created[MAX_PRODUCER] = {0}; + int consumer_created[MAX_CONSUMER] = {0}; + + ATOMIC_ADD(&producers_running, MAX_PRODUCER); + + for (int i = 0; i < MAX_PRODUCER; ++i) { +#if defined(_MSC_VER) + producers[i] = CreateThread(NULL, 0, addq, &ctx, 0, NULL); + if (producers[i]) { + producer_created[i] = 1; + } else { + ATOMIC_ADD(&errors, 1); + ATOMIC_SUB(&producers_running, 1); + } +#else + if (pthread_create(&producers[i], NULL, addq, &ctx) == 0) { + producer_created[i] = 1; + } else { + ATOMIC_ADD(&errors, 1); + ATOMIC_SUB(&producers_running, 1); + } +#endif + } + + for (int i = 0; i < MAX_CONSUMER; ++i) { +#if defined(_MSC_VER) + consumers[i] = CreateThread(NULL, 0, delq, &ctx, 0, NULL); + if (consumers[i]) { + consumer_created[i] = 1; + } else { + ATOMIC_ADD(&errors, 1); + } +#else + if (pthread_create(&consumers[i], NULL, delq, &ctx) == 0) { + consumer_created[i] = 1; + } else { + ATOMIC_ADD(&errors, 1); + } +#endif + } + + for (int i = 0; i < MAX_PRODUCER; ++i) { + if (producer_created[i]) + THREAD_WAIT(producers[i]); + } + + for (int i = 0; i < MAX_CONSUMER; ++i) { + if (consumer_created[i]) + THREAD_WAIT(consumers[i]); + } + + /* Drain if thread creation failed; accounting still has to match. */ + for (;;) { + void *raw = NULL; + int rc = lfq_try_dequeue(&ctx, &raw); + if (rc == 0) + break; + if (rc < 0) { + ATOMIC_ADD(&errors, 1); + break; + } + + struct user_data *p = (struct user_data *)raw; + if (!p || p->magic != SOMEID) { + ATOMIC_ADD(&errors, 1); + if (p) + free(p); + break; + } + + ATOMIC_ADD64(&sum_deled, ((uint64_t)p->producer << 32) | p->sequence); + ATOMIC_ADD64(&cn_deled, 1); + free(p); + } + + long retired = lfg_count_freelist(&ctx); + int clean = lfq_clean(&ctx); + int error_count = lfq_atomic_load_int(&errors); + + printf("push=%" PRIu64 " pop=%" PRIu64 + " sum_push=%" PRIu64 " sum_pop=%" PRIu64 + " retired=%ld clean=%d errors=%d\n", + cn_added, cn_deled, sum_added, sum_deled, retired, clean, error_count); + + if (cn_added != cn_deled || sum_added != sum_deled || clean != 0 || error_count != 0) { + fprintf(stderr, "multithread test FAILED\n"); + return 1; + } + + printf("multithread test PASS\n"); + return 0; +} From f4ccfaa1b6850eef2ee40cfd4bc8ef725e0471dc Mon Sep 17 00:00:00 2001 From: kautism Date: Sat, 26 Sep 2026 15:48:44 +0800 Subject: [PATCH 2/6] fix-producer-hazard-reclamation --- README.md | 2 + cross-platform.h | 20 +++---- lfq.c | 138 +++++++++++++++++++++++++++++++++-------------- lfq.h | 11 ++-- 4 files changed, 118 insertions(+), 53 deletions(-) diff --git a/README.md b/README.md index 76091b5..e64eec3 100755 --- a/README.md +++ b/README.md @@ -16,6 +16,8 @@ The queue uses the Michael-Scott linked-queue protocol: - lagging `tail` pointers are helped forward by producers and consumers; - dequeue protects both the current dummy head and its successor with two Hazard Pointers before dereferencing them; +- each producer acquires a dynamically-grown Hazard Pointer record before + dereferencing a local tail, so a stale producer tail cannot be reclaimed; - removed dummy nodes go to a separate retired list. The retired-list link is **not** overlaid on the queue's `next` field, so a hazard-protected node remains immutable until no reader can reference it; diff --git a/cross-platform.h b/cross-platform.h index b1809a7..9c835fd 100644 --- a/cross-platform.h +++ b/cross-platform.h @@ -41,24 +41,24 @@ static __forceinline bool lfq_atomic_cas_ptr(void * volatile *ptr, void **expect return false; } -static __forceinline LONG lfq_atomic_load_int(lfq_atomic_int_t *ptr) { - return InterlockedCompareExchange((volatile LONG *)ptr, 0, 0); +static __forceinline int lfq_atomic_load_int(lfq_atomic_int_t *ptr) { + return (int)InterlockedCompareExchange((volatile LONG *)ptr, 0, 0); } -static __forceinline void lfq_atomic_store_int(lfq_atomic_int_t *ptr, LONG value) { - (void)InterlockedExchange((volatile LONG *)ptr, value); +static __forceinline void lfq_atomic_store_int(lfq_atomic_int_t *ptr, int value) { + (void)InterlockedExchange((volatile LONG *)ptr, (LONG)value); } -static __forceinline bool lfq_atomic_cas_int(lfq_atomic_int_t *ptr, LONG *expected, LONG desired) { - LONG actual = InterlockedCompareExchange((volatile LONG *)ptr, desired, *expected); - if (actual == *expected) +static __forceinline bool lfq_atomic_cas_int(lfq_atomic_int_t *ptr, int *expected, int desired) { + LONG actual = InterlockedCompareExchange((volatile LONG *)ptr, (LONG)desired, (LONG)*expected); + if (actual == (LONG)*expected) return true; - *expected = actual; + *expected = (int)actual; return false; } -static __forceinline LONG lfq_atomic_fetch_add_int(lfq_atomic_int_t *ptr, LONG value) { - return InterlockedExchangeAdd((volatile LONG *)ptr, value); +static __forceinline int lfq_atomic_fetch_add_int(lfq_atomic_int_t *ptr, int value) { + return (int)InterlockedExchangeAdd((volatile LONG *)ptr, (LONG)value); } #define LFQ_ALIGNAS(n) __declspec(align(n)) diff --git a/lfq.c b/lfq.c index e52f6df..77d1636 100755 --- a/lfq.c +++ b/lfq.c @@ -7,6 +7,12 @@ #define LFQ_RECLAIM_MIN 32 #define LFQ_OP_CLOSING 0x40000000 +struct lfq_producer_hp { + struct lfq_node * volatile hazard; + lfq_atomic_int_t in_use; + struct lfq_producer_hp *next; +}; + static struct lfq_node *atomic_load_node(struct lfq_node * volatile *ptr) { return (struct lfq_node *)lfq_atomic_load_ptr((void * volatile *)ptr); } @@ -29,6 +35,40 @@ static bool atomic_cas_node(struct lfq_node * volatile *ptr, return ok; } +static struct lfq_producer_hp *producer_hp_acquire(struct lfq_ctx *ctx) { + struct lfq_producer_hp *record = + (struct lfq_producer_hp *)lfq_atomic_load_ptr((void * volatile *)&ctx->producer_hps); + + for (; record; record = record->next) { + int expected = 0; + if (lfq_atomic_cas_int(&record->in_use, &expected, 1)) { + lfq_atomic_store_ptr((void * volatile *)&record->hazard, NULL); + return record; + } + } + + record = (struct lfq_producer_hp *)calloc(1, sizeof(*record)); + if (!record) + return NULL; + + lfq_atomic_store_ptr((void * volatile *)&record->hazard, NULL); + lfq_atomic_store_int(&record->in_use, 1); + + for (;;) { + struct lfq_producer_hp *old = + (struct lfq_producer_hp *)lfq_atomic_load_ptr((void * volatile *)&ctx->producer_hps); + record->next = old; + void *expected = old; + if (lfq_atomic_cas_ptr((void * volatile *)&ctx->producer_hps, &expected, record)) + return record; + } +} + +static void producer_hp_release(struct lfq_producer_hp *record) { + lfq_atomic_store_ptr((void * volatile *)&record->hazard, NULL); + lfq_atomic_store_int(&record->in_use, 0); +} + static int begin_op(struct lfq_ctx *ctx) { if (!ctx) return -EINVAL; @@ -80,6 +120,14 @@ static bool node_is_hazard(const struct lfq_ctx *ctx, const struct lfq_node *nod if (hp_load(ctx, i) == node) return true; } + + struct lfq_producer_hp *record = + (struct lfq_producer_hp *)lfq_atomic_load_ptr((void * volatile *)&ctx->producer_hps); + for (; record; record = record->next) { + if ((struct lfq_node *)lfq_atomic_load_ptr((void * volatile *)&record->hazard) == node) + return true; + } + return false; } @@ -110,46 +158,44 @@ static void try_reclaim(struct lfq_ctx *ctx, bool force) { if (!lfq_atomic_cas_int(&ctx->reclaiming, &expected, 1)) return; - /* - * Producers dereference a local tail pointer. A producer that loaded an - * old tail before another thread advanced it does not have a hazard slot, - * so reclamation must wait until all producers have left their queue - * traversal. A producer starting after this check can only load the - * current global tail, which is never on the retired list detached below. - */ - if (!force && lfq_atomic_load_int(&ctx->active_enqueues) != 0) { + int threshold = ctx->MAXHPSIZE * LFQ_HAZARDS_PER_THREAD * 2; + if (threshold < LFQ_RECLAIM_MIN) + threshold = LFQ_RECLAIM_MIN; + + if (!force && lfq_atomic_load_int(&ctx->retired_count) < threshold) { lfq_atomic_store_int(&ctx->reclaiming, 0); return; } - if (!force) { - int threshold = ctx->MAXHPSIZE * LFQ_HAZARDS_PER_THREAD * 2; - if (threshold < LFQ_RECLAIM_MIN) - threshold = LFQ_RECLAIM_MIN; + /* + * A long scan can overlap many new retirements. Drain a few snapshots in + * one ownership period so the last operations in a burst do not leave a + * large retired list merely because they lost the reclaimer CAS. + */ + for (int pass = 0; pass < 4; ++pass) { + struct lfq_node *list = atomic_exchange_node(&ctx->retired_head, NULL); + if (!list) + break; - if (lfq_atomic_load_int(&ctx->retired_count) < threshold) { - lfq_atomic_store_int(&ctx->reclaiming, 0); - return; + int freed = 0; + while (list) { + struct lfq_node *next = list->retired_next; + if (node_is_hazard(ctx, list)) { + retired_requeue(ctx, list); + } else { + free(list); + ++freed; + } + list = next; } - } - struct lfq_node *list = atomic_exchange_node(&ctx->retired_head, NULL); - int freed = 0; + if (freed) + (void)lfq_atomic_fetch_add_int(&ctx->retired_count, -freed); - while (list) { - struct lfq_node *next = list->retired_next; - if (node_is_hazard(ctx, list)) { - retired_requeue(ctx, list); - } else { - free(list); - ++freed; - } - list = next; + if (!force && lfq_atomic_load_int(&ctx->retired_count) < threshold) + break; } - if (freed) - (void)lfq_atomic_fetch_add_int(&ctx->retired_count, -freed); - lfq_atomic_store_int(&ctx->reclaiming, 0); } @@ -267,9 +313,9 @@ int lfq_init(struct lfq_ctx *ctx, int max_consume_thread) { atomic_store_node(&ctx->head, dummy); atomic_store_node(&ctx->tail, dummy); atomic_store_node(&ctx->retired_head, NULL); + lfq_atomic_store_ptr((void * volatile *)&ctx->producer_hps, NULL); lfq_atomic_store_int(&ctx->retired_count, 0); lfq_atomic_store_int(&ctx->reclaiming, 0); - lfq_atomic_store_int(&ctx->active_enqueues, 0); lfq_atomic_store_int(&ctx->op_state, 0); return 0; @@ -317,6 +363,15 @@ int lfq_clean(struct lfq_ctx *ctx) { node = next; } + struct lfq_producer_hp *producer_hp = + (struct lfq_producer_hp *)lfq_atomic_exchange_ptr( + (void * volatile *)&ctx->producer_hps, NULL); + while (producer_hp) { + struct lfq_producer_hp *next = producer_hp->next; + free(producer_hp); + producer_hp = next; + } + free((void *)ctx->HP); free((void *)ctx->tid_map); memset(ctx, 0, sizeof(*ctx)); @@ -341,22 +396,26 @@ int lfq_enqueue(struct lfq_ctx *ctx, void *data) { node->retired_next = NULL; atomic_store_node(&node->next, NULL); - /* - * Hold the producer epoch before loading tail. Reclamation may proceed - * concurrently with producers that start later, but it must not free a - * node referenced by an already-running producer's stale local tail. - */ - (void)lfq_atomic_fetch_add_int(&ctx->active_enqueues, 1); + struct lfq_producer_hp *producer_hp = producer_hp_acquire(ctx); + if (!producer_hp) { + free(node); + end_op(ctx); + return -ENOMEM; + } for (;;) { struct lfq_node *tail = atomic_load_node(&ctx->tail); if (!tail) { - (void)lfq_atomic_fetch_add_int(&ctx->active_enqueues, -1); + producer_hp_release(producer_hp); free(node); end_op(ctx); return -EINVAL; } + lfq_atomic_store_ptr((void * volatile *)&producer_hp->hazard, tail); + if (tail != atomic_load_node(&ctx->tail)) + continue; + struct lfq_node *next = atomic_load_node(&tail->next); if (tail != atomic_load_node(&ctx->tail)) continue; @@ -371,7 +430,8 @@ int lfq_enqueue(struct lfq_ctx *ctx, void *data) { */ struct lfq_node *expected_tail = tail; (void)atomic_cas_node(&ctx->tail, &expected_tail, node); - (void)lfq_atomic_fetch_add_int(&ctx->active_enqueues, -1); + producer_hp_release(producer_hp); + try_reclaim(ctx, false); end_op(ctx); return 0; } diff --git a/lfq.h b/lfq.h index 1277147..452b6e5 100755 --- a/lfq.h +++ b/lfq.h @@ -13,15 +13,18 @@ struct lfq_node { struct lfq_node *retired_next; }; +struct lfq_producer_hp; + struct lfq_ctx { - LFQ_ALIGNAS(64) struct lfq_node * volatile head; - LFQ_ALIGNAS(64) struct lfq_node * volatile tail; + struct lfq_node * volatile head; + struct lfq_node * volatile tail; struct lfq_node * volatile retired_head; lfq_atomic_int_t retired_count; lfq_atomic_int_t reclaiming; - /* Prevent reclamation while a producer may hold a stale local tail. */ - lfq_atomic_int_t active_enqueues; + + /* Dynamically-grown producer hazard records; records live until clean(). */ + struct lfq_producer_hp * volatile producer_hps; struct lfq_node * volatile *HP; lfq_atomic_int_t *tid_map; From cc96eff8057cc525c471fed275dbd0cf5cfa6960 Mon Sep 17 00:00:00 2001 From: kautism Date: Sat, 26 Sep 2026 15:53:22 +0800 Subject: [PATCH 3/6] validate-visual-studio-build --- .github/workflows/ci.yml | 3 +++ Visual Stdio/example.vcxproj | 4 ---- Visual Stdio/lfqueue.vcxproj | 4 ---- Visual Stdio/test.vcxproj | 4 ---- 4 files changed, 3 insertions(+), 12 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d4c0920..8a7fbfd 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -57,3 +57,6 @@ jobs: bin\test_edge_cases.exe bin\test_multithread.exe bin\test_aba.exe + - name: Build Visual Studio solution + shell: cmd + run: msbuild "Visual Stdio\lfqueue.sln" /m /p:Configuration=Release /p:Platform=x64 diff --git a/Visual Stdio/example.vcxproj b/Visual Stdio/example.vcxproj index cfc9fe1..ccc4ee3 100644 --- a/Visual Stdio/example.vcxproj +++ b/Visual Stdio/example.vcxproj @@ -28,26 +28,22 @@ Application true - v143 Unicode Application false - v143 true Unicode Application true - v143 Unicode Application false - v143 true Unicode diff --git a/Visual Stdio/lfqueue.vcxproj b/Visual Stdio/lfqueue.vcxproj index 0f6c631..d53470f 100644 --- a/Visual Stdio/lfqueue.vcxproj +++ b/Visual Stdio/lfqueue.vcxproj @@ -29,26 +29,22 @@ StaticLibrary true - v143 Unicode StaticLibrary false - v143 true Unicode StaticLibrary true - v143 Unicode StaticLibrary false - v143 true Unicode diff --git a/Visual Stdio/test.vcxproj b/Visual Stdio/test.vcxproj index 1e1b953..78ceaa9 100644 --- a/Visual Stdio/test.vcxproj +++ b/Visual Stdio/test.vcxproj @@ -27,26 +27,22 @@ Application true - v143 Unicode Application false - v143 true Unicode Application true - v143 Unicode Application false - v143 true Unicode From 8bce00ccd82a35bad0dac5beaa63d52b46b7e4aa Mon Sep 17 00:00:00 2001 From: kautism Date: Sat, 26 Sep 2026 15:56:41 +0800 Subject: [PATCH 4/6] fix-msbuild-toolset-selection --- .github/workflows/ci.yml | 2 +- Visual Stdio/example.vcxproj | 4 ++++ Visual Stdio/lfqueue.vcxproj | 4 ++++ Visual Stdio/test.vcxproj | 4 ++++ 4 files changed, 13 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8a7fbfd..4066166 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -59,4 +59,4 @@ jobs: bin\test_aba.exe - name: Build Visual Studio solution shell: cmd - run: msbuild "Visual Stdio\lfqueue.sln" /m /p:Configuration=Release /p:Platform=x64 + run: msbuild "Visual Stdio\lfqueue.sln" /m /p:Configuration=Release /p:Platform=x64 /p:PlatformToolset=v145 diff --git a/Visual Stdio/example.vcxproj b/Visual Stdio/example.vcxproj index ccc4ee3..cfc9fe1 100644 --- a/Visual Stdio/example.vcxproj +++ b/Visual Stdio/example.vcxproj @@ -28,22 +28,26 @@ Application true + v143 Unicode Application false + v143 true Unicode Application true + v143 Unicode Application false + v143 true Unicode diff --git a/Visual Stdio/lfqueue.vcxproj b/Visual Stdio/lfqueue.vcxproj index d53470f..0f6c631 100644 --- a/Visual Stdio/lfqueue.vcxproj +++ b/Visual Stdio/lfqueue.vcxproj @@ -29,22 +29,26 @@ StaticLibrary true + v143 Unicode StaticLibrary false + v143 true Unicode StaticLibrary true + v143 Unicode StaticLibrary false + v143 true Unicode diff --git a/Visual Stdio/test.vcxproj b/Visual Stdio/test.vcxproj index 78ceaa9..1e1b953 100644 --- a/Visual Stdio/test.vcxproj +++ b/Visual Stdio/test.vcxproj @@ -27,22 +27,26 @@ Application true + v143 Unicode Application false + v143 true Unicode Application true + v143 Unicode Application false + v143 true Unicode From 30e72a81d257901f338e6a0312b0adcb2d6615de Mon Sep 17 00:00:00 2001 From: kautism Date: Sat, 26 Sep 2026 16:32:43 +0800 Subject: [PATCH 5/6] add-deterministic-scheduler-tests --- Makefile | 26 +++- lfq.c | 38 ++++++ lfq.h | 21 +++ test_multithread.c | 58 ++++++++ test_scheduler.c | 334 +++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 474 insertions(+), 3 deletions(-) create mode 100644 test_scheduler.c diff --git a/Makefile b/Makefile index f40fc5d..136713e 100644 --- a/Makefile +++ b/Makefile @@ -4,7 +4,7 @@ CPPFLAGS ?= LDFLAGS ?= LDLIBS ?= -lpthread -TEST_BINS = bin/test_edge_cases bin/test_p1c1 bin/test_p4c4 bin/test_p100c10 bin/test_p10c100 bin/test_aba +TEST_BINS = bin/test_edge_cases bin/test_p1c1 bin/test_p4c4 bin/test_p100c10 bin/test_p10c100 bin/test_aba bin/test_scheduler bin/test_affinity_single bin/test_affinity_split .PHONY: all test clean test-sanitize test-tsan @@ -43,6 +43,15 @@ bin/test_p10c100: test_multithread.c lfq.c lfq.h cross-platform.h | bin bin/test_aba: test_aba.c lfq.c lfq.h cross-platform.h | bin $(CC) $(CPPFLAGS) $(CFLAGS) $(LDFLAGS) test_aba.c lfq.c -o $@ $(LDLIBS) +bin/test_scheduler: test_scheduler.c lfq.c lfq.h cross-platform.h | bin + $(CC) $(CPPFLAGS) $(CFLAGS) $(LDFLAGS) -DLFQ_TEST_HOOKS test_scheduler.c lfq.c -o $@ $(LDLIBS) + +bin/test_affinity_single: test_multithread.c lfq.c lfq.h cross-platform.h | bin + $(CC) $(CPPFLAGS) $(CFLAGS) $(LDFLAGS) test_multithread.c lfq.c -o $@ $(LDLIBS) -D LFQ_AFFINITY_MODE=1 -D MAX_PRODUCER=16 -D MAX_CONSUMER=16 -D ITEMS_PER_PRODUCER=20000 + +bin/test_affinity_split: test_multithread.c lfq.c lfq.h cross-platform.h | bin + $(CC) $(CPPFLAGS) $(CFLAGS) $(LDFLAGS) test_multithread.c lfq.c -o $@ $(LDLIBS) -D LFQ_AFFINITY_MODE=2 -D MAX_PRODUCER=16 -D MAX_CONSUMER=16 -D ITEMS_PER_PRODUCER=20000 + test: $(TEST_BINS) ./bin/test_edge_cases ./bin/test_p1c1 @@ -50,6 +59,9 @@ test: $(TEST_BINS) ./bin/test_p100c10 ./bin/test_p10c100 ./bin/test_aba + ./bin/test_scheduler + ./bin/test_affinity_single + ./bin/test_affinity_split bin/test_edge_asan: test_edge_cases.c lfq.c lfq.h cross-platform.h | bin $(CC) -std=gnu99 -O1 -g -Wall -Wextra -fno-omit-frame-pointer -fsanitize=address,undefined test_edge_cases.c lfq.c -o $@ $(LDLIBS) @@ -60,10 +72,14 @@ bin/test_mt_asan: test_multithread.c lfq.c lfq.h cross-platform.h | bin bin/test_aba_asan: test_aba.c lfq.c lfq.h cross-platform.h | bin $(CC) -std=gnu99 -O1 -g -Wall -Wextra -fno-omit-frame-pointer -fsanitize=address,undefined test_aba.c lfq.c -o $@ $(LDLIBS) -D ABA_THREADS=8 -D ABA_ITERATIONS=5000 -test-sanitize: bin/test_edge_asan bin/test_mt_asan bin/test_aba_asan +bin/test_scheduler_asan: test_scheduler.c lfq.c lfq.h cross-platform.h | bin + $(CC) -std=gnu99 -O1 -g -Wall -Wextra -fno-omit-frame-pointer -fsanitize=address,undefined -DLFQ_TEST_HOOKS test_scheduler.c lfq.c -o $@ $(LDLIBS) + +test-sanitize: bin/test_edge_asan bin/test_mt_asan bin/test_aba_asan bin/test_scheduler_asan ASAN_OPTIONS=detect_leaks=1:halt_on_error=1 ./bin/test_edge_asan ASAN_OPTIONS=detect_leaks=1:halt_on_error=1 ./bin/test_mt_asan ASAN_OPTIONS=detect_leaks=1:halt_on_error=1 ./bin/test_aba_asan + ASAN_OPTIONS=detect_leaks=1:halt_on_error=1 ./bin/test_scheduler_asan bin/test_mt_tsan: test_multithread.c lfq.c lfq.h cross-platform.h | bin $(CC) -std=gnu99 -O1 -g -Wall -Wextra -fno-omit-frame-pointer -fsanitize=thread test_multithread.c lfq.c -o $@ $(LDLIBS) -D MAX_PRODUCER=8 -D MAX_CONSUMER=8 -D ITEMS_PER_PRODUCER=5000 @@ -71,9 +87,13 @@ bin/test_mt_tsan: test_multithread.c lfq.c lfq.h cross-platform.h | bin bin/test_aba_tsan: test_aba.c lfq.c lfq.h cross-platform.h | bin $(CC) -std=gnu99 -O1 -g -Wall -Wextra -fno-omit-frame-pointer -fsanitize=thread test_aba.c lfq.c -o $@ $(LDLIBS) -D ABA_THREADS=8 -D ABA_ITERATIONS=5000 -test-tsan: bin/test_mt_tsan bin/test_aba_tsan +bin/test_scheduler_tsan: test_scheduler.c lfq.c lfq.h cross-platform.h | bin + $(CC) -std=gnu99 -O1 -g -Wall -Wextra -fno-omit-frame-pointer -fsanitize=thread -DLFQ_TEST_HOOKS test_scheduler.c lfq.c -o $@ $(LDLIBS) + +test-tsan: bin/test_mt_tsan bin/test_aba_tsan bin/test_scheduler_tsan TSAN_OPTIONS=halt_on_error=1 ./bin/test_mt_tsan TSAN_OPTIONS=halt_on_error=1 ./bin/test_aba_tsan + TSAN_OPTIONS=halt_on_error=1 ./bin/test_scheduler_tsan clean: rm -f *.o liblfq.a liblfq.so.1.0.0 diff --git a/lfq.c b/lfq.c index 77d1636..d9b7dfd 100755 --- a/lfq.c +++ b/lfq.c @@ -13,6 +13,23 @@ struct lfq_producer_hp { struct lfq_producer_hp *next; }; +#ifdef LFQ_TEST_HOOKS +static lfq_test_hook_fn lfq_test_hook; +static void *lfq_test_hook_arg; + +void lfq_test_set_hook(lfq_test_hook_fn hook, void *arg) { + lfq_test_hook = hook; + lfq_test_hook_arg = arg; +} + +#define LFQ_TEST_FIRE(point, ctx, first, second) do { \ + if (lfq_test_hook) \ + lfq_test_hook((point), (ctx), (first), (second), lfq_test_hook_arg); \ +} while (0) +#else +#define LFQ_TEST_FIRE(point, ctx, first, second) ((void)0) +#endif + static struct lfq_node *atomic_load_node(struct lfq_node * volatile *ptr) { return (struct lfq_node *)lfq_atomic_load_ptr((void * volatile *)ptr); } @@ -177,6 +194,8 @@ static void try_reclaim(struct lfq_ctx *ctx, bool force) { if (!list) break; + LFQ_TEST_FIRE(LFQ_TEST_RECLAIM_AFTER_DETACH, ctx, list, NULL); + int freed = 0; while (list) { struct lfq_node *next = list->retired_next; @@ -199,6 +218,13 @@ static void try_reclaim(struct lfq_ctx *ctx, bool force) { lfq_atomic_store_int(&ctx->reclaiming, 0); } +#ifdef LFQ_TEST_HOOKS +void lfq_test_force_reclaim(struct lfq_ctx *ctx) { + if (ctx) + try_reclaim(ctx, true); +} +#endif + static int acquire_tid(struct lfq_ctx *ctx, int requested_tid) { if (requested_tid >= 0) { if (requested_tid >= ctx->MAXHPSIZE) @@ -233,6 +259,8 @@ static int dequeue_reserved(struct lfq_ctx *ctx, int tid, void **out) { if (head != atomic_load_node(&ctx->head)) continue; + LFQ_TEST_FIRE(LFQ_TEST_DEQ_AFTER_PROTECT_HEAD, ctx, head, NULL); + struct lfq_node *tail = atomic_load_node(&ctx->tail); struct lfq_node *next = atomic_load_node(&head->next); @@ -250,6 +278,8 @@ static int dequeue_reserved(struct lfq_ctx *ctx, int tid, void **out) { continue; } + LFQ_TEST_FIRE(LFQ_TEST_DEQ_AFTER_PROTECT_NEXT, ctx, head, next); + if (head == tail) { struct lfq_node *expected_tail = tail; (void)atomic_cas_node(&ctx->tail, &expected_tail, next); @@ -264,6 +294,8 @@ static int dequeue_reserved(struct lfq_ctx *ctx, int tid, void **out) { continue; } + LFQ_TEST_FIRE(LFQ_TEST_DEQ_AFTER_HEAD_CAS, ctx, head, next); + /* * Keep our hazards published while putting the removed dummy on the * retired list. Only after publication is it safe to drop them. @@ -335,6 +367,8 @@ int lfq_clean(struct lfq_ctx *ctx) { if (!lfq_atomic_cas_int(&ctx->op_state, &expected, LFQ_OP_CLOSING)) return -EBUSY; + LFQ_TEST_FIRE(LFQ_TEST_CLEAN_AFTER_GATE, ctx, atomic_load_node(&ctx->head), NULL); + if (!ctx->HP && !ctx->tid_map && !atomic_load_node(&ctx->head)) { memset(ctx, 0, sizeof(*ctx)); return 0; @@ -416,6 +450,8 @@ int lfq_enqueue(struct lfq_ctx *ctx, void *data) { if (tail != atomic_load_node(&ctx->tail)) continue; + LFQ_TEST_FIRE(LFQ_TEST_ENQ_AFTER_PROTECT_TAIL, ctx, tail, node); + struct lfq_node *next = atomic_load_node(&tail->next); if (tail != atomic_load_node(&ctx->tail)) continue; @@ -423,6 +459,8 @@ int lfq_enqueue(struct lfq_ctx *ctx, void *data) { if (!next) { struct lfq_node *expected_next = NULL; if (atomic_cas_node(&tail->next, &expected_next, node)) { + LFQ_TEST_FIRE(LFQ_TEST_ENQ_AFTER_LINK, ctx, tail, node); + /* * The link CAS above is the enqueue linearization point. * Advancing tail is only an optimization; other threads help diff --git a/lfq.h b/lfq.h index 452b6e5..6a243e7 100755 --- a/lfq.h +++ b/lfq.h @@ -34,6 +34,27 @@ struct lfq_ctx { lfq_atomic_int_t op_state; }; +#ifdef LFQ_TEST_HOOKS +enum lfq_test_hook_point { + LFQ_TEST_ENQ_AFTER_PROTECT_TAIL = 1, + LFQ_TEST_ENQ_AFTER_LINK, + LFQ_TEST_DEQ_AFTER_PROTECT_HEAD, + LFQ_TEST_DEQ_AFTER_PROTECT_NEXT, + LFQ_TEST_DEQ_AFTER_HEAD_CAS, + LFQ_TEST_RECLAIM_AFTER_DETACH, + LFQ_TEST_CLEAN_AFTER_GATE +}; + +typedef void (*lfq_test_hook_fn)(enum lfq_test_hook_point point, + struct lfq_ctx *ctx, + struct lfq_node *first, + struct lfq_node *second, + void *arg); + +void lfq_test_set_hook(lfq_test_hook_fn hook, void *arg); +void lfq_test_force_reclaim(struct lfq_ctx *ctx); +#endif + int lfq_init(struct lfq_ctx *ctx, int max_consume_thread); int lfq_clean(struct lfq_ctx *ctx); diff --git a/test_multithread.c b/test_multithread.c index 263d1f1..3ade639 100644 --- a/test_multithread.c +++ b/test_multithread.c @@ -10,6 +10,9 @@ #if !defined(_MSC_VER) #include +#if defined(__linux__) +#include +#endif #endif #ifndef MAX_PRODUCER @@ -40,8 +43,58 @@ struct user_data { uint32_t sequence; }; +#if defined(__linux__) && defined(LFQ_AFFINITY_MODE) +static int affinity_cpu_for_role(int consumer) { + cpu_set_t allowed; + CPU_ZERO(&allowed); + if (sched_getaffinity(0, sizeof(allowed), &allowed) != 0) + return -1; + + int first = -1; + int second = -1; + for (int cpu = 0; cpu < CPU_SETSIZE; ++cpu) { + if (!CPU_ISSET(cpu, &allowed)) + continue; + if (first < 0) + first = cpu; + else { + second = cpu; + break; + } + } + + if (first < 0) + return -1; + +#if LFQ_AFFINITY_MODE == 2 + if (consumer && second >= 0) + return second; +#endif + return first; +} + +static void pin_current_for_role(int consumer) { + int cpu = affinity_cpu_for_role(consumer); + if (cpu < 0) { + ATOMIC_ADD(&errors, 1); + return; + } + + cpu_set_t set; + CPU_ZERO(&set); + CPU_SET(cpu, &set); + if (pthread_setaffinity_np(pthread_self(), sizeof(set), &set) != 0) + ATOMIC_ADD(&errors, 1); +} +#else +static void pin_current_for_role(int consumer) { + (void)consumer; +} +#endif + THREAD_FN addq(void *data) { struct lfq_ctx *ctx = (struct lfq_ctx *)data; + pin_current_for_role(0); uint32_t producer = (uint32_t)(ATOMIC_ADD(&producer_id_next, 1) - 1); uint64_t local_added = 0; uint64_t local_sum = 0; @@ -77,6 +130,7 @@ THREAD_FN addq(void *data) { THREAD_FN delq(void *data) { struct lfq_ctx *ctx = (struct lfq_ctx *)data; + pin_current_for_role(1); int tid = ATOMIC_ADD(&consumer_tid_next, 1) - 1; uint64_t local_deleted = 0; uint64_t local_sum = 0; @@ -118,6 +172,10 @@ THREAD_FN delq(void *data) { } int main(void) { +#if defined(__linux__) && defined(LFQ_AFFINITY_MODE) + printf("affinity stress mode=%d (1=same CPU, 2=producer/consumer split)\n", + LFQ_AFFINITY_MODE); +#endif struct lfq_ctx ctx; if (lfq_init(&ctx, MAX_CONSUMER) != 0) { fprintf(stderr, "lfq_init failed\n"); diff --git a/test_scheduler.c b/test_scheduler.c new file mode 100644 index 0000000..dea9115 --- /dev/null +++ b/test_scheduler.c @@ -0,0 +1,334 @@ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include + +#define LFQ_TEST_HOOKS 1 +#include "lfq.h" + +#define CHECK(expr) do { if (!(expr)) { fprintf(stderr, "CHECK failed at %s:%d: %s\n", __FILE__, __LINE__, #expr); exit(1); } } while (0) + +struct gate { + pthread_mutex_t mu; + pthread_cond_t cv; + enum lfq_test_hook_point point; + int hit; + int release; +}; + +static int first_allowed_cpu(void) { + cpu_set_t set; + CPU_ZERO(&set); + CHECK(sched_getaffinity(0, sizeof(set), &set) == 0); + for (int cpu = 0; cpu < CPU_SETSIZE; ++cpu) + if (CPU_ISSET(cpu, &set)) + return cpu; + return -1; +} + +static int second_allowed_cpu(int first) { + cpu_set_t set; + CPU_ZERO(&set); + CHECK(sched_getaffinity(0, sizeof(set), &set) == 0); + for (int cpu = 0; cpu < CPU_SETSIZE; ++cpu) + if (cpu != first && CPU_ISSET(cpu, &set)) + return cpu; + return first; +} + +static void pin_current(int cpu) { + cpu_set_t set; + CPU_ZERO(&set); + CPU_SET(cpu, &set); + CHECK(pthread_setaffinity_np(pthread_self(), sizeof(set), &set) == 0); +} + +static void gate_init(struct gate *g, enum lfq_test_hook_point point) { + CHECK(pthread_mutex_init(&g->mu, NULL) == 0); + CHECK(pthread_cond_init(&g->cv, NULL) == 0); + g->point = point; + g->hit = 0; + g->release = 0; +} + +static void gate_destroy(struct gate *g) { + CHECK(pthread_cond_destroy(&g->cv) == 0); + CHECK(pthread_mutex_destroy(&g->mu) == 0); +} + +static void gate_wait_hit(struct gate *g) { + struct timespec deadline; + CHECK(clock_gettime(CLOCK_REALTIME, &deadline) == 0); + deadline.tv_sec += 5; + + CHECK(pthread_mutex_lock(&g->mu) == 0); + while (!g->hit) { + int rc = pthread_cond_timedwait(&g->cv, &g->mu, &deadline); + if (rc == ETIMEDOUT) { + fprintf(stderr, "timed out waiting for hook %d\n", (int)g->point); + exit(2); + } + CHECK(rc == 0); + } + CHECK(pthread_mutex_unlock(&g->mu) == 0); +} + +static void gate_release(struct gate *g) { + CHECK(pthread_mutex_lock(&g->mu) == 0); + g->release = 1; + CHECK(pthread_cond_broadcast(&g->cv) == 0); + CHECK(pthread_mutex_unlock(&g->mu) == 0); +} + +static void scheduler_hook(enum lfq_test_hook_point point, + struct lfq_ctx *ctx, + struct lfq_node *first, + struct lfq_node *second, + void *arg) { + (void)ctx; + (void)first; + (void)second; + struct gate *g = (struct gate *)arg; + + CHECK(pthread_mutex_lock(&g->mu) == 0); + if (point == g->point && !g->hit) { + g->hit = 1; + CHECK(pthread_cond_broadcast(&g->cv) == 0); + while (!g->release) + CHECK(pthread_cond_wait(&g->cv, &g->mu) == 0); + } + CHECK(pthread_mutex_unlock(&g->mu) == 0); +} + +struct enq_args { + struct lfq_ctx *ctx; + uintptr_t value; + int cpu; + int rc; +}; + +static void *enqueue_thread(void *raw) { + struct enq_args *a = (struct enq_args *)raw; + pin_current(a->cpu); + a->rc = lfq_enqueue(a->ctx, (void *)a->value); + return NULL; +} + +struct deq_args { + struct lfq_ctx *ctx; + int tid; + int cpu; + int rc; + void *out; +}; + +static void *dequeue_thread(void *raw) { + struct deq_args *a = (struct deq_args *)raw; + pin_current(a->cpu); + a->out = NULL; + a->rc = lfq_try_dequeue_tid(a->ctx, a->tid, &a->out); + return NULL; +} + +struct clean_args { + struct lfq_ctx *ctx; + int cpu; + int rc; +}; + +static void *clean_thread(void *raw) { + struct clean_args *a = (struct clean_args *)raw; + pin_current(a->cpu); + a->rc = lfq_clean(a->ctx); + return NULL; +} + +static void test_enqueue_completion_visible(int cpu) { + struct lfq_ctx ctx; + struct gate gate; + pthread_t p1; + void *out = NULL; + + CHECK(lfq_init(&ctx, 4) == 0); + gate_init(&gate, LFQ_TEST_ENQ_AFTER_PROTECT_TAIL); + lfq_test_set_hook(scheduler_hook, &gate); + + struct enq_args a = {&ctx, 1, cpu, -999}; + CHECK(pthread_create(&p1, NULL, enqueue_thread, &a) == 0); + gate_wait_hit(&gate); + + /* P1 is parked with the old tail protected. P2 must still complete. */ + CHECK(lfq_enqueue(&ctx, (void *)(uintptr_t)2) == 0); + + /* A completed P2 enqueue must already be visible to a later consumer. */ + CHECK(lfq_try_dequeue_tid(&ctx, 0, &out) == 1); + CHECK((uintptr_t)out == 2); + + gate_release(&gate); + CHECK(pthread_join(p1, NULL) == 0); + CHECK(a.rc == 0); + + CHECK(lfq_try_dequeue_tid(&ctx, 0, &out) == 1); + CHECK((uintptr_t)out == 1); + CHECK(lfq_try_dequeue_tid(&ctx, 0, &out) == 0); + + lfq_test_set_hook(NULL, NULL); + gate_destroy(&gate); + CHECK(lfq_clean(&ctx) == 0); + printf("schedule: completed enqueue visibility PASS\n"); +} + +static void test_consumer_hazard_survives_reclaim(int cpu) { + struct lfq_ctx ctx; + struct gate gate; + pthread_t c1; + void *out = NULL; + + CHECK(lfq_init(&ctx, 4) == 0); + CHECK(lfq_enqueue(&ctx, (void *)(uintptr_t)11) == 0); + CHECK(lfq_enqueue(&ctx, (void *)(uintptr_t)22) == 0); + + gate_init(&gate, LFQ_TEST_DEQ_AFTER_PROTECT_HEAD); + lfq_test_set_hook(scheduler_hook, &gate); + + struct deq_args a = {&ctx, 0, cpu, -999, NULL}; + CHECK(pthread_create(&c1, NULL, dequeue_thread, &a) == 0); + gate_wait_hit(&gate); + + /* C1 protects the dummy head. C2 removes it and retires it. */ + CHECK(lfq_try_dequeue_tid(&ctx, 1, &out) == 1); + CHECK((uintptr_t)out == 11); + + /* A forced scan must not reclaim C1's protected old head. */ + lfq_test_force_reclaim(&ctx); + CHECK(lfg_count_freelist(&ctx) >= 1); + + gate_release(&gate); + CHECK(pthread_join(c1, NULL) == 0); + CHECK(a.rc == 1); + CHECK((uintptr_t)a.out == 22); + + lfq_test_set_hook(NULL, NULL); + gate_destroy(&gate); + + lfq_test_force_reclaim(&ctx); + CHECK(lfq_clean(&ctx) == 0); + printf("schedule: consumer hazard vs reclaim PASS\n"); +} + +static void test_producer_stale_tail_survives_reclaim(int cpu) { + struct lfq_ctx ctx; + struct gate gate; + pthread_t p1; + void *out = NULL; + + CHECK(lfq_init(&ctx, 4) == 0); + + gate_init(&gate, LFQ_TEST_ENQ_AFTER_PROTECT_TAIL); + lfq_test_set_hook(scheduler_hook, &gate); + + struct enq_args a = {&ctx, 101, cpu, -999}; + CHECK(pthread_create(&p1, NULL, enqueue_thread, &a) == 0); + gate_wait_hit(&gate); + + /* P1 holds a stale local dummy tail. Let P2 advance the queue. */ + CHECK(lfq_enqueue(&ctx, (void *)(uintptr_t)202) == 0); + CHECK(lfq_try_dequeue_tid(&ctx, 0, &out) == 1); + CHECK((uintptr_t)out == 202); + + /* The retired dummy must remain alive because P1 still hazards it. */ + lfq_test_force_reclaim(&ctx); + CHECK(lfg_count_freelist(&ctx) >= 1); + + gate_release(&gate); + CHECK(pthread_join(p1, NULL) == 0); + CHECK(a.rc == 0); + + CHECK(lfq_try_dequeue_tid(&ctx, 0, &out) == 1); + CHECK((uintptr_t)out == 101); + + lfq_test_set_hook(NULL, NULL); + gate_destroy(&gate); + + lfq_test_force_reclaim(&ctx); + CHECK(lfq_clean(&ctx) == 0); + printf("schedule: producer stale-tail hazard PASS\n"); +} + +static void test_cleanup_rejects_active_operation(int cpu) { + struct lfq_ctx ctx; + struct gate gate; + pthread_t p1; + void *out = NULL; + + CHECK(lfq_init(&ctx, 2) == 0); + gate_init(&gate, LFQ_TEST_ENQ_AFTER_PROTECT_TAIL); + lfq_test_set_hook(scheduler_hook, &gate); + + struct enq_args a = {&ctx, 31337, cpu, -999}; + CHECK(pthread_create(&p1, NULL, enqueue_thread, &a) == 0); + gate_wait_hit(&gate); + + CHECK(lfq_clean(&ctx) == -EBUSY); + + gate_release(&gate); + CHECK(pthread_join(p1, NULL) == 0); + CHECK(a.rc == 0); + + CHECK(lfq_try_dequeue_tid(&ctx, 0, &out) == 1); + CHECK((uintptr_t)out == 31337); + + lfq_test_set_hook(NULL, NULL); + gate_destroy(&gate); + CHECK(lfq_clean(&ctx) == 0); + printf("schedule: cleanup vs active operation PASS\n"); +} + +static void test_clean_gate_blocks_new_operations(int cpu) { + struct lfq_ctx ctx; + struct gate gate; + pthread_t cleaner; + void *out = NULL; + + CHECK(lfq_init(&ctx, 2) == 0); + gate_init(&gate, LFQ_TEST_CLEAN_AFTER_GATE); + lfq_test_set_hook(scheduler_hook, &gate); + + struct clean_args a = {&ctx, cpu, -999}; + CHECK(pthread_create(&cleaner, NULL, clean_thread, &a) == 0); + gate_wait_hit(&gate); + + /* Cleanup owns the lifecycle gate, so no operation may enter now. */ + CHECK(lfq_enqueue(&ctx, (void *)(uintptr_t)1) == -EBUSY); + CHECK(lfq_try_dequeue(&ctx, &out) == -EBUSY); + + gate_release(&gate); + CHECK(pthread_join(cleaner, NULL) == 0); + CHECK(a.rc == 0); + + lfq_test_set_hook(NULL, NULL); + gate_destroy(&gate); + printf("schedule: cleanup gate blocks new operations PASS\n"); +} + +int main(void) { + int cpu0 = first_allowed_cpu(); + int cpu1 = second_allowed_cpu(cpu0); + + /* Pin main and all deterministic actors to one CPU: real OS preemption. */ + pin_current(cpu0); + printf("deterministic scheduler tests: same_cpu=%d alternate_cpu=%d\n", cpu0, cpu1); + + test_enqueue_completion_visible(cpu0); + test_consumer_hazard_survives_reclaim(cpu0); + test_producer_stale_tail_survives_reclaim(cpu0); + test_cleanup_rejects_active_operation(cpu0); + test_clean_gate_blocks_new_operations(cpu0); + + printf("deterministic scheduler tests PASS\n"); + return 0; +} From 37c25897b0c9e320049af7a0db1c4e9558569292 Mon Sep 17 00:00:00 2001 From: kautism Date: Sat, 26 Sep 2026 16:35:53 +0800 Subject: [PATCH 6/6] polish-scheduler-stress-tests --- README.md | 13 ++++++++++++- test_multithread.c | 14 ++++++++++---- 2 files changed, 22 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index e64eec3..26cdfa2 100755 --- a/README.md +++ b/README.md @@ -151,7 +151,18 @@ cleanup of non-empty queues, and post-clean behavior. `test_multithread.c` runs MPMC producer/consumer matrices and verifies both item count and a -cross-thread checksum. +cross-thread checksum. Dedicated affinity builds also force all producers and +consumers onto one CPU (heavy preemption/time-slicing) and split producers vs. +consumers across two CPUs (cache-line migration / SMP ordering pressure). + +`test_scheduler.c` +uses test-only hook points plus condition variables to force exact interleavings. +All deterministic actors are pinned to the same allowed CPU so the test parks a +thread at a known lock-free algorithm step and lets another thread run on that +same CPU. Regression cases cover completed-enqueue visibility, consumer hazard +protection during forced reclamation, stale producer tails, cleanup racing with +an active operation, and the cleanup lifecycle gate. These tests do not rely on +`sched_yield()` luck. `test_aba.c` stresses rapid node retirement/reuse. It is paired with sanitizer CI; a passing diff --git a/test_multithread.c b/test_multithread.c index 3ade639..d68ffe1 100644 --- a/test_multithread.c +++ b/test_multithread.c @@ -51,16 +51,20 @@ static int affinity_cpu_for_role(int consumer) { return -1; int first = -1; +#if LFQ_AFFINITY_MODE == 2 int second = -1; +#endif for (int cpu = 0; cpu < CPU_SETSIZE; ++cpu) { if (!CPU_ISSET(cpu, &allowed)) continue; - if (first < 0) + if (first < 0) { first = cpu; - else { - second = cpu; - break; + continue; } +#if LFQ_AFFINITY_MODE == 2 + second = cpu; +#endif + break; } if (first < 0) @@ -69,6 +73,8 @@ static int affinity_cpu_for_role(int consumer) { #if LFQ_AFFINITY_MODE == 2 if (consumer && second >= 0) return second; +#else + (void)consumer; #endif return first; }