From 4ef78b26d54876d7697e995d02f7085eed1cee16 Mon Sep 17 00:00:00 2001 From: wangyi Date: Sat, 28 Mar 2026 23:11:42 +0800 Subject: [PATCH] server/transport: add key TTL expire race test add debug log for test_kv_mt.c to observe the occasional abort in CI --- server/test/test_kv_mt.c | 22 ++-- server/test/test_transport.c | 198 +++++++++++++++++++++++++++++++++++ 2 files changed, 213 insertions(+), 7 deletions(-) diff --git a/server/test/test_kv_mt.c b/server/test/test_kv_mt.c index f31feb6..9ecd7f8 100644 --- a/server/test/test_kv_mt.c +++ b/server/test/test_kv_mt.c @@ -205,25 +205,33 @@ static void *test_kv_verify(void *arg) int ret = priskv_get_key(kv, tkv->key, tkv->keylen, &value_in_kv, &tkv->valuelen_in_kv, &keynode); if (ret != PRISKV_RESP_STATUS_OK || !keynode) { - printf("TEST KV: verify KV [%d] status [%s] [FAILED]\n", i, - priskv_resp_status_str(ret)); + printf("TEST KV: verify status FAILED at line %d [thd=%u idx=%u] ret=%s keylen=%u keynode=%p\n", + __LINE__, thdid, i, priskv_resp_status_str(ret), tkv->keylen, keynode); assert(0); } if (value_in_kv != tkv->value_in_kv) { - printf("TEST KV: verify KV [%d] value pointer [%p] against [%p] [FAILED]\n", i, - value_in_kv, tkv->value_in_kv); + printf("TEST KV: verify pointer FAILED at line %d [thd=%u idx=%u] value_in_kv=%p expected=%p valuelen_in_kv=%u saved_valuelen=%u\n", + __LINE__, thdid, i, value_in_kv, tkv->value_in_kv, + tkv->valuelen_in_kv, tkv->valuelen); assert(0); } if (tkv->valuelen != tkv->valuelen_in_kv) { - printf("TEST KV: verify KV [%d] value length [%d] against [%d] [FAILED]\n", i, - tkv->valuelen, tkv->valuelen_in_kv); + printf("TEST KV: verify length FAILED at line %d [thd=%u idx=%u] valuelen=%u valuelen_in_kv=%u\n", + __LINE__, thdid, i, tkv->valuelen, tkv->valuelen_in_kv); assert(0); } if (memcmp(value_in_kv, tkv->value, tkv->valuelen)) { - printf("TEST KV: verify KV [%d] value compare [FAILED]\n", i); + uint32_t dump_len = tkv->valuelen < 16 ? tkv->valuelen : 16; + printf("TEST KV: verify memcmp FAILED at line %d [thd=%u idx=%u] len=%u dump_len=%u\n", + __LINE__, thdid, i, tkv->valuelen, dump_len); + printf("TEST KV: first bytes (expected/actual):"); + for (uint32_t b = 0; b < dump_len; b++) { + printf(" %02x/%02x", tkv->value[b], value_in_kv[b]); + } + printf("\n"); assert(0); } diff --git a/server/test/test_transport.c b/server/test/test_transport.c index 80f52c2..84c199d 100644 --- a/server/test/test_transport.c +++ b/server/test/test_transport.c @@ -27,6 +27,7 @@ #include "priskv-protocol-helper.h" #include "priskv-utils.h" #include "../transport/transport.h" +#include "priskv-threads.h" #define MAX_KEY_LENGTH 64 #define VALUE_BLOCK_SIZE 1024 @@ -97,6 +98,41 @@ static void do_mock_req_with_flags(priskv_transport_conn *conn, uint16_t cmd, vo priskv_transport_handle_recv(conn, req, sizeof(priskv_request) + payload_len); } +/* Build ALLOC with key TTL (ms) and submit */ +static void build_and_submit_alloc_with_timeout(priskv_transport_conn *conn, const char *key, + uint16_t keylen, uint32_t alloc_len, + uint64_t key_ttl_ms) +{ + size_t req_size = sizeof(priskv_request) + keylen; + uint8_t *req_buf = alloca(req_size); + priskv_request *req = (priskv_request *)req_buf; + memset(req, 0, req_size); + req->command = htobe16(PRISKV_COMMAND_ALLOC); + req->alloc_length = htobe32(alloc_len); + req->key_length = htobe16(keylen); + req->timeout = htobe64(key_ttl_ms); + memcpy((uint8_t *)req + sizeof(priskv_request), key, keylen); + priskv_transport_handle_recv(conn, req, (uint16_t)req_size); +} + +/* Build SEAL with optional PIN and per-request pin_ttl_ms */ +static void build_and_submit_seal_with_ttl(priskv_transport_conn *conn, uint64_t token_host, + uint32_t flags, uint64_t pin_ttl_ms) +{ + const size_t req_size = sizeof(priskv_request) + sizeof(uint64_t); + uint8_t req_buf[req_size]; + priskv_request *req = (priskv_request *)req_buf; + memset(req, 0, req_size); + req->command = htobe16(PRISKV_COMMAND_SEAL); + req->flags = htobe32(flags); + req->nsgl = htobe16(0); + req->key_length = htobe16(sizeof(uint64_t)); + req->pin_ttl_ms = htobe64(pin_ttl_ms); + uint64_t be_token = htobe64(token_host); + memcpy((uint8_t *)req + sizeof(priskv_request), &be_token, sizeof(uint64_t)); + priskv_transport_handle_recv(conn, req, (uint16_t)req_size); +} + static void test_kv_transport_permissions(void *kv) { priskv_transport_driver mock_driver = { @@ -453,6 +489,166 @@ static void test_kv_transport_alloc_seal_pin_acquire_release_unpin_combo(void *k g_transport_driver = old_driver; } +/* --- Key TTL race: ALLOC (short key TTL) + SEAL, then concurrent ACQUIRE/RELEASE + * against the expire routine. Extended to operate on a set of keys and to run + * for a configurable duration to strengthen foreground/background races. --- */ +typedef struct ttl_race_worker_arg { + priskv_transport_conn *conn; + const char **keys; + const uint16_t *keylens; + int nkeys; + useconds_t sleep_us; + int duration_sec; + int *acq_ok; + int *acq_nosuch; +} ttl_race_worker_arg; + +static void *ttl_race_worker(void *arg) +{ + ttl_race_worker_arg *a = (ttl_race_worker_arg *)arg; + struct timeval start, now; + gettimeofday(&start, NULL); + + for (;;) { + gettimeofday(&now, NULL); + long elapsed_ms = (now.tv_sec - start.tv_sec) * 1000L + + (now.tv_usec - start.tv_usec) / 1000L; + if (elapsed_ms >= (long)a->duration_sec * 1000L) { + break; + } + /* Randomly pick one key from the configured key set for this attempt. */ + int idx = (a->nkeys > 1) ? (int)(random() % a->nkeys) : 0; + const char *key = a->keys[idx]; + uint16_t keylen = a->keylens[idx]; + + build_and_submit_req(a->conn, PRISKV_COMMAND_ACQUIRE, 0, key, keylen); + if (tls_status == PRISKV_RESP_STATUS_OK && tls_token != 0) { + __sync_fetch_and_add(a->acq_ok, 1); + uint64_t be_token = htobe64(tls_token); + build_and_submit_req(a->conn, PRISKV_COMMAND_RELEASE, 0, &be_token, sizeof(uint64_t)); + } else if (tls_status == PRISKV_RESP_STATUS_NO_SUCH_KEY) { + __sync_fetch_and_add(a->acq_nosuch, 1); + /* Re-seed this key with a fresh short TTL so that foreground and + * background cleanup continue to race over the full duration. */ + last_status = -1; + last_token = 0; + build_and_submit_alloc_with_timeout(a->conn, key, keylen, 128, 500 /* ms */); + if (tls_status == PRISKV_RESP_STATUS_OK && tls_token != 0) { + build_and_submit_seal_with_ttl(a->conn, tls_token, 0, 0); + } + } + usleep(a->sleep_us); + } + return NULL; +} + +static void test_kv_transport_key_ttl_expire_race(void *kv, int nthreads, int duration_sec) +{ + /* Mock driver */ + priskv_transport_driver mock_driver = { + .name = "mock", + .send_response = mock_send_response, + .request_key_off = mock_request_key_off, + .request_key = mock_request_key, + .recv_req = mock_recv_req, + }; + priskv_transport_driver *old_driver = g_transport_driver; + g_transport_driver = &mock_driver; + + /* Transport connection */ + priskv_transport_conn conn = (priskv_transport_conn){0}; + conn.kv = kv; + conn.conn_cap.max_key_length = MAX_KEY_LENGTH; + conn.conn_cap.max_sgl = 8; + pthread_spin_init(&conn.lock, PTHREAD_PROCESS_PRIVATE); + + /* Start expire routine with 1s interval. */ + priskv_threadpool *tp = priskv_threadpool_create("ttl_race", 1, 1, 0); + assert(tp); + priskv_thread *bgthread = priskv_threadpool_get_bgthread(tp, 0); + assert(bgthread); + priskv_set_expire_routine_interval(kv, 1); + priskv_expire_routine(bgthread, kv); + + /* Prepare a small set of keys sharing the same short TTL to exercise + * concurrent ACQUIRE/RELEASE and key TTL expiration across multiple + * buckets/keys for the requested duration. We also track expire stats to + * ensure the background routine, not just foreground ACQUIRE, performs key + * cleanup. */ + static const char *ttl_keys[] = { + "ttl_race_key_0", + "ttl_race_key_1", + "ttl_race_key_2", + "ttl_race_key_3", + "ttl_race_key_4", + "ttl_race_key_5", + "ttl_race_key_6", + "ttl_race_key_7", + }; + const int nkeys = (int)(sizeof(ttl_keys) / sizeof(ttl_keys[0])); + uint16_t keylens[nkeys]; + uint64_t expire_before = priskv_get_expire_kv_count(kv); + + for (int i = 0; i < nkeys; i++) { + keylens[i] = (uint16_t)(strlen(ttl_keys[i]) + 1); + + /* ALLOC with short key TTL=500ms for each key. */ + last_status = -1; + last_token = 0; + build_and_submit_alloc_with_timeout(&conn, ttl_keys[i], keylens[i], 128, 500 /* ms */); + assert(last_status == PRISKV_RESP_STATUS_OK && last_token != 0); + + /* SEAL (no PIN) for each key. */ + build_and_submit_seal_with_ttl(&conn, last_token, 0, 0); + assert(tls_status == PRISKV_RESP_STATUS_OK); + } + + /* Multi-threaded ACQUIRE/RELEASE with background expiration, operating on + * the key set above. */ + int acq_ok = 0, acq_nosuch = 0; + pthread_t *ths = calloc((size_t)nthreads, sizeof(pthread_t)); + ttl_race_worker_arg *args = calloc((size_t)nthreads, sizeof(ttl_race_worker_arg)); + for (int i = 0; i < nthreads; i++) { + args[i].conn = &conn; + args[i].keys = ttl_keys; + args[i].keylens = keylens; + args[i].nkeys = nkeys; + /* Use a short sleep to generate dense ACQUIRE/RELEASE traffic over the + * configured duration, creating more chances to race with the expire + * routine. */ + args[i].sleep_us = 1000; /* 1ms interval */ + args[i].duration_sec = duration_sec; + args[i].acq_ok = &acq_ok; + args[i].acq_nosuch = &acq_nosuch; + pthread_create(&ths[i], NULL, ttl_race_worker, &args[i]); + } + for (int i = 0; i < nthreads; i++) pthread_join(ths[i], NULL); + free(ths); + free(args); + + if (!(acq_ok > 0 && acq_nosuch > 0)) { + printf("TEST TRANSPORT: KEY TTL RACE expected both OK and NO_SUCH_KEY, got ok=%d nosuch=%d [FAILED]\n", + acq_ok, acq_nosuch); + assert(0); + } + uint64_t expire_after = priskv_get_expire_kv_count(kv); + uint64_t expire_delta = expire_after - expire_before; + /* Background expire routine may or may not win the race in this pattern. + * Foreground ACQUIRE-based cleanup is already covered by ok/nosuch checks + * above, so treat expire_kv_count as best-effort signal only. */ + if (expire_delta == 0) { + printf("TEST TRANSPORT: KEY TTL RACE ok=%d nosuch=%d expire_delta=%lu (no background eviction observed) [OK]\n", + acq_ok, acq_nosuch, expire_delta); + } else { + printf("TEST TRANSPORT: KEY TTL RACE ok=%d nosuch=%d expire_delta=%lu [OK]\n", + acq_ok, acq_nosuch, expire_delta); + } + + priskv_transport_token_cleanup(&conn); + pthread_spin_destroy(&conn.lock); + g_transport_driver = old_driver; +} + /* --- DROP behavior & permission tests --- */ static void test_kv_transport_drop_behavior(void *kv) { @@ -1120,6 +1316,8 @@ int main(int argc, char **argv) test_kv_transport_alloc_token_add_fail(kv); test_kv_transport_pin_on_seal(kv); test_kv_transport_pin_and_unpin(kv); + /* TTL race tests (transport layer): */ + test_kv_transport_key_ttl_expire_race(kv, combo_nthreads, 60); test_kv_transport_alloc_seal_pin_acquire_release_unpin_combo(kv, combo_nthreads, combo_iters); test_kv_transport_unpin_no_such_key(kv); test_kv_transport_concurrent_seal_pin(kv);