From 12810a7171073782cd3c1d56c5e08e35f3dccdcd Mon Sep 17 00:00:00 2001 From: junyao <1071307515@qq.com> Date: Tue, 25 Aug 2026 16:46:17 +0000 Subject: [PATCH 1/2] feat: Notify/Take, Incr, Expire, WatchAny on kindexpr ABI Rebase the durable queue and TTL ops onto main's kindexpr head (kvspaceClose). Char Set/Get keeps a non-empty body. --- CMakeLists.txt | 2 +- src/durable_abi.c | 314 +++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 310 insertions(+), 6 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index d736dea..afe42d5 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -21,7 +21,7 @@ set_target_properties(kvspace-c PROPERTIES # 无公共头：内部头在 src/（kvspace_shm.h、xvalue.h），权威 ABI 头在 kvspace 仓库。 target_link_libraries(kvspace-c - PUBLIC blockmalloc slotsboxmalloc + PUBLIC blockmalloc slotsboxmalloc pthread ) install(TARGETS kvspace-c diff --git a/src/durable_abi.c b/src/durable_abi.c index 0719726..ff79378 100644 --- a/src/durable_abi.c +++ b/src/durable_abi.c @@ -6,6 +6,8 @@ #include "kvspace_shm.h" #include "xvalue.h" +#include +#include #include #include #include @@ -30,6 +32,61 @@ static int parse_shm_path(const char *dsn, char *out, size_t osz) { return 0; } +#define EXP_CAP 256 +static struct { kvspace_t *kv; char key[512]; int64_t dead_ns; int used; } exp_tab[EXP_CAP]; +static pthread_mutex_t exp_mu = PTHREAD_MUTEX_INITIALIZER; + +static int64_t exp_now_ns(void) { + struct timespec t; + clock_gettime(CLOCK_MONOTONIC, &t); + return (int64_t)t.tv_sec * 1000000000LL + t.tv_nsec; +} + +static int exp_key_under(const char *key, const char *prefix) { + if (!prefix || !prefix[0] || strcmp(prefix, "/") == 0) return 1; + size_t n = strlen(prefix); + if (strncmp(key, prefix, n) != 0) return 0; + if (prefix[n - 1] == '/') return 1; + return key[n] == 0 || key[n] == '/' || key[n] == '.'; +} + +static void exp_forget(kvspace_t *kv, const char *key) { + if (!kv || !key) return; + pthread_mutex_lock(&exp_mu); + for (int i = 0; i < EXP_CAP; i++) { + if (exp_tab[i].used && exp_tab[i].kv == kv && strcmp(exp_tab[i].key, key) == 0) + exp_tab[i].used = 0; + } + pthread_mutex_unlock(&exp_mu); +} + +static void exp_forget_prefix(kvspace_t *kv, const char *prefix) { + if (!kv) return; + pthread_mutex_lock(&exp_mu); + for (int i = 0; i < EXP_CAP; i++) { + if (exp_tab[i].used && exp_tab[i].kv == kv && exp_key_under(exp_tab[i].key, prefix)) + exp_tab[i].used = 0; + } + pthread_mutex_unlock(&exp_mu); +} + +/* 0 live, 1 hidden, 2 reaped */ +static int exp_reap(kvspace_t *kv, const char *key) { + int64_t now = exp_now_ns(); + pthread_mutex_lock(&exp_mu); + int st = 0; + for (int i = 0; i < EXP_CAP; i++) { + if (!exp_tab[i].used || exp_tab[i].kv != kv || strcmp(exp_tab[i].key, key) != 0) continue; + if (now < exp_tab[i].dead_ns) { st = 1; break; } + exp_tab[i].used = 0; + st = 2; + kvspaceShmDel(kv, key); + break; + } + pthread_mutex_unlock(&exp_mu); + return st; +} + void *kvspaceConnect(const char *dsn) { char path[1024]; if (parse_shm_path(dsn, path, sizeof path) != 0) return NULL; @@ -37,7 +94,9 @@ void *kvspaceConnect(const char *dsn) { } void kvspaceClose(void *h) { - if (h) kvspaceShmClose((kvspace_t *)h); + if (!h) return; + exp_forget_prefix((kvspace_t *)h, "/"); + kvspaceShmClose((kvspace_t *)h); } void kvspaceBytesFree(uint8_t *p, uint32_t len) { @@ -50,6 +109,7 @@ int kvspaceSet(void *h, const char *const *keys, const uint8_t *vals, (void)err; (void)err_cap; uint32_t off = 0; for (uint32_t i = 0; i < n; i++) { + exp_forget((kvspace_t *)h, keys[i]); kvspaceShmSet((kvspace_t *)h, keys[i], vals + off, (int32_t)lens[i]); off += lens[i]; } @@ -57,6 +117,9 @@ int kvspaceSet(void *h, const char *const *keys, const uint8_t *vals, } int kvspaceGet(void *h, const char *key, uint8_t **out, uint32_t *out_len) { + if (exp_reap((kvspace_t *)h, key) == 2) { + *out = NULL; *out_len = 0; return 0; + } int32_t len; uint8_t *d = kvspaceShmGet((kvspace_t *)h, key, 1, &len); if (!d || len <= 0) { *out = NULL; *out_len = 0; return 0; } @@ -91,12 +154,16 @@ int kvspaceList(void *h, const char *prefix, int expand_ext, int resolve, int kvspaceDel(void *h, const char *const *keys, uint32_t nkeys, char *err, uint32_t err_cap) { (void)err; (void)err_cap; - for (uint32_t i = 0; i < nkeys; i++) kvspaceShmDel((kvspace_t *)h, keys[i]); + for (uint32_t i = 0; i < nkeys; i++) { + exp_forget((kvspace_t *)h, keys[i]); + kvspaceShmDel((kvspace_t *)h, keys[i]); + } return 0; } int kvspaceDelTree(void *h, const char *prefix, char *err, uint32_t err_cap) { (void)err; (void)err_cap; + exp_forget_prefix((kvspace_t *)h, prefix); return kvspaceShmDeltree((kvspace_t *)h, prefix); } @@ -117,6 +184,7 @@ int kvspaceRmindexExt(void *h, const char *path, char *err, uint32_t err_cap) { int kvspaceClear(void *h, char *err, uint32_t err_cap) { (void)err; (void)err_cap; + exp_forget_prefix((kvspace_t *)h, "/"); return kvspaceShmDeltree((kvspace_t *)h, "/"); } @@ -224,7 +292,9 @@ int kvspaceGetBatch(void *h, const char *prefix, const char *const *names, char key[2048]; snprintf(key, sizeof key, "%s%s", prefix, names[i]); int32_t len = 0; - uint8_t *d = kvspaceShmGet((kvspace_t *)h, key, 1, &len); + uint8_t *d = NULL; + if (exp_reap((kvspace_t *)h, key) != 2) + d = kvspaceShmGet((kvspace_t *)h, key, 1, &len); if (d && len > 0) total += (size_t)len; } uint8_t *buf = malloc(total); @@ -234,7 +304,9 @@ int kvspaceGetBatch(void *h, const char *prefix, const char *const *names, char key[2048]; snprintf(key, sizeof key, "%s%s", prefix, names[i]); int32_t len = 0; - uint8_t *d = kvspaceShmGet((kvspace_t *)h, key, 1, &len); + uint8_t *d = NULL; + if (exp_reap((kvspace_t *)h, key) != 2) + d = kvspaceShmGet((kvspace_t *)h, key, 1, &len); if (!d || len <= 0) len = 0; buf[off] = (uint8_t)(len & 0xFF); buf[off + 1] = (uint8_t)((len >> 8) & 0xFF); @@ -260,7 +332,9 @@ int kvspaceWatch(void *h, const char *key, const uint8_t *target, uint32_t targe clock_gettime(CLOCK_MONOTONIC, &t0); for (;;) { int32_t len = 0; - uint8_t *d = kvspaceShmGet((kvspace_t *)h, key, 1, &len); + uint8_t *d = NULL; + if (exp_reap((kvspace_t *)h, key) != 2) + d = kvspaceShmGet((kvspace_t *)h, key, 1, &len); if (d && (uint32_t)len == target_len && memcmp(d, target, target_len) == 0) { uint8_t *c = malloc((size_t)len); memcpy(c, d, (size_t)len); @@ -275,3 +349,233 @@ int kvspaceWatch(void *h, const char *key, const uint8_t *target, uint32_t targe usleep(1000); } } + +/* Durable Notify/Take queue. Outside the user-visible tree (Watch stays WatchValue). */ +#define NQ_PFX "/\xE2\x80\xA5notify" + +static pthread_mutex_t nq_mu = PTHREAD_MUTEX_INITIALIZER; + +static void nq_key(const char *key, char *out, size_t cap) { + snprintf(out, cap, "%s%s", NQ_PFX, key ? key : ""); +} + +static uint32_t nq_rd32(const uint8_t *p) { + return (uint32_t)p[0] | ((uint32_t)p[1] << 8) | ((uint32_t)p[2] << 16) | ((uint32_t)p[3] << 24); +} + +static void nq_wr32(uint8_t *p, uint32_t v) { + p[0] = (uint8_t)v; p[1] = (uint8_t)(v >> 8); p[2] = (uint8_t)(v >> 16); p[3] = (uint8_t)(v >> 24); +} + +static int nq_load(kvspace_t *kv, const char *qk, uint8_t **body, uint32_t *blen) { + int32_t len = 0; + uint8_t *d = kvspaceShmGet(kv, qk, 1, &len); + *body = NULL; *blen = 0; + if (!d || len <= 0) return 0; + xvalue_head_t h = kvspaceXvalueDecodeHead(d, len); + if (h.raw_len <= 0 || !h.raw) return 0; + uint8_t *c = malloc((size_t)h.raw_len); + if (!c) abort(); + memcpy(c, h.raw, (size_t)h.raw_len); + *body = c; *blen = (uint32_t)h.raw_len; + return 0; +} + +static int nq_save(kvspace_t *kv, const char *qk, const uint8_t *body, uint32_t blen) { + if (blen == 0) return kvspaceShmDel(kv, qk); + int32_t dims[1] = { (int32_t)blen }; + uint8_t *tlv = NULL; + int32_t n = kvspaceXvalueEncode(KVSPACE_KIND_UINT8, body, (int32_t)blen, dims, 1, &tlv); + if (n < 0 || !tlv) abort(); + int rc = kvspaceShmSet(kv, qk, tlv, n); + free(tlv); + return rc; +} + +int kvspaceNotify(void *h, const char *key, const uint8_t *val, uint32_t len, char *err, uint32_t err_cap) { + (void)err; (void)err_cap; + if (!h || !key || !val || len == 0) return 1; + char qk[2048]; + nq_key(key, qk, sizeof qk); + pthread_mutex_lock(&nq_mu); + uint8_t *body = NULL; uint32_t blen = 0; + nq_load((kvspace_t *)h, qk, &body, &blen); + uint8_t *nb = malloc((size_t)blen + 4 + len); + if (!nb) abort(); + if (blen) memcpy(nb, body, blen); + nq_wr32(nb + blen, len); + memcpy(nb + blen + 4, val, len); + int rc = nq_save((kvspace_t *)h, qk, nb, blen + 4 + len); + free(nb); free(body); + pthread_mutex_unlock(&nq_mu); + return rc == 0 ? 0 : 1; +} + +/* caller holds nq_mu. 1=popped, 0=empty */ +static int nq_try_pop(kvspace_t *kv, const char *key, uint8_t **out, uint32_t *out_len) { + char qk[2048]; + nq_key(key, qk, sizeof qk); + uint8_t *body = NULL; uint32_t blen = 0; + nq_load(kv, qk, &body, &blen); + if (blen >= 4) { + uint32_t fl = nq_rd32(body); + if (4 + fl <= blen) { + uint8_t *item = malloc(fl ? fl : 1); + if (!item) abort(); + if (fl) memcpy(item, body + 4, fl); + nq_save(kv, qk, body + 4 + fl, blen - 4 - fl); + free(body); + *out = item; *out_len = fl; + return 1; + } + } + free(body); + return 0; +} + +int kvspaceTake(void *h, const char *key, uint64_t timeout_ns, uint8_t **out, uint32_t *out_len) { + if (!out || !out_len) return 1; + *out = NULL; *out_len = 0; + if (!h || !key) return 1; + struct timespec t0, tn; + clock_gettime(CLOCK_MONOTONIC, &t0); + for (;;) { + pthread_mutex_lock(&nq_mu); + int hit = nq_try_pop((kvspace_t *)h, key, out, out_len); + pthread_mutex_unlock(&nq_mu); + if (hit) return 0; + clock_gettime(CLOCK_MONOTONIC, &tn); + uint64_t elapsed = (uint64_t)(tn.tv_sec - t0.tv_sec) * 1000000000ULL + + (uint64_t)(tn.tv_nsec - t0.tv_nsec); + if (elapsed >= timeout_ns) return 0; + usleep(1000); + } +} + +int kvspaceWatchAny(void *h, const char *const *keys, uint32_t nkeys, uint64_t timeout_ns, + uint8_t **out_key, uint32_t *out_key_len, uint8_t **out, uint32_t *out_len) { + if (!out_key || !out_key_len || !out || !out_len) return 1; + *out_key = NULL; *out_key_len = 0; *out = NULL; *out_len = 0; + if (!h || !keys || nkeys == 0) return 1; + struct timespec t0, tn; + clock_gettime(CLOCK_MONOTONIC, &t0); + for (;;) { + pthread_mutex_lock(&nq_mu); + for (uint32_t i = 0; i < nkeys; i++) { + if (!keys[i]) continue; + if (nq_try_pop((kvspace_t *)h, keys[i], out, out_len)) { + size_t kl = strlen(keys[i]); + uint8_t *kb = malloc(kl + 1); + if (!kb) abort(); + memcpy(kb, keys[i], kl + 1); + *out_key = kb; *out_key_len = (uint32_t)kl; + pthread_mutex_unlock(&nq_mu); + return 0; + } + } + pthread_mutex_unlock(&nq_mu); + clock_gettime(CLOCK_MONOTONIC, &tn); + uint64_t elapsed = (uint64_t)(tn.tv_sec - t0.tv_sec) * 1000000000ULL + + (uint64_t)(tn.tv_nsec - t0.tv_nsec); + if (elapsed >= timeout_ns) return 0; + usleep(1000); + } +} + +static pthread_mutex_t incr_mu = PTHREAD_MUTEX_INITIALIZER; + +static void incr_err(char *err, uint32_t err_cap, const char *msg) { + if (!err || err_cap == 0) return; + snprintf(err, err_cap, "%s", msg); +} + +int kvspaceIncr(void *h, const char *key, int64_t *out, char *err, uint32_t err_cap) { + if (!h || !key || !out) { incr_err(err, err_cap, "Incr: bad args"); return 1; } + *out = 0; + pthread_mutex_lock(&incr_mu); + exp_reap((kvspace_t *)h, key); + int32_t len = 0; + uint8_t *d = kvspaceShmGet((kvspace_t *)h, key, 1, &len); + int64_t n = 0; + if (d && len > 0) { + xvalue_head_t hd = kvspaceXvalueDecodeHead(d, len); + if (hd.kind_len < 5 || memcmp(hd.kind, "char/", 5) != 0 || !hd.raw || hd.raw_len <= 0) { + pthread_mutex_unlock(&incr_mu); + incr_err(err, err_cap, "Incr: counter is not a Char"); + return 1; + } + char *s = malloc((size_t)hd.raw_len + 1); + if (!s) abort(); + memcpy(s, hd.raw, (size_t)hd.raw_len); + s[hd.raw_len] = 0; + char *end = NULL; + n = strtoll(s, &end, 10); + int bad = end == s || (end && *end); + free(s); + if (bad) { + pthread_mutex_unlock(&incr_mu); + incr_err(err, err_cap, "Incr: unparsable counter"); + return 1; + } + } + if (n == INT64_MAX) { + pthread_mutex_unlock(&incr_mu); + incr_err(err, err_cap, "Incr: overflow"); + return 1; + } + n++; + char buf[32]; + snprintf(buf, sizeof buf, "%lld", (long long)n); + uint8_t *tlv = NULL; + int32_t tn = kvspaceXvalueNewCharUtf8(buf, &tlv); + if (tn < 0 || !tlv) abort(); + int rc = kvspaceShmSet((kvspace_t *)h, key, tlv, tn); + free(tlv); + pthread_mutex_unlock(&incr_mu); + if (rc != 0) { incr_err(err, err_cap, "Incr: set failed"); return 1; } + *out = n; + return 0; +} + +static int exp_valid_key(const char *key) { + if (!key || key[0] != '/') return 0; + size_t n = strlen(key); + if (n <= 1 || key[n - 1] == '/') return 0; + for (const char *p = key + 1; *p; ) { + const char *sl = strchr(p, '/'); + size_t seglen = sl ? (size_t)(sl - p) : strlen(p); + if (seglen == 0 || (seglen == 1 && p[0] == '.') || (seglen == 2 && p[0] == '.' && p[1] == '.')) + return 0; + p = sl ? sl + 1 : p + seglen; + } + return 1; +} + +int kvspaceExpire(void *h, const char *key, uint64_t ttl_ns, char *err, uint32_t err_cap) { + if (!h || !key) { incr_err(err, err_cap, "Expire: bad args"); return 1; } + if (ttl_ns == 0) { incr_err(err, err_cap, "Expire: ttl must be > 0"); return 1; } + if (!exp_valid_key(key)) { + incr_err(err, err_cap, key && key[0] && key[strlen(key) - 1] == '/' ? "Expire: directory" : "Expire: key is not an absolute path"); + return 1; + } + if (exp_reap((kvspace_t *)h, key) == 2) { incr_err(err, err_cap, "Expire: missing key"); return 1; } + int32_t len = 0; + uint8_t *d = kvspaceShmGet((kvspace_t *)h, key, 1, &len); + if (!d || len <= 0) { incr_err(err, err_cap, "Expire: missing key"); return 1; } + int64_t dead = exp_now_ns() + (int64_t)ttl_ns; + pthread_mutex_lock(&exp_mu); + int slot = -1; + for (int i = 0; i < EXP_CAP; i++) { + if (exp_tab[i].used && exp_tab[i].kv == (kvspace_t *)h && strcmp(exp_tab[i].key, key) == 0) { + slot = i; break; + } + if (slot < 0 && !exp_tab[i].used) slot = i; + } + if (slot < 0) { pthread_mutex_unlock(&exp_mu); abort(); } + exp_tab[slot].kv = (kvspace_t *)h; + snprintf(exp_tab[slot].key, sizeof exp_tab[slot].key, "%s", key); + exp_tab[slot].dead_ns = dead; + exp_tab[slot].used = 1; + pthread_mutex_unlock(&exp_mu); + return 0; +} From 395f61a49b57372b2a0d48e4e4c7a046d22c2e19 Mon Sep 17 00:00:00 2001 From: junyao <1071307515@qq.com> Date: Tue, 25 Aug 2026 16:54:09 +0000 Subject: [PATCH 2/2] fix: Expire delists immediately, Get holds value until TTL --- src/durable_abi.c | 85 +++++++++++++++++++++++++++++++++++++---------- 1 file changed, 67 insertions(+), 18 deletions(-) diff --git a/src/durable_abi.c b/src/durable_abi.c index ff79378..71cc958 100644 --- a/src/durable_abi.c +++ b/src/durable_abi.c @@ -33,7 +33,14 @@ static int parse_shm_path(const char *dsn, char *out, size_t osz) { } #define EXP_CAP 256 -static struct { kvspace_t *kv; char key[512]; int64_t dead_ns; int used; } exp_tab[EXP_CAP]; +static struct { + kvspace_t *kv; + char key[512]; + int64_t dead_ns; + uint8_t *val; + uint32_t vlen; + int used; +} exp_tab[EXP_CAP]; static pthread_mutex_t exp_mu = PTHREAD_MUTEX_INITIALIZER; static int64_t exp_now_ns(void) { @@ -50,12 +57,19 @@ static int exp_key_under(const char *key, const char *prefix) { return key[n] == 0 || key[n] == '/' || key[n] == '.'; } +static void exp_clear_slot(int i) { + free(exp_tab[i].val); + exp_tab[i].val = NULL; + exp_tab[i].vlen = 0; + exp_tab[i].used = 0; +} + static void exp_forget(kvspace_t *kv, const char *key) { if (!kv || !key) return; pthread_mutex_lock(&exp_mu); for (int i = 0; i < EXP_CAP; i++) { if (exp_tab[i].used && exp_tab[i].kv == kv && strcmp(exp_tab[i].key, key) == 0) - exp_tab[i].used = 0; + exp_clear_slot(i); } pthread_mutex_unlock(&exp_mu); } @@ -65,22 +79,31 @@ static void exp_forget_prefix(kvspace_t *kv, const char *prefix) { pthread_mutex_lock(&exp_mu); for (int i = 0; i < EXP_CAP; i++) { if (exp_tab[i].used && exp_tab[i].kv == kv && exp_key_under(exp_tab[i].key, prefix)) - exp_tab[i].used = 0; + exp_clear_slot(i); } pthread_mutex_unlock(&exp_mu); } -/* 0 live, 1 hidden, 2 reaped */ -static int exp_reap(kvspace_t *kv, const char *key) { +/* 0 not in table, 1 hidden-live (List gone, Get still has val), 2 reaped */ +static int exp_reap(kvspace_t *kv, const char *key, uint8_t **hold, uint32_t *hlen) { int64_t now = exp_now_ns(); pthread_mutex_lock(&exp_mu); int st = 0; for (int i = 0; i < EXP_CAP; i++) { if (!exp_tab[i].used || exp_tab[i].kv != kv || strcmp(exp_tab[i].key, key) != 0) continue; - if (now < exp_tab[i].dead_ns) { st = 1; break; } - exp_tab[i].used = 0; + if (now < exp_tab[i].dead_ns) { + st = 1; + if (hold && hlen && exp_tab[i].val) { + uint8_t *c = malloc(exp_tab[i].vlen ? exp_tab[i].vlen : 1); + if (!c) abort(); + if (exp_tab[i].vlen) memcpy(c, exp_tab[i].val, exp_tab[i].vlen); + *hold = c; + *hlen = exp_tab[i].vlen; + } + break; + } + exp_clear_slot(i); st = 2; - kvspaceShmDel(kv, key); break; } pthread_mutex_unlock(&exp_mu); @@ -117,9 +140,10 @@ int kvspaceSet(void *h, const char *const *keys, const uint8_t *vals, } int kvspaceGet(void *h, const char *key, uint8_t **out, uint32_t *out_len) { - if (exp_reap((kvspace_t *)h, key) == 2) { - *out = NULL; *out_len = 0; return 0; - } + uint8_t *held = NULL; uint32_t hl = 0; + int st = exp_reap((kvspace_t *)h, key, &held, &hl); + if (st == 2) { *out = NULL; *out_len = 0; return 0; } + if (st == 1) { *out = held; *out_len = hl; return 0; } int32_t len; uint8_t *d = kvspaceShmGet((kvspace_t *)h, key, 1, &len); if (!d || len <= 0) { *out = NULL; *out_len = 0; return 0; } @@ -293,7 +317,7 @@ int kvspaceGetBatch(void *h, const char *prefix, const char *const *names, snprintf(key, sizeof key, "%s%s", prefix, names[i]); int32_t len = 0; uint8_t *d = NULL; - if (exp_reap((kvspace_t *)h, key) != 2) + if (exp_reap((kvspace_t *)h, key, NULL, NULL) != 2) d = kvspaceShmGet((kvspace_t *)h, key, 1, &len); if (d && len > 0) total += (size_t)len; } @@ -305,7 +329,10 @@ int kvspaceGetBatch(void *h, const char *prefix, const char *const *names, snprintf(key, sizeof key, "%s%s", prefix, names[i]); int32_t len = 0; uint8_t *d = NULL; - if (exp_reap((kvspace_t *)h, key) != 2) + uint8_t *held = NULL; uint32_t hl = 0; + int st = exp_reap((kvspace_t *)h, key, &held, &hl); + if (st == 1) { d = held; len = (int32_t)hl; } + else if (st != 2) d = kvspaceShmGet((kvspace_t *)h, key, 1, &len); if (!d || len <= 0) len = 0; buf[off] = (uint8_t)(len & 0xFF); @@ -317,6 +344,7 @@ int kvspaceGetBatch(void *h, const char *prefix, const char *const *names, memcpy(buf + off, d, (size_t)len); off += (size_t)len; } + free(held); } *out = buf; *out_len = (uint32_t)off; @@ -333,15 +361,20 @@ int kvspaceWatch(void *h, const char *key, const uint8_t *target, uint32_t targe for (;;) { int32_t len = 0; uint8_t *d = NULL; - if (exp_reap((kvspace_t *)h, key) != 2) + uint8_t *held = NULL; uint32_t hl = 0; + int st = exp_reap((kvspace_t *)h, key, &held, &hl); + if (st == 1) { d = held; len = (int32_t)hl; } + else if (st != 2) d = kvspaceShmGet((kvspace_t *)h, key, 1, &len); if (d && (uint32_t)len == target_len && memcmp(d, target, target_len) == 0) { uint8_t *c = malloc((size_t)len); memcpy(c, d, (size_t)len); *out = c; *out_len = (uint32_t)len; + free(held); return 0; } + free(held); clock_gettime(CLOCK_MONOTONIC, &tn); uint64_t elapsed = (uint64_t)(tn.tv_sec - t0.tv_sec) * 1000000000ULL + (uint64_t)(tn.tv_nsec - t0.tv_nsec); @@ -493,7 +526,7 @@ int kvspaceIncr(void *h, const char *key, int64_t *out, char *err, uint32_t err_ if (!h || !key || !out) { incr_err(err, err_cap, "Incr: bad args"); return 1; } *out = 0; pthread_mutex_lock(&incr_mu); - exp_reap((kvspace_t *)h, key); + exp_reap((kvspace_t *)h, key, NULL, NULL); int32_t len = 0; uint8_t *d = kvspaceShmGet((kvspace_t *)h, key, 1, &len); int64_t n = 0; @@ -558,9 +591,21 @@ int kvspaceExpire(void *h, const char *key, uint64_t ttl_ns, char *err, uint32_t incr_err(err, err_cap, key && key[0] && key[strlen(key) - 1] == '/' ? "Expire: directory" : "Expire: key is not an absolute path"); return 1; } - if (exp_reap((kvspace_t *)h, key) == 2) { incr_err(err, err_cap, "Expire: missing key"); return 1; } + uint8_t *held = NULL; uint32_t hl = 0; + int st = exp_reap((kvspace_t *)h, key, &held, &hl); + if (st == 2) { incr_err(err, err_cap, "Expire: missing key"); return 1; } int32_t len = 0; - uint8_t *d = kvspaceShmGet((kvspace_t *)h, key, 1, &len); + uint8_t *d = NULL; + if (st == 1) { d = held; len = (int32_t)hl; } + else { + d = kvspaceShmGet((kvspace_t *)h, key, 1, &len); + if (d && len > 0) { + uint8_t *c = malloc((size_t)len); + if (!c) abort(); + memcpy(c, d, (size_t)len); + d = c; + } + } if (!d || len <= 0) { incr_err(err, err_cap, "Expire: missing key"); return 1; } int64_t dead = exp_now_ns() + (int64_t)ttl_ns; pthread_mutex_lock(&exp_mu); @@ -571,11 +616,15 @@ int kvspaceExpire(void *h, const char *key, uint64_t ttl_ns, char *err, uint32_t } if (slot < 0 && !exp_tab[i].used) slot = i; } - if (slot < 0) { pthread_mutex_unlock(&exp_mu); abort(); } + if (slot < 0) { pthread_mutex_unlock(&exp_mu); free(d); abort(); } + if (exp_tab[slot].used) exp_clear_slot(slot); exp_tab[slot].kv = (kvspace_t *)h; snprintf(exp_tab[slot].key, sizeof exp_tab[slot].key, "%s", key); exp_tab[slot].dead_ns = dead; + exp_tab[slot].val = d; + exp_tab[slot].vlen = (uint32_t)len; exp_tab[slot].used = 1; pthread_mutex_unlock(&exp_mu); + kvspaceShmDel((kvspace_t *)h, key); return 0; }