summaryrefslogtreecommitdiff
path: root/src/ipcpd/unicast/fa.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/ipcpd/unicast/fa.c')
-rw-r--r--src/ipcpd/unicast/fa.c684
1 files changed, 407 insertions, 277 deletions
diff --git a/src/ipcpd/unicast/fa.c b/src/ipcpd/unicast/fa.c
index 0d132619..c6eca175 100644
--- a/src/ipcpd/unicast/fa.c
+++ b/src/ipcpd/unicast/fa.c
@@ -1,10 +1,10 @@
/*
- * Ouroboros - Copyright (C) 2016 - 2020
+ * Ouroboros - Copyright (C) 2016 - 2026
*
* Flow allocator of the IPC Process
*
- * Dimitri Staessens <dimitri.staessens@ugent.be>
- * Sander Vrijders <sander.vrijders@ugent.be>
+ * Dimitri Staessens <dimitri@ouroboros.rocks>
+ * Sander Vrijders <sander@ouroboros.rocks>
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License version 2 as
@@ -31,32 +31,41 @@
#define FA "flow-allocator"
#define OUROBOROS_PREFIX FA
+#include <ouroboros/endian.h>
#include <ouroboros/logs.h>
#include <ouroboros/fqueue.h>
#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>
+#include "addr-auth.h"
#include "dir.h"
#include "fa.h"
#include "psched.h"
#include "ipcp.h"
#include "dt.h"
#include "ca.h"
+#include "np1.h"
#include <inttypes.h>
-#include <pthread.h>
#include <stdlib.h>
#include <string.h>
-#define TIMEOUT 10000 /* nanoseconds */
+#if defined (IPCP_FLOW_STATS) && !defined(CLOCK_REALTIME_COARSE)
+#define CLOCK_REALTIME_COARSE CLOCK_REALTIME
+#endif
+
+#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
@@ -64,23 +73,22 @@ struct fa_msg {
uint64_t s_addr;
uint64_t r_eid;
uint64_t s_eid;
- uint8_t code;
- int8_t response;
- uint16_t ece;
- /* QoS parameters from spec, aligned */
- uint8_t availability;
- uint8_t in_order;
- uint32_t delay;
uint64_t bandwidth;
+ int32_t response;
+ uint32_t delay;
uint32_t loss;
uint32_t ber;
uint32_t max_gap;
- uint16_t cypher_s;
+ uint32_t timeout;
+ uint16_t ece;
+ uint8_t code;
+ uint8_t availability;
+ uint8_t service;
} __attribute__((packed));
struct cmd {
struct list_head next;
- struct shm_du_buff * sdb;
+ struct ssm_pk_buff * spb;
};
struct fa_flow {
@@ -94,6 +102,8 @@ struct fa_flow {
size_t p_rcv_f; /* Packets received fail */
size_t b_rcv; /* Bytes received */
size_t b_rcv_f; /* Bytes received fail */
+ size_t u_snd; /* Flow updates sent */
+ size_t u_rcv; /* Flow updates received */
#endif
uint64_t s_eid; /* Local endpoint id */
uint64_t r_eid; /* Remote endpoint id */
@@ -103,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
@@ -117,23 +127,27 @@ struct {
struct psched * psched;
} fa;
-static int fa_stat_read(const char * path,
- char * buf,
- size_t len)
-{
#ifdef IPCP_FLOW_STATS
+static int fa_rib_read(const char * path,
+ char * buf,
+ size_t len)
+{
struct fa_flow * flow;
int fd;
- char r_addrstr[20];
- char s_eidstr[20];
- char r_eidstr[20];
- char tmstr[20];
+ char r_addrstr[21];
+ char s_eidstr[21];
+ char r_eidstr[21];
+ char tmstr[RIB_TM_STRLEN];
char castr[1024];
+ char * entry;
struct tm * tm;
- fd = atoi(path);
+ entry = strstr(path, RIB_SEPARATOR) + 1;
+ assert(entry);
+
+ fd = atoi(entry);
- if (fd < 0 || fd > PROG_MAX_FLOWS)
+ if (fd < 0 || fd >= PROC_MAX_FLOWS)
return -1;
if (len < 1536)
@@ -154,8 +168,8 @@ static int fa_stat_read(const char * path,
sprintf(s_eidstr, "%" PRIu64, flow->s_eid);
sprintf(r_eidstr, "%" PRIu64, flow->r_eid);
- tm = localtime(&flow->stamp);
- strftime(tmstr, sizeof(tmstr), "%F %T", tm);
+ tm = gmtime(&flow->stamp);
+ strftime(tmstr, sizeof(tmstr), RIB_TM_FORMAT, tm);
ca_print_stats(flow->ctx, castr, 1024);
@@ -172,6 +186,8 @@ static int fa_stat_read(const char * path,
"Received (bytes): %20zu\n"
"Receive failed (packets): %20zu\n"
"Receive failed (bytes): %20zu\n"
+ "Sent flow updates (packets): %20zu\n"
+ "Received flow updates (packets): %20zu\n"
"%s",
tmstr, r_addrstr,
s_eidstr, r_eidstr,
@@ -179,22 +195,16 @@ static int fa_stat_read(const char * path,
flow->p_snd_f, flow->b_snd_f,
flow->p_rcv, flow->b_rcv,
flow->b_rcv_f, flow->b_rcv_f,
+ flow->u_snd, flow->u_rcv,
castr);
pthread_rwlock_unlock(&fa.flows_lock);
return strlen(buf);
-#else
- (void) path;
- (void) buf;
- (void) len;
- return 0;
-#endif
}
-static int fa_stat_readdir(char *** buf)
+static int fa_rib_readdir(char *** buf)
{
-#ifdef IPCP_FLOW_STATS
char entry[RIB_PATH_LEN + 1];
size_t i;
int idx = 0;
@@ -202,17 +212,15 @@ static int fa_stat_readdir(char *** buf)
pthread_rwlock_rdlock(&fa.flows_lock);
if (fa.n_flows < 1) {
- pthread_rwlock_unlock(&fa.flows_lock);
- return 0;
+ *buf = NULL;
+ goto no_flows;
}
*buf = malloc(sizeof(**buf) * fa.n_flows);
- if (*buf == NULL) {
- pthread_rwlock_unlock(&fa.flows_lock);
- return -ENOMEM;
- }
+ 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];
@@ -222,67 +230,62 @@ static int fa_stat_readdir(char *** buf)
sprintf(entry, "%zu", i);
(*buf)[idx] = malloc(strlen(entry) + 1);
- if ((*buf)[idx] == NULL) {
- while (idx-- > 0)
- free((*buf)[idx]);
- free(buf);
- pthread_rwlock_unlock(&fa.flows_lock);
- return -ENOMEM;
- }
+ if ((*buf)[idx] == NULL)
+ goto fail_entry;
strcpy((*buf)[idx++], entry);
}
assert((size_t) idx == fa.n_flows);
-
+ no_flows:
pthread_rwlock_unlock(&fa.flows_lock);
return idx;
-#else
- (void) buf;
- return 0;
-#endif
+
+ fail_entry:
+ while (idx-- > 0)
+ free((*buf)[idx]);
+ free(*buf);
+ fail_entries:
+ pthread_rwlock_unlock(&fa.flows_lock);
+ return -ENOMEM;
}
-static int fa_stat_getattr(const char * path,
- struct stat * st)
+static int fa_rib_getattr(const char * path,
+ struct rib_attr * attr)
{
-#ifdef IPCP_FLOW_STATS
int fd;
+ char * entry;
struct fa_flow * flow;
- fd = atoi(path);
+ entry = strstr(path, RIB_SEPARATOR) + 1;
+ assert(entry);
- st->st_mode = S_IFREG | 0755;
- st->st_nlink = 1;
- st->st_uid = getuid();
- st->st_gid = getgid();
+ fd = atoi(entry);
flow = &fa.flows[fd];
pthread_rwlock_rdlock(&fa.flows_lock);
if (flow->stamp != 0) {
- st->st_size = 1536;
- st->st_mtime = flow->stamp;
+ attr->size = 1536;
+ attr->mtime = flow->stamp;
} else {
- st->st_size = 0;
- st->st_mtime = 0;
+ attr->size = 0;
+ attr->mtime = 0;
}
pthread_rwlock_unlock(&fa.flows_lock);
-#else
- (void) path;
- (void) st;
-#endif
+
return 0;
}
static struct rib_ops r_ops = {
- .read = fa_stat_read,
- .readdir = fa_stat_readdir,
- .getattr = fa_stat_getattr
+ .read = fa_rib_read,
+ .readdir = fa_rib_readdir,
+ .getattr = fa_rib_getattr
};
+#endif /* IPCP_FLOW_STATS */
static int eid_to_fd(uint64_t eid)
{
@@ -291,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];
@@ -316,7 +319,7 @@ static uint64_t gen_eid(int fd)
static void packet_handler(int fd,
qoscube_t qc,
- struct shm_du_buff * sdb)
+ struct ssm_pk_buff * spb)
{
struct fa_flow * flow;
uint64_t r_addr;
@@ -328,7 +331,7 @@ static void packet_handler(int fd,
pthread_rwlock_wrlock(&fa.flows_lock);
- len = shm_du_buff_tail(sdb) - shm_du_buff_head(sdb);
+ len = ssm_pk_buff_len(spb);
#ifdef IPCP_FLOW_STATS
++flow->p_snd;
@@ -343,9 +346,9 @@ static void packet_handler(int fd,
ca_wnd_wait(wnd);
- if (dt_write_packet(r_addr, qc, r_eid, sdb)) {
- ipcp_sdb_release(sdb);
- log_warn("Failed to forward packet.");
+ if (dt_write_packet(r_addr, qc, r_eid, spb)) {
+ ipcp_spb_release(spb);
+ log_dbg("Failed to forward packet.");
#ifdef IPCP_FLOW_STATS
pthread_rwlock_wrlock(&fa.flows_lock);
++flow->p_snd_f;
@@ -397,7 +400,7 @@ static void fa_flow_fini(struct fa_flow * flow)
}
static void fa_post_packet(void * comp,
- struct shm_du_buff * sdb)
+ struct ssm_pk_buff * spb)
{
struct cmd * cmd;
@@ -408,11 +411,11 @@ static void fa_post_packet(void * comp,
cmd = malloc(sizeof(*cmd));
if (cmd == NULL) {
log_err("Command failed. Out of memory.");
- ipcp_sdb_release(sdb);
+ ipcp_spb_release(spb);
return;
}
- cmd->sdb = sdb;
+ cmd->spb = spb;
pthread_mutex_lock(&fa.mtx);
@@ -423,175 +426,248 @@ static void fa_post_packet(void * comp,
pthread_mutex_unlock(&fa.mtx);
}
-static void * fa_handle_packet(void * o)
+static size_t fa_wait_for_fa_msg(struct fa_msg * msg)
{
- struct timespec ts = {0, TIMEOUT * 1000};
+ struct cmd * cmd;
+ size_t len;
- (void) o;
+ pthread_mutex_lock(&fa.mtx);
- while (true) {
- struct timespec abstime;
- int fd;
- uint8_t buf[MSGBUFSZ];
- struct fa_msg * msg;
- qosspec_t qs;
- struct cmd * cmd;
- size_t len;
- size_t msg_len;
- struct fa_flow * flow;
+ pthread_cleanup_push(__cleanup_mutex_unlock, &fa.mtx);
- pthread_mutex_lock(&fa.mtx);
+ while (list_is_empty(&fa.cmds))
+ pthread_cond_wait(&fa.cond, &fa.mtx);
- pthread_cleanup_push((void (*)(void *)) pthread_mutex_unlock,
- &fa.mtx);
+ cmd = list_last_entry(&fa.cmds, struct cmd, next);
+ list_del(&cmd->next);
- while (list_is_empty(&fa.cmds))
- pthread_cond_wait(&fa.cond, &fa.mtx);
+ pthread_cleanup_pop(true);
- cmd = list_last_entry(&fa.cmds, struct cmd, next);
- list_del(&cmd->next);
+ len = ssm_pk_buff_len(cmd->spb);
+ if (len > MSGBUFSZ || len < sizeof(*msg)) {
+ log_warn("Invalid flow allocation message (len: %zd).", len);
+ free(cmd);
+ return 0; /* No valid message */
+ }
- pthread_cleanup_pop(true);
+ memcpy(msg, ssm_pk_buff_head(cmd->spb), len);
- len = shm_du_buff_tail(cmd->sdb) - shm_du_buff_head(cmd->sdb);
+ ipcp_spb_release(cmd->spb);
- if (len > MSGBUFSZ) {
- log_err("Message over buffer size.");
- free(cmd);
- continue;
- }
+ free(cmd);
- msg = (struct fa_msg *) buf;
+ return len;
+}
- /* Depending on the message call the function in ipcp-dev.h */
+static int fa_handle_flow_req(struct fa_msg * msg,
+ size_t len)
+{
+ size_t msg_len;
+ int fd;
+ qosspec_t qs;
+ struct fa_flow * flow;
+ uint8_t * dst;
+ buffer_t data; /* Piggbacked data on flow alloc request. */
- memcpy(msg, shm_du_buff_head(cmd->sdb), len);
+ msg_len = sizeof(*msg) + ipcp_dir_hash_len();
+ if (len < msg_len) {
+ log_err("Invalid flow allocation request");
+ return -EPERM;
+ }
- ipcp_sdb_release(cmd->sdb);
+ dst = (uint8_t *)(msg + 1);
+ data.data = (uint8_t *) msg + msg_len;
+ data.len = len - msg_len;
+
+ qs.delay = ntoh32(msg->delay);
+ qs.bandwidth = ntoh64(msg->bandwidth);
+ qs.availability = msg->availability;
+ qs.loss = ntoh32(msg->loss);
+ qs.ber = ntoh32(msg->ber);
+ 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,
+ IPCP_UNICAST_MTU, &data);
+ if (fd < 0)
+ return fd;
- free(cmd);
+ flow = &fa.flows[fd];
- switch (msg->code) {
- case FLOW_REQ:
- msg_len = sizeof(*msg) + ipcp_dir_hash_len();
+ pthread_rwlock_wrlock(&fa.flows_lock);
- assert(len >= msg_len);
+ fa_flow_init(flow);
- clock_gettime(PTHREAD_COND_CLOCK, &abstime);
+ flow->s_eid = gen_eid(fd);
+ flow->r_eid = ntoh64(msg->s_eid);
+ flow->r_addr = ntoh64(msg->s_addr);
- pthread_mutex_lock(&ipcpi.alloc_lock);
+ pthread_rwlock_unlock(&fa.flows_lock);
- while (ipcpi.alloc_id != -1 &&
- ipcp_get_state() == IPCP_OPERATIONAL) {
- ts_add(&abstime, &ts, &abstime);
- pthread_cond_timedwait(&ipcpi.alloc_cond,
- &ipcpi.alloc_lock,
- &abstime);
- }
+ return fd;
+}
- if (ipcp_get_state() != IPCP_OPERATIONAL) {
- pthread_mutex_unlock(&ipcpi.alloc_lock);
- log_dbg("Won't allocate over non-operational"
- "IPCP.");
- continue;
- }
+static int fa_handle_flow_reply(struct fa_msg * msg,
+ size_t len)
+{
+ int fd;
+ struct fa_flow * flow;
+ buffer_t data; /* Piggbacked data on flow alloc request. */
+ time_t mpl = IPCP_UNICAST_MPL;
+ int response;
- assert(ipcpi.alloc_id == -1);
+ if (len < sizeof(*msg))
+ return -EINVAL;
- qs.delay = ntoh32(msg->delay);
- qs.bandwidth = ntoh64(msg->bandwidth);
- qs.availability = msg->availability;
- qs.loss = ntoh32(msg->loss);
- qs.ber = ntoh32(msg->ber);
- qs.in_order = msg->in_order;
- qs.max_gap = ntoh32(msg->max_gap);
- qs.cypher_s = ntoh16(msg->cypher_s);
+ data.data = (uint8_t *) msg + sizeof(*msg);
+ data.len = len - sizeof(*msg);
- fd = ipcp_flow_req_arr((uint8_t *) (msg + 1),
- ipcp_dir_hash_len(),
- qs,
- buf + msg_len,
- len - msg_len);
- if (fd < 0) {
- pthread_mutex_unlock(&ipcpi.alloc_lock);
- log_err("Failed to get fd for flow.");
- continue;
- }
+ pthread_rwlock_wrlock(&fa.flows_lock);
- flow = &fa.flows[fd];
+ fd = eid_to_fd(ntoh64(msg->r_eid));
+ if (fd < 0) {
+ pthread_rwlock_unlock(&fa.flows_lock);
+ log_err("Flow reply for unknown EID %" PRIu64 ".",
+ ntoh64(msg->r_eid));
+ return -ENOTALLOC;
+ }
- pthread_rwlock_wrlock(&fa.flows_lock);
+ flow = &fa.flows[fd];
- fa_flow_init(flow);
+ flow->r_eid = ntoh64(msg->s_eid);
+ response = ntoh32(msg->response);
- flow->s_eid = gen_eid(fd);
- flow->r_eid = ntoh64(msg->s_eid);
- flow->r_addr = ntoh64(msg->s_addr);
+ log_dbg("IPCP received msg response %d for flow on fd %d.",
+ response, fd);
- pthread_rwlock_unlock(&fa.flows_lock);
+ if (response < 0)
+ fa_flow_fini(flow);
+ else
+ psched_add(fa.psched, fd);
- ipcpi.alloc_id = fd;
- pthread_cond_broadcast(&ipcpi.alloc_cond);
+ pthread_rwlock_unlock(&fa.flows_lock);
- pthread_mutex_unlock(&ipcpi.alloc_lock);
+ 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;
+ }
- break;
- case FLOW_REPLY:
- assert(len >= sizeof(*msg));
+ return 0;
+}
+
+static int fa_handle_flow_update(struct fa_msg * msg,
+ size_t len)
+{
+ struct fa_flow * flow;
+ int fd;
+
+ if (len < sizeof(*msg))
+ return -EINVAL;
+
+ pthread_rwlock_wrlock(&fa.flows_lock);
+
+ fd = eid_to_fd(ntoh64(msg->r_eid));
+ if (fd < 0) {
+ pthread_rwlock_unlock(&fa.flows_lock);
+ log_err("Flow update for unknown EID %" PRIu64 ".",
+ ntoh64(msg->r_eid));
+ return -EPERM;
+ }
- pthread_rwlock_wrlock(&fa.flows_lock);
+ flow = &fa.flows[fd];
+#ifdef IPCP_FLOW_STATS
+ flow->u_rcv++;
+#endif
+ ca_ctx_update_ece(flow->ctx, ntoh16(msg->ece));
+
+ pthread_rwlock_unlock(&fa.flows_lock);
+
+ return 0;
+}
- fd = eid_to_fd(ntoh64(msg->r_eid));
- if (fd < 0) {
- pthread_rwlock_unlock(&fa.flows_lock);
- break;
- }
+static int fa_handle_flow_irm_update(struct fa_msg * msg,
+ size_t len)
+{
+ buffer_t data;
+ int fd;
+ int flow_id;
- flow = &fa.flows[fd];
+ if (len < sizeof(*msg))
+ return -EINVAL;
- flow->r_eid = ntoh64(msg->s_eid);
+ data.data = (uint8_t *) msg + sizeof(*msg);
+ data.len = len - sizeof(*msg);
- if (msg->response < 0)
- fa_flow_fini(flow);
- else
- psched_add(fa.psched, fd);
+ pthread_rwlock_rdlock(&fa.flows_lock);
- pthread_rwlock_unlock(&fa.flows_lock);
+ fd = eid_to_fd(ntoh64(msg->r_eid));
- ipcp_flow_alloc_reply(fd,
- msg->response,
- buf + sizeof(*msg),
- len - sizeof(*msg));
- break;
- case FLOW_UPDATE:
- assert(len >= sizeof(*msg));
+ pthread_rwlock_unlock(&fa.flows_lock);
- pthread_rwlock_wrlock(&fa.flows_lock);
+ if (fd < 0) {
+ log_err("Flow update for unknown EID %" PRIu64 ".",
+ ntoh64(msg->r_eid));
+ return -ENOTALLOC;
+ }
- fd = eid_to_fd(ntoh64(msg->r_eid));
- if (fd < 0) {
- pthread_rwlock_unlock(&fa.flows_lock);
- break;
- }
+ flow_id = np1_flow_id(fd);
+ if (flow_id < 0)
+ return -ENOTALLOC;
- flow = &fa.flows[fd];
+ if (ipcp_flow_update_arr(flow_id, &data) < 0) {
+ log_err("Failed to relay flow update on fd %d.", fd);
+ return -EIRMD;
+ }
- ca_ctx_update_ece(flow->ctx, ntoh16(msg->ece));
+ return 0;
+}
- pthread_rwlock_unlock(&fa.flows_lock);
+static void * fa_handle_packet(void * o)
+{
+ (void) o;
+ while (true) {
+ uint8_t buf[MSGBUFSZ];
+ struct fa_msg * msg;
+ size_t len;
+
+ msg = (struct fa_msg *) buf;
+
+ len = fa_wait_for_fa_msg(msg);
+ if (len == 0)
+ continue;
+
+ switch (msg->code) {
+ case FLOW_REQ:
+ if (fa_handle_flow_req(msg, len) < 0)
+ log_err("Error handling flow alloc request.");
+ break;
+ case FLOW_REPLY:
+ if (fa_handle_flow_reply(msg, len) < 0)
+ log_err("Error handling flow reply.");
+ break;
+ case FLOW_UPDATE:
+ 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_err("Got an unknown flow allocation message.");
+ log_warn("Recieved unknown flow allocation message.");
break;
}
}
+
+ return (void *) 0;
}
int fa_init(void)
{
pthread_condattr_t cattr;
- char fastr[256];
if (pthread_rwlock_init(&fa.flows_lock, NULL))
goto fail_rwlock;
@@ -608,21 +684,26 @@ int fa_init(void)
if (pthread_cond_init(&fa.cond, &cattr))
goto fail_cond;
- pthread_condattr_destroy(&cattr);
-
- list_head_init(&fa.cmds);
-
- sprintf(fastr, "%s", FA);
- if (rib_reg(fastr, &r_ops))
+#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)
- goto fail_rib_reg;
+ goto fail_dt_reg;
+
+ list_head_init(&fa.cmds);
+
+ pthread_condattr_destroy(&cattr);
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);
@@ -631,24 +712,34 @@ int fa_init(void)
fail_mtx:
pthread_rwlock_destroy(&fa.flows_lock);
fail_rwlock:
- log_err("Failed to initialize flow allocator.");
return -1;
}
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);
}
+static int np1_flow_read_fa(int fd,
+ struct ssm_pk_buff ** spb)
+{
+ return np1_flow_read(fd, spb, NP1_GET_POOL(fd));
+}
+
int fa_start(void)
{
+#ifndef BUILD_CONTAINER
struct sched_param par;
int pol;
int max;
+#endif
- fa.psched = psched_create(packet_handler);
+ fa.psched = psched_create(packet_handler, np1_flow_read_fa);
if (fa.psched == NULL) {
log_err("Failed to start packet scheduler.");
goto fail_psched;
@@ -659,6 +750,7 @@ int fa_start(void)
goto fail_thread;
}
+#ifndef BUILD_CONTAINER
if (pthread_getschedparam(fa.worker, &pol, &par)) {
log_err("Failed to get worker thread scheduling parameters.");
goto fail_sched;
@@ -676,16 +768,18 @@ int fa_start(void)
log_err("Failed to set scheduler priority to maximum.");
goto fail_sched;
}
+#endif
return 0;
+#ifndef BUILD_CONTAINER
fail_sched:
pthread_cancel(fa.worker);
pthread_join(fa.worker, NULL);
+#endif
fail_thread:
psched_destroy(fa.psched);
fail_psched:
- log_err("Failed to start flow allocator.");
return -1;
}
@@ -697,14 +791,13 @@ void fa_stop(void)
psched_destroy(fa.psched);
}
-int fa_alloc(int fd,
- const uint8_t * dst,
- qosspec_t qs,
- const void * data,
- size_t dlen)
+int fa_alloc(int fd,
+ const uint8_t * dst,
+ qosspec_t qs,
+ const buffer_t * data)
{
struct fa_msg * msg;
- struct shm_du_buff * sdb;
+ struct ssm_pk_buff * spb;
struct fa_flow * flow;
uint64_t addr;
qoscube_t qc = QOS_CUBE_BE;
@@ -717,31 +810,33 @@ int fa_alloc(int fd,
len = sizeof(*msg) + ipcp_dir_hash_len();
- if (ipcp_sdb_reserve(&sdb, len + dlen))
+ if (ipcp_spb_reserve(&spb, len + data->len))
return -1;
- msg = (struct fa_msg *) shm_du_buff_head(sdb);
+ msg = (struct fa_msg *) ssm_pk_buff_head(spb);
memset(msg, 0, sizeof(*msg));
eid = gen_eid(fd);
msg->code = FLOW_REQ;
msg->s_eid = hton64(eid);
- msg->s_addr = hton64(ipcpi.dt_addr);
+ msg->s_addr = hton64(addr_auth_address());
msg->delay = hton32(qs.delay);
msg->bandwidth = hton64(qs.bandwidth);
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->cypher_s = hton16(qs.cypher_s);
+ msg->timeout = hton32(qs.timeout);
memcpy(msg + 1, dst, ipcp_dir_hash_len());
- memcpy(shm_du_buff_head(sdb) + len, data, dlen);
+ if (data->len > 0)
+ memcpy(ssm_pk_buff_head(spb) + len, data->data, data->len);
- if (dt_write_packet(addr, qc, fa.eid, sdb)) {
- ipcp_sdb_release(sdb);
+ if (dt_write_packet(addr, qc, fa.eid, spb)) {
+ log_err("Failed to send flow allocation request packet.");
+ ipcp_spb_release(spb);
return -1;
}
@@ -758,74 +853,103 @@ int fa_alloc(int fd,
return 0;
}
-int fa_alloc_resp(int fd,
- int response,
- const void * data,
- size_t len)
+int fa_alloc_resp(int fd,
+ int response,
+ const buffer_t * data)
{
- struct timespec ts = {0, TIMEOUT * 1000};
- struct timespec abstime;
struct fa_msg * msg;
- struct shm_du_buff * sdb;
+ struct ssm_pk_buff * spb;
struct fa_flow * flow;
qoscube_t qc = QOS_CUBE_BE;
- clock_gettime(PTHREAD_COND_CLOCK, &abstime);
-
flow = &fa.flows[fd];
- pthread_mutex_lock(&ipcpi.alloc_lock);
-
- while (ipcpi.alloc_id != fd && ipcp_get_state() == IPCP_OPERATIONAL) {
- ts_add(&abstime, &ts, &abstime);
- pthread_cond_timedwait(&ipcpi.alloc_cond,
- &ipcpi.alloc_lock,
- &abstime);
+ if (ipcp_wait_flow_resp(fd) < 0) {
+ log_err("Failed to wait for flow response.");
+ goto fail_alloc_resp;
}
- if (ipcp_get_state() != IPCP_OPERATIONAL) {
- pthread_mutex_unlock(&ipcpi.alloc_lock);
- return -1;
+ if (ipcp_spb_reserve(&spb, sizeof(*msg) + data->len)) {
+ log_err("Failed to reserve spb (%zu bytes).",
+ sizeof(*msg) + data->len);
+ goto fail_reserve;
}
- ipcpi.alloc_id = -1;
- pthread_cond_broadcast(&ipcpi.alloc_cond);
-
- pthread_mutex_unlock(&ipcpi.alloc_lock);
-
- if (ipcp_sdb_reserve(&sdb, sizeof(*msg) + len)) {
- fa_flow_fini(flow);
- return -1;
- }
-
- msg = (struct fa_msg *) shm_du_buff_head(sdb);
+ msg = (struct fa_msg *) ssm_pk_buff_head(spb);
memset(msg, 0, sizeof(*msg));
- pthread_rwlock_wrlock(&fa.flows_lock);
-
msg->code = FLOW_REPLY;
+ msg->response = hton32(response);
+ 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);
- msg->response = response;
- memcpy(msg + 1, data, len);
+ pthread_rwlock_unlock(&fa.flows_lock);
+
+ if (dt_write_packet(flow->r_addr, qc, fa.eid, spb)) {
+ log_err("Failed to send flow allocation response packet.");
+ goto fail_packet;
+ }
if (response < 0) {
+ pthread_rwlock_wrlock(&fa.flows_lock);
fa_flow_fini(flow);
- ipcp_sdb_release(sdb);
+ pthread_rwlock_unlock(&fa.flows_lock);
} else {
psched_add(fa.psched, fd);
}
- if (dt_write_packet(flow->r_addr, qc, fa.eid, sdb)) {
- fa_flow_fini(flow);
- pthread_rwlock_unlock(&fa.flows_lock);
- ipcp_sdb_release(sdb);
+ return 0;
+
+ fail_packet:
+ ipcp_spb_release(spb);
+ fail_reserve:
+ pthread_rwlock_wrlock(&fa.flows_lock);
+ fa_flow_fini(flow);
+ pthread_rwlock_unlock(&fa.flows_lock);
+ fail_alloc_resp:
+ 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;
}
@@ -842,7 +966,7 @@ int fa_dealloc(int fd)
pthread_rwlock_unlock(&fa.flows_lock);
- flow_dealloc(fd);
+ ipcp_flow_dealloc(fd);
return 0;
}
@@ -851,34 +975,38 @@ static int fa_update_remote(int fd,
uint16_t ece)
{
struct fa_msg * msg;
- struct shm_du_buff * sdb;
+ struct ssm_pk_buff * spb;
qoscube_t qc = QOS_CUBE_BE;
struct fa_flow * flow;
uint64_t r_addr;
- if (ipcp_sdb_reserve(&sdb, sizeof(*msg))) {
+ if (ipcp_spb_reserve(&spb, sizeof(*msg))) {
+ log_err("Failed to reserve spb (%zu bytes).", sizeof(*msg));
return -1;
}
- msg = (struct fa_msg *) shm_du_buff_head(sdb);
+ msg = (struct fa_msg *) ssm_pk_buff_head(spb);
memset(msg, 0, sizeof(*msg));
flow = &fa.flows[fd];
- pthread_rwlock_rdlock(&fa.flows_lock);
+ pthread_rwlock_wrlock(&fa.flows_lock);
msg->code = FLOW_UPDATE;
msg->r_eid = hton64(flow->r_eid);
msg->ece = hton16(ece);
r_addr = flow->r_addr;
-
+#ifdef IPCP_FLOW_STATS
+ flow->u_snd++;
+#endif
pthread_rwlock_unlock(&fa.flows_lock);
- if (dt_write_packet(r_addr, qc, fa.eid, sdb)) {
- ipcp_sdb_release(sdb);
+ if (dt_write_packet(r_addr, qc, fa.eid, spb)) {
+ log_err("Failed to send flow update packet.");
+ ipcp_spb_release(spb);
return -1;
}
@@ -887,7 +1015,7 @@ static int fa_update_remote(int fd,
void fa_np1_rcv(uint64_t eid,
uint8_t ecn,
- struct shm_du_buff * sdb)
+ struct ssm_pk_buff * spb)
{
struct fa_flow * flow;
bool update;
@@ -895,14 +1023,15 @@ void fa_np1_rcv(uint64_t eid,
int fd;
size_t len;
- len = shm_du_buff_tail(sdb) - shm_du_buff_head(sdb);
+ len = ssm_pk_buff_len(spb);
pthread_rwlock_wrlock(&fa.flows_lock);
fd = eid_to_fd(eid);
if (fd < 0) {
pthread_rwlock_unlock(&fa.flows_lock);
- ipcp_sdb_release(sdb);
+ log_dbg("Received packet for unknown EID %" PRIu64 ".", eid);
+ ipcp_spb_release(spb);
return;
}
@@ -916,8 +1045,9 @@ void fa_np1_rcv(uint64_t eid,
pthread_rwlock_unlock(&fa.flows_lock);
- if (ipcp_flow_write(fd, sdb) < 0) {
- ipcp_sdb_release(sdb);
+ if (np1_flow_write(fd, spb, NP1_GET_POOL(fd)) < 0) {
+ log_dbg("Failed to write to flow %d.", fd);
+ ipcp_spb_release(spb);
#ifdef IPCP_FLOW_STATS
pthread_rwlock_wrlock(&fa.flows_lock);
++flow->p_rcv_f;
generated by cgit v1.2.3 (git 2.39.1) at 2026-07-13 04:45:09 +0000