summaryrefslogtreecommitdiff
path: root/src/lib/poa
diff options
context:
space:
mode:
Diffstat (limited to 'src/lib/poa')
-rw-r--r--src/lib/poa/addr.c142
-rw-r--r--src/lib/poa/eth.c1987
-rw-r--r--src/lib/poa/poa.c2515
-rw-r--r--src/lib/poa/poa.h364
-rw-r--r--src/lib/poa/udp.c633
5 files changed, 5641 insertions, 0 deletions
diff --git a/src/lib/poa/addr.c b/src/lib/poa/addr.c
new file mode 100644
index 00000000..d8811244
--- /dev/null
+++ b/src/lib/poa/addr.c
@@ -0,0 +1,142 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2026
+ *
+ * Points of attachment (PoA) - addresses and management messages
+ *
+ * Dimitri Staessens <dimitri@ouroboros.rocks>
+ * Sander Vrijders <sander@ouroboros.rocks>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public License
+ * version 2.1 as published by the Free Software Foundation.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., http://www.fsf.org/about/contact/.
+ */
+
+#if defined(__linux__) || defined(__CYGWIN__)
+#ifndef _DEFAULT_SOURCE /* Test include source */
+#define _DEFAULT_SOURCE
+#endif
+#endif
+
+#include "config.h"
+
+#include <ouroboros/endian.h>
+#include <ouroboros/errno.h>
+
+#include "poa.h"
+
+#include <arpa/inet.h>
+#include <stdio.h>
+#include <string.h>
+
+int poa_addr_cmp(const struct poa_addr * a,
+ const struct poa_addr * b)
+{
+ if (a->type != b->type)
+ return 1;
+
+ switch (a->type) {
+ case POA_UDP4:
+ if (a->udp4.port != b->udp4.port)
+ return 1;
+ return memcmp(&a->udp4.ip_addr, &b->udp4.ip_addr,
+ sizeof(a->udp4.ip_addr));
+ case POA_UDP6:
+ if (a->udp6.port != b->udp6.port)
+ return 1;
+ return memcmp(&a->udp6.ip_addr, &b->udp6.ip_addr,
+ sizeof(a->udp6.ip_addr));
+ case POA_ETH:
+ if (a->eth.dst.ethertype != b->eth.dst.ethertype)
+ return 1;
+ return memcmp(a->eth.dst.mac, b->eth.dst.mac, POA_MAC_SIZE);
+ default:
+ return 1;
+ }
+}
+
+/* Display/RIB entry name, e.g. "udp4.<ip>.<port>". */
+int poa_addr_name(const struct poa_addr * a,
+ char * buf,
+ size_t len)
+{
+ char ip[INET6_ADDRSTRLEN];
+ int ret;
+
+ switch (a->type) {
+ case POA_UDP4:
+ if (inet_ntop(AF_INET, &a->udp4.ip_addr,
+ ip, sizeof(ip)) == NULL)
+ return -EINVAL;
+ ret = snprintf(buf, len, "udp4.%s.%u", ip, a->udp4.port);
+ break;
+ case POA_UDP6:
+ if (inet_ntop(AF_INET6, &a->udp6.ip_addr,
+ ip, sizeof(ip)) == NULL)
+ return -EINVAL;
+ ret = snprintf(buf, len, "udp6.%s.%u", ip, a->udp6.port);
+ break;
+ case POA_ETH:
+ ret = snprintf(buf, len, "eth.%s.%04X",
+ a->eth.src.dev, a->eth.src.ethertype);
+ break;
+ default:
+ return -EINVAL;
+ }
+
+ if (ret < 0 || (size_t) ret >= len)
+ return -EMSGSIZE;
+
+ return 0;
+}
+
+void poa_mgmt_msg_ser(struct poa_mgmt_msg * msg,
+ uint8_t code,
+ uint32_t s_eid,
+ uint32_t d_eid,
+ qosspec_t qs,
+ int response,
+ size_t data_len)
+{
+ memset(msg, 0, sizeof(*msg));
+
+ msg->code = code;
+ msg->s_eid = hton32(s_eid);
+ msg->d_eid = hton32(d_eid);
+ msg->response = hton32(response);
+ msg->data_len = hton16((uint16_t) data_len);
+
+ if (code != POA_FLOW_REQ)
+ return;
+
+ msg->bandwidth = hton64(qs.bandwidth);
+ msg->delay = hton32(qs.delay);
+ msg->loss = hton32(qs.loss);
+ msg->ber = hton32(qs.ber);
+ msg->max_gap = hton32(qs.max_gap);
+ msg->timeout = hton32(qs.timeout);
+ msg->availability = qs.availability;
+ msg->service = qs.service;
+}
+
+void poa_mgmt_msg_qos(const struct poa_mgmt_msg * msg,
+ qosspec_t * qs)
+{
+ qs->bandwidth = ntoh64(msg->bandwidth);
+ qs->delay = ntoh32(msg->delay);
+ qs->loss = ntoh32(msg->loss);
+ qs->ber = ntoh32(msg->ber);
+ qs->max_gap = ntoh32(msg->max_gap);
+ qs->timeout = ntoh32(msg->timeout);
+ qs->availability = msg->availability;
+ qs->service = msg->service;
+}
+
diff --git a/src/lib/poa/eth.c b/src/lib/poa/eth.c
new file mode 100644
index 00000000..5e959ef6
--- /dev/null
+++ b/src/lib/poa/eth.c
@@ -0,0 +1,1987 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2026
+ *
+ * Points of attachment (PoA) - Ethernet transport
+ *
+ * Dimitri Staessens <dimitri@ouroboros.rocks>
+ * Sander Vrijders <sander@ouroboros.rocks>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public License
+ * version 2.1 as published by the Free Software Foundation.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., http://www.fsf.org/about/contact/.
+ */
+
+#if defined(__APPLE__)
+#define _BSD_SOURCE
+#define _DARWIN_C_SOURCE
+#elif defined(__FreeBSD__)
+#define __BSD_VISIBLE 1
+#elif defined(__linux__) || defined(__CYGWIN__)
+#ifndef _DEFAULT_SOURCE
+#define _DEFAULT_SOURCE
+#endif
+#else
+#ifndef _POSIX_C_SOURCE
+#define _POSIX_C_SOURCE 200809L
+#endif
+#endif
+
+#include "config.h"
+
+#define OUROBOROS_PREFIX "poa-eth"
+
+#include <ouroboros/endian.h>
+#include <ouroboros/errno.h>
+#include <ouroboros/hash.h>
+#include <ouroboros/logs.h>
+#include <ouroboros/pthread.h>
+#include <ouroboros/sockets.h>
+#include <ouroboros/time.h>
+
+#include "poa.h"
+
+#ifdef HAVE_RAW_SOCKETS
+#include <net/if.h>
+#include <netinet/in.h>
+#include <linux/if_ether.h>
+#include <linux/if_packet.h>
+#include <linux/netlink.h>
+#include <linux/gen_stats.h>
+#include <linux/pkt_sched.h>
+#include <linux/rtnetlink.h>
+#include <linux/sockios.h>
+#include <sys/ioctl.h>
+#include <sys/socket.h>
+#include <sys/uio.h>
+
+#include <ifaddrs.h>
+#elif defined(HAVE_BPF)
+#include <net/bpf.h>
+#include <net/if.h>
+#include <net/if_dl.h>
+#include <sys/ioctl.h>
+#include <sys/socket.h>
+#include <sys/uio.h>
+
+#include <fcntl.h>
+#include <ifaddrs.h>
+#elif defined(HAVE_NETMAP)
+#define NETMAP_WITH_LIBS
+#include <net/netmap_user.h>
+#include <net/if.h>
+#include <sys/ioctl.h>
+#include <sys/socket.h>
+
+#include <poll.h>
+#ifndef __linux__
+#include <net/if_dl.h>
+#include <ifaddrs.h>
+#endif
+#endif
+
+#include <errno.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+#define ETH_TYPE_LEN_SIZE sizeof(uint16_t)
+#define ETH_HDR_SIZE (2 * POA_MAC_SIZE + ETH_TYPE_LEN_SIZE)
+#define POA_HDR_SIZE (sizeof(struct poa_hdr))
+#define ETH_HDR_TOT_SIZE (ETH_HDR_SIZE + POA_HDR_SIZE)
+
+#define BPF_DEV_MAX 256 /* /dev/bpfN units to try */
+/* Wait for the link to come back before reading it again. */
+#define ETH_DOWN_TIMEO 100 /* ms */
+/* Budget for a qdisc dump; the send path must not wait on netlink. */
+#define POA_QDISC_TIMEO 5 /* ms */
+
+#ifndef ETH_MAX_MTU /* In if_ether.h as of Linux 4.10. */
+#define ETH_MAX_MTU 0xFFFFU
+#endif
+
+struct poa_hdr {
+ uint16_t eid;
+ uint16_t len;
+ uint8_t hcs;
+} __attribute__((packed));
+
+struct eth_hdr {
+ uint8_t dst[POA_MAC_SIZE];
+ uint8_t src[POA_MAC_SIZE];
+ uint16_t ethertype;
+ struct poa_hdr poa;
+} __attribute__((packed));
+
+
+struct eth_priv {
+ struct poa * poa;
+
+ int s_fd; /* raw socket or bpf device */
+ uint16_t ethertype; /* network order */
+ uint8_t hw_addr[POA_MAC_SIZE];
+ uint32_t mtu; /* device payload capacity */
+ int ifindex; /* link id; 0 where unknown */
+
+ pthread_t reader;
+ bool running;
+#ifdef HAVE_RAW_SOCKETS
+ struct sockaddr_ll device;
+#elif defined(HAVE_BPF)
+ size_t blen; /* bpf device buffer size */
+#elif defined(HAVE_NETMAP)
+ struct nm_desc * nmd;
+ struct pollfd poll_in;
+ struct pollfd poll_out;
+#endif
+ /* The kernel zeroes its counters on read, so accumulate. */
+ size_t kern_rcv;
+ size_t kern_drp;
+#ifdef HAVE_RAW_SOCKETS
+ /* Qdisc depth: netlink descriptor, its try-lock and cache. */
+ int nl_fd;
+ uint8_t nl_busy;
+ size_t nl_pkt;
+ size_t nl_byt;
+ uint64_t nl_time;
+#endif
+};
+
+struct eth_query {
+ struct list_head next;
+ uint8_t hash[POA_QUERY_HLEN];
+ /* The resolve's constraints; replies must satisfy them. */
+ uint16_t ethertype;
+ char c_dev[DEV_NAME_SIZE + 1];
+ uint8_t c_mac[POA_MAC_SIZE];
+ bool replied;
+ char dev[DEV_NAME_SIZE + 1];
+ uint8_t src_mac[POA_MAC_SIZE];
+ uint8_t mac[POA_MAC_SIZE];
+ uint16_t r_ethertype;
+};
+
+static struct {
+ pthread_once_t once;
+
+ struct llist pending;
+ pthread_mutex_t mtx;
+ pthread_cond_t cond;
+} queries = { .once = PTHREAD_ONCE_INIT };
+
+static void eth_hdr_ser(const struct eth_priv * priv,
+ struct eth_hdr * hdr,
+ const uint8_t * dst,
+ uint32_t eid,
+ size_t len)
+{
+ memcpy(hdr->dst, dst, POA_MAC_SIZE);
+ memcpy(hdr->src, priv->hw_addr, POA_MAC_SIZE);
+
+ hdr->ethertype = priv->ethertype;
+ hdr->poa.eid = htons((uint16_t) eid);
+ hdr->poa.len = htons((uint16_t) len);
+ mem_hash(HASH_CRC8, &hdr->poa.hcs,
+ (uint8_t *) &hdr->poa.eid, 2 * sizeof(uint16_t));
+}
+
+/* Oriented from us to the sender: our PoA, then their MAC. */
+static void frame_to_addr(const struct eth_priv * priv,
+ const struct eth_hdr * hdr,
+ struct poa_addr * addr)
+{
+ memset(addr, 0, sizeof(*addr));
+
+ addr->type = priv->poa->type;
+
+ addr->eth.src = priv->poa->local.eth.src;
+
+ memcpy(addr->eth.dst.mac, hdr->src, POA_MAC_SIZE);
+
+ addr->eth.dst.ethertype = ntohs(priv->ethertype);
+}
+
+static bool frame_is_for_us(const struct eth_priv * priv,
+ const uint8_t * dst)
+{
+ static const uint8_t bc[POA_MAC_SIZE] =
+ { 0xff, 0xff, 0xff, 0xff, 0xff, 0xff };
+
+ if (memcmp(dst, priv->hw_addr, POA_MAC_SIZE) == 0)
+ return true;
+
+ return memcmp(dst, bc, POA_MAC_SIZE) == 0;
+}
+
+/*
+ * Parse and validate a frame; on success *eid and *plen locate the
+ * payload. Errors are silent: anyone can spray an interface.
+ */
+static int frame_parse(const struct eth_priv * priv,
+ const uint8_t * buf,
+ size_t len,
+ uint32_t * eid,
+ size_t * plen)
+{
+ const struct eth_hdr * hdr = (const struct eth_hdr *) buf;
+ uint8_t hcs;
+
+ if (len < ETH_HDR_TOT_SIZE)
+ return -1;
+
+ if (!frame_is_for_us(priv, hdr->dst))
+ return -1;
+
+ if (hdr->ethertype != priv->ethertype)
+ return -1;
+
+ mem_hash(HASH_CRC8, &hcs,
+ (const uint8_t *) &hdr->poa.eid, 2 * sizeof(uint16_t));
+
+ if (hcs != hdr->poa.hcs)
+ return -1;
+
+ *eid = ntohs(hdr->poa.eid);
+
+ *plen = ntohs(hdr->poa.len);
+ if (*plen > len - ETH_HDR_TOT_SIZE)
+ return -1;
+
+ return 0;
+}
+
+#define ETH_QUERY_TIMEO 1900 /* ms total budget */
+#define ETH_QUERY_RETRIES 3 /* retransmits, 4 attempts total */
+
+/* A resolve in progress; a reply fills every matching entry. */
+static void queries_init(void)
+{
+ pthread_condattr_t cattr;
+
+ llist_init(&queries.pending);
+
+ pthread_mutex_init(&queries.mtx, NULL);
+
+ pthread_condattr_init(&cattr);
+#ifndef __APPLE__
+ pthread_condattr_setclock(&cattr, PTHREAD_COND_CLOCK);
+#endif
+ pthread_cond_init(&queries.cond, &cattr);
+ pthread_condattr_destroy(&cattr);
+}
+
+static void eth_query_cleanup(void * o)
+{
+ struct eth_query * q = (struct eth_query *) o;
+
+ pthread_mutex_lock(&queries.mtx);
+ llist_del(&q->next, &queries.pending);
+ pthread_mutex_unlock(&queries.mtx);
+}
+
+static void eth_query_arr(struct poa * poa,
+ const struct poa_addr * src,
+ const uint8_t * hash)
+{
+ uint8_t buf[sizeof(struct poa_mgmt_msg)
+ + POA_QUERY_HLEN];
+ struct poa_mgmt_msg * msg = (struct poa_mgmt_msg *) buf;
+
+ if (!poa_has_name(hash))
+ return;
+
+ poa_mgmt_msg_ser(msg, POA_NAME_REPLY, 0, 0, qos_raw, 0, POA_QUERY_HLEN);
+
+ memcpy(buf + sizeof(*msg), hash, POA_QUERY_HLEN);
+
+ if (poa->ops->poa_send_mgmt(poa, src, buf, sizeof(buf)) < 0)
+ return; /* the requester retransmits */
+
+ POA_STAT_BUMP(poa, rep_tx);
+}
+
+static bool mac_is_zero(const uint8_t * mac)
+{
+ static const uint8_t zero[POA_MAC_SIZE] = { 0 };
+
+ return memcmp(mac, zero, POA_MAC_SIZE) == 0;
+}
+
+static bool eth_query_is_match(const struct eth_query * q,
+ const struct poa * poa)
+{
+ if (q->ethertype != 0 &&
+ q->ethertype != poa->local.eth.src.ethertype)
+ return false;
+
+ if (q->c_dev[0] != '\0' &&
+ strcmp(q->c_dev, poa->local.eth.src.dev) != 0)
+ return false;
+
+ if (mac_is_zero(q->c_mac))
+ return true;
+
+ return memcmp(q->c_mac, poa->local.eth.src.mac, POA_MAC_SIZE) == 0;
+}
+
+static void eth_reply_arr(struct poa * poa,
+ const struct poa_addr * src,
+ const uint8_t * hash)
+{
+ struct list_head * p;
+
+ pthread_once(&queries.once, queries_init);
+
+ pthread_mutex_lock(&queries.mtx);
+
+ list_for_each(p, &queries.pending.list) {
+ struct eth_query * q;
+
+ q = list_entry(p, struct eth_query, next);
+ if (q->replied || memcmp(q->hash, hash, POA_QUERY_HLEN) != 0)
+ continue;
+
+ if (!eth_query_is_match(q, poa))
+ continue;
+
+ memcpy(q->mac, src->eth.dst.mac, POA_MAC_SIZE);
+ memcpy(q->src_mac, src->eth.src.mac, POA_MAC_SIZE);
+ strcpy(q->dev, src->eth.src.dev);
+
+ q->r_ethertype = poa->local.eth.src.ethertype;
+
+ q->replied = true;
+ }
+
+ pthread_cond_broadcast(&queries.cond);
+
+ pthread_mutex_unlock(&queries.mtx);
+}
+
+/* Name query frames are handled in the transport. */
+static void eth_rx_mgmt(struct poa * poa,
+ const struct poa_addr * src,
+ const uint8_t * buf,
+ size_t len)
+{
+ const struct poa_mgmt_msg * msg;
+ const uint8_t * hash;
+
+ msg = (const struct poa_mgmt_msg *) buf;
+ if (len < sizeof(*msg)) {
+ poa_rx_mgmt(poa, src, buf, len);
+ return;
+ }
+
+ if (msg->code != POA_NAME_QUERY && msg->code != POA_NAME_REPLY) {
+ poa_rx_mgmt(poa, src, buf, len);
+ return;
+ }
+
+ if (ntoh16(msg->data_len) != POA_QUERY_HLEN)
+ return; /* malformed */
+
+ if (len < sizeof(*msg) + POA_QUERY_HLEN)
+ return; /* malformed */
+
+ hash = buf + sizeof(*msg);
+
+ if (msg->code == POA_NAME_QUERY) {
+ POA_STAT_BUMP(poa, qry_rx);
+ eth_query_arr(poa, src, hash);
+ } else {
+ POA_STAT_BUMP(poa, rep_rx);
+ eth_reply_arr(poa, src, hash);
+ }
+}
+
+static int eth_query_send(const uint8_t * hash,
+ const struct poa_addr * addr)
+{
+ uint8_t buf[sizeof(struct poa_mgmt_msg)
+ + POA_QUERY_HLEN];
+ struct poa_mgmt_msg * msg = (struct poa_mgmt_msg *) buf;
+ struct poa_addr bcast;
+
+ bcast = *addr;
+
+ memset(bcast.eth.dst.mac, 0xff, POA_MAC_SIZE);
+
+ poa_mgmt_msg_ser(msg, POA_NAME_QUERY, 0, 0, qos_raw, 0, POA_QUERY_HLEN);
+
+ memcpy(buf + sizeof(*msg), hash, POA_QUERY_HLEN);
+
+ return poa_bcast_mgmt(&bcast, buf, sizeof(buf));
+}
+
+/* Complete addr for dst by broadcast query; the poa_query op. */
+static int eth_query(const char * dst,
+ const struct timespec * timeo,
+ struct poa_addr * addr)
+{
+ struct timespec dflt = TIMESPEC_INIT_MS(ETH_QUERY_TIMEO);
+ struct timespec rintv = TIMESPEC_INIT_MS(ETH_QUERY_TIMEO
+ / (ETH_QUERY_RETRIES + 1));
+ struct eth_query q;
+ struct timespec abstime;
+ struct timespec now;
+ struct timespec dl;
+ time_t rintv_ns = (time_t) TS_TO_UINT64(rintv);
+ time_t diff;
+ uint8_t hash[POA_QUERY_HLEN];
+ int n;
+ int err = -ETIMEDOUT;
+
+ if (strlen(addr->eth.src.dev) > DEV_NAME_SIZE)
+ return -EINVAL;
+
+ addr->eth.src.ethertype = addr->eth.dst.ethertype;
+
+ /* The destination is set; a zero ethertype cannot be sent. */
+ if (!mac_is_zero(addr->eth.dst.mac))
+ return addr->eth.dst.ethertype < 0x0600 ? -EINVAL : 0;
+
+ str_hash(HASH_SHA3_256, hash, dst);
+
+ pthread_once(&queries.once, queries_init);
+
+ memset(&q, 0, sizeof(q));
+ memcpy(q.hash, hash, POA_QUERY_HLEN);
+
+ q.ethertype = addr->eth.dst.ethertype;
+
+ memcpy(q.c_mac, addr->eth.src.mac, POA_MAC_SIZE);
+ strcpy(q.c_dev, addr->eth.src.dev);
+
+ pthread_mutex_lock(&queries.mtx);
+ llist_add(&q.next, &queries.pending);
+ pthread_mutex_unlock(&queries.mtx);
+
+ pthread_cleanup_push(eth_query_cleanup, &q);
+
+ clock_gettime(PTHREAD_COND_CLOCK, &now);
+ ts_add(&now, timeo != NULL ? timeo : &dflt, &abstime);
+
+ while (ts_diff_ns(&now, &abstime) < 0) {
+ n = eth_query_send(hash, addr);
+ if (n < 0) {
+ err = n;
+ break;
+ }
+
+ if (n == 0) {
+ err = -EPERM;
+ break;
+ }
+
+ ts_add(&now, &rintv, &dl);
+
+ if (ts_diff_ns(&dl, &abstime) > 0)
+ dl = abstime;
+
+ pthread_mutex_lock(&queries.mtx);
+
+ pthread_cleanup_push(__cleanup_mutex_unlock, &queries.mtx);
+
+ while (!q.replied) {
+ if (pthread_cond_timedwait(&queries.cond, &queries.mtx,
+ &dl) == ETIMEDOUT)
+ break;
+ }
+
+ if (q.replied) {
+ memcpy(addr->eth.dst.mac, q.mac, POA_MAC_SIZE);
+ memcpy(addr->eth.src.mac, q.src_mac, POA_MAC_SIZE);
+ strcpy(addr->eth.src.dev, q.dev);
+
+ addr->eth.dst.ethertype = q.r_ethertype;
+ addr->eth.src.ethertype = q.r_ethertype;
+
+ err = 0;
+ }
+
+ pthread_cleanup_pop(true);
+
+ if (err == 0)
+ break;
+
+ clock_gettime(PTHREAD_COND_CLOCK, &now);
+
+ diff = ts_diff_ns(&now, &abstime);
+ if (diff > -rintv_ns) /* skip the runt attempt */
+ break;
+ }
+
+ pthread_cleanup_pop(true);
+
+ return err;
+}
+
+static void eth_spec(const struct poa * poa,
+ struct poa_spec * spec)
+{
+ spec->type = poa->type;
+ spec->eth = poa->local.eth.src;
+}
+
+static bool eth_has_id(const struct poa * poa,
+ const struct poa_spec * spec)
+{
+ if (strnlen(spec->eth.dev, sizeof(spec->eth.dev)) > DEV_NAME_SIZE)
+ return false;
+
+ if (strcmp(poa->local.eth.src.dev, spec->eth.dev) != 0)
+ return false;
+
+ return poa->local.eth.src.ethertype == spec->eth.ethertype;
+}
+
+/* Our end of the link; an unnamed one matches any. */
+static bool eth_is_src(const struct poa * poa,
+ const struct eth_poa * src)
+{
+ if (src->dev[0] != '\0' &&
+ strcmp(poa->local.eth.src.dev, src->dev) != 0)
+ return false;
+
+ if (mac_is_zero(src->mac))
+ return true;
+
+ return memcmp(poa->local.eth.src.mac, src->mac,
+ POA_MAC_SIZE) == 0;
+}
+
+/* An ethertype of 0 in dst matches any, for broadcast queries. */
+static bool eth_match(const struct poa * poa,
+ const struct poa_addr * dst)
+{
+ uint16_t et = dst->eth.dst.ethertype;
+
+ if (et != 0 && et != poa->local.eth.src.ethertype)
+ return false;
+
+ return eth_is_src(poa, &dst->eth.src);
+}
+
+static bool eth_link_match(const struct poa * poa,
+ int id)
+{
+ struct eth_priv * priv = (struct eth_priv *) poa->priv;
+
+ return priv->ifindex == id;
+}
+
+#ifdef HAVE_RAW_SOCKETS
+
+/* MSG_DONTWAIT: the reader blocks on this socket. */
+static int eth_sendv(struct eth_priv * priv,
+ const uint8_t * dst,
+ uint32_t eid,
+ const uint8_t * body,
+ size_t len,
+ bool block,
+ const struct timespec * abstime)
+{
+ struct eth_hdr hdr;
+ struct msghdr msg;
+ struct iovec iov[2];
+ int ret;
+
+ if (len > priv->mtu - POA_HDR_SIZE)
+ return -EMSGSIZE;
+
+ eth_hdr_ser(priv, &hdr, dst, eid, len);
+
+ iov[0].iov_base = &hdr;
+ iov[0].iov_len = ETH_HDR_TOT_SIZE;
+ iov[1].iov_base = (void *) body;
+ iov[1].iov_len = len;
+
+ memset(&msg, 0, sizeof(msg));
+
+ msg.msg_name = &priv->device;
+ msg.msg_namelen = sizeof(priv->device);
+ msg.msg_iov = iov;
+ msg.msg_iovlen = len > 0 ? 2 : 1;
+ while (sendmsg(priv->s_fd, &msg, MSG_DONTWAIT) < 0) {
+ if (errno != EAGAIN && errno != EWOULDBLOCK)
+ return -EIO;
+
+ if (!block)
+ return -EAGAIN;
+
+ ret = poa_wait_out(priv->s_fd, abstime);
+ if (ret < 0)
+ return ret;
+ }
+
+ return 0;
+}
+
+static void * eth_reader(void * o)
+{
+ struct poa * poa = (struct poa *) o;
+ struct eth_priv * priv = (struct eth_priv *) poa->priv;
+ struct timespec down = TIMESPEC_INIT_MS(ETH_DOWN_TIMEO);
+ uint8_t * buf;
+ size_t bufsz;
+
+ bufsz = ETH_HDR_SIZE + priv->mtu;
+
+ buf = malloc(bufsz);
+ if (buf == NULL)
+ return (void *) -1;
+
+ pthread_cleanup_push(free, buf);
+
+ while (true) {
+ struct ssm_pk_buff * spb;
+ struct sockaddr_ll from;
+ socklen_t flen;
+ struct poa_addr src;
+ ssize_t n;
+ uint32_t eid;
+ size_t plen;
+ const uint8_t * body;
+
+ flen = sizeof(from);
+
+ n = recvfrom(priv->s_fd, buf, bufsz, 0,
+ (struct sockaddr *) &from, &flen);
+
+ if (n < 0) {
+ if (errno == EINTR)
+ continue;
+
+ POA_STAT_BUMP(poa, rcv_fail);
+
+ if (errno == ENETDOWN) {
+ nanosleep(&down, NULL);
+ continue;
+ }
+
+ log_err("Reader on %s stopped: %s.",
+ poa->local.eth.src.dev,
+ strerror(errno));
+ break;
+ }
+
+ if (from.sll_pkttype == PACKET_OUTGOING)
+ continue;
+
+ if (frame_parse(priv, buf, (size_t) n, &eid, &plen) < 0)
+ continue;
+
+ body = buf + ETH_HDR_TOT_SIZE;
+
+ if (eid == POA_MGMT_EID) {
+ frame_to_addr(priv, (struct eth_hdr *) buf, &src);
+ eth_rx_mgmt(poa, &src, body, plen);
+ continue;
+ }
+
+ if (poa_spb_reserve(&spb, plen) < 0) {
+ POA_STAT_BUMP(poa, buf_fail);
+ continue;
+ }
+
+ memcpy(ssm_pk_buff_head(spb), body, plen);
+
+ poa_rx_pkt(poa, eid, spb);
+ }
+
+ pthread_cleanup_pop(true);
+
+ return (void *) 0;
+}
+
+/*
+ * One netlink socket for the whole subsystem: RTMGRP_LINK delivers the
+ * events of every interface anyway, so a socket per PoA only added
+ * discards.
+ */
+int poa_monitor_open(void)
+{
+ struct sockaddr_nl sa;
+ int fd;
+
+ memset(&sa, 0, sizeof(sa));
+
+ sa.nl_family = AF_NETLINK;
+ sa.nl_groups = RTMGRP_LINK;
+
+ fd = socket(AF_NETLINK, SOCK_RAW, NETLINK_ROUTE);
+ if (fd < 0)
+ return -1;
+
+ if (bind(fd, (struct sockaddr *) &sa, sizeof(sa)) < 0) {
+ close(fd);
+ return -1;
+ }
+
+ return fd;
+}
+
+void poa_monitor_read(int fd)
+{
+ uint8_t buf[4096];
+ struct nlmsghdr * h;
+ ssize_t n;
+
+ n = recv(fd, buf, sizeof(buf), 0);
+ if (n < 0)
+ return;
+
+ for (h = (struct nlmsghdr *) buf;
+ NLMSG_OK(h, (unsigned int) n);
+ h = NLMSG_NEXT(h, n)) {
+ struct ifinfomsg * ifi;
+ unsigned int usable;
+ bool up;
+ size_t cnt;
+
+ if (h->nlmsg_type == NLMSG_DONE)
+ break;
+
+ if (h->nlmsg_type != RTM_NEWLINK)
+ continue;
+
+ ifi = NLMSG_DATA(h);
+
+ usable = ifi->ifi_flags & (IFF_UP | IFF_RUNNING);
+ up = usable == (IFF_UP | IFF_RUNNING);
+
+ cnt = poa_link_updown(ifi->ifi_index, up);
+ if (cnt > 0)
+ log_info("Link %d %s, %zu flows.", ifi->ifi_index,
+ up ? "up" : "down", cnt);
+ }
+}
+
+static int eth_dev_info(const char * dev,
+ int * idx,
+ uint8_t * mac,
+ uint32_t * mtu)
+{
+ struct ifreq ifr;
+ int fd;
+
+ if (strlen(dev) >= IFNAMSIZ)
+ return -EINVAL;
+
+ *idx = if_nametoindex(dev);
+ if (*idx == 0) {
+ log_err("Failed to find device %s.", dev);
+ return -ENODEV;
+ }
+
+ fd = socket(AF_UNIX, SOCK_STREAM, 0);
+ if (fd < 0)
+ return -EIO;
+
+ memset(&ifr, 0, sizeof(ifr));
+ strcpy(ifr.ifr_name, dev);
+
+ if (ioctl(fd, SIOCGIFHWADDR, &ifr) < 0) {
+ log_err("Failed to get hardware address of %s.", dev);
+ goto fail_ioctl;
+ }
+
+ memcpy(mac, ifr.ifr_hwaddr.sa_data, POA_MAC_SIZE);
+
+ if (ioctl(fd, SIOCGIFMTU, &ifr) < 0) {
+ log_err("Failed to get MTU of %s.", dev);
+ goto fail_ioctl;
+ }
+
+ close(fd);
+
+ *mtu = MIN(MIN(ETH_MAX_MTU, POA_ETH_RD_BUF), (uint32_t) ifr.ifr_mtu);
+ if (memcmp(dev, "lo", 2) == 0 && *mtu > POA_ETH_LO_MTU)
+ *mtu = POA_ETH_LO_MTU;
+
+ return 0;
+
+ fail_ioctl:
+ close(fd);
+ return -EIO;
+}
+
+/* SO_RCVBUFFORCE bypasses rmem_max; SO_RCVBUF is the fallback. */
+static void eth_set_rcvbuf(int fd,
+ int rcvbuf)
+{
+ if (setsockopt(fd, SOL_SOCKET, SO_RCVBUFFORCE,
+ &rcvbuf, sizeof(rcvbuf)) == 0)
+ return;
+
+ if (setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &rcvbuf, sizeof(rcvbuf)) < 0)
+ log_info("Failed to set SO_RCVBUF to %d.", rcvbuf);
+}
+
+/* The send buffer holds at least one frame, or sendmsg cannot fit a PDU. */
+static int eth_socket(struct eth_priv * priv,
+ int idx,
+ const uint8_t * mac)
+{
+ int sndbuf;
+ int rcvbuf;
+#ifdef POA_ETH_QDISC_BYPASS
+ int bypass = 1;
+#endif
+
+ memset(&priv->device, 0, sizeof(priv->device));
+
+ priv->device.sll_ifindex = idx;
+ priv->device.sll_family = AF_PACKET;
+ priv->device.sll_halen = POA_MAC_SIZE;
+ priv->device.sll_protocol = htons(ETH_P_ALL);
+
+ memcpy(priv->device.sll_addr, mac, POA_MAC_SIZE);
+ memcpy(priv->hw_addr, mac, POA_MAC_SIZE);
+
+ priv->s_fd = socket(AF_PACKET, SOCK_RAW, priv->ethertype);
+ if (priv->s_fd < 0) {
+ log_err("Failed to create socket: %s.", strerror(errno));
+ return -1;
+ }
+
+#ifdef POA_ETH_QDISC_BYPASS
+ if (setsockopt(priv->s_fd, SOL_PACKET, PACKET_QDISC_BYPASS,
+ &bypass, sizeof(bypass)) < 0)
+ log_info("Qdisc bypass not supported.");
+#endif
+
+ sndbuf = POA_ETH_SNDBUF;
+ if (sndbuf > 0) {
+ sndbuf = MAX(sndbuf, (int) (ETH_HDR_SIZE + priv->mtu));
+
+ if (setsockopt(priv->s_fd, SOL_SOCKET, SO_SNDBUF,
+ &sndbuf, sizeof(sndbuf)) < 0)
+ log_info("Failed to set SO_SNDBUF to %d.", sndbuf);
+ }
+
+ rcvbuf = POA_ETH_RCVBUF;
+ if (rcvbuf > 0)
+ eth_set_rcvbuf(priv->s_fd, rcvbuf);
+
+ if (bind(priv->s_fd, (struct sockaddr *) &priv->device,
+ sizeof(priv->device)) < 0) {
+ log_err("Failed to bind socket to %d.", idx);
+ close(priv->s_fd);
+ return -1;
+ }
+
+ return 0;
+}
+
+/*
+ * Pull qlen and backlog from the nested TCA_STATS2. The top-level
+ * TCA_STATS shares TCA_STATS_QUEUE's id but carries a wholly
+ * different struct, so descend first; never match on id alone.
+ */
+static void eth_qdisc_parse(struct rtattr * rta,
+ size_t rlen,
+ size_t * byt,
+ size_t * pkt)
+{
+ struct gnet_stats_queue q;
+ struct rtattr * in;
+ size_t ilen;
+
+ for (; RTA_OK(rta, rlen); rta = RTA_NEXT(rta, rlen)) {
+ if (rta->rta_type != TCA_STATS2)
+ continue;
+
+ in = (struct rtattr *) RTA_DATA(rta);
+ ilen = RTA_PAYLOAD(rta);
+ for (; RTA_OK(in, ilen); in = RTA_NEXT(in, ilen)) {
+ if (in->rta_type != TCA_STATS_QUEUE)
+ continue;
+
+ if (RTA_PAYLOAD(in) < sizeof(q))
+ continue;
+
+ memcpy(&q, RTA_DATA(in), sizeof(q));
+
+ *byt = q.backlog;
+ *pkt = q.qlen;
+ }
+ }
+}
+
+/*
+ * Egress backlog of the device's root qdisc, in bytes and packets,
+ * as the kernel queues them. The caller owns the netlink descriptor
+ * and serialises the query. An early end of dump reports failure.
+ */
+static int eth_qdisc_backlog(int fd,
+ int ifindex,
+ size_t * byt,
+ size_t * pkt)
+{
+ struct {
+ struct nlmsghdr nh;
+ struct tcmsg tc;
+ } req;
+ struct nlmsghdr * nh;
+ struct rtattr * rta;
+ struct tcmsg * tc;
+ char buf[16384];
+ ssize_t len;
+ int ret = -1;
+
+ if (fd < 0)
+ goto fail;
+
+ memset(&req, 0, sizeof(req));
+
+ req.nh.nlmsg_len = NLMSG_LENGTH(sizeof(req.tc));
+ req.nh.nlmsg_type = RTM_GETQDISC;
+ req.nh.nlmsg_flags = NLM_F_REQUEST | NLM_F_DUMP;
+ req.nh.nlmsg_seq = 1;
+ req.tc.tcm_family = AF_UNSPEC;
+ req.tc.tcm_ifindex = ifindex;
+ if (send(fd, &req, req.nh.nlmsg_len, 0) < 0)
+ goto fail;
+
+ *byt = 0;
+ *pkt = 0;
+
+ while ((len = recv(fd, buf, sizeof(buf), 0)) > 0) {
+ nh = (struct nlmsghdr *) buf;
+ for (; NLMSG_OK(nh, len); nh = NLMSG_NEXT(nh, len)) {
+ if (nh->nlmsg_type == NLMSG_DONE)
+ goto done;
+
+ if (nh->nlmsg_type == NLMSG_ERROR)
+ goto fail;
+
+ if (nh->nlmsg_type != RTM_NEWQDISC)
+ continue;
+
+ tc = (struct tcmsg *) NLMSG_DATA(nh);
+ if (tc->tcm_ifindex != ifindex)
+ continue;
+
+ if (tc->tcm_parent != TC_H_ROOT)
+ continue;
+
+ rta = (struct rtattr *)
+ ((char *) tc + NLMSG_ALIGN(sizeof(*tc)));
+
+ eth_qdisc_parse(rta, nh->nlmsg_len
+ - NLMSG_LENGTH(sizeof(*tc)),
+ byt, pkt);
+ }
+ }
+
+ goto fail; /* early end: a zero would read as empty */
+ done:
+ ret = 0;
+ fail:
+ return ret;
+}
+
+/*
+ * A netlink descriptor for the qdisc query. SO_RCVTIMEO bounds the
+ * dump: this is read from the send path, and a reply that never
+ * arrives must not park a sender thread.
+ */
+static int eth_qdisc_open(void)
+{
+ struct sockaddr_nl sa;
+ struct timeval tv = TIMEVAL_INIT_MS(POA_QDISC_TIMEO);
+ int fd;
+
+ fd = socket(AF_NETLINK, SOCK_RAW | SOCK_CLOEXEC, NETLINK_ROUTE);
+ if (fd < 0)
+ return -1;
+
+ memset(&sa, 0, sizeof(sa));
+
+ sa.nl_family = AF_NETLINK;
+ if (bind(fd, (struct sockaddr *) &sa, sizeof(sa)) < 0)
+ goto fail;
+
+ if (setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)) < 0)
+ goto fail;
+
+ return fd;
+ fail:
+ close(fd);
+
+ return -1;
+}
+
+/*
+ * Queue depth in packets, straight from the qdisc. Gated like the
+ * transport's own depth and skipped when another sender is already
+ * asking, so the send path never waits on netlink.
+ */
+static int eth_qpkts(struct poa * poa,
+ size_t * pkts,
+ size_t * byts)
+{
+ struct eth_priv * priv = (struct eth_priv *) poa->priv;
+ struct timespec now;
+ uint64_t ns;
+ size_t byt;
+ size_t pkt;
+
+ if (priv->nl_fd < 0)
+ return -1;
+
+ clock_gettime(PTHREAD_COND_CLOCK, &now);
+
+ ns = TS_TO_UINT64(now);
+ if (ns - LOAD_RELAXED(&priv->nl_time) < POA_QLEN_GATE)
+ goto cached;
+
+ if (__atomic_exchange_n(&priv->nl_busy, 1, __ATOMIC_ACQUIRE) != 0)
+ goto cached;
+
+ if (eth_qdisc_backlog(priv->nl_fd, priv->ifindex, &byt, &pkt) == 0) {
+ STORE_RELAXED(&priv->nl_pkt, pkt);
+ STORE_RELAXED(&priv->nl_byt, byt);
+ STORE_RELAXED(&priv->nl_time, ns);
+ }
+
+ __atomic_store_n(&priv->nl_busy, 0, __ATOMIC_RELEASE);
+ cached:
+ if (LOAD_RELAXED(&priv->nl_time) == 0) /* nothing measured yet */
+ return -1;
+
+ *pkts = LOAD_RELAXED(&priv->nl_pkt);
+ *byts = LOAD_RELAXED(&priv->nl_byt);
+
+ return 0;
+}
+
+static int eth_attach(struct poa * poa,
+ const struct poa_spec * spec)
+{
+ const struct eth_poa * c = &spec->eth;
+ struct eth_priv * priv;
+ uint8_t mac[POA_MAC_SIZE];
+ uint32_t mtu;
+ int idx;
+ int err;
+
+ if (strnlen(c->dev, sizeof(c->dev)) > DEV_NAME_SIZE)
+ return -EINVAL;
+
+ if (c->ethertype < 0x0600 || c->ethertype == 0xFFFF) {
+ log_err("Invalid Ethertype 0x%x.", c->ethertype);
+ return -EINVAL;
+ }
+
+ priv = malloc(sizeof(*priv));
+ if (priv == NULL)
+ return -ENOMEM;
+
+ memset(priv, 0, sizeof(*priv));
+
+ priv->poa = poa;
+ priv->nl_fd = eth_qdisc_open(); /* optional; socket fallback */
+
+ priv->ethertype = htons(c->ethertype);
+
+ err = eth_dev_info(c->dev, &idx, mac, &mtu);
+ if (err < 0)
+ goto fail_conf;
+
+ priv->mtu = mtu;
+ if (eth_socket(priv, idx, mac) < 0)
+ goto fail_conf;
+
+ poa->priv = priv;
+ poa->local.type = poa->type;
+ poa->local.eth.src.ethertype = c->ethertype;
+ priv->ifindex = idx;
+
+ memcpy(poa->local.eth.src.mac, mac, POA_MAC_SIZE);
+ strcpy(poa->local.eth.src.dev, c->dev);
+
+ return 0;
+
+ fail_conf:
+ if (priv->nl_fd >= 0)
+ close(priv->nl_fd);
+
+ free(priv);
+
+ return -EIO;
+}
+
+static void eth_detach(struct poa * poa)
+{
+ struct eth_priv * priv = (struct eth_priv *) poa->priv;
+
+ if (priv == NULL)
+ return;
+
+ close(priv->s_fd);
+
+ if (priv->nl_fd >= 0)
+ close(priv->nl_fd);
+
+ free(priv);
+
+ poa->priv = NULL;
+}
+
+static uint32_t eth_mtu(struct poa * poa,
+ const struct poa_addr * dst)
+{
+ struct eth_priv * priv = (struct eth_priv *) poa->priv;
+
+ (void) dst;
+
+ return priv->mtu - POA_HDR_SIZE;
+}
+
+/* All flows on the PoA share the socket, so this is aggregate. */
+static size_t eth_qlen(struct poa * poa)
+{
+#ifdef SIOCOUTQ
+ struct eth_priv * priv = (struct eth_priv *) poa->priv;
+ int qlen;
+
+ qlen = 0;
+ if (ioctl(priv->s_fd, SIOCOUTQ, &qlen) < 0)
+ return 0;
+
+ return (size_t) qlen;
+#else
+ (void) poa;
+
+ return 0;
+#endif
+}
+
+/*
+ * PACKET_STATISTICS zeroes the kernel counters on read, so totals
+ * accumulate here; relaxed atomics allow concurrent RIB reads.
+ * The qdisc depth rides the sender's guarded path: one dump at a time.
+ */
+static int eth_rib(struct poa * poa,
+ char * buf,
+ size_t len)
+{
+ struct eth_priv * priv = (struct eth_priv *) poa->priv;
+ struct tpacket_stats ts;
+ socklen_t optlen;
+ size_t sndbuf = 0;
+ size_t rcvbuf = 0;
+ size_t qd_byt;
+ size_t qd_pkt;
+ int val;
+ int size;
+
+ optlen = sizeof(val);
+ if (getsockopt(priv->s_fd, SOL_SOCKET, SO_SNDBUF, &val, &optlen) == 0)
+ sndbuf = (size_t) val;
+
+ optlen = sizeof(val);
+ if (getsockopt(priv->s_fd, SOL_SOCKET, SO_RCVBUF, &val, &optlen) == 0)
+ rcvbuf = (size_t) val;
+
+ optlen = sizeof(ts);
+
+ if (getsockopt(priv->s_fd, SOL_PACKET, PACKET_STATISTICS,
+ &ts, &optlen) == 0) {
+ FETCH_ADD_RELAXED(&priv->kern_rcv, ts.tp_packets);
+ FETCH_ADD_RELAXED(&priv->kern_drp, ts.tp_drops);
+ }
+
+ if (eth_qpkts(poa, &qd_pkt, &qd_byt) < 0) {
+ qd_byt = 0;
+ qd_pkt = 0;
+ }
+
+ size = snprintf(buf, len,
+ "Socket sndbuf (bytes): %zu\n"
+ "Socket rcvbuf (bytes): %zu\n"
+ "Kernel packets received: %zu\n"
+ "Kernel packets dropped: %zu\n"
+ "Qdisc backlog (bytes): %zu\n"
+ "Qdisc backlog (packets): %zu\n",
+ sndbuf, rcvbuf,
+ LOAD_RELAXED(&priv->kern_rcv),
+ LOAD_RELAXED(&priv->kern_drp),
+ qd_byt, qd_pkt);
+ if (size < 0 || (size_t) size >= len)
+ return -1;
+
+ return size;
+}
+
+#elif defined(HAVE_BPF)
+
+/*
+ * BSD and macOS reach the link layer through a cloned /dev/bpf
+ * device, bound to an interface with BIOCSETIF. One device per PoA.
+ */
+
+static int eth_sendv(struct eth_priv * priv,
+ const uint8_t * dst,
+ uint32_t eid,
+ const uint8_t * body,
+ size_t len,
+ bool block,
+ const struct timespec * abstime)
+{
+ struct eth_hdr hdr;
+ struct iovec iov[2];
+ int n;
+ int ret;
+
+ if (len > priv->mtu - POA_HDR_SIZE)
+ return -EMSGSIZE;
+
+ eth_hdr_ser(priv, &hdr, dst, eid, len);
+
+ iov[0].iov_base = &hdr;
+ iov[0].iov_len = ETH_HDR_TOT_SIZE;
+ iov[1].iov_base = (void *) body;
+ iov[1].iov_len = len;
+ n = len > 0 ? 2 : 1;
+ while (writev(priv->s_fd, iov, n) < 0) {
+ if (errno != EAGAIN && errno != EWOULDBLOCK)
+ return -EIO;
+
+ if (!block)
+ return -EAGAIN;
+
+ ret = poa_wait_out(priv->s_fd, abstime);
+ if (ret < 0)
+ return ret;
+ }
+
+ return 0;
+}
+
+/* One read yields a batch of BPF_WORDALIGN'ed frames; walk all of them. */
+static void * eth_reader(void * o)
+{
+ struct poa * poa = (struct poa *) o;
+ struct eth_priv * priv = (struct eth_priv *) poa->priv;
+ uint8_t * buf;
+
+ buf = malloc(priv->blen);
+ if (buf == NULL)
+ return (void *) -1;
+
+ pthread_cleanup_push(free, buf);
+
+ while (true) {
+ uint8_t * p;
+ uint8_t * end;
+ ssize_t n;
+
+ n = read(priv->s_fd, buf, priv->blen);
+ if (n < 0) {
+ if (errno == EINTR)
+ continue;
+
+ POA_STAT_BUMP(poa, rcv_fail);
+ break;
+ }
+
+ p = buf;
+ end = buf + n;
+ while (p + sizeof(struct bpf_hdr) <= end) {
+ struct bpf_hdr * bh = (struct bpf_hdr *) p;
+ struct ssm_pk_buff * spb;
+ struct poa_addr src;
+ const uint8_t * frame;
+ const uint8_t * body;
+ uint32_t eid;
+ size_t plen;
+
+ frame = p + bh->bh_hdrlen;
+ if (frame + bh->bh_caplen > end)
+ break;
+
+ if (frame_parse(priv, frame, bh->bh_caplen,
+ &eid, &plen) < 0)
+ goto next;
+
+ body = frame + ETH_HDR_TOT_SIZE;
+
+ if (eid == POA_MGMT_EID) {
+ frame_to_addr(priv,
+ (const struct eth_hdr *) frame,
+ &src);
+ eth_rx_mgmt(poa, &src, body, plen);
+ goto next;
+ }
+
+ if (poa_spb_reserve(&spb, plen) < 0) {
+ POA_STAT_BUMP(poa, buf_fail);
+ goto next;
+ }
+
+ memcpy(ssm_pk_buff_head(spb), body, plen);
+
+ poa_rx_pkt(poa, eid, spb);
+ next:
+ p += BPF_WORDALIGN(bh->bh_hdrlen + bh->bh_caplen);
+ }
+ }
+
+ pthread_cleanup_pop(true);
+
+ return (void *) 0;
+}
+
+static int eth_dev_info(const char * dev,
+ uint8_t * mac,
+ uint32_t * mtu)
+{
+ struct ifaddrs * ifas;
+ struct ifaddrs * ifa;
+ struct ifreq ifr;
+ int fd;
+ int found = 0;
+
+ if (getifaddrs(&ifas) < 0) {
+ log_err("Failed to list interfaces.");
+ return -1;
+ }
+
+ for (ifa = ifas; ifa != NULL; ifa = ifa->ifa_next) {
+ struct sockaddr_dl * dl;
+
+ if (ifa->ifa_addr == NULL)
+ continue;
+
+ if (ifa->ifa_addr->sa_family != AF_LINK)
+ continue;
+
+ if (strcmp(ifa->ifa_name, dev) != 0)
+ continue;
+
+ dl = (struct sockaddr_dl *) ifa->ifa_addr;
+ if (dl->sdl_alen != POA_MAC_SIZE)
+ continue;
+
+ memcpy(mac, LLADDR(dl), POA_MAC_SIZE);
+
+ found = 1;
+ break;
+ }
+
+ freeifaddrs(ifas);
+
+ if (!found) {
+ log_err("No such device: %s.", dev);
+ return -1;
+ }
+
+ fd = socket(AF_INET, SOCK_DGRAM, 0);
+ if (fd < 0)
+ return -1;
+
+ memset(&ifr, 0, sizeof(ifr));
+
+ strcpy(ifr.ifr_name, dev);
+
+ if (ioctl(fd, SIOCGIFMTU, &ifr) < 0) {
+ log_err("Failed to get MTU of %s.", dev);
+ close(fd);
+ return -1;
+ }
+
+ close(fd);
+
+ *mtu = MIN(MIN(ETH_MAX_MTU, POA_ETH_RD_BUF), (uint32_t) ifr.ifr_mtu);
+ if (memcmp(dev, "lo", 2) == 0 && *mtu > POA_ETH_LO_MTU)
+ *mtu = POA_ETH_LO_MTU;
+
+ return 0;
+}
+
+static int eth_bpf_open(void)
+{
+ char dev[32];
+ size_t i;
+
+ for (i = 0; i < BPF_DEV_MAX; ++i) {
+ int fd;
+
+ sprintf(dev, "/dev/bpf%zu", i);
+
+ fd = open(dev, O_RDWR);
+ if (fd >= 0)
+ return fd;
+ }
+
+ return -1;
+}
+
+/*
+ * BIOCSHDRCMPLT: we fill in the source address ourselves.
+ * BIOCSSEESENT: our own egress must not come back at us.
+ * BIOCIMMEDIATE: deliver on arrival, do not wait for a full buffer.
+ */
+static int eth_bpf_setup(struct eth_priv * priv,
+ const char * dev)
+{
+ struct ifreq ifr;
+ int enable = 1;
+ int disable = 0;
+ int blen = 0;
+
+ memset(&ifr, 0, sizeof(ifr));
+
+ strcpy(ifr.ifr_name, dev);
+
+ if (ioctl(priv->s_fd, BIOCSETIF, &ifr) < 0) {
+ log_err("Failed to bind bpf device to %s.", dev);
+ return -1;
+ }
+
+ if (ioctl(priv->s_fd, BIOCGBLEN, &blen) < 0 || blen <= 0) {
+ log_err("Failed to get the bpf buffer length.");
+ return -1;
+ }
+
+ priv->blen = (size_t) blen;
+ if (ioctl(priv->s_fd, BIOCSHDRCMPLT, &enable) < 0) {
+ log_err("Failed to set BIOCSHDRCMPLT.");
+ return -1;
+ }
+
+ if (ioctl(priv->s_fd, BIOCSSEESENT, &disable) < 0) {
+ log_err("Failed to set BIOCSSEESENT.");
+ return -1;
+ }
+
+ if (ioctl(priv->s_fd, BIOCIMMEDIATE, &enable) < 0) {
+ log_err("Failed to set BIOCIMMEDIATE.");
+ return -1;
+ }
+
+ return 0;
+}
+
+static int eth_attach(struct poa * poa,
+ const struct poa_spec * spec)
+{
+ const struct eth_poa * c = &spec->eth;
+ struct eth_priv * priv;
+ uint8_t mac[POA_MAC_SIZE];
+ uint32_t mtu;
+
+ if (strnlen(c->dev, sizeof(c->dev)) > DEV_NAME_SIZE)
+ return -EINVAL;
+
+ if (c->ethertype < 0x0600 || c->ethertype == 0xFFFF) {
+ log_err("Invalid Ethertype 0x%x.", c->ethertype);
+ return -EINVAL;
+ }
+
+ priv = malloc(sizeof(*priv));
+ if (priv == NULL)
+ return -ENOMEM;
+
+ memset(priv, 0, sizeof(*priv));
+
+ priv->poa = poa;
+ priv->s_fd = -1;
+
+ priv->ethertype = htons(c->ethertype);
+
+ if (eth_dev_info(c->dev, mac, &mtu) < 0)
+ goto fail_conf;
+
+ priv->mtu = mtu;
+
+ memcpy(priv->hw_addr, mac, POA_MAC_SIZE);
+
+ priv->s_fd = eth_bpf_open();
+ if (priv->s_fd < 0) {
+ log_err("Failed to open a bpf device.");
+ goto fail_conf;
+ }
+
+ if (eth_bpf_setup(priv, c->dev) < 0)
+ goto fail_device;
+
+ poa->priv = priv;
+ poa->local.type = poa->type;
+ poa->local.eth.src.ethertype = c->ethertype;
+
+ memcpy(poa->local.eth.src.mac, mac, POA_MAC_SIZE);
+ strcpy(poa->local.eth.src.dev, c->dev);
+
+ log_info("Using Berkeley Packet Filter on %s.", c->dev);
+
+ return 0;
+
+ fail_device:
+ close(priv->s_fd);
+ fail_conf:
+ free(priv);
+
+ return -EIO;
+}
+
+static void eth_detach(struct poa * poa)
+{
+ struct eth_priv * priv = (struct eth_priv *) poa->priv;
+
+ if (priv == NULL)
+ return;
+
+ close(priv->s_fd);
+
+ free(priv);
+
+ poa->priv = NULL;
+}
+
+static uint32_t eth_mtu(struct poa * poa,
+ const struct poa_addr * dst)
+{
+ struct eth_priv * priv = (struct eth_priv *) poa->priv;
+
+ (void) dst;
+
+ return priv->mtu - POA_HDR_SIZE;
+}
+
+/* A bpf device has no send queue to report; mb-ECN cannot mark here. */
+static size_t eth_qlen(struct poa * poa)
+{
+ (void) poa;
+
+ return 0;
+}
+
+/* The bpf device buffer is all the receive queue there is. */
+static int eth_rib(struct poa * poa,
+ char * buf,
+ size_t len)
+{
+ struct eth_priv * priv = (struct eth_priv *) poa->priv;
+ struct bpf_stat bs;
+ int size;
+
+ if (ioctl(priv->s_fd, BIOCGSTATS, &bs) == 0) {
+ FETCH_ADD_RELAXED(&priv->kern_rcv, bs.bs_recv);
+ FETCH_ADD_RELAXED(&priv->kern_drp, bs.bs_drop);
+ }
+
+ size = snprintf(buf, len,
+ "Socket rcvbuf (bytes): %zu\n"
+ "Kernel packets received: %zu\n"
+ "Kernel packets dropped: %zu\n",
+ priv->blen,
+ LOAD_RELAXED(&priv->kern_rcv),
+ LOAD_RELAXED(&priv->kern_drp));
+ if (size < 0 || (size_t) size >= len)
+ return -1;
+
+ return size;
+}
+
+#elif defined(HAVE_NETMAP)
+
+/*
+ * netmap gives one PoA the whole port: reads copy out of the NIC
+ * ring, writes inject into it.
+ */
+
+/* nm_inject takes one contiguous frame, so the header is copied in. */
+static int eth_sendv(struct eth_priv * priv,
+ const uint8_t * dst,
+ uint32_t eid,
+ const uint8_t * body,
+ size_t len,
+ bool block,
+ const struct timespec * abstime)
+{
+ uint8_t * frame;
+ size_t flen;
+ int ret;
+
+ if (len > priv->mtu - POA_HDR_SIZE)
+ return -EMSGSIZE;
+
+ flen = ETH_HDR_TOT_SIZE + len;
+
+ frame = malloc(flen);
+ if (frame == NULL)
+ return -ENOMEM;
+
+ eth_hdr_ser(priv, (struct eth_hdr *) frame, dst, eid, len);
+
+ if (len > 0)
+ memcpy(frame + ETH_HDR_TOT_SIZE, body, len);
+
+ if (block)
+ ret = poa_wait_out(priv->poll_out.fd, abstime);
+ else
+ ret = poll(&priv->poll_out, 1, 0) > 0 ? 0 : -EAGAIN;
+
+ if (ret < 0)
+ goto fail;
+
+ ret = nm_inject(priv->nmd, frame, flen) == (int) flen ? 0 : -EIO;
+ fail:
+ free(frame);
+
+ return ret;
+}
+
+/* A slot stays owned by the ring, so each frame is copied out. */
+static void * eth_reader(void * o)
+{
+ struct poa * poa = (struct poa *) o;
+ struct eth_priv * priv = (struct eth_priv *) poa->priv;
+
+ while (true) {
+ struct ssm_pk_buff * spb;
+ struct poa_addr src;
+ struct nm_pkthdr hdr;
+ const uint8_t * frame;
+ const uint8_t * body;
+ uint32_t eid;
+ size_t plen;
+
+ if (poll(&priv->poll_in, 1, -1) < 0) {
+ if (errno == EINTR)
+ continue;
+
+ POA_STAT_BUMP(poa, rcv_fail);
+ break;
+ }
+
+ if (priv->poll_in.revents == 0)
+ continue;
+
+ frame = nm_nextpkt(priv->nmd, &hdr);
+ if (frame == NULL)
+ continue;
+
+ if (frame_parse(priv, frame, hdr.len, &eid, &plen) < 0)
+ continue;
+
+ body = frame + ETH_HDR_TOT_SIZE;
+
+ if (eid == POA_MGMT_EID) {
+ frame_to_addr(priv, (const struct eth_hdr *) frame,
+ &src);
+ eth_rx_mgmt(poa, &src, body, plen);
+ continue;
+ }
+
+ if (poa_spb_reserve(&spb, plen) < 0) {
+ POA_STAT_BUMP(poa, buf_fail);
+ continue;
+ }
+
+ memcpy(ssm_pk_buff_head(spb), body, plen);
+
+ poa_rx_pkt(poa, eid, spb);
+ }
+
+ return (void *) 0;
+}
+
+static int eth_dev_info(const char * dev,
+ uint8_t * mac,
+ uint32_t * mtu)
+{
+ struct ifreq ifr;
+#ifndef __linux__
+ struct ifaddrs * ifas;
+ struct ifaddrs * ifa;
+ int found = 0;
+#endif
+ int fd;
+
+ if (strlen(dev) >= IFNAMSIZ)
+ return -EINVAL;
+
+ fd = socket(AF_INET, SOCK_DGRAM, 0);
+ if (fd < 0)
+ return -EIO;
+
+ memset(&ifr, 0, sizeof(ifr));
+
+ strcpy(ifr.ifr_name, dev);
+
+#ifdef __linux__
+ if (ioctl(fd, SIOCGIFHWADDR, &ifr) < 0) {
+ log_err("Failed to get hardware address of %s.", dev);
+ goto fail_ioctl;
+ }
+
+ memcpy(mac, ifr.ifr_hwaddr.sa_data, POA_MAC_SIZE);
+#else
+ if (getifaddrs(&ifas) < 0)
+ goto fail_ioctl;
+
+ for (ifa = ifas; ifa != NULL; ifa = ifa->ifa_next) {
+ struct sockaddr_dl * dl;
+
+ if (ifa->ifa_addr == NULL)
+ continue;
+
+ if (ifa->ifa_addr->sa_family != AF_LINK)
+ continue;
+
+ if (strcmp(ifa->ifa_name, dev) != 0)
+ continue;
+
+ dl = (struct sockaddr_dl *) ifa->ifa_addr;
+ if (dl->sdl_alen != POA_MAC_SIZE)
+ continue;
+
+ memcpy(mac, LLADDR(dl), POA_MAC_SIZE);
+
+ found = 1;
+ break;
+ }
+
+ freeifaddrs(ifas);
+
+ if (!found) {
+ log_err("No hardware address for %s.", dev);
+ goto fail_ioctl;
+ }
+#endif
+ if (ioctl(fd, SIOCGIFMTU, &ifr) < 0) {
+ log_err("Failed to get MTU of %s.", dev);
+ goto fail_ioctl;
+ }
+
+ close(fd);
+
+ *mtu = MIN(MIN(ETH_MAX_MTU, POA_ETH_RD_BUF), (uint32_t) ifr.ifr_mtu);
+ if (memcmp(dev, "lo", 2) == 0 && *mtu > POA_ETH_LO_MTU)
+ *mtu = POA_ETH_LO_MTU;
+
+ return 0;
+
+ fail_ioctl:
+ close(fd);
+
+ return -EIO;
+}
+
+static int eth_nm_open(struct eth_priv * priv,
+ const char * dev)
+{
+ char ifn[IFNAMSIZ + sizeof("netmap:")];
+
+ strcpy(ifn, "netmap:");
+ strcat(ifn, dev);
+
+ priv->nmd = nm_open(ifn, NULL, 0, NULL);
+ if (priv->nmd == NULL) {
+ log_err("Failed to open netmap device for %s.", dev);
+ return -1;
+ }
+
+ memset(&priv->poll_in, 0, sizeof(priv->poll_in));
+ memset(&priv->poll_out, 0, sizeof(priv->poll_out));
+
+ priv->poll_in.fd = NETMAP_FD(priv->nmd);
+ priv->poll_in.events = POLLIN;
+ priv->poll_out.fd = NETMAP_FD(priv->nmd);
+ priv->poll_out.events = POLLOUT;
+
+ return 0;
+}
+
+static int eth_attach(struct poa * poa,
+ const struct poa_spec * spec)
+{
+ const struct eth_poa * c = &spec->eth;
+ struct eth_priv * priv;
+ uint8_t mac[POA_MAC_SIZE];
+ uint32_t mtu;
+
+ if (strnlen(c->dev, sizeof(c->dev)) > DEV_NAME_SIZE)
+ return -EINVAL;
+
+ if (c->ethertype < 0x0600 || c->ethertype == 0xFFFF) {
+ log_err("Invalid Ethertype 0x%x.", c->ethertype);
+ return -EINVAL;
+ }
+
+ priv = malloc(sizeof(*priv));
+ if (priv == NULL)
+ return -ENOMEM;
+
+ memset(priv, 0, sizeof(*priv));
+
+ priv->poa = poa;
+
+ priv->ethertype = htons(c->ethertype);
+
+ if (eth_dev_info(c->dev, mac, &mtu) < 0)
+ goto fail_conf;
+
+ priv->mtu = mtu;
+
+ memcpy(priv->hw_addr, mac, POA_MAC_SIZE);
+
+ if (eth_nm_open(priv, c->dev) < 0)
+ goto fail_conf;
+
+ poa->priv = priv;
+ poa->local.type = poa->type;
+ poa->local.eth.src.ethertype = c->ethertype;
+
+ memcpy(poa->local.eth.src.mac, mac, POA_MAC_SIZE);
+ strcpy(poa->local.eth.src.dev, c->dev);
+
+ log_info("Using netmap on %s.", c->dev);
+
+ return 0;
+
+ fail_conf:
+ free(priv);
+
+ return -EIO;
+}
+
+static void eth_detach(struct poa * poa)
+{
+ struct eth_priv * priv = (struct eth_priv *) poa->priv;
+
+ if (priv == NULL)
+ return;
+
+ nm_close(priv->nmd);
+
+ free(priv);
+
+ poa->priv = NULL;
+}
+
+static uint32_t eth_mtu(struct poa * poa,
+ const struct poa_addr * dst)
+{
+ struct eth_priv * priv = (struct eth_priv *) poa->priv;
+
+ (void) dst;
+
+ return priv->mtu - POA_HDR_SIZE;
+}
+
+/* The ring is drained by the NIC; there is no queue to report. */
+static size_t eth_qlen(struct poa * poa)
+{
+ (void) poa;
+
+ return 0;
+}
+
+#endif /* HAVE_RAW_SOCKETS */
+
+#ifndef HAVE_RAW_SOCKETS
+
+/* Only netlink reports link events; no other backend has a monitor. */
+int poa_monitor_open(void)
+{
+ return -1;
+}
+
+void poa_monitor_read(int fd)
+{
+ (void) fd;
+}
+
+#endif
+
+/* One reader per socket, so a flow cannot be reordered on receive. */
+static int eth_start(struct poa * poa)
+{
+ struct eth_priv * priv = (struct eth_priv *) poa->priv;
+
+ if (pthread_create(&priv->reader, NULL, eth_reader, poa) != 0)
+ return -1;
+
+ priv->running = true;
+
+ return 0;
+}
+
+static void eth_stop(struct poa * poa)
+{
+ struct eth_priv * priv = (struct eth_priv *) poa->priv;
+
+ if (!priv->running)
+ return;
+
+ pthread_cancel(priv->reader);
+ pthread_join(priv->reader, NULL);
+
+ priv->running = false;
+}
+
+static int eth_send(struct poa * poa,
+ const struct poa_addr * dst,
+ uint32_t eid,
+ struct ssm_pk_buff * spb,
+ bool block,
+ const struct timespec * abstime)
+{
+ return eth_sendv((struct eth_priv *) poa->priv, dst->eth.dst.mac, eid,
+ ssm_pk_buff_head(spb), ssm_pk_buff_len(spb),
+ block, abstime);
+}
+
+static int eth_send_mgmt(struct poa * poa,
+ const struct poa_addr * dst,
+ const uint8_t * buf,
+ size_t len)
+{
+ struct timespec timeo = TIMESPEC_INIT_MS(POA_MGMT_SND_TIMEO);
+ struct timespec abstime;
+
+ clock_gettime(PTHREAD_COND_CLOCK, &abstime);
+ ts_add(&abstime, &timeo, &abstime);
+
+ return eth_sendv((struct eth_priv *) poa->priv, dst->eth.dst.mac,
+ POA_MGMT_EID, buf, len, true, &abstime);
+}
+
+const struct poa_ops eth_poa_ops = {
+ .poa_attach = eth_attach,
+ .poa_detach = eth_detach,
+ .poa_start = eth_start,
+ .poa_stop = eth_stop,
+ .poa_send = eth_send,
+ .poa_send_mgmt = eth_send_mgmt,
+ .poa_query = eth_query,
+ .poa_mtu = eth_mtu,
+ .poa_qlen = eth_qlen,
+#ifdef HAVE_RAW_SOCKETS
+ .poa_qpkts = eth_qpkts,
+#endif
+#ifndef HAVE_NETMAP
+ .poa_rib = eth_rib,
+#endif
+ .poa_spec = eth_spec,
+ .poa_has_id = eth_has_id,
+ .poa_match = eth_match,
+ .poa_link_match = eth_link_match,
+ .mpl = POA_ETH_MPL
+};
diff --git a/src/lib/poa/poa.c b/src/lib/poa/poa.c
new file mode 100644
index 00000000..3ad17c4f
--- /dev/null
+++ b/src/lib/poa/poa.c
@@ -0,0 +1,2515 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2026
+ *
+ * Points of attachment (PoA) - transport independent core
+ *
+ * Included by dev.c; uses dev.c statics (proc, flow_init, ...).
+ *
+ * Dimitri Staessens <dimitri@ouroboros.rocks>
+ * Sander Vrijders <sander@ouroboros.rocks>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public License
+ * version 2.1 as published by the Free Software Foundation.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., http://www.fsf.org/about/contact/.
+ */
+
+#define POA_MAX_EIDS PROC_MAX_FLOWS
+#define POA_ALLOC_TIMEO 10000 /* ms, overall FLOW_REQ deadline */
+#define POA_RETRY_TIMEO 300 /* ms, FLOW_REQ retransmit period */
+/* Must fit a certificate chain: post-quantum ones are large. */
+#define POA_MGMT_BUFSZ POA_MGMT_FRAME_SIZE
+#define POA_MGMT_QMAX 64 /* queued management frames per PoA */
+#define POA_PEND_TIMEO 10 /* s, reap a request that never completes */
+#define POA_SWEEP_TIMEO 1000 /* ms, sweep interval */
+#define POA_SWEEP_MAX 16 /* requests reaped per sweep */
+#define POA_DEFER_MAX 64 /* replies waiting for their flow id */
+
+/* EWMA over 8 samples. */
+#define POA_AVG_SHIFT 3
+/* Queue cost is sampled every 64th packet: qlen is a syscall. */
+#define POA_COST_MASK 63
+/* Reuse a qlen read for this long; the mark moves on doublings. */
+
+#define POA_RIB "poa"
+/* Fits the RIB labels below with 20-digit counters. */
+#define POA_RIB_STRLEN 2048
+
+enum poa_flow_state {
+ POA_FLOW_NULL = 0,
+ POA_FLOW_PENDING,
+ POA_FLOW_ALLOCATED,
+ POA_FLOW_DEAD
+};
+
+enum poa_state {
+ POA_NULL = 0,
+ POA_INIT, /* poa_init(); this process may attach */
+ POA_RUNNING, /* the threads are up */
+ POA_OPERATIONAL /* bootstrapped or enrolled in a layer */
+};
+
+struct poa_flow {
+ struct poa * poa;
+
+ int flow_id;
+ uint32_t eid;
+ uint32_t r_eid;
+ struct poa_addr r_addr;
+
+ enum poa_flow_state state;
+
+ struct ssm_rbuff * rx_rb;
+
+ struct list_head pend; /* on poa->pending while unanswered */
+ struct timespec t0;
+
+ /* Our answer, kept to re-send when a request is repeated. */
+ bool answered;
+ bool answer_sent;
+
+ /* Handed to a caller that will attach it; not the sweeper's. */
+ bool claimed;
+ int answer;
+ buffer_t answer_data;
+
+ /* Handshake rendezvous with the reader thread. */
+ pthread_mutex_t mtx;
+ pthread_cond_t cond;
+ bool replied;
+ bool pending;
+ int response;
+ buffer_t resp_data;
+};
+
+/*
+ * An answer can be ready before the flow it answers has an id, and
+ * the caller must not be kept waiting for one: it answers to the IRMd,
+ * which gives up long before we would.
+ */
+struct poa_deferred {
+ struct list_head next;
+ struct timespec t0;
+ int flow_id;
+ int response;
+ buffer_t data;
+};
+
+struct poa_mgmt_frame {
+ struct list_head next;
+ struct poa * poa;
+ struct poa_addr src;
+ size_t len;
+ uint8_t buf[POA_MGMT_BUFSZ];
+};
+
+/*
+ * A detach is performed by the management thread, so that it cannot
+ * run alongside the accept side. The caller waits for the result.
+ */
+struct poa_detach_req {
+ struct list_head next;
+ struct poa_spec spec;
+ int result;
+ bool done;
+};
+
+static struct {
+ struct list_head list;
+
+ struct poa_flow * id_to_pf[SYS_MAX_FLOWS];
+
+ struct llist deferred;
+
+ pthread_mutex_t mtx; /* guards id_to_pf */
+ pthread_cond_t cond;
+
+ /* One management thread and one link monitor for all PoAs. */
+ struct llist mgmt_frames;
+ struct list_head detach;
+ pthread_mutex_t mgmt_mtx;
+ pthread_cond_t mgmt_cond; /* work for the handler */
+ pthread_cond_t done_cond; /* a detach has completed */
+ pthread_t mgmt_handler;
+ bool mgmt_stop;
+
+ pthread_t monitor;
+ int mon_fd;
+
+ enum poa_state state;
+
+ /* Hashes of the names this process answers queries for. */
+ uint8_t name_hash[POA_QUERY_HLEN];
+ uint8_t layer_hash[POA_QUERY_HLEN];
+
+ pthread_rwlock_t lock; /* the PoAs and their flows */
+} poas;
+
+static int mgmt_send(struct poa * poa,
+ const struct poa_addr * dst,
+ uint8_t code,
+ uint32_t s_eid,
+ uint32_t d_eid,
+ qosspec_t qs,
+ int response,
+ const buffer_t * data)
+{
+ uint8_t buf[POA_MGMT_BUFSZ];
+ struct poa_mgmt_msg * msg = (struct poa_mgmt_msg *) buf;
+ size_t len;
+
+ len = sizeof(*msg);
+ if (data != NULL && data->len > 0) {
+ if (len + data->len > sizeof(buf))
+ return -EMSGSIZE;
+ memcpy(buf + len, data->data, data->len);
+
+ len += data->len;
+ }
+
+ poa_mgmt_msg_ser(msg, code, s_eid, d_eid, qs, response,
+ data != NULL ? data->len : 0);
+
+ POA_STAT_BUMP(poa, mgmt_tx);
+
+ return poa->ops->poa_send_mgmt(poa, dst, buf, len);
+}
+
+/* Caller holds poas.lock for writing. */
+static struct poa_flow * pf_create(struct poa * poa,
+ const struct poa_addr * r_addr)
+{
+ struct poa_flow * pf;
+ pthread_condattr_t cattr;
+ int eid;
+
+ pf = malloc(sizeof(*pf));
+ if (pf == NULL)
+ goto fail_malloc;
+
+ memset(pf, 0, sizeof(*pf));
+
+ if (pthread_mutex_init(&pf->mtx, NULL) != 0)
+ goto fail_mtx;
+
+ if (pthread_condattr_init(&cattr) != 0)
+ goto fail_cond;
+#ifndef __APPLE__
+ pthread_condattr_setclock(&cattr, PTHREAD_COND_CLOCK);
+#endif
+ if (pthread_cond_init(&pf->cond, &cattr) != 0) {
+ pthread_condattr_destroy(&cattr);
+ goto fail_cond;
+ }
+
+ pthread_condattr_destroy(&cattr);
+
+ eid = bmp_allocate(poa->eids);
+ if (!bmp_is_id_valid(poa->eids, eid))
+ goto fail_eid;
+
+ pf->poa = poa;
+ pf->eid = (uint32_t) eid;
+ pf->flow_id = -1;
+ pf->state = POA_FLOW_PENDING;
+ pf->r_addr = *r_addr;
+
+ clock_gettime(PTHREAD_COND_CLOCK, &pf->t0);
+
+ rcu_wrlock(&poa->guard);
+
+ rcu_publish(pf);
+ rcu_assign(poa->eid_to_pf[eid], pf);
+
+ rcu_wrunlock(&poa->guard);
+
+ list_add_tail(&pf->pend, &poa->flows);
+
+ POA_STAT_BUMP(poa, n_flows);
+
+ return pf;
+
+ fail_eid:
+ pthread_cond_destroy(&pf->cond);
+ fail_cond:
+ pthread_mutex_destroy(&pf->mtx);
+ fail_mtx:
+ free(pf);
+ fail_malloc:
+ return NULL;
+}
+
+/* Caller holds poas.mtx. */
+static void deferred_purge(int flow_id)
+{
+ struct list_head * p;
+ struct list_head * h;
+
+ list_for_each_safe(p, h, &poas.deferred.list) {
+ struct poa_deferred * d;
+
+ d = list_entry(p, struct poa_deferred, next);
+ if (d->flow_id != flow_id)
+ continue;
+
+ llist_del(&d->next, &poas.deferred);
+ freebuf(d->data);
+ free(d);
+ }
+}
+
+/* The id may have moved to a newer flow; only its owner clears it. */
+static void pf_destroy(struct poa_flow * pf)
+{
+ struct poa * poa = pf->poa;
+
+ pthread_rwlock_wrlock(&poas.lock);
+
+ rcu_wrlock(&poa->guard);
+
+ rcu_assign(poa->eid_to_pf[pf->eid], NULL);
+ rcu_assign(pf->rx_rb, NULL);
+
+ rcu_wrunlock(&poa->guard);
+
+ bmp_release(poa->eids, pf->eid);
+
+ if (!list_is_empty(&pf->pend)) {
+ list_del(&pf->pend);
+ POA_STAT_SUB(poa, n_flows, 1);
+ }
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ pthread_mutex_lock(&poas.mtx);
+
+ if (pf->flow_id >= 0 && poas.id_to_pf[pf->flow_id] == pf) {
+ poas.id_to_pf[pf->flow_id] = NULL;
+
+ deferred_purge(pf->flow_id);
+ }
+
+ pthread_mutex_unlock(&poas.mtx);
+
+ rcu_reclaim(&poa->guard); /* a receive may still hold pf */
+
+ freebuf(pf->resp_data);
+ freebuf(pf->answer_data);
+
+ pthread_cond_destroy(&pf->cond);
+ pthread_mutex_destroy(&pf->mtx);
+
+ free(pf);
+}
+
+/*
+ * Keeps the answer for a repeat, then sends it. An accept must wait
+ * until the flow can receive; a refusal needs no receiver.
+ */
+static int pf_answer(struct poa_flow * pf,
+ int response,
+ const buffer_t * data)
+{
+ int err;
+
+ pthread_rwlock_wrlock(&poas.lock);
+
+ freebuf(pf->answer_data);
+
+ if (data != NULL && data->len > 0) {
+ pf->answer_data.data = malloc(data->len);
+ if (pf->answer_data.data != NULL) {
+ memcpy(pf->answer_data.data, data->data, data->len);
+
+ pf->answer_data.len = data->len;
+ }
+ }
+
+ pf->answer = response;
+ pf->answered = true;
+ if (response == 0 && pf->state != POA_FLOW_ALLOCATED) {
+ pthread_rwlock_unlock(&poas.lock);
+ return 0;
+ }
+
+ pf->answer_sent = true;
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ err = mgmt_send(pf->poa, &pf->r_addr, POA_FLOW_REPLY, pf->eid,
+ pf->r_eid, qos_raw, response, data);
+
+ if (err == -ETIMEDOUT || err == -EAGAIN)
+ err = 0; /* stored; a repeat request resends it */
+
+ return err;
+}
+
+/* Takes an answer left for a flow that had no id yet. */
+static struct poa_deferred * deferred_take(int flow_id)
+{
+ struct list_head * p;
+ struct list_head * h;
+
+ list_for_each_safe(p, h, &poas.deferred.list) {
+ struct poa_deferred * d;
+
+ d = list_entry(p, struct poa_deferred, next);
+ if (d->flow_id != flow_id)
+ continue;
+
+ llist_del(&d->next, &poas.deferred);
+
+ return d;
+ }
+
+ return NULL;
+}
+
+/* Publishes the flow_id so an answer can find this flow. */
+static void pf_set_flow_id(struct poa_flow * pf,
+ int flow_id)
+{
+ struct poa_deferred * d;
+
+ pthread_mutex_lock(&poas.mtx);
+
+ pf->flow_id = flow_id;
+ poas.id_to_pf[flow_id] = pf;
+
+ d = deferred_take(flow_id);
+
+ pthread_cond_broadcast(&poas.cond);
+
+ pthread_mutex_unlock(&poas.mtx);
+
+ if (d != NULL) {
+ pf_answer(pf, d->response, &d->data);
+ freebuf(d->data);
+ free(d);
+ }
+}
+
+/*
+ * Between the request arriving and the accept returning, the flow has
+ * an id but no fd yet; flow_init claims the PoA here.
+ */
+static void pf_set_pending(struct poa_flow * pf)
+{
+ pthread_mutex_lock(&poas.mtx);
+
+ pf->pending = true;
+
+ pthread_mutex_unlock(&poas.mtx);
+}
+
+static void pf_clr_pending(struct poa_flow * pf)
+{
+ pthread_mutex_lock(&poas.mtx);
+
+ pf->pending = false;
+
+ pthread_mutex_unlock(&poas.mtx);
+}
+
+/* A process that attaches no PoA has nothing pending. */
+struct poa_flow * poa_flow_take_pending(int flow_id)
+{
+ struct poa_flow * pf;
+
+ if (poas.state == POA_NULL)
+ return NULL;
+
+ if (flow_id < 0 || flow_id >= SYS_MAX_FLOWS)
+ return NULL;
+
+ pthread_mutex_lock(&poas.mtx);
+
+ pf = poas.id_to_pf[flow_id];
+ if (pf != NULL && pf->pending)
+ pf->pending = false;
+ else
+ pf = NULL;
+
+ pthread_mutex_unlock(&poas.mtx);
+
+ return pf;
+}
+
+static struct poa_flow * pf_get(int flow_id)
+{
+ struct poa_flow * pf;
+
+ if (flow_id < 0 || flow_id >= SYS_MAX_FLOWS)
+ return NULL;
+
+ pthread_mutex_lock(&poas.mtx);
+
+ pf = poas.id_to_pf[flow_id];
+
+ pthread_mutex_unlock(&poas.mtx);
+
+ return pf;
+}
+
+#ifdef PROC_FLOW_STATS
+
+/* Caller holds poas.lock. */
+static struct poa * poa_by_rib_name(const char * name)
+{
+ struct list_head * p;
+
+ list_for_each(p, &poas.list) {
+ struct poa * poa = list_entry(p, struct poa, next);
+
+ if (strcmp(poa->name, name) == 0)
+ return poa;
+ }
+
+ return NULL;
+}
+
+static int poa_rib_read(const char * path,
+ char * buf,
+ size_t len)
+{
+ struct poa * poa;
+ const char * entry;
+ size_t qlen;
+ size_t avg;
+ size_t cost;
+ int size;
+ int ret;
+
+ entry = strstr(path, RIB_SEPARATOR) + 1;
+
+ if (len < POA_RIB_STRLEN)
+ return -1;
+
+ pthread_rwlock_rdlock(&poas.lock);
+
+ poa = poa_by_rib_name(entry);
+ if (poa == NULL)
+ goto fail;
+
+ qlen = poa->ops->poa_qlen(poa);
+ avg = poa->avg_len;
+ cost = poa->avg_len + poa->avg_ovh;
+
+ size = snprintf(buf, len,
+ "Active flows: %zu\n"
+ "Packets received: %zu\n"
+ "Bytes received: %zu\n"
+ "Packets sent: %zu\n"
+ "Bytes sent: %zu\n"
+ "Management frames rcvd: %zu\n"
+ "Management frames sent: %zu\n"
+ "Bad EID packets: %zu\n"
+ "Delivery (N+1) failures: %zu\n"
+ "Buffer alloc failures: %zu\n"
+ "Packet read failures: %zu\n"
+ "Packet send failures: %zu\n"
+ "Name queries sent: %zu\n"
+ "Name queries received: %zu\n"
+ "Name replies sent: %zu\n"
+ "Name replies received: %zu\n"
+ "Queued (transport): %zu\n"
+ "Queued (packets): %zu\n"
+ "Mean packet size (bytes): %zu\n"
+ "Mean packet cost: %zu\n",
+ POA_STAT_LOAD(poa, n_flows),
+ POA_STAT_LOAD(poa, rx_pkts),
+ POA_STAT_LOAD(poa, rx_bytes),
+ POA_STAT_LOAD(poa, tx_pkts),
+ POA_STAT_LOAD(poa, tx_bytes),
+ POA_STAT_LOAD(poa, mgmt_rx),
+ POA_STAT_LOAD(poa, mgmt_tx),
+ POA_STAT_LOAD(poa, bad_eid),
+ POA_STAT_LOAD(poa, dlv_fail),
+ POA_STAT_LOAD(poa, buf_fail),
+ POA_STAT_LOAD(poa, rcv_fail),
+ POA_STAT_LOAD(poa, snd_fail),
+ POA_STAT_LOAD(poa, qry_tx),
+ POA_STAT_LOAD(poa, qry_rx),
+ POA_STAT_LOAD(poa, rep_tx),
+ POA_STAT_LOAD(poa, rep_rx),
+ qlen, cost > 0 ? qlen / cost : 0, avg, cost);
+ if (size < 0 || (size_t) size >= len)
+ goto fail;
+
+ if (poa->ops->poa_rib != NULL) {
+ ret = poa->ops->poa_rib(poa, buf + size, len - size);
+ if (ret < 0)
+ goto fail;
+
+ size += ret;
+ }
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ return size;
+
+ fail:
+ pthread_rwlock_unlock(&poas.lock);
+
+ return -1;
+}
+
+static int poa_rib_readdir(char *** buf)
+{
+ struct list_head * p;
+ size_t n = 0;
+ int idx = 0;
+
+ pthread_rwlock_rdlock(&poas.lock);
+
+ list_for_each(p, &poas.list)
+ ++n;
+
+ if (n == 0) {
+ *buf = NULL;
+ goto no_poas;
+ }
+
+ *buf = malloc(sizeof(**buf) * n);
+ if (*buf == NULL)
+ goto fail_entries;
+
+ list_for_each(p, &poas.list) {
+ struct poa * poa = list_entry(p, struct poa, next);
+
+ (*buf)[idx] = malloc(strlen(poa->name) + 1);
+ if ((*buf)[idx] == NULL)
+ goto fail_entry;
+
+ strcpy((*buf)[idx++], poa->name);
+ }
+ no_poas:
+ pthread_rwlock_unlock(&poas.lock);
+
+ return idx;
+
+ fail_entry:
+ while (idx-- > 0)
+ free((*buf)[idx]);
+
+ free(*buf);
+ fail_entries:
+ pthread_rwlock_unlock(&poas.lock);
+
+ return -ENOMEM;
+}
+
+static int poa_rib_getattr(const char * path,
+ struct rib_attr * attr)
+{
+ (void) path;
+
+ attr->size = POA_RIB_STRLEN;
+ attr->mtime = 0;
+
+ return 0;
+}
+
+static struct rib_ops poa_r_ops = {
+ .read = poa_rib_read,
+ .readdir = poa_rib_readdir,
+ .getattr = poa_rib_getattr
+};
+
+#endif /* PROC_FLOW_STATS */
+
+int poa_init(const char * name)
+{
+ pthread_condattr_t cattr;
+
+ assert(name != NULL);
+
+ memset(&poas, 0, sizeof(poas));
+
+ str_hash(HASH_SHA3_256, poas.name_hash, name);
+
+ poas.mon_fd = -1;
+ if (pthread_mutex_init(&poas.mtx, NULL) != 0)
+ goto fail_mtx;
+
+ if (pthread_cond_init(&poas.cond, NULL) != 0)
+ goto fail_cond;
+
+ if (pthread_mutex_init(&poas.mgmt_mtx, NULL) != 0)
+ goto fail_mgmt_mtx;
+
+ if (pthread_condattr_init(&cattr) != 0)
+ goto fail_cattr;
+#ifndef __APPLE__
+ pthread_condattr_setclock(&cattr, PTHREAD_COND_CLOCK);
+#endif
+ if (pthread_cond_init(&poas.mgmt_cond, &cattr) != 0) {
+ pthread_condattr_destroy(&cattr);
+ goto fail_cattr;
+ }
+
+ pthread_condattr_destroy(&cattr);
+
+ if (pthread_cond_init(&poas.done_cond, NULL) != 0)
+ goto fail_done_cond;
+
+ if (pthread_rwlock_init(&poas.lock, NULL) != 0)
+ goto fail_lock;
+
+ list_head_init(&poas.list);
+ llist_init(&poas.deferred);
+ llist_init(&poas.mgmt_frames);
+ list_head_init(&poas.detach);
+
+ poas.state = POA_INIT;
+
+#ifdef PROC_FLOW_STATS
+ if (rib_reg(POA_RIB, &poa_r_ops) < 0)
+ goto fail_rib;
+#endif
+ return 0;
+
+#ifdef PROC_FLOW_STATS
+ fail_rib:
+ pthread_rwlock_destroy(&poas.lock);
+#endif
+
+ fail_lock:
+ pthread_cond_destroy(&poas.done_cond);
+ fail_done_cond:
+ pthread_cond_destroy(&poas.mgmt_cond);
+ fail_cattr:
+ pthread_mutex_destroy(&poas.mgmt_mtx);
+ fail_mgmt_mtx:
+ pthread_cond_destroy(&poas.cond);
+ fail_cond:
+ pthread_mutex_destroy(&poas.mtx);
+ fail_mtx:
+ return -1;
+}
+
+int poa_set_layer(const char * layer)
+{
+ if (layer == NULL)
+ return -EINVAL;
+
+ pthread_rwlock_wrlock(&poas.lock);
+
+ str_hash(HASH_SHA3_256, poas.layer_hash, layer);
+
+ poas.state = POA_OPERATIONAL;
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ return 0;
+}
+
+bool poa_has_name(const uint8_t * hash)
+{
+ bool match = false;
+
+ pthread_rwlock_rdlock(&poas.lock);
+
+ if (poas.state >= POA_INIT)
+ match = memcmp(hash, poas.name_hash, POA_QUERY_HLEN) == 0;
+
+ if (!match && poas.state >= POA_OPERATIONAL)
+ match = memcmp(hash, poas.layer_hash, POA_QUERY_HLEN) == 0;
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ return match;
+}
+
+int poa_spb_reserve(struct ssm_pk_buff ** spb,
+ size_t len)
+{
+ return ssm_pool_alloc_b(proc.pool, len, NULL, spb, NULL) < 0 ? -1 : 0;
+}
+
+void poa_spb_release(struct ssm_pk_buff * spb)
+{
+ ssm_pool_remove(proc.pool, ssm_pk_buff_get_off(spb));
+}
+
+void poa_rx_pkt(struct poa * poa,
+ uint32_t eid,
+ struct ssm_pk_buff * spb)
+{
+ struct poa_flow * pf;
+ struct ssm_rbuff * rx_rb;
+ size_t len;
+
+ len = ssm_pk_buff_len(spb); /* the ring write takes it over */
+
+ if (eid >= poa->n_eids) {
+ POA_STAT_BUMP(poa, bad_eid);
+ poa_spb_release(spb);
+ return;
+ }
+
+ rcu_rdlock(&poa->guard);
+
+ pf = rcu_deref(poa->eid_to_pf[eid]);
+
+ rcu_consume(pf);
+
+ if (pf == NULL)
+ goto fail_eid;
+
+ rx_rb = rcu_deref(pf->rx_rb);
+
+ rcu_consume(rx_rb);
+
+ if (rx_rb == NULL)
+ goto fail_eid;
+
+ if (ssm_rbuff_write(rx_rb, ssm_pk_buff_get_off(spb)) < 0) {
+ POA_STAT_BUMP(poa, dlv_fail);
+ rcu_rdunlock(&poa->guard);
+ poa_spb_release(spb);
+ return;
+ }
+
+ POA_STAT_BUMP(poa, rx_pkts);
+ POA_STAT_ADD(poa, rx_bytes, len);
+
+ ssm_flow_set_notify(proc.fqset, pf->flow_id, FLOW_PKT);
+
+ rcu_rdunlock(&poa->guard);
+
+ return;
+
+ fail_eid:
+ POA_STAT_BUMP(poa, bad_eid);
+
+ rcu_rdunlock(&poa->guard);
+
+ poa_spb_release(spb);
+}
+
+static int poa_flow_req_arr(struct poa_flow * pf,
+ qosspec_t qs,
+ uint32_t mtu,
+ const buffer_t * data)
+{
+ struct flow_info flow;
+ uint8_t buf[SOCK_BUF_SIZE];
+ buffer_t msg = {SOCK_BUF_SIZE, buf};
+ buffer_t out = BUF_INIT;
+ int err;
+
+ memset(&flow, 0, sizeof(flow));
+
+ flow.n_pid = getpid();
+ flow.n_1_pid = getpid();
+ flow.qs = qs;
+ flow.mpl = pf->poa->mpl;
+ flow.mtu = mtu;
+ if (ipcp_poa_flow_req_arr__irm_req_ser(&msg, &flow, data) < 0)
+ return -ENOMEM;
+
+ pf_set_pending(pf);
+
+ err = send_recv_msg(&msg);
+ if (err < 0)
+ goto fail;
+
+ err = poa_flow__irm_result_des(&msg, &flow, &out);
+ if (err < 0)
+ goto fail;
+
+ freebuf(out);
+
+ if (flow.id < 0 || flow.id >= SYS_MAX_FLOWS) {
+ err = -EBADF;
+ goto fail;
+ }
+
+ pf_set_flow_id(pf, flow.id);
+
+ return 0;
+ fail:
+ pf_clr_pending(pf);
+ return err;
+}
+
+static void handle_flow_req(struct poa * poa,
+ const struct poa_addr * src,
+ const struct poa_mgmt_msg * msg,
+ const uint8_t * data,
+ size_t data_len)
+{
+ struct list_head * p;
+ struct poa_flow * pf = NULL;
+ qosspec_t qs;
+ buffer_t buf;
+ buffer_t answer_data = BUF_INIT;
+ uint32_t r_eid;
+ uint32_t mtu;
+ uint32_t eid = 0;
+ int answer = 0;
+ bool found = false;
+ bool served = false;
+
+ r_eid = ntoh32(msg->s_eid);
+
+ poa_mgmt_msg_qos(msg, &qs);
+
+ pthread_rwlock_wrlock(&poas.lock);
+
+ list_for_each(p, &poa->flows) {
+ pf = list_entry(p, struct poa_flow, pend);
+ if (pf->r_eid != r_eid || poa_addr_cmp(&pf->r_addr, src) != 0)
+ continue;
+
+ answer = pf->answer;
+ eid = pf->eid;
+ served = true;
+
+ found = pf->answer_sent;
+ if (found && pf->answer_data.len > 0) {
+ answer_data.data = malloc(pf->answer_data.len);
+ if (answer_data.data != NULL) {
+ memcpy(answer_data.data, pf->answer_data.data,
+ pf->answer_data.len);
+ answer_data.len = pf->answer_data.len;
+ }
+ }
+ break;
+ }
+
+ if (!served) {
+ pf = pf_create(poa, src);
+ if (pf != NULL)
+ pf->r_eid = r_eid;
+ }
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ if (found) {
+ mgmt_send(poa, src, POA_FLOW_REPLY, eid, r_eid, qos_raw, answer,
+ answer_data.len > 0 ? &answer_data : NULL);
+ freebuf(answer_data);
+ }
+
+ if (served || pf == NULL)
+ return;
+
+ buf.len = data_len;
+ buf.data = (uint8_t *) data;
+ mtu = poa->ops->poa_mtu(poa, src);
+ if (poa_flow_req_arr(pf, qs, mtu, &buf) < 0) {
+ mgmt_send(poa, src, POA_FLOW_REPLY, pf->eid, r_eid, qos_raw,
+ -1, NULL);
+ pf_destroy(pf);
+ }
+}
+
+static void handle_flow_reply(struct poa * poa,
+ const struct poa_addr * src,
+ const struct poa_mgmt_msg * msg,
+ const uint8_t * data,
+ size_t data_len)
+{
+ struct poa_flow * pf;
+ uint32_t eid;
+
+ eid = ntoh32(msg->d_eid);
+
+ pthread_rwlock_rdlock(&poas.lock);
+
+ pf = eid < poa->n_eids ? poa->eid_to_pf[eid] : NULL;
+ if (pf == NULL || pf->state != POA_FLOW_PENDING) {
+ pthread_rwlock_unlock(&poas.lock);
+ return;
+ }
+
+ if (poa_addr_cmp(&pf->r_addr, src) != 0) {
+ pthread_rwlock_unlock(&poas.lock);
+ return;
+ }
+
+ pthread_mutex_lock(&pf->mtx);
+
+ if (pf->replied) {
+ pthread_mutex_unlock(&pf->mtx);
+ pthread_rwlock_unlock(&poas.lock);
+ return;
+ }
+
+ if (data_len > 0) {
+ pf->resp_data.data = malloc(data_len);
+ if (pf->resp_data.data != NULL) {
+ memcpy(pf->resp_data.data, data, data_len);
+
+ pf->resp_data.len = data_len;
+ }
+ }
+
+ pf->r_eid = ntoh32(msg->s_eid);
+ pf->response = ntoh32(msg->response);
+ pf->replied = true;
+
+ pthread_cond_broadcast(&pf->cond);
+
+ pthread_mutex_unlock(&pf->mtx);
+
+ pthread_rwlock_unlock(&poas.lock);
+}
+
+static void handle_flow_update(struct poa * poa,
+ const struct poa_addr * src,
+ const struct poa_mgmt_msg * msg,
+ const uint8_t * data,
+ size_t data_len)
+{
+ struct poa_flow * pf;
+ buffer_t buf;
+ uint32_t eid;
+
+ eid = ntoh32(msg->d_eid);
+
+ pthread_rwlock_rdlock(&poas.lock);
+
+ pf = eid < poa->n_eids ? poa->eid_to_pf[eid] : NULL;
+ if (pf == NULL || pf->state != POA_FLOW_ALLOCATED) {
+ pthread_rwlock_unlock(&poas.lock);
+ return;
+ }
+
+ if (poa_addr_cmp(&pf->r_addr, src) != 0) {
+ pthread_rwlock_unlock(&poas.lock);
+ return;
+ }
+
+ eid = (uint32_t) pf->flow_id;
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ buf.len = data_len;
+ buf.data = (uint8_t *) data;
+
+ ipcp_flow_update_arr((int) eid, &buf);
+}
+
+static void mgmt_frame_handle(struct poa_mgmt_frame * frame)
+{
+ const struct poa_mgmt_msg * msg;
+ const uint8_t * data;
+ size_t data_len;
+
+ msg = (const struct poa_mgmt_msg *) frame->buf;
+ if (frame->len < sizeof(*msg))
+ return;
+
+ data_len = ntoh16(msg->data_len);
+ if (data_len > frame->len - sizeof(*msg))
+ return;
+
+ data = frame->buf + sizeof(*msg);
+
+ switch (msg->code) {
+ case POA_FLOW_REQ:
+ handle_flow_req(frame->poa, &frame->src, msg, data, data_len);
+ break;
+ case POA_FLOW_REPLY:
+ handle_flow_reply(frame->poa, &frame->src, msg, data, data_len);
+ break;
+ case POA_FLOW_UPDATE:
+ handle_flow_update(frame->poa, &frame->src, msg, data,
+ data_len);
+ break;
+ default:
+ break;
+ }
+}
+
+static bool pf_steal(struct poa_flow * pf)
+{
+ bool stolen = false;
+
+ pthread_mutex_lock(&poas.mtx);
+
+ if (pf->pending) {
+ pf->pending = false;
+ poas.id_to_pf[pf->flow_id] = NULL;
+
+ deferred_purge(pf->flow_id);
+
+ stolen = true;
+ }
+
+ pthread_mutex_unlock(&poas.mtx);
+
+ return stolen;
+}
+
+static void sweep_pending(void)
+{
+ struct poa_flow * dead[POA_SWEEP_MAX];
+ struct list_head * p;
+ struct list_head * q;
+ struct timespec now;
+ size_t n = 0;
+ size_t i;
+
+ clock_gettime(PTHREAD_COND_CLOCK, &now);
+
+ pthread_rwlock_rdlock(&poas.lock);
+
+ list_for_each(p, &poas.list) {
+ struct poa * poa = list_entry(p, struct poa, next);
+
+ list_for_each(q, &poa->flows) {
+ struct poa_flow * pf;
+
+ if (n == POA_SWEEP_MAX)
+ break;
+
+ pf = list_entry(q, struct poa_flow, pend);
+ if (pf->state != POA_FLOW_PENDING || pf->claimed)
+ continue;
+
+ if (now.tv_sec - pf->t0.tv_sec < POA_PEND_TIMEO)
+ continue;
+
+ if (pf->flow_id >= 0 && !pf_steal(pf))
+ continue;
+
+ dead[n++] = pf;
+ }
+
+ if (n == POA_SWEEP_MAX)
+ break;
+ }
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ for (i = 0; i < n; ++i)
+ pf_destroy(dead[i]);
+
+ pthread_mutex_lock(&poas.mtx);
+
+ list_for_each_safe(p, q, &poas.deferred.list) {
+ struct poa_deferred * d;
+
+ d = list_entry(p, struct poa_deferred, next);
+ if (now.tv_sec - d->t0.tv_sec < POA_PEND_TIMEO)
+ continue;
+
+ llist_del(&d->next, &poas.deferred);
+ freebuf(d->data);
+ free(d);
+ }
+
+ pthread_mutex_unlock(&poas.mtx);
+}
+
+void poa_rx_mgmt(struct poa * poa,
+ const struct poa_addr * src,
+ const uint8_t * buf,
+ size_t len)
+{
+ struct poa_mgmt_frame * frame;
+
+ if (len < sizeof(struct poa_mgmt_msg) || len > POA_MGMT_BUFSZ)
+ return;
+
+ POA_STAT_BUMP(poa, mgmt_rx);
+
+ frame = malloc(offsetof(struct poa_mgmt_frame, buf) + len);
+ if (frame == NULL)
+ return;
+
+ frame->poa = poa;
+ frame->src = *src;
+ frame->len = len;
+
+ memcpy(frame->buf, buf, len);
+
+ pthread_mutex_lock(&poas.mgmt_mtx);
+
+ if (poa->n_mgmt >= POA_MGMT_QMAX) {
+ pthread_mutex_unlock(&poas.mgmt_mtx);
+ free(frame);
+ return;
+ }
+
+ ++poa->n_mgmt;
+
+ llist_add_tail(&frame->next, &poas.mgmt_frames);
+
+ pthread_cond_signal(&poas.mgmt_cond);
+
+ pthread_mutex_unlock(&poas.mgmt_mtx);
+}
+
+static void mgmt_frames_purge(const struct poa * poa)
+{
+ struct list_head * p;
+ struct list_head * h;
+
+ pthread_mutex_lock(&poas.mgmt_mtx);
+
+ list_for_each_safe(p, h, &poas.mgmt_frames.list) {
+ struct poa_mgmt_frame * frame;
+
+ frame = list_entry(p, struct poa_mgmt_frame, next);
+ if (frame->poa != poa)
+ continue;
+
+ --frame->poa->n_mgmt;
+
+ llist_del(&frame->next, &poas.mgmt_frames);
+
+ free(frame);
+ }
+
+ pthread_mutex_unlock(&poas.mgmt_mtx);
+}
+
+static bool poa_has_id(const struct poa * poa,
+ const struct poa_spec * spec)
+{
+ if (poa->type != spec->type)
+ return false;
+
+ return poa->ops->poa_has_id(poa, spec);
+}
+
+/*
+ * The PoA carrying dst, as the transport judges it. -EPERM if none
+ * matches, -EINVAL if several do. Caller holds poas.lock.
+ */
+static int poa_lookup(const struct poa_addr * dst,
+ struct poa ** out)
+{
+ struct list_head * p;
+ struct poa * found = NULL;
+
+ list_for_each(p, &poas.list) {
+ struct poa * poa = list_entry(p, struct poa, next);
+
+ if (poa->type != dst->type)
+ continue;
+
+ if (!poa->ops->poa_match(poa, dst))
+ continue;
+
+ if (found != NULL) /* nothing given, two candidates */
+ return -EINVAL;
+
+ found = poa;
+ }
+
+ if (found == NULL)
+ return -EPERM;
+
+ *out = found;
+
+ return 0;
+}
+
+/* Sends are deadlined, bounding the lock hold on a full queue. */
+int poa_bcast_mgmt(const struct poa_addr * dst,
+ const uint8_t * buf,
+ size_t len)
+{
+ struct list_head * p;
+ int n = 0;
+
+ pthread_rwlock_rdlock(&poas.lock);
+
+ list_for_each(p, &poas.list) {
+ struct poa * poa = list_entry(p, struct poa, next);
+
+ if (poa->type != dst->type)
+ continue;
+
+ if (!poa->ops->poa_match(poa, dst))
+ continue;
+
+ if (poa->ops->poa_send_mgmt(poa, dst, buf, len) < 0)
+ continue;
+
+ /* All management broadcasts are name queries. */
+ POA_STAT_BUMP(poa, qry_tx);
+ ++n;
+ }
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ return n;
+}
+
+static bool deadline_is_malformed(const struct timespec * timeo)
+{
+ if (timeo == NULL)
+ return false;
+
+ if (timeo->tv_sec < 0 || timeo->tv_nsec < 0)
+ return true;
+
+ return timeo->tv_nsec >= BILLION;
+}
+
+/*
+ * Complete addr for dst on any backend that can query. The ops are
+ * collected under poas.lock but called outside it: a query blocks up
+ * to its deadline and takes the lock again to broadcast. The tables
+ * are static, so nothing dangles; a struct poa cannot be carried
+ * across the unlock. The deadline applies per backend.
+ */
+int poa_query(const char * dst,
+ const struct timespec * timeo,
+ struct poa_addr * addr)
+{
+ const struct poa_ops * cand[POA_MAX_POAS];
+ enum poa_type type[POA_MAX_POAS];
+ struct list_head * p;
+ size_t n = 0;
+ size_t i;
+ int err = -ENOTSUP;
+
+ if (dst == NULL || addr == NULL)
+ return -EINVAL;
+
+ if (deadline_is_malformed(timeo))
+ return -EINVAL;
+
+ pthread_rwlock_rdlock(&poas.lock);
+
+ list_for_each(p, &poas.list) {
+ struct poa * poa = list_entry(p, struct poa, next);
+
+ if (poa->ops->poa_query == NULL)
+ continue;
+
+ for (i = 0; i < n && cand[i] != poa->ops; i++)
+ ;
+ if (i < n)
+ continue;
+
+ /* One type per backend: eth. Revisit if that changes. */
+ cand[n] = poa->ops;
+ type[n++] = poa->type;
+ }
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ for (i = 0; i < n; i++) {
+ memset(addr, 0, sizeof(*addr));
+
+ addr->type = type[i];
+
+ err = cand[i]->poa_query(dst, timeo, addr);
+ if (err == 0)
+ return 0;
+ }
+
+ return err;
+}
+
+static int poa_check(const struct poa_addr * dst)
+{
+ struct poa * poa;
+ int err;
+
+ pthread_rwlock_rdlock(&poas.lock);
+
+ err = poa_lookup(dst, &poa);
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ return err;
+}
+
+static int poa_alloc(const struct poa_addr * dst,
+ qosspec_t qs,
+ const buffer_t * req,
+ buffer_t * resp,
+ struct poa_flow ** pf_out,
+ uint32_t * mtu,
+ const struct timespec * timeo)
+{
+ struct timespec dflt = TIMESPEC_INIT_MS(POA_ALLOC_TIMEO);
+ struct timespec rintv = TIMESPEC_INIT_MS(POA_RETRY_TIMEO);
+ struct poa_flow * pf;
+ struct poa * poa;
+ struct timespec abstime;
+ struct timespec now;
+ struct timespec retry;
+ int err;
+
+ pthread_rwlock_wrlock(&poas.lock);
+
+ err = poa_lookup(dst, &poa);
+ if (err < 0) {
+ pthread_rwlock_unlock(&poas.lock);
+ return err;
+ }
+
+ pf = pf_create(poa, dst);
+ if (pf == NULL) {
+ pthread_rwlock_unlock(&poas.lock);
+ return -ENOMEM;
+ }
+
+ pf->claimed = true;
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ clock_gettime(PTHREAD_COND_CLOCK, &abstime);
+ ts_add(&abstime, timeo != NULL ? timeo : &dflt, &abstime);
+
+ pthread_mutex_lock(&pf->mtx);
+
+ while (!pf->replied) {
+ pthread_mutex_unlock(&pf->mtx);
+
+ err = mgmt_send(poa, dst, POA_FLOW_REQ, pf->eid, 0, qs, 0, req);
+
+ pthread_mutex_lock(&pf->mtx);
+
+ if (err < 0 && err != -ETIMEDOUT && err != -EAGAIN) {
+ err = -EIO;
+ goto fail;
+ }
+
+ if (pf->replied)
+ break;
+
+ clock_gettime(PTHREAD_COND_CLOCK, &now);
+ ts_add(&now, &rintv, &retry);
+
+ if (ts_diff_ns(&retry, &abstime) > 0)
+ retry = abstime;
+
+ pthread_cond_timedwait(&pf->cond, &pf->mtx, &retry);
+
+ if (pf->replied)
+ break;
+
+ clock_gettime(PTHREAD_COND_CLOCK, &now);
+
+ if (ts_diff_ns(&now, &abstime) >= 0) {
+ err = -ETIMEDOUT;
+ goto fail;
+ }
+ }
+
+ if (pf->response != 0) {
+ err = -ECONNREFUSED;
+ goto fail;
+ }
+
+ *resp = pf->resp_data;
+ pf->resp_data.len = 0;
+ pf->resp_data.data = NULL;
+
+ pthread_mutex_unlock(&pf->mtx);
+
+ *mtu = poa->ops->poa_mtu(poa, dst);
+ *pf_out = pf;
+
+ return 0;
+
+ fail:
+ pthread_mutex_unlock(&pf->mtx);
+ pf_destroy(pf);
+ return err;
+}
+
+static void poa_alloc_fail(struct poa_flow * pf)
+{
+ pf_destroy(pf);
+}
+
+void poa_flow_attach(struct poa_flow * pf,
+ int flow_id,
+ struct ssm_rbuff * rx_rb)
+{
+ struct poa * poa = pf->poa;
+
+ if (pf->flow_id != flow_id)
+ pf_set_flow_id(pf, flow_id);
+
+ pthread_rwlock_wrlock(&poas.lock);
+
+ pf->state = POA_FLOW_ALLOCATED;
+
+ rcu_wrlock(&poa->guard);
+
+ rcu_publish(rx_rb);
+ rcu_assign(pf->rx_rb, rx_rb);
+
+ rcu_wrunlock(&poa->guard);
+
+ pthread_rwlock_unlock(&poas.lock);
+}
+
+void poa_flow_detach(struct poa_flow * pf)
+{
+ struct poa * poa = pf->poa;
+
+ pthread_rwlock_wrlock(&poas.lock);
+
+ pf->state = POA_FLOW_DEAD;
+
+ rcu_wrlock(&poa->guard);
+
+ rcu_assign(pf->rx_rb, NULL);
+
+ rcu_wrunlock(&poa->guard);
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ pf_destroy(pf);
+}
+
+static size_t flows_updown(struct poa * poa,
+ bool up)
+{
+ struct list_head * p;
+ size_t n = 0;
+
+ list_for_each(p, &poa->flows) {
+ struct poa_flow * pf;
+ struct flow * flow;
+
+ pf = list_entry(p, struct poa_flow, pend);
+ if (pf->state != POA_FLOW_ALLOCATED || pf->flow_id < 0)
+ continue;
+
+ flow = &proc.flows[proc.id_to_fd[pf->flow_id].fd];
+ if (flow->info.id != pf->flow_id)
+ continue;
+
+ if (((flow->oflags & FLOWFDOWN) != 0) != !up)
+ ++n;
+
+ if (up) {
+ flow->oflags &= ~FLOWFDOWN;
+
+ ssm_rbuff_clr_bits(flow->rx_rb, RB_FLOWDOWN);
+ } else {
+ flow->oflags |= FLOWFDOWN;
+
+ ssm_rbuff_set_bits(flow->rx_rb, RB_FLOWDOWN);
+ }
+
+ ssm_flow_set_notify(proc.fqset, pf->flow_id,
+ up ? FLOW_UP : FLOW_DOWN);
+ }
+
+ return n;
+}
+
+size_t poa_link_updown(int id,
+ bool up)
+{
+ struct list_head * p;
+ size_t n = 0;
+
+ pthread_rwlock_wrlock(&proc.lock);
+ pthread_rwlock_rdlock(&poas.lock);
+
+ list_for_each(p, &poas.list) {
+ struct poa * poa = list_entry(p, struct poa, next);
+
+ if (poa->ops->poa_link_match == NULL)
+ continue;
+
+ if (!poa->ops->poa_link_match(poa, id))
+ continue;
+
+ n += flows_updown(poa, up);
+ }
+
+ pthread_rwlock_unlock(&poas.lock);
+ pthread_rwlock_unlock(&proc.lock);
+
+ return n;
+}
+
+static size_t poa_ewma(size_t avg,
+ size_t sz)
+{
+ if (avg == 0)
+ return sz;
+
+ avg = avg + (sz >> POA_AVG_SHIFT) - (avg >> POA_AVG_SHIFT);
+
+ return avg == 0 ? 1 : avg;
+}
+
+static void poa_avg_len_update(struct poa * poa,
+ size_t sz)
+{
+ STORE_RELAXED(&poa->avg_len, poa_ewma(LOAD_RELAXED(&poa->avg_len), sz));
+}
+
+static size_t poa_qlen(struct poa * poa)
+{
+ struct timespec now;
+ uint64_t ns;
+ size_t qlen;
+
+ clock_gettime(PTHREAD_COND_CLOCK, &now);
+
+ ns = TS_TO_UINT64(now);
+ if (ns - LOAD_RELAXED(&poa->q_time) < POA_QLEN_GATE)
+ return LOAD_RELAXED(&poa->q_cache);
+
+ qlen = poa->ops->poa_qlen(poa);
+
+ STORE_RELAXED(&poa->q_cache, qlen);
+ STORE_RELAXED(&poa->q_time, ns);
+
+ return qlen;
+}
+
+static void poa_cost_sample(struct poa * poa,
+ size_t before,
+ size_t len)
+{
+ size_t after;
+
+ after = poa->ops->poa_qlen(poa);
+ if (after <= before)
+ return; /* drained; nothing to learn */
+
+ after -= before;
+ if (after < len || after > (len << 2) + 1024)
+ return;
+
+ STORE_RELAXED(&poa->avg_ovh,
+ poa_ewma(LOAD_RELAXED(&poa->avg_ovh), after - len));
+}
+
+int poa_flow_tx(struct poa_flow * pf,
+ struct ssm_pk_buff * spb,
+ bool block,
+ const struct timespec * abstime)
+{
+ struct poa * poa = pf->poa;
+ size_t len = ssm_pk_buff_len(spb);
+ size_t before = 0;
+ bool sample;
+ int ret;
+
+ sample = (LOAD_RELAXED(&poa->n_tx) & POA_COST_MASK) == 0;
+ if (sample)
+ before = poa->ops->poa_qlen(poa);
+
+ ret = poa->ops->poa_send(poa, &pf->r_addr, pf->r_eid, spb, block,
+ abstime);
+ if (ret < 0) { /* the caller releases the buffer */
+ POA_STAT_BUMP(poa, snd_fail);
+ return ret;
+ }
+
+ POA_STAT_BUMP(poa, tx_pkts);
+ POA_STAT_ADD(poa, tx_bytes, len);
+
+ FETCH_ADD_RELAXED(&poa->n_tx, 1);
+
+ poa_avg_len_update(poa, len);
+
+ if (sample)
+ poa_cost_sample(poa, before, len);
+
+ poa_spb_release(spb);
+
+ return 0;
+}
+
+size_t poa_flow_qlen(const struct poa_flow * pf)
+{
+ struct poa * poa = pf->poa;
+ uint64_t bytes;
+ size_t cost;
+ size_t pkts;
+ size_t byts;
+
+ if (poa->ops->poa_qpkts != NULL &&
+ poa->ops->poa_qpkts(poa, &pkts, &byts) == 0)
+ return byts;
+
+ cost = LOAD_RELAXED(&poa->avg_len) + LOAD_RELAXED(&poa->avg_ovh);
+ if (LOAD_RELAXED(&poa->avg_ovh) == 0 || cost == 0)
+ return poa_qlen(poa); /* overstated beats false empty */
+
+ bytes = (uint64_t) poa_qlen(poa) * LOAD_RELAXED(&poa->avg_len);
+
+ return (size_t) (bytes / cost);
+}
+
+size_t poa_flow_qpkts(const struct poa_flow * pf)
+{
+ struct poa * poa = pf->poa;
+ size_t cost;
+ size_t pkts;
+ size_t byts;
+
+ if (poa->ops->poa_qpkts != NULL &&
+ poa->ops->poa_qpkts(poa, &pkts, &byts) == 0)
+ return pkts;
+
+ cost = LOAD_RELAXED(&poa->avg_len) + LOAD_RELAXED(&poa->avg_ovh);
+ if (LOAD_RELAXED(&poa->avg_ovh) == 0 || cost == 0)
+ return 0;
+
+ return poa_qlen(poa) / cost;
+}
+
+size_t poa_flow_mean_len(const struct poa_flow * pf)
+{
+ return LOAD_RELAXED(&pf->poa->avg_len);
+}
+
+int poa_flow_qid(const struct poa_flow * pf)
+{
+ return pf->poa->qid;
+}
+
+void poa_flow_ready(struct poa_flow * pf)
+{
+ buffer_t data;
+ int answer;
+
+ if (pf == NULL)
+ return;
+
+ clrbuf(data);
+
+ pthread_rwlock_wrlock(&poas.lock);
+
+ if (!pf->answered || pf->answer_sent) {
+ pthread_rwlock_unlock(&poas.lock);
+ return;
+ }
+
+ answer = pf->answer;
+
+ if (pf->answer_data.len > 0) {
+ data.data = malloc(pf->answer_data.len);
+ if (data.data != NULL) {
+ memcpy(data.data, pf->answer_data.data,
+ pf->answer_data.len);
+ data.len = pf->answer_data.len;
+ }
+ }
+
+ pf->answer_sent = true;
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ mgmt_send(pf->poa, &pf->r_addr, POA_FLOW_REPLY, pf->eid,
+ pf->r_eid, qos_raw, answer, &data);
+
+ freebuf(data);
+}
+
+int poa_flow_alloc_resp(int flow_id,
+ int response,
+ const buffer_t * data)
+{
+ struct poa_deferred * d;
+ struct poa_flow * pf;
+
+ if (flow_id < 0 || flow_id >= SYS_MAX_FLOWS)
+ return -EPERM;
+
+ pthread_mutex_lock(&poas.mtx);
+
+ pf = poas.id_to_pf[flow_id];
+ if (pf != NULL) {
+ pthread_mutex_unlock(&poas.mtx);
+ return pf_answer(pf, response, data);
+ }
+
+ if (poas.deferred.len >= POA_DEFER_MAX) {
+ pthread_mutex_unlock(&poas.mtx);
+ return -ENOMEM;
+ }
+
+ d = malloc(sizeof(*d));
+ if (d == NULL) {
+ pthread_mutex_unlock(&poas.mtx);
+ return -ENOMEM;
+ }
+
+ memset(d, 0, sizeof(*d));
+
+ clock_gettime(PTHREAD_COND_CLOCK, &d->t0);
+
+ d->flow_id = flow_id;
+ d->response = response;
+
+ if (data != NULL && data->len > 0) {
+ d->data.data = malloc(data->len);
+ if (d->data.data == NULL) {
+ free(d);
+ pthread_mutex_unlock(&poas.mtx);
+ return -ENOMEM;
+ }
+ memcpy(d->data.data, data->data, data->len);
+
+ d->data.len = data->len;
+ }
+
+ llist_add_tail(&d->next, &poas.deferred);
+
+ pthread_mutex_unlock(&poas.mtx);
+
+ return 0;
+}
+
+int poa_flow_update(int flow_id,
+ const buffer_t * data)
+{
+ struct poa_flow * pf;
+
+ pf = pf_get(flow_id);
+ if (pf == NULL)
+ return -EPERM;
+
+ return mgmt_send(pf->poa, &pf->r_addr, POA_FLOW_UPDATE, pf->eid,
+ pf->r_eid, qos_raw, 0, data);
+}
+
+/* The PoA state is released when the flow itself is torn down. */
+int poa_flow_dealloc(int flow_id)
+{
+ (void) flow_id;
+
+ return 0;
+}
+
+/* PoA id 0 = management channel. */
+static struct poa * poa_create(enum poa_type type,
+ const struct poa_ops * ops,
+ size_t n_eids)
+{
+ struct poa * poa;
+
+ poa = malloc(sizeof(*poa));
+ if (poa == NULL)
+ goto fail_malloc;
+
+ memset(poa, 0, sizeof(*poa));
+
+ poa->eid_to_pf = malloc(sizeof(*poa->eid_to_pf) * n_eids);
+ if (poa->eid_to_pf == NULL)
+ goto fail_map;
+
+ memset(poa->eid_to_pf, 0, sizeof(*poa->eid_to_pf) * n_eids);
+
+ poa->eids = bmp_create(n_eids - 1, 1);
+ if (poa->eids == NULL)
+ goto fail_bmp;
+
+ if (rcu_guard_init(&poa->guard) != 0)
+ goto fail_guard;
+
+ list_head_init(&poa->next);
+ list_head_init(&poa->flows);
+
+ poa->type = type;
+ poa->ops = ops;
+ poa->mpl = ops->mpl;
+ poa->n_eids = n_eids;
+ poa->qid = -1;
+
+ return poa;
+
+ fail_guard:
+ bmp_destroy(poa->eids);
+ fail_bmp:
+ free(poa->eid_to_pf);
+ fail_map:
+ free(poa);
+ fail_malloc:
+ return NULL;
+}
+
+static void poa_destroy(struct poa * poa)
+{
+ rcu_guard_fini(&poa->guard);
+
+ bmp_destroy(poa->eids);
+
+ free(poa->eid_to_pf);
+ free(poa);
+}
+
+static void poa_teardown(struct poa * poa)
+{
+ if (poas.state >= POA_RUNNING)
+ poa->ops->poa_stop(poa);
+
+ mgmt_frames_purge(poa);
+
+ poa->ops->poa_detach(poa);
+
+ poa_destroy(poa);
+}
+
+static void poa_detach_all(void)
+{
+ pthread_rwlock_wrlock(&poas.lock);
+
+ while (!list_is_empty(&poas.list)) {
+ struct poa * poa;
+
+ poa = list_first_entry(&poas.list, struct poa, next);
+
+ list_del(&poa->next);
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ poa_teardown(poa);
+
+ pthread_rwlock_wrlock(&poas.lock);
+ }
+
+ pthread_rwlock_unlock(&poas.lock);
+}
+
+static int poa_do_detach(const struct poa_detach_req * req)
+{
+ struct list_head * p;
+ struct poa * found = NULL;
+
+ pthread_rwlock_wrlock(&poas.lock);
+
+ list_for_each(p, &poas.list) {
+ struct poa * poa = list_entry(p, struct poa, next);
+
+ if (!poa_has_id(poa, &req->spec))
+ continue;
+
+ found = poa;
+ break;
+ }
+
+ if (found == NULL) {
+ pthread_rwlock_unlock(&poas.lock);
+ return -ENOENT;
+ }
+
+ if (!list_is_empty(&found->flows)) {
+ pthread_rwlock_unlock(&poas.lock);
+ return -EBUSY;
+ }
+
+ list_del(&found->next);
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ poa_teardown(found);
+
+ return 0;
+}
+
+static int poa_del(const struct poa_spec * spec)
+{
+ struct poa_detach_req req;
+ int cs;
+ int ret;
+
+ memset(&req, 0, sizeof(req));
+
+ req.spec = *spec;
+
+ pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &cs);
+
+ if (poas.state < POA_RUNNING) { /* set before workers run */
+ ret = poa_do_detach(&req);
+ goto out;
+ }
+
+ pthread_mutex_lock(&poas.mgmt_mtx);
+
+ if (poas.mgmt_stop) { /* stopped: poa_fini reaps these */
+ pthread_mutex_unlock(&poas.mgmt_mtx);
+
+ ret = -EBUSY;
+ goto out;
+ }
+
+ list_add_tail(&req.next, &poas.detach);
+
+ pthread_cond_signal(&poas.mgmt_cond);
+
+ while (!req.done)
+ pthread_cond_wait(&poas.done_cond, &poas.mgmt_mtx);
+
+ pthread_mutex_unlock(&poas.mgmt_mtx);
+
+ ret = req.result;
+ out:
+ pthread_setcancelstate(cs, NULL);
+
+ return ret;
+}
+
+static __inline__ bool mgmt_idle(void)
+{
+ if (poas.mgmt_stop)
+ return false;
+
+ if (!llist_is_empty(&poas.mgmt_frames))
+ return false;
+
+ return list_is_empty(&poas.detach);
+}
+
+static void detach_run(void)
+{
+ while (!list_is_empty(&poas.detach)) {
+ struct poa_detach_req * req;
+
+ req = list_first_entry(&poas.detach, struct poa_detach_req,
+ next);
+ list_del(&req->next);
+
+ pthread_mutex_unlock(&poas.mgmt_mtx);
+
+ req->result = poa_do_detach(req);
+
+ pthread_mutex_lock(&poas.mgmt_mtx);
+
+ req->done = true;
+
+ pthread_cond_broadcast(&poas.done_cond);
+ }
+}
+
+static void * mgmt_handler(void * o)
+{
+ struct timespec intv = TIMESPEC_INIT_MS(POA_SWEEP_TIMEO);
+
+ (void) o;
+
+ while (true) {
+ struct poa_mgmt_frame * frame;
+ struct timespec abstime;
+
+ pthread_mutex_lock(&poas.mgmt_mtx);
+
+ detach_run();
+
+ while (mgmt_idle()) {
+ clock_gettime(PTHREAD_COND_CLOCK, &abstime);
+ ts_add(&abstime, &intv, &abstime);
+
+ if (pthread_cond_timedwait(&poas.mgmt_cond,
+ &poas.mgmt_mtx,
+ &abstime) == ETIMEDOUT) {
+ pthread_mutex_unlock(&poas.mgmt_mtx);
+ sweep_pending();
+ pthread_mutex_lock(&poas.mgmt_mtx);
+ }
+ }
+
+ if (poas.mgmt_stop) {
+ detach_run(); /* nobody else serves these */
+
+ pthread_mutex_unlock(&poas.mgmt_mtx);
+ break;
+ }
+
+ if (llist_is_empty(&poas.mgmt_frames)) {
+ pthread_mutex_unlock(&poas.mgmt_mtx);
+ continue;
+ }
+
+ frame = llist_first_entry(&poas.mgmt_frames,
+ struct poa_mgmt_frame, next);
+ llist_del(&frame->next, &poas.mgmt_frames);
+
+ --frame->poa->n_mgmt;
+
+ pthread_mutex_unlock(&poas.mgmt_mtx);
+
+ mgmt_frame_handle(frame);
+
+ free(frame);
+ }
+
+ return (void *) 0;
+}
+
+#ifndef HAVE_ETH
+/* Only the Ethernet transport reports link events. */
+int poa_monitor_open(void)
+{
+ return -1;
+}
+
+void poa_monitor_read(int fd)
+{
+ (void) fd;
+}
+#endif
+
+static void * poa_monitor(void * o)
+{
+ (void) o;
+
+ while (true)
+ poa_monitor_read(poas.mon_fd);
+
+ return (void *) 0;
+}
+
+static int monitor_start(void)
+{
+ poas.mon_fd = poa_monitor_open();
+ if (poas.mon_fd < 0)
+ return 0;
+
+ if (pthread_create(&poas.monitor, NULL, poa_monitor, NULL) == 0)
+ return 0;
+
+ close(poas.mon_fd);
+
+ poas.mon_fd = -1;
+
+ return -1;
+}
+
+static void monitor_stop(void)
+{
+ if (poas.mon_fd < 0)
+ return;
+
+ pthread_cancel(poas.monitor);
+ pthread_join(poas.monitor, NULL);
+
+ close(poas.mon_fd);
+
+ poas.mon_fd = -1;
+}
+
+int poa_start(void)
+{
+ struct list_head * p;
+ struct list_head * h;
+
+ if (poas.state == POA_NULL)
+ return 0;
+
+ if (pthread_create(&poas.mgmt_handler, NULL, mgmt_handler, NULL) != 0)
+ goto fail_mgmt;
+
+ if (monitor_start() < 0)
+ goto fail_monitor;
+
+ pthread_rwlock_rdlock(&poas.lock);
+
+ list_for_each(p, &poas.list) {
+ struct poa * poa = list_entry(p, struct poa, next);
+
+ if (poa->ops->poa_start(poa) < 0)
+ goto fail_reader;
+ }
+
+ poas.state = POA_RUNNING;
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ return 0;
+
+ fail_reader:
+ list_for_each(h, &poas.list) {
+ struct poa * poa = list_entry(h, struct poa, next);
+
+ if (h == p)
+ break;
+
+ poa->ops->poa_stop(poa);
+ }
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ monitor_stop();
+ fail_monitor:
+ pthread_mutex_lock(&poas.mgmt_mtx);
+
+ poas.mgmt_stop = true;
+
+ pthread_cond_broadcast(&poas.mgmt_cond);
+ pthread_mutex_unlock(&poas.mgmt_mtx);
+
+ pthread_join(poas.mgmt_handler, NULL);
+ fail_mgmt:
+ return -1;
+}
+
+void poa_stop(void)
+{
+ struct list_head * p;
+
+ if (poas.state < POA_RUNNING)
+ return;
+
+ pthread_mutex_lock(&poas.mgmt_mtx);
+
+ poas.mgmt_stop = true;
+
+ pthread_cond_broadcast(&poas.mgmt_cond);
+ pthread_mutex_unlock(&poas.mgmt_mtx);
+
+ pthread_join(poas.mgmt_handler, NULL);
+
+ monitor_stop();
+
+ pthread_rwlock_rdlock(&poas.lock);
+
+ list_for_each(p, &poas.list) {
+ struct poa * poa = list_entry(p, struct poa, next);
+
+ poa->ops->poa_stop(poa);
+ }
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ poas.state = POA_INIT;
+}
+
+void poa_fini(void)
+{
+ if (poas.state == POA_NULL)
+ return;
+
+ poa_stop();
+
+ poas.state = POA_NULL;
+
+ poa_detach_all();
+
+#ifdef PROC_FLOW_STATS
+ rib_unreg(POA_RIB);
+#endif
+
+ pthread_rwlock_destroy(&poas.lock);
+ pthread_cond_destroy(&poas.done_cond);
+ pthread_cond_destroy(&poas.mgmt_cond);
+ pthread_mutex_destroy(&poas.mgmt_mtx);
+ pthread_cond_destroy(&poas.cond);
+ pthread_mutex_destroy(&poas.mtx);
+}
+
+/*
+ * Lowest queue id no attached PoA holds; detaching frees it by
+ * leaving the list. Caller holds poas.lock.
+ */
+static int poa_qid_alloc(void)
+{
+ struct list_head * p;
+ bool used[POA_MAX_POAS];
+ int i;
+
+ memset(used, 0, sizeof(used));
+
+ list_for_each(p, &poas.list) {
+ struct poa * poa = list_entry(p, struct poa, next);
+
+ if (poa->qid >= 0 && poa->qid < POA_MAX_POAS)
+ used[poa->qid] = true;
+ }
+
+ for (i = 0; i < POA_MAX_POAS; i++)
+ if (!used[i])
+ return i;
+
+ return -1;
+}
+
+static int poa_add(const struct poa_spec * spec,
+ const struct poa_ops * ops,
+ size_t n_eids)
+{
+ struct list_head * p;
+ struct poa * poa;
+ int err;
+
+ pthread_rwlock_rdlock(&poas.lock);
+
+ list_for_each(p, &poas.list) {
+ poa = list_entry(p, struct poa, next);
+ if (poa_has_id(poa, spec)) {
+ pthread_rwlock_unlock(&poas.lock);
+ return -EPERM;
+ }
+ }
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ poa = poa_create(spec->type, ops, n_eids);
+ if (poa == NULL)
+ return -ENOMEM;
+
+ err = poa->ops->poa_attach(poa, spec);
+ if (err < 0)
+ goto fail_bind;
+
+ err = poa_addr_name(&poa->local, poa->name, sizeof(poa->name));
+ if (err < 0)
+ goto fail_start;
+
+ err = -1;
+
+ pthread_rwlock_wrlock(&poas.lock);
+
+ poa->qid = poa_qid_alloc();
+ if (poa->qid < 0) {
+ pthread_rwlock_unlock(&poas.lock);
+ goto fail_start;
+ }
+
+ if (poas.state >= POA_RUNNING && poa->ops->poa_start(poa) < 0) {
+ pthread_rwlock_unlock(&poas.lock);
+ goto fail_start;
+ }
+
+ list_add_tail(&poa->next, &poas.list);
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ return 0;
+
+ fail_start:
+ poa->ops->poa_detach(poa);
+ fail_bind:
+ poa_destroy(poa);
+ return err;
+}
+
+/* The single place a type is bound to its transport. */
+int poa_attach(const struct poa_spec * poa)
+{
+ if (poa == NULL)
+ return -EINVAL;
+
+ switch (poa->type) {
+ case POA_UDP4:
+ /* FALLTHRU */
+ case POA_UDP6:
+ return poa_add(poa, &udp_poa_ops, POA_MAX_EIDS);
+ case POA_ETH:
+#ifdef HAVE_ETH
+ return poa_add(poa, &eth_poa_ops, POA_MAX_EIDS);
+#else
+ return -ENOTSUP;
+#endif
+ default:
+ return -ENOTSUP;
+ }
+}
+
+int poa_detach(const struct poa_spec * poa)
+{
+ if (poa == NULL)
+ return -EINVAL;
+
+ return poa_del(poa);
+}
+
+ssize_t poa_list(struct poa_spec * specs,
+ size_t max)
+{
+ struct list_head * p;
+ size_t n = 0;
+
+ if (specs == NULL)
+ return -EINVAL;
+
+ pthread_rwlock_rdlock(&poas.lock);
+
+ list_for_each(p, &poas.list) {
+ struct poa * poa = list_entry(p, struct poa, next);
+
+ if (n++ >= max)
+ continue;
+
+ memset(specs, 0, sizeof(*specs));
+
+ poa->ops->poa_spec(poa, specs);
+
+ specs++;
+ }
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ return (ssize_t) n;
+}
+
+/*
+ * Complete peer for dst on the backend serving its type. The ops are
+ * borrowed under poas.lock and called outside it (see poa_query); if
+ * every PoA of the type detaches in between, the query's broadcast
+ * reaches nothing and reports -EPERM, as the lookup would.
+ */
+static int poa_peer_resolve(const char * dst,
+ struct poa_addr * peer)
+{
+ const struct poa_ops * ops = NULL;
+ struct list_head * p;
+
+ pthread_rwlock_rdlock(&poas.lock);
+
+ list_for_each(p, &poas.list) {
+ struct poa * poa = list_entry(p, struct poa, next);
+
+ if (poa->type == peer->type) {
+ ops = poa->ops;
+ break;
+ }
+ }
+
+ pthread_rwlock_unlock(&poas.lock);
+
+ if (ops == NULL) /* nothing could carry the flow */
+ return -EPERM;
+
+ if (ops->poa_query == NULL) /* these addresses arrive complete */
+ return 0;
+
+ return ops->poa_query(dst, NULL, peer);
+}
+
+/*
+ * Three steps: the IRMd creates the flow and prepares the key exchange,
+ * the PoA handshakes with the peer, the IRMd completes the
+ * exchange and hands us the key.
+ */
+int poa_flow_alloc(const char * dst,
+ const struct poa_addr * addr,
+ qosspec_t * qs,
+ const struct timespec * timeo)
+{
+ struct flow_info flow;
+ struct poa_flow * pf;
+ struct poa_addr peer;
+ struct crypt_sk crypt;
+ struct timespec t0;
+ struct timespec t1;
+ uint8_t key[SYMMKEYSZ];
+ uint8_t buf[SOCK_BUF_SIZE];
+ buffer_t msg = {SOCK_BUF_SIZE, buf};
+ buffer_t req;
+ buffer_t resp;
+ uint32_t mtu = 0;
+ int err;
+
+ if (addr == NULL)
+ return -EINVAL;
+
+ if (qs != NULL && qs->service == SVC_STREAM && qs->loss != 0)
+ return -EINVAL;
+
+ peer = *addr;
+
+ err = poa_peer_resolve(dst, &peer);
+ if (err < 0)
+ return err;
+
+ addr = &peer;
+
+ err = poa_check(addr);
+ if (err < 0)
+ return err;
+
+ memset(&flow, 0, sizeof(flow));
+
+ flow.n_pid = getpid();
+ flow.n_1_pid = getpid();
+ flow.qs = qs == NULL ? qos_raw : *qs;
+ if (poa_flow_alloc__irm_req_ser(&msg, &flow, dst) < 0)
+ return -ENOMEM;
+
+ err = send_recv_msg(&msg);
+ if (err < 0)
+ return err;
+
+ clrbuf(req);
+ clrbuf(resp);
+
+ err = poa_flow__irm_result_des(&msg, &flow, &req);
+ if (err < 0)
+ return err;
+
+ clock_gettime(PTHREAD_COND_CLOCK, &t0);
+
+ err = poa_alloc(addr, flow.qs, &req, &resp, &pf, &mtu, timeo);
+
+ freebuf(req);
+
+ if (err < 0)
+ goto fail_alloc;
+
+ clock_gettime(PTHREAD_COND_CLOCK, &t1);
+
+ flow.mtu = mtu;
+ flow.mpl = pf->poa->mpl;
+ msg.len = SOCK_BUF_SIZE;
+ msg.data = buf;
+ if (poa_flow_alloc_r__irm_req_ser(&msg, &flow, &resp, 0) < 0) {
+ err = -ENOMEM;
+ goto fail_resp;
+ }
+
+ freebuf(resp);
+
+ err = send_recv_msg(&msg);
+ if (err < 0)
+ goto fail_msg;
+
+ crypt.key = key;
+ crypt.epoch = 0;
+ crypt.role = CRYPT_ROLE_INIT;
+
+ err = flow__irm_result_des(&msg, &flow, &crypt);
+ if (err < 0)
+ goto fail_msg;
+
+ err = flow_init(&flow, &crypt, ts_diff_ns(&t1, &t0), pf);
+
+ crypt_secure_clear(key, SYMMKEYSZ);
+
+ if (err < 0)
+ goto fail_msg;
+
+ if (qs != NULL)
+ *qs = flow.qs;
+
+ return err;
+
+ fail_resp:
+ freebuf(resp);
+ fail_msg:
+ poa_alloc_fail(pf);
+ return err;
+ fail_alloc:
+ msg.len = SOCK_BUF_SIZE;
+ msg.data = buf;
+ if (poa_flow_alloc_r__irm_req_ser(&msg, &flow, NULL, err) == 0)
+ send_recv_msg(&msg);
+
+ return err;
+}
+
diff --git a/src/lib/poa/poa.h b/src/lib/poa/poa.h
new file mode 100644
index 00000000..014986a3
--- /dev/null
+++ b/src/lib/poa/poa.h
@@ -0,0 +1,364 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2026
+ *
+ * Points of attachment (PoA) - internal API
+ *
+ * Dimitri Staessens <dimitri@ouroboros.rocks>
+ * Sander Vrijders <sander@ouroboros.rocks>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public License
+ * version 2.1 as published by the Free Software Foundation.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., http://www.fsf.org/about/contact/.
+ */
+
+#ifndef OUROBOROS_LIB_POA_POA_H
+#define OUROBOROS_LIB_POA_POA_H
+
+#include <ouroboros/atomics.h>
+#include <ouroboros/ipcp-dev.h>
+#include <ouroboros/list.h>
+#include <ouroboros/qos.h>
+#include <ouroboros/rcu.h>
+#include <ouroboros/ssm_flow_set.h>
+#include <ouroboros/ssm_pool.h>
+#include <ouroboros/ssm_rbuff.h>
+#include <ouroboros/time.h>
+#include <ouroboros/utils.h>
+
+#include <errno.h>
+#include <limits.h>
+#include <poll.h>
+#include <pthread.h>
+#include <stdbool.h>
+#include <stdint.h>
+
+#define POA_MGMT_EID 0 /* reserved for the mgmt channel */
+
+#define POA_FLOW_REQ 1
+#define POA_FLOW_REPLY 2
+#define POA_FLOW_UPDATE 3
+#define POA_NAME_QUERY 4
+#define POA_NAME_REPLY 5
+
+#define POA_QUERY_HLEN 32 /* SHA3-256, the query hash algorithm */
+
+/* Fits "udp6.<ip6>.<port>", the longest display/RIB entry name. */
+#define POA_NAME_STRLEN 63
+
+/* Management message; every transport frames it the same way. */
+struct poa_mgmt_msg {
+ uint8_t code;
+ uint8_t resv[3];
+ uint32_t s_eid;
+ uint32_t d_eid;
+ int32_t response;
+ uint64_t bandwidth;
+ uint32_t delay;
+ uint32_t loss;
+ uint32_t ber;
+ uint32_t max_gap;
+ uint32_t timeout;
+ uint8_t availability;
+ uint8_t service;
+ uint16_t data_len;
+} __attribute__((packed));
+
+struct poa;
+struct poa_flow;
+
+#ifdef PROC_FLOW_STATS
+#define POA_STAT_BUMP(poa, field) FETCH_ADD_RELAXED(&(poa)->stat.field, 1)
+#define POA_STAT_ADD(poa, field, v) FETCH_ADD_RELAXED(&(poa)->stat.field, (v))
+#define POA_STAT_SUB(poa, field, v) FETCH_SUB_RELAXED(&(poa)->stat.field, (v))
+#define POA_STAT_LOAD(poa, field) LOAD_RELAXED(&(poa)->stat.field)
+#else
+#define POA_STAT_BUMP(poa, field) ((void) (poa))
+#define POA_STAT_ADD(poa, field, v) ((void) (poa), (void) (v))
+#define POA_STAT_SUB(poa, field, v) ((void) (poa), (void) (v))
+#define POA_STAT_LOAD(poa, field) ((void) (poa), (size_t) 0)
+#endif
+
+struct poa_stat {
+ size_t n_flows; /* gauge; the RIB reads it without poas.lock */
+ size_t rx_pkts; /* packets delivered to a flow */
+ size_t rx_bytes; /* payload bytes delivered */
+ size_t tx_pkts; /* packets handed to the transport */
+ size_t tx_bytes; /* payload bytes handed to the transport */
+ size_t mgmt_rx; /* frames queued for the mgmt handler */
+ size_t mgmt_tx; /* management frames sent */
+ size_t bad_eid; /* no flow on the EID a peer sent */
+ size_t dlv_fail; /* the rx ring above would not take it */
+ size_t buf_fail; /* no buffer to receive into */
+ size_t rcv_fail; /* transport read failed; the reader exits */
+ size_t snd_fail; /* transport send failed */
+ size_t qry_tx; /* name queries broadcast */
+ size_t qry_rx; /* name queries received */
+ size_t rep_tx; /* name replies sent, query matched */
+ size_t rep_rx; /* name replies received */
+};
+
+/* Spacing between transmit-depth samples; a depth costs a syscall. */
+#define POA_QLEN_GATE (100 * 1000) /* ns */
+
+
+/* Transport operations; public poa_X() dispatches to ops->poa_X. */
+struct poa_ops {
+ /* Parse own arm of the spec; validate; fill local and priv. */
+ int (* poa_attach)(struct poa * poa,
+ const struct poa_spec * spec);
+
+ void (* poa_detach)(struct poa * poa);
+
+ int (* poa_start)(struct poa * poa);
+
+ void (* poa_stop)(struct poa * poa);
+
+ /* Full queue: -EAGAIN unless block; then wait, to abstime if set. */
+ int (* poa_send)(struct poa * poa,
+ const struct poa_addr * dst,
+ uint32_t eid,
+ struct ssm_pk_buff * spb,
+ bool block,
+ const struct timespec * abstime);
+
+ int (* poa_send_mgmt)(struct poa * poa,
+ const struct poa_addr * dst,
+ const uint8_t * buf,
+ size_t len);
+
+ int (* poa_query)(const char * dst,
+ const struct timespec * timeo,
+ struct poa_addr * addr);
+
+ uint32_t (* poa_mtu)(struct poa * poa,
+ const struct poa_addr * dst);
+
+ /* Bytes queued in the transmit path of the PoA. */
+ size_t (* poa_qlen)(struct poa * poa);
+
+ /* Depth from the queue itself; NULL infers it from qlen. */
+ int (* poa_qpkts)(struct poa * poa,
+ size_t * pkts,
+ size_t * byts);
+
+ int (* poa_rib)(struct poa * poa,
+ char * buf,
+ size_t len);
+
+ /* Identity as a spec, e.g. for poa_list. */
+ void (* poa_spec)(const struct poa * poa,
+ struct poa_spec * spec);
+
+ /* Same identity as spec? Caller matched poa->type already. */
+ bool (* poa_has_id)(const struct poa * poa,
+ const struct poa_spec * spec);
+
+ /* Carries dst? Caller matched poa->type already. */
+ bool (* poa_match)(const struct poa * poa,
+ const struct poa_addr * dst);
+
+ /*
+ * Flows ride the link this id names; NULL: no link events.
+ * Ids are meaningful only to the backend whose monitor
+ * produced them; a single backend owns the monitor.
+ */
+ bool (* poa_link_match)(const struct poa * poa,
+ int id);
+
+ /* Maximum packet lifetime in the transport, seconds. */
+ time_t mpl;
+};
+struct poa {
+ struct list_head next;
+
+ enum poa_type type;
+ const struct poa_ops * ops;
+ void * priv;
+
+ struct poa_addr local; /* what peers dial us on */
+
+ /* Display/RIB entry name for local, e.g. "udp4.<ip>.<port>". */
+ char name[POA_NAME_STRLEN + 1];
+
+ time_t mpl;
+
+ /* Identifies the transmit queue the flows on this PoA share. */
+ int qid;
+
+ /* Mean sent packet size (bytes), EWMA over the send path. */
+ size_t avg_len;
+ /* Cost of one packet in the queue, in the transport's terms. */
+ size_t avg_ovh;
+ size_t n_tx;
+
+ /* Last queue depth read, and when, in the transport's terms. */
+ size_t q_cache;
+ uint64_t q_time;
+
+ /* Queued management frames, capped; poas.mgmt_mtx guards. */
+ size_t n_mgmt;
+
+#ifdef PROC_FLOW_STATS
+ struct poa_stat stat;
+#endif
+
+ struct bmp * eids;
+ struct poa_flow ** eid_to_pf;
+ size_t n_eids;
+
+ struct list_head flows; /* live flows, for repeats */
+
+ /* Keeps a flow and its ring alive under the receive path. */
+ struct rcu_guard guard;
+};
+
+/* poa/poa.c is part of the dev.c translation unit. */
+int poa_init(const char * name);
+
+int poa_start(void);
+
+void poa_stop(void);
+
+void poa_fini(void);
+
+/* Also answer name queries for the layer once enrolled. */
+int poa_set_layer(const char * layer);
+
+/* Hash of a name this process answers queries for? */
+bool poa_has_name(const uint8_t * hash);
+
+int poa_flow_tx(struct poa_flow * pf,
+ struct ssm_pk_buff * spb,
+ bool block,
+ const struct timespec * abstime);
+
+size_t poa_flow_qlen(const struct poa_flow * pf);
+
+size_t poa_flow_qpkts(const struct poa_flow * pf);
+
+int poa_flow_qid(const struct poa_flow * pf);
+
+size_t poa_flow_mean_len(const struct poa_flow * pf);
+
+void poa_flow_attach(struct poa_flow * pf,
+ int flow_id,
+ struct ssm_rbuff * rx_rb);
+
+void poa_flow_ready(struct poa_flow * pf);
+
+void poa_flow_detach(struct poa_flow * pf);
+
+struct poa_flow * poa_flow_take_pending(int flow_id);
+
+/* Addresses and management messages (poa/addr.c). */
+int poa_addr_cmp(const struct poa_addr * a,
+ const struct poa_addr * b);
+
+/* Display/RIB entry name, e.g. "udp4.<ip>.<port>". */
+int poa_addr_name(const struct poa_addr * a,
+ char * buf,
+ size_t len);
+
+void poa_mgmt_msg_ser(struct poa_mgmt_msg * msg,
+ uint8_t code,
+ uint32_t s_eid,
+ uint32_t d_eid,
+ qosspec_t qs,
+ int response,
+ size_t data_len);
+
+void poa_mgmt_msg_qos(const struct poa_mgmt_msg * msg,
+ qosspec_t * qs);
+
+/* Called by the transports. */
+void poa_rx_pkt(struct poa * poa,
+ uint32_t eid,
+ struct ssm_pk_buff * spb);
+
+void poa_rx_mgmt(struct poa * poa,
+ const struct poa_addr * src,
+ const uint8_t * buf,
+ size_t len);
+
+/* Reserve a buffer for a received packet, with transport headroom. */
+int poa_spb_reserve(struct ssm_pk_buff ** spb,
+ size_t len);
+
+void poa_spb_release(struct ssm_pk_buff * spb);
+
+/*
+ * All flows on PoAs whose poa_link_match reports this link id go up
+ * or down with it. Returns the number of flows whose state changed.
+ */
+size_t poa_link_updown(int id,
+ bool up);
+
+/*
+ * Link monitor: one socket for the whole subsystem, opened by
+ * poa_start(). Returns -1 where the transport has no monitor.
+ */
+int poa_monitor_open(void);
+
+/* Reads one batch of link events; cancellation point. */
+void poa_monitor_read(int fd);
+
+/* Broadcast a mgmt frame on every PoA matching dst; # sent. */
+int poa_bcast_mgmt(const struct poa_addr * dst,
+ const uint8_t * buf,
+ size_t len);
+
+/* Transport op tables. */
+extern const struct poa_ops udp_poa_ops;
+extern const struct poa_ops eth_poa_ops;
+
+/*
+ * Waits for a descriptor to take another packet, up to abstime.
+ * A NULL deadline waits indefinitely. Transports call this when
+ * their send reports the transmit queue full.
+ */
+static __inline__ int poa_wait_out(int fd,
+ const struct timespec * abstime)
+{
+ struct pollfd pfd;
+ struct timespec now;
+ long ms = -1;
+ bool clamped = false;
+ int ret;
+
+ if (abstime != NULL) {
+ clock_gettime(PTHREAD_COND_CLOCK, &now);
+
+ if (ts_diff_ns(abstime, &now) <= 0)
+ return -ETIMEDOUT;
+
+ ms = ts_diff_ms(abstime, &now) + 1; /* sub-ms must wait */
+ if (ms > INT_MAX) { /* poll takes an int */
+ ms = INT_MAX;
+ clamped = true;
+ }
+ }
+
+ pfd.fd = fd;
+ pfd.events = POLLOUT;
+ pfd.revents = 0;
+
+ ret = poll(&pfd, 1, (int) ms);
+ if (ret < 0)
+ return errno == EINTR ? 0 : -EIO;
+
+ if (ret == 0)
+ return clamped ? 0 : -ETIMEDOUT; /* clamped: retry */
+
+ return 0;
+}
+
+#endif /* OUROBOROS_LIB_POA_POA_H */
diff --git a/src/lib/poa/udp.c b/src/lib/poa/udp.c
new file mode 100644
index 00000000..6753347a
--- /dev/null
+++ b/src/lib/poa/udp.c
@@ -0,0 +1,633 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2026
+ *
+ * Points of attachment (PoA) - UDP transport
+ *
+ * Dimitri Staessens <dimitri@ouroboros.rocks>
+ * Sander Vrijders <sander@ouroboros.rocks>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public License
+ * version 2.1 as published by the Free Software Foundation.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., http://www.fsf.org/about/contact/.
+ */
+
+#if defined(__APPLE__)
+#define _BSD_SOURCE
+#define _DARWIN_C_SOURCE
+#elif defined(__FreeBSD__)
+#define __BSD_VISIBLE 1
+#elif defined(__linux__) || defined(__CYGWIN__)
+#ifndef _DEFAULT_SOURCE
+#define _DEFAULT_SOURCE
+#endif
+#else
+#ifndef _POSIX_C_SOURCE
+#define _POSIX_C_SOURCE 200809L
+#endif
+#endif
+
+#include "config.h"
+
+#define OUROBOROS_PREFIX "poa-udp"
+
+#include <ouroboros/endian.h>
+#include <ouroboros/errno.h>
+#include <ouroboros/logs.h>
+#include <ouroboros/time.h>
+#include <ouroboros/utils.h>
+
+#include "poa.h"
+
+#ifdef __linux__
+#include <linux/sockios.h>
+#endif
+#include <arpa/inet.h>
+#include <netinet/in.h>
+#include <sys/ioctl.h>
+#include <sys/socket.h>
+#include <sys/uio.h>
+
+#include <errno.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+#define UDP_HDR_LEN sizeof(uint32_t) /* PoA id */
+#define UDP_MAX_PAYLOAD (POA_UDP_RD_BUF - UDP_HDR_LEN)
+/* The reader buffer must fit a full mgmt frame at any tuning. */
+#define UDP_MAX_PACKET MAX(POA_UDP_RD_BUF, POA_MGMT_FRAME_SIZE + UDP_HDR_LEN)
+#define UDP_IP4_OVERH 28U /* IPv4 + UDP */
+#define UDP_IP6_OVERH 48U /* IPv6 + UDP */
+/* Wait for the link to come back before reading it again. */
+#define UDP_DOWN_TIMEO 100 /* ms */
+
+union udp_saddr {
+ struct sockaddr sa;
+ struct sockaddr_in in;
+ struct sockaddr_in6 in6;
+};
+
+struct udp_priv {
+ int s_fd;
+ int af;
+ union udp_saddr s_saddr;
+ pthread_t reader;
+ bool running;
+};
+
+static socklen_t saddr_len(int af)
+{
+ if (af == AF_INET)
+ return sizeof(struct sockaddr_in);
+
+ return sizeof(struct sockaddr_in6);
+}
+
+static void addr_to_saddr(const struct poa_addr * addr,
+ union udp_saddr * saddr)
+{
+ memset(saddr, 0, sizeof(*saddr));
+
+ if (addr->type == POA_UDP4) {
+ saddr->in.sin_family = AF_INET;
+ saddr->in.sin_addr = addr->udp4.ip_addr;
+ saddr->in.sin_port = htons(addr->udp4.port);
+ } else {
+ saddr->in6.sin6_family = AF_INET6;
+ saddr->in6.sin6_addr = addr->udp6.ip_addr;
+ saddr->in6.sin6_port = htons(addr->udp6.port);
+ }
+}
+
+static void saddr_to_addr(const union udp_saddr * saddr,
+ struct poa_addr * addr)
+{
+ memset(addr, 0, sizeof(*addr));
+
+ if (saddr->sa.sa_family == AF_INET) {
+ addr->type = POA_UDP4;
+ addr->udp4.ip_addr = saddr->in.sin_addr;
+ addr->udp4.port = ntohs(saddr->in.sin_port);
+ } else {
+ addr->type = POA_UDP6;
+ addr->udp6.ip_addr = saddr->in6.sin6_addr;
+ addr->udp6.port = ntohs(saddr->in6.sin6_port);
+ }
+}
+
+/* A datagram longer than the buffer arrives truncated: drop it. */
+static void * udp_reader(void * o)
+{
+ struct poa * poa = (struct poa *) o;
+ struct udp_priv * priv = (struct udp_priv *) poa->priv;
+ struct timespec down = TIMESPEC_INIT_MS(UDP_DOWN_TIMEO);
+ uint8_t * buf;
+
+ buf = malloc(UDP_MAX_PACKET);
+ if (buf == NULL)
+ return (void *) -1;
+
+ pthread_cleanup_push(free, buf);
+
+ while (true) {
+ struct ssm_pk_buff * spb;
+ union udp_saddr r_saddr;
+ struct poa_addr src;
+ struct msghdr mh;
+ struct iovec iov;
+ ssize_t n;
+ uint32_t eid;
+ size_t plen;
+
+ iov.iov_base = buf;
+ iov.iov_len = UDP_MAX_PACKET;
+
+ memset(&mh, 0, sizeof(mh));
+
+ mh.msg_name = &r_saddr;
+ mh.msg_namelen = sizeof(r_saddr);
+ mh.msg_iov = &iov;
+ mh.msg_iovlen = 1;
+
+ n = recvmsg(priv->s_fd, &mh, 0);
+ if (n < 0) {
+ if (errno == EINTR)
+ continue;
+
+ POA_STAT_BUMP(poa, rcv_fail);
+
+ if (errno == ENETDOWN) {
+ nanosleep(&down, NULL);
+ continue;
+ }
+
+ log_err("Reader stopped: %s.", strerror(errno));
+ break;
+ }
+
+ if ((mh.msg_flags & MSG_TRUNC) != 0) {
+ POA_STAT_BUMP(poa, buf_fail);
+ continue;
+ }
+
+ if ((size_t) n < UDP_HDR_LEN)
+ continue;
+
+ eid = ntoh32(*(uint32_t *) buf);
+ plen = (size_t) n - UDP_HDR_LEN;
+
+ saddr_to_addr(&r_saddr, &src);
+
+ if (eid == POA_MGMT_EID) {
+ poa_rx_mgmt(poa, &src, buf + UDP_HDR_LEN, plen);
+ continue;
+ }
+
+ if (poa_spb_reserve(&spb, plen) < 0) {
+ POA_STAT_BUMP(poa, buf_fail);
+ continue;
+ }
+
+ memcpy(ssm_pk_buff_head(spb), buf + UDP_HDR_LEN, plen);
+
+ poa_rx_pkt(poa, eid, spb);
+ }
+
+ pthread_cleanup_pop(true);
+
+ return (void *) 0;
+}
+
+/* Reads the bound address back: an ephemeral port is only known after. */
+static int udp_attach(struct poa * poa,
+ const struct poa_spec * spec)
+{
+ struct udp_priv * priv;
+ socklen_t len;
+ int af;
+
+ af = spec->type == POA_UDP4 ? AF_INET : AF_INET6;
+
+ priv = malloc(sizeof(*priv));
+ if (priv == NULL)
+ return -ENOMEM;
+
+ memset(priv, 0, sizeof(*priv));
+
+ priv->af = af;
+ priv->s_fd = socket(af, SOCK_DGRAM, IPPROTO_UDP);
+ if (priv->s_fd < 0) {
+ log_err("Failed to create socket: %s.", strerror(errno));
+ goto fail_socket;
+ }
+
+ if (af == AF_INET) {
+ priv->s_saddr.in.sin_family = AF_INET;
+ priv->s_saddr.in.sin_addr = spec->udp4.ip_addr;
+ priv->s_saddr.in.sin_port = htons(spec->udp4.port);
+ } else {
+ int on = 1;
+
+ if (setsockopt(priv->s_fd, IPPROTO_IPV6, IPV6_V6ONLY,
+ &on, sizeof(on)) < 0) {
+ log_err("Failed to set IPV6_V6ONLY: %s.",
+ strerror(errno));
+ goto fail_bind;
+ }
+
+ priv->s_saddr.in6.sin6_family = AF_INET6;
+ priv->s_saddr.in6.sin6_addr = spec->udp6.ip_addr;
+ priv->s_saddr.in6.sin6_port = htons(spec->udp6.port);
+ }
+
+ if (bind(priv->s_fd, &priv->s_saddr.sa, saddr_len(af)) < 0) {
+ log_err("Failed to bind: %s.", strerror(errno));
+ goto fail_bind;
+ }
+
+ poa->priv = priv;
+ len = saddr_len(af);
+ if (getsockname(priv->s_fd, &priv->s_saddr.sa, &len) < 0)
+ log_warn("Failed to read the bound address: %s.",
+ strerror(errno));
+
+ saddr_to_addr(&priv->s_saddr, &poa->local);
+
+ return 0;
+
+ fail_bind:
+ close(priv->s_fd);
+ fail_socket:
+ poa->priv = NULL;
+
+ free(priv);
+
+ return -EIO;
+}
+
+static void udp_detach(struct poa * poa)
+{
+ struct udp_priv * priv = (struct udp_priv *) poa->priv;
+
+ if (priv == NULL)
+ return;
+
+ close(priv->s_fd);
+
+ free(priv);
+
+ poa->priv = NULL;
+}
+
+/* One reader per socket, so a flow cannot be reordered on receive. */
+static int udp_start(struct poa * poa)
+{
+ struct udp_priv * priv = (struct udp_priv *) poa->priv;
+
+ if (pthread_create(&priv->reader, NULL, udp_reader, poa) != 0)
+ return -1;
+
+ priv->running = true;
+
+ return 0;
+}
+
+static void udp_stop(struct poa * poa)
+{
+ struct udp_priv * priv = (struct udp_priv *) poa->priv;
+
+ if (!priv->running)
+ return;
+
+ pthread_cancel(priv->reader);
+ pthread_join(priv->reader, NULL);
+
+ priv->running = false;
+}
+
+/*
+ * The PoA id is a separate iovec, so the buffer needs no headroom.
+ * MSG_DONTWAIT: the reader blocks on this socket.
+ */
+static int udp_sendv(struct udp_priv * priv,
+ const struct poa_addr * dst,
+ uint32_t eid,
+ const uint8_t * body,
+ size_t len,
+ bool block,
+ const struct timespec * abstime)
+{
+ union udp_saddr saddr;
+ struct msghdr msg;
+ struct iovec iov[2];
+ uint32_t hdr;
+ int ret;
+
+ if (len > UDP_MAX_PAYLOAD)
+ return -EMSGSIZE;
+
+ addr_to_saddr(dst, &saddr);
+
+ hdr = hton32(eid);
+
+ iov[0].iov_base = &hdr;
+ iov[0].iov_len = sizeof(hdr);
+ iov[1].iov_base = (void *) body;
+ iov[1].iov_len = len;
+
+ memset(&msg, 0, sizeof(msg));
+
+ msg.msg_name = &saddr;
+ msg.msg_namelen = saddr_len(priv->af);
+ msg.msg_iov = iov;
+ msg.msg_iovlen = len > 0 ? 2 : 1;
+ while (sendmsg(priv->s_fd, &msg, MSG_DONTWAIT) < 0) {
+ if (errno != EAGAIN && errno != EWOULDBLOCK)
+ return -EIO;
+
+ if (!block)
+ return -EAGAIN;
+
+ ret = poa_wait_out(priv->s_fd, abstime);
+ if (ret < 0)
+ return ret;
+ }
+
+ return 0;
+}
+
+static int udp_send(struct poa * poa,
+ const struct poa_addr * dst,
+ uint32_t eid,
+ struct ssm_pk_buff * spb,
+ bool block,
+ const struct timespec * abstime)
+{
+ return udp_sendv((struct udp_priv *) poa->priv, dst, eid,
+ ssm_pk_buff_head(spb), ssm_pk_buff_len(spb),
+ block, abstime);
+}
+
+static int udp_send_mgmt(struct poa * poa,
+ const struct poa_addr * dst,
+ const uint8_t * buf,
+ size_t len)
+{
+ struct timespec timeo = TIMESPEC_INIT_MS(POA_MGMT_SND_TIMEO);
+ struct timespec abstime;
+
+ clock_gettime(PTHREAD_COND_CLOCK, &abstime);
+ ts_add(&abstime, &timeo, &abstime);
+
+ return udp_sendv((struct udp_priv *) poa->priv, dst, POA_MGMT_EID,
+ buf, len, true, &abstime);
+}
+
+/* The PoA id header eats into the usable MTU. */
+static uint32_t udp_mtu(struct poa * poa,
+ const struct poa_addr * dst)
+{
+ struct udp_priv * priv = (struct udp_priv *) poa->priv;
+ uint32_t fallback;
+ uint32_t overh;
+#if defined(__linux__) && (defined(IP_MTU) || defined(IPV6_MTU))
+ union udp_saddr saddr;
+ socklen_t len;
+ int sock;
+ int mtu = 0;
+#endif
+ if (priv->af == AF_INET) {
+ fallback = POA_UDP4_MTU;
+ overh = UDP_IP4_OVERH;
+ } else {
+ fallback = POA_UDP6_MTU;
+ overh = UDP_IP6_OVERH;
+ }
+
+ fallback -= UDP_HDR_LEN;
+ if (fallback > UDP_MAX_PAYLOAD)
+ fallback = UDP_MAX_PAYLOAD;
+
+#if defined(__linux__) && (defined(IP_MTU) || defined(IPV6_MTU))
+
+ addr_to_saddr(dst, &saddr);
+
+ sock = socket(priv->af, SOCK_DGRAM, IPPROTO_UDP);
+ if (sock < 0)
+ return fallback;
+
+ if (connect(sock, &saddr.sa, saddr_len(priv->af)) < 0)
+ goto fallback;
+
+ len = sizeof(mtu);
+
+#if defined(IP_MTU)
+ if (priv->af == AF_INET) {
+ if (getsockopt(sock, IPPROTO_IP, IP_MTU, &mtu, &len) < 0)
+ goto fallback;
+ }
+#endif
+#if defined(IPV6_MTU)
+ if (priv->af == AF_INET6) {
+ if (getsockopt(sock, IPPROTO_IPV6, IPV6_MTU, &mtu, &len) < 0)
+ goto fallback;
+ }
+#endif
+ close(sock);
+
+ if (mtu <= (int) (overh + UDP_HDR_LEN))
+ return fallback;
+
+ return MIN((uint32_t) mtu - overh - UDP_HDR_LEN, UDP_MAX_PAYLOAD);
+
+ fallback:
+ close(sock);
+#else
+ (void) dst;
+ (void) overh;
+#endif
+ return fallback;
+}
+
+/* All flows on the PoA share the socket, so this is aggregate. */
+static size_t udp_qlen(struct poa * poa)
+{
+#if defined(__linux__) && defined(SIOCOUTQ)
+ struct udp_priv * priv = (struct udp_priv *) poa->priv;
+ int qlen;
+
+ qlen = 0;
+ if (ioctl(priv->s_fd, SIOCOUTQ, &qlen) < 0)
+ return 0;
+
+ return (size_t) qlen;
+#else
+ (void) poa;
+
+ return 0;
+#endif
+}
+
+/* The kernel keeps no per-socket drop count for UDP. */
+static int udp_rib(struct poa * poa,
+ char * buf,
+ size_t len)
+{
+ struct udp_priv * priv = (struct udp_priv *) poa->priv;
+ socklen_t optlen;
+ size_t sndbuf = 0;
+ size_t rcvbuf = 0;
+ int val;
+ int size;
+
+ optlen = sizeof(val);
+ if (getsockopt(priv->s_fd, SOL_SOCKET, SO_SNDBUF, &val, &optlen) == 0)
+ sndbuf = (size_t) val;
+
+ optlen = sizeof(val);
+ if (getsockopt(priv->s_fd, SOL_SOCKET, SO_RCVBUF, &val, &optlen) == 0)
+ rcvbuf = (size_t) val;
+
+ size = snprintf(buf, len,
+ "Socket sndbuf (bytes): %zu\n"
+ "Socket rcvbuf (bytes): %zu\n",
+ sndbuf, rcvbuf);
+ if (size < 0 || (size_t) size >= len)
+ return -1;
+
+ return size;
+}
+
+/*
+ * Asks the kernel which address it would send from: connect() does the
+ * real route lookup and sends nothing, so this honours the default
+ * route, metrics and policy rules alike.
+ */
+static int udp_src_addr(const struct poa_addr * dst,
+ struct poa_addr * src)
+{
+ union udp_saddr saddr;
+ socklen_t len;
+ int af;
+ int fd;
+
+ if (dst->type != POA_UDP4 && dst->type != POA_UDP6)
+ return -EINVAL;
+
+ af = dst->type == POA_UDP4 ? AF_INET : AF_INET6;
+
+ addr_to_saddr(dst, &saddr);
+
+ fd = socket(af, SOCK_DGRAM, IPPROTO_UDP);
+ if (fd < 0)
+ return -EIO;
+
+ if (connect(fd, &saddr.sa, saddr_len(af)) < 0)
+ goto fail;
+
+ len = saddr_len(af);
+ if (getsockname(fd, &saddr.sa, &len) < 0)
+ goto fail;
+
+ close(fd);
+
+ saddr_to_addr(&saddr, src);
+
+ return 0;
+
+ fail:
+ close(fd);
+
+ return -EIO;
+}
+
+static void udp_spec(const struct poa * poa,
+ struct poa_spec * spec)
+{
+ spec->type = poa->type;
+
+ if (poa->type == POA_UDP4)
+ spec->udp4 = poa->local.udp4;
+ else
+ spec->udp6 = poa->local.udp6;
+}
+
+static bool udp_has_id(const struct poa * poa,
+ const struct poa_spec * spec)
+{
+ if (poa->type == POA_UDP4) {
+ if (poa->local.udp4.port != spec->udp4.port)
+ return false;
+
+ return memcmp(&poa->local.udp4.ip_addr,
+ &spec->udp4.ip_addr,
+ sizeof(spec->udp4.ip_addr)) == 0;
+ }
+
+ if (poa->local.udp6.port != spec->udp6.port)
+ return false;
+
+ return memcmp(&poa->local.udp6.ip_addr, &spec->udp6.ip_addr,
+ sizeof(spec->udp6.ip_addr)) == 0;
+}
+
+static bool udp_addr_is_any(const struct poa_addr * addr)
+{
+ static const struct in6_addr any6 = IN6ADDR_ANY_INIT;
+
+ if (addr->type == POA_UDP4)
+ return addr->udp4.ip_addr.s_addr == htonl(INADDR_ANY);
+
+ return memcmp(&addr->udp6.ip_addr, &any6, sizeof(any6)) == 0;
+}
+
+/*
+ * Our end of the flow: the IP the kernel would send to dst from. A
+ * PoA bound to the wildcard is the catch-all and a failed probe
+ * matches any. Ports are not compared: the probe's is ephemeral.
+ */
+static bool udp_match(const struct poa * poa,
+ const struct poa_addr * dst)
+{
+ struct poa_addr src;
+
+ if (udp_addr_is_any(&poa->local))
+ return true;
+
+ if (udp_src_addr(dst, &src) < 0)
+ return true;
+
+ if (poa->type == POA_UDP4)
+ return memcmp(&poa->local.udp4.ip_addr,
+ &src.udp4.ip_addr,
+ sizeof(src.udp4.ip_addr)) == 0;
+
+ return memcmp(&poa->local.udp6.ip_addr, &src.udp6.ip_addr,
+ sizeof(src.udp6.ip_addr)) == 0;
+}
+
+const struct poa_ops udp_poa_ops = {
+ .poa_attach = udp_attach,
+ .poa_detach = udp_detach,
+ .poa_start = udp_start,
+ .poa_stop = udp_stop,
+ .poa_send = udp_send,
+ .poa_send_mgmt = udp_send_mgmt,
+ .poa_mtu = udp_mtu,
+ .poa_qlen = udp_qlen,
+ .poa_rib = udp_rib,
+ .poa_spec = udp_spec,
+ .poa_has_id = udp_has_id,
+ .poa_match = udp_match,
+ .mpl = POA_UDP_MPL
+};
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-01 11:04:38 +0000