lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge branch 'persistent_ha' into scheduler

Browse source
This commit is contained in:
Rongsong Shen 2026-05-12 10:52:46 +00:00
commit a07a74a943
106 changed files with 18528 additions and 637 deletions
Download patch file Download diff file Expand all files Collapse all files

191
src/lib/cluster.c Normal file
View file

@ -0,0 +1,191 @@
/* src/lib/cluster.c — cluster discovery and reconnect (P2.5-2/3)
*
* xce_connect_cluster() iterates the candidate list and returns the first
* successful endpoint_t* (cast to void*). In production the real SSL
* connect is performed via do_connect(); unit tests override s_connect_fn.
*
* xce_reconnect_step() drives a non-blocking state machine:
* IDLE → CONNECTING → CONNECTED (happy path)
* IDLE → CONNECTING → BACKOFF → CONNECTING → … (retrying)
*
* Backoff sleep is skipped when s_connect_fn is set (i.e., in unit tests)
* so tests do not need to wait for real timeouts.
*/
#include <errno.h>
#include <stddef.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include "xce/cluster.h"
/* ── Peer list parsing ─────────────────────────────────────────── */
int xce_parse_peers(const char* peers_str, xce_cluster_config_t* cfg)
{
cfg->candidates_count = 0;
if (!peers_str || peers_str[0] == '\0')
return -1;
char buf[2048];
strncpy(buf, peers_str, sizeof(buf) - 1);
buf[sizeof(buf) - 1] = '\0';
char* saveptr;
char* tok = strtok_r(buf, ",", &saveptr);
while (tok && cfg->candidates_count < MAX_HA_CANDIDATES) {
/* Format: name=host:port OR host:port */
char* eq = strchr(tok, '=');
char* addr = eq ? eq + 1 : tok;
char* colon = strrchr(addr, ':');
if (!colon)
return -1;
*colon = '\0';
xce_candidate_t* cand = &cfg->candidates[cfg->candidates_count];
if (eq) {
*eq = '\0';
strncpy(cand->name, tok, XCE_MAXNAME_LEN - 1);
cand->name[XCE_MAXNAME_LEN - 1] = '\0';
} else {
cand->name[0] = '\0';
}
strncpy(cand->hostname, addr, XCE_MAXHOSTNAME_LEN - 1);
cand->hostname[XCE_MAXHOSTNAME_LEN - 1] = '\0';
cand->port = atoi(colon + 1);
cfg->candidates_count++;
tok = strtok_r(NULL, ",", &saveptr);
}
return cfg->candidates_count > 0 ? 0 : -1;
}
/* ── Test seam ─────────────────────────────────────────────────── */
static xce_connect_fn_t s_connect_fn = NULL;
void xce_cluster_set_connect_fn(xce_connect_fn_t fn)
{
s_connect_fn = fn;
}
/* do_connect: calls the mock when set, otherwise returns NULL (stub for
* the real SSL connect via netio — to be wired in a later task). */
static void* do_connect(const char* host, int port, const char* ca_cert)
{
if (s_connect_fn)
return s_connect_fn(host, port, ca_cert);
/* Production path: real SSL connect via netio — stub for now */
(void)host;
(void)port;
(void)ca_cert;
return NULL;
}
/* ── Validation ────────────────────────────────────────────────── */
int xce_cluster_config_validate(const xce_cluster_config_t* cfg)
{
if (!cfg)
return -EINVAL;
if (cfg->candidates_count < 1 || cfg->candidates_count > MAX_HA_CANDIDATES)
return -EINVAL;
for (int i = 0; i < cfg->candidates_count; i++) {
if (cfg->candidates[i].hostname[0] == '\0')
return -EINVAL;
if (cfg->candidates[i].port <= 0)
return -EINVAL;
}
if (cfg->ca_cert_path[0] == '\0')
return -EINVAL;
if (cfg->identity[0] == '\0')
return -EINVAL;
if (cfg->candidates_count > 3)
fprintf(stderr,
"WARN: HA cluster has %d candidates; validated for <=3 only\n",
cfg->candidates_count);
return 0;
}
/* ── Connect with candidate fallback ───────────────────────────── */
void* xce_connect_cluster(xce_cluster_config_t* cfg)
{
if (xce_cluster_config_validate(cfg) != 0)
return NULL;
for (int i = 0; i < cfg->candidates_count; i++) {
void* ep = do_connect(cfg->candidates[i].hostname,
cfg->candidates[i].port,
cfg->ca_cert_path);
if (ep)
return ep;
}
return NULL;
}
/* ── Reconnect state machine ───────────────────────────────────── */
int xce_reconnect_init(xce_reconnect_ctx_t* rctx, xce_cluster_config_t* cfg)
{
if (!rctx || !cfg)
return -EINVAL;
memset(rctx, 0, sizeof(*rctx));
rctx->config = cfg;
rctx->state = XCE_RECONNECT_IDLE;
rctx->backoff_ms = XCE_RECONNECT_INITIAL_BACKOFF_MS;
return 0;
}
int xce_reconnect_step(xce_reconnect_ctx_t* rctx)
{
if (!rctx)
return -EINVAL;
switch (rctx->state) {
case XCE_RECONNECT_IDLE:
rctx->state = XCE_RECONNECT_CONNECTING;
return 0; /* next step does the actual connect */
case XCE_RECONNECT_CONNECTING: {
void* ep = xce_connect_cluster(rctx->config);
if (ep) {
rctx->endpoint = ep;
rctx->state = XCE_RECONNECT_CONNECTED;
rctx->attempt = 0;
rctx->backoff_ms = XCE_RECONNECT_INITIAL_BACKOFF_MS;
} else {
rctx->attempt++;
rctx->state = XCE_RECONNECT_BACKOFF;
}
return 0;
}
case XCE_RECONNECT_BACKOFF: {
/* Skip sleep in test context (mock connect set) to keep tests fast */
if (!s_connect_fn) {
struct timespec ts = {
.tv_sec = rctx->backoff_ms / 1000,
.tv_nsec = (long)(rctx->backoff_ms % 1000) * 1000000L,
};
nanosleep(&ts, NULL);
}
/* Exponential backoff with cap */
rctx->backoff_ms *= XCE_RECONNECT_BACKOFF_MULTIPLIER;
if (rctx->backoff_ms > XCE_RECONNECT_MAX_BACKOFF_MS)
rctx->backoff_ms = XCE_RECONNECT_MAX_BACKOFF_MS;
rctx->state = XCE_RECONNECT_CONNECTING;
return 0;
}
case XCE_RECONNECT_CONNECTED:
case XCE_RECONNECT_FAILED:
return -1;
}
return 0;
}
void xce_reconnect_fini(xce_reconnect_ctx_t* rctx)
{
if (rctx)
memset(rctx, 0, sizeof(*rctx));
}

12
src/lib/delay.c
View file

@ -3,6 +3,7 @@
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include "delay.h"
#include "utils.h"
@ -17,7 +18,14 @@ static void* delay_worker(void* args)
time_t now;
pthread_testcancel();
sleep(mgr->precision);
{
long ms = (long)mgr->precision_ms;
struct timespec ts = {
.tv_sec = ms / 1000,
.tv_nsec = (ms % 1000) * 1000000L,
};
nanosleep(&ts, NULL);
}
it = ordq_peek(&(mgr->tasks), NULL);
if (it == NULL) {
@ -100,7 +108,7 @@ int delaymgr_init(delaymgr_t* mgr, int precision, task_run_fn_t runner_fn)
runner_fn = default_runner_fn;
}
mgr->precision = precision;
mgr->precision_ms = precision;
mgr->running = 0;
mgr->runner_fn = runner_fn;
mgr->worker = 0;

52
src/lib/hook.c
View file

