summaryrefslogtreecommitdiff
path: root/src/ipcpd
diff options
context:
space:
mode:
Diffstat (limited to 'src/ipcpd')
-rw-r--r--src/ipcpd/broadcast/dt.c2
-rw-r--r--src/ipcpd/broadcast/main.c7
-rw-r--r--src/ipcpd/config.h.in13
-rw-r--r--src/ipcpd/eth/eth.c671
-rw-r--r--src/ipcpd/ipcp.c37
-rw-r--r--src/ipcpd/ipcp.h4
-rw-r--r--src/ipcpd/local/main.c58
-rw-r--r--src/ipcpd/udp/udp.c155
-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
21 files changed, 1111 insertions, 307 deletions
diff --git a/src/ipcpd/broadcast/dt.c b/src/ipcpd/broadcast/dt.c
index 30e89a4f..95483e33 100644
--- a/src/ipcpd/broadcast/dt.c
+++ b/src/ipcpd/broadcast/dt.c
@@ -28,7 +28,7 @@
#include "config.h"
-#define BROADCAST_MTU 1400 /* FIXME: avoid packet copy. */
+#define BROADCAST_MTU IPCP_BROADCAST_MTU /* FIXME: avoid packet copy. */
#define DT "dt"
#define OUROBOROS_PREFIX DT
diff --git a/src/ipcpd/broadcast/main.c b/src/ipcpd/broadcast/main.c
index b3cbdc79..d18cac82 100644
--- a/src/ipcpd/broadcast/main.c
+++ b/src/ipcpd/broadcast/main.c
@@ -242,7 +242,7 @@ static int broadcast_ipcp_join(int fd,
notifier_event(NOTIFY_DT_CONN_ADD, &conn);
- ipcp_flow_alloc_reply(fd, 0, mpl, &data);
+ ipcp_flow_alloc_reply(fd, 0, mpl, IPCP_BROADCAST_MTU, &data);
return 0;
}
@@ -307,12 +307,13 @@ int main(int argc,
ipcp_sigwait();
if (ipcp_get_state() == IPCP_SHUTDOWN) {
+ ipcp_stop();
stop_components();
finalize_components();
+ } else {
+ ipcp_stop();
}
- ipcp_stop();
-
enroll_fini();
connmgr_fini();
diff --git a/src/ipcpd/config.h.in b/src/ipcpd/config.h.in
index 0b4252e5..7edec526 100644
--- a/src/ipcpd/config.h.in
+++ b/src/ipcpd/config.h.in
@@ -23,8 +23,8 @@
#define PTHREAD_COND_CLOCK @PTHREAD_COND_CLOCK@
#define SYS_MAX_FLOWS @SYS_MAX_FLOWS@
-#define PROG_RES_FDS @PROG_RES_FDS@
-#define PROG_MAX_FLOWS @PROG_MAX_FLOWS@
+#define PROC_RES_FDS @PROC_RES_FDS@
+#define PROC_MAX_FLOWS @PROC_MAX_FLOWS@
#define SOCKET_TIMEOUT @SOCKET_TIMEOUT@
#define CONNECT_TIMEOUT @CONNECT_TIMEOUT@
@@ -46,11 +46,13 @@
#define IPCP_SCHED_THR_MUL @IPCP_SCHED_THR_MUL@
#define PFT_SIZE @PFT_SIZE@
#define IPCP_UNICAST_MPL @IPCP_UNICAST_MPL@
+#define IPCP_UNICAST_MTU @IPCP_UNICAST_MTU@
#define CONNMGR_RCV_TIMEOUT @CONNMGR_RCV_TIMEOUT@
#cmakedefine DISABLE_CORE_LOCK
#cmakedefine BUILD_CONTAINER
#cmakedefine IPCP_FLOW_STATS
+#cmakedefine IPCP_ETH_FLOW_STATS
#cmakedefine IPCP_DEBUG_LOCAL
#ifdef CONFIG_OUROBOROS_DEBUG
#cmakedefine DEBUG_PROTO_DHT
@@ -65,6 +67,8 @@
#define IPCP_UDP_RD_THR @IPCP_UDP_RD_THR@
#define IPCP_UDP_WR_THR @IPCP_UDP_WR_THR@
#define IPCP_UDP_MPL @IPCP_UDP_MPL@
+#define IPCP_UDP4_MTU @IPCP_UDP4_MTU@
+#define IPCP_UDP6_MTU @IPCP_UDP6_MTU@
/* eth */
#cmakedefine HAVE_NETMAP
@@ -76,10 +80,13 @@
#define IPCP_ETH_LO_MTU @IPCP_ETH_LO_MTU@
#define IPCP_ETH_MGMT_FRAME_SIZE @IPCP_ETH_MGMT_FRAME_SIZE@
#define IPCP_ETH_MPL @IPCP_ETH_MPL@
+#define IPCP_ETH_SNDBUF @IPCP_ETH_SNDBUF@
+#define IPCP_ETH_RCVBUF @IPCP_ETH_RCVBUF@
/* local */
#define IPCP_LOCAL_MPL @IPCP_LOCAL_MPL@
+#define IPCP_LOCAL_MTU @IPCP_LOCAL_MTU@
/* broadcast */
-/* local */
#define IPCP_BROADCAST_MPL @IPCP_BROADCAST_MPL@
+#define IPCP_BROADCAST_MTU @IPCP_BROADCAST_MTU@
diff --git a/src/ipcpd/eth/eth.c b/src/ipcpd/eth/eth.c
index 4be7775e..2d3bd249 100644
--- a/src/ipcpd/eth/eth.c
+++ b/src/ipcpd/eth/eth.c
@@ -37,19 +37,30 @@
#include "config.h"
+#include <ouroboros/atomics.h>
#include <ouroboros/endian.h>
#include <ouroboros/hash.h>
#include <ouroboros/errno.h>
#include <ouroboros/list.h>
#include <ouroboros/utils.h>
#include <ouroboros/bitmap.h>
+#include <ouroboros/crc8.h>
#include <ouroboros/dev.h>
#include <ouroboros/ipcp-dev.h>
#include <ouroboros/fqueue.h>
#include <ouroboros/logs.h>
+#include <ouroboros/np1_flow.h>
#include <ouroboros/time.h>
#include <ouroboros/fccntl.h>
#include <ouroboros/pthread.h>
+#include <ouroboros/rib.h>
+
+#ifndef IPCP_ETH_FLOW_STATS
+#undef FETCH_ADD_RELAXED
+#define FETCH_ADD_RELAXED(p, v) ((void) 0)
+#undef FETCH_SUB_RELAXED
+#define FETCH_SUB_RELAXED(p, v) ((void) 0)
+#endif
#include "ipcp.h"
#include "np1.h"
@@ -72,6 +83,7 @@
#include <linux/if_ether.h>
#include <linux/netlink.h>
#include <linux/rtnetlink.h>
+#include <linux/sockios.h>
#endif
#ifdef __FreeBSD__
@@ -122,7 +134,8 @@
#define MGMT_EID 0
#define DIX_EID_SIZE sizeof(uint16_t)
#define DIX_LENGTH_SIZE sizeof(uint16_t)
-#define DIX_HEADER_SIZE (DIX_EID_SIZE + DIX_LENGTH_SIZE)
+#define DIX_HCS_SIZE CRC8_HASH_LEN
+#define DIX_HEADER_SIZE (DIX_EID_SIZE + DIX_LENGTH_SIZE + DIX_HCS_SIZE)
#define ETH_HEADER_TOT_SIZE (ETH_HEADER_SIZE + DIX_HEADER_SIZE)
#define MAX_EIDS (1 << (8 * DIX_EID_SIZE))
#define ETH_MAX_PACKET_SIZE (ETH_MTU - DIX_HEADER_SIZE)
@@ -130,21 +143,26 @@
#elif defined(BUILD_ETH_LLC)
#define THIS_TYPE IPCP_ETH_LLC
#define MGMT_SAP 0x01
-#define LLC_HEADER_SIZE 3
+#define LLC_FIELDS_SIZE 3
+#define LLC_HCS_SIZE CRC8_HASH_LEN
+#define LLC_HEADER_SIZE (LLC_FIELDS_SIZE + LLC_HCS_SIZE)
#define ETH_HEADER_TOT_SIZE (ETH_HEADER_SIZE + LLC_HEADER_SIZE)
#define MAX_SAPS 64
#define ETH_MAX_PACKET_SIZE (ETH_MTU - LLC_HEADER_SIZE)
#define ETH_FRAME_SIZE (ETH_HEADER_SIZE + ETH_MTU_MAX)
#endif
-#define NAME_QUERY_TIMEO 2000 /* ms */
-#define MGMT_TIMEO 100 /* ms */
+#define NAME_QUERY_TIMEO 1900 /* ms total budget */
+#define NAME_QUERY_RETRIES 3 /* retransmits, 4 attempts total */
+#define MGMT_TIMEO 100 /* ms */
#define MGMT_FRAME_SIZE IPCP_ETH_MGMT_FRAME_SIZE
+#define ETH_RIB_PATH "eth"
#define FLOW_REQ 0
#define FLOW_REPLY 1
#define NAME_QUERY_REQ 2
#define NAME_QUERY_REPLY 3
+#define FLOW_IRM_UPDATE 4
struct mgmt_msg {
#if defined(BUILD_ETH_DIX)
@@ -165,7 +183,7 @@ struct mgmt_msg {
uint32_t delay;
uint32_t timeout;
int32_t response;
- uint8_t in_order;
+ uint8_t service;
#if defined (BUILD_ETH_DIX)
uint8_t code;
uint8_t availability;
@@ -185,6 +203,7 @@ struct eth_frame {
uint8_t ssap;
uint8_t cf;
#endif
+ uint8_t hcs;
uint8_t payload;
} __attribute__((packed));
@@ -196,6 +215,17 @@ struct ef {
int8_t r_sap;
#endif
uint8_t r_addr[MAC_SIZE];
+#ifdef IPCP_ETH_FLOW_STATS
+ struct {
+ time_t stamp;
+ size_t p_rcv;
+ size_t b_rcv;
+ size_t p_dlv_f;
+ size_t p_snd;
+ size_t b_snd;
+ size_t p_snd_f;
+ } stat;
+#endif
};
struct mgmt_frame {
@@ -233,6 +263,22 @@ struct {
struct ef * fd_to_ef;
fset_t * np1_flows;
pthread_rwlock_t flows_lock;
+#ifdef IPCP_ETH_FLOW_STATS
+ struct {
+ size_t n_flows;
+ size_t n_rcv;
+ size_t n_snd;
+ size_t n_mgmt_rcv;
+ size_t n_mgmt_snd;
+ size_t n_bad_id;
+ size_t n_dlv_f;
+ size_t n_buf_f;
+ size_t n_rcv_f;
+ size_t n_snd_f;
+ size_t kern_rcv;
+ size_t kern_drp;
+ } stat;
+#endif
pthread_t packet_writer[IPCP_ETH_WR_THR];
pthread_t packet_reader[IPCP_ETH_RD_THR];
@@ -284,7 +330,14 @@ static int eth_data_init(void)
eth_data.fd_to_ef[i].r_sap = -1;
#endif
memset(&eth_data.fd_to_ef[i].r_addr, 0, MAC_SIZE);
+#ifdef IPCP_ETH_FLOW_STATS
+ memset(&eth_data.fd_to_ef[i].stat, 0,
+ sizeof(eth_data.fd_to_ef[i].stat));
+#endif
}
+#ifdef IPCP_ETH_FLOW_STATS
+ memset(&eth_data.stat, 0, sizeof(eth_data.stat));
+#endif
eth_data.shim_data = shim_data_create();
if (eth_data.shim_data == NULL)
@@ -357,6 +410,232 @@ static void eth_data_fini(void)
free(eth_data.fd_to_ef);
}
+#ifdef IPCP_ETH_FLOW_STATS
+static int eth_rib_read(const char * path,
+ char * buf,
+ size_t len)
+{
+ struct ef * flow;
+ int fd;
+ char tmstr[RIB_TM_STRLEN];
+ struct tm * tm;
+ time_t stamp;
+ char * entry;
+
+ entry = strstr(path, RIB_SEPARATOR) + 1;
+ assert(entry);
+
+ if (len < 2048)
+ return 0;
+
+ buf[0] = '\0';
+
+ if (strcmp(entry, "summary") == 0) {
+ int n;
+#if defined(HAVE_RAW_SOCKETS)
+ int rcvbuf = 0;
+ int sndbuf = 0;
+ int queued = 0;
+ socklen_t optlen = sizeof(rcvbuf);
+# if defined(__linux__)
+ int outq = 0;
+ struct tpacket_stats tp_stats;
+ socklen_t tp_len = sizeof(tp_stats);
+# endif
+
+ getsockopt(eth_data.s_fd, SOL_SOCKET,
+ SO_RCVBUF, &rcvbuf, &optlen);
+ optlen = sizeof(sndbuf);
+ getsockopt(eth_data.s_fd, SOL_SOCKET,
+ SO_SNDBUF, &sndbuf, &optlen);
+ ioctl(eth_data.s_fd, FIONREAD, &queued);
+# if defined(__linux__)
+ ioctl(eth_data.s_fd, SIOCOUTQ, &outq);
+ if (getsockopt(eth_data.s_fd, SOL_PACKET,
+ PACKET_STATISTICS,
+ &tp_stats, &tp_len) == 0) {
+ FETCH_ADD_RELAXED(&eth_data.stat.kern_rcv,
+ tp_stats.tp_packets);
+ FETCH_ADD_RELAXED(&eth_data.stat.kern_drp,
+ tp_stats.tp_drops);
+ }
+# endif
+#endif
+ n = sprintf(buf,
+ "Active flows: %20zu\n"
+ "Total frames received: %20zu\n"
+ "Total frames sent: %20zu\n"
+ "Management frames received: %20zu\n"
+ "Management frames sent: %20zu\n"
+ "Bad EID/SAP frames: %20zu\n"
+ "Delivery (N+1) failures: %20zu\n"
+ "Buffer alloc failures: %20zu\n"
+ "Frame read failures: %20zu\n"
+ "Frame send failures: %20zu\n",
+ LOAD_RELAXED(&eth_data.stat.n_flows),
+ LOAD_RELAXED(&eth_data.stat.n_rcv),
+ LOAD_RELAXED(&eth_data.stat.n_snd),
+ LOAD_RELAXED(&eth_data.stat.n_mgmt_rcv),
+ LOAD_RELAXED(&eth_data.stat.n_mgmt_snd),
+ LOAD_RELAXED(&eth_data.stat.n_bad_id),
+ LOAD_RELAXED(&eth_data.stat.n_dlv_f),
+ LOAD_RELAXED(&eth_data.stat.n_buf_f),
+ LOAD_RELAXED(&eth_data.stat.n_rcv_f),
+ LOAD_RELAXED(&eth_data.stat.n_snd_f));
+#if defined(HAVE_RAW_SOCKETS)
+ n += sprintf(buf + n,
+ "Socket rcvbuf (bytes): %20d\n"
+ "Socket sndbuf (bytes): %20d\n"
+ "Socket ingress (bytes): %20d\n",
+ rcvbuf, sndbuf, queued);
+# if defined(__linux__)
+ n += sprintf(buf + n,
+ "Socket egress (bytes): %20d\n"
+ "Kernel frames received: %20zu\n"
+ "Kernel frames dropped: %20zu\n",
+ outq,
+ LOAD_RELAXED(&eth_data.stat.kern_rcv),
+ LOAD_RELAXED(&eth_data.stat.kern_drp));
+# endif
+#endif
+ return n;
+ }
+
+ fd = atoi(entry);
+
+ if (fd < 0 || fd >= SYS_MAX_FLOWS)
+ return -1;
+
+ flow = &eth_data.fd_to_ef[fd];
+
+ pthread_rwlock_rdlock(&eth_data.flows_lock);
+
+ stamp = flow->stat.stamp;
+ if (stamp == 0) {
+ pthread_rwlock_unlock(&eth_data.flows_lock);
+ return 0;
+ }
+
+ pthread_rwlock_unlock(&eth_data.flows_lock);
+
+ tm = gmtime(&stamp);
+ strftime(tmstr, sizeof(tmstr), RIB_TM_FORMAT, tm);
+
+ sprintf(buf,
+ "Flow established at: %20s\n"
+ "Sent (packets): %20zu\n"
+ "Sent (bytes): %20zu\n"
+ "Send failed (packets): %20zu\n"
+ "Received (packets): %20zu\n"
+ "Received (bytes): %20zu\n"
+ "Delivery (N+1) failures: %20zu\n",
+ tmstr,
+ LOAD_RELAXED(&flow->stat.p_snd),
+ LOAD_RELAXED(&flow->stat.b_snd),
+ LOAD_RELAXED(&flow->stat.p_snd_f),
+ LOAD_RELAXED(&flow->stat.p_rcv),
+ LOAD_RELAXED(&flow->stat.b_rcv),
+ LOAD_RELAXED(&flow->stat.p_dlv_f));
+
+ return strlen(buf);
+}
+
+static int eth_rib_readdir(char *** buf)
+{
+ char entry[RIB_PATH_LEN + 1];
+ size_t i;
+ int idx = 0;
+ int n_entries;
+
+ pthread_rwlock_rdlock(&eth_data.flows_lock);
+
+ n_entries = (int) LOAD_RELAXED(&eth_data.stat.n_flows) + 1;
+
+ *buf = malloc(sizeof(**buf) * n_entries);
+ if (*buf == NULL)
+ goto fail_entries;
+
+ (*buf)[idx] = malloc(strlen("summary") + 1);
+ if ((*buf)[idx] == NULL)
+ goto fail_entry;
+
+ strcpy((*buf)[idx++], "summary");
+
+ for (i = 0; i < SYS_MAX_FLOWS && idx < n_entries; ++i) {
+ if (eth_data.fd_to_ef[i].stat.stamp == 0)
+ continue;
+
+ sprintf(entry, "%zu", i);
+
+ (*buf)[idx] = malloc(strlen(entry) + 1);
+ if ((*buf)[idx] == NULL)
+ goto fail_entry;
+
+ strcpy((*buf)[idx++], entry);
+ }
+
+ pthread_rwlock_unlock(&eth_data.flows_lock);
+
+ return idx;
+
+ fail_entry:
+ while (idx-- > 0)
+ free((*buf)[idx]);
+
+ free(*buf);
+ fail_entries:
+ pthread_rwlock_unlock(&eth_data.flows_lock);
+ return -ENOMEM;
+}
+
+static int eth_rib_getattr(const char * path,
+ struct rib_attr * attr)
+{
+ int fd;
+ char * entry;
+ struct ef * flow;
+
+ entry = strstr(path, RIB_SEPARATOR) + 1;
+ assert(entry);
+
+ if (strcmp(entry, "summary") == 0) {
+ attr->size = 2048;
+ attr->mtime = 0;
+ return 0;
+ }
+
+ fd = atoi(entry);
+
+ if (fd < 0 || fd >= SYS_MAX_FLOWS) {
+ attr->size = 0;
+ attr->mtime = 0;
+ return 0;
+ }
+
+ flow = &eth_data.fd_to_ef[fd];
+
+ pthread_rwlock_rdlock(&eth_data.flows_lock);
+
+ if (flow->stat.stamp != 0) {
+ attr->size = 2048;
+ attr->mtime = flow->stat.stamp;
+ } else {
+ attr->size = 0;
+ attr->mtime = 0;
+ }
+
+ pthread_rwlock_unlock(&eth_data.flows_lock);
+
+ return 0;
+}
+
+static struct rib_ops eth_r_ops = {
+ .read = eth_rib_read,
+ .readdir = eth_rib_readdir,
+ .getattr = eth_rib_getattr
+};
+#endif /* IPCP_ETH_FLOW_STATS */
+
#ifdef BUILD_ETH_LLC
static uint8_t reverse_bits(uint8_t b)
{
@@ -409,12 +688,18 @@ static int eth_ipcp_send_frame(const uint8_t * dst_addr,
e_frame->ethertype = eth_data.ethertype;
e_frame->eid = htons(deid);
e_frame->length = htons(len);
+ mem_hash(HASH_CRC8, &e_frame->hcs,
+ (uint8_t *) &e_frame->eid,
+ DIX_EID_SIZE + DIX_LENGTH_SIZE);
frame_len = ETH_HEADER_TOT_SIZE + len;
#elif defined(BUILD_ETH_LLC)
e_frame->length = htons(LLC_HEADER_SIZE + len);
e_frame->dsap = dsap;
e_frame->ssap = ssap;
e_frame->cf = cf;
+ mem_hash(HASH_CRC8, &e_frame->hcs,
+ (uint8_t *) &e_frame->dsap,
+ LLC_FIELDS_SIZE);
frame_len = ETH_HEADER_TOT_SIZE + len;
#endif
@@ -440,10 +725,7 @@ static int eth_ipcp_send_frame(const uint8_t * dst_addr,
}
assert(FD_ISSET(eth_data.s_fd, &fds));
- if (sendto(eth_data.s_fd,
- frame,
- frame_len,
- 0,
+ if (sendto(eth_data.s_fd, frame, frame_len, 0,
(struct sockaddr *) &eth_data.device,
sizeof(eth_data.device)) <= 0) {
log_dbg("Failed to send message: %s.", strerror(errno));
@@ -451,6 +733,8 @@ static int eth_ipcp_send_frame(const uint8_t * dst_addr,
}
#endif /* HAVE_NETMAP */
+ FETCH_ADD_RELAXED(&eth_data.stat.n_snd, 1);
+
return 0;
}
@@ -490,7 +774,7 @@ static int eth_ipcp_alloc(const uint8_t * dst_addr,
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);
@@ -508,6 +792,9 @@ static int eth_ipcp_alloc(const uint8_t * dst_addr,
buf, len + data->len);
free(buf);
+ if (ret == 0)
+ FETCH_ADD_RELAXED(&eth_data.stat.n_mgmt_snd, 1);
+
return ret;
}
@@ -558,11 +845,65 @@ static int eth_ipcp_alloc_resp(uint8_t * dst_addr,
return -1;
}
+ FETCH_ADD_RELAXED(&eth_data.stat.n_mgmt_snd, 1);
+
free(buf);
return 0;
}
+static int eth_ipcp_flow_update(int fd,
+ const buffer_t * data)
+{
+ struct mgmt_msg * msg;
+ struct ef * flow;
+ uint8_t * buf;
+ uint8_t r_addr[MAC_SIZE];
+ int ret;
+
+ buf = malloc(sizeof(*msg) + ETH_HEADER_TOT_SIZE + data->len);
+ if (buf == NULL)
+ return -1;
+
+ memset(buf, 0, sizeof(*msg) + ETH_HEADER_TOT_SIZE + data->len);
+
+ msg = (struct mgmt_msg *) (buf + ETH_HEADER_TOT_SIZE);
+
+ msg->code = FLOW_IRM_UPDATE;
+
+ pthread_rwlock_rdlock(&eth_data.flows_lock);
+
+ flow = &eth_data.fd_to_ef[fd];
+#if defined(BUILD_ETH_DIX)
+ msg->seid = htons((uint16_t) fd);
+ msg->deid = htons((uint16_t) flow->r_eid);
+#elif defined(BUILD_ETH_LLC)
+ msg->ssap = flow->sap;
+ msg->dsap = (uint8_t) flow->r_sap;
+#endif
+ memcpy(r_addr, flow->r_addr, MAC_SIZE);
+
+ pthread_rwlock_unlock(&eth_data.flows_lock);
+
+ if (data->len > 0)
+ memcpy(msg + 1, data->data, data->len);
+
+ ret = eth_ipcp_send_frame(r_addr,
+#if defined(BUILD_ETH_DIX)
+ MGMT_EID,
+#elif defined(BUILD_ETH_LLC)
+ reverse_bits(MGMT_SAP),
+ reverse_bits(MGMT_SAP),
+#endif
+ buf, sizeof(*msg) + data->len);
+ free(buf);
+
+ if (ret == 0)
+ FETCH_ADD_RELAXED(&eth_data.stat.n_mgmt_snd, 1);
+
+ return ret;
+}
+
static int eth_ipcp_req(uint8_t * r_addr,
#if defined(BUILD_ETH_DIX)
uint16_t r_eid,
@@ -575,7 +916,8 @@ static int eth_ipcp_req(uint8_t * r_addr,
{
int fd;
- fd = ipcp_wait_flow_req_arr(dst, qs, IPCP_ETH_MPL, data);
+ fd = ipcp_wait_flow_req_arr(dst, qs, IPCP_ETH_MPL,
+ ETH_MAX_PACKET_SIZE, data);
if (fd < 0) {
log_err("Could not get new flow from IRMd.");
return -1;
@@ -622,7 +964,7 @@ static int eth_ipcp_alloc_reply(uint8_t * r_addr,
fd = eth_data.ef_to_fd[dsap];
#endif
if (fd < 0) {
- pthread_rwlock_unlock(& eth_data.flows_lock);
+ pthread_rwlock_unlock(&eth_data.flows_lock);
log_err("No flow found with that SAP.");
return -1; /* -EFLOWNOTFOUND */
}
@@ -647,7 +989,8 @@ static int eth_ipcp_alloc_reply(uint8_t * r_addr,
#elif defined(BUILD_ETH_LLC)
log_dbg("Flow reply, fd %d, SSAP %d, DSAP %d.", fd, ssap, dsap);
#endif
- if ((ret = ipcp_flow_alloc_reply(fd, response, mpl, data)) < 0) {
+ if ((ret = ipcp_flow_alloc_reply(fd, response, mpl,
+ ETH_MAX_PACKET_SIZE, data)) < 0) {
log_err("Failed to reply to flow allocation.");
return -1;
}
@@ -689,6 +1032,8 @@ static int eth_ipcp_name_query_req(const uint8_t * hash,
return -1;
}
+ FETCH_ADD_RELAXED(&eth_data.stat.n_mgmt_snd, 1);
+
free(buf);
}
@@ -709,6 +1054,44 @@ static int eth_ipcp_name_query_reply(const uint8_t * hash,
return 0;
}
+static int eth_ipcp_flow_update_arr(const uint8_t * buf,
+ size_t len)
+{
+ struct mgmt_msg * msg;
+ buffer_t data;
+ int fd;
+ int flow_id;
+
+ msg = (struct mgmt_msg *) buf;
+
+ data.data = (uint8_t *) buf + sizeof(*msg);
+ data.len = len - sizeof(*msg);
+
+ pthread_rwlock_rdlock(&eth_data.flows_lock);
+#if defined(BUILD_ETH_DIX)
+ fd = ntohs(msg->deid);
+#elif defined(BUILD_ETH_LLC)
+ fd = eth_data.ef_to_fd[msg->dsap];
+#endif
+ pthread_rwlock_unlock(&eth_data.flows_lock);
+
+ if (fd < 0 || fd >= SYS_MAX_FLOWS) {
+ log_err("Flow update for unknown endpoint.");
+ return -1;
+ }
+
+ flow_id = np1_flow_id(fd);
+ if (flow_id < 0)
+ return -1;
+
+ if (ipcp_flow_update_arr(flow_id, &data) < 0) {
+ log_err("Failed to relay flow update on fd %d.", fd);
+ return -1;
+ }
+
+ return 0;
+}
+
static int eth_ipcp_mgmt_frame(const uint8_t * buf,
size_t len,
uint8_t * r_addr)
@@ -718,20 +1101,24 @@ static int eth_ipcp_mgmt_frame(const uint8_t * buf,
qosspec_t qs;
buffer_t data;
+ if (len < sizeof(*msg))
+ return -1;
+
msg = (struct mgmt_msg *) buf;
switch (msg->code) {
case FLOW_REQ:
msg_len = sizeof(*msg) + ipcp_dir_hash_len();
- assert(len >= msg_len);
+ if (len < msg_len)
+ return -1;
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.service = msg->service;
qs.max_gap = ntoh32(msg->max_gap);
qs.timeout = ntoh32(msg->timeout);
@@ -752,8 +1139,6 @@ static int eth_ipcp_mgmt_frame(const uint8_t * buf,
}
break;
case FLOW_REPLY:
- assert(len >= sizeof(*msg));
-
data.data = (uint8_t *) buf + sizeof(*msg);
data.len = len - sizeof(*msg);
@@ -768,10 +1153,17 @@ static int eth_ipcp_mgmt_frame(const uint8_t * buf,
ntoh32(msg->response),
&data);
break;
+ case FLOW_IRM_UPDATE:
+ eth_ipcp_flow_update_arr(buf, len);
+ break;
case NAME_QUERY_REQ:
+ if (len < sizeof(*msg) + ipcp_dir_hash_len())
+ return -1;
eth_ipcp_name_query_req(buf + sizeof(*msg), r_addr);
break;
case NAME_QUERY_REPLY:
+ if (len < sizeof(*msg) + ipcp_dir_hash_len())
+ return -1;
eth_ipcp_name_query_reply(buf + sizeof(*msg), r_addr);
break;
default:
@@ -844,6 +1236,12 @@ static void * eth_ipcp_packet_reader(void * o)
fd_set fds;
int frame_len;
#endif
+#if defined(HAVE_RAW_SOCKETS)
+ struct sockaddr_ll src;
+ socklen_t slen;
+#endif
+ size_t eth_len;
+ uint8_t hcs;
struct eth_frame * e_frame;
struct mgmt_frame * frame;
@@ -881,24 +1279,58 @@ static void * eth_ipcp_packet_reader(void * o)
if (select(eth_data.s_fd + 1, &fds, NULL, NULL, NULL) < 0)
continue;
assert(FD_ISSET(eth_data.s_fd, &fds));
- if (ipcp_spb_reserve(&spb, ETH_MTU))
+ if (ipcp_spb_reserve(&spb, ETH_MTU)) {
+ FETCH_ADD_RELAXED(&eth_data.stat.n_buf_f, 1);
continue;
- buf = ssm_pk_buff_head_alloc(spb, ETH_HEADER_TOT_SIZE);
+ }
+ buf = ssm_pk_buff_push(spb, ETH_HEADER_TOT_SIZE);
if (buf == NULL) {
log_dbg("Failed to allocate header.");
ipcp_spb_release(spb);
+ FETCH_ADD_RELAXED(&eth_data.stat.n_buf_f, 1);
continue;
}
- frame_len = recv(eth_data.s_fd, buf,
- ETH_MTU + ETH_HEADER_TOT_SIZE, 0);
+ slen = sizeof(src);
+ /* MSG_DONTWAIT: RD_THR>1 race-loser bails with EAGAIN. */
+ frame_len = recvfrom(eth_data.s_fd, buf,
+ ETH_MTU + ETH_HEADER_TOT_SIZE,
+ MSG_DONTWAIT,
+ (struct sockaddr *) &src, &slen);
#endif
- if (frame_len <= 0) {
- log_dbg("Failed to receive frame.");
+ if (frame_len == 0) {
+ ipcp_spb_release(spb);
+ continue; /* Spurious */
+ }
+
+ if (frame_len < 0) {
ipcp_spb_release(spb);
+
+ if (errno == EAGAIN || errno == EWOULDBLOCK)
+ continue;
+
+ log_dbg("Failed to rcv frame: %s.", strerror(errno));
+ FETCH_ADD_RELAXED(&eth_data.stat.n_rcv_f, 1);
continue;
}
#endif
+#if defined(HAVE_NETMAP)
+ eth_len = hdr.len;
+#elif defined(HAVE_BPF)
+ eth_len = ((struct bpf_hdr *) buf)->bh_caplen;
+#else
+ eth_len = (size_t) frame_len;
+#endif
+ /* Defense in depth: reject before parsing dereferences. */
+ if (eth_len < ETH_HEADER_TOT_SIZE)
+ goto fail_frame;
+
+#if defined(HAVE_RAW_SOCKETS)
+ /* Drop our own egress. */
+ if (src.sll_pkttype == PACKET_OUTGOING)
+ goto fail_frame;
+#endif
+
#if defined(HAVE_BPF) && !defined(HAVE_NETMAP)
e_frame = (struct eth_frame *)
(buf + ((struct bpf_hdr *) buf)->bh_hdrlen);
@@ -916,6 +1348,8 @@ static void * eth_ipcp_packet_reader(void * o)
e_frame->dst_hwaddr,
MAC_SIZE) &&
memcmp(br_addr, e_frame->dst_hwaddr, MAC_SIZE)) {
+ FETCH_ADD_RELAXED(&eth_data.stat.n_bad_id, 1);
+ goto fail_frame;
}
#endif
length = ntohs(e_frame->length);
@@ -923,17 +1357,41 @@ static void * eth_ipcp_packet_reader(void * o)
if (e_frame->ethertype != eth_data.ethertype)
goto fail_frame;
+ if (length > ETH_MTU)
+ goto fail_frame;
+
deid = ntohs(e_frame->eid);
- if (deid == MGMT_EID) {
#elif defined (BUILD_ETH_LLC)
if (length > 0x05FF) /* DIX */
goto fail_frame;
+ if (length < LLC_HEADER_SIZE || length > ETH_MTU)
+ goto fail_frame;
+
length -= LLC_HEADER_SIZE;
dsap = reverse_bits(e_frame->dsap);
ssap = reverse_bits(e_frame->ssap);
+#endif
+
+ if (eth_len < ETH_HEADER_TOT_SIZE + (size_t) length)
+ goto fail_frame;
+
+#if defined(BUILD_ETH_DIX)
+ mem_hash(HASH_CRC8, &hcs,
+ (uint8_t *) &e_frame->eid,
+ DIX_EID_SIZE + DIX_LENGTH_SIZE);
+#elif defined(BUILD_ETH_LLC)
+ mem_hash(HASH_CRC8, &hcs,
+ (uint8_t *) &e_frame->dsap,
+ LLC_FIELDS_SIZE);
+#endif
+ if (hcs != e_frame->hcs)
+ goto fail_frame;
+#if defined(BUILD_ETH_DIX)
+ if (deid == MGMT_EID) {
+#elif defined (BUILD_ETH_LLC)
if (ssap == MGMT_SAP && dsap == MGMT_SAP) {
#endif
ipcp_spb_release(spb); /* No need for the N+1 buffer. */
@@ -941,13 +1399,13 @@ static void * eth_ipcp_packet_reader(void * o)
if (length > MGMT_FRAME_SIZE) {
log_warn("Management frame size %u exceeds %u.",
length, MGMT_FRAME_SIZE);
- goto fail_frame;
+ continue;
}
frame = malloc(sizeof(*frame));
if (frame == NULL) {
log_err("Failed to allocate frame.");
- goto fail_frame;
+ continue;
}
memcpy(frame->buf, &e_frame->payload, length);
@@ -958,6 +1416,8 @@ static void * eth_ipcp_packet_reader(void * o)
list_add(&frame->next, &eth_data.mgmt_frames);
pthread_cond_signal(&eth_data.mgmt_cond);
pthread_mutex_unlock(&eth_data.mgmt_lock);
+ FETCH_ADD_RELAXED(&eth_data.stat.n_rcv, 1);
+ FETCH_ADD_RELAXED(&eth_data.stat.n_mgmt_rcv, 1);
} else {
pthread_rwlock_rdlock(&eth_data.flows_lock);
@@ -968,6 +1428,7 @@ static void * eth_ipcp_packet_reader(void * o)
#endif
if (fd < 0) {
pthread_rwlock_unlock(&eth_data.flows_lock);
+ FETCH_ADD_RELAXED(&eth_data.stat.n_bad_id, 1);
goto fail_frame;
}
@@ -976,13 +1437,18 @@ static void * eth_ipcp_packet_reader(void * o)
|| memcmp(eth_data.fd_to_ef[fd].r_addr,
e_frame->src_hwaddr, MAC_SIZE)) {
pthread_rwlock_unlock(&eth_data.flows_lock);
+ FETCH_ADD_RELAXED(&eth_data.stat.n_bad_id, 1);
goto fail_frame;
}
#endif
+ FETCH_ADD_RELAXED(&eth_data.fd_to_ef[fd].stat.p_rcv, 1);
+ FETCH_ADD_RELAXED(&eth_data.fd_to_ef[fd].stat.b_rcv,
+ length);
+ FETCH_ADD_RELAXED(&eth_data.stat.n_rcv, 1);
pthread_rwlock_unlock(&eth_data.flows_lock);
#ifndef HAVE_NETMAP
- ssm_pk_buff_head_release(spb, ETH_HEADER_TOT_SIZE);
+ ssm_pk_buff_pop(spb, ETH_HEADER_TOT_SIZE);
ssm_pk_buff_truncate(spb, length);
#else
if (ipcp_spb_reserve(&spb, length))
@@ -991,8 +1457,13 @@ static void * eth_ipcp_packet_reader(void * o)
buf = ssm_pk_buff_head(spb);
memcpy(buf, &e_frame->payload, length);
#endif
- if (np1_flow_write(fd, spb, NP1_GET_POOL(fd)) < 0)
+ if (np1_flow_write(fd, spb, NP1_GET_POOL(fd)) < 0) {
ipcp_spb_release(spb);
+ FETCH_ADD_RELAXED(
+ &eth_data.fd_to_ef[fd].stat.p_dlv_f,
+ 1);
+ FETCH_ADD_RELAXED(&eth_data.stat.n_dlv_f, 1);
+ }
continue;
fail_frame:
@@ -1048,10 +1519,11 @@ static void * eth_ipcp_packet_writer(void * o)
len = ssm_pk_buff_len(spb);
- if (ssm_pk_buff_head_alloc(spb, ETH_HEADER_TOT_SIZE)
+ if (ssm_pk_buff_push(spb, ETH_HEADER_TOT_SIZE)
== NULL) {
log_dbg("Failed to allocate header.");
ipcp_spb_release(spb);
+ FETCH_ADD_RELAXED(&eth_data.stat.n_buf_f, 1);
continue;
}
@@ -1075,8 +1547,20 @@ static void * eth_ipcp_packet_writer(void * o)
dsap, ssap,
#endif
ssm_pk_buff_head(spb),
- len))
+ len)) {
log_dbg("Failed to send frame.");
+ FETCH_ADD_RELAXED(
+ &eth_data.fd_to_ef[fd].stat.p_snd_f,
+ 1);
+ FETCH_ADD_RELAXED(&eth_data.stat.n_snd_f, 1);
+ } else {
+ FETCH_ADD_RELAXED(
+ &eth_data.fd_to_ef[fd].stat.p_snd,
+ 1);
+ FETCH_ADD_RELAXED(
+ &eth_data.fd_to_ef[fd].stat.b_snd,
+ len);
+ }
ipcp_spb_release(spb);
}
}
@@ -1424,12 +1908,14 @@ static int eth_init_bpf(struct ifreq * ifr)
return -1;
}
#elif defined(HAVE_RAW_SOCKETS)
+#define SOCKOPT()
static int eth_init_raw_socket(struct ifreq * ifr)
{
int idx;
- int flags;
+ int sndbuf;
+ int rcvbuf;
#if defined(IPCP_ETH_QDISC_BYPASS)
- int qdisc_bypass = 1;
+ int qdisc_bypass = 1;
#endif /* ENABLE_QDISC_BYPASS */
idx = if_nametoindex(ifr->ifr_name);
@@ -1437,6 +1923,7 @@ static int eth_init_raw_socket(struct ifreq * ifr)
log_err("Failed to retrieve interface index.");
return -1;
}
+
memset(&(eth_data.device), 0, sizeof(eth_data.device));
eth_data.device.sll_ifindex = idx;
eth_data.device.sll_family = AF_PACKET;
@@ -1453,17 +1940,6 @@ static int eth_init_raw_socket(struct ifreq * ifr)
goto fail_socket;
}
- flags = fcntl(eth_data.s_fd, F_GETFL, 0);
- if (flags < 0) {
- log_err("Failed to get flags.");
- goto fail_device;
- }
-
- if (fcntl(eth_data.s_fd, F_SETFL, flags | O_NONBLOCK)) {
- log_err("Failed to set socket non-blocking.");
- goto fail_device;
- }
-
#if defined(IPCP_ETH_QDISC_BYPASS)
if (setsockopt(eth_data.s_fd, SOL_PACKET, PACKET_QDISC_BYPASS,
&qdisc_bypass, sizeof(qdisc_bypass))) {
@@ -1471,6 +1947,25 @@ static int eth_init_raw_socket(struct ifreq * ifr)
}
#endif
+ sndbuf = IPCP_ETH_SNDBUF;
+ if (sndbuf > 0) {
+ /* Never below one frame or sendto() cannot fit an SDU. */
+ sndbuf = MAX(sndbuf, (int) (ETH_HEADER_TOT_SIZE + ETH_MTU));
+ if (setsockopt(eth_data.s_fd, SOL_SOCKET, SO_SNDBUF,
+ &sndbuf, sizeof(sndbuf)))
+ log_info("Failed to set SO_SNDBUF to %d.", sndbuf);
+ }
+
+ rcvbuf = IPCP_ETH_RCVBUF;
+ if (rcvbuf > 0) {
+ /* SO_RCVBUFFORCE bypasses rmem_max; SO_RCVBUF is fallback. */
+ if (setsockopt(eth_data.s_fd, SOL_SOCKET, SO_RCVBUFFORCE,
+ &rcvbuf, sizeof(rcvbuf))
+ && setsockopt(eth_data.s_fd, SOL_SOCKET, SO_RCVBUF,
+ &rcvbuf, sizeof(rcvbuf)))
+ log_info("Failed to set SO_RCVBUF to %d.", rcvbuf);
+ }
+
if (bind(eth_data.s_fd, (struct sockaddr *) &eth_data.device,
sizeof(eth_data.device)) < 0) {
log_err("Failed to bind socket to interface.");
@@ -1543,6 +2038,12 @@ static int eth_ipcp_bootstrap(struct ipcp_config * conf)
return -1;
}
#endif /* HAVE_NETMAP */
+#ifdef IPCP_ETH_FLOW_STATS
+ if (rib_reg(ETH_RIB_PATH, &eth_r_ops)) {
+ log_err("Failed to register RIB.");
+ goto fail_rib_reg;
+ }
+#endif
#if defined(__linux__)
if (pthread_create(&eth_data.if_monitor, NULL,
eth_ipcp_if_monitor, NULL)) {
@@ -1606,6 +2107,10 @@ static int eth_ipcp_bootstrap(struct ipcp_config * conf)
#if defined(__linux__)
fail_monitor:
#endif
+#ifdef IPCP_ETH_FLOW_STATS
+ rib_unreg(ETH_RIB_PATH);
+ fail_rib_reg:
+#endif
#if defined(HAVE_NETMAP)
nm_close(eth_data.nmd);
#elif defined(HAVE_BPF)
@@ -1637,12 +2142,14 @@ static int eth_ipcp_unreg(const uint8_t * hash)
static int eth_ipcp_query(const uint8_t * hash)
{
uint8_t r_addr[MAC_SIZE];
- struct timespec timeout = TIMESPEC_INIT_MS(NAME_QUERY_TIMEO);
+ struct timespec timeout;
struct dir_query * query;
int ret;
+ int attempt;
uint8_t * buf;
struct mgmt_msg * msg;
size_t len;
+ long per_ms;
if (shim_data_dir_has(eth_data.shim_data, hash))
return 0;
@@ -1662,32 +2169,46 @@ static int eth_ipcp_query(const uint8_t * hash)
memset(r_addr, 0xff, MAC_SIZE);
- query = shim_data_dir_query_create(eth_data.shim_data, hash);
- if (query == NULL) {
- free(buf);
- return -1;
- }
+ per_ms = NAME_QUERY_TIMEO / (NAME_QUERY_RETRIES + 1);
- if (eth_ipcp_send_frame(r_addr,
+ ret = -1;
+ for (attempt = 0; attempt <= NAME_QUERY_RETRIES; ++attempt) {
+ query = shim_data_dir_query_create(eth_data.shim_data, hash);
+ if (query == NULL) {
+ ret = -1;
+ break;
+ }
+
+ if (eth_ipcp_send_frame(r_addr,
#if defined(BUILD_ETH_DIX)
- MGMT_EID,
+ MGMT_EID,
#elif defined(BUILD_ETH_LLC)
- reverse_bits(MGMT_SAP),
- reverse_bits(MGMT_SAP),
+ reverse_bits(MGMT_SAP),
+ reverse_bits(MGMT_SAP),
#endif
- buf, len)) {
- log_err("Failed to send management frame.");
+ buf, len)) {
+ log_err("Failed to send management frame.");
+ shim_data_dir_query_destroy(eth_data.shim_data,
+ query);
+ ret = -1;
+ break;
+ }
+
+ FETCH_ADD_RELAXED(&eth_data.stat.n_mgmt_snd, 1);
+
+ timeout.tv_sec = per_ms / 1000;
+ timeout.tv_nsec = (per_ms % 1000) * 1000000L;
+
+ ret = shim_data_dir_query_wait(query, &timeout);
+
shim_data_dir_query_destroy(eth_data.shim_data, query);
- free(buf);
- return -1;
+
+ if (ret != -ETIMEDOUT)
+ break;
}
free(buf);
- ret = shim_data_dir_query_wait(query, &timeout);
-
- shim_data_dir_query_destroy(eth_data.shim_data, query);
-
return ret;
}
@@ -1748,6 +2269,14 @@ static int eth_ipcp_flow_alloc(int fd,
}
fset_add(eth_data.np1_flows, fd);
+#ifdef IPCP_ETH_FLOW_STATS
+ pthread_rwlock_wrlock(&eth_data.flows_lock);
+ memset(&eth_data.fd_to_ef[fd].stat, 0,
+ sizeof(eth_data.fd_to_ef[fd].stat));
+ eth_data.fd_to_ef[fd].stat.stamp = time(NULL);
+ FETCH_ADD_RELAXED(&eth_data.stat.n_flows, 1);
+ pthread_rwlock_unlock(&eth_data.flows_lock);
+#endif
#if defined(BUILD_ETH_LLC)
log_dbg("Assigned SAP %d for fd %d.", ssap, fd);
#endif
@@ -1808,6 +2337,14 @@ static int eth_ipcp_flow_alloc_resp(int fd,
}
fset_add(eth_data.np1_flows, fd);
+#ifdef IPCP_ETH_FLOW_STATS
+ pthread_rwlock_wrlock(&eth_data.flows_lock);
+ memset(&eth_data.fd_to_ef[fd].stat, 0,
+ sizeof(eth_data.fd_to_ef[fd].stat));
+ eth_data.fd_to_ef[fd].stat.stamp = time(NULL);
+ FETCH_ADD_RELAXED(&eth_data.stat.n_flows, 1);
+ pthread_rwlock_unlock(&eth_data.flows_lock);
+#endif
#if defined(BUILD_ETH_LLC)
log_dbg("Assigned SAP %d for fd %d.", ssap, fd);
#endif
@@ -1836,6 +2373,12 @@ static int eth_ipcp_flow_dealloc(int fd)
#endif
memset(&eth_data.fd_to_ef[fd].r_addr, 0, MAC_SIZE);
+#ifdef IPCP_ETH_FLOW_STATS
+ memset(&eth_data.fd_to_ef[fd].stat, 0,
+ sizeof(eth_data.fd_to_ef[fd].stat));
+ FETCH_SUB_RELAXED(&eth_data.stat.n_flows, 1);
+#endif
+
pthread_rwlock_unlock(&eth_data.flows_lock);
ipcp_flow_dealloc(fd);
@@ -1854,7 +2397,8 @@ static struct ipcp_ops eth_ops = {
.ipcp_flow_alloc = eth_ipcp_flow_alloc,
.ipcp_flow_join = NULL,
.ipcp_flow_alloc_resp = eth_ipcp_flow_alloc_resp,
- .ipcp_flow_dealloc = eth_ipcp_flow_dealloc
+ .ipcp_flow_dealloc = eth_ipcp_flow_dealloc,
+ .ipcp_flow_update = eth_ipcp_flow_update
};
int main(int argc,
@@ -1902,6 +2446,9 @@ int main(int argc,
#ifdef __linux__
pthread_join(eth_data.if_monitor, NULL);
#endif
+#ifdef IPCP_ETH_FLOW_STATS
+ rib_unreg(ETH_RIB_PATH);
+#endif
}
ipcp_stop();
diff --git a/src/ipcpd/ipcp.c b/src/ipcpd/ipcp.c
index 5ad2401f..dcee4b9c 100644
--- a/src/ipcpd/ipcp.c
+++ b/src/ipcpd/ipcp.c
@@ -363,6 +363,7 @@ static void * acceptloop(void * o)
int ipcp_wait_flow_req_arr(const uint8_t * dst,
qosspec_t qs,
time_t mpl,
+ uint32_t mtu,
const buffer_t * data)
{
struct timespec ts = TIMESPEC_INIT_MS(ALLOC_TIMEOUT);
@@ -392,7 +393,7 @@ int ipcp_wait_flow_req_arr(const uint8_t * dst,
assert(ipcpd.alloc_id == -1);
- fd = ipcp_flow_req_arr(&hash, qs, mpl, data);
+ fd = ipcp_flow_req_arr(&hash, qs, mpl, mtu, data);
if (fd < 0) {
pthread_mutex_unlock(&ipcpd.alloc_lock);
log_err("Failed to get fd for flow.");
@@ -819,6 +820,33 @@ static void do_flow_dealloc(int flow_id,
log_info("Finished deallocating flow %d.", flow_id);
}
+static void do_flow_update(int flow_id,
+ const buffer_t * data,
+ ipcp_msg_t * ret_msg)
+{
+ int fd;
+
+ if (ipcpd.ops->ipcp_flow_update == NULL) {
+ log_err("Failed to update flow: operation unsupported.");
+ ret_msg->result = -ENOTSUP;
+ return;
+ }
+
+ if (ipcp_get_state() != IPCP_OPERATIONAL) {
+ ret_msg->result = -EIPCPSTATE;
+ return;
+ }
+
+ fd = np1_flow_fd(flow_id);
+ if (fd < 0) {
+ log_warn("Flow update for unknown flow_id %d.", flow_id);
+ ret_msg->result = -1;
+ return;
+ }
+
+ ret_msg->result = ipcpd.ops->ipcp_flow_update(fd, data);
+}
+
static void * mainloop(void * o)
{
int sfd;
@@ -917,6 +945,13 @@ static void * mainloop(void * o)
case IPCP_MSG_CODE__IPCP_FLOW_DEALLOC:
do_flow_dealloc(msg->flow_id, msg->timeo_sec, &ret_msg);
break;
+ case IPCP_MSG_CODE__IPCP_FLOW_UPDATE:
+ assert(msg->pk.len > 0 ? msg->pk.data != NULL
+ : msg->pk.data == NULL);
+ data.len = msg->pk.len;
+ data.data = msg->pk.data;
+ do_flow_update(msg->flow_id, &data, &ret_msg);
+ break;
default:
ret_msg.result = -1;
log_err("Unknown message code: %d.", msg->code);
diff --git a/src/ipcpd/ipcp.h b/src/ipcpd/ipcp.h
index 26a780a3..210157ec 100644
--- a/src/ipcpd/ipcp.h
+++ b/src/ipcpd/ipcp.h
@@ -68,6 +68,9 @@ struct ipcp_ops {
const buffer_t * data);
int (* ipcp_flow_dealloc)(int fd);
+
+ int (* ipcp_flow_update)(int fd,
+ const buffer_t * data);
};
int ipcp_init(int argc,
@@ -98,6 +101,7 @@ enum ipcp_state ipcp_get_state(void);
int ipcp_wait_flow_req_arr(const uint8_t * dst,
qosspec_t qs,
time_t mpl,
+ uint32_t mtu,
const buffer_t * data);
int ipcp_wait_flow_resp(const int fd);
diff --git a/src/ipcpd/local/main.c b/src/ipcpd/local/main.c
index 377a7df3..c0aeb51e 100644
--- a/src/ipcpd/local/main.c
+++ b/src/ipcpd/local/main.c
@@ -38,6 +38,7 @@
#include <ouroboros/ipcp.h>
#include <ouroboros/ipcp-dev.h>
#include <ouroboros/local-dev.h>
+#include <ouroboros/np1_flow.h>
#include "ipcp.h"
#include "np1.h"
@@ -203,7 +204,8 @@ static int local_ipcp_flow_alloc(int fd,
HASH_VAL32(dst), fd);
assert(dst);
- out_fd = ipcp_wait_flow_req_arr(dst, qs, IPCP_LOCAL_MPL, data);
+ out_fd = ipcp_wait_flow_req_arr(dst, qs, IPCP_LOCAL_MPL,
+ IPCP_LOCAL_MTU, data);
if (out_fd < 0) {
log_dbg("Flow allocation failed: %d", out_fd);
return -1;
@@ -236,15 +238,6 @@ static int local_ipcp_flow_alloc_resp(int fd,
return -1;
}
- if (response < 0) {
- pthread_rwlock_wrlock(&local_data.lock);
- if (local_data.in_out[fd] != -1)
- local_data.in_out[local_data.in_out[fd]] = fd;
- local_data.in_out[fd] = -1;
- pthread_rwlock_unlock(&local_data.lock);
- return 0;
- }
-
pthread_rwlock_rdlock(&local_data.lock);
out_fd = local_data.in_out[fd];
@@ -263,9 +256,17 @@ static int local_ipcp_flow_alloc_resp(int fd,
return -1;
}
+ if (response < 0) {
+ ipcp_flow_alloc_reply(out_fd, response, mpl,
+ IPCP_LOCAL_MTU, data);
+ log_info("Flow allocation rejected, fds (%d, %d).", out_fd, fd);
+ return 0;
+ }
+
fset_add(local_data.flows, fd);
- if (ipcp_flow_alloc_reply(out_fd, response, mpl, data) < 0) {
+ if (ipcp_flow_alloc_reply(out_fd, response, mpl,
+ IPCP_LOCAL_MTU, data) < 0) {
log_err("Failed to reply to allocation");
fset_del(local_data.flows, fd);
return -1;
@@ -297,6 +298,38 @@ static int local_ipcp_flow_dealloc(int fd)
return 0;
}
+/* Loopback relay: deliver the update back to the peer end (same IRMd). */
+static int local_ipcp_flow_update(int fd,
+ const buffer_t * data)
+{
+ int out_fd;
+ int out_flow_id;
+
+ pthread_rwlock_rdlock(&local_data.lock);
+
+ out_fd = local_data.in_out[fd];
+
+ pthread_rwlock_unlock(&local_data.lock);
+
+ if (out_fd == -1) {
+ log_err("Flow update on fd %d with no peer.", fd);
+ return -1;
+ }
+
+ out_flow_id = np1_flow_id(out_fd);
+ if (out_flow_id < 0) {
+ log_err("No flow_id for peer fd %d.", out_fd);
+ return -1;
+ }
+
+ if (ipcp_flow_update_arr(out_flow_id, data) < 0) {
+ log_err("Failed to relay flow update to fd %d.", out_fd);
+ return -1;
+ }
+
+ return 0;
+}
+
static struct ipcp_ops local_ops = {
.ipcp_bootstrap = local_ipcp_bootstrap,
.ipcp_enroll = NULL,
@@ -308,7 +341,8 @@ static struct ipcp_ops local_ops = {
.ipcp_flow_alloc = local_ipcp_flow_alloc,
.ipcp_flow_join = NULL,
.ipcp_flow_alloc_resp = local_ipcp_flow_alloc_resp,
- .ipcp_flow_dealloc = local_ipcp_flow_dealloc
+ .ipcp_flow_dealloc = local_ipcp_flow_dealloc,
+ .ipcp_flow_update = local_ipcp_flow_update
};
int main(int argc,
diff --git a/src/ipcpd/udp/udp.c b/src/ipcpd/udp/udp.c
index 452bbc1a..db57e2f4 100644
--- a/src/ipcpd/udp/udp.c
+++ b/src/ipcpd/udp/udp.c
@@ -28,6 +28,8 @@
#include <ouroboros/list.h>
#include <ouroboros/utils.h>
#include <ouroboros/dev.h>
+#include <ouroboros/ipcp-dev.h>
+#include <ouroboros/np1_flow.h>
#include <ouroboros/fqueue.h>
#include <ouroboros/errno.h>
#include <ouroboros/logs.h>
@@ -47,9 +49,14 @@
#include <stdlib.h>
#include <sys/wait.h>
#include <fcntl.h>
+#include <unistd.h>
+#if defined(__linux__)
+#include <netinet/ip.h>
+#endif
#define FLOW_REQ 1
#define FLOW_REPLY 2
+#define FLOW_IRM_UPDATE 3
#define OUR_HEADER_LEN sizeof(uint32_t) /* adds eid */
@@ -87,7 +94,7 @@ struct mgmt_msg {
uint8_t code;
/* QoS parameters from spec */
uint8_t availability;
- uint8_t in_order;
+ uint8_t service;
} __attribute__((packed));
struct mgmt_frame {
@@ -130,6 +137,53 @@ static const char * __inet_ntop(const struct __ADDR * addr,
return inet_ntop(__AF, addr, buf, __ADDRSTRLEN);
}
+#if defined(BUILD_IPCP_UDP4)
+#define UDP_MTU_FALLBACK IPCP_UDP4_MTU
+#define UDP_IP_OVERHEAD 28U /* IPv4 + UDP */
+#else
+#define UDP_MTU_FALLBACK IPCP_UDP6_MTU
+#define UDP_IP_OVERHEAD 48U /* IPv6 + UDP */
+#endif
+
+static uint32_t udp_query_mtu(const struct __SOCKADDR * saddr)
+{
+#if defined(__linux__) && (defined(IP_MTU) || defined(IPV6_MTU))
+ int sock;
+ int mtu = 0;
+ socklen_t len = sizeof(mtu);
+
+ sock = socket(__AF, SOCK_DGRAM, IPPROTO_UDP);
+ if (sock < 0)
+ return UDP_MTU_FALLBACK;
+
+ if (connect(sock, (const struct sockaddr *) saddr,
+ sizeof(*saddr)) < 0)
+ goto fallback;
+
+#if defined(BUILD_IPCP_UDP4) && defined(IP_MTU)
+ if (getsockopt(sock, IPPROTO_IP, IP_MTU, &mtu, &len) < 0)
+ goto fallback;
+#elif defined(BUILD_IPCP_UDP6) && defined(IPV6_MTU)
+ if (getsockopt(sock, IPPROTO_IPV6, IPV6_MTU, &mtu, &len) < 0)
+ goto fallback;
+#else
+ goto fallback;
+#endif
+ close(sock);
+
+ if (mtu <= (int) UDP_IP_OVERHEAD)
+ return UDP_MTU_FALLBACK;
+
+ return (uint32_t) mtu - UDP_IP_OVERHEAD;
+
+ fallback:
+ close(sock);
+#else
+ (void) saddr;
+#endif
+ return UDP_MTU_FALLBACK;
+}
+
static int udp_data_init(void)
{
int i;
@@ -220,7 +274,7 @@ static int udp_ipcp_port_alloc(const struct __SOCKADDR * r_saddr,
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);
@@ -277,6 +331,48 @@ static int udp_ipcp_port_alloc_resp(const struct __SOCKADDR * r_saddr,
return 0;
}
+static int udp_ipcp_flow_update(int fd,
+ const buffer_t * data)
+{
+ struct mgmt_msg * msg;
+ struct __SOCKADDR r_saddr;
+ uint32_t d_eid;
+
+ msg = malloc(sizeof(*msg) + data->len);
+ if (msg == NULL)
+ return -1;
+
+ memset(msg, 0, sizeof(*msg) + data->len);
+
+ pthread_rwlock_rdlock(&udp_data.flows_lock);
+
+ r_saddr = udp_data.fd_to_uf[fd].r_saddr;
+ d_eid = (uint32_t) udp_data.fd_to_uf[fd].d_eid;
+
+ pthread_rwlock_unlock(&udp_data.flows_lock);
+
+ msg->eid = hton32(MGMT_EID);
+ msg->code = FLOW_IRM_UPDATE;
+ msg->s_eid = hton32(d_eid);
+ msg->d_eid = hton32((uint32_t) fd);
+
+ if (data->len > 0)
+ memcpy(msg + 1, data->data, data->len);
+
+ if (sendto(udp_data.s_fd, msg, sizeof(*msg) + data->len,
+ SENDTO_FLAGS,
+ (const struct sockaddr *) &r_saddr,
+ sizeof(r_saddr)) < 0) {
+ log_err("Failed to send flow update: %s.", strerror(errno));
+ free(msg);
+ return -1;
+ }
+
+ free(msg);
+
+ return 0;
+}
+
static int udp_ipcp_port_req(struct __SOCKADDR * c_saddr,
int d_eid,
const uint8_t * dst,
@@ -285,7 +381,8 @@ static int udp_ipcp_port_req(struct __SOCKADDR * c_saddr,
{
int fd;
- fd = ipcp_wait_flow_req_arr(dst, qs, IPCP_UDP_MPL, data);
+ fd = ipcp_wait_flow_req_arr(dst, qs, IPCP_UDP_MPL,
+ udp_query_mtu(c_saddr), data);
if (fd < 0) {
log_err("Could not get new flow from IRMd.");
return -1;
@@ -332,7 +429,8 @@ static int udp_ipcp_port_alloc_reply(const struct __SOCKADDR * saddr,
pthread_rwlock_unlock(&udp_data.flows_lock);
- if (ipcp_flow_alloc_reply(s_eid, response, mpl, data) < 0) {
+ if (ipcp_flow_alloc_reply(s_eid, response, mpl,
+ udp_query_mtu(saddr), data) < 0) {
log_err("Failed to reply to flow allocation.");
return -1;
}
@@ -343,6 +441,37 @@ static int udp_ipcp_port_alloc_reply(const struct __SOCKADDR * saddr,
return 0;
}
+static int udp_ipcp_flow_update_arr(const uint8_t * buf,
+ size_t len)
+{
+ struct mgmt_msg * msg;
+ buffer_t data;
+ int fd;
+ int flow_id;
+
+ msg = (struct mgmt_msg *) buf;
+
+ fd = (int) ntoh32(msg->s_eid);
+ if (fd < 0 || fd >= SYS_MAX_FLOWS) {
+ log_err("Flow update for invalid eid %d.", fd);
+ return -1;
+ }
+
+ data.len = len - sizeof(*msg);
+ data.data = (uint8_t *) buf + sizeof(*msg);
+
+ flow_id = np1_flow_id(fd);
+ if (flow_id < 0)
+ return -1;
+
+ if (ipcp_flow_update_arr(flow_id, &data) < 0) {
+ log_err("Failed to relay flow update on fd %d.", fd);
+ return -1;
+ }
+
+ return 0;
+}
+
static int udp_ipcp_mgmt_frame(struct __SOCKADDR c_saddr,
const uint8_t * buf,
size_t len)
@@ -352,13 +481,18 @@ static int udp_ipcp_mgmt_frame(struct __SOCKADDR c_saddr,
qosspec_t qs;
buffer_t data;
+ /* Defence against malformed/corrupted wire input. */
+ if (len < sizeof(*msg))
+ return -1;
+
msg = (struct mgmt_msg *) buf;
switch (msg->code) {
case FLOW_REQ:
msg_len = sizeof(*msg) + ipcp_dir_hash_len();
- assert(len >= msg_len);
+ if (len < msg_len)
+ return -1;
data.len = len - msg_len;
data.data = (uint8_t *) buf + msg_len;
@@ -369,7 +503,7 @@ static int udp_ipcp_mgmt_frame(struct __SOCKADDR c_saddr,
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);
@@ -377,8 +511,6 @@ static int udp_ipcp_mgmt_frame(struct __SOCKADDR c_saddr,
(uint8_t *) (msg + 1), qs,
&data);
case FLOW_REPLY:
- assert(len >= sizeof(*msg));
-
data.len = len - sizeof(*msg);
data.data = (uint8_t *) buf + sizeof(*msg);
@@ -387,6 +519,8 @@ static int udp_ipcp_mgmt_frame(struct __SOCKADDR c_saddr,
ntoh32(msg->d_eid),
ntoh32(msg->response),
&data);
+ case FLOW_IRM_UPDATE:
+ return udp_ipcp_flow_update_arr(buf, len);
default:
log_err("Unknown message received %d.", msg->code);
return -1;
@@ -549,7 +683,7 @@ static void * udp_ipcp_packet_writer(void * o)
continue;
}
- buf = ssm_pk_buff_head_alloc(spb, OUR_HEADER_LEN);
+ buf = ssm_pk_buff_push(spb, OUR_HEADER_LEN);
if (buf == NULL) {
log_dbg("Failed to allocate header.");
ipcp_spb_release(spb);
@@ -1140,7 +1274,8 @@ static struct ipcp_ops udp_ops = {
.ipcp_flow_alloc = udp_ipcp_flow_alloc,
.ipcp_flow_join = NULL,
.ipcp_flow_alloc_resp = udp_ipcp_flow_alloc_resp,
- .ipcp_flow_dealloc = udp_ipcp_flow_dealloc
+ .ipcp_flow_dealloc = udp_ipcp_flow_dealloc,
+ .ipcp_flow_update = udp_ipcp_flow_update
};
int main(int argc,
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 09:24:31 +0000