From 66aa838332a93c2ff59cbfbe23728c22b2e4ea30 Mon Sep 17 00:00:00 2001 From: Jjateen Gundesha Date: Mon, 17 Aug 2026 10:35:23 +0530 Subject: [PATCH 1/2] fix: exclude concurrent seqlock writers - a bare fetch_add lets two writers drive the sequence even mid-write, so reads tear - a mutex cannot fix it: the slow paths write other processes' slots - take the slot with a CAS, even to odd - add a GPU-free two-writer regression test Signed-off-by: Jjateen Gundesha --- src/multiprocess/multiprocess_memory_limit.c | 99 +++++++++-- test/CMakeLists.txt | 31 +++- test/test_seqlock_writer_exclusion.c | 169 +++++++++++++++++++ 3 files changed, 276 insertions(+), 23 deletions(-) create mode 100644 test/test_seqlock_writer_exclusion.c diff --git a/src/multiprocess/multiprocess_memory_limit.c b/src/multiprocess/multiprocess_memory_limit.c index 4114ed35..9c65104c 100755 --- a/src/multiprocess/multiprocess_memory_limit.c +++ b/src/multiprocess/multiprocess_memory_limit.c @@ -7,6 +7,7 @@ #include #include #include +#include #include #include #include @@ -448,6 +449,72 @@ uint64_t nvml_get_device_memory_usage(const int dev) { return usage; } +/* Seqlock writers must exclude each other, or the sequence counter can be + * driven back to even while another writer's stores are still in flight and a + * reader will accept a torn snapshot. + * + * A process-local mutex is not sufficient here: the slow paths below write + * another process's slot, so two writers to the same slot can live in + * different processes. Exclusion therefore has to sit in the shared region + * itself, which is what the CAS provides. + * + * Only the writer that wins the even->odd transition enters; a waiting writer + * never publishes an odd value, so the window in which readers see a write in + * progress is no longer than it is today. + */ +static inline void seqlock_cpu_relax(void) { +#if defined(__x86_64__) || defined(__i386__) + __asm__ __volatile__("pause" ::: "memory"); +#elif defined(__aarch64__) + __asm__ __volatile__("yield" ::: "memory"); +#else + __asm__ __volatile__("" ::: "memory"); /* barrier only, no pause opcode */ +#endif +} + +/* The critical section is a handful of atomic adds with no syscall in it, so a + * real wait is nanoseconds. Escalate anyway so a waiter never sits on a core: + * pause, then yield, then short sleeps. */ +#define SEQLOCK_SPIN_LIMIT 1000u /* pause only */ +#define SEQLOCK_YIELD_LIMIT 11000u /* then sched_yield */ +#define SEQLOCK_SLEEP_LIMIT 31000u /* then 100us naps, indefinitely */ + +static inline void seqlock_write_begin(shrreg_proc_slot_t* slot) { + uint64_t seq; + unsigned spins = 0; + for (;;) { + seq = atomic_load_explicit(&slot->seqlock, memory_order_relaxed); + if (seq & 1) { /* another writer holds the slot */ + if (spins < SEQLOCK_SPIN_LIMIT) { + seqlock_cpu_relax(); + } else if (spins < SEQLOCK_YIELD_LIMIT) { + sched_yield(); + } else if (spins < SEQLOCK_SLEEP_LIMIT) { + usleep(100); + } else { + /* Keep waiting. Reclaiming the slot here is not safe: a live + * writer can be paused for longer than any threshold, and + * forcing the counter would either publish an even sequence + * while that writer is still storing, or turn an already + * released even sequence back to odd with no holder. Proving + * the holder is dead needs an owner field this structure does + * not have. */ + usleep(100); + } + spins++; + continue; + } + if (atomic_compare_exchange_weak_explicit( + &slot->seqlock, &seq, seq + 1, + memory_order_acq_rel, memory_order_relaxed)) + return; + } +} + +static inline void seqlock_write_end(shrreg_proc_slot_t* slot) { + atomic_fetch_add_explicit(&slot->seqlock, 1, memory_order_release); +} + // Lock-free memory add using atomics with seqlock for consistent reads int add_gpu_device_memory_usage(int32_t pid, int cudadev, size_t usage, int type) { LOG_INFO("add_gpu_device_memory_lockfree:%d %d->%d %lu", pid, cudadev, cuda_to_nvml_map(cudadev), usage); @@ -459,8 +526,8 @@ int add_gpu_device_memory_usage(int32_t pid, int cudadev, size_t usage, int type if (pid == getpid() && region_info.my_slot != NULL) { shrreg_proc_slot_t* slot = region_info.my_slot; - // Seqlock protocol: increment to odd (write in progress) - atomic_fetch_add_explicit(&slot->seqlock, 1, memory_order_release); + // Seqlock protocol: take the slot for writing (even -> odd) + seqlock_write_begin(slot); // Perform updates with release semantics for visibility atomic_fetch_add_explicit(&slot->used[dev].total, usage, memory_order_release); @@ -476,8 +543,8 @@ int add_gpu_device_memory_usage(int32_t pid, int cudadev, size_t usage, int type break; } - // Seqlock protocol: increment to even (write complete) - atomic_fetch_add_explicit(&slot->seqlock, 1, memory_order_release); + // Seqlock protocol: release the slot (odd -> even) + seqlock_write_end(slot); LOG_INFO("gpu_device_memory_added_lockfree:%d %d %lu", pid, dev, usage); return 0; @@ -491,8 +558,8 @@ int add_gpu_device_memory_usage(int32_t pid, int cudadev, size_t usage, int type if (slot_pid == pid) { shrreg_proc_slot_t* slot = ®ion_info.shared_region->procs[i]; - // Seqlock protocol: increment to odd (write in progress) - atomic_fetch_add_explicit(&slot->seqlock, 1, memory_order_release); + // Seqlock protocol: take the slot for writing (even -> odd) + seqlock_write_begin(slot); // Perform updates atomic_fetch_add_explicit(&slot->used[dev].total, usage, memory_order_release); @@ -508,8 +575,8 @@ int add_gpu_device_memory_usage(int32_t pid, int cudadev, size_t usage, int type break; } - // Seqlock protocol: increment to even (write complete) - atomic_fetch_add_explicit(&slot->seqlock, 1, memory_order_release); + // Seqlock protocol: release the slot (odd -> even) + seqlock_write_end(slot); LOG_INFO("gpu_device_memory_added_lockfree:%d %d %lu", pid, dev, usage); return 0; @@ -530,8 +597,8 @@ int rm_gpu_device_memory_usage(int32_t pid, int cudadev, size_t usage, int type) if (pid == getpid() && region_info.my_slot != NULL) { shrreg_proc_slot_t* slot = region_info.my_slot; - // Seqlock protocol: increment to odd (write in progress) - atomic_fetch_add_explicit(&slot->seqlock, 1, memory_order_release); + // Seqlock protocol: take the slot for writing (even -> odd) + seqlock_write_begin(slot); // Perform updates with release semantics atomic_fetch_sub_explicit(&slot->used[dev].total, usage, memory_order_release); @@ -547,8 +614,8 @@ int rm_gpu_device_memory_usage(int32_t pid, int cudadev, size_t usage, int type) break; } - // Seqlock protocol: increment to even (write complete) - atomic_fetch_add_explicit(&slot->seqlock, 1, memory_order_release); + // Seqlock protocol: release the slot (odd -> even) + seqlock_write_end(slot); uint64_t new_total = atomic_load_explicit(&slot->used[dev].total, memory_order_acquire); LOG_INFO("after delete_lockfree:%lu", new_total); @@ -563,8 +630,8 @@ int rm_gpu_device_memory_usage(int32_t pid, int cudadev, size_t usage, int type) if (slot_pid == pid) { shrreg_proc_slot_t* slot = ®ion_info.shared_region->procs[i]; - // Seqlock protocol: increment to odd (write in progress) - atomic_fetch_add_explicit(&slot->seqlock, 1, memory_order_release); + // Seqlock protocol: take the slot for writing (even -> odd) + seqlock_write_begin(slot); // Perform updates atomic_fetch_sub_explicit(&slot->used[dev].total, usage, memory_order_release); @@ -580,8 +647,8 @@ int rm_gpu_device_memory_usage(int32_t pid, int cudadev, size_t usage, int type) break; } - // Seqlock protocol: increment to even (write complete) - atomic_fetch_add_explicit(&slot->seqlock, 1, memory_order_release); + // Seqlock protocol: release the slot (odd -> even) + seqlock_write_end(slot); uint64_t new_total = atomic_load_explicit(&slot->used[dev].total, memory_order_acquire); LOG_INFO("after delete_lockfree:%lu", new_total); diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index ae39f322..0fa9e2b9 100755 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -12,15 +12,24 @@ foreach(TEST_SCRIPT ${TEST_SCRIPTS}) get_filename_component(TEST_TARGET_NAME ${RELATIVE_TEST_PATH} NAME_WE) set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/${TEST_TARGET_DIR}) - if (TEST_TARGET_NAME STREQUAL "test_postinit_owner_death") + if (TEST_TARGET_NAME STREQUAL "test_postinit_owner_death" OR + TEST_TARGET_NAME STREQUAL "test_seqlock_writer_exclusion") # Build this focused regression test with the production shared-region # implementation. It does not invoke any CUDA/NVML entry point at # runtime. Section garbage collection drops unrelated GPU-facing # production functions. - add_executable(${TEST_TARGET_NAME} - ${TEST_SCRIPT} - ${CMAKE_CURRENT_SOURCE_DIR}/../src/multiprocess/multiprocess_memory_limit.c - ${CMAKE_CURRENT_SOURCE_DIR}/../src/log_utils.c) + if (TEST_TARGET_NAME STREQUAL "test_seqlock_writer_exclusion") + # includes the translation unit directly to observe the sequence + # counter, so it must not also be linked in + add_executable(${TEST_TARGET_NAME} + ${TEST_SCRIPT} + ${CMAKE_CURRENT_SOURCE_DIR}/../src/log_utils.c) + else() + add_executable(${TEST_TARGET_NAME} + ${TEST_SCRIPT} + ${CMAKE_CURRENT_SOURCE_DIR}/../src/multiprocess/multiprocess_memory_limit.c + ${CMAKE_CURRENT_SOURCE_DIR}/../src/log_utils.c) + endif() target_compile_definitions(${TEST_TARGET_NAME} PRIVATE _GNU_SOURCE) target_compile_options(${TEST_TARGET_NAME} PRIVATE @@ -34,7 +43,8 @@ foreach(TEST_SCRIPT ${TEST_SCRIPTS}) endif() list(APPEND TEST_TARGET_NAMES_LIST ${TEST_TARGET_NAME}) - if (TEST_TARGET_NAME STREQUAL "test_postinit_owner_death") + if (TEST_TARGET_NAME STREQUAL "test_postinit_owner_death" OR + TEST_TARGET_NAME STREQUAL "test_seqlock_writer_exclusion") target_link_libraries(${TEST_TARGET_NAME} -lrt -lpthread) else() target_link_libraries(${TEST_TARGET_NAME} -lrt -lpthread @@ -50,7 +60,14 @@ set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}) add_test(NAME postinit_owner_death COMMAND test_postinit_owner_death) -set_tests_properties(postinit_owner_death PROPERTIES TIMEOUT 20) +set_tests_properties(postinit_owner_death PROPERTIES TIMEOUT 20 + RESOURCE_LOCK shared_region) + +add_test(NAME seqlock_writer_exclusion + COMMAND test_seqlock_writer_exclusion) +set_tests_properties(seqlock_writer_exclusion PROPERTIES TIMEOUT 60 + SKIP_RETURN_CODE 77 + RESOURCE_LOCK shared_region) add_custom_target(python_test ALL diff --git a/test/test_seqlock_writer_exclusion.c b/test/test_seqlock_writer_exclusion.c new file mode 100644 index 00000000..56270906 --- /dev/null +++ b/test/test_seqlock_writer_exclusion.c @@ -0,0 +1,169 @@ +/* + * GPU-free regression test for seqlock writer-writer exclusion. + * + * get_gpu_memory_usage() reads each process slot under a seqlock. That + * protocol is only sound when writers exclude one another: if two writers + * interleave their sequence increments, the counter can return to an even + * value while stores are still in flight and a reader will accept a torn + * snapshot. + * + * Two writers hammer the same slot through the public entry points while a + * reader samples the counter. The test fails if it ever observes an even + * sequence number together with a slot total that is not a multiple of the + * per-write quantum, which is only reachable when a second writer has driven + * the counter even mid-write. + * + * Nothing here calls a CUDA or NVML entry point. + */ +#include +#include +#include +#include +#include +#include +#include +#include + +/* region_info is static, so reaching the slot to observe the sequence counter + * means compiling against the translation unit itself. The CMake target for + * this test therefore does not list the .c file separately. */ +#include "multiprocess/multiprocess_memory_limit.c" // NOLINT(build/include) + +/* The production mapping lives in the NVML-facing utilization watcher, which a + * GPU-free test must not pull in. Slot accounting here uses a single device, + * so the identity mapping is the correct stand-in. */ +unsigned int cuda_to_nvml_map(unsigned int cudadev) { return cudadev; } + +#define QUANTUM 4096u /* every write moves total by this much */ +#define ITERATIONS 200000 +#define DEV 0 + +static _Atomic int stop_readers; +static _Atomic int readers_ready; +static _Atomic int64_t torn_reads; +static _Atomic int64_t odd_observations; + +static void *writer_thread(void *arg) { + (void)arg; + for (int64_t i = 0; i < ITERATIONS; i++) { + add_gpu_device_memory_usage(getpid(), DEV, QUANTUM, 2); + rm_gpu_device_memory_usage(getpid(), DEV, QUANTUM, 2); + } + return NULL; +} + +/* + * Sample the slot the way get_gpu_memory_usage() does. A snapshot taken + * while the sequence number is even must be self-consistent, so the running + * total has to be a whole number of quanta. + */ +static void *reader_thread(void *arg) { + shrreg_proc_slot_t *slot = region_info.my_slot; + (void)arg; + + atomic_fetch_add_explicit(&readers_ready, 1, memory_order_release); + while (!atomic_load_explicit(&stop_readers, memory_order_relaxed)) { + uint64_t seq1 = atomic_load_explicit(&slot->seqlock, memory_order_acquire); + if (seq1 & 1) { + atomic_fetch_add_explicit(&odd_observations, 1, memory_order_relaxed); + continue; + } + uint64_t total = atomic_load_explicit(&slot->used[DEV].total, + memory_order_acquire); + uint64_t data = atomic_load_explicit(&slot->used[DEV].data_size, + memory_order_acquire); + atomic_thread_fence(memory_order_acquire); + uint64_t seq2 = atomic_load_explicit(&slot->seqlock, memory_order_acquire); + + if (seq1 != seq2) + continue; /* a write landed, resample */ + + /* Accepted as a consistent snapshot. total and data_size move + * together in one critical section, so they must agree. */ + if (total % QUANTUM != 0 || data % QUANTUM != 0 || total != data) + atomic_fetch_add_explicit(&torn_reads, 1, memory_order_relaxed); + } + return NULL; +} + +int main(void) { + pthread_t w1, w2, r1, r2; + + /* overwrite, so an inherited value cannot change what this test measures */ + if (setenv("CUDA_DEVICE_MEMORY_LIMIT", "0", 1) != 0) { + perror("FAIL: setenv"); + return 1; + } + ensure_initialized(); + + if (region_info.my_slot == NULL) { + fprintf(stderr, "SKIP: no slot allocated for this process\n"); + return 77; /* ctest "skipped" convention */ + } + + /* Without this check a failed create leaves an uninitialised pthread_t for + * join, and the assertions below would still see zero torn reads and report + * PASS having exercised no concurrency at all. */ + if (pthread_create(&r1, NULL, reader_thread, NULL) != 0 || + pthread_create(&r2, NULL, reader_thread, NULL) != 0) { + fprintf(stderr, "FAIL: pthread_create failed\n"); + return 1; + } + /* Hold the writers until both readers are sampling. Otherwise the scheduler + * can run the writers to completion first and the test passes without ever + * exercising the torn-snapshot path. */ + while (atomic_load_explicit(&readers_ready, memory_order_acquire) < 2) + sched_yield(); + + if (pthread_create(&w1, NULL, writer_thread, NULL) != 0 || + pthread_create(&w2, NULL, writer_thread, NULL) != 0) { + fprintf(stderr, "FAIL: pthread_create failed\n"); + return 1; + } + + pthread_join(w1, NULL); + pthread_join(w2, NULL); + atomic_store_explicit(&stop_readers, 1, memory_order_relaxed); + pthread_join(r1, NULL); + pthread_join(r2, NULL); + + int64_t torn = atomic_load(&torn_reads); + int64_t odd = atomic_load(&odd_observations); + shrreg_proc_slot_t *slot = region_info.my_slot; + uint64_t final_seq = atomic_load(&slot->seqlock); + uint64_t final_total = atomic_load(&slot->used[DEV].total); + + printf("writers=2 readers=2 iterations=%d\n", ITERATIONS); + printf("odd sequence observations : %" PRId64 "\n", odd); + printf("torn reads : %" PRId64 "\n", torn); + printf("final seqlock : %" PRIu64 " (%s)\n", final_seq, + (final_seq & 1) ? "ODD - a writer did not release" : "even"); + printf("final total : %" PRIu64 " (expected 0)\n", final_total); + + if (torn != 0) { + fprintf(stderr, "FAIL: %" PRId64 " torn reads accepted under an even sequence\n", + torn); + return 1; + } + /* two writers, ITERATIONS rounds, an add and a remove each, two + * transitions per critical section */ + const uint64_t expected_seq = 2ULL * ITERATIONS * 2ULL * 2ULL; + if (final_seq != expected_seq) { + fprintf(stderr, "FAIL: sequence is %" PRIu64 ", expected %" PRIu64 "\n", + final_seq, expected_seq); + return 1; + } + if (final_total != 0) { + fprintf(stderr, "FAIL: add/rm pairs did not balance, lost updates\n"); + return 1; + } + /* Zero means no reader ever sampled while a write was in progress, so the + * run proves nothing about tearing whatever the other counters say. */ + if (odd == 0) { + fprintf(stderr, "FAIL: no reader observed an active writer, " + "readers and writers did not overlap\n"); + return 1; + } + printf("PASS\n"); + return 0; +} From 6a8ee6f96d0c2c6b58315e18c0c81ed7cb66e549 Mon Sep 17 00:00:00 2001 From: Jjateen Gundesha Date: Thu, 20 Aug 2026 10:59:20 +0530 Subject: [PATCH 2/2] fix: reclaim a seqlock left odd by a dead writer - record the holder pid in reserved slot space - reclaim only if that pid is gone - acquire is enough on the claiming CAS - drop the sleep limit, both arms were equal - reader reuses the writer cpu relax helper - add a cross-process regression test Signed-off-by: Jjateen Gundesha --- src/multiprocess/multiprocess_memory_limit.c | 94 ++++++--- src/multiprocess/multiprocess_memory_limit.h | 8 +- test/CMakeLists.txt | 15 +- test/test_seqlock_multiprocess.c | 200 +++++++++++++++++++ 4 files changed, 284 insertions(+), 33 deletions(-) create mode 100644 test/test_seqlock_multiprocess.c diff --git a/src/multiprocess/multiprocess_memory_limit.c b/src/multiprocess/multiprocess_memory_limit.c index 9c65104c..8f9177f2 100755 --- a/src/multiprocess/multiprocess_memory_limit.c +++ b/src/multiprocess/multiprocess_memory_limit.c @@ -280,6 +280,17 @@ size_t get_gpu_memory_monitor(const int dev) { } // Lock-free memory usage aggregation with seqlock for consistent snapshots +static inline void seqlock_cpu_relax(void) { +#if defined(__x86_64__) || defined(__i386__) + __asm__ __volatile__("pause" ::: "memory"); +#elif defined(__aarch64__) + __asm__ __volatile__("yield" ::: "memory"); +#else + __asm__ __volatile__("" ::: "memory"); /* barrier only, no pause opcode */ +#endif +} + + size_t get_gpu_memory_usage(const int dev) { LOG_INFO("get_gpu_memory_usage_lockfree dev=%d", dev); ensure_initialized(); @@ -305,12 +316,9 @@ size_t get_gpu_memory_usage(const int dev) { while (seq1 & 1) { // Exponential backoff to reduce contention if (retry_count < 5) { - // First 5 retries: just CPU pause (fast path) - #if defined(__x86_64__) || defined(__i386__) - __asm__ __volatile__("pause" ::: "memory"); - #elif defined(__aarch64__) - __asm__ __volatile__("yield" ::: "memory"); - #endif + // First 5 retries: just CPU pause (fast path). Shared with + // the writer so both sides get the portable fallback. + seqlock_cpu_relax(); } else if (retry_count < 20) { // Next 15 retries: 1μs delay usleep(1); @@ -462,22 +470,49 @@ uint64_t nvml_get_device_memory_usage(const int dev) { * never publishes an odd value, so the window in which readers see a write in * progress is no longer than it is today. */ -static inline void seqlock_cpu_relax(void) { -#if defined(__x86_64__) || defined(__i386__) - __asm__ __volatile__("pause" ::: "memory"); -#elif defined(__aarch64__) - __asm__ __volatile__("yield" ::: "memory"); -#else - __asm__ __volatile__("" ::: "memory"); /* barrier only, no pause opcode */ -#endif -} - /* The critical section is a handful of atomic adds with no syscall in it, so a * real wait is nanoseconds. Escalate anyway so a waiter never sits on a core: * pause, then yield, then short sleeps. */ -#define SEQLOCK_SPIN_LIMIT 1000u /* pause only */ -#define SEQLOCK_YIELD_LIMIT 11000u /* then sched_yield */ -#define SEQLOCK_SLEEP_LIMIT 31000u /* then 100us naps, indefinitely */ +#define SEQLOCK_SPIN_LIMIT 1000u /* pause only */ +#define SEQLOCK_YIELD_LIMIT 11000u /* then sched_yield */ +#define SEQLOCK_RECLAIM_AFTER 31000u /* then start trying to recover */ +#define SEQLOCK_RECLAIM_EVERY 10000u /* and retry about once a second */ + +/* A holder killed between begin and end leaves the sequence odd for good, which + * blocks every later writer and every reader of that slot. Recovery has to prove + * the holder is gone rather than infer it from a timeout, because a live writer + * can be descheduled for longer than any threshold we could pick. + * + * The writer therefore records its pid in the slot while it holds the sequence, + * and this only resets the counter when that pid no longer exists. Taking + * lock_shrreg() serialises the reclaim so two waiters cannot both perform it. + * + * Known residual: a writer killed in the window between winning the CAS and + * storing its pid leaves the sequence odd with seq_owner still 0. That is + * treated as an unknown holder and never reclaimed, so the slot stays wedged as + * it would have before this change. Closing it means publishing the sequence + * and the owner in one atomic word, which changes how the reader interprets the + * counter. */ +static void seqlock_try_reclaim(shrreg_proc_slot_t* slot) { + lock_shrreg(); + uint64_t seq = atomic_load_explicit(&slot->seqlock, memory_order_acquire); + int32_t owner = atomic_load_explicit(&slot->seq_owner, memory_order_relaxed); + if ((seq & 1) && owner > 0 && kill(owner, 0) == -1 && errno == ESRCH) { + LOG_WARN("seqlock owner %d is gone, reclaiming slot", owner); + /* the dead writer stopped partway through, so this slot's counters no + * longer describe anything and are cleared with the sequence */ + int dev; + for (dev = 0; dev < CUDA_DEVICE_MAX_COUNT; dev++) { + atomic_store_explicit(&slot->used[dev].total, 0, memory_order_relaxed); + atomic_store_explicit(&slot->used[dev].context_size, 0, memory_order_relaxed); + atomic_store_explicit(&slot->used[dev].module_size, 0, memory_order_relaxed); + atomic_store_explicit(&slot->used[dev].data_size, 0, memory_order_relaxed); + } + atomic_store_explicit(&slot->seq_owner, 0, memory_order_relaxed); + atomic_store_explicit(&slot->seqlock, seq + 1, memory_order_release); + } + unlock_shrreg(); +} static inline void seqlock_write_begin(shrreg_proc_slot_t* slot) { uint64_t seq; @@ -489,29 +524,30 @@ static inline void seqlock_write_begin(shrreg_proc_slot_t* slot) { seqlock_cpu_relax(); } else if (spins < SEQLOCK_YIELD_LIMIT) { sched_yield(); - } else if (spins < SEQLOCK_SLEEP_LIMIT) { - usleep(100); } else { - /* Keep waiting. Reclaiming the slot here is not safe: a live - * writer can be paused for longer than any threshold, and - * forcing the counter would either publish an even sequence - * while that writer is still storing, or turn an already - * released even sequence back to odd with no holder. Proving - * the holder is dead needs an owner field this structure does - * not have. */ usleep(100); + if (spins >= SEQLOCK_RECLAIM_AFTER && + (spins - SEQLOCK_RECLAIM_AFTER) % SEQLOCK_RECLAIM_EVERY == 0) + seqlock_try_reclaim(slot); } spins++; continue; } + /* Acquire is all this needs: it stops the stores in the critical section + * being hoisted above the claim. Nothing is published here, and a reader + * that sees an odd sequence only backs off, so release would pair with + * nothing. seqlock_write_end carries the release. */ if (atomic_compare_exchange_weak_explicit( &slot->seqlock, &seq, seq + 1, - memory_order_acq_rel, memory_order_relaxed)) + memory_order_acquire, memory_order_relaxed)) { + atomic_store_explicit(&slot->seq_owner, getpid(), memory_order_relaxed); return; + } } } static inline void seqlock_write_end(shrreg_proc_slot_t* slot) { + atomic_store_explicit(&slot->seq_owner, 0, memory_order_relaxed); atomic_fetch_add_explicit(&slot->seqlock, 1, memory_order_release); } diff --git a/src/multiprocess/multiprocess_memory_limit.h b/src/multiprocess/multiprocess_memory_limit.h index 3a880400..2c95d0f9 100755 --- a/src/multiprocess/multiprocess_memory_limit.h +++ b/src/multiprocess/multiprocess_memory_limit.h @@ -87,7 +87,13 @@ typedef struct { device_util_t device_util[CUDA_DEVICE_MAX_COUNT]; _Atomic int32_t status; _Atomic uint64_t seqlock; // Sequence lock for consistent snapshots - uint64_t unused[2]; + // Pid of the writer currently holding seqlock, 0 when no one holds it. + // Taken from the space reserved by unused[], so sizeof this struct and the + // shared region layout are unchanged: a process built without it leaves the + // field 0 and the reclaim path simply declines to act. + _Atomic int32_t seq_owner; + int32_t seq_owner_pad; + uint64_t unused[1]; } shrreg_proc_slot_t; typedef char uuid[96]; diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 0fa9e2b9..d8caf342 100755 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -13,12 +13,14 @@ foreach(TEST_SCRIPT ${TEST_SCRIPTS}) set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/${TEST_TARGET_DIR}) if (TEST_TARGET_NAME STREQUAL "test_postinit_owner_death" OR - TEST_TARGET_NAME STREQUAL "test_seqlock_writer_exclusion") + TEST_TARGET_NAME STREQUAL "test_seqlock_writer_exclusion" OR + TEST_TARGET_NAME STREQUAL "test_seqlock_multiprocess") # Build this focused regression test with the production shared-region # implementation. It does not invoke any CUDA/NVML entry point at # runtime. Section garbage collection drops unrelated GPU-facing # production functions. - if (TEST_TARGET_NAME STREQUAL "test_seqlock_writer_exclusion") + if (TEST_TARGET_NAME STREQUAL "test_seqlock_writer_exclusion" OR + TEST_TARGET_NAME STREQUAL "test_seqlock_multiprocess") # includes the translation unit directly to observe the sequence # counter, so it must not also be linked in add_executable(${TEST_TARGET_NAME} @@ -44,7 +46,8 @@ foreach(TEST_SCRIPT ${TEST_SCRIPTS}) list(APPEND TEST_TARGET_NAMES_LIST ${TEST_TARGET_NAME}) if (TEST_TARGET_NAME STREQUAL "test_postinit_owner_death" OR - TEST_TARGET_NAME STREQUAL "test_seqlock_writer_exclusion") + TEST_TARGET_NAME STREQUAL "test_seqlock_writer_exclusion" OR + TEST_TARGET_NAME STREQUAL "test_seqlock_multiprocess") target_link_libraries(${TEST_TARGET_NAME} -lrt -lpthread) else() target_link_libraries(${TEST_TARGET_NAME} -lrt -lpthread @@ -69,6 +72,12 @@ set_tests_properties(seqlock_writer_exclusion PROPERTIES TIMEOUT 60 SKIP_RETURN_CODE 77 RESOURCE_LOCK shared_region) +add_test(NAME seqlock_multiprocess + COMMAND test_seqlock_multiprocess) +set_tests_properties(seqlock_multiprocess PROPERTIES TIMEOUT 120 + SKIP_RETURN_CODE 77 + RESOURCE_LOCK shared_region) + add_custom_target(python_test ALL COMMAND cp -r ${CMAKE_CURRENT_SOURCE_DIR}/python ${CMAKE_CURRENT_BINARY_DIR}) diff --git a/test/test_seqlock_multiprocess.c b/test/test_seqlock_multiprocess.c new file mode 100644 index 00000000..6a8a65a3 --- /dev/null +++ b/test/test_seqlock_multiprocess.c @@ -0,0 +1,200 @@ +/* + * Cross-process regression test for seqlock writer exclusion. + * + * test_seqlock_writer_exclusion covers two threads in one process, which only + * exercises the pid == getpid() fast path on the cached slot pointer. The + * reason the exclusion has to live in the shared region, rather than in a + * pthread mutex, is the slow path: add_gpu_device_memory_usage and + * rm_gpu_device_memory_usage look a slot up by pid and write another process's + * slot. A process-local mutex cannot exclude a writer in a different process. + * + * Every participant here writes every slot by pid, so all but one of its writes + * take the slow path, while sampling every slot the way get_gpu_memory_usage + * does. A snapshot accepted under an even sequence must have total and + * data_size in agreement, since both move inside one critical section. + * + * Nothing here calls a CUDA or NVML entry point. + */ +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "multiprocess/multiprocess_memory_limit.c" // NOLINT(build/include) + +/* both live in the NVML-facing utilization watcher, which a GPU-free test must + * not pull in */ +unsigned int cuda_to_nvml_map(unsigned int cudadev) { return cudadev; } +int setspec(void) { return 0; } + +#define PARTICIPANTS 4 /* the parent plus three children */ +#define QUANTUM 4096u +#define ITERATIONS 20000 +#define DEV 0 + +/* placed in an anonymous shared mapping before the fork, so every participant + * sees the same object */ +typedef struct { + _Atomic int32_t pid[PARTICIPANTS]; + _Atomic int64_t torn; + _Atomic int64_t overlaps; + _Atomic int ready; +} board_t; + +static board_t *board; + +static void publish(int index) { + atomic_store_explicit(&board->pid[index], getpid(), memory_order_release); +} + +static void wait_for_everyone(void) { + for (;;) { + int seen = 0; + for (int i = 0; i < PARTICIPANTS; i++) + if (atomic_load_explicit(&board->pid[i], memory_order_acquire) > 0) + seen++; + if (seen == PARTICIPANTS) + return; + sched_yield(); + } +} + +/* Sample every registered slot the way get_gpu_memory_usage does. */ +static void sample_all(void) { + shared_region_t *region = region_info.shared_region; + int proc_num = atomic_load_explicit(®ion->proc_num, memory_order_acquire); + for (int i = 0; i < proc_num; i++) { + shrreg_proc_slot_t *slot = ®ion->procs[i]; + uint64_t s1 = atomic_load_explicit(&slot->seqlock, memory_order_acquire); + if (s1 & 1) { + atomic_fetch_add_explicit(&board->overlaps, 1, memory_order_relaxed); + continue; + } + uint64_t total = atomic_load_explicit(&slot->used[DEV].total, + memory_order_acquire); + uint64_t data = atomic_load_explicit(&slot->used[DEV].data_size, + memory_order_acquire); + atomic_thread_fence(memory_order_acquire); + uint64_t s2 = atomic_load_explicit(&slot->seqlock, memory_order_acquire); + if (s1 != s2) + continue; + if (total != data) + atomic_fetch_add_explicit(&board->torn, 1, memory_order_relaxed); + } +} + +/* Write every slot by pid. Only the entry matching getpid() takes the fast + * path; the rest go through the slow path lookup, which is the case this test + * exists for. */ +static void hammer(void) { + for (int n = 0; n < ITERATIONS; n++) { + for (int i = 0; i < PARTICIPANTS; i++) { + int32_t target = atomic_load_explicit(&board->pid[i], + memory_order_relaxed); + if (target <= 0) + continue; + add_gpu_device_memory_usage(target, DEV, QUANTUM, 2); + rm_gpu_device_memory_usage(target, DEV, QUANTUM, 2); + } + if ((n & 0x3f) == 0) + sample_all(); + } +} + +int main(void) { + board = mmap(NULL, sizeof(board_t), PROT_READ | PROT_WRITE, + MAP_SHARED | MAP_ANONYMOUS, -1, 0); + if (board == MAP_FAILED) { + perror("FAIL: mmap"); + return 1; + } + if (setenv("CUDA_DEVICE_MEMORY_LIMIT", "0", 1) != 0) { + perror("FAIL: setenv"); + return 1; + } + + /* Fork before touching the region. Initialising first would leave every + * child sharing the parent's cached slot instead of registering its own. */ + pid_t kids[PARTICIPANTS - 1]; + for (int k = 0; k < PARTICIPANTS - 1; k++) { + kids[k] = fork(); + if (kids[k] < 0) { + perror("FAIL: fork"); + return 1; + } + if (kids[k] == 0) { + ensure_initialized(); + if (region_info.my_slot == NULL) + _exit(77); + publish(k + 1); + wait_for_everyone(); + hammer(); + _exit(0); + } + } + + ensure_initialized(); + if (region_info.my_slot == NULL) { + fprintf(stderr, "SKIP: no slot allocated\n"); + for (int k = 0; k < PARTICIPANTS - 1; k++) kill(kids[k], SIGKILL); + return 77; + } + publish(0); + wait_for_everyone(); + hammer(); + + int skipped = 0; + for (int k = 0; k < PARTICIPANTS - 1; k++) { + int st = 0; + waitpid(kids[k], &st, 0); + if (WIFEXITED(st) && WEXITSTATUS(st) == 77) { + skipped = 1; + } else if (!WIFEXITED(st) || WEXITSTATUS(st) != 0) { + fprintf(stderr, "FAIL: child %d did not exit cleanly (status %d)\n", + kids[k], st); + return 1; + } + } + if (skipped) { + fprintf(stderr, "SKIP: a participant could not register a slot\n"); + return 77; + } + + int64_t torn = atomic_load(&board->torn); + int64_t overlaps = atomic_load(&board->overlaps); + printf("participants=%d iterations=%d\n", PARTICIPANTS, ITERATIONS); + printf("writes in progress observed : %" PRId64 "\n", overlaps); + printf("torn snapshots : %" PRId64 "\n", torn); + + shared_region_t *region = region_info.shared_region; + int proc_num = atomic_load_explicit(®ion->proc_num, memory_order_acquire); + for (int i = 0; i < proc_num; i++) { + uint64_t seq = atomic_load(®ion->procs[i].seqlock); + uint64_t total = atomic_load(®ion->procs[i].used[DEV].total); + if (seq & 1) { + fprintf(stderr, "FAIL: slot %d left odd, a writer did not release\n", i); + return 1; + } + if (total != 0) { + fprintf(stderr, "FAIL: slot %d total is %" PRIu64 ", add/rm did not " + "balance\n", i, total); + return 1; + } + } + if (torn != 0) { + fprintf(stderr, "FAIL: %" PRId64 " torn snapshots across processes\n", torn); + return 1; + } + if (overlaps == 0) { + fprintf(stderr, "FAIL: never sampled during a write, so the run proves " + "nothing\n"); + return 1; + } + printf("PASS\n"); + return 0; +}