summaryrefslogtreecommitdiff
path: root/src/lib/dev.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/lib/dev.c')
-rw-r--r--src/lib/dev.c1737
1 files changed, 1002 insertions, 735 deletions
diff --git a/src/lib/dev.c b/src/lib/dev.c
index c8b7d93b..92310b9e 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -1,5 +1,5 @@
/*
- * Ouroboros - Copyright (C) 2016 - 2021
+ * Ouroboros - Copyright (C) 2016 - 2024
*
* API for applications
*
@@ -28,26 +28,36 @@
#include "config.h"
-#include <ouroboros/hash.h>
-#include <ouroboros/cacep.h>
-#include <ouroboros/errno.h>
+#include <ouroboros/bitmap.h>
+#include <ouroboros/cep.h>
+#include <ouroboros/crypt.h>
#include <ouroboros/dev.h>
+#include <ouroboros/errno.h>
+#include <ouroboros/fccntl.h>
+#include <ouroboros/flow.h>
+#include <ouroboros/fqueue.h>
+#include <ouroboros/hash.h>
+#include <ouroboros/ipcp.h>
#include <ouroboros/ipcp-dev.h>
+#include <ouroboros/list.h>
#include <ouroboros/local-dev.h>
-#include <ouroboros/sockets.h>
-#include <ouroboros/fccntl.h>
-#include <ouroboros/bitmap.h>
+#include <ouroboros/np1_flow.h>
#include <ouroboros/pthread.h>
#include <ouroboros/random.h>
+#include <ouroboros/serdes-irm.h>
#include <ouroboros/shm_flow_set.h>
#include <ouroboros/shm_rdrbuff.h>
#include <ouroboros/shm_rbuff.h>
+#include <ouroboros/sockets.h>
#include <ouroboros/utils.h>
-#include <ouroboros/fqueue.h>
#ifdef PROC_FLOW_STATS
#include <ouroboros/rib.h>
#endif
+#ifdef HAVE_LIBGCRYPT
+#include <gcrypt.h>
+#endif
+#include <assert.h>
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
@@ -65,51 +75,34 @@
#define CRCLEN (sizeof(uint32_t))
#define SECMEMSZ 16384
-#define SYMMKEYSZ 32
#define MSGBUFSZ 2048
-struct flow_set {
- size_t idx;
-};
-
-struct fqueue {
- int fqueue[2 * SHM_BUFFER_SIZE]; /* Safe copy from shm. */
- size_t fqsize;
- size_t next;
-};
-
-enum port_state {
- PORT_NULL = 0,
- PORT_INIT,
- PORT_ID_PENDING,
- PORT_ID_ASSIGNED,
- PORT_DESTROY
-};
-
-struct port {
+/* map flow_ids to flow descriptors; track state of the flow */
+struct fmap {
int fd;
-
- enum port_state state;
- pthread_mutex_t state_lock;
- pthread_cond_t state_cond;
+ /* TODO: use actual flow state */
+ enum flow_state state;
};
#define frcti_to_flow(frcti) \
((struct flow *)((uint8_t *) frcti - offsetof(struct flow, frcti)))
struct flow {
+ struct list_head next;
+
+ struct flow_info info;
+
struct shm_rbuff * rx_rb;
struct shm_rbuff * tx_rb;
struct shm_flow_set * set;
- int flow_id;
- int oflags;
- qosspec_t qs;
+
+ uint16_t oflags;
ssize_t part_idx;
- void * ctx;
- uint8_t key[SYMMKEYSZ];
+ struct crypt_info crypt;
- pid_t pid;
+ struct timespec snd_act;
+ struct timespec rcv_act;
bool snd_timesout;
bool rcv_timesout;
@@ -119,148 +112,299 @@ struct flow {
struct frcti * frcti;
};
-struct {
- char * prog;
- pid_t pid;
+struct flow_set {
+ size_t idx;
+ pthread_rwlock_t lock;
+};
+struct fqueue {
+ struct flowevent fqueue[SHM_BUFFER_SIZE]; /* Safe copy from shm. */
+ size_t fqsize;
+ size_t next;
+};
+
+struct {
struct shm_rdrbuff * rdrb;
struct shm_flow_set * fqset;
- struct timerwheel * tw;
-
struct bmp * fds;
struct bmp * fqueues;
struct flow * flows;
- struct port * ports;
+ struct fmap * id_to_fd;
+ struct list_head flow_list;
+
+ pthread_mutex_t mtx;
+ pthread_cond_t cond;
+
+ pthread_t tx;
+ pthread_t rx;
+ size_t n_frcti;
+ fset_t * frct_set;
pthread_rwlock_t lock;
} ai;
-#include "frct.c"
-
-static void port_destroy(struct port * p)
+static void flow_destroy(struct fmap * p)
{
- pthread_mutex_lock(&p->state_lock);
+ pthread_mutex_lock(&ai.mtx);
- if (p->state == PORT_DESTROY) {
- pthread_mutex_unlock(&p->state_lock);
+ if (p->state == FLOW_DESTROY) {
+ pthread_mutex_unlock(&ai.mtx);
return;
}
- if (p->state == PORT_ID_PENDING)
- p->state = PORT_DESTROY;
+ if (p->state == FLOW_ALLOC_PENDING)
+ p->state = FLOW_DESTROY;
else
- p->state = PORT_NULL;
+ p->state = FLOW_NULL;
+
+ pthread_cond_signal(&ai.cond);
- pthread_cond_signal(&p->state_cond);
+ pthread_cleanup_push(__cleanup_mutex_unlock, &ai.mtx);
- while (p->state != PORT_NULL)
- pthread_cond_wait(&p->state_cond, &p->state_lock);
+ while (p->state != FLOW_NULL)
+ pthread_cond_wait(&ai.cond, &ai.mtx);
- p->fd = -1;
- p->state = PORT_INIT;
+ p->fd = -1;
+ p->state = FLOW_INIT;
- pthread_mutex_unlock(&p->state_lock);
+ pthread_cleanup_pop(true);
}
-static void port_set_state(struct port * p,
- enum port_state state)
+static void flow_set_state(struct fmap * p,
+ enum flow_state state)
{
- pthread_mutex_lock(&p->state_lock);
+ pthread_mutex_lock(&ai.mtx);
- if (p->state == PORT_DESTROY) {
- pthread_mutex_unlock(&p->state_lock);
+ if (p->state == FLOW_DESTROY) {
+ pthread_mutex_unlock(&ai.mtx);
return;
}
p->state = state;
- pthread_cond_broadcast(&p->state_cond);
+ pthread_cond_broadcast(&ai.cond);
- pthread_mutex_unlock(&p->state_lock);
+ pthread_mutex_unlock(&ai.mtx);
}
-static enum port_state port_wait_assign(int flow_id)
+static enum flow_state flow_wait_assign(int flow_id)
{
- enum port_state state;
- struct port * p;
+ enum flow_state state;
+ struct fmap * p;
- p = &ai.ports[flow_id];
+ p = &ai.id_to_fd[flow_id];
- pthread_mutex_lock(&p->state_lock);
+ pthread_mutex_lock(&ai.mtx);
- if (p->state == PORT_ID_ASSIGNED) {
- pthread_mutex_unlock(&p->state_lock);
- return PORT_ID_ASSIGNED;
+ if (p->state == FLOW_ALLOCATED) {
+ pthread_mutex_unlock(&ai.mtx);
+ return FLOW_ALLOCATED;
}
- if (p->state == PORT_INIT)
- p->state = PORT_ID_PENDING;
+ if (p->state == FLOW_INIT)
+ p->state = FLOW_ALLOC_PENDING;
+
+ pthread_cleanup_push(__cleanup_mutex_unlock, &ai.mtx);
- while (p->state == PORT_ID_PENDING)
- pthread_cond_wait(&p->state_cond, &p->state_lock);
+ while (p->state == FLOW_ALLOC_PENDING)
+ pthread_cond_wait(&ai.cond, &ai.mtx);
- if (p->state == PORT_DESTROY) {
- p->state = PORT_NULL;
- pthread_cond_broadcast(&p->state_cond);
+ if (p->state == FLOW_DESTROY) {
+ p->state = FLOW_NULL;
+ pthread_cond_broadcast(&ai.cond);
}
state = p->state;
- assert(state != PORT_INIT);
+ pthread_cleanup_pop(true);
- pthread_mutex_unlock(&p->state_lock);
+ assert(state != FLOW_INIT);
return state;
}
-static int proc_announce(char * prog)
+static int proc_announce(const char * prog)
+{
+ uint8_t buf[SOCK_BUF_SIZE];
+ buffer_t msg = {buf, SOCK_BUF_SIZE};
+ int err;
+
+ if (proc_announce__irm_req_ser(&msg, prog) < 0)
+ return -ENOMEM;
+
+ err = send_recv_msg(&msg);
+ if (err < 0)
+ return err;
+
+ return irm__irm_result_des(&msg);
+}
+
+/* IRMd will clean up the mess if this fails */
+static void proc_exit(void)
{
- irm_msg_t msg = IRM_MSG__INIT;
- irm_msg_t * recv_msg;
- int ret = -1;
-
- msg.code = IRM_MSG_CODE__IRM_PROC_ANNOUNCE;
- msg.has_pid = true;
- msg.pid = ai.pid;
- msg.prog = prog;
-
- recv_msg = send_recv_irm_msg(&msg);
- if (recv_msg == NULL) {
- return -EIRMD;
+ uint8_t buf[SOCK_BUF_SIZE];
+ buffer_t msg = {buf, SOCK_BUF_SIZE};
+
+ if (proc_exit__irm_req_ser(&msg) < 0)
+ return;
+
+ send_recv_msg(&msg);
+}
+
+#include "frct.c"
+
+void * flow_tx(void * o)
+{
+ struct timespec tic = TIMESPEC_INIT_NS(TICTIME);
+
+ (void) o;
+
+ while (true) {
+ timerwheel_move();
+
+ nanosleep(&tic, NULL);
}
- if (!recv_msg->has_result || (ret = recv_msg->result)) {
- irm_msg__free_unpacked(recv_msg, NULL);
- return ret;
+ return (void *) 0;
+}
+
+static void flow_send_keepalive(struct flow * flow,
+ struct timespec now)
+{
+ struct shm_du_buff * sdb;
+ ssize_t idx;
+ uint8_t * ptr;
+
+ idx = shm_rdrbuff_alloc(ai.rdrb, 0, &ptr, &sdb);
+ if (idx < 0)
+ return;
+
+ pthread_rwlock_wrlock(&ai.lock);
+
+ flow->snd_act = now;
+
+ if (shm_rbuff_write(flow->tx_rb, idx))
+ shm_rdrbuff_remove(ai.rdrb, idx);
+ else
+ shm_flow_set_notify(flow->set, flow->info.id, FLOW_PKT);
+
+ pthread_rwlock_unlock(&ai.lock);
+}
+
+/* Needs rdlock on ai. */
+static void _flow_keepalive(struct flow * flow)
+{
+ struct timespec now;
+ struct timespec s_act;
+ struct timespec r_act;
+ int flow_id;
+ time_t timeo;
+ uint32_t acl;
+
+ s_act = flow->snd_act;
+ r_act = flow->rcv_act;
+
+ flow_id = flow->info.id;
+ timeo = flow->info.qs.timeout;
+
+ acl = shm_rbuff_get_acl(flow->rx_rb);
+ if (timeo == 0 || acl & (ACL_FLOWPEER | ACL_FLOWDOWN))
+ return;
+
+ clock_gettime(PTHREAD_COND_CLOCK, &now);
+
+ if (ts_diff_ns(&r_act, &now) > (int64_t) timeo * MILLION) {
+ shm_rbuff_set_acl(flow->rx_rb, ACL_FLOWPEER);
+ shm_flow_set_notify(ai.fqset, flow_id, FLOW_PEER);
+ return;
}
- irm_msg__free_unpacked(recv_msg, NULL);
+ if (ts_diff_ns(&s_act, &now) > (int64_t) timeo * (MILLION >> 2)) {
+ pthread_rwlock_unlock(&ai.lock);
- return ret;
+ flow_send_keepalive(flow, now);
+
+ pthread_rwlock_rdlock(&ai.lock);
+ }
+}
+
+static void handle_keepalives(void)
+{
+ struct list_head * p;
+ struct list_head * h;
+
+ pthread_rwlock_rdlock(&ai.lock);
+
+ list_for_each_safe(p, h, &ai.flow_list) {
+ struct flow * flow;
+ flow = list_entry(p, struct flow, next);
+ _flow_keepalive(flow);
+ }
+
+ pthread_rwlock_unlock(&ai.lock);
+}
+
+static void __cleanup_fqueue_destroy(void * fq)
+{
+ fqueue_destroy((fqueue_t *) fq);
+}
+
+void * flow_rx(void * o)
+{
+ struct timespec tic = TIMESPEC_INIT_NS(TICTIME);
+ int ret;
+ struct fqueue * fq;
+
+ (void) o;
+
+ fq = fqueue_create();
+
+ pthread_cleanup_push(__cleanup_fqueue_destroy, fq);
+
+ /* fevent will filter all FRCT packets for us */
+ while ((ret = fevent(ai.frct_set, fq, &tic)) != 0) {
+ if (ret == -ETIMEDOUT) {
+ handle_keepalives();
+ continue;
+ }
+
+ while (fqueue_next(fq) >= 0)
+ ; /* no need to act */
+ }
+
+ pthread_cleanup_pop(true);
+
+ return (void *) 0;
}
static void flow_clear(int fd)
{
memset(&ai.flows[fd], 0, sizeof(ai.flows[fd]));
- ai.flows[fd].flow_id = -1;
- ai.flows[fd].pid = -1;
+ ai.flows[fd].info.id = -1;
}
-#include "crypt.c"
-
-static void flow_fini(int fd)
+static void __flow_fini(int fd)
{
assert(fd >= 0 && fd < SYS_MAX_FLOWS);
- if (ai.flows[fd].flow_id != -1) {
- port_destroy(&ai.ports[ai.flows[fd].flow_id]);
- bmp_release(ai.fds, fd);
- }
+ if (ai.flows[fd].frcti != NULL) {
+ ai.n_frcti--;
+ if (ai.n_frcti == 0) {
+ pthread_cancel(ai.tx);
+ pthread_join(ai.tx, NULL);
+ }
+
+ shm_flow_set_del(ai.fqset, 0, ai.flows[fd].info.id);
- if (ai.flows[fd].frcti != NULL)
frcti_destroy(ai.flows[fd].frcti);
+ }
+
+ if (ai.flows[fd].info.id != -1) {
+ flow_destroy(&ai.id_to_fd[ai.flows[fd].info.id]);
+ bmp_release(ai.fds, fd);
+ }
if (ai.flows[fd].rx_rb != NULL) {
shm_rbuff_set_acl(ai.flows[fd].rx_rb, ACL_FLOWDOWN);
@@ -274,24 +418,36 @@ static void flow_fini(int fd)
if (ai.flows[fd].set != NULL) {
shm_flow_set_notify(ai.flows[fd].set,
- ai.flows[fd].flow_id,
+ ai.flows[fd].info.id,
FLOW_DEALLOC);
shm_flow_set_close(ai.flows[fd].set);
}
- if (ai.flows[fd].ctx != NULL)
- crypt_fini(ai.flows[fd].ctx);
+ crypt_fini(&ai.flows[fd].crypt);
+
+ list_del(&ai.flows[fd].next);
flow_clear(fd);
}
-static int flow_init(int flow_id,
- pid_t pid,
- qosspec_t qs,
- uint8_t * s)
+static void flow_fini(int fd)
{
- int fd;
- int err = -ENOMEM;
+ pthread_rwlock_wrlock(&ai.lock);
+
+ __flow_fini(fd);
+
+ pthread_rwlock_unlock(&ai.lock);
+}
+
+static int flow_init(struct flow_info * info,
+ buffer_t * sk)
+{
+ struct timespec now;
+ struct flow * flow;
+ int fd;
+ int err = -ENOMEM;
+
+ clock_gettime(PTHREAD_COND_CLOCK, &now);
pthread_rwlock_wrlock(&ai.lock);
@@ -301,46 +457,75 @@ static int flow_init(int flow_id,
goto fail_fds;
}
- ai.flows[fd].rx_rb = shm_rbuff_open(ai.pid, flow_id);
- if (ai.flows[fd].rx_rb == NULL)
+ flow = &ai.flows[fd];
+
+ flow->info = *info;
+
+ flow->rx_rb = shm_rbuff_open(info->n_pid, info->id);
+ if (flow->rx_rb == NULL)
goto fail_rx_rb;
- ai.flows[fd].tx_rb = shm_rbuff_open(pid, flow_id);
- if (ai.flows[fd].tx_rb == NULL)
+ flow->tx_rb = shm_rbuff_open(info->n_1_pid, info->id);
+ if (flow->tx_rb == NULL)
goto fail_tx_rb;
- ai.flows[fd].set = shm_flow_set_open(pid);
- if (ai.flows[fd].set == NULL)
+ flow->set = shm_flow_set_open(info->n_1_pid);
+ if (flow->set == NULL)
goto fail_set;
- ai.flows[fd].flow_id = flow_id;
- ai.flows[fd].oflags = FLOWFDEFAULT;
- ai.flows[fd].pid = pid;
- ai.flows[fd].part_idx = NO_PART;
- ai.flows[fd].qs = qs;
+ flow->oflags = FLOWFDEFAULT;
+ flow->part_idx = NO_PART;
+ flow->snd_act = now;
+ flow->rcv_act = now;
+
+ flow->crypt.flags = info->qs.cypher_s; /* TODO: move cypher_s */
+
+ memset(flow->crypt.key, 0, SYMMKEYSZ);
- if (qs.cypher_s > 0) {
- assert(s != NULL);
- if (crypt_init(&ai.flows[fd].ctx) < 0)
- goto fail_ctx;
+ if (flow->crypt.flags > 0 && sk!= NULL && sk->data != NULL)
+ memcpy(flow->crypt.key, sk->data , sk->len);
+
+ if (crypt_init(&flow->crypt) < 0)
+ goto fail_crypt;
+
+ assert(flow->frcti == NULL);
+
+ if (info->qs.in_order != 0) {
+ flow->frcti = frcti_create(fd, DELT_A, DELT_R, info->mpl);
+ if (flow->frcti == NULL)
+ goto fail_frcti;
- memcpy(ai.flows[fd].key, s, SYMMKEYSZ);
+ if (shm_flow_set_add(ai.fqset, 0, info->id))
+ goto fail_flow_set_add;
+
+ ++ai.n_frcti;
+ if (ai.n_frcti == 1 &&
+ pthread_create(&ai.tx, NULL, flow_tx, NULL) < 0)
+ goto fail_tx_thread;
}
- ai.ports[flow_id].fd = fd;
+ list_add_tail(&flow->next, &ai.flow_list);
+
+ ai.id_to_fd[info->id].fd = fd;
- port_set_state(&ai.ports[flow_id], PORT_ID_ASSIGNED);
+ flow_set_state(&ai.id_to_fd[info->id], FLOW_ALLOCATED);
pthread_rwlock_unlock(&ai.lock);
return fd;
- fail_ctx:
- shm_flow_set_close(ai.flows[fd].set);
+ fail_tx_thread:
+ shm_flow_set_del(ai.fqset, 0, info->id);
+ fail_flow_set_add:
+ frcti_destroy(flow->frcti);
+ fail_frcti:
+ crypt_fini(&flow->crypt);
+ fail_crypt:
+ shm_flow_set_close(flow->set);
fail_set:
- shm_rbuff_close(ai.flows[fd].tx_rb);
+ shm_rbuff_close(flow->tx_rb);
fail_tx_rb:
- shm_rbuff_close(ai.flows[fd].rx_rb);
+ shm_rbuff_close(flow->rx_rb);
fail_rx_rb:
bmp_release(ai.fds, fd);
fail_fds:
@@ -362,110 +547,143 @@ static void init(int argc,
char ** argv,
char ** envp)
{
- const char * prog = argv[0];
- int i;
+ char * prog = argv[0];
+ int i;
#ifdef PROC_FLOW_STATS
- char procstr[32];
+ char procstr[32];
#endif
(void) argc;
(void) envp;
- assert(ai.prog == NULL);
-
if (check_python(argv[0]))
prog = argv[1];
- ai.pid = getpid();
+ prog = path_strip(prog);
+ if (prog == NULL) {
+ fprintf(stderr, "FATAL: Could not determine program name.\n");
+ goto fail_prog;
+ }
+
+ if (proc_announce(prog)) {
+ fprintf(stderr, "FATAL: Could not announce to IRMd.\n");
+ goto fail_prog;
+ }
+
#ifdef HAVE_LIBGCRYPT
if (!gcry_control(GCRYCTL_INITIALIZATION_FINISHED_P)) {
- if (!gcry_check_version(GCRYPT_VERSION))
- goto fail_fds;
+ if (!gcry_check_version(GCRYPT_VERSION)) {
+ fprintf(stderr, "FATAL: Could not get gcry version.\n");
+ goto fail_prog;
+ }
gcry_control(GCRYCTL_DISABLE_SECMEM, 0);
gcry_control(GCRYCTL_INITIALIZATION_FINISHED, 0);
}
#endif
ai.fds = bmp_create(PROG_MAX_FLOWS - PROG_RES_FDS, PROG_RES_FDS);
- if (ai.fds == NULL)
+ if (ai.fds == NULL) {
+ fprintf(stderr, "FATAL: Could not create fd bitmap.\n");
goto fail_fds;
+ }
ai.fqueues = bmp_create(PROG_MAX_FQUEUES, 0);
- if (ai.fqueues == NULL)
+ if (ai.fqueues == NULL) {
+ fprintf(stderr, "FATAL: Could not create fqueue bitmap.\n");
goto fail_fqueues;
+ }
ai.rdrb = shm_rdrbuff_open();
- if (ai.rdrb == NULL)
+ if (ai.rdrb == NULL) {
+ fprintf(stderr, "FATAL: Could not open packet buffer.\n");
goto fail_rdrb;
+ }
ai.flows = malloc(sizeof(*ai.flows) * PROG_MAX_FLOWS);
- if (ai.flows == NULL)
+ if (ai.flows == NULL) {
+ fprintf(stderr, "FATAL: Could not malloc flows.\n");
goto fail_flows;
+ }
for (i = 0; i < PROG_MAX_FLOWS; ++i)
flow_clear(i);
- ai.ports = malloc(sizeof(*ai.ports) * SYS_MAX_FLOWS);
- if (ai.ports == NULL)
- goto fail_ports;
+ ai.id_to_fd = malloc(sizeof(*ai.id_to_fd) * SYS_MAX_FLOWS);
+ if (ai.id_to_fd == NULL) {
+ fprintf(stderr, "FATAL: Could not malloc id_to_fd.\n");
+ goto fail_id_to_fd;
+ }
- if (prog != NULL) {
- ai.prog = strdup(path_strip((char *) prog));
- if (ai.prog == NULL)
- goto fail_prog;
+ for (i = 0; i < SYS_MAX_FLOWS; ++i)
+ ai.id_to_fd[i].state = FLOW_INIT;
- if (proc_announce((char *) ai.prog))
- goto fail_announce;
+ if (pthread_mutex_init(&ai.mtx, NULL)) {
+ fprintf(stderr, "FATAL: Could not init mutex.\n");
+ goto fail_mtx;
}
- for (i = 0; i < SYS_MAX_FLOWS; ++i) {
- ai.ports[i].state = PORT_INIT;
- if (pthread_mutex_init(&ai.ports[i].state_lock, NULL)) {
- int j;
- for (j = 0; j < i; ++j)
- pthread_mutex_destroy(&ai.ports[j].state_lock);
- goto fail_announce;
- }
- if (pthread_cond_init(&ai.ports[i].state_cond, NULL)) {
- int j;
- for (j = 0; j < i; ++j)
- pthread_cond_destroy(&ai.ports[j].state_cond);
- goto fail_state_cond;
- }
+ if (pthread_cond_init(&ai.cond, NULL) < 0) {
+ fprintf(stderr, "FATAL: Could not init condvar.\n");
+ goto fail_cond;
}
- if (pthread_rwlock_init(&ai.lock, NULL))
- goto fail_lock;
+ if (pthread_rwlock_init(&ai.lock, NULL) < 0) {
+ fprintf(stderr, "FATAL: Could not initialize flow lock.\n");
+ goto fail_flow_lock;
+ }
ai.fqset = shm_flow_set_open(getpid());
- if (ai.fqset == NULL)
+ if (ai.fqset == NULL) {
+ fprintf(stderr, "FATAL: Could not open flow set.\n");
goto fail_fqset;
+ }
- if (timerwheel_init() < 0)
+ ai.frct_set = fset_create();
+ if (ai.frct_set == NULL || ai.frct_set->idx != 0) {
+ fprintf(stderr, "FATAL: Could not create FRCT set.\n");
+ goto fail_frct_set;
+ }
+
+ if (timerwheel_init() < 0) {
+ fprintf(stderr, "FATAL: Could not initialize timerwheel.\n");
goto fail_timerwheel;
+ }
#if defined PROC_FLOW_STATS
if (strstr(argv[0], "ipcpd") == NULL) {
sprintf(procstr, "proc.%d", getpid());
- /* Don't bail on fail, it just won't show metrics */
- rib_init(procstr);
+ if (rib_init(procstr) < 0) {
+ fprintf(stderr, "FATAL: Could not initialize RIB.\n");
+ goto fail_rib_init;
+ }
}
#endif
+ if (pthread_create(&ai.rx, NULL, flow_rx, NULL) < 0) {
+ fprintf(stderr, "FATAL: Could not start monitor thread.\n");
+ goto fail_monitor;
+ }
+
+ list_head_init(&ai.flow_list);
+
return;
+ fail_monitor:
+#if defined PROC_FLOW_STATS
+ rib_fini();
+ fail_rib_init:
+#endif
+ timerwheel_fini();
fail_timerwheel:
+ fset_destroy(ai.frct_set);
+ fail_frct_set:
shm_flow_set_close(ai.fqset);
fail_fqset:
pthread_rwlock_destroy(&ai.lock);
- fail_lock:
- for (i = 0; i < SYS_MAX_FLOWS; ++i)
- pthread_cond_destroy(&ai.ports[i].state_cond);
- fail_state_cond:
- for (i = 0; i < SYS_MAX_FLOWS; ++i)
- pthread_mutex_destroy(&ai.ports[i].state_lock);
- fail_announce:
- free(ai.prog);
- fail_prog:
- free(ai.ports);
- fail_ports:
+ fail_flow_lock:
+ pthread_cond_destroy(&ai.cond);
+ fail_cond:
+ pthread_mutex_destroy(&ai.mtx);
+ fail_mtx:
+ free(ai.id_to_fd);
+ fail_id_to_fd:
free(ai.flows);
fail_flows:
shm_rdrbuff_close(ai.rdrb);
@@ -474,57 +692,60 @@ static void init(int argc,
fail_fqueues:
bmp_destroy(ai.fds);
fail_fds:
- fprintf(stderr, "FATAL: ouroboros-dev init failed. "
- "Make sure an IRMd is running.\n\n");
memset(&ai, 0, sizeof(ai));
+ fail_prog:
exit(EXIT_FAILURE);
}
static void fini(void)
{
- int i = 0;
-#ifdef PROC_FLOW_STATS
- rib_fini();
-#endif
+ int i;
if (ai.fds == NULL)
return;
- if (ai.prog != NULL)
- free(ai.prog);
+ pthread_cancel(ai.rx);
+ pthread_join(ai.rx, NULL);
pthread_rwlock_wrlock(&ai.lock);
for (i = 0; i < PROG_MAX_FLOWS; ++i) {
- if (ai.flows[i].flow_id != -1) {
+ if (ai.flows[i].info.id != -1) {
ssize_t idx;
shm_rbuff_set_acl(ai.flows[i].rx_rb, ACL_FLOWDOWN);
while ((idx = shm_rbuff_read(ai.flows[i].rx_rb)) >= 0)
shm_rdrbuff_remove(ai.rdrb, idx);
- flow_fini(i);
+ __flow_fini(i);
}
}
- shm_flow_set_close(ai.fqset);
+ pthread_cond_destroy(&ai.cond);
+ pthread_mutex_destroy(&ai.mtx);
- for (i = 0; i < SYS_MAX_FLOWS; ++i) {
- pthread_mutex_destroy(&ai.ports[i].state_lock);
- pthread_cond_destroy(&ai.ports[i].state_cond);
- }
+ pthread_rwlock_unlock(&ai.lock);
+#ifdef PROC_FLOW_STATS
+ rib_fini();
+#endif
timerwheel_fini();
- shm_rdrbuff_close(ai.rdrb);
+ fset_destroy(ai.frct_set);
+
+ shm_flow_set_close(ai.fqset);
+
+ pthread_rwlock_destroy(&ai.lock);
free(ai.flows);
- free(ai.ports);
+ free(ai.id_to_fd);
+
+ shm_rdrbuff_close(ai.rdrb);
bmp_destroy(ai.fds);
bmp_destroy(ai.fqueues);
- pthread_rwlock_unlock(&ai.lock);
+ proc_exit();
- pthread_rwlock_destroy(&ai.lock);
+ memset(&ai, 0, sizeof(ai));
}
#if defined(__MACH__) && defined(__APPLE__)
@@ -541,310 +762,241 @@ __attribute__((section(FINI_SECTION))) __typeof__(fini) * __fini = fini;
int flow_accept(qosspec_t * qs,
const struct timespec * timeo)
{
- irm_msg_t msg = IRM_MSG__INIT;
- irm_msg_t * recv_msg;
- int fd;
- void * pkp; /* public key pair */
- uint8_t s[SYMMKEYSZ]; /* secret key for flow */
- uint8_t buf[MSGBUFSZ];
- int err = -EIRMD;
- ssize_t key_len;
-
- memset(s, 0, SYMMKEYSZ);
-
- msg.code = IRM_MSG_CODE__IRM_FLOW_ACCEPT;
- msg.has_pid = true;
- msg.pid = ai.pid;
-
- 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;
- }
-
- key_len = crypt_dh_pkp_create(&pkp, buf);
- if (key_len < 0) {
- err = -ECRYPT;
- goto fail_crypt_pkp;
- }
- if (key_len > 0) {
- msg.has_pk = true;
- msg.pk.data = buf;
- msg.pk.len = (uint32_t) key_len;
- }
-
- pthread_cleanup_push(crypt_dh_pkp_destroy, pkp);
-
- recv_msg = send_recv_irm_msg(&msg);
-
- pthread_cleanup_pop(false);
-
- if (recv_msg == NULL)
- goto fail_recv;
-
- if (!recv_msg->has_result)
- goto fail_result;
-
- if (recv_msg->result != 0) {
- err = recv_msg->result;
- goto fail_result;
- }
-
- if (!recv_msg->has_pid || !recv_msg->has_flow_id ||
- recv_msg->qosspec == NULL)
- goto fail_result;
-
- if (recv_msg->pk.len != 0 &&
- crypt_dh_derive(pkp, recv_msg->pk.data,
- recv_msg->pk.len, s) < 0) {
- err = -ECRYPT;
- goto fail_result;
- }
+ struct flow_info flow;
+ uint8_t buf[SOCK_BUF_SIZE];
+ buffer_t msg = {buf, SOCK_BUF_SIZE};
+ buffer_t sk;
+ int fd;
+ int err;
- crypt_dh_pkp_destroy(pkp);
+#ifdef QOS_DISABLE_CRC
+ if (qs != NULL)
+ qs->ber = 1;
+#endif
+ memset(&flow, 0, sizeof(flow));
- fd = flow_init(recv_msg->flow_id, recv_msg->pid,
- msg_to_spec(recv_msg->qosspec), s);
+ flow.n_pid = getpid();
+ flow.qs = qs == NULL ? qos_raw : *qs;
- irm_msg__free_unpacked(recv_msg, NULL);
+ if (flow_accept__irm_req_ser(&msg, &flow, timeo))
+ return -ENOMEM;
- if (fd < 0)
- return fd;
+ err = send_recv_msg(&msg);
+ if (err < 0)
+ return err;
- pthread_rwlock_wrlock(&ai.lock);
+ err = flow__irm_result_des(&msg, &flow, &sk);
+ if (err < 0)
+ return err;
- assert(ai.flows[fd].frcti == NULL);
+ fd = flow_init(&flow, &sk);
- if (ai.flows[fd].qs.in_order != 0) {
- ai.flows[fd].frcti = frcti_create(fd);
- if (ai.flows[fd].frcti == NULL) {
- pthread_rwlock_unlock(&ai.lock);
- flow_dealloc(fd);
- return -ENOMEM;
- }
- }
+ freebuf(sk);
if (qs != NULL)
- *qs = ai.flows[fd].qs;
-
- pthread_rwlock_unlock(&ai.lock);
+ *qs = flow.qs;
return fd;
-
- fail_result:
- irm_msg__free_unpacked(recv_msg, NULL);
- fail_recv:
- crypt_dh_pkp_destroy(pkp);
- fail_crypt_pkp:
- 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;
- qosspec_msg_t qs_msg = QOSSPEC_MSG__INIT;
- irm_msg_t * recv_msg;
- int fd;
- void * pkp = NULL; /* public key pair */
- uint8_t s[SYMMKEYSZ]; /* secret key for flow */
- uint8_t buf[MSGBUFSZ];
- int err = -EIRMD;
-
- memset(s, 0, SYMMKEYSZ);
+ struct flow_info flow;
+ uint8_t buf[SOCK_BUF_SIZE];
+ buffer_t msg = {buf, SOCK_BUF_SIZE};
+ buffer_t sk; /* symmetric key */
+ int fd;
+ int err;
#ifdef QOS_DISABLE_CRC
if (qs != NULL)
qs->ber = 1;
#endif
- msg.code = join ? IRM_MSG_CODE__IRM_FLOW_JOIN
- : IRM_MSG_CODE__IRM_FLOW_ALLOC;
- msg.dst = (char *) dst;
- msg.has_pid = true;
- msg.pid = ai.pid;
- qs_msg = spec_to_msg(qs);
- msg.qosspec = &qs_msg;
- 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;
- }
+ memset(&flow, 0, sizeof(flow));
- if (!join && qs != NULL && qs->cypher_s != 0) {
- ssize_t key_len;
+ flow.n_pid = getpid();
+ flow.qs = qs == NULL ? qos_raw : *qs;
- key_len = crypt_dh_pkp_create(&pkp, buf);
- if (key_len < 0) {
- err = -ECRYPT;
- goto fail_crypt_pkp;
- }
+ if (flow_alloc__irm_req_ser(&msg, &flow, dst, timeo))
+ return -ENOMEM;
- msg.has_pk = true;
- msg.pk.data = buf;
- msg.pk.len = (uint32_t) key_len;
- }
+ err = send_recv_msg(&msg);
+ if (err < 0)
+ return err;
- recv_msg = send_recv_irm_msg(&msg);
- if (recv_msg == NULL)
- goto fail_send;
+ err = flow__irm_result_des(&msg, &flow, &sk);
+ if (err < 0)
+ return err;
- if (!recv_msg->has_result)
- goto fail_result;
+ fd = flow_init(&flow, &sk);
- if (recv_msg->result != 0) {
- err = recv_msg->result;
- goto fail_result;
- }
+ freebuf(sk);
- if (!recv_msg->has_pid || !recv_msg->has_flow_id)
- goto fail_result;
+ if (qs != NULL)
+ *qs = flow.qs;
- if (!join && qs != NULL && qs->cypher_s != 0) {
- if (!recv_msg->has_pk || recv_msg->pk.len == 0) {
- err = -ECRYPT;
- goto fail_result;
- }
+ return fd;
+}
- if (crypt_dh_derive(pkp, recv_msg->pk.data,
- recv_msg->pk.len, s) < 0) {
- err = -ECRYPT;
- goto fail_result;
- }
+int flow_join(const char * dst,
+ qosspec_t * qs,
+ const struct timespec * timeo)
+{
+ struct flow_info flow;
+ uint8_t buf[SOCK_BUF_SIZE];
+ buffer_t msg = {buf, SOCK_BUF_SIZE};
+ int fd;
+ int err;
- crypt_dh_pkp_destroy(pkp);
- }
+#ifdef QOS_DISABLE_CRC
+ if (qs != NULL)
+ qs->ber = 1;
+#endif
+ if (qs != NULL && qs->cypher_s > 0)
+ return -ENOTSUP; /* TODO: Encrypted broadcast */
- fd = flow_init(recv_msg->flow_id, recv_msg->pid,
- qs == NULL ? qos_raw : *qs, s);
+ memset(&flow, 0, sizeof(flow));
- irm_msg__free_unpacked(recv_msg, NULL);
+ flow.n_pid = getpid();
+ flow.qs = qs == NULL ? qos_raw : *qs;
- if (fd < 0)
- return fd;
+ if (flow_alloc__irm_req_ser(&msg, &flow, dst, timeo))
+ return -ENOMEM;
- pthread_rwlock_wrlock(&ai.lock);
+ err = send_recv_msg(&msg);
+ if (err < 0)
+ return err;
- assert(ai.flows[fd].frcti == NULL);
+ err = flow__irm_result_des(&msg, &flow, NULL);
+ if (err < 0)
+ return err;
- if (ai.flows[fd].qs.in_order != 0) {
- ai.flows[fd].frcti = frcti_create(fd);
- if (ai.flows[fd].frcti == NULL) {
- pthread_rwlock_unlock(&ai.lock);
- flow_dealloc(fd);
- return -ENOMEM;
- }
- }
+ fd = flow_init(&flow, NULL);
- pthread_rwlock_unlock(&ai.lock);
+ if (qs != NULL)
+ *qs = flow.qs;
return fd;
-
- fail_result:
- irm_msg__free_unpacked(recv_msg, NULL);
- fail_send:
- crypt_dh_pkp_destroy(pkp);
- fail_crypt_pkp:
- 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;
-
- return __flow_alloc(dst, qs, timeo, true);
}
+#define PKT_BUF_LEN 2048
int flow_dealloc(int fd)
{
- irm_msg_t msg = IRM_MSG__INIT;
- irm_msg_t * recv_msg;
- struct flow * f;
- time_t timeo;
+ struct flow_info info;
+ uint8_t pkt[PKT_BUF_LEN];
+ uint8_t buf[SOCK_BUF_SIZE];
+ buffer_t msg = {buf, SOCK_BUF_SIZE};
+ struct timespec tic = TIMESPEC_INIT_NS(TICTIME);
+ struct timespec timeo = TIMESPEC_INIT_S(0);
+ struct flow * flow;
+ int err;
if (fd < 0 || fd >= SYS_MAX_FLOWS )
return -EINVAL;
- msg.code = IRM_MSG_CODE__IRM_FLOW_DEALLOC;
- msg.has_flow_id = true;
- msg.has_pid = true;
- msg.pid = ai.pid;
- msg.has_timeo_sec = true;
- msg.has_timeo_nsec = true;
- msg.timeo_nsec = 0;
+ memset(&info, 0, sizeof(flow));
- f = &ai.flows[fd];
+ flow = &ai.flows[fd];
pthread_rwlock_rdlock(&ai.lock);
- if (f->flow_id < 0) {
+ if (flow->info.id < 0) {
pthread_rwlock_unlock(&ai.lock);
return -ENOTALLOC;
}
- msg.flow_id = f->flow_id;
+ flow->oflags = FLOWFDEFAULT | FLOWFRNOPART;
- timeo = frcti_dealloc(f->frcti);
- while (timeo < 0) { /* keep the flow active for rtx */
- ssize_t ret;
- uint8_t buf[128];
- struct timespec tic = {0, TICTIME};
+ flow->rcv_timesout = true;
+ flow->rcv_timeo = tic;
+
+ pthread_rwlock_unlock(&ai.lock);
+
+ flow_read(fd, buf, SOCK_BUF_SIZE);
- f->oflags = FLOWFDEFAULT | FLOWFRNOPART;
+ pthread_rwlock_rdlock(&ai.lock);
- f->rcv_timesout = true;
- f->rcv_timeo = tic;
+ timeo.tv_sec = frcti_dealloc(flow->frcti);
+ while (timeo.tv_sec < 0) { /* keep the flow active for rtx */
+ ssize_t ret;
pthread_rwlock_unlock(&ai.lock);
- ret = flow_read(fd, buf, 128);
+ ret = flow_read(fd, pkt, PKT_BUF_LEN);
pthread_rwlock_rdlock(&ai.lock);
- timeo = frcti_dealloc(f->frcti);
+ timeo.tv_sec = frcti_dealloc(flow->frcti);
- if (ret == -EFLOWDOWN && timeo < 0)
- timeo = -timeo;
+ if (ret == -EFLOWDOWN && timeo.tv_sec < 0)
+ timeo.tv_sec = -timeo.tv_sec;
}
- msg.timeo_sec = timeo;
+ pthread_cleanup_push(__cleanup_rwlock_unlock, &ai.lock);
- shm_rbuff_fini(ai.flows[fd].tx_rb);
+ shm_rbuff_fini(flow->tx_rb);
- pthread_rwlock_unlock(&ai.lock);
+ pthread_cleanup_pop(true);
- recv_msg = send_recv_irm_msg(&msg);
- if (recv_msg == NULL)
- return -EIRMD;
+ info.id = flow->info.id;
+ info.n_pid = getpid();
- if (!recv_msg->has_result) {
- irm_msg__free_unpacked(recv_msg, NULL);
- return -EIRMD;
- }
+ if (flow_dealloc__irm_req_ser(&msg, &info, &timeo) < 0)
+ return -ENOMEM;
- irm_msg__free_unpacked(recv_msg, NULL);
+ err = send_recv_msg(&msg);
+ if (err < 0)
+ return err;
- pthread_rwlock_wrlock(&ai.lock);
+ err = irm__irm_result_des(&msg);
flow_fini(fd);
+ return err;
+}
+
+int ipcp_flow_dealloc(int fd)
+{
+ struct flow_info info;
+ uint8_t buf[SOCK_BUF_SIZE];
+ buffer_t msg = {buf, SOCK_BUF_SIZE};
+ struct flow * flow;
+ int err;
+
+ if (fd < 0 || fd >= SYS_MAX_FLOWS )
+ return -EINVAL;
+
+ flow = &ai.flows[fd];
+
+ memset(&info, 0, sizeof(flow));
+
+ pthread_rwlock_rdlock(&ai.lock);
+
+ if (flow->info.id < 0) {
+ pthread_rwlock_unlock(&ai.lock);
+ return -ENOTALLOC;
+ }
+
+ info.id = flow->info.id;
+ info.n_1_pid = flow->info.n_1_pid;
+
pthread_rwlock_unlock(&ai.lock);
- return 0;
+ if (ipcp_flow_dealloc__irm_req_ser(&msg, &info) < 0)
+ return -ENOMEM;
+
+ err = send_recv_msg(&msg);
+ if (err < 0)
+ return err;
+
+ err = irm__irm_result_des(&msg);
+
+ flow_fini(fd);
+
+ return err;
}
int fccntl(int fd,
@@ -853,6 +1005,7 @@ int fccntl(int fd,
{
uint32_t * fflags;
uint16_t * cflags;
+ uint16_t csflags;
va_list l;
struct timespec * timeo;
qosspec_t * qs;
@@ -870,7 +1023,7 @@ int fccntl(int fd,
pthread_rwlock_wrlock(&ai.lock);
- if (flow->flow_id < 0) {
+ if (flow->info.id < 0) {
pthread_rwlock_unlock(&ai.lock);
va_end(l);
return -ENOTALLOC;
@@ -909,13 +1062,13 @@ int fccntl(int fd,
goto einval;
if (!flow->rcv_timesout)
goto eperm;
- *timeo = flow->snd_timeo;
+ *timeo = flow->rcv_timeo;
break;
case FLOWGQOSSPEC:
qs = va_arg(l, qosspec_t *);
if (qs == NULL)
goto einval;
- *qs = flow->qs;
+ *qs = flow->info.qs;
break;
case FLOWGRXQLEN:
qlen = va_arg(l, size_t *);
@@ -942,13 +1095,13 @@ int fccntl(int fd,
rx_acl |= ACL_FLOWDOWN;
tx_acl |= ACL_FLOWDOWN;
shm_flow_set_notify(flow->set,
- flow->flow_id,
+ flow->info.id,
FLOW_DOWN);
} else {
rx_acl &= ~ACL_FLOWDOWN;
tx_acl &= ~ACL_FLOWDOWN;
shm_flow_set_notify(flow->set,
- flow->flow_id,
+ flow->info.id,
FLOW_UP);
}
@@ -963,15 +1116,13 @@ int fccntl(int fd,
*fflags = flow->oflags;
break;
case FRCTSFLAGS:
- cflags = va_arg(l, uint16_t *);
- if (cflags == NULL)
- goto einval;
+ csflags = (uint16_t) va_arg(l, uint32_t);
if (flow->frcti == NULL)
goto eperm;
- frcti_setflags(flow->frcti, *cflags);
+ frcti_setflags(flow->frcti, csflags);
break;
case FRCTGFLAGS:
- cflags = (uint16_t *) va_arg(l, int *);
+ cflags = (uint16_t *) va_arg(l, uint32_t *);
if (cflags == NULL)
goto einval;
if (flow->frcti == NULL)
@@ -1024,6 +1175,60 @@ static int add_crc(struct shm_du_buff * sdb)
return 0;
}
+static int flow_tx_sdb(struct flow * flow,
+ struct shm_du_buff * sdb,
+ bool block,
+ struct timespec * abstime)
+{
+ struct timespec now;
+ ssize_t idx;
+ int ret;
+
+ clock_gettime(PTHREAD_COND_CLOCK, &now);
+
+ pthread_rwlock_wrlock(&ai.lock);
+
+ flow->snd_act = now;
+
+ pthread_rwlock_unlock(&ai.lock);
+
+ idx = shm_du_buff_get_idx(sdb);
+
+ pthread_rwlock_rdlock(&ai.lock);
+
+ if (shm_du_buff_len(sdb) > 0) {
+ if (frcti_snd(flow->frcti, sdb) < 0)
+ goto enomem;
+
+ if (crypt_encrypt(&flow->crypt, sdb) < 0)
+ goto enomem;
+
+ if (flow->info.qs.ber == 0 && add_crc(sdb) != 0)
+ goto enomem;
+ }
+
+ pthread_cleanup_push(__cleanup_rwlock_unlock, &ai.lock);
+
+ if (!block)
+ ret = shm_rbuff_write(flow->tx_rb, idx);
+ else
+ ret = shm_rbuff_write_b(flow->tx_rb, idx, abstime);
+
+ if (ret < 0)
+ shm_rdrbuff_remove(ai.rdrb, idx);
+ else
+ shm_flow_set_notify(flow->set, flow->info.id, FLOW_PKT);
+
+ pthread_cleanup_pop(true);
+
+ return 0;
+
+enomem:
+ pthread_rwlock_unlock(&ai.lock);
+ shm_rdrbuff_remove(ai.rdrb, idx);
+ return -ENOMEM;
+}
+
ssize_t flow_write(int fd,
const void * buf,
size_t count)
@@ -1034,31 +1239,27 @@ ssize_t flow_write(int fd,
int flags;
struct timespec abs;
struct timespec * abstime = NULL;
- struct timespec tic = {0, TICTIME};
- struct timespec tictime;
struct shm_du_buff * sdb;
uint8_t * ptr;
- if (buf == NULL)
- return 0;
+ if (buf == NULL && count != 0)
+ return -EINVAL;
- if (fd < 0 || fd > PROG_MAX_FLOWS)
+ if (fd < 0 || fd >= PROG_MAX_FLOWS)
return -EBADF;
flow = &ai.flows[fd];
clock_gettime(PTHREAD_COND_CLOCK, &abs);
- pthread_rwlock_rdlock(&ai.lock);
+ pthread_rwlock_wrlock(&ai.lock);
- if (flow->flow_id < 0) {
+ if (flow->info.id < 0) {
pthread_rwlock_unlock(&ai.lock);
return -ENOTALLOC;
}
- ts_add(&tic, &abs, &tictime);
-
- if (ai.flows[fd].snd_timesout) {
+ if (flow->snd_timesout) {
ts_add(&abs, &flow->snd_timeo, &abs);
abstime = &abs;
}
@@ -1075,54 +1276,66 @@ ssize_t flow_write(int fd,
return -EAGAIN;
idx = shm_rdrbuff_alloc(ai.rdrb, count, &ptr, &sdb);
} else {
- while ((ret = frcti_window_wait(flow->frcti, &tictime)) < 0) {
- if (ret != -ETIMEDOUT)
- return ret;
-
- if (abstime != NULL && ts_diff_ns(&tictime, &abs) <= 0)
- return -ETIMEDOUT;
-
- frcti_tick(flow->frcti);
-
- ts_add(&tictime, &tic, &tictime);
- }
+ ret = frcti_window_wait(flow->frcti, abstime);
+ if (ret < 0)
+ return ret;
idx = shm_rdrbuff_alloc_b(ai.rdrb, count, &ptr, &sdb, abstime);
}
if (idx < 0)
return idx;
- memcpy(ptr, buf, count);
+ if (count > 0)
+ memcpy(ptr, buf, count);
- pthread_rwlock_rdlock(&ai.lock);
+ ret = flow_tx_sdb(flow, sdb, !(flags & FLOWFWNOBLOCK), abstime);
+
+ return ret < 0 ? (ssize_t) ret : (ssize_t) count;
+}
+
+static bool invalid_pkt(struct flow * flow,
+ struct shm_du_buff * sdb)
+{
+ if (shm_du_buff_len(sdb) == 0)
+ return true;
- if (frcti_snd(flow->frcti, sdb) < 0)
- goto enomem;
+ if (flow->info.qs.ber == 0 && chk_crc(sdb) != 0)
+ return true;
- if (flow->qs.cypher_s > 0 && crypt_encrypt(flow, sdb) < 0)
- goto enomem;
+ if (crypt_decrypt(&flow->crypt, sdb) < 0)
+ return true;
- if (flow->qs.ber == 0 && add_crc(sdb) != 0)
- goto enomem;
+ return false;
+}
- if (flags & FLOWFWNOBLOCK)
- ret = shm_rbuff_write(flow->tx_rb, idx);
- else
- ret = shm_rbuff_write_b(flow->tx_rb, idx, abstime);
+static ssize_t flow_rx_sdb(struct flow * flow,
+ struct shm_du_buff ** sdb,
+ bool block,
+ struct timespec * abstime)
+{
+ ssize_t idx;
+ struct timespec now;
- if (ret < 0)
- shm_rdrbuff_remove(ai.rdrb, idx);
- else
- shm_flow_set_notify(flow->set, flow->flow_id, FLOW_PKT);
+ idx = block ? shm_rbuff_read_b(flow->rx_rb, abstime) :
+ shm_rbuff_read(flow->rx_rb);
+ if (idx < 0)
+ return idx;
- pthread_rwlock_unlock(&ai.lock);
+ clock_gettime(PTHREAD_COND_CLOCK, &now);
- return ret < 0 ? (ssize_t) ret : (ssize_t) count;
+ pthread_rwlock_wrlock(&ai.lock);
+
+ flow->rcv_act = now;
- enomem:
pthread_rwlock_unlock(&ai.lock);
- shm_rdrbuff_remove(ai.rdrb, idx);
- return -ENOMEM;
+
+ *sdb = shm_rdrbuff_get(ai.rdrb, idx);
+ if (invalid_pkt(flow, *sdb)) {
+ shm_rdrbuff_remove(ai.rdrb, idx);
+ return -EAGAIN;
+ }
+
+ return idx;
}
ssize_t flow_read(int fd,
@@ -1132,120 +1345,100 @@ ssize_t flow_read(int fd,
ssize_t idx;
ssize_t n;
uint8_t * packet;
- struct shm_rbuff * rb;
struct shm_du_buff * sdb;
struct timespec abs;
- struct timespec tic = {0, TICTIME};
- struct timespec tictime;
+ struct timespec now;
struct timespec * abstime = NULL;
struct flow * flow;
- bool noblock;
+ bool block;
bool partrd;
- if (fd < 0 || fd > PROG_MAX_FLOWS)
+ if (fd < 0 || fd >= PROG_MAX_FLOWS)
return -EBADF;
flow = &ai.flows[fd];
- clock_gettime(PTHREAD_COND_CLOCK, &abs);
+ clock_gettime(PTHREAD_COND_CLOCK, &now);
pthread_rwlock_rdlock(&ai.lock);
- if (flow->part_idx == DONE_PART) {
+ if (flow->info.id < 0) {
pthread_rwlock_unlock(&ai.lock);
- flow->part_idx = NO_PART;
- return 0;
+ return -ENOTALLOC;
}
- if (flow->flow_id < 0) {
+ if (flow->part_idx == DONE_PART) {
pthread_rwlock_unlock(&ai.lock);
- return -ENOTALLOC;
+ flow->part_idx = NO_PART;
+ return 0;
}
- rb = flow->rx_rb;
- noblock = flow->oflags & FLOWFRNOBLOCK;
+ block = !(flow->oflags & FLOWFRNOBLOCK);
partrd = !(flow->oflags & FLOWFRNOPART);
- ts_add(&tic, &abs, &tictime);
-
- if (ai.flows[fd].rcv_timesout) {
- ts_add(&abs, &flow->rcv_timeo, &abs);
+ if (flow->rcv_timesout) {
+ ts_add(&now, &flow->rcv_timeo, &abs);
abstime = &abs;
}
idx = flow->part_idx;
-
if (idx < 0) {
while ((idx = frcti_queued_pdu(flow->frcti)) < 0) {
pthread_rwlock_unlock(&ai.lock);
- idx = noblock ? shm_rbuff_read(rb) :
- shm_rbuff_read_b(rb, &tictime);
+ idx = flow_rx_sdb(flow, &sdb, block, abstime);
if (idx < 0) {
- frcti_tick(flow->frcti);
-
- if (idx != -ETIMEDOUT)
+ if (block && idx != -EAGAIN)
+ return idx;
+ if (!block)
return idx;
- if (abstime != NULL
- && ts_diff_ns(&tictime, &abs) <= 0)
- return -ETIMEDOUT;
-
- ts_add(&tictime, &tic, &tictime);
- pthread_rwlock_rdlock(&ai.lock);
- continue;
- }
-
- sdb = shm_rdrbuff_get(ai.rdrb, idx);
- if (flow->qs.ber == 0 && chk_crc(sdb) != 0) {
pthread_rwlock_rdlock(&ai.lock);
- shm_rdrbuff_remove(ai.rdrb, idx);
continue;
}
pthread_rwlock_rdlock(&ai.lock);
- if (flow->qs.cypher_s > 0
- && crypt_decrypt(flow, sdb) < 0) {
- pthread_rwlock_unlock(&ai.lock);
- shm_rdrbuff_remove(ai.rdrb, idx);
- return -ENOMEM;
- }
-
frcti_rcv(flow->frcti, sdb);
}
}
- frcti_tick(flow->frcti);
+ sdb = shm_rdrbuff_get(ai.rdrb, idx);
pthread_rwlock_unlock(&ai.lock);
- n = shm_rdrbuff_read(&packet, ai.rdrb, idx);
+ packet = shm_du_buff_head(sdb);
+
+ n = shm_du_buff_len(sdb);
assert(n >= 0);
if (n <= (ssize_t) count) {
memcpy(buf, packet, n);
- shm_rdrbuff_remove(ai.rdrb, idx);
+ ipcp_sdb_release(sdb);
pthread_rwlock_wrlock(&ai.lock);
flow->part_idx = (partrd && n == (ssize_t) count) ?
DONE_PART : NO_PART;
+ flow->rcv_act = now;
+
pthread_rwlock_unlock(&ai.lock);
return n;
} else {
if (partrd) {
memcpy(buf, packet, count);
- sdb = shm_rdrbuff_get(ai.rdrb, idx);
shm_du_buff_head_release(sdb, n);
pthread_rwlock_wrlock(&ai.lock);
flow->part_idx = idx;
+
+ flow->rcv_act = now;
+
pthread_rwlock_unlock(&ai.lock);
return count;
} else {
- shm_rdrbuff_remove(ai.rdrb, idx);
+ ipcp_sdb_release(sdb);
return -EMSGSIZE;
}
}
@@ -1253,26 +1446,31 @@ ssize_t flow_read(int fd,
/* fqueue functions. */
-struct flow_set * fset_create()
+struct flow_set * fset_create(void)
{
- struct flow_set * set = malloc(sizeof(*set));
+ struct flow_set * set;
+
+ set = malloc(sizeof(*set));
if (set == NULL)
- return NULL;
+ goto fail_malloc;
assert(ai.fqueues);
pthread_rwlock_wrlock(&ai.lock);
set->idx = bmp_allocate(ai.fqueues);
- if (!bmp_is_id_valid(ai.fqueues, set->idx)) {
- pthread_rwlock_unlock(&ai.lock);
- free(set);
- return NULL;
- }
+ if (!bmp_is_id_valid(ai.fqueues, set->idx))
+ goto fail_bmp_alloc;
pthread_rwlock_unlock(&ai.lock);
return set;
+
+ fail_bmp_alloc:
+ pthread_rwlock_unlock(&ai.lock);
+ free(set);
+ fail_malloc:
+ return NULL;
}
void fset_destroy(struct flow_set * set)
@@ -1291,13 +1489,13 @@ void fset_destroy(struct flow_set * set)
free(set);
}
-struct fqueue * fqueue_create()
+struct fqueue * fqueue_create(void)
{
struct fqueue * fq = malloc(sizeof(*fq));
if (fq == NULL)
return NULL;
- memset(fq->fqueue, -1, (SHM_BUFFER_SIZE) * sizeof(*fq->fqueue));
+ memset(fq->fqueue, -1, SHM_BUFFER_SIZE * sizeof(*fq->fqueue));
fq->fqsize = 0;
fq->next = 0;
@@ -1320,41 +1518,57 @@ void fset_zero(struct flow_set * set)
int fset_add(struct flow_set * set,
int fd)
{
- int ret;
- size_t packets;
- size_t i;
+ struct flow * flow;
+ int ret;
- if (set == NULL || fd < 0 || fd > SYS_MAX_FLOWS)
+ if (set == NULL || fd < 0 || fd >= SYS_MAX_FLOWS)
return -EINVAL;
- pthread_rwlock_wrlock(&ai.lock);
+ flow = &ai.flows[fd];
- if (ai.flows[fd].flow_id < 0) {
- pthread_rwlock_unlock(&ai.lock);
- return -EINVAL;
+ pthread_rwlock_rdlock(&ai.lock);
+
+ if (flow->info.id < 0) {
+ ret = -EINVAL;
+ goto fail;
}
- ret = shm_flow_set_add(ai.fqset, set->idx, ai.flows[fd].flow_id);
+ if (flow->frcti != NULL)
+ shm_flow_set_del(ai.fqset, 0, ai.flows[fd].info.id);
+
+ ret = shm_flow_set_add(ai.fqset, set->idx, ai.flows[fd].info.id);
+ if (ret < 0)
+ goto fail;
- packets = shm_rbuff_queued(ai.flows[fd].rx_rb);
- for (i = 0; i < packets; i++)
- shm_flow_set_notify(ai.fqset, ai.flows[fd].flow_id, FLOW_PKT);
+ if (shm_rbuff_queued(ai.flows[fd].rx_rb))
+ shm_flow_set_notify(ai.fqset, ai.flows[fd].info.id, FLOW_PKT);
pthread_rwlock_unlock(&ai.lock);
return ret;
+
+ fail:
+ pthread_rwlock_unlock(&ai.lock);
+ return ret;
}
void fset_del(struct flow_set * set,
int fd)
{
- if (set == NULL || fd < 0 || fd > SYS_MAX_FLOWS)
+ struct flow * flow;
+
+ if (set == NULL || fd < 0 || fd >= SYS_MAX_FLOWS)
return;
+ flow = &ai.flows[fd];
+
pthread_rwlock_rdlock(&ai.lock);
- if (ai.flows[fd].flow_id >= 0)
- shm_flow_set_del(ai.fqset, set->idx, ai.flows[fd].flow_id);
+ if (flow->info.id >= 0)
+ shm_flow_set_del(ai.fqset, set->idx, flow->info.id);
+
+ if (flow->frcti != NULL)
+ shm_flow_set_add(ai.fqset, 0, ai.flows[fd].info.id);
pthread_rwlock_unlock(&ai.lock);
}
@@ -1362,28 +1576,86 @@ void fset_del(struct flow_set * set,
bool fset_has(const struct flow_set * set,
int fd)
{
- bool ret = false;
+ bool ret;
- if (set == NULL || fd < 0 || fd > SYS_MAX_FLOWS)
+ if (set == NULL || fd < 0 || fd >= SYS_MAX_FLOWS)
return false;
pthread_rwlock_rdlock(&ai.lock);
- if (ai.flows[fd].flow_id < 0) {
+ if (ai.flows[fd].info.id < 0) {
pthread_rwlock_unlock(&ai.lock);
return false;
}
- ret = (shm_flow_set_has(ai.fqset, set->idx, ai.flows[fd].flow_id) == 1);
+ ret = (shm_flow_set_has(ai.fqset, set->idx, ai.flows[fd].info.id) == 1);
pthread_rwlock_unlock(&ai.lock);
return ret;
}
+/* Filter fqueue events for non-data packets */
+static int fqueue_filter(struct fqueue * fq)
+{
+ struct shm_du_buff * sdb;
+ int fd;
+ ssize_t idx;
+ struct frcti * frcti;
+
+ while (fq->next < fq->fqsize) {
+ if (fq->fqueue[fq->next].event != FLOW_PKT)
+ return 1;
+
+ pthread_rwlock_rdlock(&ai.lock);
+
+ fd = ai.id_to_fd[fq->fqueue[fq->next].flow_id].fd;
+ if (fd < 0) {
+ ++fq->next;
+ pthread_rwlock_unlock(&ai.lock);
+ continue;
+ }
+
+ frcti = ai.flows[fd].frcti;
+ if (frcti == NULL) {
+ pthread_rwlock_unlock(&ai.lock);
+ return 1;
+ }
+
+ if (__frcti_pdu_ready(frcti) >= 0) {
+ pthread_rwlock_unlock(&ai.lock);
+ return 1;
+ }
+
+ pthread_rwlock_unlock(&ai.lock);
+
+ idx = flow_rx_sdb(&ai.flows[fd], &sdb, false, NULL);
+ if (idx < 0)
+ return 0;
+
+ pthread_rwlock_rdlock(&ai.lock);
+
+ sdb = shm_rdrbuff_get(ai.rdrb, idx);
+
+ __frcti_rcv(frcti, sdb);
+
+ if (__frcti_pdu_ready(frcti) >= 0) {
+ pthread_rwlock_unlock(&ai.lock);
+ return 1;
+ }
+
+ pthread_rwlock_unlock(&ai.lock);
+
+ ++fq->next;
+ }
+
+ return 0;
+}
+
int fqueue_next(struct fqueue * fq)
{
- int fd;
+ int fd;
+ struct flowevent * e;
if (fq == NULL)
return -EINVAL;
@@ -1391,16 +1663,16 @@ int fqueue_next(struct fqueue * fq)
if (fq->fqsize == 0 || fq->next == fq->fqsize)
return -EPERM;
+ if (fq->next != 0 && fqueue_filter(fq) == 0)
+ return -EPERM;
+
pthread_rwlock_rdlock(&ai.lock);
- if (fq->next != 0 && frcti_filter(fq) == 0) {
- pthread_rwlock_unlock(&ai.lock);
- return -EPERM;
- }
+ e = fq->fqueue + fq->next;
- fd = ai.ports[fq->fqueue[fq->next]].fd;
+ fd = ai.id_to_fd[e->flow_id].fd;
- fq->next += 2;
+ ++fq->next;
pthread_rwlock_unlock(&ai.lock);
@@ -1415,7 +1687,7 @@ enum fqtype fqueue_type(struct fqueue * fq)
if (fq->fqsize == 0 || fq->next == 0)
return -EPERM;
- return fq->fqueue[fq->next - 1];
+ return fq->fqueue[(fq->next - 1)].event;
}
ssize_t fevent(struct flow_set * set,
@@ -1423,8 +1695,6 @@ ssize_t fevent(struct flow_set * set,
const struct timespec * timeo)
{
ssize_t ret = 0;
- struct timespec tic = {0, TICTIME};
- struct timespec tictime;
struct timespec abs;
struct timespec * t = NULL;
@@ -1436,46 +1706,43 @@ ssize_t fevent(struct flow_set * set,
clock_gettime(PTHREAD_COND_CLOCK, &abs);
- ts_add(&tic, &abs, &tictime);
- t = &tictime;
-
- if (timeo != NULL)
+ if (timeo != NULL) {
ts_add(&abs, timeo, &abs);
+ t = &abs;
+ }
while (ret == 0) {
ret = shm_flow_set_wait(ai.fqset, set->idx, fq->fqueue, t);
- if (ret == -ETIMEDOUT) {
- if (timeo != NULL && ts_diff_ns(t, &abs) < 0) {
- fq->fqsize = 0;
- return -ETIMEDOUT;
- }
- ret = 0;
- ts_add(t, &tic, t);
- pthread_rwlock_rdlock(&ai.lock);
- timerwheel_move();
- pthread_rwlock_unlock(&ai.lock);
- continue;
- }
+ if (ret == -ETIMEDOUT)
+ return -ETIMEDOUT;
- fq->fqsize = ret << 1;
+ fq->fqsize = ret;
fq->next = 0;
- ret = frcti_filter(fq);
+ ret = fqueue_filter(fq);
}
- assert(ret);
+ assert(ret != 0);
return 1;
}
/* ipcp-dev functions. */
-int np1_flow_alloc(pid_t n_pid,
- int flow_id,
- qosspec_t qs)
+int np1_flow_alloc(pid_t n_pid,
+ int flow_id)
{
- qs.cypher_s = 0; /* No encryption ctx for np1 */
- return flow_init(flow_id, n_pid, qs, NULL);
+ struct flow_info flow;
+
+ memset(&flow, 0, sizeof(flow));
+
+ flow.id = flow_id;
+ flow.n_pid = getpid();
+ flow.qs = qos_np1;
+ flow.mpl = 0;
+ flow.n_1_pid = n_pid; /* This "flow" is upside-down! */
+
+ return flow_init(&flow, NULL);
}
int np1_flow_dealloc(int flow_id,
@@ -1493,7 +1760,7 @@ int np1_flow_dealloc(int flow_id,
pthread_rwlock_rdlock(&ai.lock);
- fd = ai.ports[flow_id].fd;
+ fd = ai.id_to_fd[flow_id].fd;
pthread_rwlock_unlock(&ai.lock);
@@ -1504,140 +1771,106 @@ int np1_flow_resp(int flow_id)
{
int fd;
- if (port_wait_assign(flow_id) != PORT_ID_ASSIGNED)
+ if (flow_wait_assign(flow_id) != FLOW_ALLOCATED)
return -1;
pthread_rwlock_rdlock(&ai.lock);
- fd = ai.ports[flow_id].fd;
+ fd = ai.id_to_fd[flow_id].fd;
pthread_rwlock_unlock(&ai.lock);
return fd;
}
-int ipcp_create_r(int result)
+int ipcp_create_r(const struct ipcp_info * info)
{
- irm_msg_t msg = IRM_MSG__INIT;
- irm_msg_t * recv_msg;
- int ret;
-
- msg.code = IRM_MSG_CODE__IPCP_CREATE_R;
- msg.has_pid = true;
- msg.pid = getpid();
- msg.has_result = true;
- msg.result = result;
-
- recv_msg = send_recv_irm_msg(&msg);
- if (recv_msg == NULL)
- return -EIRMD;
-
- if (!recv_msg->has_result) {
- irm_msg__free_unpacked(recv_msg, NULL);
- return -1;
- }
+ uint8_t buf[SOCK_BUF_SIZE];
+ buffer_t msg = {buf, SOCK_BUF_SIZE};
+ int err;
- ret = recv_msg->result;
- irm_msg__free_unpacked(recv_msg, NULL);
+ if (ipcp_create_r__irm_req_ser(&msg,info) < 0)
+ return -ENOMEM;
- return ret;
+ err = send_recv_msg(&msg);
+ if (err < 0)
+ return err;
+
+ return irm__irm_result_des(&msg);
}
-int ipcp_flow_req_arr(const uint8_t * dst,
- size_t len,
- qosspec_t qs,
- const void * data,
- size_t dlen)
+int ipcp_flow_req_arr(const buffer_t * dst,
+ qosspec_t qs,
+ time_t mpl,
+ const buffer_t * data)
{
- irm_msg_t msg = IRM_MSG__INIT;
- irm_msg_t * recv_msg;
- qosspec_msg_t qs_msg;
- int fd;
-
- assert(dst != NULL);
-
- msg.code = IRM_MSG_CODE__IPCP_FLOW_REQ_ARR;
- msg.has_pid = true;
- msg.pid = getpid();
- msg.has_hash = true;
- msg.hash.len = len;
- msg.hash.data = (uint8_t *) dst;
- qs_msg = spec_to_msg(&qs);
- msg.qosspec = &qs_msg;
- msg.has_pk = true;
- msg.pk.data = (uint8_t *) data;
- msg.pk.len = dlen;
-
- recv_msg = send_recv_irm_msg(&msg);
- if (recv_msg == NULL)
- return -EIRMD;
-
- if (!recv_msg->has_flow_id || !recv_msg->has_pid) {
- irm_msg__free_unpacked(recv_msg, NULL);
- return -1;
- }
+ struct flow_info flow;
+ uint8_t buf[SOCK_BUF_SIZE];
+ buffer_t msg = {buf, SOCK_BUF_SIZE};
+ int err;
- if (recv_msg->has_result && recv_msg->result) {
- irm_msg__free_unpacked(recv_msg, NULL);
- return -1;
- }
+ memset(&flow, 0, sizeof(flow));
- qs.cypher_s = 0; /* No encryption ctx for np1 */
- fd = flow_init(recv_msg->flow_id, recv_msg->pid, qs, NULL);
+ assert(dst != NULL && dst->len != 0 && dst->data != NULL);
- irm_msg__free_unpacked(recv_msg, NULL);
+ flow.n_1_pid = getpid();
+ flow.qs = qs;
+ flow.mpl = mpl;
- return fd;
+ if (ipcp_flow_req_arr__irm_req_ser(&msg, dst, &flow, data) < 0)
+ return -ENOMEM;
+
+ err = send_recv_msg(&msg);
+ if (err < 0)
+ return err;
+
+ err = flow__irm_result_des(&msg, &flow, NULL);
+ if (err < 0)
+ return err;
+
+ /* inverted for np1_flow */
+ flow.n_1_pid = flow.n_pid;
+ flow.n_pid = getpid();
+ flow.mpl = 0;
+
+ return flow_init(&flow, NULL);
}
-int ipcp_flow_alloc_reply(int fd,
- int response,
- const void * data,
- size_t len)
+int ipcp_flow_alloc_reply(int fd,
+ int response,
+ time_t mpl,
+ const buffer_t * data)
{
- irm_msg_t msg = IRM_MSG__INIT;
- irm_msg_t * recv_msg;
- int ret;
+ struct flow_info flow;
+ uint8_t buf[SOCK_BUF_SIZE];
+ buffer_t msg = {buf, SOCK_BUF_SIZE};
+ int err;
assert(fd >= 0 && fd < SYS_MAX_FLOWS);
- msg.code = IRM_MSG_CODE__IPCP_FLOW_ALLOC_REPLY;
- msg.has_flow_id = true;
- msg.has_pk = true;
- msg.pk.data = (uint8_t *) data;
- msg.pk.len = (uint32_t) len;
-
pthread_rwlock_rdlock(&ai.lock);
- msg.flow_id = ai.flows[fd].flow_id;
+ flow.id = ai.flows[fd].info.id;
pthread_rwlock_unlock(&ai.lock);
- msg.has_response = true;
- msg.response = response;
+ flow.mpl = mpl;
- recv_msg = send_recv_irm_msg(&msg);
- if (recv_msg == NULL)
- return -EIRMD;
-
- if (!recv_msg->has_result) {
- irm_msg__free_unpacked(recv_msg, NULL);
- return -1;
- }
-
- ret = recv_msg->result;
+ if (ipcp_flow_alloc_reply__irm_msg_ser(&msg, &flow, response, data) < 0)
+ return -ENOMEM;
- irm_msg__free_unpacked(recv_msg, NULL);
+ err = send_recv_msg(&msg);
+ if (err < 0)
+ return err;
- return ret;
+ return irm__irm_result_des(&msg);
}
int ipcp_flow_read(int fd,
struct shm_du_buff ** sdb)
{
- struct flow * flow;
- struct shm_rbuff * rb;
- ssize_t idx = -1;
+ struct flow * flow;
+ ssize_t idx = -1;
assert(fd >= 0 && fd < SYS_MAX_FLOWS);
assert(sdb);
@@ -1646,50 +1879,39 @@ int ipcp_flow_read(int fd,
pthread_rwlock_rdlock(&ai.lock);
- assert(flow->flow_id >= 0);
-
- rb = flow->rx_rb;
+ assert(flow->info.id >= 0);
- while ((idx = frcti_queued_pdu(flow->frcti)) < 0) {
+ while (frcti_queued_pdu(flow->frcti) < 0) {
pthread_rwlock_unlock(&ai.lock);
- idx = shm_rbuff_read(rb);
+ idx = flow_rx_sdb(flow, sdb, false, NULL);
if (idx < 0)
return idx;
pthread_rwlock_rdlock(&ai.lock);
- *sdb = shm_rdrbuff_get(ai.rdrb, idx);
- if (flow->qs.ber == 0 && chk_crc(*sdb) != 0)
- continue;
-
frcti_rcv(flow->frcti, *sdb);
}
- frcti_tick(flow->frcti);
-
pthread_rwlock_unlock(&ai.lock);
- *sdb = shm_rdrbuff_get(ai.rdrb, idx);
-
return 0;
}
int ipcp_flow_write(int fd,
struct shm_du_buff * sdb)
{
- struct flow * flow;
- int ret;
- ssize_t idx;
+ struct flow * flow;
+ int ret;
assert(fd >= 0 && fd < SYS_MAX_FLOWS);
assert(sdb);
flow = &ai.flows[fd];
- pthread_rwlock_rdlock(&ai.lock);
+ pthread_rwlock_wrlock(&ai.lock);
- if (flow->flow_id < 0) {
+ if (flow->info.id < 0) {
pthread_rwlock_unlock(&ai.lock);
return -ENOTALLOC;
}
@@ -1699,30 +1921,74 @@ int ipcp_flow_write(int fd,
return -EPERM;
}
- assert(flow->tx_rb);
+ pthread_rwlock_unlock(&ai.lock);
- idx = shm_du_buff_get_idx(sdb);
+ ret = flow_tx_sdb(flow, sdb, true, NULL);
+
+ return ret;
+}
+
+int np1_flow_read(int fd,
+ struct shm_du_buff ** sdb)
+{
+ struct flow * flow;
+ ssize_t idx = -1;
+
+ assert(fd >= 0 && fd < SYS_MAX_FLOWS);
+ assert(sdb);
- if (frcti_snd(flow->frcti, sdb) < 0) {
+ flow = &ai.flows[fd];
+
+ assert(flow->info.id >= 0);
+
+ pthread_rwlock_rdlock(&ai.lock);
+
+ idx = shm_rbuff_read(flow->rx_rb);;
+ if (idx < 0) {
pthread_rwlock_unlock(&ai.lock);
- return -ENOMEM;
+ return idx;
}
- if (flow->qs.ber == 0 && add_crc(sdb) != 0) {
+ pthread_rwlock_unlock(&ai.lock);
+
+ *sdb = shm_rdrbuff_get(ai.rdrb, idx);
+
+ return 0;
+}
+
+int np1_flow_write(int fd,
+ struct shm_du_buff * sdb)
+{
+ struct flow * flow;
+ int ret;
+ ssize_t idx;
+
+ assert(fd >= 0 && fd < SYS_MAX_FLOWS);
+ assert(sdb);
+
+ flow = &ai.flows[fd];
+
+ pthread_rwlock_rdlock(&ai.lock);
+
+ if (flow->info.id < 0) {
pthread_rwlock_unlock(&ai.lock);
- shm_rdrbuff_remove(ai.rdrb, idx);
- return -ENOMEM;
+ return -ENOTALLOC;
}
- ret = shm_rbuff_write_b(flow->tx_rb, idx, NULL);
- if (ret == 0)
- shm_flow_set_notify(flow->set, flow->flow_id, FLOW_PKT);
- else
- shm_rdrbuff_remove(ai.rdrb, idx);
+ if ((flow->oflags & FLOWFACCMODE) == FLOWFRDONLY) {
+ pthread_rwlock_unlock(&ai.lock);
+ return -EPERM;
+ }
pthread_rwlock_unlock(&ai.lock);
- assert(ret <= 0);
+ idx = shm_du_buff_get_idx(sdb);
+
+ ret = shm_rbuff_write_b(flow->tx_rb, idx, NULL);
+ if (ret < 0)
+ shm_rdrbuff_remove(ai.rdrb, idx);
+ else
+ shm_flow_set_notify(flow->set, flow->info.id, FLOW_PKT);
return ret;
}
@@ -1746,7 +2012,7 @@ int ipcp_flow_fini(int fd)
pthread_rwlock_rdlock(&ai.lock);
- if (ai.flows[fd].flow_id < 0) {
+ if (ai.flows[fd].info.id < 0) {
pthread_rwlock_unlock(&ai.lock);
return -1;
}
@@ -1755,7 +2021,7 @@ int ipcp_flow_fini(int fd)
shm_rbuff_set_acl(ai.flows[fd].tx_rb, ACL_FLOWDOWN);
shm_flow_set_notify(ai.flows[fd].set,
- ai.flows[fd].flow_id,
+ ai.flows[fd].info.id,
FLOW_DEALLOC);
rx_rb = ai.flows[fd].rx_rb;
@@ -1776,9 +2042,9 @@ int ipcp_flow_get_qoscube(int fd,
pthread_rwlock_rdlock(&ai.lock);
- assert(ai.flows[fd].flow_id >= 0);
+ assert(ai.flows[fd].info.id >= 0);
- *cube = qos_spec_to_cube(ai.flows[fd].qs);
+ *cube = qos_spec_to_cube(ai.flows[fd].info.qs);
pthread_rwlock_unlock(&ai.lock);
@@ -1791,7 +2057,7 @@ size_t ipcp_flow_queued(int fd)
pthread_rwlock_rdlock(&ai.lock);
- assert(ai.flows[fd].flow_id >= 0);
+ assert(ai.flows[fd].info.id >= 0);
q = shm_rbuff_queued(ai.flows[fd].tx_rb);
@@ -1827,13 +2093,14 @@ int local_flow_write(int fd,
pthread_rwlock_rdlock(&ai.lock);
- if (flow->flow_id < 0) {
+ if (flow->info.id < 0) {
pthread_rwlock_unlock(&ai.lock);
return -ENOTALLOC;
}
+
ret = shm_rbuff_write_b(flow->tx_rb, idx, NULL);
if (ret == 0)
- shm_flow_set_notify(flow->set, flow->flow_id, FLOW_PKT);
+ shm_flow_set_notify(flow->set, flow->info.id, FLOW_PKT);
else
shm_rdrbuff_remove(ai.rdrb, idx);
generated by cgit v1.2.3 (git 2.39.1) at 2024-04-27 16:52:39 +0000