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.c215
1 files changed, 160 insertions, 55 deletions
diff --git a/src/ipcpd/unicast/fa.c b/src/ipcpd/unicast/fa.c
index ddf78e22..ac2ecaea 100644
--- a/src/ipcpd/unicast/fa.c
+++ b/src/ipcpd/unicast/fa.c
@@ -31,12 +31,15 @@
#define FA "flow-allocator"
#define OUROBOROS_PREFIX FA
+#include <ouroboros/atomics.h>
#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/qoscube.h>
#include <ouroboros/rib.h>
#include <ouroboros/random.h>
#include <ouroboros/pthread.h>
@@ -58,12 +61,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
@@ -79,9 +83,10 @@ struct fa_msg {
uint32_t max_gap;
uint32_t timeout;
uint16_t ece;
+ uint8_t cap;
uint8_t code;
uint8_t availability;
- uint8_t in_order;
+ uint8_t service;
} __attribute__((packed));
struct cmd {
@@ -107,11 +112,13 @@ struct fa_flow {
uint64_t r_eid; /* Remote endpoint id */
uint64_t r_addr; /* Remote address */
void * ctx; /* Congestion avoidance context */
+ uint64_t fair; /* SFQ virtual finish tag (bytes) */
+ uint8_t l_ecn; /* Local first-hop mark (relaxed) */
};
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 +132,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 +152,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 +206,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 +225,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];
@@ -250,20 +250,16 @@ static int fa_rib_readdir(char *** buf)
fail_entry:
while (idx-- > 0)
free((*buf)[idx]);
+
free(*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 +282,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 +291,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 +300,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];
@@ -329,18 +323,21 @@ static uint64_t gen_eid(int fd)
return ((uint64_t) rnd << 32) + fd;
}
-static void packet_handler(int fd,
- qoscube_t qc,
- struct ssm_pk_buff * spb)
+static time_t packet_handler(int fd,
+ qoscube_t qc,
+ struct ssm_pk_buff * spb)
{
struct fa_flow * flow;
uint64_t r_addr;
uint64_t r_eid;
- ca_wnd_t wnd;
+ time_t wait;
size_t len;
+ uint8_t ecn;
flow = &fa.flows[fd];
+ ecn = 0;
+
pthread_rwlock_wrlock(&fa.flows_lock);
len = ssm_pk_buff_len(spb);
@@ -349,16 +346,16 @@ static void packet_handler(int fd,
++flow->p_snd;
flow->b_snd += len;
#endif
- wnd = ca_ctx_update_snd(flow->ctx, len);
+ wait = ca_ctx_update_snd(flow->ctx, len,
+ LOAD_RELAXED(&flow->l_ecn), &flow->fair);
r_addr = flow->r_addr;
r_eid = flow->r_eid;
pthread_rwlock_unlock(&fa.flows_lock);
- ca_wnd_wait(wnd);
-
- if (dt_write_packet(r_addr, qc, r_eid, spb)) {
+ if (dt_write_packet(r_addr, qc, r_eid, spb, &ecn)) {
+ STORE_RELAXED(&flow->l_ecn, ecn);
ipcp_spb_release(spb);
log_dbg("Failed to forward packet.");
#ifdef IPCP_FLOW_STATS
@@ -367,8 +364,12 @@ static void packet_handler(int fd,
flow->b_snd_f += len;
pthread_rwlock_unlock(&fa.flows_lock);
#endif
- return;
+ return wait;
}
+
+ STORE_RELAXED(&flow->l_ecn, ecn);
+
+ return wait;
}
static int fa_flow_init(struct fa_flow * flow)
@@ -382,9 +383,7 @@ static int fa_flow_init(struct fa_flow * flow)
flow->s_eid = -1;
flow->r_addr = INVALID_ADDR;
- flow->ctx = ca_ctx_create();
- if (flow->ctx == NULL)
- return -1;
+ /* ctx is acquired once (r_addr, qc) are known (ca_ctx_get). */
#ifdef IPCP_FLOW_STATS
clock_gettime(CLOCK_REALTIME_COARSE, &now);
@@ -398,7 +397,8 @@ static int fa_flow_init(struct fa_flow * flow)
static void fa_flow_fini(struct fa_flow * flow)
{
- ca_ctx_destroy(flow->ctx);
+ if (flow->ctx != NULL)
+ ca_ctx_put(flow->ctx);
memset(flow, 0, sizeof(*flow));
@@ -496,11 +496,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;
@@ -514,6 +515,13 @@ static int fa_handle_flow_req(struct fa_msg * msg,
flow->r_eid = ntoh64(msg->s_eid);
flow->r_addr = ntoh64(msg->s_addr);
+ flow->ctx = ca_ctx_get(flow->r_addr, qos_spec_to_cube(qs));
+ if (flow->ctx == NULL) {
+ fa_flow_fini(flow);
+ pthread_rwlock_unlock(&fa.flows_lock);
+ return -ENOMEM;
+ }
+
pthread_rwlock_unlock(&fa.flows_lock);
return fd;
@@ -528,7 +536,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 +567,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 +582,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);
@@ -589,10 +599,47 @@ static int fa_handle_flow_update(struct fa_msg * msg,
#ifdef IPCP_FLOW_STATS
flow->u_rcv++;
#endif
- ca_ctx_update_ece(flow->ctx, ntoh16(msg->ece));
+ ca_ctx_update_ece(flow->ctx, ntoh16(msg->ece), msg->cap);
+
+ pthread_rwlock_unlock(&fa.flows_lock);
+
+ 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;
}
@@ -624,6 +671,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 +703,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 +719,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 +736,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 +845,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);
@@ -797,7 +853,7 @@ int fa_alloc(int fd,
if (data->len > 0)
memcpy(ssm_pk_buff_head(spb) + len, data->data, data->len);
- if (dt_write_packet(addr, qc, fa.eid, spb)) {
+ if (dt_write_packet(addr, qc, fa.eid, spb, NULL)) {
log_err("Failed to send flow allocation request packet.");
ipcp_spb_release(spb);
return -1;
@@ -811,6 +867,13 @@ int fa_alloc(int fd,
flow->r_addr = addr;
flow->s_eid = eid;
+ flow->ctx = ca_ctx_get(addr, qos_spec_to_cube(qs));
+ if (flow->ctx == NULL) {
+ fa_flow_fini(flow);
+ pthread_rwlock_unlock(&fa.flows_lock);
+ return -1;
+ }
+
pthread_rwlock_unlock(&fa.flows_lock);
return 0;
@@ -853,7 +916,7 @@ int fa_alloc_resp(int fd,
pthread_rwlock_unlock(&fa.flows_lock);
- if (dt_write_packet(flow->r_addr, qc, fa.eid, spb)) {
+ if (dt_write_packet(flow->r_addr, qc, fa.eid, spb, NULL)) {
log_err("Failed to send flow allocation response packet.");
goto fail_packet;
}
@@ -878,6 +941,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, NULL)) {
+ 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)
@@ -897,7 +998,8 @@ int fa_dealloc(int fd)
}
static int fa_update_remote(int fd,
- uint16_t ece)
+ uint16_t ece,
+ uint8_t cap)
{
struct fa_msg * msg;
struct ssm_pk_buff * spb;
@@ -921,6 +1023,7 @@ static int fa_update_remote(int fd,
msg->code = FLOW_UPDATE;
msg->r_eid = hton64(flow->r_eid);
msg->ece = hton16(ece);
+ msg->cap = cap;
r_addr = flow->r_addr;
#ifdef IPCP_FLOW_STATS
@@ -929,7 +1032,7 @@ static int fa_update_remote(int fd,
pthread_rwlock_unlock(&fa.flows_lock);
- if (dt_write_packet(r_addr, qc, fa.eid, spb)) {
+ if (dt_write_packet(r_addr, qc, fa.eid, spb, NULL)) {
log_err("Failed to send flow update packet.");
ipcp_spb_release(spb);
return -1;
@@ -940,11 +1043,13 @@ static int fa_update_remote(int fd,
void fa_np1_rcv(uint64_t eid,
uint8_t ecn,
+ uint8_t cap,
struct ssm_pk_buff * spb)
{
struct fa_flow * flow;
bool update;
uint16_t ece;
+ uint8_t fcap;
int fd;
size_t len;
@@ -966,7 +1071,7 @@ void fa_np1_rcv(uint64_t eid,
++flow->p_rcv;
flow->b_rcv += len;
#endif
- update = ca_ctx_update_rcv(flow->ctx, len, ecn, &ece);
+ update = ca_ctx_update_rcv(flow->ctx, len, ecn, cap, &ece, &fcap);
pthread_rwlock_unlock(&fa.flows_lock);
@@ -982,5 +1087,5 @@ void fa_np1_rcv(uint64_t eid,
}
if (update)
- fa_update_remote(eid, ece);
+ fa_update_remote(eid, ece, fcap);
}
generated by cgit v1.2.3 (git 2.39.1) at 2026-08-05 07:52:00 +0000