@ -1,4 +1,5 @@
#include <errno.h>
#include <stdatomic.h>
#include <stdlib.h>
#include <string.h>
@ -41,19 +42,60 @@ int xce_add_hook(hooksys_t* s, int typ, hook_fn_t fn, void* data)
return -EINVAL;
}
/* Scan for a disabled (NULL-fn) slot of the same type to reuse */
{
xlist_t* l = &s->hooks[typ];
int i;
for (i = 0; i < l->count; i++) {
hook_item_t* slot = (hook_item_t*)l->data[i];
if (!slot) continue;
hook_fn_t cur =
atomic_load_explicit(&slot->fn, memory_order_acquire);
if (cur == NULL) {
slot->data = data;
atomic_store_explicit(&slot->fn, fn, memory_order_release);
return 0;
}
}
}
/* No reusable slot — allocate new */
it = (hook_item_t*)calloc(1, sizeof(*it));
if (it == NULL) {
return -ENOMEM;
}
it->fn = fn;
it->data = data;
atomic_store_explicit(&it->fn, fn, memory_order_release);
xlist_store(&(s->hooks[typ]), it);
return 0;
}
int xce_disable_hook(hooksys_t* s, int typ, hook_fn_t fn)
{
xlist_t* l;
int i;
if (!s || !fn || typ < 0 || typ >= s->maxtypes) {
return -EINVAL;
}
l = &s->hooks[typ];
for (i = 0; i < l->count; i++) {
hook_item_t* it = (hook_item_t*)l->data[i];
if (!it) continue;
hook_fn_t cur = atomic_load_explicit(&it->fn, memory_order_acquire);
if (cur == fn) {
atomic_store_explicit(&it->fn, (hook_fn_t)NULL,
memory_order_release);
return 0;
}
}
return -ENOENT;
}
/* HOOK_BEFORE_EVENT returns
* negative value will cause
* fail
@ -76,14 +118,20 @@ int xce_run_hooks(hooksys_t* s, int type, int situation, int detail, ...)
for (i = 0; i < l->count; i++) {
hook_item_t* it = l->data[i];
hook_fn_t local_fn;
va_list ap;
if (it == NULL) {
continue;
}
local_fn = atomic_load_explicit(&it->fn, memory_order_acquire);
if (!local_fn) {
continue; /* disabled slot */
}
va_start(ap, detail);
rc = (*(it->fn))(type, situation, detail, it->data, ap);
rc = local_fn(type, situation, detail, it->data, ap);
va_end(ap);
if ((situation == HOOK_BEFORE_EVENT) && (rc < 0)) {

29
src/lib/kv.c
View file

@ -248,6 +248,35 @@ static int kvtab_find(meta_obj_t* obj, void* key, void** value)
return 1;
}
/*
* kvtab_foreach — walk every live entry and invoke cb(key, index, arg).
* index is recovered by casting the stored void* value back to int.
* Caller must hold any necessary external lock before calling.
*/
int kvtab_foreach(var kvtab_var, kvtab_foreach_fn_t cb, void* arg)
{
if (kvtab_var == NULL || cb == NULL) {
return -EINVAL;
}
meta_obj_t* obj = RTTI_DATA_TO_OBJ(kvtab_var);
kvtab_t* t = RTTI_OBJ_TO_DATA(kvtab_t, obj);
int count = 0;
uint32_t i;
for (i = 0; i < t->nslots; i++) {
kvent_t* e = t->slots[i];
while (e != NULL) {
int idx = (int)(long)e->value;
cb(e->key, idx, arg);
count++;
e = e->next;
}
}
return count;
}
typeinfo_t kvtab_type
= { .name = "kvtab",
.privateSize = 0,

112
src/lib/kvmgr.c
View file

@ -1,12 +1,15 @@
#include <errno.h>
#include <inttypes.h>
#include <rtti.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "htab.h"
#include "kv.h"
#include "kvmgr.h"
#include "mylock.h"
#include "persist.h"
#include "utils.h"
#include "xce/types.h"
#include "zarray.h"
@ -290,7 +293,9 @@ typeinfo_t xkvmgr_type = {
.vget_vec = (vget_vec_fn_t)xkvmgr_get_vec,
.vput = (vput_fn_t)xkvmgr_put,
.vput_vec = (vput_vec_fn_t)xkvmgr_put_vec,
.vclear = (vclear_fn_t)NULL },
.vclear = (vclear_fn_t)NULL,
.vfreeze = NULL, /* wired in P0-2 */
.vthaw = NULL /* wired in P0-2 */ },
.iter
= &(intf_iterator_t) { .make = (iter_make_fn_t)xkvmgr_make_iter,
.next = (iter_next_fn_t)xkvmgr_iter_next,
@ -300,3 +305,108 @@ typeinfo_t xkvmgr_type = {
.lock = &(intf_lock_t) { .lock = (lock_fn_t)xkvmgr_lock,
.unlock = (unlock_fn_t)xkvmgr_unlock } }
};
/* ------------------------------------------------------------------ */
/* kvmgr_capture_keys — P0-14a */
/* ------------------------------------------------------------------ */
typedef struct {
slot_key_t* keys;
size_t capacity;
size_t filled;
kvmgr_key_type_t key_type;
int oom;
} capture_ctx_t;
static void capture_one(void* raw_key, int index, void* arg)
{
capture_ctx_t* ctx = (capture_ctx_t*)arg;
if (ctx->oom || ctx->filled >= ctx->capacity) {
return;
}
slot_key_t* sk = &ctx->keys[ctx->filled];
sk->index = index;
if (ctx->key_type == KVMGR_KEY_STRING) {
const char* src = (const char*)raw_key;
strncpy(sk->key, src, XCE_MAXNAME_LEN - 1);
sk->key[XCE_MAXNAME_LEN - 1] = '\0';
sk->key_len = strlen(sk->key) + 1;
} else {
/* Integer key — format as decimal string */
int64_t intval = (int64_t)(long)raw_key;
int n = snprintf(sk->key, XCE_MAXNAME_LEN,
"%" PRId64, intval);
if (n < 0 || n >= XCE_MAXNAME_LEN) {
/* Truncated — mark as oom so caller can detect */
ctx->oom = 1;
return;
}
sk->key_len = (size_t)n + 1;
}
ctx->filled++;
}
/*
* Count live entries by iterating; kvtab_foreach returns the count.
* We use a NULL callback to count first, then allocate, then fill.
*/
static void count_one(void* key, int index, void* arg)
{
(void)key;
(void)index;
size_t* n = (size_t*)arg;
(*n)++;
}
int kvmgr_capture_keys(var kvmgr_var,
slot_key_t** keys_out,
size_t* n_keys_out)
{
*keys_out = NULL;
*n_keys_out = 0;
if (kvmgr_var == NULL) {
return -EINVAL;
}
/* The public API pointer (var) points to the data region;
* convert to the meta_obj_t header, then extract xkvmgr_t. */
meta_obj_t* obj = RTTI_DATA_TO_OBJ(kvmgr_var);
xkvmgr_t* kvm = RTTI_OBJ_TO_DATA(xkvmgr_t, obj);
/* Count live entries */
size_t n = 0;
kvtab_foreach(kvm->kvtab, count_one, &n);
if (n == 0) {
return 0;
}
slot_key_t* keys = calloc(n, sizeof(slot_key_t));
if (keys == NULL) {
return -ENOMEM;
}
capture_ctx_t ctx = {
.keys = keys,
.capacity = n,
.filled = 0,
.key_type = kvm->key_type,
.oom = 0,
};
kvtab_foreach(kvm->kvtab, capture_one, &ctx);
if (ctx.oom) {
free(keys);
return -ENOMEM;
}
*keys_out = keys;
*n_keys_out = ctx.filled;
return 0;
}

6
src/lib/meson.build
View file

