Skip to content
Open
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
4 changes: 4 additions & 0 deletions CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
24 changes: 18 additions & 6 deletions src/durable_abi.c
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Expand All @@ -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);
}
Expand All @@ -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) {
Expand Down
34 changes: 31 additions & 3 deletions src/kvspace.c
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}

Expand All @@ -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;
Expand Down Expand Up @@ -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))
Expand Down
3 changes: 3 additions & 0 deletions src/kvspace_shm.h
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
83 changes: 83 additions & 0 deletions tests/test_value_refs.c
Original file line number Diff line number Diff line change
@@ -0,0 +1,83 @@
#define _GNU_SOURCE
#include "kvspace_shm.h"
#include "xvalue_head.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#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;
}
Loading