diff options
Diffstat (limited to 'src/ipcpd/unicast/dir')
| -rw-r--r-- | src/ipcpd/unicast/dir/dht.c | 424 | ||||
| -rw-r--r-- | src/ipcpd/unicast/dir/dht.h | 2 | ||||
| -rw-r--r-- | src/ipcpd/unicast/dir/dht.proto | 2 | ||||
| -rw-r--r-- | src/ipcpd/unicast/dir/ops.h | 2 | ||||
| -rw-r--r-- | src/ipcpd/unicast/dir/pol.h | 2 | ||||
| -rw-r--r-- | src/ipcpd/unicast/dir/tests/CMakeLists.txt | 44 | ||||
| -rw-r--r-- | src/ipcpd/unicast/dir/tests/dht_test.c | 47 |
7 files changed, 226 insertions, 297 deletions
diff --git a/src/ipcpd/unicast/dir/dht.c b/src/ipcpd/unicast/dir/dht.c index 6b06def9..8eeea800 100644 --- a/src/ipcpd/unicast/dir/dht.c +++ b/src/ipcpd/unicast/dir/dht.c @@ -1,5 +1,5 @@ /* - * Ouroboros - Copyright (C) 2016 - 2024 + * Ouroboros - Copyright (C) 2016 - 2026 * * Distributed Hash Table based on Kademlia * @@ -148,15 +148,8 @@ struct dht_entry { uint8_t * key; - struct { - struct list_head list; - size_t len; - } vals; /* We don't own these, only replicate */ - - struct { - struct list_head list; - size_t len; - } lvals; /* We own these, must be republished */ + struct llist vals; /* We don't own these, only replicate */ + struct llist lvals; /* We own these, must be republished */ }; struct contact { @@ -183,38 +176,24 @@ struct peer_entry { struct dht_req { struct list_head next; - uint8_t * key; - time_t t_exp; + uint8_t * key; + time_t t_exp; - struct { - struct list_head list; - size_t len; - } peers; - - struct { - struct list_head list; - size_t len; - } cache; + struct llist peers; + struct llist cache; }; struct bucket { - struct { - struct list_head list; - size_t len; - } contacts; + struct llist contacts; + struct llist alts; - struct { - struct list_head list; - size_t len; - } alts; + time_t t_refr; - time_t t_refr; + size_t depth; + uint8_t mask; - size_t depth; - uint8_t mask; - - struct bucket * parent; - struct bucket * children[1L << DHT_BETA]; + struct bucket * parent; + struct bucket * children[1L << DHT_BETA]; }; struct cmd { @@ -236,8 +215,8 @@ struct { struct { /* Kademlia parameters */ uint32_t alpha; /* Number of concurrent requests */ size_t k; /* Number of replicas to store */ - time_t t_expire; /* Expiry time for values (s) */ - time_t t_refresh; /* Refresh time for contacts (s) */ + time_t t_exp; /* Expiry time for values (s) */ + time_t t_refr; /* Refresh time for contacts (s) */ time_t t_repl; /* Replication time for values (s) */ }; @@ -261,8 +240,7 @@ struct { } contacts; struct { - struct list_head list; - size_t len; + struct llist ll; size_t vals; size_t lvals; } kv; @@ -271,10 +249,9 @@ struct { } db; struct { - struct list_head list; - size_t len; - pthread_cond_t cond; - pthread_mutex_t mtx; + struct llist ll; + pthread_cond_t cond; + pthread_mutex_t mtx; } reqs; struct { @@ -321,7 +298,7 @@ static int dht_rib_statfile(char * buf, pthread_rwlock_rdlock(&dht.db.lock); - keys = dht.db.kv.len; + keys = dht.db.kv.ll.len; lvals = dht.db.kv.lvals; vals = dht.db.kv.vals; @@ -335,7 +312,7 @@ static int dht_rib_statfile(char * buf, tmstr, ADDR_VAL32(&dht.addr), dht.alpha, dht.k, - dht.t_expire, dht.t_refresh, dht.t_repl, + dht.t_exp, dht.t_refr, dht.t_repl, keys, vals, lvals); return strlen(buf); @@ -350,14 +327,14 @@ static size_t dht_db_file_len(void) pthread_rwlock_rdlock(&dht.db.lock); - if (dht.db.kv.len == 0) { + if (llist_is_empty(&dht.db.kv.ll)) { pthread_rwlock_unlock(&dht.db.lock); sz += 14; /* No entries */ return sz; } sz += 39 * 3 + 1; /* tally + extra newline */ - sz += dht.db.kv.len * (25 + 19 + 23 + 1); + sz += dht.db.kv.ll.len * (25 + 19 + 23 + 1); vals = dht.db.kv.vals + dht.db.kv.lvals; @@ -382,7 +359,7 @@ static int dht_rib_dbfile(char * buf, pthread_rwlock_rdlock(&dht.db.lock); - if (dht.db.kv.len == 0) { + if (llist_is_empty(&dht.db.kv.ll)) { i += snprintf(buf, len, " No entries.\n"); pthread_rwlock_unlock(&dht.db.lock); return i; @@ -393,9 +370,9 @@ static int dht_rib_dbfile(char * buf, "Number of keys: %10zu\n" "Number of local values: %10zu\n" "Number of non-local values: %10zu\n\n", - dht.db.kv.len, dht.db.kv.vals, dht.db.kv.lvals); + dht.db.kv.ll.len, dht.db.kv.vals, dht.db.kv.lvals); - list_for_each(p, &dht.db.kv.list) { + llist_for_each(p, &dht.db.kv.ll) { struct dht_entry * e = list_entry(p, struct dht_entry, next); struct list_head * h; @@ -403,7 +380,7 @@ static int dht_rib_dbfile(char * buf, KEY_VAL(e->key)); i += snprintf(buf + i, len - i, " Local entries:\n"); - list_for_each(h, &e->vals.list) { + llist_for_each(h, &e->vals) { struct val_entry * v; v = list_entry(h, struct val_entry, next); @@ -416,7 +393,7 @@ static int dht_rib_dbfile(char * buf, i += snprintf(buf + i, len - i, " " VAL_FMT - ", t_replicated=%.*s, t_expire=%.*s\n", + ", t_replicated=%.*s, t_exp=%.*s\n", VAL_VAL(v->val), RIB_TM_STRLEN, tmstr, RIB_TM_STRLEN, exstr); @@ -426,7 +403,7 @@ static int dht_rib_dbfile(char * buf, i += snprintf(buf + i, len - i, " Non-local entries:\n"); - list_for_each(h, &e->lvals.list) { + llist_for_each(h, &e->lvals) { struct val_entry * v; v= list_entry(h, struct val_entry, next); @@ -439,7 +416,7 @@ static int dht_rib_dbfile(char * buf, i += snprintf(buf + i, len - i, " " VAL_FMT - ", t_replicated=%.*s, t_expire=%.*s\n", + ", t_replicated=%.*s, t_exp=%.*s\n", VAL_VAL(v->val), RIB_TM_STRLEN, tmstr, RIB_TM_STRLEN, exstr); @@ -694,11 +671,8 @@ static struct dht_entry * dht_entry_create(const uint8_t * key) goto fail_entry; list_head_init(&e->next); - list_head_init(&e->vals.list); - list_head_init(&e->lvals.list); - - e->vals.len = 0; - e->lvals.len = 0; + llist_init(&e->vals); + llist_init(&e->lvals); e->key = dht_dup_key(key); if (e->key == NULL) @@ -718,25 +692,23 @@ static void dht_entry_destroy(struct dht_entry * e) assert(e != NULL); - list_for_each_safe(p, h, &e->vals.list) { + llist_for_each_safe(p, h, &e->vals) { struct val_entry * v = list_entry(p, struct val_entry, next); - list_del(&v->next); + llist_del(&v->next, &e->vals); val_entry_destroy(v); - --e->vals.len; --dht.db.kv.vals; } - list_for_each_safe(p, h, &e->lvals.list) { + llist_for_each_safe(p, h, &e->lvals) { struct val_entry * v = list_entry(p, struct val_entry, next); - list_del(&v->next); + llist_del(&v->next, &e->lvals); val_entry_destroy(v); - --e->lvals.len; --dht.db.kv.lvals; } free(e->key); - assert(e->vals.len == 0 && e->lvals.len == 0); + assert(llist_is_empty(&e->vals) && llist_is_empty(&e->lvals)); free(e); } @@ -750,7 +722,7 @@ static struct val_entry * dht_entry_get_lval(const struct dht_entry * e, assert(val.data != NULL); assert(val.len > 0); - list_for_each(p, &e->lvals.list) { + llist_for_each(p, &e->lvals) { struct val_entry * v = list_entry(p, struct val_entry, next); if (bufcmp(&v->val, &val) == 0) return v; @@ -768,7 +740,7 @@ static struct val_entry * dht_entry_get_val(const struct dht_entry * e, assert(val.data != NULL); assert(val.len > 0); - list_for_each(p, &e->vals.list) { + llist_for_each(p, &e->vals) { struct val_entry * v = list_entry(p, struct val_entry, next); if (bufcmp(&v->val, &val) == 0) return v; @@ -805,8 +777,7 @@ static int dht_entry_update_val(struct dht_entry * e, if (v == NULL) return -ENOMEM; - list_add_tail(&v->next, &e->vals.list); - ++e->vals.len; + llist_add_tail(&v->next, &e->vals); ++dht.db.kv.vals; return 0; @@ -833,12 +804,11 @@ static int dht_entry_update_lval(struct dht_entry * e, v = dht_entry_get_lval(e, val); if (v == NULL) { log_dbg(KV_FMT " Adding lval.", KV_VAL(e->key, val)); - v = val_entry_create(val, now.tv_sec + dht.t_expire); + v = val_entry_create(val, now.tv_sec + dht.t_exp); if (v == NULL) return -ENOMEM; - list_add_tail(&v->next, &e->lvals.list); - ++e->lvals.len; + llist_add_tail(&v->next, &e->lvals); ++dht.db.kv.lvals; return 0; @@ -862,9 +832,8 @@ static int dht_entry_remove_lval(struct dht_entry * e, log_dbg(KV_FMT " Removing lval.", KV_VAL(e->key, val)); - list_del(&v->next); + llist_del(&v->next, &e->lvals); val_entry_destroy(v); - --e->lvals.len; --dht.db.kv.lvals; return 0; @@ -881,15 +850,14 @@ static void dht_entry_remove_expired_vals(struct dht_entry * e) clock_gettime(CLOCK_REALTIME_COARSE, &now); - list_for_each_safe(p, h, &e->vals.list) { + llist_for_each_safe(p, h, &e->vals) { struct val_entry * v = list_entry(p, struct val_entry, next); if (!IS_EXPIRED(v, &now)) continue; log_dbg(KV_FMT " Value expired." , KV_VAL(e->key, v->val)); - list_del(&v->next); + llist_del(&v->next, &e->vals); val_entry_destroy(v); - --e->vals.len; --dht.db.kv.vals; } } @@ -900,7 +868,7 @@ static struct dht_entry * __dht_kv_find_entry(const uint8_t * key) assert(key != NULL); - list_for_each(p, &dht.db.kv.list) { + llist_for_each(p, &dht.db.kv.ll) { struct dht_entry * e = list_entry(p, struct dht_entry, next); if (!memcmp(key, e->key, dht.id.len)) return e; @@ -919,16 +887,15 @@ static void dht_kv_remove_expired_entries(void) pthread_rwlock_wrlock(&dht.db.lock); - list_for_each_safe(p, h, &dht.db.kv.list) { + llist_for_each_safe(p, h, &dht.db.kv.ll) { struct dht_entry * e = list_entry(p, struct dht_entry, next); dht_entry_remove_expired_vals(e); if (e->lvals.len > 0 || e->vals.len > 0) continue; log_dbg(KEY_FMT " Entry removed. ", KEY_VAL(e->key)); - list_del(&e->next); + llist_del(&e->next, &dht.db.kv.ll); dht_entry_destroy(e); - --dht.db.kv.len; } pthread_rwlock_unlock(&dht.db.lock); @@ -987,15 +954,13 @@ static struct dht_req * dht_req_create(const uint8_t * key) req->t_exp = now.tv_sec + DHT_T_RESP; - list_head_init(&req->peers.list); - req->peers.len = 0; + llist_init(&req->peers); req->key = dht_dup_key(key); if (req->key == NULL) goto fail_dup_key; - list_head_init(&req->cache.list); - req->cache.len = 0; + llist_init(&req->cache); return req; @@ -1013,34 +978,32 @@ static void dht_req_destroy(struct dht_req * req) assert(req); assert(req->key); - list_for_each_safe(p, h, &req->peers.list) { + llist_for_each_safe(p, h, &req->peers) { struct peer_entry * e = list_entry(p, struct peer_entry, next); - list_del(&e->next); + llist_del(&e->next, &req->peers); free(e->id); free(e); - --req->peers.len; } - list_for_each_safe(p, h, &req->cache.list) { + llist_for_each_safe(p, h, &req->cache) { struct val_entry * e = list_entry(p, struct val_entry, next); - list_del(&e->next); + llist_del(&e->next, &req->cache); val_entry_destroy(e); - --req->cache.len; } free(req->key); - assert(req->peers.len == 0); + assert(llist_is_empty(&req->peers)); free(req); } -static struct peer_entry * dht_req_get_peer(struct dht_req * req, - struct peer_entry * e) +static struct peer_entry * dht_req_get_peer(struct dht_req * req, + const struct peer_entry * e) { struct list_head * p; - list_for_each(p, &req->peers.list) { + llist_for_each(p, &req->peers) { struct peer_entry * x = list_entry(p, struct peer_entry, next); if (x->addr == e->addr) return x; @@ -1050,8 +1013,8 @@ static struct peer_entry * dht_req_get_peer(struct dht_req * req, } #define IS_MAGIC(peer) ((peer)->cookie == dht.magic) -void dht_req_add_peer(struct dht_req * req, - struct peer_entry * e) +static int dht_req_add_peer(struct dht_req * req, + const struct peer_entry * e) { struct peer_entry * x; /* existing */ struct list_head * p; /* iterator */ @@ -1063,16 +1026,17 @@ void dht_req_add_peer(struct dht_req * req, /* * Dedupe messages to the same peer, unless - * 1) The previous request was FIND_NODE and now it's FIND_VALUE - * 2) We urgently need contacts from emergency peer (magic cookie) + * 1) The previous was FIND_NODE and now it's FIND_VALUE + * 2) We urgently need contacts (magic cookie) */ x = dht_req_get_peer(req, e); if (x != NULL && x->code >= e->code && !IS_MAGIC(e)) - goto skip; + return -1; /* Find how this contact ranks in distance to the key */ - list_for_each(p, &req->peers.list) { - struct peer_entry * y = list_entry(p, struct peer_entry, next); + llist_for_each(p, &req->peers) { + struct peer_entry * y; + y = list_entry(p, struct peer_entry, next); if (IS_CLOSER(y->id, e->id)) { pos++; continue; @@ -1080,36 +1044,32 @@ void dht_req_add_peer(struct dht_req * req, break; } - /* Add a new peer to this request if we need to */ - if (pos < dht.alpha || !IS_MAGIC(e)) { - x = malloc(sizeof(*x)); - if (x == NULL) { - log_err("Failed to malloc peer entry."); - goto skip; - } + if (pos >= dht.alpha && IS_MAGIC(e)) + return -1; - x->cookie = e->cookie; - x->addr = e->addr; - x->code = e->code; - x->t_sent = e->t_sent; - x->id = dht_dup_key(e->id); - if (x->id == NULL) { - log_err("Failed to dup peer ID."); - free(x); - goto skip; - } + x = malloc(sizeof(*x)); + if (x == NULL) { + log_err("Failed to malloc peer entry."); + return -1; + } - if (IS_MAGIC(e)) - list_add(&x->next, p); - else - list_add_tail(&x->next, p); - ++req->peers.len; - return; + x->cookie = e->cookie; + x->addr = e->addr; + x->code = e->code; + x->t_sent = e->t_sent; + x->id = dht_dup_key(e->id); + if (x->id == NULL) { + log_err("Failed to dup peer ID."); + free(x); + return -1; } - skip: - list_del(&e->next); - free(e->id); - free(e); + + if (IS_MAGIC(e)) + llist_add_at(&x->next, p, &req->peers); + else + llist_add_tail_at(&x->next, p, &req->peers); + + return 0; } static size_t dht_req_add_peers(struct dht_req * req, @@ -1123,8 +1083,13 @@ static size_t dht_req_add_peers(struct dht_req * req, assert(pl != NULL); list_for_each_safe(p, h, pl) { - struct peer_entry * e = list_entry(p, struct peer_entry, next); - dht_req_add_peer(req, e); + struct peer_entry * e; + e = list_entry(p, struct peer_entry, next); + if (dht_req_add_peer(req, e) < 0) { + list_del(&e->next); + free(e->id); + free(e); + } } return n; @@ -1137,7 +1102,7 @@ static bool dht_req_has_peer(struct dht_req * req, assert(req != NULL); - list_for_each(p, &req->peers.list) { + llist_for_each(p, &req->peers) { struct peer_entry * e = list_entry(p, struct peer_entry, next); if (e->cookie == cookie) return true; @@ -1209,7 +1174,7 @@ static struct dht_req * __dht_kv_req_get_req(const uint8_t * key) { struct list_head * p; - list_for_each(p, &dht.reqs.list) { + llist_for_each(p, &dht.reqs.ll) { struct dht_req * r = list_entry(p, struct dht_req, next); if (memcmp(r->key, key, dht.id.len) == 0) return r; @@ -1228,7 +1193,7 @@ static struct dht_req * __dht_kv_get_req_cache(const uint8_t * key) if (req == NULL) return NULL; - if (req->cache.len == 0) + if (llist_is_empty(&req->cache)) return NULL; return req; @@ -1244,8 +1209,7 @@ static void __dht_kv_req_remove(const uint8_t * key) if (req == NULL) return; - list_del(&req->next); - --dht.reqs.len; + llist_del(&req->next, &dht.reqs.ll); dht_req_destroy(req); } @@ -1301,9 +1265,9 @@ static int dht_kv_update_req(const uint8_t * key, req = __dht_kv_req_get_req(key); if (req == NULL) { - if (dht.reqs.len == DHT_MAX_REQS) { + if (dht.reqs.ll.len == DHT_MAX_REQS) { log_err(KEY_FMT " Max reqs reached (%zu).", - KEY_VAL(key), dht.reqs.len); + KEY_VAL(key), dht.reqs.ll.len); peer_list_destroy(pl); goto fail_req; } @@ -1312,8 +1276,7 @@ static int dht_kv_update_req(const uint8_t * key, log_err(KEY_FMT "Failed to create req.", KEY_VAL(key)); goto fail_req; } - list_add_tail(&req->next, &dht.reqs.list); - ++dht.reqs.len; + llist_add_tail(&req->next, &dht.reqs.ll); } if (req->cache.len > 0) /* Already have values */ @@ -1322,9 +1285,9 @@ static int dht_kv_update_req(const uint8_t * key, dht_req_add_peers(req, pl); req->t_exp = now.tv_sec + DHT_T_RESP; - if (dht.reqs.len > DHT_WARN_REQS) { + if (dht.reqs.ll.len > DHT_WARN_REQS) { log_warn("Number of outstanding requests (%zu) exceeds %u.", - dht.reqs.len, DHT_WARN_REQS); + dht.reqs.ll.len, DHT_WARN_REQS); } pthread_mutex_unlock(&dht.reqs.mtx); @@ -1368,8 +1331,7 @@ static int dht_kv_respond_req(uint8_t * key, continue; } - list_add_tail(&e->next, &req->cache.list); - ++req->cache.len; + llist_add_tail(&e->next, &req->cache); } pthread_cond_broadcast(&dht.reqs.cond); @@ -1434,7 +1396,7 @@ static ssize_t dht_kv_wait_req(const uint8_t * key, memset(*vals, 0, max * sizeof(**vals)); - list_for_each(p, &req->cache.list) { + llist_for_each(p, &req->cache) { struct val_entry * v; if (i == max) break; /* We have enough values */ @@ -1535,10 +1497,10 @@ static ssize_t dht_kv_contact_list(const uint8_t * key, goto fail_bucket; } - b->t_refr = t.tv_sec + dht.t_refresh; + b->t_refr = t.tv_sec + dht.t_refr; if (b->contacts.len == dht.k || b->parent == NULL) { - list_for_each(p, &b->contacts.list) { + llist_for_each(p, &b->contacts) { struct contact * c; struct contact * d; c = list_entry(p, struct contact, next); @@ -1554,7 +1516,7 @@ static ssize_t dht_kv_contact_list(const uint8_t * key, } else { struct bucket * d = b->parent; for (i = 0; i < (1L << DHT_BETA) && len < dht.k; ++i) { - list_for_each(p, &d->children[i]->contacts.list) { + llist_for_each(p, &d->children[i]->contacts) { struct contact * c; struct contact * d; c = list_entry(p, struct contact, next); @@ -1661,11 +1623,11 @@ static void __dht_kv_bucket_refresh_list(struct bucket * b, __dht_kv_bucket_refresh_list(b->children[i], t, r); } - if (b->contacts.len == 0) + if (llist_is_empty(&b->contacts)) return; - c = list_first_entry(&b->contacts.list, struct contact, next); - if (t > c->t_seen + dht.t_refresh) { + c = llist_first_entry(&b->contacts, struct contact, next); + if (t > c->t_seen + dht.t_refr) { d = contact_create(c->id, c->addr); if (d != NULL) list_add(&d->next, r); @@ -1682,14 +1644,12 @@ static struct bucket * bucket_create(void) if (b == NULL) return NULL; - list_head_init(&b->contacts.list); - b->contacts.len = 0; + llist_init(&b->contacts); - list_head_init(&b->alts.list); - b->alts.len = 0; + llist_init(&b->alts); clock_gettime(CLOCK_REALTIME_COARSE, &t); - b->t_refr = t.tv_sec + dht.t_refresh; + b->t_refr = t.tv_sec + dht.t_refr; for (i = 0; i < (1L << DHT_BETA); ++i) b->children[i] = NULL; @@ -1713,18 +1673,16 @@ static void bucket_destroy(struct bucket * b) if (b->children[i] != NULL) bucket_destroy(b->children[i]); - list_for_each_safe(p, h, &b->contacts.list) { + llist_for_each_safe(p, h, &b->contacts) { struct contact * c = list_entry(p, struct contact, next); - list_del(&c->next); + llist_del(&c->next, &b->contacts); contact_destroy(c); - --b->contacts.len; } - list_for_each_safe(p, h, &b->alts.list) { + llist_for_each_safe(p, h, &b->alts) { struct contact * c = list_entry(p, struct contact, next); - list_del(&c->next); + llist_del(&c->next, &b->alts); contact_destroy(c); - --b->alts.len; } free(b); @@ -1759,13 +1717,11 @@ static int move_contacts(struct bucket * b, assert(b != NULL); assert(c != NULL); - list_for_each_safe(p, h, &b->contacts.list) { + llist_for_each_safe(p, h, &b->contacts) { d = list_entry(p, struct contact, next); if (bucket_has_id(c, d->id)) { - list_del(&d->next); - --b->contacts.len; - list_add_tail(&d->next, &c->contacts.list); - ++c->contacts.len; + llist_del(&d->next, &b->contacts); + llist_add_tail(&d->next, &c->contacts); } } @@ -1779,8 +1735,8 @@ static int split_bucket(struct bucket * b) size_t b_len; assert(b); - assert(b->alts.len == 0); - assert(b->contacts.len != 0); + assert(llist_is_empty(&b->alts)); + assert(!llist_is_empty(&b->contacts)); assert(b->children[0] == NULL); b_len = b->contacts.len; @@ -1836,39 +1792,33 @@ static int dht_kv_update_contacts(const uint8_t * id, goto fail_update; } - list_for_each_safe(p, h, &b->contacts.list) { + llist_for_each_safe(p, h, &b->contacts) { struct contact * d = list_entry(p, struct contact, next); if (d->addr == addr) { - list_del(&d->next); + llist_del(&d->next, &b->contacts); contact_destroy(d); - --b->contacts.len; } } if (b->contacts.len == dht.k) { if (bucket_has_id(b, dht.id.data)) { - list_add_tail(&c->next, &b->contacts.list); - ++b->contacts.len; + llist_add_tail(&c->next, &b->contacts); if (split_bucket(b)) { - list_del(&c->next); + llist_del(&c->next, &b->contacts); contact_destroy(c); - --b->contacts.len; } } else if (b->alts.len == dht.k) { struct contact * d; - d = list_first_entry(&b->alts.list, + d = llist_first_entry(&b->alts, struct contact, next); - list_del(&d->next); + llist_del(&d->next, &b->alts); contact_destroy(d); - list_add_tail(&c->next, &b->alts.list); - ++b->alts.len; + llist_add_tail(&c->next, &b->alts); } else { - list_add_tail(&c->next, &b->alts.list); - ++b->alts.len; + llist_add_tail(&c->next, &b->alts); } } else { - list_add_tail(&c->next, &b->contacts.list); - ++b->contacts.len; + llist_add_tail(&c->next, &b->contacts); } pthread_rwlock_unlock(&dht.db.lock); @@ -2116,7 +2066,7 @@ static ssize_t dht_kv_retrieve(const uint8_t * key, i = 0; - list_for_each(p, &e->vals.list) { + llist_for_each(p, &e->vals) { struct val_entry * v; if (i == n) break; /* We have enough values */ @@ -2129,7 +2079,7 @@ static ssize_t dht_kv_retrieve(const uint8_t * key, memcpy((*vals)[i++].data, v->val.data, v->val.len); } - list_for_each(p, &e->lvals.list) { + llist_for_each(p, &e->lvals) { struct val_entry * v; if (i == n) break; /* We have enough values */ @@ -2266,7 +2216,7 @@ static int dht_send_msg(dht_msg_t * msg, uint64_t addr) { size_t len; - struct shm_du_buff * sdb; + struct ssm_pk_buff * spb; if (msg == NULL) return 0; @@ -2279,21 +2229,21 @@ static int dht_send_msg(dht_msg_t * msg, goto fail_msg; } - if (ipcp_sdb_reserve(&sdb, len)) { - log_warn("%s failed to get sdb.", DHT_CODE(msg)); + if (ipcp_spb_reserve(&spb, len)) { + log_warn("%s failed to get spb.", DHT_CODE(msg)); goto fail_msg; } - dht_msg__pack(msg, shm_du_buff_head(sdb)); + dht_msg__pack(msg, ssm_pk_buff_head(spb)); - if (dt_write_packet(addr, QOS_CUBE_BE, dht.eid, sdb) < 0) { + if (dt_write_packet(addr, QOS_CUBE_BE, dht.eid, spb) < 0) { log_warn("%s write failed", DHT_CODE(msg)); goto fail_send; } return 0; fail_send: - ipcp_sdb_release(sdb); + ipcp_spb_release(spb); fail_msg: return -1; } @@ -2584,15 +2534,14 @@ static void __add_dht_kv_entry(struct dht_entry * e) assert(e != NULL); - list_for_each(p, &dht.db.kv.list) { + llist_for_each(p, &dht.db.kv.ll) { struct dht_entry * d = list_entry(p, struct dht_entry, next); if (IS_CLOSER(d->key, e->key)) continue; break; } - list_add_tail(&e->next, p); - ++dht.db.kv.len; + llist_add_tail_at(&e->next, p, &dht.db.kv.ll); } /* incoming store message */ @@ -2629,9 +2578,8 @@ static int dht_kv_store(const uint8_t * key, return 0; fail_add: if (new) { - list_del(&e->next); + llist_del(&e->next, &dht.db.kv.ll); dht_entry_destroy(e); - --dht.db.kv.len; } fail: pthread_rwlock_unlock(&dht.db.lock); @@ -2669,14 +2617,13 @@ static int dht_kv_publish(const uint8_t * key, pthread_rwlock_unlock(&dht.db.lock); - dht_kv_store_remote(key, val, now.tv_sec + dht.t_expire); + dht_kv_store_remote(key, val, now.tv_sec + dht.t_exp); return 0; fail_add: if (new) { - list_del(&e->next); + llist_del(&e->next, &dht.db.kv.ll); dht_entry_destroy(e); - --dht.db.kv.len; } fail: pthread_rwlock_unlock(&dht.db.lock); @@ -2858,7 +2805,7 @@ static void do_dht_kv_store(const dht_store_msg_t * store) key = store->key.data; exp = store->exp; - if (dht_kv_store(store->key.data, val, store->exp) < 0) { + if (dht_kv_store(key, val, store->exp) < 0) { log_err(KV_FMT " Failed to store.", KV_VAL(key, val)); return; } @@ -3189,7 +3136,7 @@ static void * dht_handle_packet(void * o) } #ifndef __DHT_TEST__ static void dht_post_packet(void * comp, - struct shm_du_buff * sdb) + struct ssm_pk_buff * spb) { struct cmd * cmd; @@ -3201,17 +3148,17 @@ static void dht_post_packet(void * comp, goto fail_cmd; } - cmd->cbuf.data = malloc(shm_du_buff_len(sdb)); + cmd->cbuf.data = malloc(ssm_pk_buff_len(spb)); if (cmd->cbuf.data == NULL) { log_err("Command buffer malloc failed."); goto fail_buf; } - cmd->cbuf.len = shm_du_buff_len(sdb); + cmd->cbuf.len = ssm_pk_buff_len(spb); - memcpy(cmd->cbuf.data, shm_du_buff_head(sdb), cmd->cbuf.len); + memcpy(cmd->cbuf.data, ssm_pk_buff_head(spb), cmd->cbuf.len); - ipcp_sdb_release(sdb); + ipcp_spb_release(spb); pthread_mutex_lock(&dht.cmds.mtx); @@ -3226,7 +3173,7 @@ static void dht_post_packet(void * comp, fail_buf: free(cmd); fail_cmd: - ipcp_sdb_release(sdb); + ipcp_spb_release(spb); return; } #endif @@ -3449,15 +3396,14 @@ static void dht_kv_remove_expired_reqs(void) pthread_mutex_lock(&dht.reqs.mtx); - list_for_each_safe(p, h, &dht.reqs.list) { + llist_for_each_safe(p, h, &dht.reqs.ll) { struct dht_req * e; e = list_entry(p, struct dht_req, next); if (IS_EXPIRED(e, &now)) { log_dbg(KEY_FMT " Removing expired request.", KEY_VAL(e->key)); - list_del(&e->next); + llist_del(&e->next, &dht.reqs.ll); dht_req_destroy(e); - --dht.reqs.len; } } @@ -3489,7 +3435,7 @@ static void dht_entry_get_repl_lists(const struct dht_entry * e, struct list_head * p; struct val_entry * n; - list_for_each(p, &e->vals.list) { + llist_for_each(p, &e->vals) { struct val_entry * v = list_entry(p, struct val_entry, next); if (MUST_REPLICATE(v, now) && !IS_EXPIRED(v, now)) { n = val_entry_create(v->val, v->t_exp); @@ -3500,11 +3446,11 @@ static void dht_entry_get_repl_lists(const struct dht_entry * e, } } - list_for_each(p, &e->lvals.list) { + llist_for_each(p, &e->lvals) { struct val_entry * v = list_entry(p, struct val_entry, next); if (MUST_REPLICATE(v, now) && MUST_REPUBLISH(v, now)) { /* Add expire time here, to allow creating val_entry */ - n = val_entry_create(v->val, now->tv_sec + dht.t_expire); + n = val_entry_create(v->val, now->tv_sec + dht.t_exp); if (n == NULL) continue; @@ -3533,10 +3479,10 @@ static int dht_kv_next_values(uint8_t * key, pthread_rwlock_rdlock(&dht.db.lock); - if (dht.db.kv.len == 0) + if (llist_is_empty(&dht.db.kv.ll)) goto no_entries; - list_for_each_safe(p, h, &dht.db.kv.list) { + llist_for_each_safe(p, h, &dht.db.kv.ll) { e = list_entry(p, struct dht_entry, next); if (IS_CLOSER(e->key, key)) continue; /* Already processed */ @@ -3578,7 +3524,7 @@ static void dht_kv_republish_value(const uint8_t * key, assert(MUST_REPLICATE(v, now)); if (MUST_REPUBLISH(v, now)) - assert(v->t_exp >= now->tv_sec + dht.t_expire); + assert(v->t_exp >= now->tv_sec + dht.t_exp); if (dht_kv_store_remote(key, v->val, v->t_exp) == 0) { log_dbg(KV_FMT " Republished.", KV_VAL(key, v->val)); @@ -3786,8 +3732,8 @@ static void * work(void * o) nanosleep(&now, NULL); } - intv = gcd(dht.t_expire, (dht.t_expire - DHT_N_REPUB * dht.t_repl)); - intv = gcd(intv, gcd(dht.t_repl, dht.t_refresh)) / 2; + intv = gcd(dht.t_exp, (dht.t_exp - DHT_N_REPUB * dht.t_repl)); + intv = gcd(intv, gcd(dht.t_repl, dht.t_refr)) / 2; intv = MAX(1, intv / n); log_dbg("DHT worker starting %ld seconds interval.", intv * n); @@ -3868,13 +3814,13 @@ int dht_init(struct dir_dht_config * conf) dht.id.len = DHT_TEST_KEY_LEN; dht.addr = DHT_TEST_ADDR; #endif - dht.t0 = now.tv_sec; - dht.alpha = conf->params.alpha; - dht.k = conf->params.k; - dht.t_expire = conf->params.t_expire; - dht.t_refresh = conf->params.t_refresh; - dht.t_repl = conf->params.t_replicate; - dht.peer = conf->peer; + dht.t0 = now.tv_sec; + dht.alpha = conf->params.alpha; + dht.k = conf->params.k; + dht.t_exp = conf->params.t_expire; + dht.t_refr = conf->params.t_refresh; + dht.t_repl = conf->params.t_replicate; + dht.peer = conf->peer; dht.magic = generate_cookie(); @@ -3899,8 +3845,7 @@ int dht_init(struct dir_dht_config * conf) goto fail_cmds_cond; } - list_head_init(&dht.reqs.list); - dht.reqs.len = 0; + llist_init(&dht.reqs.ll); if (pthread_mutex_init(&dht.reqs.mtx, NULL)) { log_err("Failed to initialize request mutex."); @@ -3922,8 +3867,7 @@ int dht_init(struct dir_dht_config * conf) goto fail_reqs_cond; } - list_head_init(&dht.db.kv.list); - dht.db.kv.len = 0; + llist_init(&dht.db.kv.ll); dht.db.kv.vals = 0; dht.db.kv.lvals = 0; @@ -3960,9 +3904,9 @@ int dht_init(struct dir_dht_config * conf) log_dbg(" address: " ADDR_FMT32 ".", ADDR_VAL32(&dht.addr)); log_dbg(" peer: " ADDR_FMT32 ".", ADDR_VAL32(&dht.peer)); log_dbg(" magic cookie: " HASH_FMT64 ".", HASH_VAL64(&dht.magic)); - log_info(" parameters: alpha=%u, k=%zu, t_expire=%ld, " - "t_refresh=%ld, t_replicate=%ld.", - dht.alpha, dht.k, dht.t_expire, dht.t_refresh, dht.t_repl); + log_info(" parameters: alpha=%u, k=%zu, t_exp=%ld, " + "t_refr=%ld, t_replicate=%ld.", + dht.alpha, dht.k, dht.t_exp, dht.t_refr, dht.t_repl); #endif dht.state = DHT_INIT; @@ -4015,11 +3959,10 @@ void dht_fini(void) pthread_mutex_lock(&dht.reqs.mtx); - list_for_each_safe(p, h, &dht.reqs.list) { + llist_for_each_safe(p, h, &dht.reqs.ll) { struct dht_req * r = list_entry(p, struct dht_req, next); - list_del(&r->next); + llist_del(&r->next, &dht.reqs.ll); dht_req_destroy(r); - dht.reqs.len--; } pthread_mutex_unlock(&dht.reqs.mtx); @@ -4029,11 +3972,10 @@ void dht_fini(void) pthread_rwlock_wrlock(&dht.db.lock); - list_for_each_safe(p, h, &dht.db.kv.list) { + llist_for_each_safe(p, h, &dht.db.kv.ll) { struct dht_entry * e = list_entry(p, struct dht_entry, next); - list_del(&e->next); + llist_del(&e->next, &dht.db.kv.ll); dht_entry_destroy(e); - dht.db.kv.len--; } if (dht.db.contacts.root != NULL) @@ -4043,10 +3985,10 @@ void dht_fini(void) pthread_rwlock_destroy(&dht.db.lock); - assert(dht.db.kv.len == 0); + assert(llist_is_empty(&dht.db.kv.ll)); assert(dht.db.kv.vals == 0); assert(dht.db.kv.lvals == 0); - assert(dht.reqs.len == 0); + assert(llist_is_empty(&dht.reqs.ll)); freebuf(dht.id); } diff --git a/src/ipcpd/unicast/dir/dht.h b/src/ipcpd/unicast/dir/dht.h index 852a5130..ed3bb9f0 100644 --- a/src/ipcpd/unicast/dir/dht.h +++ b/src/ipcpd/unicast/dir/dht.h @@ -1,5 +1,5 @@ /* - * Ouroboros - Copyright (C) 2016 - 2024 + * Ouroboros - Copyright (C) 2016 - 2026 * * Distributed Hash Table based on Kademlia * diff --git a/src/ipcpd/unicast/dir/dht.proto b/src/ipcpd/unicast/dir/dht.proto index ea74805f..02b6b341 100644 --- a/src/ipcpd/unicast/dir/dht.proto +++ b/src/ipcpd/unicast/dir/dht.proto @@ -1,5 +1,5 @@ /* - * Ouroboros - Copyright (C) 2016 - 2024 + * Ouroboros - Copyright (C) 2016 - 2026 * * DHT protocol, based on Kademlia * diff --git a/src/ipcpd/unicast/dir/ops.h b/src/ipcpd/unicast/dir/ops.h index 8c6e5eb5..6c336ee0 100644 --- a/src/ipcpd/unicast/dir/ops.h +++ b/src/ipcpd/unicast/dir/ops.h @@ -1,5 +1,5 @@ /* - * Ouroboros - Copyright (C) 2016 - 2024 + * Ouroboros - Copyright (C) 2016 - 2026 * * Directory policy ops * diff --git a/src/ipcpd/unicast/dir/pol.h b/src/ipcpd/unicast/dir/pol.h index eae4b2e7..8ccf4f95 100644 --- a/src/ipcpd/unicast/dir/pol.h +++ b/src/ipcpd/unicast/dir/pol.h @@ -1,5 +1,5 @@ /* - * Ouroboros - Copyright (C) 2016 - 2024 + * Ouroboros - Copyright (C) 2016 - 2026 * * Directory policies * diff --git a/src/ipcpd/unicast/dir/tests/CMakeLists.txt b/src/ipcpd/unicast/dir/tests/CMakeLists.txt index 897f1ec2..eded823f 100644 --- a/src/ipcpd/unicast/dir/tests/CMakeLists.txt +++ b/src/ipcpd/unicast/dir/tests/CMakeLists.txt @@ -3,38 +3,36 @@ get_filename_component(CURRENT_SOURCE_PARENT_DIR get_filename_component(CURRENT_BINARY_PARENT_DIR ${CMAKE_CURRENT_BINARY_DIR} DIRECTORY) -include_directories(${CMAKE_CURRENT_SOURCE_DIR}) -include_directories(${CMAKE_CURRENT_BINARY_DIR}) - -include_directories(${CURRENT_SOURCE_PARENT_DIR}) -include_directories(${CURRENT_BINARY_PARENT_DIR}) - -include_directories(${CMAKE_SOURCE_DIR}/include) -include_directories(${CMAKE_BINARY_DIR}/include) - get_filename_component(PARENT_PATH ${CMAKE_CURRENT_SOURCE_DIR} DIRECTORY) get_filename_component(PARENT_DIR ${PARENT_PATH} NAME) +compute_test_prefix() + create_test_sourcelist(${PARENT_DIR}_tests test_suite.c # Add new tests here dht_test.c - ) +) -protobuf_generate_c(DHT_PROTO_SRCS KAD_PROTO_HDRS ../dht.proto) +protobuf_generate_c(DHT_PROTO_SRCS KAD_PROTO_HDRS ${CURRENT_SOURCE_PARENT_DIR}/dht.proto) add_executable(${PARENT_DIR}_test ${${PARENT_DIR}_tests} ${DHT_PROTO_SRCS}) -target_link_libraries(${PARENT_DIR}_test ouroboros-common) -add_dependencies(check ${PARENT_DIR}_test) +target_include_directories(${PARENT_DIR}_test PRIVATE + ${CMAKE_CURRENT_SOURCE_DIR} + ${CMAKE_CURRENT_BINARY_DIR} + ${CURRENT_SOURCE_PARENT_DIR} + ${CURRENT_BINARY_PARENT_DIR} + ${CMAKE_SOURCE_DIR}/include + ${CMAKE_BINARY_DIR}/include + ${CMAKE_SOURCE_DIR}/src/ipcpd + ${CMAKE_BINARY_DIR}/src/ipcpd + ${CMAKE_SOURCE_DIR}/src/ipcpd/unicast + ${CMAKE_BINARY_DIR}/src/ipcpd/unicast +) + +disable_test_logging_for_target(${PARENT_DIR}_test) -set(tests_to_run ${${PARENT_DIR}_tests}) -if(CMAKE_VERSION VERSION_LESS "3.29.0") - remove(tests_to_run test_suite.c) -else () - list(POP_FRONT tests_to_run) -endif() +target_link_libraries(${PARENT_DIR}_test ouroboros-common) +add_dependencies(build_tests ${PARENT_DIR}_test) -foreach (test ${tests_to_run}) - get_filename_component(test_name ${test} NAME_WE) - add_test(${test_name} ${C_TEST_PATH}/${PARENT_DIR}_test ${test_name}) -endforeach (test) +ouroboros_register_tests(TARGET ${PARENT_DIR}_test TESTS ${${PARENT_DIR}_tests}) diff --git a/src/ipcpd/unicast/dir/tests/dht_test.c b/src/ipcpd/unicast/dir/tests/dht_test.c index cb6b0f9f..1f7026b3 100644 --- a/src/ipcpd/unicast/dir/tests/dht_test.c +++ b/src/ipcpd/unicast/dir/tests/dht_test.c @@ -1,5 +1,5 @@ /* - * Ouroboros - Copyright (C) 2016 - 2024 + * Ouroboros - Copyright (C) 2016 - 2026 * * Unit tests of the DHT * @@ -27,7 +27,7 @@ #define _POSIX_C_SOURCE 200112L #endif -#include <ouroboros/test.h> +#include <test/test.h> #include <ouroboros/list.h> #include <ouroboros/utils.h> @@ -46,10 +46,9 @@ /* forward declare for use in the dht code */ /* Packet sink for DHT tests */ struct { - bool enabled; + bool enabled; - struct list_head list; - size_t len; + struct llist msgs; } sink; struct message { @@ -66,8 +65,6 @@ static int sink_send_msg(buffer_t * pkt, assert(pkt != NULL); assert(addr != 0); - assert(!list_is_empty(&sink.list) || sink.len == 0); - if (!sink.enabled) goto finish; @@ -83,9 +80,8 @@ static int sink_send_msg(buffer_t * pkt, m->dst = addr; - list_add_tail(&m->next, &sink.list); + llist_add_tail(&m->next, &sink.msgs); - ++sink.len; finish: freebuf(*pkt); @@ -103,8 +99,7 @@ static int sink_send_msg(buffer_t * pkt, static void sink_init(void) { - list_head_init(&sink.list); - sink.len = 0; + llist_init(&sink.msgs); sink.enabled = true; } @@ -113,22 +108,20 @@ static void sink_clear(void) struct list_head * p; struct list_head * h; - list_for_each_safe(p, h, &sink.list) { + llist_for_each_safe(p, h, &sink.msgs) { struct message * m = list_entry(p, struct message, next); - list_del(&m->next); + llist_del(&m->next, &sink.msgs); dht_msg__free_unpacked((dht_msg_t *) m->msg, NULL); free(m); - --sink.len; } - assert(list_is_empty(&sink.list)); + assert(llist_is_empty(&sink.msgs)); } static void sink_fini(void) { sink_clear(); - - assert(list_is_empty(&sink.list) || sink.len != 0); + sink.enabled = false; } static dht_msg_t * sink_read(void) @@ -136,16 +129,12 @@ static dht_msg_t * sink_read(void) struct message * m; dht_msg_t * msg; - assert(!list_is_empty(&sink.list) || sink.len == 0); - - if (list_is_empty(&sink.list)) + if (llist_is_empty(&sink.msgs)) return NULL; - m = list_first_entry(&sink.list, struct message, next); - - --sink.len; + m = llist_first_entry(&sink.msgs, struct message, next); - list_del(&m->next); + llist_del(&m->next, &sink.msgs); msg = m->msg; @@ -978,7 +967,7 @@ static int test_dht_kv_find_node_rsp_msg_contacts(void) } if ((size_t) n < dht.k) { - printf("Failed to get enough contacts (%zu < %zu).\n", n, dht.k); + printf("Failed to get all contacts (%zu < %zu).\n", n, dht.k); goto fail_fill; } @@ -1204,7 +1193,7 @@ static int test_dht_kv_find_value_rsp_msg_contacts(void) } if ((size_t) n < dht.k) { - printf("Failed to get enough contacts (%zu < %zu).\n", n, dht.k); + printf("Failed to get all contacts (%zu < %zu).\n", n, dht.k); goto fail_fill; } @@ -1591,7 +1580,7 @@ static int test_dht_reg_unreg(void) goto fail_reg; } - if (sink.len != 0) { + if (!llist_is_empty(&sink.msgs)) { printf("Packet sent without contacts!"); goto fail_msg; } @@ -1642,7 +1631,7 @@ static int test_dht_reg_unreg_contacts(void) goto fail_reg; } - if (sink.len != dht.alpha) { + if (sink.msgs.len != dht.alpha) { printf("Packet sent to too few contacts!\n"); goto fail_msg; } @@ -1784,7 +1773,7 @@ static int test_dht_query(void) goto fail_get; } - if (sink.len != 0) { + if (!llist_is_empty(&sink.msgs)) { printf("Packet sent without contacts!"); goto fail_test; } |