summaryrefslogtreecommitdiff
path: root/src/ipcpd
diff options
context:
space:
mode:
Diffstat (limited to 'src/ipcpd')
-rw-r--r--src/ipcpd/eth/eth.c5
-rw-r--r--src/ipcpd/ipcp.c56
-rw-r--r--src/ipcpd/local/main.c34
-rw-r--r--src/ipcpd/np1.h41
-rw-r--r--src/ipcpd/udp/udp.c5
-rw-r--r--src/ipcpd/unicast/fa.c11
6 files changed, 128 insertions, 24 deletions
diff --git a/src/ipcpd/eth/eth.c b/src/ipcpd/eth/eth.c
index f36e0b13..9d281297 100644
--- a/src/ipcpd/eth/eth.c
+++ b/src/ipcpd/eth/eth.c
@@ -52,6 +52,7 @@
#include <ouroboros/pthread.h>
#include "ipcp.h"
+#include "np1.h"
#include "shim-data.h"
#include <signal.h>
@@ -990,7 +991,7 @@ static void * eth_ipcp_packet_reader(void * o)
buf = ssm_pk_buff_head(spb);
memcpy(buf, &e_frame->payload, length);
#endif
- if (np1_flow_write(fd, spb) < 0)
+ if (np1_flow_write(fd, spb, NP1_GET_POOL(fd)) < 0)
ipcp_spb_release(spb);
continue;
@@ -1040,7 +1041,7 @@ static void * eth_ipcp_packet_writer(void * o)
if (fqueue_type(fq) != FLOW_PKT)
continue;
- if (np1_flow_read(fd, &spb)) {
+ if (np1_flow_read(fd, &spb, NP1_GET_POOL(fd))) {
log_dbg("Bad read from fd %d.", fd);
continue;
}
diff --git a/src/ipcpd/ipcp.c b/src/ipcpd/ipcp.c
index ebb9b1c5..3ea77da9 100644
--- a/src/ipcpd/ipcp.c
+++ b/src/ipcpd/ipcp.c
@@ -52,6 +52,7 @@
#include <ouroboros/utils.h>
#include "ipcp.h"
+#include "np1.h"
#include <signal.h>
#include <string.h>
@@ -131,6 +132,8 @@ struct {
pthread_t acceptor;
} ipcpd;
+struct np1_state np1;
+
struct cmd {
struct list_head next;
@@ -633,9 +636,11 @@ static void do_flow_alloc(pid_t pid,
uint8_t * dst,
qosspec_t qs,
const buffer_t * data,
+ uid_t uid,
ipcp_msg_t * ret_msg)
{
- int fd;
+ int fd;
+ struct ssm_pool * pool = NULL;
log_info("Allocating flow %d for %d to " HASH_FMT32 ".",
flow_id, pid, HASH_VAL32(dst));
@@ -662,6 +667,17 @@ static void do_flow_alloc(pid_t pid,
return;
}
+ if (uid != 0) {
+ pool = ssm_pool_open(uid);
+ if (pool == NULL) {
+ log_err("Failed to open PUP for uid %d.", uid);
+ ret_msg->result = -ENOMEM;
+ return;
+ }
+ }
+
+ NP1_SET_POOL(fd, pool);
+
ret_msg->result = ipcpd.ops->ipcp_flow_alloc(fd, dst, qs, data);
log_info("Finished allocating flow %d to " HASH_FMT32 ".",
@@ -672,9 +688,11 @@ static void do_flow_alloc(pid_t pid,
static void do_flow_join(pid_t pid,
int flow_id,
const uint8_t * dst,
+ uid_t uid,
ipcp_msg_t * ret_msg)
{
- int fd;
+ int fd;
+ struct ssm_pool * pool = NULL;
log_info("Joining layer " HASH_FMT32 ".", HASH_VAL32(dst));
@@ -699,6 +717,17 @@ static void do_flow_join(pid_t pid,
return;
}
+ if (uid != 0) {
+ pool = ssm_pool_open(uid);
+ if (pool == NULL) {
+ log_err("Failed to open PUP for uid %d.", uid);
+ ret_msg->result = -ENOMEM;
+ return;
+ }
+ }
+
+ NP1_SET_POOL(fd, pool);
+
ret_msg->result = ipcpd.ops->ipcp_flow_join(fd, dst);
log_info("Finished joining layer " HASH_FMT32 ".", HASH_VAL32(dst));
@@ -706,10 +735,12 @@ static void do_flow_join(pid_t pid,
static void do_flow_alloc_resp(int resp,
int flow_id,
+ uid_t uid,
const buffer_t * data,
ipcp_msg_t * ret_msg)
{
- int fd = -1;
+ int fd = -1;
+ struct ssm_pool * pool = NULL;
log_info("Responding %d to alloc on flow_id %d.", resp, flow_id);
@@ -737,6 +768,17 @@ static void do_flow_alloc_resp(int resp,
return;
}
+ if (uid != 0) {
+ pool = ssm_pool_open(uid);
+ if (pool == NULL) {
+ log_err("Failed to open PUP for uid %d.", uid);
+ ret_msg->result = -ENOMEM;
+ return;
+ }
+ }
+
+ NP1_SET_POOL(fd, pool);
+
ret_msg->result = ipcpd.ops->ipcp_flow_alloc_resp(fd, resp, data);
log_info("Finished responding %d to allocation request.",
@@ -857,12 +899,12 @@ static void * mainloop(void * o)
qs = qos_spec_msg_to_s(msg->qosspec);
do_flow_alloc(msg->pid, msg->flow_id,
msg->hash.data, qs,
- &data, &ret_msg);
+ &data, msg->uid, &ret_msg);
break;
case IPCP_MSG_CODE__IPCP_FLOW_JOIN:
assert(msg->hash.len == ipcp_dir_hash_len());
do_flow_join(msg->pid, msg->flow_id,
- msg->hash.data, &ret_msg);
+ msg->hash.data, msg->uid, &ret_msg);
break;
case IPCP_MSG_CODE__IPCP_FLOW_ALLOC_RESP:
assert(msg->pk.len > 0 ? msg->pk.data != NULL
@@ -870,7 +912,7 @@ static void * mainloop(void * o)
data.len = msg->pk.len;
data.data = msg->pk.data;
do_flow_alloc_resp(msg->response, msg->flow_id,
- &data, &ret_msg);
+ msg->uid, &data, &ret_msg);
break;
case IPCP_MSG_CODE__IPCP_FLOW_DEALLOC:
do_flow_dealloc(msg->flow_id, msg->timeo_sec, &ret_msg);
@@ -1035,6 +1077,8 @@ int ipcp_init(int argc,
ipcpd.alloc_id = -1;
+ memset(&np1, 0, sizeof(np1));
+
pthread_condattr_destroy(&cattr);
ipcp_set_state(IPCP_INIT);
diff --git a/src/ipcpd/local/main.c b/src/ipcpd/local/main.c
index e0f3cc5a..5372197f 100644
--- a/src/ipcpd/local/main.c
+++ b/src/ipcpd/local/main.c
@@ -40,6 +40,7 @@
#include <ouroboros/local-dev.h>
#include "ipcp.h"
+#include "np1.h"
#include "shim-data.h"
#include <string.h>
@@ -103,32 +104,41 @@ static void local_data_fini(void){
static void * local_ipcp_packet_loop(void * o)
{
+ int src_fd;
+ int dst_fd;
+ struct timespec * timeout;
+#ifdef CONFIG_IPCP_LOCAL_POLLING
+ struct timespec ts_poll = {0, 0};
+#endif
(void) o;
ipcp_lock_to_core();
- while (true) {
- int fd;
- ssize_t idx;
+#ifdef CONFIG_IPCP_LOCAL_POLLING
+ timeout = &ts_poll; /* Spin poll with zero timeout */
+#else
+ timeout = NULL; /* Block until event */
+#endif
- fevent(local_data.flows, local_data.fq, NULL);
+ while (true) {
+ fevent(local_data.flows, local_data.fq, timeout);
- while ((fd = fqueue_next(local_data.fq)) >= 0) {
+ while ((src_fd = fqueue_next(local_data.fq)) >= 0) {
if (fqueue_type(local_data.fq) != FLOW_PKT)
continue;
- idx = local_flow_read(fd);
- if (idx < 0)
- continue;
-
pthread_rwlock_rdlock(&local_data.lock);
- fd = local_data.in_out[fd];
+ dst_fd = local_data.in_out[src_fd];
pthread_rwlock_unlock(&local_data.lock);
- if (fd != -1)
- local_flow_write(fd, idx);
+ if (dst_fd == -1)
+ continue;
+
+ local_flow_transfer(src_fd, dst_fd,
+ NP1_GET_POOL(src_fd),
+ NP1_GET_POOL(dst_fd));
}
}
diff --git a/src/ipcpd/np1.h b/src/ipcpd/np1.h
new file mode 100644
index 00000000..a3f21200
--- /dev/null
+++ b/src/ipcpd/np1.h
@@ -0,0 +1,41 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2024
+ *
+ * N+1 flow pool tracking for IPCPs
+ *
+ * Dimitri Staessens <dimitri@ouroboros.rocks>
+ * Sander Vrijders <sander@ouroboros.rocks>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2 as
+ * published by the Free Software Foundation.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., http://www.fsf.org/about/contact/.
+ */
+
+#ifndef OUROBOROS_IPCPD_NP1_H
+#define OUROBOROS_IPCPD_NP1_H
+
+#include "config.h"
+
+#include <ouroboros/ssm_pool.h>
+
+#define NP1_LOAD(ptr) (__atomic_load_n((ptr), __ATOMIC_ACQUIRE))
+#define NP1_STORE(ptr, v) (__atomic_store_n((ptr), (v), __ATOMIC_RELEASE))
+#define NP1_GET_POOL(fd) (NP1_LOAD(&np1.pool[(fd)]))
+#define NP1_SET_POOL(fd, p) (NP1_STORE(&np1.pool[(fd)], (p)))
+
+struct np1_state {
+ struct ssm_pool * pool[SYS_MAX_FLOWS];
+};
+
+extern struct np1_state np1;
+
+#endif /* OUROBOROS_IPCPD_NP1_H */
diff --git a/src/ipcpd/udp/udp.c b/src/ipcpd/udp/udp.c
index 5e29cb52..5803e674 100644
--- a/src/ipcpd/udp/udp.c
+++ b/src/ipcpd/udp/udp.c
@@ -34,6 +34,7 @@
#include <ouroboros/pthread.h>
#include "ipcp.h"
+#include "np1.h"
#include "shim-data.h"
#include <string.h>
@@ -492,7 +493,7 @@ static void * udp_ipcp_packet_reader(void * o)
head = ssm_pk_buff_head(spb);
memcpy(head, data, n);
- if (np1_flow_write(eid, spb) < 0)
+ if (np1_flow_write(eid, spb, NP1_GET_POOL(eid)) < 0)
ipcp_spb_release(spb);
}
@@ -536,7 +537,7 @@ static void * udp_ipcp_packet_writer(void * o)
if (fqueue_type(fq) != FLOW_PKT)
continue;
- if (np1_flow_read(fd, &spb)) {
+ if (np1_flow_read(fd, &spb, NP1_GET_POOL(fd))) {
log_dbg("Bad read from fd %d.", fd);
continue;
}
diff --git a/src/ipcpd/unicast/fa.c b/src/ipcpd/unicast/fa.c
index 06e4b043..07f5b46c 100644
--- a/src/ipcpd/unicast/fa.c
+++ b/src/ipcpd/unicast/fa.c
@@ -48,6 +48,7 @@
#include "ipcp.h"
#include "dt.h"
#include "ca.h"
+#include "np1.h"
#include <inttypes.h>
#include <stdlib.h>
@@ -687,6 +688,12 @@ void fa_fini(void)
pthread_rwlock_destroy(&fa.flows_lock);
}
+static int np1_flow_read_fa(int fd,
+ struct ssm_pk_buff ** spb)
+{
+ return np1_flow_read(fd, spb, NP1_GET_POOL(fd));
+}
+
int fa_start(void)
{
#ifndef BUILD_CONTAINER
@@ -695,7 +702,7 @@ int fa_start(void)
int max;
#endif
- fa.psched = psched_create(packet_handler, np1_flow_read);
+ fa.psched = psched_create(packet_handler, np1_flow_read_fa);
if (fa.psched == NULL) {
log_err("Failed to start packet scheduler.");
goto fail_psched;
@@ -963,7 +970,7 @@ void fa_np1_rcv(uint64_t eid,
pthread_rwlock_unlock(&fa.flows_lock);
- if (ipcp_flow_write(fd, spb) < 0) {
+ if (np1_flow_write(fd, spb, NP1_GET_POOL(fd)) < 0) {
log_dbg("Failed to write to flow %d.", fd);
ipcp_spb_release(spb);
#ifdef IPCP_FLOW_STATS
generated by cgit v1.2.3 (git 2.39.1) at 2026-02-04 18:01:28 +0000