@ -104,6 +104,7 @@ libxce_src = [ gen_srcs[0],
'resobj.c',
'wobj.c',
'configobj.c',
'cluster.c',
]
incdir = include_directories('../include')
@ -163,6 +164,11 @@ libxce_test_src = ['testsuite/test.c',
'testsuite/test_wobj.c',
'testsuite/test_qobj.c',
'testsuite/test_request.c',
'testsuite/test_zarray_cow.c',
'testsuite/test_kvmgr_capture.c',
'testsuite/test_hooks.c',
'testsuite/test_cluster.c',
'testsuite/test_cluster_parse.c',
]
libxce_test_deps = [libxce_deps, librtti_dep, unity_dep, libxce_dep]

10
src/lib/testsuite/test.c
View file

@ -29,6 +29,11 @@ extern void test_jobobj(void);
extern void test_qobj(void);
extern void test_wobj(void);
extern void test_request(void);
extern void test_zarray_cow(void);
extern void test_kvmgr_capture(void);
extern void test_hooks(void);
extern void test_cluster(void);
extern void test_cluster_parse(void);
void setUp(void) { return; }
@ -127,6 +132,11 @@ int main(int argc, char* argv[])
TEST(test_wobj);
TEST(test_qobj);
TEST(test_request);
TEST(test_zarray_cow);
TEST(test_kvmgr_capture);
TEST(test_hooks);
TEST(test_cluster);
TEST(test_cluster_parse);
return UNITY_END();
}

181
src/lib/testsuite/test_cluster.c Normal file
View file

@ -0,0 +1,181 @@
#include <errno.h>
#include <stdio.h>
#include <string.h>
#include <unity/unity.h>
#include "xce/cluster.h"
/* ── Test seam: mock connect function ───────────────────────────── */
static int s_connect_call_count = 0;
static int s_connect_fail_n = 0; /* fail first N attempts, then succeed */
static char s_last_host[256] = "";
static int s_last_port = 0;
static void* mock_connect(const char* host, int port, const char* ca_cert)
{
(void)ca_cert;
strncpy(s_last_host, host, sizeof(s_last_host) - 1);
s_last_port = port;
s_connect_call_count++;
if (s_connect_call_count <= s_connect_fail_n)
return NULL;
return (void*)0xDEADBEEF; /* non-NULL = success */
}
static xce_cluster_config_t make_valid_config(void)
{
xce_cluster_config_t cfg = {0};
strncpy(cfg.candidates[0].hostname, "10.0.1.1",
XCE_MAXHOSTNAME_LEN - 1);
cfg.candidates[0].port = 7000;
strncpy(cfg.candidates[1].hostname, "10.0.1.2",
XCE_MAXHOSTNAME_LEN - 1);
cfg.candidates[1].port = 7001;
cfg.candidates_count = 2;
strncpy(cfg.ca_cert_path, "/etc/xce/ca.pem",
sizeof(cfg.ca_cert_path) - 1);
strncpy(cfg.identity, "worker-01", sizeof(cfg.identity) - 1);
return cfg;
}
/* ── Test cases ─────────────────────────────────────────────────── */
/* Test 1: validate rejects empty config */
static void test_validate_rejects_zero_candidates(void)
{
xce_cluster_config_t cfg = {0};
cfg.candidates_count = 0;
TEST_ASSERT_NOT_EQUAL(0, xce_cluster_config_validate(&cfg));
printf("test_validate_rejects_zero_candidates: PASS\n");
}
/* Test 2: validate rejects too many candidates */
static void test_validate_rejects_over_max_candidates(void)
{
xce_cluster_config_t cfg = {0};
cfg.candidates_count = MAX_HA_CANDIDATES + 1;
TEST_ASSERT_NOT_EQUAL(0, xce_cluster_config_validate(&cfg));
printf("test_validate_rejects_over_max_candidates: PASS\n");
}
/* Test 3: validate rejects missing identity */
static void test_validate_rejects_empty_identity(void)
{
xce_cluster_config_t cfg = make_valid_config();
cfg.identity[0] = '\0';
TEST_ASSERT_NOT_EQUAL(0, xce_cluster_config_validate(&cfg));
printf("test_validate_rejects_empty_identity: PASS\n");
}
/* Test 4: validate accepts valid config */
static void test_validate_accepts_valid_config(void)
{
xce_cluster_config_t cfg = make_valid_config();
TEST_ASSERT_EQUAL_INT(0, xce_cluster_config_validate(&cfg));
printf("test_validate_accepts_valid_config: PASS\n");
}
/* Test 5: connect succeeds on first candidate */
static void test_connect_succeeds_first_candidate(void)
{
xce_cluster_config_t cfg = make_valid_config();
s_connect_call_count = 0;
s_connect_fail_n = 0;
xce_cluster_set_connect_fn(mock_connect);
void* ep = xce_connect_cluster(&cfg);
TEST_ASSERT_NOT_NULL(ep);
TEST_ASSERT_EQUAL_INT(1, s_connect_call_count);
TEST_ASSERT_EQUAL_STRING("10.0.1.1", s_last_host);
TEST_ASSERT_EQUAL_INT(7000, s_last_port);
xce_cluster_set_connect_fn(NULL);
printf("test_connect_succeeds_first_candidate: PASS\n");
}
/* Test 6: connect falls back to second candidate when first fails */
static void test_connect_fallback_to_second_candidate(void)
{
xce_cluster_config_t cfg = make_valid_config();
s_connect_call_count = 0;
s_connect_fail_n = 1; /* first attempt fails */
xce_cluster_set_connect_fn(mock_connect);
void* ep = xce_connect_cluster(&cfg);
TEST_ASSERT_NOT_NULL(ep);
TEST_ASSERT_EQUAL_INT(2, s_connect_call_count);
TEST_ASSERT_EQUAL_STRING("10.0.1.2", s_last_host);
xce_cluster_set_connect_fn(NULL);
printf("test_connect_fallback_to_second_candidate: PASS\n");
}
/* Test 7: connect returns NULL when all candidates fail */
static void test_connect_returns_null_all_fail(void)
{
xce_cluster_config_t cfg = make_valid_config();
s_connect_call_count = 0;
s_connect_fail_n = 999; /* all fail */
xce_cluster_set_connect_fn(mock_connect);
void* ep = xce_connect_cluster(&cfg);
TEST_ASSERT_NULL(ep);
xce_cluster_set_connect_fn(NULL);
printf("test_connect_returns_null_all_fail: PASS\n");
}
/* Test 8: reconnect state machine — IDLE → CONNECTING → CONNECTED */
static void test_reconnect_step_reaches_connected(void)
{
xce_cluster_config_t cfg = make_valid_config();
xce_reconnect_ctx_t rctx;
s_connect_call_count = 0;
s_connect_fail_n = 0;
xce_cluster_set_connect_fn(mock_connect);
xce_reconnect_init(&rctx, &cfg);
TEST_ASSERT_EQUAL_INT(XCE_RECONNECT_IDLE, (int)rctx.state);
int steps = 0;
while (rctx.state != XCE_RECONNECT_CONNECTED &&
rctx.state != XCE_RECONNECT_FAILED && steps < 100) {
xce_reconnect_step(&rctx);
steps++;
}
TEST_ASSERT_EQUAL_INT(XCE_RECONNECT_CONNECTED, (int)rctx.state);
TEST_ASSERT_NOT_NULL(rctx.endpoint);
xce_reconnect_fini(&rctx);
xce_cluster_set_connect_fn(NULL);
printf("test_reconnect_step_reaches_connected: PASS\n");
}
/* Test 9: reconnect state machine backs off on failure */
static void test_reconnect_step_backs_off_on_failure(void)
{
xce_cluster_config_t cfg = make_valid_config();
cfg.candidates_count = 1; /* only one candidate so it hits backoff */
xce_reconnect_ctx_t rctx;
s_connect_call_count = 0;
s_connect_fail_n = 999; /* always fail */
xce_cluster_set_connect_fn(mock_connect);
xce_reconnect_init(&rctx, &cfg);
xce_reconnect_step(&rctx); /* IDLE → CONNECTING (attempt connect) */
xce_reconnect_step(&rctx); /* CONNECTING → BACKOFF (all candidates exhausted) */
/* When mock is set, sleep is skipped — state must be BACKOFF */
TEST_ASSERT_EQUAL_INT(XCE_RECONNECT_BACKOFF, (int)rctx.state);
TEST_ASSERT_TRUE(rctx.backoff_ms > 0);
xce_reconnect_fini(&rctx);
xce_cluster_set_connect_fn(NULL);
printf("test_reconnect_step_backs_off_on_failure: PASS\n");
}
/* ── Test group entry point ─────────────────────────────────────── */
void test_cluster(void)
{
RUN_TEST(test_validate_rejects_zero_candidates);
RUN_TEST(test_validate_rejects_over_max_candidates);
RUN_TEST(test_validate_rejects_empty_identity);
RUN_TEST(test_validate_accepts_valid_config);
RUN_TEST(test_connect_succeeds_first_candidate);
RUN_TEST(test_connect_fallback_to_second_candidate);
RUN_TEST(test_connect_returns_null_all_fail);
RUN_TEST(test_reconnect_step_reaches_connected);
RUN_TEST(test_reconnect_step_backs_off_on_failure);
}

