summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/irmd/main.c129
-rw-r--r--src/lib/dev.c88
-rw-r--r--src/lib/protobuf.c23
3 files changed, 183 insertions, 57 deletions
diff --git a/src/irmd/main.c b/src/irmd/main.c
index b2f83388..5ef9a82e 100644
--- a/src/irmd/main.c
+++ b/src/irmd/main.c
@@ -1415,12 +1415,103 @@ static int flow_accept(pid_t pid,
return 0;
}
+static int flow_join(pid_t pid,
+ const char * dst,
+ qosspec_t qs,
+ struct timespec * timeo,
+ struct irm_flow * f_out)
+{
+ struct irm_flow * f;
+ struct ipcp_entry * ipcp;
+ int flow_id;
+ int state;
+ uint8_t * hash;
+
+ log_info("Allocating flow for %d to %s.", pid, dst);
+
+ ipcp = get_ipcp_entry_by_layer(dst);
+ if (ipcp == NULL) {
+ log_info("Layer %s unreachable.", dst);
+ return -1;
+ }
+
+ pthread_rwlock_wrlock(&irmd.flows_lock);
+
+ flow_id = bmp_allocate(irmd.flow_ids);
+ if (!bmp_is_id_valid(irmd.flow_ids, flow_id)) {
+ pthread_rwlock_unlock(&irmd.flows_lock);
+ log_err("Could not allocate flow_id.");
+ return -EBADF;
+ }
+
+ f = irm_flow_create(pid, ipcp->pid, flow_id, qs);
+ if (f == NULL) {
+ bmp_release(irmd.flow_ids, flow_id);
+ pthread_rwlock_unlock(&irmd.flows_lock);
+ log_err("Could not allocate flow_id.");
+ return -ENOMEM;
+ }
+
+ list_add(&f->next, &irmd.irm_flows);
+
+ pthread_rwlock_unlock(&irmd.flows_lock);
+
+ assert(irm_flow_get_state(f) == FLOW_ALLOC_PENDING);
+
+ hash = malloc(IPCP_HASH_LEN(ipcp));
+ if (hash == NULL)
+ /* sanitizer cleans this */
+ return -ENOMEM;
+
+ str_hash(ipcp->dir_hash_algo, hash, dst);
+
+ if (ipcp_flow_join(ipcp->pid, flow_id, pid, hash,
+ IPCP_HASH_LEN(ipcp), qs)) {
+ irm_flow_set_state(f, FLOW_NULL);
+ /* sanitizer cleans this */
+ log_info("Flow_join failed.");
+ free(hash);
+ return -EAGAIN;
+ }
+
+ free(hash);
+
+ state = irm_flow_wait_state(f, FLOW_ALLOCATED, timeo);
+ if (state != FLOW_ALLOCATED) {
+ if (state == -ETIMEDOUT) {
+ log_dbg("Flow allocation timed out");
+ return -ETIMEDOUT;
+ }
+
+ log_info("Pending flow to %s torn down.", dst);
+ return -EPIPE;
+ }
+
+ pthread_rwlock_wrlock(&irmd.flows_lock);
+
+ assert(irm_flow_get_state(f) == FLOW_ALLOCATED);
+
+ f_out->flow_id = f->flow_id;
+ f_out->n_pid = f->n_pid;
+ f_out->n_1_pid = f->n_1_pid;
+ f_out->data = f->data; /* pass owner */
+ f_out->len = f->len;
+ f_out->mpl = f->mpl;
+ f->data = NULL;
+ f->len = 0;
+
+ pthread_rwlock_unlock(&irmd.flows_lock);
+
+ log_info("Flow on flow_id %d allocated.", flow_id);
+
+ return 0;
+}
+
static int flow_alloc(pid_t pid,
const char * dst,
qosspec_t qs,
struct timespec * timeo,
struct irm_flow * f_out,
- bool join,
const void * data,
size_t len)
{
@@ -1432,8 +1523,7 @@ static int flow_alloc(pid_t pid,
log_info("Allocating flow for %d to %s.", pid, dst);
- ipcp = join ? get_ipcp_entry_by_layer(dst)
- : get_ipcp_by_dst_name(dst, pid);
+ ipcp = get_ipcp_by_dst_name(dst, pid);
if (ipcp == NULL) {
log_info("Destination %s unreachable.", dst);
return -1;
@@ -1469,24 +1559,13 @@ static int flow_alloc(pid_t pid,
str_hash(ipcp->dir_hash_algo, hash, dst);
- if (join) {
- if (ipcp_flow_join(ipcp->pid, flow_id, pid, hash,
- IPCP_HASH_LEN(ipcp), qs)) {
- irm_flow_set_state(f, FLOW_NULL);
- /* sanitizer cleans this */
- log_info("Flow_join failed.");
- free(hash);
- return -EAGAIN;
- }
- } else {
- if (ipcp_flow_alloc(ipcp->pid, flow_id, pid, hash,
- IPCP_HASH_LEN(ipcp), qs, data, len)) {
- irm_flow_set_state(f, FLOW_NULL);
- /* sanitizer cleans this */
- log_info("Flow_allocation failed.");
- free(hash);
- return -EAGAIN;
- }
+ if (ipcp_flow_alloc(ipcp->pid, flow_id, pid, hash,
+ IPCP_HASH_LEN(ipcp), qs, data, len)) {
+ irm_flow_set_state(f, FLOW_NULL);
+ /* sanitizer cleans this */
+ log_info("Flow_allocation failed.");
+ free(hash);
+ return -EAGAIN;
}
free(hash);
@@ -2136,7 +2215,7 @@ static void * mainloop(void * o)
: msg->pk.data == NULL);
result = flow_alloc(msg->pid, msg->dst,
qos_spec_msg_to_s(msg->qosspec),
- timeo, &e, false, msg->pk.data,
+ timeo, &e, msg->pk.data,
msg->pk.len);
if (result == 0) {
ret_msg->has_flow_id = true;
@@ -2152,9 +2231,9 @@ static void * mainloop(void * o)
break;
case IRM_MSG_CODE__IRM_FLOW_JOIN:
assert(msg->pk.len == 0 && msg->pk.data == NULL);
- result = flow_alloc(msg->pid, msg->dst,
- qos_spec_msg_to_s(msg->qosspec),
- timeo, &e, true, NULL, 0);
+ result = flow_join(msg->pid, msg->dst,
+ qos_spec_msg_to_s(msg->qosspec),
+ timeo, &e);
if (result == 0) {
ret_msg->has_flow_id = true;
ret_msg->flow_id = e.flow_id;
diff --git a/src/lib/dev.c b/src/lib/dev.c
index e84efe55..b0c0175c 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -476,7 +476,6 @@ static int flow_init(int flow_id,
flow->rcv_act = now;
if (qs.cypher_s > 0) {
- assert(s != NULL);
if (crypt_init(&flow->ctx) < 0)
goto fail_ctx;
@@ -822,10 +821,9 @@ int flow_accept(qosspec_t * qs,
return err;
}
-static int __flow_alloc(const char * dst,
- qosspec_t * qs,
- const struct timespec * timeo,
- bool join)
+int flow_alloc(const char * dst,
+ qosspec_t * qs,
+ const struct timespec * timeo)
{
irm_msg_t msg = IRM_MSG__INIT;
irm_msg_t * recv_msg;
@@ -841,12 +839,11 @@ static int __flow_alloc(const char * dst,
if (qs != NULL)
qs->ber = 1;
#endif
- msg.code = join ? IRM_MSG_CODE__IRM_FLOW_JOIN
- : IRM_MSG_CODE__IRM_FLOW_ALLOC;
+ msg.code = IRM_MSG_CODE__IRM_FLOW_ALLOC;
msg.dst = (char *) dst;
msg.has_pid = true;
msg.pid = getpid();
- msg.qosspec = qos_spec_s_to_msg(qs);
+ msg.qosspec = qos_spec_s_to_msg(qs == NULL ? &qos_raw : qs);
if (timeo != NULL) {
msg.has_timeo_sec = true;
@@ -855,7 +852,7 @@ static int __flow_alloc(const char * dst,
msg.timeo_nsec = timeo->tv_nsec;
}
- if (!join && qs != NULL && qs->cypher_s != 0) {
+ if (qs != NULL && qs->cypher_s != 0) {
ssize_t key_len;
key_len = crypt_dh_pkp_create(&pkp, buf);
@@ -887,7 +884,7 @@ static int __flow_alloc(const char * dst,
!recv_msg->has_mpl)
goto fail_result;
- if (!join && qs != NULL && qs->cypher_s != 0) {
+ if (qs != NULL && qs->cypher_s != 0) {
if (!recv_msg->has_pk || recv_msg->pk.len == 0) {
err = -ECRYPT;
goto fail_result;
@@ -902,6 +899,9 @@ static int __flow_alloc(const char * dst,
crypt_dh_pkp_destroy(pkp);
}
+ /* TODO: Make sure qosspec is set in msg */
+ if (qs != NULL && recv_msg->qosspec != NULL)
+ *qs = qos_spec_msg_to_s(recv_msg->qosspec);
fd = flow_init(recv_msg->flow_id, recv_msg->pid,
qs == NULL ? qos_raw : *qs, s,
@@ -909,6 +909,7 @@ static int __flow_alloc(const char * dst,
irm_msg__free_unpacked(recv_msg, NULL);
+
return fd;
fail_result:
@@ -919,21 +920,68 @@ static int __flow_alloc(const char * dst,
return err;
}
-int flow_alloc(const char * dst,
- qosspec_t * qs,
- const struct timespec * timeo)
-{
- return __flow_alloc(dst, qs, timeo, false);
-}
-
int flow_join(const char * dst,
qosspec_t * qs,
const struct timespec * timeo)
{
- if (qs != NULL && qs->cypher_s != 0)
- return -ECRYPT;
+ irm_msg_t msg = IRM_MSG__INIT;
+ irm_msg_t * recv_msg;
+ uint8_t s[SYMMKEYSZ];
+ int fd;
+ int err = -EIRMD;
+
+#ifdef QOS_DISABLE_CRC
+ if (qs != NULL)
+ qs->ber = 1;
+#endif
+ if (qs != NULL && qs->cypher_s > 0)
+ return -ENOTSUP; /* TODO: Encrypted broadcast */
+
+ memset(s, 0, SYMMKEYSZ);
+
+ msg.code = IRM_MSG_CODE__IRM_FLOW_JOIN;
+ msg.dst = (char *) dst;
+ msg.has_pid = true;
+ msg.pid = getpid();
+ msg.qosspec = qos_spec_s_to_msg(qs == NULL ? &qos_raw : qs);
+
+ if (timeo != NULL) {
+ msg.has_timeo_sec = true;
+ msg.has_timeo_nsec = true;
+ msg.timeo_sec = timeo->tv_sec;
+ msg.timeo_nsec = timeo->tv_nsec;
+ }
+
+ recv_msg = send_recv_irm_msg(&msg);
+ qosspec_msg__free_unpacked(msg.qosspec, NULL);
+
+ if (recv_msg == NULL)
+ goto fail_send;
+
+ if (!recv_msg->has_result)
+ goto fail_result;
+
+ if (recv_msg->result != 0) {
+ err = recv_msg->result;
+ goto fail_result;
+ }
- return __flow_alloc(dst, qs, timeo, true);
+ if (!recv_msg->has_pid || !recv_msg->has_flow_id ||
+ !recv_msg->has_mpl)
+ goto fail_result;
+
+ fd = flow_init(recv_msg->flow_id, recv_msg->pid,
+ qs == NULL ? qos_raw : *qs, s,
+ recv_msg->mpl);
+
+ irm_msg__free_unpacked(recv_msg, NULL);
+
+ return fd;
+
+ fail_result:
+ irm_msg__free_unpacked(recv_msg, NULL);
+ fail_send:
+ return err;
}
int flow_dealloc(int fd)
diff --git a/src/lib/protobuf.c b/src/lib/protobuf.c
index 0855305f..830efd40 100644
--- a/src/lib/protobuf.c
+++ b/src/lib/protobuf.c
@@ -301,26 +301,25 @@ struct ipcp_config ipcp_config_msg_to_s(const ipcp_config_msg_t * msg)
qosspec_msg_t * qos_spec_s_to_msg(const struct qos_spec * s)
{
- struct qos_spec spec;
qosspec_msg_t * msg;
+ assert(s != NULL);
+
msg = malloc(sizeof(*msg));
if (msg == NULL)
return NULL;
qosspec_msg__init(msg);
- spec = (s == NULL ? qos_raw : *s);
-
- msg->delay = spec.delay;
- msg->bandwidth = spec.bandwidth;
- msg->availability = spec.availability;
- msg->loss = spec.loss;
- msg->ber = spec.ber;
- msg->in_order = spec.in_order;
- msg->max_gap = spec.max_gap;
- msg->cypher_s = spec.cypher_s;
- msg->timeout = spec.timeout;
+ msg->delay = s->delay;
+ msg->bandwidth = s->bandwidth;
+ msg->availability = s->availability;
+ msg->loss = s->loss;
+ msg->ber = s->ber;
+ msg->in_order = s->in_order;
+ msg->max_gap = s->max_gap;
+ msg->cypher_s = s->cypher_s;
+ msg->timeout = s->timeout;
return msg;
}
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-08 17:59:13 +0000