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. --- src/lib/dev.c | 500 ++++++++++++++++++++++++++++++++++++++-------------------- 1 file changed, 329 insertions(+), 171 deletions(-) (limited to 'src/lib/dev.c') diff --git a/src/lib/dev.c b/src/lib/dev.c index 77c2d06a..f735e72b 100644 --- a/src/lib/dev.c +++ b/src/lib/dev.c @@ -3,7 +3,8 @@ * * API for applications * - * Sander Vrijders + * 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 @@ -26,20 +27,24 @@ #include #include #include +#include #include -#include +#include #include -#include +#include #include #include #include struct flow_set { - bool dirty; - bool b[IRMD_MAX_FLOWS]; /* working copy */ - bool s[IRMD_MAX_FLOWS]; /* safe copy */ - pthread_rwlock_t lock; + size_t idx; +}; + +struct fqueue { + int fqueue[SHM_BUFFER_SIZE]; /* safe copy from shm */ + size_t fqsize; + size_t next; }; enum port_state { @@ -124,7 +129,9 @@ enum port_state port_wait_assign(struct port * p) } struct flow { - struct shm_ap_rbuff * rb; + struct shm_rbuff * rx_rb; + struct shm_rbuff * tx_rb; + struct shm_flow_set * set; int port_id; int oflags; @@ -139,10 +146,11 @@ struct { pid_t api; struct shm_rdrbuff * rdrb; - struct shm_ap_rbuff * rb; + struct shm_flow_set * fqset; pthread_rwlock_t data_lock; struct bmp * fds; + struct bmp * fqueues; struct flow * flows; struct port * ports; @@ -194,40 +202,52 @@ int ap_init(char * ap_name) if (ai.fds == NULL) return -ENOMEM; - ai.rdrb = shm_rdrbuff_open(); - if (ai.rdrb == NULL) { + ai.fqueues = bmp_create(AP_MAX_FQUEUES, 0); + if (ai.fqueues == NULL) { + bmp_destroy(ai.fds); + return -ENOMEM; + } + + ai.fqset = shm_flow_set_create(); + if (ai.fqset == NULL) { + bmp_destroy(ai.fqueues); bmp_destroy(ai.fds); return -1; } - ai.rb = shm_ap_rbuff_create(); - if (ai.rb == NULL) { - shm_rdrbuff_close(ai.rdrb); + ai.rdrb = shm_rdrbuff_open(); + if (ai.rdrb == NULL) { + shm_flow_set_destroy(ai.fqset); + bmp_destroy(ai.fqueues); bmp_destroy(ai.fds); return -1; } ai.flows = malloc(sizeof(*ai.flows) * AP_MAX_FLOWS); if (ai.flows == NULL) { - shm_ap_rbuff_destroy(ai.rb); shm_rdrbuff_close(ai.rdrb); + shm_flow_set_destroy(ai.fqset); + bmp_destroy(ai.fqueues); bmp_destroy(ai.fds); return -1; } for (i = 0; i < AP_MAX_FLOWS; ++i) { - ai.flows[i].rb = NULL; + ai.flows[i].rx_rb = NULL; + ai.flows[i].tx_rb = NULL; + ai.flows[i].set = NULL; ai.flows[i].port_id = -1; - ai.flows[i].oflags = 0; - ai.flows[i].api = -1; + ai.flows[i].oflags = 0; + ai.flows[i].api = -1; ai.flows[i].timeout = NULL; } ai.ports = malloc(sizeof(*ai.ports) * IRMD_MAX_FLOWS); - if (ai.flows == NULL) { + if (ai.ports == NULL) { free(ai.flows); - shm_ap_rbuff_destroy(ai.rb); shm_rdrbuff_close(ai.rdrb); + shm_flow_set_destroy(ai.fqset); + bmp_destroy(ai.fqueues); bmp_destroy(ai.fds); return -1; } @@ -253,16 +273,10 @@ void ap_fini() pthread_rwlock_wrlock(&ai.data_lock); - /* remove all remaining sdus */ - while ((i = shm_ap_rbuff_pop_idx(ai.rb)) >= 0) - shm_rdrbuff_remove(ai.rdrb, i); - - if (ai.fds != NULL) - bmp_destroy(ai.fds); - if (ai.rb != NULL) - shm_ap_rbuff_destroy(ai.rb); - if (ai.rdrb != NULL) - shm_rdrbuff_close(ai.rdrb); + bmp_destroy(ai.fds); + bmp_destroy(ai.fqueues); + shm_flow_set_destroy(ai.fqset); + shm_rdrbuff_close(ai.rdrb); if (ai.daf_name != NULL) free(ai.daf_name); @@ -270,8 +284,15 @@ void ap_fini() pthread_rwlock_rdlock(&ai.flows_lock); for (i = 0; i < AP_MAX_FLOWS; ++i) { - if (ai.flows[i].rb != NULL) - shm_ap_rbuff_close(ai.flows[i].rb); + if (ai.flows[i].tx_rb != NULL) { + 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].tx_rb); + shm_flow_set_close(ai.flows[i].set); + } + if (ai.flows[i].timeout != NULL) free(ai.flows[i].timeout); } @@ -328,8 +349,8 @@ int flow_accept(char ** ae_name) return -1; } - ai.flows[fd].rb = shm_ap_rbuff_open(recv_msg->api); - if (ai.flows[fd].rb == NULL) { + ai.flows[fd].rx_rb = shm_rbuff_create(recv_msg->port_id); + if (ai.flows[fd].rx_rb == NULL) { bmp_release(ai.fds, fd); pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); @@ -337,10 +358,24 @@ int flow_accept(char ** ae_name) return -1; } + 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].tx_rb); + pthread_rwlock_unlock(&ai.flows_lock); + pthread_rwlock_unlock(&ai.data_lock); + irm_msg__free_unpacked(recv_msg, NULL); + return -1; + } + + if (ae_name != NULL) { *ae_name = strdup(recv_msg->ae_name); if (*ae_name == NULL) { - shm_ap_rbuff_close(ai.flows[fd].rb); + shm_rbuff_destroy(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); pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); @@ -356,8 +391,6 @@ int flow_accept(char ** ae_name) ai.ports[recv_msg->port_id].fd = fd; ai.ports[recv_msg->port_id].state = PORT_ID_ASSIGNED; - shm_ap_rbuff_open_port(ai.rb, recv_msg->port_id); - pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); @@ -410,6 +443,17 @@ int flow_alloc_resp(int fd, int response) ret = recv_msg->result; + pthread_rwlock_wrlock(&ai.flows_lock); + + 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); + return -1; + } + + pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); irm_msg__free_unpacked(recv_msg, NULL); @@ -461,8 +505,11 @@ int flow_alloc(char * dst_name, char * src_ae_name, struct qos_spec * qos) return -1; } - ai.flows[fd].rb = shm_ap_rbuff_open(recv_msg->api); - if (ai.flows[fd].rb == NULL) { + 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); + if (ai.flows[fd].rx_rb == NULL) { bmp_release(ai.fds, fd); pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); @@ -470,9 +517,26 @@ int flow_alloc(char * dst_name, char * src_ae_name, struct qos_spec * qos) return -1; } - 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].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; @@ -480,8 +544,6 @@ int flow_alloc(char * dst_name, char * src_ae_name, struct qos_spec * qos) pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); - shm_ap_rbuff_open_port(ai.rb, recv_msg->port_id); - irm_msg__free_unpacked(recv_msg, NULL); return fd; @@ -548,7 +610,7 @@ int flow_dealloc(int fd) return -ENOTALLOC; } - if (shm_ap_rbuff_close_port(ai.rb, ai.flows[fd].port_id) == -EBUSY) { + if (shm_rbuff_block(ai.flows[fd].rx_rb) == -EBUSY) { pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); return -EBUSY; @@ -559,8 +621,10 @@ int flow_dealloc(int fd) port_destroy(&ai.ports[msg.port_id]); ai.flows[fd].port_id = -1; - shm_ap_rbuff_close(ai.flows[fd].rb); - ai.flows[fd].rb = NULL; + shm_rbuff_destroy(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; ai.flows[fd].oflags = 0; ai.flows[fd].api = -1; if (ai.flows[fd].timeout != NULL) { @@ -604,9 +668,9 @@ int flow_cntl(int fd, int cmd, int oflags) case FLOW_F_SETFL: /* SET FLOW FLAGS */ ai.flows[fd].oflags = oflags; if (oflags & FLOW_O_WRONLY) - shm_ap_rbuff_close_port(ai.rb, ai.flows[fd].port_id); + shm_rbuff_block(ai.flows[fd].rx_rb); if (oflags & FLOW_O_RDWR) - shm_ap_rbuff_open_port(ai.rb, ai.flows[fd].port_id); + shm_rbuff_unblock(ai.flows[fd].rx_rb); pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); return old; @@ -620,7 +684,6 @@ int flow_cntl(int fd, int cmd, int oflags) ssize_t flow_write(int fd, void * buf, size_t count) { ssize_t idx; - struct rb_entry e; if (buf == NULL) return 0; @@ -653,13 +716,10 @@ ssize_t flow_write(int fd, void * buf, size_t count) if (idx < 0) { pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); - return -idx; + return idx; } - e.index = idx; - e.port_id = ai.flows[fd].port_id; - - if (shm_ap_rbuff_write(ai.flows[fd].rb, &e) < 0) { + if (shm_rbuff_write(ai.flows[fd].tx_rb, idx) < 0) { shm_rdrbuff_remove(ai.rdrb, idx); pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); @@ -667,7 +727,7 @@ ssize_t flow_write(int fd, void * buf, size_t count) } } else { /* blocking */ struct shm_rdrbuff * rdrb = ai.rdrb; - pid_t api = ai.flows[fd].api; + pid_t api = ai.flows[fd].api; pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); @@ -681,17 +741,16 @@ ssize_t flow_write(int fd, void * buf, size_t count) pthread_rwlock_rdlock(&ai.data_lock); pthread_rwlock_rdlock(&ai.flows_lock); - e.index = idx; - e.port_id = ai.flows[fd].port_id; - - if (shm_ap_rbuff_write(ai.flows[fd].rb, &e) < 0) { - shm_rdrbuff_remove(ai.rdrb, e.index); + if (shm_rbuff_write(ai.flows[fd].tx_rb, idx) < 0) { + shm_rdrbuff_remove(ai.rdrb, idx); pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); return -ENOTALLOC; } } + shm_flow_set_notify(ai.flows[fd].set, ai.flows[fd].port_id); + pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); @@ -717,15 +776,14 @@ ssize_t flow_read(int fd, void * buf, size_t count) } if (ai.flows[fd].oflags & FLOW_O_NONBLOCK) { - idx = shm_ap_rbuff_read_port(ai.rb, ai.flows[fd].port_id); + idx = shm_rbuff_read(ai.flows[fd].rx_rb); pthread_rwlock_unlock(&ai.flows_lock); } else { - struct shm_ap_rbuff * rb = ai.rb; - int port_id = ai.flows[fd].port_id; - struct timespec * timeout = ai.flows[fd].timeout; + struct shm_rbuff * rb = ai.flows[fd].rx_rb; + struct timespec * timeout = ai.flows[fd].timeout; pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); - idx = shm_ap_rbuff_read_port_b(rb, port_id, timeout); + idx = shm_rbuff_read_b(rb, timeout); pthread_rwlock_rdlock(&ai.data_lock); } @@ -757,79 +815,163 @@ struct flow_set * flow_set_create() if (set == NULL) return NULL; - if (pthread_rwlock_init(&set->lock, NULL)) { + assert(ai.fqueues); + + set->idx = bmp_allocate(ai.fqueues); + if (!bmp_is_id_valid(ai.fqueues, set->idx)) { free(set); return NULL; } - memset(set->b, 0, IRMD_MAX_FLOWS); - memset(set->s, 0, IRMD_MAX_FLOWS); + return set; +} - set->dirty = true; +void flow_set_destroy(struct flow_set * set) +{ + if (set == NULL) + return; - return set; + flow_set_zero(set); + bmp_release(ai.fqueues, set->idx); + free(set); } -void flow_set_zero(struct flow_set * set) +struct fqueue * fqueue_create() { - pthread_rwlock_wrlock(&set->lock); - memset(set->b, 0, IRMD_MAX_FLOWS); - set->dirty = true; - pthread_rwlock_unlock(&set->lock); + struct fqueue * fq = malloc(sizeof(*fq)); + if (fq == NULL) + return NULL; + + memset(fq->fqueue, -1, SHM_BUFFER_SIZE); + fq->fqsize = 0; + fq->next = 0; + + return fq; } -void flow_set_add(struct flow_set * set, int fd) +void fqueue_destroy(struct fqueue * fq) { - pthread_rwlock_wrlock(&set->lock); - set->b[ai.flows[fd].port_id] = true; - set->dirty = true; - pthread_rwlock_unlock(&set->lock); + if (fq == NULL) + return + free(fq); } -void flow_set_del(struct flow_set * set, int fd) +void flow_set_zero(struct flow_set * set) { - pthread_rwlock_wrlock(&set->lock); - set->b[ai.flows[fd].port_id] = false; - set->dirty = true; - pthread_rwlock_unlock(&set->lock); + if (set == NULL) + return; + + pthread_rwlock_rdlock(&ai.data_lock); + + shm_flow_set_zero(ai.fqset, set->idx); + + pthread_rwlock_unlock(&ai.data_lock); } -bool flow_set_has(struct flow_set * set, int fd) +int flow_set_add(struct flow_set * set, int fd) { - bool ret; - pthread_rwlock_rdlock(&set->lock); - ret = set->b[ai.flows[fd].port_id]; - pthread_rwlock_unlock(&set->lock); + int ret; + + if (set == NULL) + return -EINVAL; + + pthread_rwlock_rdlock(&ai.data_lock); + pthread_rwlock_rdlock(&ai.flows_lock); + + ret = shm_flow_set_add(ai.fqset, set->idx, ai.flows[fd].port_id); + + pthread_rwlock_unlock(&ai.flows_lock); + pthread_rwlock_unlock(&ai.data_lock); + return ret; } -void flow_set_destroy(struct flow_set * set) +void flow_set_del(struct flow_set * set, int fd) { - pthread_rwlock_destroy(&set->lock); - free(set); + if (set == NULL) + return; + + pthread_rwlock_rdlock(&ai.data_lock); + pthread_rwlock_rdlock(&ai.flows_lock); + + if (ai.flows[fd].port_id >= 0) + shm_flow_set_del(ai.fqset, set->idx, ai.flows[fd].port_id); + + pthread_rwlock_unlock(&ai.flows_lock); + pthread_rwlock_unlock(&ai.data_lock); } -static void flow_set_cpy(struct flow_set * set) +bool flow_set_has(struct flow_set * set, int fd) { - pthread_rwlock_rdlock(&set->lock); - if (set->dirty) - memcpy(set->s, set->b, IRMD_MAX_FLOWS); - set->dirty = false; - pthread_rwlock_unlock(&set->lock); + bool ret = false; + + if (set == NULL || fd < 0) + return false; + + pthread_rwlock_rdlock(&ai.data_lock); + pthread_rwlock_rdlock(&ai.flows_lock); + + if (ai.flows[fd].port_id < 0) { + pthread_rwlock_unlock(&ai.flows_lock); + pthread_rwlock_unlock(&ai.data_lock); + return false; + } + + ret = (shm_flow_set_has(ai.fqset, set->idx, ai.flows[fd].port_id) == 1); + + pthread_rwlock_unlock(&ai.flows_lock); + pthread_rwlock_unlock(&ai.data_lock); + + return ret; } -int flow_select(struct flow_set * set, const struct timespec * timeout) +int fqueue_next(struct fqueue * fq) { - int port_id; - if (set == NULL) { - port_id = shm_ap_rbuff_peek_b(ai.rb, NULL, timeout); - } else { - flow_set_cpy(set); - port_id = shm_ap_rbuff_peek_b(ai.rb, (bool *) set->s, timeout); + int fd; + + if (fq == NULL) + return -EINVAL; + + if (fq->next == fq->fqsize) { + fq->fqsize = 0; + fq->next = 0; + return -EPERM; } - if (port_id < 0) - return port_id; - return ai.ports[port_id].fd; + + pthread_rwlock_rdlock(&ai.data_lock); + pthread_rwlock_rdlock(&ai.flows_lock); + + fd = ai.ports[fq->fqueue[fq->next++]].fd; + + pthread_rwlock_unlock(&ai.flows_lock); + pthread_rwlock_unlock(&ai.data_lock); + + return fd; +} + +int flow_event_wait(struct flow_set * set, + struct fqueue * fq, + const struct timespec * timeout) +{ + int ret; + + if (set == NULL) + return -EINVAL; + + if (fq->fqsize > 0) + return 0; + + ret = shm_flow_set_wait(ai.fqset, set->idx, fq->fqueue, timeout); + if (ret == -ETIMEDOUT) + return -ETIMEDOUT; + + if (ret < 0) + return ret; + + fq->fqsize = ret; + fq->next = 0; + + return 0; } /* ipcp-dev functions */ @@ -848,8 +990,8 @@ int np1_flow_alloc(pid_t n_api, int port_id) return -1; } - ai.flows[fd].rb = shm_ap_rbuff_open(n_api); - if (ai.flows[fd].rb == NULL) { + ai.flows[fd].rx_rb = shm_rbuff_create(port_id); + if (ai.flows[fd].rx_rb == NULL) { bmp_release(ai.fds, fd); pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); @@ -863,8 +1005,6 @@ int np1_flow_alloc(pid_t n_api, int port_id) ai.ports[port_id].fd = fd; port_set_state(&ai.ports[port_id], PORT_ID_ASSIGNED); - shm_ap_rbuff_open_port(ai.rb, port_id); - pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); @@ -890,7 +1030,6 @@ int np1_flow_dealloc(int port_id) int np1_flow_resp(pid_t n_api, int port_id) { int fd; - struct shm_ap_rbuff * rb; port_wait_assign(&ai.ports[port_id]); @@ -904,18 +1043,26 @@ int np1_flow_resp(pid_t n_api, int port_id) return fd; } - rb = shm_ap_rbuff_open(n_api); - if (rb == NULL) { + 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); port_destroy(&ai.ports[port_id]); pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); return -1; } - ai.flows[fd].rb = rb; - - shm_ap_rbuff_open_port(ai.rb, port_id); + ai.flows[fd].set = shm_flow_set_open(n_api); + 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); + port_destroy(&ai.ports[port_id]); + pthread_rwlock_unlock(&ai.flows_lock); + pthread_rwlock_unlock(&ai.data_lock); + return -1; + } pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); @@ -929,9 +1076,9 @@ int ipcp_create_r(pid_t api) irm_msg_t * recv_msg = NULL; int ret = -1; - msg.code = IRM_MSG_CODE__IPCP_CREATE_R; - msg.has_api = true; - msg.api = api; + msg.code = IRM_MSG_CODE__IPCP_CREATE_R; + msg.has_api = true; + msg.api = api; recv_msg = send_recv_irm_msg(&msg); if (recv_msg == NULL) @@ -958,11 +1105,11 @@ int ipcp_flow_req_arr(pid_t api, char * dst_name, char * src_ae_name) if (dst_name == NULL || src_ae_name == NULL) return -EINVAL; - msg.code = IRM_MSG_CODE__IPCP_FLOW_REQ_ARR; - msg.has_api = true; - msg.api = api; - msg.dst_name = dst_name; - msg.ae_name = src_ae_name; + msg.code = IRM_MSG_CODE__IPCP_FLOW_REQ_ARR; + msg.has_api = true; + msg.api = api; + msg.dst_name = dst_name; + msg.ae_name = src_ae_name; pthread_rwlock_rdlock(&ai.data_lock); pthread_rwlock_wrlock(&ai.flows_lock); @@ -974,7 +1121,7 @@ int ipcp_flow_req_arr(pid_t api, char * dst_name, char * src_ae_name) return -1; /* -ENOMOREFDS */ } - ai.flows[fd].rb = NULL; + ai.flows[fd].tx_rb = NULL; pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); @@ -996,8 +1143,16 @@ 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); + if (ai.flows[fd].rx_rb == NULL) { + ai.flows[fd].port_id = -1; + port_destroy(&ai.ports[port_id]); + pthread_rwlock_unlock(&ai.flows_lock); + pthread_rwlock_unlock(&ai.data_lock); + return -1; + } + ai.flows[fd].port_id = port_id; - ai.flows[fd].rb = NULL; ai.ports[port_id].fd = fd; port_set_state(&(ai.ports[port_id]), PORT_ID_ASSIGNED); @@ -1019,16 +1174,13 @@ int ipcp_flow_alloc_reply(int fd, int response) pthread_rwlock_rdlock(&ai.data_lock); pthread_rwlock_rdlock(&ai.flows_lock); - msg.port_id = ai.flows[fd].port_id; + msg.port_id = ai.flows[fd].port_id; pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); msg.has_response = true; msg.response = response; - if (response) - shm_ap_rbuff_open_port(ai.rb, msg.port_id); - recv_msg = send_recv_irm_msg(&msg); if (recv_msg == NULL) return -1; @@ -1039,6 +1191,26 @@ int ipcp_flow_alloc_reply(int fd, int response) } ret = recv_msg->result; + + pthread_rwlock_wrlock(&ai.flows_lock); + + 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); + 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); + return -1; + } + + pthread_rwlock_unlock(&ai.flows_lock); + irm_msg__free_unpacked(recv_msg, NULL); return ret; @@ -1061,7 +1233,7 @@ int ipcp_flow_read(int fd, struct shm_du_buff ** sdb) pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); - idx = shm_ap_rbuff_read_port(ai.rb, port_id); + idx = shm_rbuff_read(ai.flows[fd].rx_rb); if (idx < 0) { pthread_rwlock_rdlock(&ai.data_lock); pthread_rwlock_rdlock(&ai.flows_lock); @@ -1081,7 +1253,7 @@ int ipcp_flow_read(int fd, struct shm_du_buff ** sdb) int ipcp_flow_write(int fd, struct shm_du_buff * sdb) { - struct rb_entry e; + ssize_t idx; if (sdb == NULL) return -EINVAL; @@ -1095,16 +1267,16 @@ int ipcp_flow_write(int fd, struct shm_du_buff * sdb) return -EPERM; } - if (ai.flows[fd].rb == NULL) { + if (ai.flows[fd].tx_rb == NULL) { pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); return -EPERM; } - e.index = shm_du_buff_get_idx(sdb); - e.port_id = ai.flows[fd].port_id; + idx = shm_du_buff_get_idx(sdb); - shm_ap_rbuff_write(ai.flows[fd].rb, &e); + shm_rbuff_write(ai.flows[fd].tx_rb, idx); + shm_flow_set_notify(ai.flows[fd].set, ai.flows[fd].port_id); pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); @@ -1112,46 +1284,28 @@ int ipcp_flow_write(int fd, struct shm_du_buff * sdb) return 0; } -struct rb_entry * local_flow_read(int fd) +ssize_t local_flow_read(int fd) { - int port_id; - struct rb_entry * e = NULL; - - pthread_rwlock_rdlock(&ai.data_lock); - pthread_rwlock_rdlock(&ai.flows_lock); - - port_id = ai.flows[fd].port_id; - - pthread_rwlock_unlock(&ai.flows_lock); - pthread_rwlock_unlock(&ai.data_lock); - - if (port_id != -1) { - e = malloc(sizeof(*e)); - if (e == NULL) - return NULL; - e->index = shm_ap_rbuff_read_port(ai.rb, port_id); - } - - return e; + return shm_rbuff_read(ai.flows[fd].rx_rb); } -int local_flow_write(int fd, struct rb_entry * e) +int local_flow_write(int fd, ssize_t idx) { - if (e == NULL || fd < 0) + if (fd < 0) return -EINVAL; pthread_rwlock_rdlock(&ai.data_lock); pthread_rwlock_rdlock(&ai.flows_lock); - if (ai.flows[fd].rb == NULL) { + if (ai.flows[fd].tx_rb == NULL) { pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); return -EPERM; } - e->port_id = ai.flows[fd].port_id; + shm_rbuff_write(ai.flows[fd].tx_rb, idx); - shm_ap_rbuff_write(ai.flows[fd].rb, e); + shm_flow_set_notify(ai.flows[fd].set, ai.flows[fd].port_id); pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); @@ -1159,22 +1313,26 @@ int local_flow_write(int fd, struct rb_entry * e) return 0; } -int ipcp_read_shim(struct shm_du_buff ** sdb) +int ipcp_read_shim(int fd, struct shm_du_buff ** sdb) { - int fd; - struct rb_entry * e = shm_ap_rbuff_read(ai.rb); + ssize_t idx; pthread_rwlock_rdlock(&ai.data_lock); pthread_rwlock_rdlock(&ai.flows_lock); - fd = ai.ports[e->port_id].fd; + if (ai.flows[fd].rx_rb == NULL) { + pthread_rwlock_unlock(&ai.flows_lock); + pthread_rwlock_unlock(&ai.data_lock); + return -EPERM; + } - *sdb = shm_rdrbuff_get(ai.rdrb, e->index); + idx = shm_rbuff_read(ai.flows[fd].rx_rb); + *sdb = shm_rdrbuff_get(ai.rdrb, idx); pthread_rwlock_unlock(&ai.flows_lock); pthread_rwlock_unlock(&ai.data_lock); - return fd; + return 0; } void ipcp_flow_del(struct shm_du_buff * sdb) -- 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 'src/lib/dev.c') 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