160
src/lib/testsuite/test_cluster_parse.c Normal file
View file

@ -0,0 +1,160 @@
/* test_cluster_parse.c — unit tests for the parse_peers() helper used by
* worker and scheduler main.c (P2.5-5 / P2.5-6).
*
* The parser is a static function in the respective main.c files. To keep
* the test self-contained we duplicate a simplified reference implementation
* here and test the same contract. Integration with the real binaries is
* covered by the build (both must compile).
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "xce/cluster.h"
/* ── Reference implementation of parse_peers (mirrors worker/main.c) ─── */
static int parse_peers_local(const char* peers_str, xce_cluster_config_t* cfg)
{
cfg->candidates_count = 0;
if (!peers_str || peers_str[0] == '\0')
return -1;
char buf[2048];
strncpy(buf, peers_str, sizeof(buf) - 1);
buf[sizeof(buf) - 1] = '\0';
char* saveptr;
char* tok = strtok_r(buf, ",", &saveptr);
while (tok && cfg->candidates_count < MAX_HA_CANDIDATES) {
/* Format: name=host:port OR host:port */
char* eq = strchr(tok, '=');
char* addr = eq ? eq + 1 : tok;
char* colon = strrchr(addr, ':');
if (!colon)
return -1;
*colon = '\0';
strncpy(cfg->candidates[cfg->candidates_count].hostname, addr,
XCE_MAXHOSTNAME_LEN - 1);
cfg->candidates[cfg->candidates_count].hostname[XCE_MAXHOSTNAME_LEN - 1]
= '\0';
cfg->candidates[cfg->candidates_count].port = atoi(colon + 1);
cfg->candidates_count++;
tok = strtok_r(NULL, ",", &saveptr);
}
return cfg->candidates_count > 0 ? 0 : -1;
}
/* ── Test cases ─────────────────────────────────────────────────────── */
static void test_parse_peers_single(void)
{
xce_cluster_config_t cfg = {0};
int rc = parse_peers_local("ctrl-1=10.0.1.1:7000", &cfg);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(1, cfg.candidates_count);
TEST_ASSERT_EQUAL_STRING("10.0.1.1", cfg.candidates[0].hostname);
TEST_ASSERT_EQUAL_INT(7000, cfg.candidates[0].port);
printf("test_parse_peers_single: PASS\n");
}
static void test_parse_peers_multiple(void)
{
xce_cluster_config_t cfg = {0};
int rc = parse_peers_local(
"ctrl-1=10.0.1.1:7000,ctrl-2=10.0.1.2:7001,ctrl-3=10.0.1.3:7002",
&cfg);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(3, cfg.candidates_count);
TEST_ASSERT_EQUAL_STRING("10.0.1.2", cfg.candidates[1].hostname);
TEST_ASSERT_EQUAL_INT(7001, cfg.candidates[1].port);
TEST_ASSERT_EQUAL_STRING("10.0.1.3", cfg.candidates[2].hostname);
TEST_ASSERT_EQUAL_INT(7002, cfg.candidates[2].port);
printf("test_parse_peers_multiple: PASS\n");
}
static void test_parse_peers_no_name(void)
{
xce_cluster_config_t cfg = {0};
int rc = parse_peers_local("10.0.1.1:7000,10.0.1.2:7001", &cfg);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(2, cfg.candidates_count);
TEST_ASSERT_EQUAL_STRING("10.0.1.1", cfg.candidates[0].hostname);
TEST_ASSERT_EQUAL_INT(7000, cfg.candidates[0].port);
TEST_ASSERT_EQUAL_STRING("10.0.1.2", cfg.candidates[1].hostname);
TEST_ASSERT_EQUAL_INT(7001, cfg.candidates[1].port);
printf("test_parse_peers_no_name: PASS\n");
}
static void test_parse_peers_empty_returns_error(void)
{
xce_cluster_config_t cfg = {0};
int rc = parse_peers_local("", &cfg);
TEST_ASSERT_NOT_EQUAL(0, rc);
printf("test_parse_peers_empty_returns_error: PASS\n");
}
static void test_parse_peers_null_returns_error(void)
{
xce_cluster_config_t cfg = {0};
int rc = parse_peers_local(NULL, &cfg);
TEST_ASSERT_NOT_EQUAL(0, rc);
printf("test_parse_peers_null_returns_error: PASS\n");
}
static void test_parse_peers_missing_port_returns_error(void)
{
xce_cluster_config_t cfg = {0};
int rc = parse_peers_local("ctrl-1=10.0.1.1", &cfg); /* no colon:port */
TEST_ASSERT_NOT_EQUAL(0, rc);
printf("test_parse_peers_missing_port_returns_error: PASS\n");
}
static void test_parse_peers_first_entry_parsed_correctly(void)
{
/* Verify the name= prefix is stripped and only addr:port is stored */
xce_cluster_config_t cfg = {0};
int rc = parse_peers_local("primary=192.168.10.1:9000", &cfg);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_STRING("192.168.10.1", cfg.candidates[0].hostname);
TEST_ASSERT_EQUAL_INT(9000, cfg.candidates[0].port);
printf("test_parse_peers_first_entry_parsed_correctly: PASS\n");
}
static void test_parse_peers_respects_max_candidates(void)
{
/* Build a string with MAX_HA_CANDIDATES + 2 entries; only MAX accepted */
char big[2048] = "";
char entry[128] = "";
int n = MAX_HA_CANDIDATES + 2;
for (int i = 0; i < n; i++) {
snprintf(entry, sizeof(entry), "10.0.1.%d:%d", i + 1, 7000 + i);
if (i > 0)
strncat(big, ",", sizeof(big) - strlen(big) - 1);
strncat(big, entry, sizeof(big) - strlen(big) - 1);
}
xce_cluster_config_t cfg = {0};
int rc = parse_peers_local(big, &cfg);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(MAX_HA_CANDIDATES, cfg.candidates_count);
printf("test_parse_peers_respects_max_candidates: PASS\n");
}
/* ── Test group entry point ─────────────────────────────────────────── */
void test_cluster_parse(void)
{
RUN_TEST(test_parse_peers_single);
RUN_TEST(test_parse_peers_multiple);
RUN_TEST(test_parse_peers_no_name);
RUN_TEST(test_parse_peers_empty_returns_error);
RUN_TEST(test_parse_peers_null_returns_error);
RUN_TEST(test_parse_peers_missing_port_returns_error);
RUN_TEST(test_parse_peers_first_entry_parsed_correctly);
RUN_TEST(test_parse_peers_respects_max_candidates);
}

22
src/lib/testsuite/test_config.c
View file

