Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions src/foundation/lock_registry.c
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@ struct cbm_lock_registry {
cbm_lock_registry_abort_failure_t test_abort_failure;
bool test_abort_failure_armed;
atomic_uint_fast64_t test_condition_wait_calls;
atomic_uint_fast64_t test_waiter_enqueue_calls;
atomic_size_t test_condition_waiters_now;
struct cbm_lock_registry *next_live;
struct cbm_lock_registry *next_retired;
Expand Down Expand Up @@ -228,6 +229,7 @@ static void lock_registry_waiter_push(cbm_lock_registry_t *registry, lock_regist
}
entry->waiter_tail = waiter;
registry->waiter_count++;
(void)atomic_fetch_add_explicit(&registry->test_waiter_enqueue_calls, 1, memory_order_relaxed);
}

static bool lock_registry_waiter_remove(cbm_lock_registry_t *registry, lock_registry_entry_t *entry,
Expand Down Expand Up @@ -914,6 +916,7 @@ cbm_private_file_lock_status_t cbm_lock_registry_free(cbm_lock_registry_t **regi
registry->test_abort_failure = 0;
registry->test_abort_failure_armed = false;
atomic_store_explicit(&registry->test_condition_wait_calls, 0, memory_order_relaxed);
atomic_store_explicit(&registry->test_waiter_enqueue_calls, 0, memory_order_relaxed);
atomic_store_explicit(&registry->test_condition_waiters_now, 0, memory_order_relaxed);
registry->next_retired = lock_registry_retired;
lock_registry_retired = registry;
Expand Down Expand Up @@ -987,6 +990,15 @@ uint64_t cbm_lock_registry_condition_wait_call_count_for_test(const cbm_lock_reg
: 0;
}

/* Enqueue is monotonic, so a test can observe that a waiter joined the queue
* after the fact instead of catching a queue depth that the waiter's own
* deadline erases again. */
uint64_t cbm_lock_registry_waiter_enqueue_count_for_test(const cbm_lock_registry_t *registry) {
return registry
? atomic_load_explicit(&registry->test_waiter_enqueue_calls, memory_order_relaxed)
: 0;
}

size_t cbm_lock_registry_condition_waiter_count_for_test(const cbm_lock_registry_t *registry) {
return registry
? atomic_load_explicit(&registry->test_condition_waiters_now, memory_order_relaxed)
Expand Down
1 change: 1 addition & 0 deletions src/foundation/lock_registry_internal.h
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ bool cbm_lock_registry_is_retired_for_test(const cbm_lock_registry_t *registry);
size_t cbm_lock_registry_attempting_waiter_count_for_test(cbm_lock_registry_t *registry);
uint64_t cbm_lock_registry_condition_wait_call_count_for_test(const cbm_lock_registry_t *registry);
size_t cbm_lock_registry_condition_waiter_count_for_test(const cbm_lock_registry_t *registry);
uint64_t cbm_lock_registry_waiter_enqueue_count_for_test(const cbm_lock_registry_t *registry);

typedef enum {
CBM_LOCK_REGISTRY_RELEASE_RW = 1,
Expand Down
45 changes: 24 additions & 21 deletions tests/test_lock_registry.c
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ enum {
LOCK_REGISTRY_STRESS_THREADS = 8,
LOCK_REGISTRY_STRESS_ITERATIONS = 160,
LOCK_REGISTRY_PARKING_WAITERS = 64,
LOCK_REGISTRY_DEADLINE_WAIT_MS = 200,
};

typedef struct {
Expand Down Expand Up @@ -1020,7 +1021,8 @@ typedef struct {
atomic_bool ready;
atomic_bool go;
atomic_bool finished;
uint64_t deadline_ms;
uint64_t wait_ms;
uint64_t started_ms;
uint64_t returned_ms;
cbm_private_file_lock_status_t status;
cbm_lock_lease_t *lease;
Expand All @@ -1033,9 +1035,14 @@ static void *lock_registry_deadline_waiter_run(void *opaque) {
while (!atomic_load_explicit(&waiter->go, memory_order_acquire)) {
lock_registry_test_yield();
}
waiter->status =
cbm_lock_registry_acquire(waiter->registry, "absolute-deadline", CBM_PRIVATE_FILE_LOCK_EX,
waiter->deadline_ms, &waiter->cancel_token, &waiter->lease);
/* The absolute deadline is anchored here, in the thread it bounds, right
* before the call it bounds. An anchor taken by the observer is already
* running while this thread is still waiting to be scheduled, so on a
* loaded host it can expire before the acquire even starts. */
waiter->started_ms = cbm_now_ms();
waiter->status = cbm_lock_registry_acquire(
waiter->registry, "absolute-deadline", CBM_PRIVATE_FILE_LOCK_EX,
waiter->started_ms + waiter->wait_ms, &waiter->cancel_token, &waiter->lease);
waiter->returned_ms = cbm_now_ms();
atomic_store_explicit(&waiter->finished, true, memory_order_release);
return NULL;
Expand Down Expand Up @@ -1066,7 +1073,7 @@ TEST(lock_registry_absolute_deadline_survives_repeated_wakes) {
bool head_started = holder_status == CBM_PRIVATE_FILE_LOCK_OK &&
cbm_thread_create(&head_thread, 0, lock_registry_waiter_run, &head) == 0;
bool head_attempting = false;
uint64_t head_deadline = cbm_now_ms() + 500;
uint64_t head_deadline = cbm_now_ms() + LOCK_REGISTRY_TEST_TIMEOUT_MS;
while (head_started && cbm_now_ms() < head_deadline) {
head_attempting = cbm_lock_registry_waiter_count(fixture.registry) == 1 &&
cbm_lock_registry_attempting_waiter_count_for_test(fixture.registry) == 1;
Expand All @@ -1077,6 +1084,7 @@ TEST(lock_registry_absolute_deadline_survives_repeated_wakes) {
}

lock_registry_deadline_waiter_t tail = {.registry = fixture.registry,
.wait_ms = LOCK_REGISTRY_DEADLINE_WAIT_MS,
.status = CBM_PRIVATE_FILE_LOCK_IO};
atomic_init(&tail.cancel_token, false);
atomic_init(&tail.ready, false);
Expand All @@ -1086,31 +1094,20 @@ TEST(lock_registry_absolute_deadline_survives_repeated_wakes) {
bool tail_started =
head_attempting &&
cbm_thread_create(&tail_thread, 0, lock_registry_deadline_waiter_run, &tail) == 0;
uint64_t ready_deadline = cbm_now_ms() + 500;
uint64_t ready_deadline = cbm_now_ms() + LOCK_REGISTRY_TEST_TIMEOUT_MS;
while (tail_started && !atomic_load_explicit(&tail.ready, memory_order_acquire) &&
cbm_now_ms() < ready_deadline) {
lock_registry_test_yield();
}
bool tail_ready = tail_started && atomic_load_explicit(&tail.ready, memory_order_acquire);
uint64_t deadline_start = cbm_now_ms();
tail.deadline_ms = deadline_start + 200;
uint64_t enqueues_before_tail =
cbm_lock_registry_waiter_enqueue_count_for_test(fixture.registry);
atomic_store_explicit(&tail.go, true, memory_order_release);

bool tail_queued = false;
uint64_t queue_deadline = deadline_start + 100;
while (tail_ready && cbm_now_ms() < queue_deadline) {
tail_queued = cbm_lock_registry_waiter_count(fixture.registry) == 2 &&
cbm_lock_registry_attempting_waiter_count_for_test(fixture.registry) == 1;
if (tail_queued) {
break;
}
lock_registry_test_yield();
}

cbm_lock_cancel_token_t unrelated_token;
atomic_init(&unrelated_token, false);
bool broadcasts_ok = true;
uint64_t observe_deadline = deadline_start + 600;
uint64_t observe_deadline = cbm_now_ms() + LOCK_REGISTRY_TEST_TIMEOUT_MS;
while (tail_ready && !atomic_load_explicit(&tail.finished, memory_order_acquire) &&
cbm_now_ms() < observe_deadline) {
broadcasts_ok = cbm_lock_registry_request_cancel(fixture.registry, &unrelated_token) ==
Expand All @@ -1120,7 +1117,13 @@ TEST(lock_registry_absolute_deadline_survives_repeated_wakes) {
}
bool returned_at_deadline =
tail_ready && atomic_load_explicit(&tail.finished, memory_order_acquire);
uint64_t elapsed_ms = returned_at_deadline ? tail.returned_ms - deadline_start : UINT64_MAX;
/* The enqueue count only ever grows, so this stays true once the tail has
* joined the queue behind the attempting head. Reading it after the tail
* has returned removes the window the observer used to have to catch. */
uint64_t enqueues_after_tail =
cbm_lock_registry_waiter_enqueue_count_for_test(fixture.registry);
bool tail_queued = returned_at_deadline && enqueues_after_tail > enqueues_before_tail;
uint64_t elapsed_ms = returned_at_deadline ? tail.returned_ms - tail.started_ms : UINT64_MAX;

if (!returned_at_deadline && tail_started) {
(void)cbm_lock_registry_request_cancel(fixture.registry, &tail.cancel_token);
Expand Down
Loading