summaryrefslogtreecommitdiff
path: root/src/ipcpd/unicast
diff options
context:
space:
mode:
Diffstat (limited to 'src/ipcpd/unicast')
-rw-r--r--src/ipcpd/unicast/dir/dht.c53
-rw-r--r--src/ipcpd/unicast/dt.c216
-rw-r--r--src/ipcpd/unicast/fa.c145
-rw-r--r--src/ipcpd/unicast/fa.h3
-rw-r--r--src/ipcpd/unicast/main.c5
-rw-r--r--src/ipcpd/unicast/pff/alternate.c2
-rw-r--r--src/ipcpd/unicast/pff/multipath.c2
-rw-r--r--src/ipcpd/unicast/pff/pft.c19
-rw-r--r--src/ipcpd/unicast/pff/pft.h4
-rw-r--r--src/ipcpd/unicast/pff/simple.c2
-rw-r--r--src/ipcpd/unicast/pff/tests/pft_test.c10
-rw-r--r--src/ipcpd/unicast/routing/graph.c8
-rw-r--r--src/ipcpd/unicast/routing/link-state.c2
13 files changed, 256 insertions, 215 deletions
diff --git a/src/ipcpd/unicast/dir/dht.c b/src/ipcpd/unicast/dir/dht.c
index 8eeea800..753d9a73 100644
--- a/src/ipcpd/unicast/dir/dht.c
+++ b/src/ipcpd/unicast/dir/dht.c
@@ -1597,6 +1597,7 @@ static ssize_t dht_kv_get_contacts(const uint8_t * key,
fail_contact:
while (i-- > 0)
dht_contact_msg__free_unpacked((*msgs)[i], NULL);
+
free(*msgs);
*msgs = NULL;
fail_msgs:
@@ -1763,6 +1764,7 @@ static int split_bucket(struct bucket * b)
fail_child:
while (i-- > 0)
bucket_destroy(b->children[i]);
+
return -1;
}
@@ -2849,6 +2851,7 @@ static dht_msg_t * do_dht_kv_find_node_req(const dht_find_req_msg_t * req)
fail_msg:
while (len-- > 0)
dht_contact_msg__free_unpacked(contacts[len], NULL);
+
free(contacts);
fail_contacts:
return NULL;
@@ -2953,6 +2956,7 @@ static dht_msg_t * do_dht_kv_find_value_req(const dht_find_req_msg_t * req)
fail_vals:
while (n_contacts-- > 0)
dht_contact_msg__free_unpacked(contacts[n_contacts], NULL);
+
free(contacts);
fail_contacts:
return NULL;
@@ -3312,6 +3316,42 @@ static int emergency_peer(struct list_head * pl)
return -ENOMEM;
}
+static bool __dht_kv_bucket_has_addr(struct bucket * b,
+ uint64_t addr)
+{
+ struct list_head * p;
+ size_t i;
+
+ assert(b != NULL);
+
+ if (*b->children != NULL)
+ for (i = 0; i < (1L << DHT_BETA); ++i)
+ if (__dht_kv_bucket_has_addr(b->children[i], addr))
+ return true;
+
+ llist_for_each(p, &b->contacts) {
+ struct contact * c;
+ c = list_entry(p, struct contact, next);
+ if (c->addr == addr)
+ return true;
+ }
+
+ return false;
+}
+
+static bool dht_kv_knows_peer(void)
+{
+ bool found;
+
+ pthread_rwlock_rdlock(&dht.db.lock);
+
+ found = __dht_kv_bucket_has_addr(dht.db.contacts.root, dht.peer);
+
+ pthread_rwlock_unlock(&dht.db.lock);
+
+ return found;
+}
+
static int dht_kv_seed_bootstrap_peer(void)
{
struct list_head pl;
@@ -3323,6 +3363,9 @@ static int dht_kv_seed_bootstrap_peer(void)
return 0;
}
+ if (dht_kv_knows_peer())
+ return 0;
+
if (emergency_peer(&pl) < 0) {
log_err("Could not create emergency peer.");
goto fail_peer;
@@ -3338,7 +3381,8 @@ static int dht_kv_seed_bootstrap_peer(void)
peer_list_destroy(&pl);
- return 0;
+ /* Sent, but not bootstrapped until the peer is in the DHT. */
+ return -EAGAIN;
fail_query:
peer_list_destroy(&pl);
fail_peer:
@@ -3427,6 +3471,8 @@ static void value_list_destroy(struct list_head * vl)
#define MUST_REPLICATE(v, now) ((now)->tv_sec > (v)->t_repl + dht.t_repl)
#define MUST_REPUBLISH(v, now) /* Close to expiry deadline */ \
(((v)->t_exp - (now)->tv_sec) < (DHT_N_REPUB * dht.t_repl))
+/* A local value must be (re)stored if near expiry or never stored. */
+#define MUST_STORE_LVAL(v, now) (MUST_REPUBLISH(v, now) || (v)->t_repl == 0)
static void dht_entry_get_repl_lists(const struct dht_entry * e,
struct list_head * repl,
struct list_head * rebl,
@@ -3448,7 +3494,7 @@ static void dht_entry_get_repl_lists(const struct dht_entry * e,
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)) {
+ if (MUST_REPLICATE(v, now) && MUST_STORE_LVAL(v, now)) {
/* Add expire time here, to allow creating val_entry */
n = val_entry_create(v->val, now->tv_sec + dht.t_exp);
if (n == NULL)
@@ -3738,6 +3784,9 @@ static void * work(void * o)
log_dbg("DHT worker starting %ld seconds interval.", intv * n);
+ /* Flush names registered before we had peers to store them. */
+ dht_kv_replicate();
+
while (true) {
int i = 0;
while (tasks[i] != NULL) {
diff --git a/src/ipcpd/unicast/dt.c b/src/ipcpd/unicast/dt.c
index 252477f4..e89cb17e 100644
--- a/src/ipcpd/unicast/dt.c
+++ b/src/ipcpd/unicast/dt.c
@@ -31,6 +31,7 @@
#define DT "dt"
#define OUROBOROS_PREFIX DT
+#include <ouroboros/atomics.h>
#include <ouroboros/bitmap.h>
#include <ouroboros/errno.h>
#include <ouroboros/logs.h>
@@ -139,7 +140,7 @@ static void dt_pci_shrink(struct ssm_pk_buff * spb)
{
assert(spb);
- ssm_pk_buff_head_release(spb, dt_pci_info.head_size);
+ ssm_pk_buff_pop(spb, dt_pci_info.head_size);
}
struct {
@@ -168,22 +169,33 @@ struct {
size_t f_nhp_pkt[QOS_CUBE_MAX];
size_t f_nhp_bytes[QOS_CUBE_MAX];
pthread_mutex_t lock;
- } stat[PROG_MAX_FLOWS];
+ } stat[PROC_MAX_FLOWS];
size_t n_flows;
#endif
struct bmp * res_fds;
- struct comp_info comps[PROG_RES_FDS];
+ struct comp_info comps[PROC_RES_FDS];
pthread_rwlock_t lock;
pthread_t listener;
} dt;
+/*
+ * Flow stats are lock-free relaxed atomics on the data path; the per-flow
+ * lock still guards the stamp/addr/n_flows lifecycle (see stat_used).
+ */
+#ifdef IPCP_FLOW_STATS
+#define dt_stat_inc(idx, name, qc, len) \
+ do { \
+ FETCH_ADD_RELAXED(&dt.stat[idx].name ## _pkt[qc], 1); \
+ FETCH_ADD_RELAXED(&dt.stat[idx].name ## _bytes[qc], (len)); \
+ } while (0)
+#define dt_stat_load(idx, field, qc) LOAD_RELAXED(&dt.stat[idx].field[qc])
+
static int dt_rib_read(const char * path,
char * buf,
size_t len)
{
-#ifdef IPCP_FLOW_STATS
int fd;
int i;
char str[QOS_BLOCK_LEN + 1];
@@ -220,7 +232,7 @@ static int dt_rib_read(const char * path,
tm = gmtime(&dt.stat[fd].stamp);
strftime(tmstr, sizeof(tmstr), RIB_TM_FORMAT, tm);
- if (fd >= PROG_RES_FDS) {
+ if (fd >= PROC_RES_FDS) {
fccntl(fd, FLOWGRXQLEN, &rxqlen);
fccntl(fd, FLOWGTXQLEN, &txqlen);
}
@@ -249,20 +261,20 @@ static int dt_rib_read(const char * path,
" failed nhop (packets): %20zu\n"
" failed nhop (bytes): %20zu\n",
i,
- dt.stat[fd].snd_pkt[i],
- dt.stat[fd].snd_bytes[i],
- dt.stat[fd].rcv_pkt[i],
- dt.stat[fd].rcv_bytes[i],
- dt.stat[fd].lcl_w_pkt[i],
- dt.stat[fd].lcl_w_bytes[i],
- dt.stat[fd].lcl_r_pkt[i],
- dt.stat[fd].lcl_r_bytes[i],
- dt.stat[fd].r_drp_pkt[i],
- dt.stat[fd].r_drp_bytes[i],
- dt.stat[fd].w_drp_pkt[i],
- dt.stat[fd].w_drp_bytes[i],
- dt.stat[fd].f_nhp_pkt[i],
- dt.stat[fd].f_nhp_bytes[i]
+ dt_stat_load(fd, snd_pkt, i),
+ dt_stat_load(fd, snd_bytes, i),
+ dt_stat_load(fd, rcv_pkt, i),
+ dt_stat_load(fd, rcv_bytes, i),
+ dt_stat_load(fd, lcl_w_pkt, i),
+ dt_stat_load(fd, lcl_w_bytes, i),
+ dt_stat_load(fd, lcl_r_pkt, i),
+ dt_stat_load(fd, lcl_r_bytes, i),
+ dt_stat_load(fd, r_drp_pkt, i),
+ dt_stat_load(fd, r_drp_bytes, i),
+ dt_stat_load(fd, w_drp_pkt, i),
+ dt_stat_load(fd, w_drp_bytes, i),
+ dt_stat_load(fd, f_nhp_pkt, i),
+ dt_stat_load(fd, f_nhp_bytes, i)
);
strcat(buf, str);
}
@@ -270,17 +282,10 @@ static int dt_rib_read(const char * path,
pthread_mutex_unlock(&dt.stat[fd].lock);
return RIB_FILE_STRLEN;
-#else
- (void) path;
- (void) buf;
- (void) len;
- return 0;
-#endif
}
static int dt_rib_readdir(char *** buf)
{
-#ifdef IPCP_FLOW_STATS
char entry[RIB_PATH_LEN + 1];
size_t i;
int idx = 0;
@@ -296,7 +301,7 @@ static int dt_rib_readdir(char *** buf)
if (*buf == NULL)
goto fail_entries;
- for (i = 0; i < PROG_MAX_FLOWS; ++i) {
+ for (i = 0; i < PROC_MAX_FLOWS; ++i) {
pthread_mutex_lock(&dt.stat[i].lock);
if (dt.stat[i].stamp == 0) {
@@ -327,16 +332,11 @@ static int dt_rib_readdir(char *** buf)
fail_entries:
pthread_rwlock_unlock(&dt.lock);
return -ENOMEM;
-#else
- (void) buf;
- return 0;
-#endif
}
static int dt_rib_getattr(const char * path,
struct rib_attr * attr)
{
-#ifdef IPCP_FLOW_STATS
int fd;
char * entry;
@@ -356,10 +356,7 @@ static int dt_rib_getattr(const char * path,
}
pthread_mutex_unlock(&dt.stat[fd].lock);
-#else
- (void) path;
- (void) attr;
-#endif
+
return 0;
}
@@ -369,7 +366,12 @@ static struct rib_ops r_ops = {
.getattr = dt_rib_getattr
};
-#ifdef IPCP_FLOW_STATS
+/*
+ * Hold dt.lock + per-stat together: dt_rib_readdir samples n_flows
+ * under rdlock and walks stamps under per-stat; updates must be
+ * atomic w.r.t. that snapshot or the malloc(n_flows) buffer can
+ * overflow.
+ */
static void stat_used(int fd,
uint64_t addr)
{
@@ -377,6 +379,7 @@ static void stat_used(int fd,
clock_gettime(CLOCK_REALTIME_COARSE, &now);
+ pthread_rwlock_wrlock(&dt.lock);
pthread_mutex_lock(&dt.stat[fd].lock);
memset(&dt.stat[fd], 0, sizeof(dt.stat[fd]));
@@ -384,14 +387,13 @@ static void stat_used(int fd,
dt.stat[fd].stamp = (addr != INVALID_ADDR) ? now.tv_sec : 0;
dt.stat[fd].addr = addr;
- pthread_mutex_unlock(&dt.stat[fd].lock);
-
- pthread_rwlock_wrlock(&dt.lock);
-
(addr != INVALID_ADDR) ? ++dt.n_flows : --dt.n_flows;
+ pthread_mutex_unlock(&dt.stat[fd].lock);
pthread_rwlock_unlock(&dt.lock);
}
+#else
+#define dt_stat_inc(idx, name, qc, len) ((void) 0)
#endif
static void handle_event(void * self,
@@ -440,15 +442,10 @@ static void packet_handler(int fd,
len = ssm_pk_buff_len(spb);
#ifndef IPCP_FLOW_STATS
- (void) fd;
-#else
- pthread_mutex_lock(&dt.stat[fd].lock);
-
- ++dt.stat[fd].rcv_pkt[qc];
- dt.stat[fd].rcv_bytes[qc] += len;
-
- pthread_mutex_unlock(&dt.stat[fd].lock);
+ (void) fd;
#endif
+ dt_stat_inc(fd, rcv, qc, len);
+
memset(&dt_pci, 0, sizeof(dt_pci));
head = ssm_pk_buff_head(spb);
@@ -458,14 +455,7 @@ static void packet_handler(int fd,
if (dt_pci.ttl == 0) {
log_dbg("TTL was zero.");
ipcp_spb_release(spb);
-#ifdef IPCP_FLOW_STATS
- pthread_mutex_lock(&dt.stat[fd].lock);
-
- ++dt.stat[fd].r_drp_pkt[qc];
- dt.stat[fd].r_drp_bytes[qc] += len;
-
- pthread_mutex_unlock(&dt.stat[fd].lock);
-#endif
+ dt_stat_inc(fd, r_drp, qc, len);
return;
}
@@ -475,14 +465,7 @@ static void packet_handler(int fd,
log_dbg("No next hop for %" PRIu64 ".",
dt_pci.dst_addr);
ipcp_spb_release(spb);
-#ifdef IPCP_FLOW_STATS
- pthread_mutex_lock(&dt.stat[fd].lock);
-
- ++dt.stat[fd].f_nhp_pkt[qc];
- dt.stat[fd].f_nhp_bytes[qc] += len;
-
- pthread_mutex_unlock(&dt.stat[fd].lock);
-#endif
+ dt_stat_inc(fd, f_nhp, qc, len);
return;
}
@@ -494,27 +477,14 @@ static void packet_handler(int fd,
if (ret == -EFLOWDOWN)
notifier_event(NOTIFY_DT_FLOW_DOWN, &ofd);
ipcp_spb_release(spb);
-#ifdef IPCP_FLOW_STATS
- pthread_mutex_lock(&dt.stat[ofd].lock);
-
- ++dt.stat[ofd].w_drp_pkt[qc];
- dt.stat[ofd].w_drp_bytes[qc] += len;
-
- pthread_mutex_unlock(&dt.stat[ofd].lock);
-#endif
+ dt_stat_inc(ofd, w_drp, qc, len);
return;
}
-#ifdef IPCP_FLOW_STATS
- pthread_mutex_lock(&dt.stat[ofd].lock);
- ++dt.stat[ofd].snd_pkt[qc];
- dt.stat[ofd].snd_bytes[qc] += len;
-
- pthread_mutex_unlock(&dt.stat[ofd].lock);
-#endif
+ dt_stat_inc(ofd, snd, qc, len);
} else {
dt_pci_shrink(spb);
- if (dt_pci.eid >= PROG_RES_FDS) {
+ if (dt_pci.eid >= PROC_RES_FDS) {
uint8_t ecn = *(head + dt_pci_info.ecn_o);
fa_np1_rcv(dt_pci.eid, ecn, spb);
return;
@@ -526,20 +496,9 @@ static void packet_handler(int fd,
ipcp_spb_release(spb);
return;
}
-#ifdef IPCP_FLOW_STATS
- pthread_mutex_lock(&dt.stat[fd].lock);
-
- ++dt.stat[fd].lcl_r_pkt[qc];
- dt.stat[fd].lcl_r_bytes[qc] += len;
-
- pthread_mutex_unlock(&dt.stat[fd].lock);
- pthread_mutex_lock(&dt.stat[dt_pci.eid].lock);
-
- ++dt.stat[dt_pci.eid].snd_pkt[qc];
- dt.stat[dt_pci.eid].snd_bytes[qc] += len;
+ dt_stat_inc(fd, lcl_r, qc, len);
+ dt_stat_inc(dt_pci.eid, snd, qc, len);
- pthread_mutex_unlock(&dt.stat[dt_pci.eid].lock);
-#endif
dt.comps[dt_pci.eid].post_packet(dt.comps[dt_pci.eid].comp,
spb);
}
@@ -569,7 +528,9 @@ int dt_init(struct dt_config cfg)
{
int i;
int j;
+#ifdef IPCP_FLOW_STATS
char dtstr[RIB_NAME_STRLEN + 1];
+#endif
enum pol_pff pp;
struct conn_info info;
@@ -636,13 +597,13 @@ int dt_init(struct dt_config cfg)
goto fail_rwlock_init;
}
- dt.res_fds = bmp_create(PROG_RES_FDS, 0);
+ dt.res_fds = bmp_create(PROC_RES_FDS, 0);
if (dt.res_fds == NULL)
goto fail_res_fds;
#ifdef IPCP_FLOW_STATS
memset(dt.stat, 0, sizeof(dt.stat));
- for (i = 0; i < PROG_MAX_FLOWS; ++i)
+ for (i = 0; i < PROC_MAX_FLOWS; ++i)
if (pthread_mutex_init(&dt.stat[i].lock, NULL)) {
log_err("Failed to init mutex for flow %d.", i);
for (j = 0; j < i; ++j)
@@ -651,18 +612,19 @@ int dt_init(struct dt_config cfg)
}
dt.n_flows = 0;
-#endif
+
sprintf(dtstr, "%s." ADDR_FMT32, DT, ADDR_VAL32(&dt.addr));
if (rib_reg(dtstr, &r_ops)) {
log_err("Failed to register RIB.");
goto fail_rib_reg;
}
+#endif
return 0;
- fail_rib_reg:
#ifdef IPCP_FLOW_STATS
- for (i = 0; i < PROG_MAX_FLOWS; ++i)
+ fail_rib_reg:
+ for (i = 0; i < PROC_MAX_FLOWS; ++i)
pthread_mutex_destroy(&dt.stat[i].lock);
fail_stat_lock:
#endif
@@ -685,13 +647,15 @@ int dt_init(struct dt_config cfg)
void dt_fini(void)
{
+#ifdef IPCP_FLOW_STATS
char dtstr[RIB_NAME_STRLEN + 1];
+#endif
int i;
+#ifdef IPCP_FLOW_STATS
sprintf(dtstr, "%s.%" PRIu64, DT, dt.addr);
rib_unreg(dtstr);
-#ifdef IPCP_FLOW_STATS
- for (i = 0; i < PROG_MAX_FLOWS; ++i)
+ for (i = 0; i < PROC_MAX_FLOWS; ++i)
pthread_mutex_destroy(&dt.stat[i].lock);
#endif
bmp_destroy(dt.res_fds);
@@ -791,7 +755,7 @@ int dt_reg_comp(void * comp,
void dt_unreg_comp(int eid)
{
- assert(eid >= 0 && eid < PROG_RES_FDS);
+ assert(eid >= 0 && eid < PROC_RES_FDS);
pthread_rwlock_wrlock(&dt.lock);
@@ -823,33 +787,21 @@ int dt_write_packet(uint64_t dst_addr,
#ifdef IPCP_FLOW_STATS
len = ssm_pk_buff_len(spb);
- if (eid < PROG_RES_FDS) {
- pthread_mutex_lock(&dt.stat[eid].lock);
-
- ++dt.stat[eid].lcl_r_pkt[qc];
- dt.stat[eid].lcl_r_bytes[qc] += len;
-
- pthread_mutex_unlock(&dt.stat[eid].lock);
- }
+ if (eid < PROC_RES_FDS)
+ dt_stat_inc(eid, lcl_r, qc, len);
#endif
fd = pff_nhop(dt.pff[qc], dst_addr);
if (fd < 0) {
log_dbg("Could not get nhop for " ADDR_FMT32 ".",
ADDR_VAL32(&dst_addr));
#ifdef IPCP_FLOW_STATS
- if (eid < PROG_RES_FDS) {
- pthread_mutex_lock(&dt.stat[eid].lock);
-
- ++dt.stat[eid].lcl_r_pkt[qc];
- dt.stat[eid].lcl_r_bytes[qc] += len;
-
- pthread_mutex_unlock(&dt.stat[eid].lock);
- }
+ if (eid < PROC_RES_FDS)
+ dt_stat_inc(eid, lcl_r, qc, len);
#endif
return -EPERM;
}
- head = ssm_pk_buff_head_alloc(spb, dt_pci_info.head_size);
+ head = ssm_pk_buff_push(spb, dt_pci_info.head_size);
if (head == NULL) {
log_dbg("Failed to allocate DT header.");
goto fail_write;
@@ -874,31 +826,17 @@ int dt_write_packet(uint64_t dst_addr,
goto fail_write;
}
#ifdef IPCP_FLOW_STATS
- pthread_mutex_lock(&dt.stat[fd].lock);
-
- if (dt_pci.eid < PROG_RES_FDS) {
- ++dt.stat[fd].lcl_w_pkt[qc];
- dt.stat[fd].lcl_w_bytes[qc] += len;
- }
- ++dt.stat[fd].snd_pkt[qc];
- dt.stat[fd].snd_bytes[qc] += len;
-
- pthread_mutex_unlock(&dt.stat[fd].lock);
+ if (dt_pci.eid < PROC_RES_FDS)
+ dt_stat_inc(fd, lcl_w, qc, len);
+ dt_stat_inc(fd, snd, qc, len);
#endif
return 0;
fail_write:
#ifdef IPCP_FLOW_STATS
- pthread_mutex_lock(&dt.stat[fd].lock);
-
- if (eid < PROG_RES_FDS) {
- ++dt.stat[fd].lcl_w_pkt[qc];
- dt.stat[fd].lcl_w_bytes[qc] += len;
- }
- ++dt.stat[fd].w_drp_pkt[qc];
- dt.stat[fd].w_drp_bytes[qc] += len;
-
- pthread_mutex_unlock(&dt.stat[fd].lock);
+ if (eid < PROC_RES_FDS)
+ dt_stat_inc(fd, lcl_w, qc, len);
+ dt_stat_inc(fd, w_drp, qc, len);
#endif
return -1;
}
diff --git a/src/ipcpd/unicast/fa.c b/src/ipcpd/unicast/fa.c
index ddf78e22..c6eca175 100644
--- a/src/ipcpd/unicast/fa.c
+++ b/src/ipcpd/unicast/fa.c
@@ -37,6 +37,7 @@
#include <ouroboros/errno.h>
#include <ouroboros/dev.h>
#include <ouroboros/ipcp-dev.h>
+#include <ouroboros/np1_flow.h>
#include <ouroboros/rib.h>
#include <ouroboros/random.h>
#include <ouroboros/pthread.h>
@@ -58,12 +59,13 @@
#define CLOCK_REALTIME_COARSE CLOCK_REALTIME
#endif
-#define TIMEOUT 10 * MILLION /* nanoseconds */
+#define TIMEOUT 10 * MILLION /* nanoseconds */
+#define MSGBUFSZ 32768
-#define FLOW_REQ 0
-#define FLOW_REPLY 1
-#define FLOW_UPDATE 2
-#define MSGBUFSZ 2048
+#define FLOW_REQ 0
+#define FLOW_REPLY 1
+#define FLOW_UPDATE 2
+#define FLOW_IRM_UPDATE 3
#define STAT_FILE_LEN 0
@@ -81,7 +83,7 @@ struct fa_msg {
uint16_t ece;
uint8_t code;
uint8_t availability;
- uint8_t in_order;
+ uint8_t service;
} __attribute__((packed));
struct cmd {
@@ -111,7 +113,7 @@ struct fa_flow {
struct {
pthread_rwlock_t flows_lock;
- struct fa_flow flows[PROG_MAX_FLOWS];
+ struct fa_flow flows[PROC_MAX_FLOWS];
#ifdef IPCP_FLOW_STATS
size_t n_flows;
#endif
@@ -125,11 +127,11 @@ struct {
struct psched * psched;
} fa;
+#ifdef IPCP_FLOW_STATS
static int fa_rib_read(const char * path,
char * buf,
size_t len)
{
-#ifdef IPCP_FLOW_STATS
struct fa_flow * flow;
int fd;
char r_addrstr[21];
@@ -145,7 +147,7 @@ static int fa_rib_read(const char * path,
fd = atoi(entry);
- if (fd < 0 || fd >= PROG_MAX_FLOWS)
+ if (fd < 0 || fd >= PROC_MAX_FLOWS)
return -1;
if (len < 1536)
@@ -199,17 +201,10 @@ static int fa_rib_read(const char * path,
pthread_rwlock_unlock(&fa.flows_lock);
return strlen(buf);
-#else
- (void) path;
- (void) buf;
- (void) len;
- return 0;
-#endif
}
static int fa_rib_readdir(char *** buf)
{
-#ifdef IPCP_FLOW_STATS
char entry[RIB_PATH_LEN + 1];
size_t i;
int idx = 0;
@@ -225,7 +220,7 @@ static int fa_rib_readdir(char *** buf)
if (*buf == NULL)
goto fail_entries;
- for (i = 0; i < PROG_MAX_FLOWS; ++i) {
+ for (i = 0; i < PROC_MAX_FLOWS; ++i) {
struct fa_flow * flow;
flow = &fa.flows[i];
@@ -254,16 +249,11 @@ static int fa_rib_readdir(char *** buf)
fail_entries:
pthread_rwlock_unlock(&fa.flows_lock);
return -ENOMEM;
-#else
- (void) buf;
- return 0;
-#endif
}
static int fa_rib_getattr(const char * path,
struct rib_attr * attr)
{
-#ifdef IPCP_FLOW_STATS
int fd;
char * entry;
struct fa_flow * flow;
@@ -286,10 +276,7 @@ static int fa_rib_getattr(const char * path,
}
pthread_rwlock_unlock(&fa.flows_lock);
-#else
- (void) path;
- (void) attr;
-#endif
+
return 0;
}
@@ -298,6 +285,7 @@ static struct rib_ops r_ops = {
.readdir = fa_rib_readdir,
.getattr = fa_rib_getattr
};
+#endif /* IPCP_FLOW_STATS */
static int eid_to_fd(uint64_t eid)
{
@@ -306,7 +294,7 @@ static int eid_to_fd(uint64_t eid)
fd = eid & 0xFFFFFFFF;
- if (fd < 0 || fd >= PROG_MAX_FLOWS)
+ if (fd < 0 || fd >= PROC_MAX_FLOWS)
return -1;
flow = &fa.flows[fd];
@@ -496,11 +484,12 @@ static int fa_handle_flow_req(struct fa_msg * msg,
qs.availability = msg->availability;
qs.loss = ntoh32(msg->loss);
qs.ber = ntoh32(msg->ber);
- qs.in_order = msg->in_order;
+ qs.service = msg->service;
qs.max_gap = ntoh32(msg->max_gap);
qs.timeout = ntoh32(msg->timeout);
- fd = ipcp_wait_flow_req_arr(dst, qs, IPCP_UNICAST_MPL, &data);
+ fd = ipcp_wait_flow_req_arr(dst, qs, IPCP_UNICAST_MPL,
+ IPCP_UNICAST_MTU, &data);
if (fd < 0)
return fd;
@@ -528,7 +517,8 @@ static int fa_handle_flow_reply(struct fa_msg * msg,
time_t mpl = IPCP_UNICAST_MPL;
int response;
- assert(len >= sizeof(*msg));
+ if (len < sizeof(*msg))
+ return -EINVAL;
data.data = (uint8_t *) msg + sizeof(*msg);
data.len = len - sizeof(*msg);
@@ -558,7 +548,8 @@ static int fa_handle_flow_reply(struct fa_msg * msg,
pthread_rwlock_unlock(&fa.flows_lock);
- if (ipcp_flow_alloc_reply(fd, response, mpl, &data) < 0) {
+ if (ipcp_flow_alloc_reply(fd, response, mpl,
+ IPCP_UNICAST_MTU, &data) < 0) {
log_err("Failed to reply for flow allocation on fd %d.", fd);
return -EIRMD;
}
@@ -572,8 +563,8 @@ static int fa_handle_flow_update(struct fa_msg * msg,
struct fa_flow * flow;
int fd;
- (void) len;
- assert(len >= sizeof(*msg));
+ if (len < sizeof(*msg))
+ return -EINVAL;
pthread_rwlock_wrlock(&fa.flows_lock);
@@ -596,6 +587,43 @@ static int fa_handle_flow_update(struct fa_msg * msg,
return 0;
}
+static int fa_handle_flow_irm_update(struct fa_msg * msg,
+ size_t len)
+{
+ buffer_t data;
+ int fd;
+ int flow_id;
+
+ if (len < sizeof(*msg))
+ return -EINVAL;
+
+ data.data = (uint8_t *) msg + sizeof(*msg);
+ data.len = len - sizeof(*msg);
+
+ pthread_rwlock_rdlock(&fa.flows_lock);
+
+ fd = eid_to_fd(ntoh64(msg->r_eid));
+
+ pthread_rwlock_unlock(&fa.flows_lock);
+
+ if (fd < 0) {
+ log_err("Flow update for unknown EID %" PRIu64 ".",
+ ntoh64(msg->r_eid));
+ return -ENOTALLOC;
+ }
+
+ flow_id = np1_flow_id(fd);
+ if (flow_id < 0)
+ return -ENOTALLOC;
+
+ if (ipcp_flow_update_arr(flow_id, &data) < 0) {
+ log_err("Failed to relay flow update on fd %d.", fd);
+ return -EIRMD;
+ }
+
+ return 0;
+}
+
static void * fa_handle_packet(void * o)
{
(void) o;
@@ -624,6 +652,10 @@ static void * fa_handle_packet(void * o)
if (fa_handle_flow_update(msg, len) < 0)
log_err("Error handling flow update.");
break;
+ case FLOW_IRM_UPDATE:
+ if (fa_handle_flow_irm_update(msg, len) < 0)
+ log_err("Error handling flow update.");
+ break;
default:
log_warn("Recieved unknown flow allocation message.");
break;
@@ -652,8 +684,10 @@ int fa_init(void)
if (pthread_cond_init(&fa.cond, &cattr))
goto fail_cond;
+#ifdef IPCP_FLOW_STATS
if (rib_reg(FA, &r_ops))
goto fail_rib_reg;
+#endif
fa.eid = dt_reg_comp(&fa, &fa_post_packet, FA);
if ((int) fa.eid < 0)
@@ -666,8 +700,10 @@ int fa_init(void)
return 0;
fail_dt_reg:
+#ifdef IPCP_FLOW_STATS
rib_unreg(FA);
fail_rib_reg:
+#endif
pthread_cond_destroy(&fa.cond);
fail_cond:
pthread_condattr_destroy(&cattr);
@@ -681,8 +717,9 @@ int fa_init(void)
void fa_fini(void)
{
+#ifdef IPCP_FLOW_STATS
rib_unreg(FA);
-
+#endif
pthread_cond_destroy(&fa.cond);;
pthread_mutex_destroy(&fa.mtx);
pthread_rwlock_destroy(&fa.flows_lock);
@@ -789,7 +826,7 @@ int fa_alloc(int fd,
msg->availability = qs.availability;
msg->loss = hton32(qs.loss);
msg->ber = hton32(qs.ber);
- msg->in_order = qs.in_order;
+ msg->service = qs.service;
msg->max_gap = hton32(qs.max_gap);
msg->timeout = hton32(qs.timeout);
@@ -878,6 +915,44 @@ int fa_alloc_resp(int fd,
return -1;
}
+int fa_irm_update(int fd,
+ const buffer_t * data)
+{
+ struct fa_msg * msg;
+ struct ssm_pk_buff * spb;
+ struct fa_flow * flow;
+ qoscube_t qc = QOS_CUBE_BE;
+ uint64_t r_addr;
+
+ flow = &fa.flows[fd];
+
+ if (ipcp_spb_reserve(&spb, sizeof(*msg) + data->len))
+ return -1;
+
+ msg = (struct fa_msg *) ssm_pk_buff_head(spb);
+ memset(msg, 0, sizeof(*msg));
+
+ msg->code = FLOW_IRM_UPDATE;
+ if (data->len > 0)
+ memcpy(msg + 1, data->data, data->len);
+
+ pthread_rwlock_rdlock(&fa.flows_lock);
+
+ msg->r_eid = hton64(flow->r_eid);
+ msg->s_eid = hton64(flow->s_eid);
+ r_addr = flow->r_addr;
+
+ pthread_rwlock_unlock(&fa.flows_lock);
+
+ if (dt_write_packet(r_addr, qc, fa.eid, spb)) {
+ log_err("Failed to send flow update packet.");
+ ipcp_spb_release(spb);
+ return -1;
+ }
+
+ return 0;
+}
+
int fa_dealloc(int fd)
{
if (ipcp_flow_fini(fd) < 0)
diff --git a/src/ipcpd/unicast/fa.h b/src/ipcpd/unicast/fa.h
index 0c19dc25..f31b40e9 100644
--- a/src/ipcpd/unicast/fa.h
+++ b/src/ipcpd/unicast/fa.h
@@ -45,6 +45,9 @@ int fa_alloc_resp(int fd,
int fa_dealloc(int fd);
+int fa_irm_update(int fd,
+ const buffer_t * data);
+
void fa_np1_rcv(uint64_t eid,
uint8_t ecn,
struct ssm_pk_buff * spb);
diff --git a/src/ipcpd/unicast/main.c b/src/ipcpd/unicast/main.c
index 583a04ff..1155b88b 100644
--- a/src/ipcpd/unicast/main.c
+++ b/src/ipcpd/unicast/main.c
@@ -273,7 +273,8 @@ static struct ipcp_ops unicast_ops = {
.ipcp_flow_alloc = fa_alloc,
.ipcp_flow_join = NULL,
.ipcp_flow_alloc_resp = fa_alloc_resp,
- .ipcp_flow_dealloc = fa_dealloc
+ .ipcp_flow_dealloc = fa_dealloc,
+ .ipcp_flow_update = fa_irm_update
};
int main(int argc,
@@ -307,8 +308,8 @@ int main(int argc,
ipcp_sigwait();
if (ipcp_get_state() == IPCP_SHUTDOWN) {
- stop_components();
ipcp_stop();
+ stop_components();
finalize_components();
} else {
ipcp_stop();
diff --git a/src/ipcpd/unicast/pff/alternate.c b/src/ipcpd/unicast/pff/alternate.c
index be1c35c0..1c508c1b 100644
--- a/src/ipcpd/unicast/pff/alternate.c
+++ b/src/ipcpd/unicast/pff/alternate.c
@@ -211,7 +211,7 @@ struct pff_i * alternate_pff_create(void)
if (pthread_rwlock_init(&tmp->lock, NULL))
goto fail_lock;
- tmp->pft = pft_create(PFT_SIZE, false);
+ tmp->pft = pft_create(PFT_SIZE);
if (tmp->pft == NULL)
goto fail_pft;
diff --git a/src/ipcpd/unicast/pff/multipath.c b/src/ipcpd/unicast/pff/multipath.c
index c636e789..9ba59592 100644
--- a/src/ipcpd/unicast/pff/multipath.c
+++ b/src/ipcpd/unicast/pff/multipath.c
@@ -63,7 +63,7 @@ struct pff_i * multipath_pff_create(void)
if (pthread_rwlock_init(&tmp->lock, NULL))
goto fail_rwlock;
- tmp->pft = pft_create(PFT_SIZE, false);
+ tmp->pft = pft_create(PFT_SIZE);
if (tmp->pft == NULL)
goto fail_pft;
diff --git a/src/ipcpd/unicast/pff/pft.c b/src/ipcpd/unicast/pff/pft.c
index a0d70799..d0e562d6 100644
--- a/src/ipcpd/unicast/pff/pft.c
+++ b/src/ipcpd/unicast/pff/pft.c
@@ -43,12 +43,10 @@ struct pft_entry {
struct pft {
struct list_head * buckets;
- bool hash_key;
uint64_t buckets_size;
};
-struct pft * pft_create(uint64_t buckets,
- bool hash_key)
+struct pft * pft_create(uint64_t buckets)
{
struct pft * tmp;
unsigned int i;
@@ -69,7 +67,6 @@ struct pft * pft_create(uint64_t buckets,
if (tmp == NULL)
return NULL;
- tmp->hash_key = hash_key;
tmp->buckets_size = buckets;
tmp->buckets = malloc(buckets * sizeof(*tmp->buckets));
@@ -113,22 +110,10 @@ void pft_flush(struct pft * pft)
}
}
-static uint64_t hash(uint64_t key)
-{
- uint64_t res[2];
-
- mem_hash(HASH_MD5, res, (uint8_t *) &key, sizeof(key));
-
- return res[0];
-}
-
static uint64_t calc_key(struct pft * pft,
uint64_t dst)
{
- if (pft->hash_key)
- dst = hash(dst);
-
- return (dst & (pft->buckets_size - 1));
+ return hash_mix64(dst) & (pft->buckets_size - 1);
}
int pft_insert(struct pft * pft,
diff --git a/src/ipcpd/unicast/pff/pft.h b/src/ipcpd/unicast/pff/pft.h
index 3bb9cff7..15bbe451 100644
--- a/src/ipcpd/unicast/pff/pft.h
+++ b/src/ipcpd/unicast/pff/pft.h
@@ -24,14 +24,12 @@
#define OUROBOROS_PFT_H
#include <stdint.h>
-#include <stdbool.h>
#include <stdlib.h>
struct pft;
/* Buckets is rounded up to the nearest power of 2 */
-struct pft * pft_create(uint64_t buckets,
- bool hash_key);
+struct pft * pft_create(uint64_t buckets);
void pft_destroy(struct pft * table);
diff --git a/src/ipcpd/unicast/pff/simple.c b/src/ipcpd/unicast/pff/simple.c
index be542bdb..7befa42f 100644
--- a/src/ipcpd/unicast/pff/simple.c
+++ b/src/ipcpd/unicast/pff/simple.c
@@ -63,7 +63,7 @@ struct pff_i * simple_pff_create(void)
return NULL;
}
- tmp->pft = pft_create(PFT_SIZE, false);
+ tmp->pft = pft_create(PFT_SIZE);
if (tmp->pft == NULL) {
pthread_rwlock_destroy(&tmp->lock);
free(tmp);
diff --git a/src/ipcpd/unicast/pff/tests/pft_test.c b/src/ipcpd/unicast/pff/tests/pft_test.c
index 4962c241..20e73a94 100644
--- a/src/ipcpd/unicast/pff/tests/pft_test.c
+++ b/src/ipcpd/unicast/pff/tests/pft_test.c
@@ -38,15 +38,7 @@ int pft_test(int argc,
(void) argc;
(void) argv;
- pft = pft_create(TBL_SIZE, true);
- if (pft == NULL) {
- printf("Failed to create.\n");
- return -1;
- }
-
- pft_destroy(pft);
-
- pft = pft_create(TBL_SIZE, false);
+ pft = pft_create(TBL_SIZE);
if (pft == NULL) {
printf("Failed to create.\n");
return -1;
diff --git a/src/ipcpd/unicast/routing/graph.c b/src/ipcpd/unicast/routing/graph.c
index 0226c762..c168eb7d 100644
--- a/src/ipcpd/unicast/routing/graph.c
+++ b/src/ipcpd/unicast/routing/graph.c
@@ -603,9 +603,9 @@ static int graph_routing_table_lfa(struct graph * graph,
struct list_head * table,
int ** dist)
{
- int * n_dist[PROG_MAX_FLOWS];
- uint64_t addrs[PROG_MAX_FLOWS];
- int n_index[PROG_MAX_FLOWS];
+ int * n_dist[PROC_MAX_FLOWS];
+ uint64_t addrs[PROC_MAX_FLOWS];
+ int n_index[PROC_MAX_FLOWS];
struct list_head * p;
struct list_head * q;
struct vertex * v;
@@ -618,7 +618,7 @@ static int graph_routing_table_lfa(struct graph * graph,
if (graph_routing_table_simple(graph, s_addr, table, dist))
goto fail_table;
- for (j = 0; j < PROG_MAX_FLOWS; j++) {
+ for (j = 0; j < PROC_MAX_FLOWS; j++) {
n_dist[j] = NULL;
n_index[j] = -1;
addrs[j] = -1;
diff --git a/src/ipcpd/unicast/routing/link-state.c b/src/ipcpd/unicast/routing/link-state.c
index 051dd98d..c4ea9e1c 100644
--- a/src/ipcpd/unicast/routing/link-state.c
+++ b/src/ipcpd/unicast/routing/link-state.c
@@ -415,7 +415,7 @@ static void calculate_pff(struct routing_i * instance)
struct list_head table;
struct list_head * p;
struct list_head * q;
- int fds[PROG_MAX_FLOWS];
+ int fds[PROC_MAX_FLOWS];
assert(instance);
generated by cgit v1.2.3 (git 2.39.1) at 2026-07-16 12:10:34 +0000