@ -27,6 +27,27 @@ static void test_array(void)
TEST_ASSERT_TRUE(MAX_ARRAY_IDX == 32767);
}
static void test_persistence_constants(void)
{
TEST_ASSERT_TRUE(MAX_STANDBYS == 3);
TEST_ASSERT_TRUE(MAX_HA_CANDIDATES == 4);
TEST_ASSERT_TRUE(WAL_FORMAT_VERSION == 1);
TEST_ASSERT_TRUE(WAL_ENTRY_TYPE_MAX == 100);
TEST_ASSERT_TRUE(MAX_WAL_PAYLOAD_BYTES == (1 << 20));
TEST_ASSERT_TRUE(MAX_SNAPSHOT_BYTES == (256 * 1024 * 1024));
TEST_ASSERT_TRUE(DEFAULT_WAL_SYNC_MS == 10);
TEST_ASSERT_TRUE(DEFAULT_SNAP_INTERVAL_S == 60);
TEST_ASSERT_TRUE(DEFAULT_SNAP_WAL_SIZE_MB == 64);
TEST_ASSERT_TRUE(DEFAULT_SNAP_KEEP_N == 2);
TEST_ASSERT_TRUE(DEFAULT_FINISHED_RETAIN_S == 3600);
TEST_ASSERT_TRUE(DEFAULT_WORKER_RETAIN_S == 86400);
TEST_ASSERT_TRUE(DEFAULT_SYNC_TIMEOUT_MS == 50);
TEST_ASSERT_TRUE(DEFAULT_HB_INTERVAL_MS == 3000);
TEST_ASSERT_TRUE(DEFAULT_TRANSIENT_JOB_TIMEOUT_S == 300);
TEST_ASSERT_TRUE(WAL_HDR_CRC_SIZE == 32);
TEST_ASSERT_TRUE(WAL_CRC_NONE == 0xFFFFFFFFu);
}
void test_config(void)
{
printf("\n>> config test\n\n");
@ -36,4 +57,5 @@ void test_config(void)
RUN_TEST(test_timestamp);
RUN_TEST(test_tag);
RUN_TEST(test_array);
RUN_TEST(test_persistence_constants);
}

15
src/lib/testsuite/test_delay.c
View file

@ -26,7 +26,7 @@ static void test_delay1()
int delay_period = 2;
time_t t_start = time(NULL);
delaymgr_init(&mgr, 1, NULL);
delaymgr_init(&mgr, 1000 /* precision_ms */, NULL);
delaymgr_start(&mgr);
delay_task(&mgr, delay_period, runcount, (delay_task_entry_fn_t)delay_work,
&count);
@ -48,9 +48,22 @@ static void test_delay1()
return;
}
static void test_delaymgr_precision_ms_field(void)
{
delaymgr_t mgr;
memset(&mgr, 0, sizeof(mgr));
/* precision_ms field must exist and be stored correctly */
int rc = delaymgr_init(&mgr, 200 /* precision_ms */, NULL);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(200, mgr.precision_ms);
delaymgr_fini(&mgr);
printf("test_delaymgr_precision_ms_field: PASS\n");
}
void test_delay(void)
{
printf("\n>> delay test\n\n");
RUN_TEST(test_delay1);
RUN_TEST(test_delaymgr_precision_ms_field);
}

112
src/lib/testsuite/test_hooks.c Normal file
View file

@ -0,0 +1,112 @@
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include "hook.h"
#include <unity/unity.h>
/* Shared call counter for hook tests */
static int s_call_count = 0;
static int counting_fn(int type, int sit, int det, void* data, va_list ap)
{
(void)type;
(void)sit;
(void)det;
(void)ap;
(*(int*)data)++;
return 0;
}
/* --- test cases --- */
static void test_disable_hook_stops_invocation(void)
{
hooksys_t hs;
int rc;
s_call_count = 0;
rc = xce_hook_init(&hs, 2 /* maxtypes */);
TEST_ASSERT_EQUAL_INT(0, rc);
rc = xce_add_hook(&hs, HOOK_AFTER_EVENT, counting_fn, &s_call_count);
TEST_ASSERT_EQUAL_INT(0, rc);
/* Before disable: hook fires once */
xce_run_hooks(&hs, HOOK_AFTER_EVENT, 0, 0);
TEST_ASSERT_EQUAL_INT(1, s_call_count);
/* Disable the hook */
rc = xce_disable_hook(&hs, HOOK_AFTER_EVENT, counting_fn);
TEST_ASSERT_EQUAL_INT(0, rc);
/* After disable: hook must not fire */
xce_run_hooks(&hs, HOOK_AFTER_EVENT, 0, 0);
TEST_ASSERT_EQUAL_INT(1, s_call_count); /* unchanged */
xce_hook_fini(&hs);
printf("test_disable_hook_stops_invocation: PASS\n");
}
static void test_disable_hook_returns_enoent_if_not_found(void)
{
hooksys_t hs;
int rc;
rc = xce_hook_init(&hs, 2);
TEST_ASSERT_EQUAL_INT(0, rc);
/* Disabling a hook never added must return non-zero (ENOENT) */
rc = xce_disable_hook(&hs, HOOK_AFTER_EVENT, counting_fn);
TEST_ASSERT_NOT_EQUAL(0, rc);
xce_hook_fini(&hs);
printf("test_disable_hook_returns_enoent_if_not_found: PASS\n");
}
static void test_add_hook_reuses_null_slot(void)
{
hooksys_t hs;
int rc;
int count_before_reuse;
s_call_count = 0;
rc = xce_hook_init(&hs, 2);
TEST_ASSERT_EQUAL_INT(0, rc);
rc = xce_add_hook(&hs, HOOK_AFTER_EVENT, counting_fn, &s_call_count);
TEST_ASSERT_EQUAL_INT(0, rc);
/* Record list length after first add */
count_before_reuse = hs.hooks[HOOK_AFTER_EVENT].count;
TEST_ASSERT_EQUAL_INT(1, count_before_reuse);
/* Disable the slot — fn becomes NULL atomically */
rc = xce_disable_hook(&hs, HOOK_AFTER_EVENT, counting_fn);
TEST_ASSERT_EQUAL_INT(0, rc);
/* Re-add the same fn — must reuse the NULL slot, not grow the list */
rc = xce_add_hook(&hs, HOOK_AFTER_EVENT, counting_fn, &s_call_count);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(count_before_reuse, hs.hooks[HOOK_AFTER_EVENT].count);
/* Fire: must get exactly 1 call (not 2 if slot was duplicated) */
xce_run_hooks(&hs, HOOK_AFTER_EVENT, 0, 0);
TEST_ASSERT_EQUAL_INT(1, s_call_count);
xce_hook_fini(&hs);
printf("test_add_hook_reuses_null_slot: PASS\n");
}
/* --- suite entry point --- */
void test_hooks(void)
{
printf("\n>> hook system tests\n\n");
RUN_TEST(test_disable_hook_stops_invocation);
RUN_TEST(test_disable_hook_returns_enoent_if_not_found);
RUN_TEST(test_add_hook_reuses_null_slot);
}

244
src/lib/testsuite/test_kvmgr_capture.c Normal file
View file

