diff --git a/CMakeLists.txt b/CMakeLists.txt index 7b362e6..d8032b7 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -56,6 +56,10 @@ if(KVSPACE_BUILD_TESTS) target_include_directories(test_handle_cache PRIVATE src) target_link_libraries(test_handle_cache PRIVATE kvspace-c) add_test(NAME handle_cache COMMAND test_handle_cache) + add_executable(test_value_refs tests/test_value_refs.c) + target_include_directories(test_value_refs PRIVATE src) + target_link_libraries(test_value_refs PRIVATE kvspace-c) + add_test(NAME value_refs COMMAND test_value_refs) add_executable(test_xvalue_head tests/test_xvalue_head.c src/xvalue_head.c src/xvalue_meta.c) target_include_directories(test_xvalue_head PRIVATE src) diff --git a/src/durable_abi.c b/src/durable_abi.c index a6ca2b3..9d848ee 100644 --- a/src/durable_abi.c +++ b/src/durable_abi.c @@ -88,12 +88,7 @@ int kvspaceGet(void *h, const char *key, int resolve, uint8_t **out, return 0; } -int kvspaceSetValue(void *h, const char *key, const uint8_t *value, - uint32_t value_len, uint8_t ro, uint32_t vid, - char *err, uint32_t err_cap) { - int rc = value_len > INT32_MAX ? -2 : - kvspaceShmSetValue((kvspace_t *)h, key, value, - (int32_t)value_len, ro, vid); +static int set_value_result(int rc, const char *key, char *err, uint32_t err_cap) { if (rc == -2) { if (err && err_cap) snprintf(err, err_cap, "invalid XValue"); @@ -107,6 +102,15 @@ int kvspaceSetValue(void *h, const char *key, const uint8_t *value, return 0; } +int kvspaceSetValue(void *h, const char *key, const uint8_t *value, + uint32_t value_len, uint8_t ro, uint32_t vid, + char *err, uint32_t err_cap) { + int rc = value_len > INT32_MAX ? -2 : + kvspaceShmSetValue((kvspace_t *)h, key, value, + (int32_t)value_len, ro, vid); + return set_value_result(rc, key, err, err_cap); +} + int kvspaceResolveRef(void *h, const char *key, kvspaceRef_t *ref) { return kvspaceShmResolveRef((kvspace_t *)h, key, ref); } @@ -125,6 +129,14 @@ int kvspaceGetByRef(void *h, kvspaceRef_t *ref, const char *key_fallback, return 0; } +int kvspaceSetValueByRef(void *h, kvspaceRef_t *ref, const char *key, + const uint8_t *value, uint32_t value_len, + uint8_t ro, uint32_t vid, char *err, uint32_t err_cap) { + int rc = value_len > INT32_MAX ? -2 : + kvspaceShmSetValueByRef(h, ref, key, value, (int32_t)value_len, ro, vid); + return set_value_result(rc, key, err, err_cap); +} + int kvspaceSetPartByRef(void *h, kvspaceRef_t *ref, const char *key_fallback, uint32_t offset, const uint8_t *buf, uint32_t buf_len, char *err, uint32_t err_cap) { diff --git a/src/kvspace.c b/src/kvspace.c index 2955461..76bfa47 100644 --- a/src/kvspace.c +++ b/src/kvspace.c @@ -1454,8 +1454,9 @@ uint8_t *kvspaceShmGetByRef(kvspace_t *kv, kvspaceRef_t *ref, if (!key_fallback) return NULL; uint8_t *raw = kvspaceShmGet(kv, key_fallback, 0, ol); - if (raw) - kvspaceShmResolveRef(kv, key_fallback, ref); + kvspaceRef_t refreshed; + if (raw && kvspaceShmResolveRef(kv, key_fallback, &refreshed) == 0) + *ref = refreshed; return raw; } @@ -1478,7 +1479,9 @@ int kvspaceShmSetPartByRef(kvspace_t *kv, kvspaceRef_t *ref, if (!d || offset + buf_len > (uint32_t)rl) return -1; memcpy(d + offset, buf, buf_len); - kvspaceShmResolveRef(kv, key_fallback, ref); + kvspaceRef_t refreshed; + if (kvspaceShmResolveRef(kv, key_fallback, &refreshed) == 0) + *ref = refreshed; return 0; } uint8_t *raw; @@ -1647,6 +1650,31 @@ int kvspaceShmSetValue(kvspace_t *kv, const char *key, const uint8_t *val, return shm_set_validated(kv, key, val, val_len, &head, ro, vid); } +int kvspaceShmSetValueByRef(kvspace_t *kv, kvspaceRef_t *ref, const char *key, + const uint8_t *val, int32_t val_len, + uint8_t ro, uint32_t vid) { + kvspaceXh wire; + if (!kv || !ref || !key || !val || val_len <= 0 || + kvspaceXhDecode(val, (uint64_t)val_len, &wire) != 0 || + wire.total != (uint64_t)val_len) + return -2; + if (kv_sync(kv) != 0 || strlen(key) >= 1024 || strstr(key, "//") || + reserved_meta_path(key)) + return -1; + int32_t id = ref->gen == 0 ? ref_leaf(kv, ref, key) : -1; + art_hdr_t *h = id >= 0 ? art_hdr(kv, id) : NULL; + if (h && h->has_value) { + if (shm_set_local(kv, key, val, val_len, h) != 0) + return -1; + return sync_metadata(kv, key, ro, vid); + } + int rc = shm_set_validated(kv, key, val, val_len, &wire, ro, vid); + kvspaceRef_t refreshed; + if (rc == 0 && kvspaceShmResolveRef(kv, key, &refreshed) == 0) + *ref = refreshed; + return rc; +} + int kvspaceShmWriteInPlace(kvspace_t *kv, const char *key, int resolve, int32_t body_len, uint8_t **body) { if (!kv || !key || !body || body_len < 0 || reserved_meta_path(key)) diff --git a/src/kvspace_shm.h b/src/kvspace_shm.h index 150df0a..222eb43 100644 --- a/src/kvspace_shm.h +++ b/src/kvspace_shm.h @@ -54,6 +54,9 @@ int kvspaceShmSet(kvspace_t *kv, const char *key, const uint8_t *val, int32_t val_len); int kvspaceShmSetValue(kvspace_t *kv, const char *key, const uint8_t *val, int32_t val_len, uint8_t ro, uint32_t vid); +int kvspaceShmSetValueByRef(kvspace_t *kv, kvspaceRef_t *ref, const char *key, + const uint8_t *val, int32_t val_len, + uint8_t ro, uint32_t vid); // 零拷贝写原语——返回 SHM 常驻 body 偏移指针供调用方直接写；写即持久，无收尾。 // WriteInPlace: key 必须已存在、body_len 必须等于原 body_len（同 kind diff --git a/tests/test_value_refs.c b/tests/test_value_refs.c new file mode 100644 index 0000000..e30c1be --- /dev/null +++ b/tests/test_value_refs.c @@ -0,0 +1,83 @@ +#define _GNU_SOURCE +#include "kvspace_shm.h" +#include "xvalue_head.h" +#include +#include +#include +#include + +#define CHECK(c) do { if (!(c)) { fprintf(stderr, "%d: %s\n", __LINE__, #c); return 1; } } while (0) + +static int same(kvspace_t *kv, const char *key, const uint8_t *value, uint64_t n) { + int32_t len = 0; + uint8_t *got = kvspaceShmGet(kv, key, 0, &len); + return got && len == (int32_t)n && memcmp(got, value, (size_t)n) == 0; +} + +int main(void) { + char dir[] = "/tmp/kvspace-value-refs-XXXXXX", path[128]; + CHECK(mkdtemp(dir)); + snprintf(path, sizeof path, "%s/db", dir); + kvspace_t *a = kvspaceShmOpen(path, 8UL * 64 * 64); + kvspace_t *b = kvspaceShmOpen(path, 8UL * 64 * 64); + CHECK(a && b); + uint8_t *value = NULL, raw[4096] = {42}; + uint64_t n = 0; + CHECK(kvspaceXhNewScalar("int64", raw, 8, &value, &n) == 0); + CHECK(kvspaceShmSet(a, "/unrelated", value, (int32_t)n) == 0); + CHECK(kvspaceShmSet(a, "/local/k", value, (int32_t)n) == 0); + free(value); + kvspaceRef_t ref; + CHECK(kvspaceShmResolveRef(a, "/local/k", &ref) == 0); + const uint64_t sizes[] = {1, 8, 32, 4096, 3, 512, 1}; + for (unsigned i = 0; i < sizeof sizes / sizeof sizes[0]; i++) { + raw[0] = (uint8_t)(i + 50); + CHECK(kvspaceXhNewSlack(1, raw, sizes[i], sizes[i], &value, &n) == 0); + CHECK(kvspaceShmSetValueByRef(a, &ref, "/local/k", value, (int32_t)n, 1, i + 1) == 0); + CHECK(same(b, "/local/k", value, n)); + uint8_t ro; + uint32_t vid; + CHECK(kvspaceShmMetaGetAt(b, "/local/k", &ro, &vid) == 0 && ro == 1 && vid == i + 1); + CHECK(kvspaceShmSetValueByRef(a, &ref, "/local/k", value, (int32_t)n, 0, 0) == 0); + CHECK(kvspaceShmMetaGetAt(b, "/local/k", &ro, &vid) == 0 && ro == 0 && vid == 0); + free(value); + } + uint64_t dims[] = {2, 2}; + CHECK(kvspaceXhNewTensor(dims, 2, "float64", raw, 32, &value, &n) == 0); + CHECK(kvspaceShmSetValueByRef(a, &ref, "/local/k", value, (int32_t)n, 0, 0) == 0); + CHECK(same(b, "/local/k", value, n)); + free(value); + CHECK(kvspaceXhNewPtr("float64", "/unrelated", 10, &value, &n) == 0); + CHECK(kvspaceShmSetValueByRef(a, &ref, "/local/k", value, (int32_t)n, 0, 0) == 0); + CHECK(same(b, "/local/k", value, n)); + free(value); + CHECK(kvspaceXhNewScalar("int64", raw, 8, &value, &n) == 0); + CHECK(kvspaceShmSet(b, "/local/k", value, (int32_t)n) == 0); + for (int i = 0; i < 60; i++) { + char key[64]; + snprintf(key, sizeof key, "/local/k%c", i + 32); + CHECK(kvspaceShmSet(b, key, value, (int32_t)n) == 0); + } + CHECK(kvspaceShmSetValueByRef(a, &ref, "/local/k", value, (int32_t)n, 0, 0) == 0); + CHECK(same(b, "/local/k", value, n)); + uint8_t invalid[3] = {31}; + CHECK(kvspaceShmSetValueByRef(a, &ref, "/local/k", invalid, sizeof invalid, 0, 0) != 0); + CHECK(same(b, "/local/k", value, n)); + CHECK(kvspaceShmDel(b, "/local/k") == 0); + CHECK(kvspaceShmMkindex(b, "/remote/", 0) == 0); + CHECK(kvspaceShmSet(b, "/remote/k", value, (int32_t)n) == 0); + CHECK(kvspaceShmExtindex(b, "/local/", "/remote/") == 0); + for (int i = 0; i < 3; i++) { + int32_t len = 0; + uint8_t *got = kvspaceShmGetByRef(a, &ref, "/local/k", &len); + CHECK(got && len == (int32_t)n && memcmp(got, value, (size_t)n) == 0); + } + CHECK(kvspaceShmSetValueByRef(a, &ref, "/local/k", value, (int32_t)n, 0, 0) != 0); + CHECK(kvspaceShmDelextindex(b, "/local/") == 0); + CHECK(kvspaceShmSetValueByRef(a, &ref, "/local/k", value, (int32_t)n, 0, 0) == 0); + CHECK(same(b, "/local/k", value, n)); + free(value); + kvspaceShmClose(b); + kvspaceShmClose(a); + return 0; +}