From f516b51169020ea1957010fbd1005d746f01b1d9 Mon Sep 17 00:00:00 2001 From: dimitri staessens Date: Wed, 19 Oct 2016 22:25:46 +0200 Subject: lib: Demultiplex the fast path The fast path will now use an incoming ring buffer per flow per process. This necessitated the development of a new method for the asynchronous io call, which is now based on an event queue system for scalability (fqueue). The ipcpd's and tools have been updated to this API. --- include/ouroboros/CMakeLists.txt | 4 +-- include/ouroboros/config.h.in | 4 ++- include/ouroboros/fqueue.h | 62 ++++++++++++++++++++++++++++++++ include/ouroboros/ipcp-dev.h | 3 -- include/ouroboros/local-dev.h | 8 ++--- include/ouroboros/select.h | 52 --------------------------- include/ouroboros/shm_ap_rbuff.h | 73 -------------------------------------- include/ouroboros/shm_flow_set.h | 63 ++++++++++++++++++++++++++++++++ include/ouroboros/shm_rbuff.h | 53 +++++++++++++++++++++++++++ include/ouroboros/wrap/ouroboros.i | 4 +-- 10 files changed, 188 insertions(+), 138 deletions(-) create mode 100644 include/ouroboros/fqueue.h delete mode 100644 include/ouroboros/select.h delete mode 100644 include/ouroboros/shm_ap_rbuff.h create mode 100644 include/ouroboros/shm_flow_set.h create mode 100644 include/ouroboros/shm_rbuff.h (limited to 'include') diff --git a/include/ouroboros/CMakeLists.txt b/include/ouroboros/CMakeLists.txt index f24857ed..41feb65e 100644 --- a/include/ouroboros/CMakeLists.txt +++ b/include/ouroboros/CMakeLists.txt @@ -7,11 +7,11 @@ set(HEADER_FILES dev.h errno.h fcntl.h + fqueue.h irm.h irm_config.h nsm.h - qos.h - select.h) + qos.h) install(FILES ${HEADER_FILES} DESTINATION usr/include/ouroboros) diff --git a/include/ouroboros/config.h.in b/include/ouroboros/config.h.in index 143ae7c8..a9d65aec 100644 --- a/include/ouroboros/config.h.in +++ b/include/ouroboros/config.h.in @@ -36,6 +36,7 @@ #define IPCP_NORMAL_EXEC "@IPCP_NORMAL_TARGET@" #define IPCP_LOCAL_EXEC "@IPCP_LOCAL_TARGET@" #define AP_MAX_FLOWS 256 +#define AP_MAX_FQUEUES 64 #define SHM_RDRB_BLOCK_SIZE sysconf(_SC_PAGESIZE) #define SHM_RDRB_MULTI_BLOCK #define SHM_RDRB_PREFIX "/ouroboros.rdrb." @@ -43,7 +44,8 @@ #define SHM_BUFFER_SIZE (1 << 14) #define DU_BUFF_HEADSPACE 128 #define DU_BUFF_TAILSPACE 0 -#define SHM_AP_RBUFF_PREFIX "/ouroboros.rbuff." +#define SHM_RBUFF_PREFIX "/ouroboros.rbuff." +#define SHM_FLOW_SET_PREFIX "/ouroboros.sets." #define IRMD_MAX_FLOWS 4096 #define IRMD_THREADPOOL_SIZE 5 #define LOG_DIR "/@LOG_DIR@/" diff --git a/include/ouroboros/fqueue.h b/include/ouroboros/fqueue.h new file mode 100644 index 00000000..943d6510 --- /dev/null +++ b/include/ouroboros/fqueue.h @@ -0,0 +1,62 @@ +/* + * Ouroboros - Copyright (C) 2016 + * + * Flow queues + * + * Dimitri Staessens + * Sander Vrijders + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version. + * + * 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., 675 Mass Ave, Cambridge, MA 02139, USA. + */ + +#ifndef OUROBOROS_FQUEUE_H +#define OUROBOROS_FQUEUE_H + +#include +#include + +struct flow_set; + +struct fqueue; + +typedef struct flow_set flow_set_t; +typedef struct fqueue fqueue_t; + +flow_set_t * flow_set_create(); + +void flow_set_destroy(flow_set_t * set); + +fqueue_t * fqueue_create(); + +void fqueue_destroy(struct fqueue * fq); + +void flow_set_zero(flow_set_t * set); + +int flow_set_add(flow_set_t * set, + int fd); + +bool flow_set_has(flow_set_t * set, + int fd); + +void flow_set_del(flow_set_t * set, + int fd); + +int fqueue_next(fqueue_t * fq); + +int flow_event_wait(flow_set_t * set, + fqueue_t * fq, + const struct timespec * timeout); + +#endif /* OUROBOROS_SELECT_H */ diff --git a/include/ouroboros/ipcp-dev.h b/include/ouroboros/ipcp-dev.h index 9343aeaa..3ab05bd7 100644 --- a/include/ouroboros/ipcp-dev.h +++ b/include/ouroboros/ipcp-dev.h @@ -47,7 +47,4 @@ int ipcp_flow_write(int fd, void ipcp_flow_del(struct shm_du_buff * sdb); -/* returns flow descriptor and du buff */ -int ipcp_read_shim(struct shm_du_buff ** sdb); - #endif /* OUROBOROS_IPCP_DEV_H */ diff --git a/include/ouroboros/local-dev.h b/include/ouroboros/local-dev.h index 77ff47e9..30f440b1 100644 --- a/include/ouroboros/local-dev.h +++ b/include/ouroboros/local-dev.h @@ -20,14 +20,12 @@ * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. */ -#include - #ifndef OUROBOROS_LOCAL_DEV_H #define OUROBOROS_LOCAL_DEV_H -struct rb_entry * local_flow_read(int fd); +ssize_t local_flow_read(int fd); -int local_flow_write(int fd, - struct rb_entry * e); +int local_flow_write(int fd, + ssize_t idx); #endif /* OUROBOROS_LOCAL_DEV_H */ diff --git a/include/ouroboros/select.h b/include/ouroboros/select.h deleted file mode 100644 index de309b8d..00000000 --- a/include/ouroboros/select.h +++ /dev/null @@ -1,52 +0,0 @@ -/* - * Ouroboros - Copyright (C) 2016 - * - * A select call for flows - * - * Dimitri Staessens - * Sander Vrijders - * - * This program is free software; you can redistribute it and/or modify - * it under the terms of the GNU General Public License as published by - * the Free Software Foundation; either version 2 of the License, or - * (at your option) any later version. - * - * 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., 675 Mass Ave, Cambridge, MA 02139, USA. - */ - -#ifndef OUROBOROS_SELECT_H -#define OUROBOROS_SELECT_H - -#include -#include - -struct flow_set; - -typedef struct flow_set flow_set_t; - -flow_set_t * flow_set_create(); - -void flow_set_destroy(flow_set_t * set); - -void flow_set_zero(flow_set_t * set); - -void flow_set_add(flow_set_t * set, - int fd); - -void flow_set_del(flow_set_t * set, - int fd); - -bool flow_set_has(flow_set_t * set, - int fd); - -int flow_select(flow_set_t * set, - const struct timespec * timeout); - -#endif /* OUROBOROS_SELECT_H */ diff --git a/include/ouroboros/shm_ap_rbuff.h b/include/ouroboros/shm_ap_rbuff.h deleted file mode 100644 index 453e4bf8..00000000 --- a/include/ouroboros/shm_ap_rbuff.h +++ /dev/null @@ -1,73 +0,0 @@ -/* - * Ouroboros - Copyright (C) 2016 - * - * Ring buffer for application processes - * - * Dimitri Staessens - * Sander Vrijders - * - * This program is free software; you can redistribute it and/or modify - * it under the terms of the GNU General Public License as published by - * the Free Software Foundation; either version 2 of the License, or - * (at your option) any later version. - * - * 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., 675 Mass Ave, Cambridge, MA 02139, USA. - */ - -#ifndef OUROBOROS_SHM_AP_RBUFF_H -#define OUROBOROS_SHM_AP_RBUFF_H - -#include -#include -#include -#include - -struct shm_ap_rbuff; - -struct rb_entry { - ssize_t index; - int port_id; -}; - -struct shm_ap_rbuff * shm_ap_rbuff_create(); - -struct shm_ap_rbuff * shm_ap_rbuff_open(pid_t api); - -void shm_ap_rbuff_close(struct shm_ap_rbuff * rb); - -void shm_ap_rbuff_destroy(struct shm_ap_rbuff * rb); - -void shm_ap_rbuff_open_port(struct shm_ap_rbuff * rb, - int port_id); - -int shm_ap_rbuff_close_port(struct shm_ap_rbuff * rb, - int port_id); - -int shm_ap_rbuff_write(struct shm_ap_rbuff * rb, - struct rb_entry * e); - -struct rb_entry * shm_ap_rbuff_read(struct shm_ap_rbuff * rb); - -int shm_ap_rbuff_pop_idx(struct shm_ap_rbuff * rb); - -int shm_ap_rbuff_peek_b(struct shm_ap_rbuff * rb, - bool * set, - const struct timespec * timeout); - -ssize_t shm_ap_rbuff_read_port(struct shm_ap_rbuff * rb, - int port_id); - -ssize_t shm_ap_rbuff_read_port_b(struct shm_ap_rbuff * rb, - int port_id, - const struct timespec * timeout); - -void shm_ap_rbuff_reset(struct shm_ap_rbuff * rb); - -#endif /* OUROBOROS_SHM_AP_RBUFF_H */ diff --git a/include/ouroboros/shm_flow_set.h b/include/ouroboros/shm_flow_set.h new file mode 100644 index 00000000..32db5d36 --- /dev/null +++ b/include/ouroboros/shm_flow_set.h @@ -0,0 +1,63 @@ +/* + * Ouroboros - Copyright (C) 2016 + * + * Management of flow_sets for fqueue + * + * Dimitri Staessens + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version. + * + * 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., 675 Mass Ave, Cambridge, MA 02139, USA. + */ + +#ifndef OUROBOROS_SHM_FLOW_SET_H +#define OUROBOROS_SHM_FLOW_SET_H + +#include + +#include + +struct shm_flow_set; + +struct shm_flow_set * shm_flow_set_create(); + +void shm_flow_set_destroy(struct shm_flow_set * set); + +struct shm_flow_set * shm_flow_set_open(pid_t api); + +void shm_flow_set_close(struct shm_flow_set * set); + +void shm_flow_set_zero(struct shm_flow_set * shm_set, + ssize_t idx); + +int shm_flow_set_add(struct shm_flow_set * shm_set, + ssize_t idx, + int port_id); + +int shm_flow_set_has(struct shm_flow_set * shm_set, + ssize_t idx, + int port_id); + +void shm_flow_set_del(struct shm_flow_set * shm_set, + ssize_t idx, + int port_id); + +void shm_flow_set_notify(struct shm_flow_set * set, + int port_id); + +int shm_flow_set_wait(const struct shm_flow_set * shm_set, + ssize_t idx, + int * fqueue, + const struct timespec * timeout); + +#endif /* OUROBOROS_SHM_FLOW_SET_H */ diff --git a/include/ouroboros/shm_rbuff.h b/include/ouroboros/shm_rbuff.h new file mode 100644 index 00000000..03660b88 --- /dev/null +++ b/include/ouroboros/shm_rbuff.h @@ -0,0 +1,53 @@ +/* + * Ouroboros - Copyright (C) 2016 + * + * Ring buffer for incoming SDUs + * + * Dimitri Staessens + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version. + * + * 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., 675 Mass Ave, Cambridge, MA 02139, USA. + */ + +#ifndef OUROBOROS_SHM_RBUFF_H +#define OUROBOROS_SHM_RBUFF_H + +#include +#include + +struct shm_rbuff; + +struct shm_rbuff * shm_rbuff_create(int port_id); + +struct shm_rbuff * shm_rbuff_open(pid_t api, int port_id); + +void shm_rbuff_close(struct shm_rbuff * rb); + +void shm_rbuff_destroy(struct shm_rbuff * rb); + +int shm_rbuff_block(struct shm_rbuff * rb); + +void shm_rbuff_unblock(struct shm_rbuff * rb); + +int shm_rbuff_write(struct shm_rbuff * rb, + ssize_t idx); + +ssize_t shm_rbuff_read(struct shm_rbuff * rb); + +ssize_t shm_rbuff_read_b(struct shm_rbuff * rb, + const struct timespec * timeout); + +void shm_rbuff_reset(struct shm_rbuff * rb); + +#endif /* OUROBOROS_SHM_RBUFF_H */ diff --git a/include/ouroboros/wrap/ouroboros.i b/include/ouroboros/wrap/ouroboros.i index 394b505a..26cc6076 100644 --- a/include/ouroboros/wrap/ouroboros.i +++ b/include/ouroboros/wrap/ouroboros.i @@ -26,11 +26,11 @@ #include "ouroboros/dev.h" #include "ouroboros/errno.h" #include "ouroboros/fcntl.h" +#include "ouroboros/fqueue.h" #include "ouroboros/irm.h" #include "ouroboros/irm_config.h" #include "ouroboros/nsm.h" #include "ouroboros/qos.h" -#include "ouroboros/select.h" %} typedef int pid_t; @@ -39,8 +39,8 @@ typedef int pid_t; %include "ouroboros/dev.h" %include "ouroboros/errno.h" %include "ouroboros/fcntl.h" +%include "ouroboros/fqueue.h" %include "ouroboros/irm.h" %include "ouroboros/irm_config.h" %include "ouroboros/nsm.h" %include "ouroboros/qos.h" -%include "ouroboros/select.h" -- cgit v1.2.3 From 02976060919566d1a217b818ca8f33297700d56d Mon Sep 17 00:00:00 2001 From: dimitri staessens Date: Thu, 20 Oct 2016 19:52:02 +0200 Subject: lib: Move rbuff creation/destruction to IRMd This stabilises flow allocation now that the rbuffs are created upon flow allocation. Only the IRMd can sync this process sufficiently. --- include/ouroboros/shm_rbuff.h | 2 +- src/irmd/irm_flow.c | 6 ++++ src/irmd/irm_flow.h | 20 +++++++------ src/irmd/main.c | 65 +++++++++++++++++++++++++++++++++---------- src/lib/dev.c | 60 +++++++++++++++++++-------------------- src/lib/shm_flow_set.c | 1 - src/lib/shm_rbuff.c | 37 ++++++------------------ 7 files changed, 106 insertions(+), 85 deletions(-) (limited to 'include') diff --git a/include/ouroboros/shm_rbuff.h b/include/ouroboros/shm_rbuff.h index 03660b88..4c4e8c64 100644 --- a/include/ouroboros/shm_rbuff.h +++ b/include/ouroboros/shm_rbuff.h @@ -28,7 +28,7 @@ struct shm_rbuff; -struct shm_rbuff * shm_rbuff_create(int port_id); +struct shm_rbuff * shm_rbuff_create(pid_t api, int port_id); struct shm_rbuff * shm_rbuff_open(pid_t api, int port_id); diff --git a/src/irmd/irm_flow.c b/src/irmd/irm_flow.c index df1302b4..dc5d22d8 100644 --- a/src/irmd/irm_flow.c +++ b/src/irmd/irm_flow.c @@ -36,6 +36,9 @@ struct irm_flow * irm_flow_create() f->n_api = -1; f->n_1_api = -1; f->port_id = -1; + f->n_rb = NULL; + f->n_1_rb = NULL; + f->state = FLOW_NULL; if (pthread_cond_init(&f->state_cond, NULL)) { @@ -78,6 +81,9 @@ void irm_flow_destroy(struct irm_flow * f) pthread_cond_destroy(&f->state_cond); pthread_mutex_destroy(&f->state_lock); + shm_rbuff_destroy(f->n_rb); + shm_rbuff_destroy(f->n_1_rb); + free(f); } diff --git a/src/irmd/irm_flow.h b/src/irmd/irm_flow.h index 5ec6d90e..507295bd 100644 --- a/src/irmd/irm_flow.h +++ b/src/irmd/irm_flow.h @@ -24,6 +24,7 @@ #define OUROBOROS_IRMD_IRM_FLOW_H #include +#include #include #include @@ -38,18 +39,21 @@ enum flow_state { }; struct irm_flow { - struct list_head next; + struct list_head next; - int port_id; + int port_id; - pid_t n_api; - pid_t n_1_api; + pid_t n_api; + pid_t n_1_api; - struct timespec t0; + struct shm_rbuff * n_rb; + struct shm_rbuff * n_1_rb; - enum flow_state state; - pthread_cond_t state_cond; - pthread_mutex_t state_lock; + struct timespec t0; + + enum flow_state state; + pthread_cond_t state_cond; + pthread_mutex_t state_lock; }; struct irm_flow * irm_flow_create(); diff --git a/src/irmd/main.c b/src/irmd/main.c index 67941e41..8d9d04ac 100644 --- a/src/irmd/main.c +++ b/src/irmd/main.c @@ -1164,6 +1164,24 @@ static struct irm_flow * flow_alloc(pid_t api, port_id = f->port_id = bmp_allocate(irmd->port_ids); f->n_1_api = ipcp; + f->n_rb = shm_rbuff_create(api, port_id); + if (f->n_rb == NULL) { + pthread_rwlock_unlock(&irmd->flows_lock); + pthread_rwlock_unlock(&irmd->state_lock); + LOG_ERR("Could not create ringbuffer for AP-I %d.", api); + irm_flow_destroy(f); + return NULL; + } + + f->n_1_rb = shm_rbuff_create(ipcp, port_id); + if (f->n_1_rb == NULL) { + pthread_rwlock_unlock(&irmd->flows_lock); + pthread_rwlock_unlock(&irmd->state_lock); + LOG_ERR("Could not create ringbuffer for AP-I %d.", ipcp); + irm_flow_destroy(f); + return NULL; + } + list_add(&f->next, &irmd->irm_flows); pthread_rwlock_unlock(&irmd->flows_lock); @@ -1346,7 +1364,7 @@ static struct irm_flow * flow_req_arr(pid_t api, pthread_rwlock_unlock(&irmd->reg_lock); pthread_rwlock_unlock(&irmd->state_lock); LOG_ERR("Unknown name: %s.", dst_name); - free(f); + irm_flow_destroy(f); return NULL; } @@ -1359,14 +1377,14 @@ static struct irm_flow * flow_req_arr(pid_t api, pthread_rwlock_unlock(&irmd->reg_lock); pthread_rwlock_unlock(&irmd->state_lock); LOG_ERR("No AP's for %s.", dst_name); - free(f); + irm_flow_destroy(f); return NULL; case REG_NAME_AUTO_ACCEPT: c_api = malloc(sizeof(*c_api)); if (c_api == NULL) { pthread_rwlock_unlock(&irmd->reg_lock); pthread_rwlock_unlock(&irmd->state_lock); - free(f); + irm_flow_destroy(f); return NULL; } @@ -1384,7 +1402,7 @@ static struct irm_flow * flow_req_arr(pid_t api, pthread_rwlock_unlock(&irmd->state_lock); LOG_ERR("Could not get start apn for reg_entry %s.", re->name); - free(f); + irm_flow_destroy(f); free(c_api); return NULL; } @@ -1411,6 +1429,7 @@ static struct irm_flow * flow_req_arr(pid_t api, pthread_mutex_unlock(&re->state_lock); pthread_rwlock_unlock(&irmd->reg_lock); pthread_rwlock_unlock(&irmd->state_lock); + irm_flow_destroy(f); return NULL; } @@ -1424,6 +1443,7 @@ static struct irm_flow * flow_req_arr(pid_t api, pthread_rwlock_unlock(&irmd->reg_lock); pthread_rwlock_unlock(&irmd->state_lock); LOG_ERR("Invalid api returned."); + irm_flow_destroy(f); return NULL; } @@ -1432,7 +1452,7 @@ static struct irm_flow * flow_req_arr(pid_t api, pthread_rwlock_unlock(&irmd->reg_lock); pthread_rwlock_unlock(&irmd->state_lock); LOG_ERR("IRMd in wrong state."); - free(f); + irm_flow_destroy(f); return NULL; } @@ -1441,6 +1461,26 @@ static struct irm_flow * flow_req_arr(pid_t api, pthread_rwlock_wrlock(&irmd->flows_lock); f->port_id = bmp_allocate(irmd->port_ids); + f->n_rb = shm_rbuff_create(f->n_api, f->port_id); + if (f->n_rb == NULL) { + bmp_release(irmd->port_ids, f->port_id); + pthread_rwlock_unlock(&irmd->flows_lock); + pthread_rwlock_unlock(&irmd->state_lock); + LOG_ERR("Could not create ringbuffer for AP-I %d.", f->n_api); + irm_flow_destroy(f); + return NULL; + } + + f->n_1_rb = shm_rbuff_create(f->n_1_api, f->port_id); + if (f->n_1_rb == NULL) { + bmp_release(irmd->port_ids, f->port_id); + pthread_rwlock_unlock(&irmd->flows_lock); + pthread_rwlock_unlock(&irmd->state_lock); + LOG_ERR("Could not create ringbuffer for AP-I %d.", f->n_1_api); + irm_flow_destroy(f); + return NULL; + } + list_add(&f->next, &irmd->irm_flows); pthread_rwlock_unlock(&irmd->flows_lock); @@ -1455,10 +1495,13 @@ static struct irm_flow * flow_req_arr(pid_t api, e = api_table_get(&irmd->api_table, h_api); if (e == NULL) { - LOG_ERR("Could not get api table entry for %d.", h_api); pthread_rwlock_unlock(&irmd->reg_lock); + pthread_rwlock_wrlock(&irmd->flows_lock); + bmp_release(irmd->port_ids, f->port_id); + pthread_rwlock_unlock(&irmd->flows_lock); pthread_rwlock_unlock(&irmd->state_lock); - free(f); + LOG_ERR("Could not get api table entry for %d.", h_api); + irm_flow_destroy(f); return NULL; } @@ -1692,26 +1735,18 @@ void * irm_sanitize() } if (kill(f->n_api, 0) < 0) { - struct shm_rbuff * rb = - shm_rbuff_open(f->n_api, f->port_id); bmp_release(irmd->port_ids, f->port_id); list_del(&f->next); LOG_INFO("AP-I %d gone, flow %d deallocated.", f->n_api, f->port_id); ipcp_flow_dealloc(f->n_1_api, f->port_id); - if (rb != NULL) - shm_rbuff_destroy(rb); irm_flow_destroy(f); continue; } if (kill(f->n_1_api, 0) < 0) { - struct shm_rbuff * rb = - shm_rbuff_open(f->n_1_api, f->port_id); list_del(&f->next); LOG_ERR("IPCP %d gone, flow %d removed.", f->n_1_api, f->port_id); - if (rb != NULL) - shm_rbuff_destroy(rb); irm_flow_destroy(f); } } diff --git a/src/lib/dev.c b/src/lib/dev.c index f735e72b..146070b7 100644 --- a/src/lib/dev.c +++ b/src/lib/dev.c @@ -288,7 +288,7 @@ void ap_fini() int idx; while ((idx = shm_rbuff_read(ai.flows[i].rx_rb)) >= 0) shm_rdrbuff_remove(ai.rdrb, idx); - shm_rbuff_destroy(ai.flows[i].rx_rb); + shm_rbuff_close(ai.flows[i].rx_rb); shm_rbuff_close(ai.flows[i].tx_rb); shm_flow_set_close(ai.flows[i].set); } @@ -349,7 +349,7 @@ int flow_accept(char ** ae_name) return -1; } - ai.flows[fd].rx_rb = shm_rbuff_create(recv_msg->port_id); + ai.flows[fd].rx_rb = shm_rbuff_open(ai.api, recv_msg->port_id); if (ai.flows[fd].rx_rb == NULL) { bmp_release(ai.fds, fd); pthread_rwlock_unlock(&ai.flows_lock); @@ -361,7 +361,7 @@ int flow_accept(char ** ae_name) ai.flows[fd].set = shm_flow_set_open(recv_msg->api); if (ai.flows[fd].set == NULL) { bmp_release(ai.fds, fd); - shm_rbuff_destroy(ai.flows[fd].rx_rb); + shm_rbuff_close(ai.flows[fd].rx_rb); shm_rbuff_close(ai.flows[fd].tx_rb); pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); @@ -373,7 +373,7 @@ int flow_accept(char ** ae_name) if (ae_name != NULL) { *ae_name = strdup(recv_msg->ae_name); if (*ae_name == NULL) { - shm_rbuff_destroy(ai.flows[fd].tx_rb); + shm_rbuff_close(ai.flows[fd].tx_rb); shm_rbuff_close(ai.flows[fd].tx_rb); shm_flow_set_close(ai.flows[fd].set); bmp_release(ai.fds, fd); @@ -508,7 +508,7 @@ int flow_alloc(char * dst_name, char * src_ae_name, struct qos_spec * qos) ai.flows[fd].port_id = recv_msg->port_id; ai.flows[fd].oflags = FLOW_O_DEFAULT; ai.flows[fd].api = recv_msg->api; - ai.flows[fd].rx_rb = shm_rbuff_create(recv_msg->port_id); + ai.flows[fd].rx_rb = shm_rbuff_open(ai.api, recv_msg->port_id); if (ai.flows[fd].rx_rb == NULL) { bmp_release(ai.fds, fd); pthread_rwlock_unlock(&ai.flows_lock); @@ -517,27 +517,6 @@ int flow_alloc(char * dst_name, char * src_ae_name, struct qos_spec * qos) return -1; } - ai.flows[fd].tx_rb = shm_rbuff_open(recv_msg->api, recv_msg->port_id); - if (ai.flows[fd].tx_rb == NULL) { - shm_rbuff_destroy(ai.flows[fd].rx_rb); - bmp_release(ai.fds, fd); - pthread_rwlock_unlock(&ai.flows_lock); - pthread_rwlock_unlock(&ai.data_lock); - irm_msg__free_unpacked(recv_msg, NULL); - return -1; - } - - ai.flows[fd].set = shm_flow_set_open(recv_msg->api); - if (ai.flows[fd].set == NULL) { - shm_rbuff_close(ai.flows[fd].tx_rb); - shm_rbuff_destroy(ai.flows[fd].rx_rb); - bmp_release(ai.fds, fd); - pthread_rwlock_unlock(&ai.flows_lock); - pthread_rwlock_unlock(&ai.data_lock); - irm_msg__free_unpacked(recv_msg, NULL); - return -1; - } - ai.ports[recv_msg->port_id].fd = fd; ai.ports[recv_msg->port_id].state = PORT_ID_ASSIGNED; @@ -572,6 +551,23 @@ int flow_alloc_res(int fd) msg.port_id = ai.flows[fd].port_id; + ai.flows[fd].tx_rb = shm_rbuff_open(ai.flows[fd].api, + ai.flows[fd].port_id); + if (ai.flows[fd].tx_rb == NULL) { + pthread_rwlock_unlock(&ai.flows_lock); + pthread_rwlock_unlock(&ai.data_lock); + irm_msg__free_unpacked(recv_msg, NULL); + return -1; + } + + ai.flows[fd].set = shm_flow_set_open(ai.flows[fd].api); + if (ai.flows[fd].set == NULL) { + pthread_rwlock_unlock(&ai.flows_lock); + pthread_rwlock_unlock(&ai.data_lock); + irm_msg__free_unpacked(recv_msg, NULL); + return -1; + } + pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); @@ -599,7 +595,7 @@ int flow_dealloc(int fd) msg.code = IRM_MSG_CODE__IRM_FLOW_DEALLOC; msg.has_port_id = true; msg.has_api = true; - msg.api = getpid(); + msg.api = ai.api; pthread_rwlock_rdlock(&ai.data_lock); pthread_rwlock_wrlock(&ai.flows_lock); @@ -621,7 +617,7 @@ int flow_dealloc(int fd) port_destroy(&ai.ports[msg.port_id]); ai.flows[fd].port_id = -1; - shm_rbuff_destroy(ai.flows[fd].rx_rb); + shm_rbuff_close(ai.flows[fd].rx_rb); ai.flows[fd].rx_rb = NULL; shm_rbuff_close(ai.flows[fd].tx_rb); ai.flows[fd].tx_rb = NULL; @@ -990,7 +986,7 @@ int np1_flow_alloc(pid_t n_api, int port_id) return -1; } - ai.flows[fd].rx_rb = shm_rbuff_create(port_id); + ai.flows[fd].rx_rb = shm_rbuff_open(ai.api, port_id); if (ai.flows[fd].rx_rb == NULL) { bmp_release(ai.fds, fd); pthread_rwlock_unlock(&ai.flows_lock); @@ -1046,7 +1042,7 @@ int np1_flow_resp(pid_t n_api, int port_id) ai.flows[fd].tx_rb = shm_rbuff_open(n_api, port_id); if (ai.flows[fd].tx_rb == NULL) { ai.flows[fd].port_id = -1; - shm_rbuff_destroy(ai.flows[fd].rx_rb); + shm_rbuff_close(ai.flows[fd].rx_rb); port_destroy(&ai.ports[port_id]); pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); @@ -1057,7 +1053,7 @@ int np1_flow_resp(pid_t n_api, int port_id) if (ai.flows[fd].set == NULL) { shm_rbuff_close(ai.flows[fd].tx_rb); ai.flows[fd].port_id = -1; - shm_rbuff_destroy(ai.flows[fd].rx_rb); + shm_rbuff_close(ai.flows[fd].rx_rb); port_destroy(&ai.ports[port_id]); pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); @@ -1143,7 +1139,7 @@ int ipcp_flow_req_arr(pid_t api, char * dst_name, char * src_ae_name) pthread_rwlock_rdlock(&ai.data_lock); pthread_rwlock_wrlock(&ai.flows_lock); - ai.flows[fd].rx_rb = shm_rbuff_create(port_id); + ai.flows[fd].rx_rb = shm_rbuff_open(ai.api, port_id); if (ai.flows[fd].rx_rb == NULL) { ai.flows[fd].port_id = -1; port_destroy(&ai.ports[port_id]); diff --git a/src/lib/shm_flow_set.c b/src/lib/shm_flow_set.c index c960bd25..04de9fc5 100644 --- a/src/lib/shm_flow_set.c +++ b/src/lib/shm_flow_set.c @@ -315,7 +315,6 @@ int shm_flow_set_has(struct shm_flow_set * shm_set, assert(!(port_id < 0) && port_id < IRMD_MAX_FLOWS); assert(!(idx < 0) && idx < AP_MAX_FQUEUES); - pthread_mutex_lock(shm_set->lock); if (shm_set->mtable[port_id] == idx) diff --git a/src/lib/shm_rbuff.c b/src/lib/shm_rbuff.c index cf094488..a933fbff 100644 --- a/src/lib/shm_rbuff.c +++ b/src/lib/shm_rbuff.c @@ -68,7 +68,7 @@ struct shm_rbuff { int port_id; /* port_id of the flow */ }; -struct shm_rbuff * shm_rbuff_create(int port_id) +struct shm_rbuff * shm_rbuff_create(pid_t api, int port_id) { struct shm_rbuff * rb; int shm_fd; @@ -78,7 +78,7 @@ struct shm_rbuff * shm_rbuff_create(int port_id) char fn[FN_MAX_CHARS]; mode_t mask; - sprintf(fn, SHM_RBUFF_PREFIX "%d.%d", getpid(), port_id); + sprintf(fn, SHM_RBUFF_PREFIX "%d.%d", api, port_id); rb = malloc(sizeof(*rb)); if (rb == NULL) { @@ -148,9 +148,12 @@ struct shm_rbuff * shm_rbuff_create(int port_id) *rb->head = 0; *rb->tail = 0; - rb->api = getpid(); + rb->api = api; rb->port_id = port_id; + if (munmap(rb->shm_base, SHM_RBUFF_FILE_SIZE) == -1) + LOG_DBG("Couldn't unmap shared memory."); + return rb; } @@ -221,36 +224,14 @@ void shm_rbuff_close(struct shm_rbuff * rb) void shm_rbuff_destroy(struct shm_rbuff * rb) { char fn[25]; - struct lockfile * lf = NULL; - - assert(rb); - if (rb->api != getpid()) { - lf = lockfile_open(); - if (lf == NULL) { - LOG_ERR("Failed to open lockfile."); - return; - } - - if (lockfile_owner(lf) == getpid()) { - LOG_DBG("Ringbuffer %d destroyed by IRMd %d.", - rb->api, getpid()); - lockfile_close(lf); - } else { - LOG_ERR("AP-I %d tried to destroy rbuff owned by %d.", - getpid(), rb->api); - lockfile_close(lf); - return; - } - } + if (rb == NULL) + return; sprintf(fn, SHM_RBUFF_PREFIX "%d.%d", rb->api, rb->port_id); - if (munmap(rb->shm_base, SHM_RBUFF_FILE_SIZE) == -1) - LOG_DBG("Couldn't unmap shared memory."); - if (shm_unlink(fn) == -1) - LOG_DBG("Failed to unlink shm."); + LOG_DBG("Failed to unlink shm %s.", fn); free(rb); } -- cgit v1.2.3