summaryrefslogtreecommitdiff
path: root/src/ipcpd/common/connmgr.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/ipcpd/common/connmgr.c')
-rw-r--r--src/ipcpd/common/connmgr.c292
1 files changed, 200 insertions, 92 deletions
diff --git a/src/ipcpd/common/connmgr.c b/src/ipcpd/common/connmgr.c
index 53a66992..e0ad80cb 100644
--- a/src/ipcpd/common/connmgr.c
+++ b/src/ipcpd/common/connmgr.c
@@ -1,5 +1,5 @@
/*
- * Ouroboros - Copyright (C) 2016 - 2021
+ * Ouroboros - Copyright (C) 2016 - 2026
*
* Handles connections between components
*
@@ -22,26 +22,27 @@
#define OUROBOROS_PREFIX "connection-manager"
+#include <ouroboros/cep.h>
#include <ouroboros/dev.h>
-#include <ouroboros/cacep.h>
#include <ouroboros/errno.h>
+#include <ouroboros/fccntl.h>
+#include <ouroboros/ipcp-dev.h>
#include <ouroboros/list.h>
#include <ouroboros/logs.h>
#include <ouroboros/notifier.h>
#include <ouroboros/pthread.h>
+#include <ouroboros/qos.h>
#include "connmgr.h"
#include "ipcp.h"
-#include <string.h>
-#include <stdlib.h>
#include <assert.h>
+#include <stdlib.h>
+#include <string.h>
-enum connmgr_state {
- CONNMGR_NULL = 0,
- CONNMGR_INIT,
- CONNMGR_RUNNING
-};
+#define CONNMGR_ETH_PROBE_TIMEO 20 /* ms, one query attempt */
+#define CONNMGR_ETH_RETRY_TIMEO 1500 /* ms, the remaining tries */
+#define CONNMGR_DHT_TIMEO 1000 /* ms, bounded lower-layer */
struct conn_el {
struct list_head next;
@@ -60,11 +61,18 @@ struct comp {
struct {
struct comp comps[COMPID_MAX];
- enum connmgr_state state;
pthread_t acceptor;
} connmgr;
+static bool is_eth_query(const struct poa_addr * addr)
+{
+ static const uint8_t zero[POA_MAC_SIZE] = { 0 };
+
+ return addr->type == POA_ETH &&
+ memcmp(addr->eth.dst.mac, zero, POA_MAC_SIZE) == 0;
+}
+
static int get_id_by_name(const char * name)
{
enum comp_id i;
@@ -126,55 +134,70 @@ static int add_comp_conn(enum comp_id id,
return 0;
}
+/* qs is also an in-parameter, and flow_accept writes it back. */
static void * flow_acceptor(void * o)
{
- int fd;
- qosspec_t qs;
- struct conn_info rcv_info;
- struct conn_info fail_info;
+ int fd;
+ struct conn_info rcv_info;
+ struct conn_info fail_info;
+ struct timespec timeo = TIMESPEC_INIT_MS(CONNMGR_RCV_TIMEOUT);
+ int err;
(void) o;
memset(&fail_info, 0, sizeof(fail_info));
while (true) {
- int id;
+ qosspec_t qs = qos_raw;
+ int id;
fd = flow_accept(&qs, NULL);
if (fd < 0) {
if (fd != -EIRMD)
- log_warn("Flow accept failed: %d", fd);
+ log_err("Flow accept failed: %d", fd);
continue;
}
- if (cacep_rcv(fd, &rcv_info)) {
- log_dbg("Error establishing application connection.");
+ log_info("Handling incoming flow %d.",fd);
+
+ fccntl(fd, FLOWSRCVTIMEO, &timeo);
+
+ err = cep_rcv(fd, &rcv_info);
+ if (err < 0) {
+ log_err("Error receiving OCEP info: %d.", err);
flow_dealloc(fd);
continue;
}
+ log_info("Request to connect to %s.", rcv_info.comp_name);
+
id = get_id_by_name(rcv_info.comp_name);
if (id < 0) {
- log_dbg("Connection request for unknown component %s.",
+ log_err("Connection request for unknown component %s.",
rcv_info.comp_name);
- cacep_snd(fd, &fail_info);
+ cep_snd(fd, &fail_info);
flow_dealloc(fd);
continue;
}
- assert(id < COMPID_MAX);
-
- if (cacep_snd(fd, &connmgr.comps[id].info)) {
- log_dbg("Failed to respond to request.");
+ err = cep_snd(fd, &connmgr.comps[id].info);
+ if (err < 0) {
+ log_err("Failed responding to OCEP request: %d.", err);
flow_dealloc(fd);
continue;
}
- if (add_comp_conn(id, fd, qs, &rcv_info)) {
- log_dbg("Failed to add new connection.");
+ fccntl(fd, FLOWSRCVTIMEO, NULL);
+
+ err = add_comp_conn(id, fd, qs, &rcv_info);
+ if (err < 0) {
+ log_err("Failed to add new connection: %d.", err);
flow_dealloc(fd);
continue;
}
+
+ log_info("Finished handling incoming flow %d for %s.",
+ fd, rcv_info.comp_name);
}
return (void *) 0;
@@ -213,10 +236,10 @@ static void handle_event(void * self,
int connmgr_init(void)
{
- connmgr.state = CONNMGR_INIT;
-
- if (notifier_reg(handle_event, NULL))
+ if (notifier_reg(handle_event, NULL)) {
+ log_err("Failed to register notifier.");
return -1;
+ }
return 0;
}
@@ -225,29 +248,26 @@ void connmgr_fini(void)
{
int i;
- notifier_unreg(handle_event);
-
- if (connmgr.state == CONNMGR_RUNNING)
- pthread_join(connmgr.acceptor, NULL);
-
for (i = 0; i < COMPID_MAX; ++i)
connmgr_comp_fini(i);
+
+ notifier_unreg(handle_event);
}
int connmgr_start(void)
{
- if (pthread_create(&connmgr.acceptor, NULL, flow_acceptor, NULL))
+ if (pthread_create(&connmgr.acceptor, NULL, flow_acceptor, NULL)) {
+ log_err("Failed to create pthread: %s.", strerror(errno));
return -1;
-
- connmgr.state = CONNMGR_RUNNING;
+ }
return 0;
}
void connmgr_stop(void)
{
- if (connmgr.state == CONNMGR_RUNNING)
- pthread_cancel(connmgr.acceptor);
+ pthread_cancel(connmgr.acceptor);
+ pthread_join(connmgr.acceptor, NULL);
}
int connmgr_comp_init(enum comp_id id,
@@ -259,12 +279,14 @@ int connmgr_comp_init(enum comp_id id,
comp = connmgr.comps + id;
- if (pthread_mutex_init(&comp->lock, NULL))
- return -1;
+ if (pthread_mutex_init(&comp->lock, NULL)) {
+ log_err("Failed to initialize mutex: %s.", strerror(errno));
+ goto fail_mutex;
+ }
if (pthread_cond_init(&comp->cond, NULL)) {
- pthread_mutex_destroy(&comp->lock);
- return -1;
+ log_err("Failed to initialize condvar: %s.", strerror(errno));
+ goto fail_cond;
}
list_head_init(&comp->conns);
@@ -273,6 +295,11 @@ int connmgr_comp_init(enum comp_id id,
memcpy(&connmgr.comps[id].info, info, sizeof(connmgr.comps[id].info));
return 0;
+
+ fail_cond:
+ pthread_mutex_destroy(&comp->lock);
+ fail_mutex:
+ return -1;
}
void connmgr_comp_fini(enum comp_id id)
@@ -310,32 +337,44 @@ void connmgr_comp_fini(enum comp_id id)
memset(&connmgr.comps[id].info, 0, sizeof(connmgr.comps[id].info));
}
-int connmgr_ipcp_connect(const char * dst,
- const char * component,
- qosspec_t qs)
+int connmgr_ipcp_connect(const char * dst,
+ const char * component,
+ qosspec_t qs,
+ const struct poa_addr * addr)
{
struct conn_el * ce;
int id;
+ int ret;
assert(dst);
assert(component);
+ if (qs.service == SVC_STREAM) {
+ log_err("No stream service on component flows.");
+ return -ENOTSUP;
+ }
+
ce = malloc(sizeof(*ce));
if (ce == NULL) {
- log_dbg("Out of memory.");
- return -1;
+ log_err("Out of memory.");
+ goto fail_malloc;
}
id = get_id_by_name(component);
if (id < 0) {
- log_dbg("No such component: %s", component);
- free(ce);
- return -1;
+ log_err("No such component: %s", component);
+ goto fail_id;
}
- if (connmgr_alloc(id, dst, &qs, &ce->conn)) {
- free(ce);
- return -1;
+ pthread_cleanup_push(free, ce);
+
+ ret = connmgr_alloc(id, dst, &qs, addr, &ce->conn);
+
+ pthread_cleanup_pop(false);
+
+ if (ret < 0) {
+ log_err("Failed to allocate flow.");
+ goto fail_id;
}
if (strlen(dst) > DST_MAX_STRLEN) {
@@ -353,6 +392,11 @@ int connmgr_ipcp_connect(const char * dst,
pthread_mutex_unlock(&connmgr.comps[id].lock);
return 0;
+
+ fail_id:
+ free(ce);
+ fail_malloc:
+ return -1;
}
int connmgr_ipcp_disconnect(const char * dst,
@@ -366,8 +410,10 @@ int connmgr_ipcp_disconnect(const char * dst,
assert(component);
id = get_id_by_name(component);
- if (id < 0)
+ if (id < 0) {
+ log_err("No such component: %s.", component);
return -1;
+ }
pthread_mutex_lock(&connmgr.comps[id].lock);
@@ -388,67 +434,123 @@ int connmgr_ipcp_disconnect(const char * dst,
return 0;
}
-int connmgr_alloc(enum comp_id id,
- const char * dst,
- qosspec_t * qs,
- struct conn * conn)
+/*
+ * Without an address, a peer may be on the wire or reachable through
+ * the layer below. A PoA query is cheap and creates no flow, so it
+ * goes first; the layer below gets a bounded try before the query
+ * retries, and the last try is unbounded.
+ */
+static int alloc_any(const char * dst,
+ qosspec_t * qs)
{
+ struct timespec probe = TIMESPEC_INIT_MS(CONNMGR_ETH_PROBE_TIMEO);
+ struct timespec retry = TIMESPEC_INIT_MS(CONNMGR_ETH_RETRY_TIMEO);
+ struct timespec below = TIMESPEC_INIT_MS(CONNMGR_DHT_TIMEO);
+ struct poa_addr addr;
+ int fd;
+
+ if (poa_query(dst, &probe, &addr) == 0) {
+ fd = poa_flow_alloc(dst, &addr, qs, NULL);
+ if (fd >= 0)
+ return fd;
+ }
+
+ fd = flow_alloc(dst, qs, &below);
+ if (fd >= 0)
+ return fd;
+
+ if (poa_query(dst, &retry, &addr) == 0) {
+ fd = poa_flow_alloc(dst, &addr, qs, NULL);
+ if (fd >= 0)
+ return fd;
+ }
+
+ return flow_alloc(dst, qs, NULL);
+}
+
+/* A literal peer address bypasses the layer below. */
+int connmgr_alloc(enum comp_id id,
+ const char * dst,
+ qosspec_t * qs,
+ const struct poa_addr * addr,
+ struct conn * conn)
+{
+ struct comp * comp;
+ int fd;
+ struct timespec timeo = TIMESPEC_INIT_MS(CONNMGR_RCV_TIMEOUT);
+
assert(id >= 0 && id < COMPID_MAX);
assert(dst);
- conn->flow_info.fd = flow_alloc(dst, qs, NULL);
- if (conn->flow_info.fd < 0) {
- log_dbg("Failed to allocate flow to %s.", dst);
- return -1;
+ comp = connmgr.comps + id;
+
+ if (addr != NULL)
+ fd = poa_flow_alloc(dst, addr, qs, NULL);
+ else
+ fd = alloc_any(dst, qs);
+
+ if (fd == -EPERM && addr != NULL) {
+ log_err("No PoA attached to reach %s.", dst);
+ goto fail_alloc;
}
+ if (fd == -EINVAL && addr != NULL) {
+ log_err("More than one PoA could reach %s", dst);
+ goto fail_alloc;
+ }
+
+ if (fd == -ETIMEDOUT && addr != NULL && is_eth_query(addr)) {
+ log_err("No answer to name query for %s.", dst);
+ goto fail_alloc;
+ }
+
+ if (fd < 0) {
+ log_err("Failed to allocate flow to %s.", dst);
+ goto fail_alloc;
+ }
+
+ conn->flow_info.fd = fd;
+
if (qs != NULL)
conn->flow_info.qs = *qs;
else
memset(&conn->flow_info.qs, 0, sizeof(conn->flow_info.qs));
- log_dbg("Sending cacep info for protocol %s to fd %d.",
- connmgr.comps[id].info.protocol, conn->flow_info.fd);
+ log_dbg("Sending OCEP info for protocol %s to fd %d.",
+ comp->info.protocol, conn->flow_info.fd);
- if (cacep_snd(conn->flow_info.fd, &connmgr.comps[id].info)) {
- log_dbg("Failed to create application connection.");
- flow_dealloc(conn->flow_info.fd);
- return -1;
+ fccntl(fd, FLOWSRCVTIMEO, &timeo);
+
+ if (cep_snd(fd, &comp->info)) {
+ log_err("Failed to send OCEP info.");
+ goto fail_cep;
}
- if (cacep_rcv(conn->flow_info.fd, &conn->conn_info)) {
- log_dbg("Failed to connect to application.");
- flow_dealloc(conn->flow_info.fd);
- return -1;
+ if (cep_rcv(fd, &conn->conn_info)) {
+ log_err("Failed to receive OCEP info.");
+ goto fail_cep;
}
- if (strcmp(connmgr.comps[id].info.protocol, conn->conn_info.protocol)) {
- log_dbg("Unknown protocol (requested %s, got %s).",
- connmgr.comps[id].info.protocol,
- conn->conn_info.protocol);
- flow_dealloc(conn->flow_info.fd);
- return -1;
+ if (strcmp(comp->info.protocol, conn->conn_info.protocol)) {
+ log_err("Unknown protocol (requested %s, got %s).",
+ comp->info.protocol, conn->conn_info.protocol);
+ goto fail_cep;
}
- if (connmgr.comps[id].info.pref_version !=
- conn->conn_info.pref_version) {
- log_dbg("Unknown protocol version.");
- flow_dealloc(conn->flow_info.fd);
- return -1;
+ if (comp->info.pref_version != conn->conn_info.pref_version) {
+ log_err("Unknown protocol version %d.",
+ conn->conn_info.pref_version);
+ goto fail_cep;
}
- if (connmgr.comps[id].info.pref_syntax != conn->conn_info.pref_syntax) {
- log_dbg("Unknown protocol syntax.");
- flow_dealloc(conn->flow_info.fd);
- return -1;
+ if (comp->info.pref_syntax != conn->conn_info.pref_syntax) {
+ log_err("Unknown protocol syntax.");
+ goto fail_cep;
}
switch (id) {
case COMPID_DT:
notifier_event(NOTIFY_DT_CONN_ADD, conn);
-#if defined(BUILD_IPCP_UNICAST) && defined(IPCP_CONN_WAIT_DIR)
- dir_wait_running();
-#endif
break;
case COMPID_MGMT:
notifier_event(NOTIFY_MGMT_CONN_ADD, conn);
@@ -458,6 +560,11 @@ int connmgr_alloc(enum comp_id id,
}
return 0;
+
+ fail_cep:
+ flow_dealloc(conn->flow_info.fd);
+ fail_alloc:
+ return -1;
}
int connmgr_dealloc(enum comp_id id,
@@ -467,7 +574,7 @@ int connmgr_dealloc(enum comp_id id,
case COMPID_DT:
notifier_event(NOTIFY_DT_CONN_DEL, conn);
break;
-#if defined(BUILD_IPCP_UNICAST) && defined(IPCP_CONN_WAIT_DIR)
+#if defined(BUILD_IPCP_UNICAST)
case COMPID_MGMT:
notifier_event(NOTIFY_MGMT_CONN_DEL, conn);
break;
@@ -503,6 +610,7 @@ int connmgr_wait(enum comp_id id,
el = list_first_entry((&comp->pending), struct conn_el, next);
if (el == NULL) {
pthread_mutex_unlock(&comp->lock);
+ log_err("Failed to get connection element.");
return -1;
}
generated by cgit v1.2.3 (git 2.46.0) at 2026-09-12 19:26:59 +0000