@ -0,0 +1,244 @@
/*
* Test suite for kvmgr_capture_keys() — P0-14a.
*
* Tests follow TDD: written BEFORE implementation.
* All tests exercise the public API declared in kvmgr.h.
*/
#include <rtti.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "kv.h"
#include "kvmgr.h"
#include "persist.h"
#include "utils.h"
/* ------------------------------------------------------------------ */
/* helpers */
/* ------------------------------------------------------------------ */
static int find_key_in_captured(slot_key_t* keys, size_t n,
const char* target_key, int* index_out)
{
size_t i;
for (i = 0; i < n; i++) {
if (strcmp(keys[i].key, target_key) == 0) {
if (index_out != NULL) {
*index_out = keys[i].index;
}
return 1;
}
}
return 0;
}
/* ------------------------------------------------------------------ */
/* test 1: capture from a string-keyed kvmgr with 3 entries */
/* ------------------------------------------------------------------ */
static void test_capture_keys_string_kvmgr(void)
{
kvmgr_params_t params = { .key_type = KVMGR_KEY_STRING,
.nslots = 8,
.delta = 4,
.datakiller = NULL };
var kvmgr = makeInstance(&xkvmgr_type, &params);
TEST_ASSERT_NOT_NULL(kvmgr);
/* Add 3 string-keyed entries whose values are arbitrary pointers;
the index returned by key_add is the slot index we capture. */
int idx_a = key_add(kvmgr, (void*)"alpha", (void*)"data_alpha");
int idx_b = key_add(kvmgr, (void*)"beta", (void*)"data_beta");
int idx_c = key_add(kvmgr, (void*)"gamma", (void*)"data_gamma");
TEST_ASSERT_TRUE(idx_a >= 0);
TEST_ASSERT_TRUE(idx_b >= 0);
TEST_ASSERT_TRUE(idx_c >= 0);
slot_key_t* keys = NULL;
size_t n_keys = 0;
int rc = kvmgr_capture_keys(kvmgr, &keys, &n_keys);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_size_t(3, n_keys);
TEST_ASSERT_NOT_NULL(keys);
/* Every inserted key must appear exactly once with the correct index */
int found_idx = -1;
TEST_ASSERT_TRUE(find_key_in_captured(keys, n_keys, "alpha", &found_idx));
TEST_ASSERT_EQUAL_INT(idx_a, found_idx);
TEST_ASSERT_TRUE(find_key_in_captured(keys, n_keys, "beta", &found_idx));
TEST_ASSERT_EQUAL_INT(idx_b, found_idx);
TEST_ASSERT_TRUE(find_key_in_captured(keys, n_keys, "gamma", &found_idx));
TEST_ASSERT_EQUAL_INT(idx_c, found_idx);
/* key_len must equal strlen(key)+1 */
size_t i;
for (i = 0; i < n_keys; i++) {
TEST_ASSERT_EQUAL_size_t(strlen(keys[i].key) + 1, keys[i].key_len);
}
free(keys);
destroy(kvmgr);
}
/* ------------------------------------------------------------------ */
/* test 2: empty kvmgr returns 0 entries and NULL keys_out */
/* ------------------------------------------------------------------ */
static void test_capture_keys_empty_kvmgr(void)
{
kvmgr_params_t params = { .key_type = KVMGR_KEY_STRING,
.nslots = 4,
.delta = 4,
.datakiller = NULL };
var kvmgr = makeInstance(&xkvmgr_type, &params);
TEST_ASSERT_NOT_NULL(kvmgr);
slot_key_t* keys = (slot_key_t*)(uintptr_t)0xdeadbeef; /* sentinel */
size_t n_keys = 99;
int rc = kvmgr_capture_keys(kvmgr, &keys, &n_keys);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_size_t(0, n_keys);
TEST_ASSERT_NULL(keys);
destroy(kvmgr);
}
/* ------------------------------------------------------------------ */
/* test 3: captured keys are deep copies — safe after kvmgr destroyed */
/* ------------------------------------------------------------------ */
static void test_capture_keys_are_deep_copies(void)
{
kvmgr_params_t params = { .key_type = KVMGR_KEY_STRING,
.nslots = 4,
.delta = 4,
.datakiller = NULL };
var kvmgr = makeInstance(&xkvmgr_type, &params);
TEST_ASSERT_NOT_NULL(kvmgr);
key_add(kvmgr, (void*)"keyA", (void*)"valA");
key_add(kvmgr, (void*)"keyB", (void*)"valB");
slot_key_t* keys = NULL;
size_t n_keys = 0;
int rc = kvmgr_capture_keys(kvmgr, &keys, &n_keys);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_size_t(2, n_keys);
/* Stash manual copies of the captured key strings */
char copy_a[256], copy_b[256];
int found_a = 0, found_b = 0;
size_t i;
for (i = 0; i < n_keys; i++) {
if (strcmp(keys[i].key, "keyA") == 0) {
strncpy(copy_a, keys[i].key, sizeof(copy_a) - 1);
copy_a[sizeof(copy_a) - 1] = '\0';
found_a = 1;
} else if (strcmp(keys[i].key, "keyB") == 0) {
strncpy(copy_b, keys[i].key, sizeof(copy_b) - 1);
copy_b[sizeof(copy_b) - 1] = '\0';
found_b = 1;
}
}
TEST_ASSERT_TRUE(found_a);
TEST_ASSERT_TRUE(found_b);
/* Destroy the kvmgr — this frees all kvtab entries including the
original key strings. Captured keys[] must remain valid. */
destroy(kvmgr);
TEST_ASSERT_EQUAL_STRING("keyA", copy_a);
TEST_ASSERT_EQUAL_STRING("keyB", copy_b);
/* Directly verify the captured keys array is still readable */
int still_a = 0, still_b = 0;
for (i = 0; i < n_keys; i++) {
if (strcmp(keys[i].key, "keyA") == 0) still_a = 1;
if (strcmp(keys[i].key, "keyB") == 0) still_b = 1;
}
TEST_ASSERT_TRUE(still_a);
TEST_ASSERT_TRUE(still_b);
free(keys);
}
/* ------------------------------------------------------------------ */
/* test 4: capture from an integer-keyed kvmgr with 3 entries */
/* ------------------------------------------------------------------ */
static void test_capture_keys_int_kvmgr(void)
{
kvmgr_params_t params = { .key_type = KVMGR_KEY_INT,
.nslots = 8,
.delta = 4,
.datakiller = NULL };
var kvmgr = makeInstance(&xkvmgr_type, &params);
TEST_ASSERT_NOT_NULL(kvmgr);
/* Add 3 integer-keyed entries; the integer keys are jobids.
For integer-keyed kvmgrs, key_add expects a (void*)(long)intval. */
int64_t jobid_1001 = 1001;
int64_t jobid_1002 = 1002;
int64_t jobid_1003 = 1003;
int idx_1001 =
key_add(kvmgr, (void*)(long)jobid_1001, (void*)"data_1001");
int idx_1002 =
key_add(kvmgr, (void*)(long)jobid_1002, (void*)"data_1002");
int idx_1003 =
key_add(kvmgr, (void*)(long)jobid_1003, (void*)"data_1003");
TEST_ASSERT_TRUE(idx_1001 >= 0);
TEST_ASSERT_TRUE(idx_1002 >= 0);
TEST_ASSERT_TRUE(idx_1003 >= 0);
slot_key_t* keys = NULL;
size_t n_keys = 0;
int rc = kvmgr_capture_keys(kvmgr, &keys, &n_keys);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_size_t(3, n_keys);
TEST_ASSERT_NOT_NULL(keys);
/* Every inserted integer key must appear as a decimal string with the
correct index */
int found_idx = -1;
TEST_ASSERT_TRUE(find_key_in_captured(keys, n_keys, "1001", &found_idx));
TEST_ASSERT_EQUAL_INT(idx_1001, found_idx);
TEST_ASSERT_TRUE(find_key_in_captured(keys, n_keys, "1002", &found_idx));
TEST_ASSERT_EQUAL_INT(idx_1002, found_idx);
TEST_ASSERT_TRUE(find_key_in_captured(keys, n_keys, "1003", &found_idx));
TEST_ASSERT_EQUAL_INT(idx_1003, found_idx);
/* key_len must equal strlen(key)+1 */
size_t i;
for (i = 0; i < n_keys; i++) {
TEST_ASSERT_EQUAL_size_t(strlen(keys[i].key) + 1, keys[i].key_len);
}
free(keys);
destroy(kvmgr);
}
/* ------------------------------------------------------------------ */
/* entry point called from test.c */
/* ------------------------------------------------------------------ */
void test_kvmgr_capture(void)
{
printf("\n>> kvmgr_capture_keys test\n");
RUN_TEST(test_capture_keys_string_kvmgr);
RUN_TEST(test_capture_keys_empty_kvmgr);
RUN_TEST(test_capture_keys_are_deep_copies);
RUN_TEST(test_capture_keys_int_kvmgr);
}

136
src/lib/testsuite/test_zarray_cow.c Normal file
View file

@ -0,0 +1,136 @@
#include <assert.h>
#include <rtti.h>
#include <stdio.h>
#include <unity/unity.h>
#include "zarray.h"
/* xzarray_type is defined in zarray.c */
extern typeinfo_t xzarray_type;
/* Helper: create an xzarray instance */
static meta_obj_t* make_test_xzarray(void)
{
xzarray_params_t params = { .delta = 4, .datakiller = NULL };
return (meta_obj_t*)makeInstance(&xzarray_type, &params);
}
static void test_freeze_thaw_slots_exist(void)
{
/* After P0-2, vfreeze and vthaw must be wired (non-NULL). */
const intf_varray_t* iface
= (const intf_varray_t*)xzarray_type.interfaces.varray;
TEST_ASSERT_NOT_NULL(iface);
/* Access slots without crash — verifies struct layout has them */
(void)iface->vfreeze;
(void)iface->vthaw;
/* Slots must be non-NULL after P0-2 wires the real implementations */
TEST_ASSERT_NOT_NULL(iface->vfreeze);
TEST_ASSERT_NOT_NULL(iface->vthaw);
printf("test_freeze_thaw_slots_exist: PASS\n");
}
static void test_freeze_creates_shadow(void)
{
meta_obj_t* arr = make_test_xzarray();
TEST_ASSERT_NOT_NULL(arr);
/* Store 3 items */
char* items[3] = { "alpha", "beta", "gamma" };
for (int i = 0; i < 3; i++) {
int idx = vstore(arr, items[i]);
TEST_ASSERT_TRUE(idx >= 0);
}
/* Freeze should succeed */
TEST_ASSERT_EQUAL_INT(0, vfreeze(arr));
/* After freeze, vget must still return stored values (COW mode active) */
void* out = NULL;
TEST_ASSERT_EQUAL_INT(0, vget(arr, 0, &out));
TEST_ASSERT_NOT_NULL(out);
/* Thaw to clean up */
vthaw(arr, NULL, NULL);
destroy(arr);
}
static void test_freeze_thaw_preserves_all_entries(void)
{
meta_obj_t* arr = make_test_xzarray();
TEST_ASSERT_NOT_NULL(arr);
int N = 5;
char* items[5] = { "a", "b", "c", "d", "e" };
int idxs[5];
for (int i = 0; i < N; i++) {
idxs[i] = vstore(arr, items[i]);
TEST_ASSERT_TRUE(idxs[i] >= 0);
}
TEST_ASSERT_EQUAL_INT(0, vfreeze(arr));
TEST_ASSERT_EQUAL_INT(0, vthaw(arr, NULL, NULL));
/* All 5 entries must be reachable after thaw */
for (int i = 0; i < N; i++) {
void* out = NULL;
TEST_ASSERT_EQUAL_INT(0, vget(arr, idxs[i], &out));
TEST_ASSERT_EQUAL_PTR(items[i], out);
}
destroy(arr);
}
static void test_mutation_during_cow_visible_after_thaw(void)
{
meta_obj_t* arr = make_test_xzarray();
TEST_ASSERT_NOT_NULL(arr);
char* original = "original";
char* updated = "updated";
int idx = vstore(arr, original);
TEST_ASSERT_TRUE(idx >= 0);
TEST_ASSERT_EQUAL_INT(0, vfreeze(arr));
/* Overwrite index during COW — write to new active array */
int put_idxs[1] = { idx };
void* put_data[1] = { updated };
vput_vec(arr, 1, put_idxs, put_data);
TEST_ASSERT_EQUAL_INT(0, vthaw(arr, NULL, NULL));
/* After thaw, the updated value must win over the shadow value */
void* out = NULL;
TEST_ASSERT_EQUAL_INT(0, vget(arr, idx, &out));
TEST_ASSERT_EQUAL_PTR(updated, out);
destroy(arr);
}
static void test_double_freeze_fails(void)
{
meta_obj_t* arr = make_test_xzarray();
TEST_ASSERT_NOT_NULL(arr);
TEST_ASSERT_EQUAL_INT(0, vfreeze(arr));
/* Second freeze must fail (already in COW mode) */
TEST_ASSERT_TRUE(vfreeze(arr) != 0);
vthaw(arr, NULL, NULL);
destroy(arr);
}
void test_zarray_cow(void)
{
printf("\n>> zarray COW interface slot test\n\n");
RUN_TEST(test_freeze_thaw_slots_exist);
RUN_TEST(test_freeze_creates_shadow);
RUN_TEST(test_freeze_thaw_preserves_all_entries);
RUN_TEST(test_mutation_during_cow_visible_after_thaw);
RUN_TEST(test_double_freeze_fails);
}

24
src/lib/xcectx.c
View file

@ -133,6 +133,29 @@ static int handle_msg(endpoint_t* ep)
(void*)ep);
}
/* Called by netio when the TCP connection drops (EPOLLHUP / EPOLLERR / read EOF).
* Unblocks any thread stuck in cbuf_get (e.g. xce_hello waiting for a reply)
* so that checking_connection can detect the failure and attempt to rejoin. */
static int handle_disconnect(endpoint_t* ep)
{
if (ep == NULL)
return 0;
xce_ctx_t* ctx = (xce_ctx_t*)ep->owner;
if (ctx == NULL)
return 0;
ctx->state = XCE_CTX_ERR;
/* Inject an error into wbuf so any blocked cbuf_get returns immediately.
* cbuf_tryput returns -EAGAIN if the buffer already has an item — that is
* fine: the existing item will wake the waiter. */
int err = -ECONNRESET;
cbuf_tryput(&ctx->wbuf, &err);
return 0;
}
int xce_ctx_init(xce_ctx_t* ctx, char* name, char* key)
{
int rc;
@ -237,6 +260,7 @@ int xce_ctx_start(xce_ctx_t* ctx)
endpoint_set_owner(ctx->ep, ctx);
endpoint_set_msg_handler(ctx->ep, (msg_handler_t)handle_msg);
endpoint_set_err_handler(ctx->ep, (err_handler_t)handle_disconnect);
rc = iomgr_monitor(ctx->ep);
if (rc < 0) {
ctx->state = XCE_CTX_ERR;

148
src/lib/zarray.c
View file

@ -1,5 +1,7 @@
#include <assert.h>
#include <errno.h>
#include <rtti.h>
#include <stdatomic.h>
#include <stdlib.h>
#include <string.h>
@ -351,13 +353,16 @@ void* zarray_get(zarray_t* array, int idx)
}
typedef struct {
mylock_t lock;
} xzarray_private_t;
_Atomic _Bool inuse;
void* data;
} xzarray_item_t;
typedef struct {
unsigned int inuse : 1;
void* data;
} xzarray_item_t;
mylock_t lock;
xzarray_item_t* shadow; /* COW snapshot (may be NULL if array was empty at freeze) */
size_t shadow_caps; /* capacity of shadow array */
bool in_cow; /* true while between freeze and thaw */
} xzarray_private_t;
typedef struct {
int count;
@ -392,6 +397,10 @@ static int xzarray_init(meta_obj_t* obj, va_list ap)
a->pos = 0;
a->killer = params->datakiller;
priv->shadow = NULL;
priv->shadow_caps = 0;
priv->in_cow = false;
mylock_init(&(priv->lock));
return 1;
@ -403,8 +412,16 @@ static void xzarray_destroy(meta_obj_t* obj)
xzarray_t* a = RTTI_OBJ_TO_DATA(xzarray_t, obj);
int i;
for (i = 0; i < a->caps; i++) {
if (a->data[i].inuse) {
/* Free shadow if still present (e.g. destroy called mid-COW) */
if (priv->in_cow) {
free(priv->shadow);
priv->shadow = NULL;
priv->shadow_caps = 0;
priv->in_cow = false;
}
for (i = 0; i < (int)a->caps; i++) {
if (atomic_load_explicit(&a->data[i].inuse, memory_order_acquire)) {
if (a->killer != NULL) {
a->killer(a->data[i].data);
}
@ -437,14 +454,14 @@ static int xzarray_ensure_items(xzarray_t* array, int mincap)
array->data = p;
for (i = array->caps; i < nc; i++) {
array->data[i].inuse = 0;
atomic_store_explicit(&array->data[i].inuse, false, memory_order_release);
array->data[i].data = NULL;
}
/* this will make we can get next free
* location quickly
*/
if (array->data[array->pos].inuse) {
if (atomic_load_explicit(&array->data[array->pos].inuse, memory_order_acquire)) {
array->pos = array->caps;
}
@ -467,9 +484,9 @@ static int xzarray_store(meta_obj_t* obj, void* data)
idx = a->pos;
for (i = 0; i < a->caps; i++) {
if (a->data[idx].inuse == 0) {
a->data[idx].inuse = 1;
a->data[idx].data = data;
if (!atomic_load_explicit(&a->data[idx].inuse, memory_order_acquire)) {
a->data[idx].data = data;
atomic_store_explicit(&a->data[idx].inuse, true, memory_order_release);
a->count++;
rc = idx;
@ -503,7 +520,15 @@ static int xzarray_get_vec(meta_obj_t* obj, int count, int* idxs, void** data)
for (i = 0; i < count; i++) {
int idx = idxs[i];
if ((idx >= a->caps) || (a->data[idx].inuse == 0)) {
/* During COW, check active first, then shadow */
bool in_active = (idx < (int)a->caps) &&
atomic_load_explicit(&a->data[idx].inuse, memory_order_acquire);
bool in_shadow = !in_active && priv->in_cow &&
priv->shadow != NULL &&
(idx < (int)priv->shadow_caps) &&
atomic_load_explicit(&priv->shadow[idx].inuse, memory_order_acquire);
if (!in_active && !in_shadow) {
rc = -ENOENT;
break;
}
@ -513,7 +538,13 @@ static int xzarray_get_vec(meta_obj_t* obj, int count, int* idxs, void** data)
for (i = 0; i < count; i++) {
int idx = idxs[i];
data[i] = a->data[idx].data;
/* Active takes precedence over shadow */
if ((idx < (int)a->caps) &&
atomic_load_explicit(&a->data[idx].inuse, memory_order_acquire)) {
data[i] = a->data[idx].data;
} else {
data[i] = priv->shadow[idx].data;
}
}
}
return rc;
@ -557,14 +588,16 @@ static int xzarray_put_vec(meta_obj_t* obj, int count, int* idxs, void** data)
for (i = 0; i < count; i++) {
int idx = idxs[i];
if (a->data[idx].inuse == 1) {
if (atomic_load_explicit(&a->data[idx].inuse, memory_order_acquire)) {
if (a->killer != NULL) {
a->killer(a->data[idx].data);
}
} else {
a->count++;
}
a->data[idx].data = data[i];
a->data[idx].inuse = 1;
a->data[idx].data = data[i];
atomic_store_explicit(&a->data[idx].inuse, true, memory_order_release);
}
}
@ -584,15 +617,18 @@ static int xzarray_clear(meta_obj_t* obj, int idx)
xzarray_private_t* priv = (xzarray_private_t*)RTTI_OBJ_TO_PRIV(obj);
int rc = -EINVAL;
if (idx < a->caps) {
assert(!priv->in_cow && "xzarray_clear called during COW — deferred cleanup required");
if (idx < (int)a->caps) {
rc = 0;
if (a->data[idx].inuse) {
if (atomic_load_explicit(&a->data[idx].inuse, memory_order_acquire)) {
if (a->killer != NULL) {
a->killer(a->data[idx].data);
}
a->count--;
}
a->data[idx].inuse = 0;
atomic_store_explicit(&a->data[idx].inuse, false, memory_order_release);
a->data[idx].data = NULL;
a->pos = idx; /* we can find the available idx next time */
}
@ -614,8 +650,8 @@ static int xzarray_iter_next(iterator_t* iter, void** data)
xzarray_t* a = RTTI_OBJ_TO_DATA(xzarray_t, iter->obj);
int i;
for (i = (int)(long)iter->data; i < a->caps; i++) {
if (a->data[i].inuse) {
for (i = (int)(long)iter->data; i < (int)a->caps; i++) {
if (atomic_load_explicit(&a->data[i].inuse, memory_order_acquire)) {
*data = a->data[i].data;
iter->data = (void*)(long)(i + 1);
@ -656,6 +692,70 @@ static void xzarray_iter_destroy(iterator_t* iter)
return;
}
/* P0-2: COW freeze — O(1) pointer swap, pre-grow active array */
static int xzarray_freeze(meta_obj_t* obj)
{
xzarray_private_t* priv = (xzarray_private_t*)RTTI_OBJ_TO_PRIV(obj);
xzarray_t* a = RTTI_OBJ_TO_DATA(xzarray_t, obj);
if (priv->in_cow)
return -EALREADY; /* already frozen */
/* Pre-grow active to 1.5x + 16 to avoid realloc during COW (A6-026) */
size_t new_caps = a->caps + (a->caps / 2) + 16;
xzarray_item_t* active = calloc(new_caps, sizeof(xzarray_item_t));
if (!active)
return -ENOMEM;
priv->shadow = a->data; /* may be NULL if array was empty */
priv->shadow_caps = a->caps;
priv->in_cow = true;
a->data = active;
a->caps = new_caps;
a->count = 0;
/* Start pos past shadow to prevent index collisions with shadow entries */
a->pos = (int)priv->shadow_caps;
return 0;
}
/* P0-2: COW thaw — merge shadow into active, free shadow */
static int xzarray_thaw(meta_obj_t* obj, retain_fn_t should_retain, void* arg)
{
xzarray_private_t* priv = (xzarray_private_t*)RTTI_OBJ_TO_PRIV(obj);
xzarray_t* a = RTTI_OBJ_TO_DATA(xzarray_t, obj);
if (!priv->in_cow)
return -EINVAL; /* not in COW mode */
/* Merge: for each shadow entry not already written to active, copy it */
for (size_t i = 0; i < priv->shadow_caps; i++) {
if (!atomic_load_explicit(&priv->shadow[i].inuse, memory_order_acquire))
continue;
void* item = priv->shadow[i].data;
/* GC: skip entries that should not be retained */
if (should_retain != NULL && !should_retain(item, arg))
continue;
/* Only copy to active if active slot not already updated during COW */
if (i < a->caps &&
!atomic_load_explicit(&a->data[i].inuse, memory_order_acquire)) {
a->data[i].data = item;
atomic_store_explicit(&a->data[i].inuse, true, memory_order_release);
a->count++;
}
/* If active[i].inuse == true: mutation during COW wins — keep it */
}
free(priv->shadow);
priv->shadow = NULL;
priv->shadow_caps = 0;
priv->in_cow = false;
a->pos = 0;
return 0;
}
typeinfo_t xzarray_type = {
.name = "xzarray",
.privateSize = sizeof(xzarray_private_t),
@ -668,7 +768,9 @@ typeinfo_t xzarray_type = {
.vget_vec = (vget_vec_fn_t)xzarray_get_vec,
.vput = (vput_fn_t)xzarray_put,
.vput_vec = (vput_vec_fn_t)xzarray_put_vec,
.vclear = (vclear_fn_t)xzarray_clear },
.vclear = (vclear_fn_t)xzarray_clear,
.vfreeze = (vfreeze_fn_t)xzarray_freeze,
.vthaw = (vthaw_fn_t)xzarray_thaw },
.iter
= &(intf_iterator_t) { .make = (iter_make_fn_t)xzarray_make_iter,
.next = (iter_next_fn_t)xzarray_iter_next,