diff options
Diffstat (limited to 'src/ipcpd/local/main.c')
| -rw-r--r-- | src/ipcpd/local/main.c | 494 | 
1 files changed, 99 insertions, 395 deletions
| diff --git a/src/ipcpd/local/main.c b/src/ipcpd/local/main.c index c0809429..1ccec0c0 100644 --- a/src/ipcpd/local/main.c +++ b/src/ipcpd/local/main.c @@ -22,17 +22,10 @@  #include <ouroboros/config.h>  #include "ipcp.h" -#include "flow.h"  #include <ouroboros/errno.h> -#include <ouroboros/shm_rdrbuff.h> -#include <ouroboros/shm_ap_rbuff.h> -#include <ouroboros/list.h> -#include <ouroboros/utils.h> -#include <ouroboros/ipcp.h> -#include <ouroboros/irm_config.h> -#include <ouroboros/bitmap.h> -#include <ouroboros/shared.h>  #include <ouroboros/dev.h> +#include <ouroboros/ipcp-dev.h> +#include <ouroboros/local-dev.h>  #define OUROBOROS_PREFIX "ipcpd/local"  #include <ouroboros/logs.h> @@ -46,176 +39,51 @@  #define THIS_TYPE IPCP_LOCAL -#define shim_data(type) ((struct ipcp_local_data *) type->data) -  /* global for trapping signal */  int irmd_api; -/* this IPCP's data */ -#ifdef MAKE_CHECK -extern struct ipcp * _ipcp; /* defined in test */ -#else -struct ipcp * _ipcp; -#endif - -/* - * copied from ouroboros/dev. The shim needs access to the internals - * because it doesn't follow all steps necessary steps to get - * the info - */ - -/* the shim needs access to these internals */ -struct shim_ap_data { -        pid_t                 api; -        struct shm_rdrbuff *  rdrb; -        struct bmp *          fds; -        struct shm_ap_rbuff * rb; - -        int                   in_out[AP_MAX_FLOWS]; +struct { +        int                   in_out[IRMD_MAX_FLOWS]; -        struct flow           flows[AP_MAX_FLOWS]; -        pthread_rwlock_t      flows_lock; - -        pthread_t             mainloop; +        pthread_rwlock_t      lock;          pthread_t             sduloop; +} local_data; -} * _ap_instance; - -static int shim_ap_init() +void local_data_init()  {          int i; +        for (i = 0; i < IRMD_MAX_FLOWS; ++i) +                local_data.in_out[i] = -1; -        _ap_instance = malloc(sizeof(struct shim_ap_data)); -        if (_ap_instance == NULL) { -                return -1; -        } - -        _ap_instance->api = getpid(); - -        _ap_instance->fds = bmp_create(AP_MAX_FLOWS, 0); -        if (_ap_instance->fds == NULL) { -                free(_ap_instance); -                return -1; -        } - -        _ap_instance->rdrb = shm_rdrbuff_open(); -        if (_ap_instance->rdrb == NULL) { -                bmp_destroy(_ap_instance->fds); -                free(_ap_instance); -                return -1; -        } - -        _ap_instance->rb = shm_ap_rbuff_create_n(); -        if (_ap_instance->rb == NULL) { -                shm_rdrbuff_close(_ap_instance->rdrb); -                bmp_destroy(_ap_instance->fds); -                free(_ap_instance); -                return -1; -        } - -        for (i = 0; i < AP_MAX_FLOWS; i ++) { -                _ap_instance->flows[i].rb = NULL; -                _ap_instance->flows[i].port_id = -1; -                _ap_instance->flows[i].state = FLOW_NULL; -                _ap_instance->in_out[i] = -1; -        } - -        pthread_rwlock_init(&_ap_instance->flows_lock, NULL); - -        return 0; -} - -void shim_ap_fini() -{ -        int i = 0; - -        if (_ap_instance == NULL) -                return; - -        pthread_rwlock_rdlock(&_ipcp->state_lock); - -        if (_ipcp->state != IPCP_SHUTDOWN) -                LOG_WARN("Cleaning up AP while not in shutdown."); - -        if (_ap_instance->fds != NULL) -                bmp_destroy(_ap_instance->fds); - -        /* remove all remaining sdus */ -        while ((i = shm_ap_rbuff_peek_idx(_ap_instance->rb)) >= 0) -                shm_rdrbuff_remove(_ap_instance->rdrb, i); - -        if (_ap_instance->rdrb != NULL) -                shm_rdrbuff_close(_ap_instance->rdrb); -        if (_ap_instance->rb != NULL) -                shm_ap_rbuff_destroy(_ap_instance->rb); - -        pthread_rwlock_wrlock(&_ap_instance->flows_lock); - -        for (i = 0; i < AP_MAX_FLOWS; i ++) -                if (_ap_instance->flows[i].rb != NULL) -                        shm_ap_rbuff_close(_ap_instance->flows[i].rb); - -        pthread_rwlock_unlock(&_ap_instance->flows_lock); -        pthread_rwlock_unlock(&_ipcp->state_lock); - -        free(_ap_instance); +        pthread_rwlock_init(&local_data.lock, NULL);  } -/* only call this under flows_lock */ -static int port_id_to_fd(int port_id) +void local_data_fini()  { -        int i; - -        for (i = 0; i < AP_MAX_FLOWS; ++i) { -                if (_ap_instance->flows[i].port_id == port_id -                    && _ap_instance->flows[i].state != FLOW_NULL) -                        return i; -        } - -        return -1; +        pthread_rwlock_destroy(&local_data.lock);  } -/* - * end copy from dev.c - */ - -/* FIXME: if we move _ap_instance to dev.h, we can reuse it everywhere */  static void * ipcp_local_sdu_loop(void * o)  {          while (true) { -                struct rb_entry * e; -                int fd; - -                e = shm_ap_rbuff_read(_ap_instance->rb); -                if (e == NULL) { -                        continue; -                } +                struct rb_entry e; +                int fd = local_flow_read(&e); -                pthread_rwlock_rdlock(&_ipcp->state_lock); +                pthread_rwlock_rdlock(&ipcpi.state_lock); -                if (_ipcp->state != IPCP_ENROLLED) { -                        pthread_rwlock_unlock(&_ipcp->state_lock); +                if (ipcp_get_state() != IPCP_ENROLLED) { +                        pthread_rwlock_unlock(&ipcpi.state_lock);                          return (void *) 1; /* -ENOTENROLLED */                  } -                pthread_rwlock_rdlock(&_ap_instance->flows_lock); -                fd = _ap_instance->in_out[port_id_to_fd(e->port_id)]; -                if (fd == -1) { -                        pthread_rwlock_unlock(&_ap_instance->flows_lock); -                        pthread_rwlock_unlock(&_ipcp->state_lock); -                        free(e); -                        continue; -                } - -                e->port_id = _ap_instance->flows[fd].port_id; - -                while (shm_ap_rbuff_write(_ap_instance->flows[fd].rb, e) < 0) -                        ; +                pthread_rwlock_rdlock(&local_data.lock); +                fd = local_data.in_out[fd]; +                pthread_rwlock_unlock(&local_data.lock); -                pthread_rwlock_unlock(&_ap_instance->flows_lock); -                pthread_rwlock_unlock(&_ipcp->state_lock); +                if (fd != -1) +                        local_flow_write(fd, &e); -                free(e); +                pthread_rwlock_unlock(&ipcpi.state_lock);          }          return (void *) 1; @@ -223,10 +91,6 @@ static void * ipcp_local_sdu_loop(void * o)  void ipcp_sig_handler(int sig, siginfo_t * info, void * c)  { -        sigset_t  sigset; -        sigemptyset(&sigset); -        sigaddset(&sigset, SIGINT); -          switch(sig) {          case SIGINT:          case SIGTERM: @@ -236,11 +100,11 @@ void ipcp_sig_handler(int sig, siginfo_t * info, void * c)                          LOG_DBG("IPCP %d terminating by order of %d. Bye.",                                  getpid(), info->si_pid); -                        pthread_rwlock_wrlock(&_ipcp->state_lock); +                        pthread_rwlock_wrlock(&ipcpi.state_lock); -                        ipcp_set_state(_ipcp, IPCP_SHUTDOWN); +                        ipcp_set_state(IPCP_SHUTDOWN); -                        pthread_rwlock_unlock(&_ipcp->state_lock); +                        pthread_rwlock_unlock(&ipcpi.state_lock);                  }          default:                  return; @@ -254,307 +118,154 @@ static int ipcp_local_bootstrap(struct dif_config * conf)                  return -1;          } -        pthread_rwlock_wrlock(&_ipcp->state_lock); +        pthread_rwlock_wrlock(&ipcpi.state_lock); -        if (ipcp_get_state(_ipcp) != IPCP_INIT) { -                pthread_rwlock_unlock(&_ipcp->state_lock); +        if (ipcp_get_state() != IPCP_INIT) { +                pthread_rwlock_unlock(&ipcpi.state_lock);                  LOG_ERR("IPCP in wrong state.");                  return -1;          } -        ipcp_set_state(_ipcp, IPCP_ENROLLED); +        ipcp_set_state(IPCP_ENROLLED); -        pthread_create(&_ap_instance->sduloop, -                       NULL, -                       ipcp_local_sdu_loop, -                       NULL); +        pthread_create(&local_data.sduloop, NULL, ipcp_local_sdu_loop, NULL); -        pthread_rwlock_unlock(&_ipcp->state_lock); +        pthread_rwlock_unlock(&ipcpi.state_lock); -        LOG_DBG("Bootstrapped local IPCP with api %d.", -                getpid()); +        LOG_INFO("Bootstrapped local IPCP with api %d.", getpid());          return 0;  }  static int ipcp_local_name_reg(char * name)  { -        pthread_rwlock_rdlock(&_ipcp->state_lock); +        pthread_rwlock_rdlock(&ipcpi.state_lock); -        if (ipcp_data_add_reg_entry(_ipcp->data, name)) { -                pthread_rwlock_unlock(&_ipcp->state_lock); +        if (ipcp_data_add_reg_entry(ipcpi.data, name)) { +                pthread_rwlock_unlock(&ipcpi.state_lock);                  LOG_DBGF("Failed to add %s to local registry.", name);                  return -1;          } -        pthread_rwlock_unlock(&_ipcp->state_lock); +        pthread_rwlock_unlock(&ipcpi.state_lock); -        LOG_DBG("Registered %s.", name); +        LOG_INFO("Registered %s.", name);          return 0;  }  static int ipcp_local_name_unreg(char * name)  { -        pthread_rwlock_rdlock(&_ipcp->state_lock); +        pthread_rwlock_rdlock(&ipcpi.state_lock); -        ipcp_data_del_reg_entry(_ipcp->data, name); +        ipcp_data_del_reg_entry(ipcpi.data, name); -        pthread_rwlock_unlock(&_ipcp->state_lock); +        pthread_rwlock_unlock(&ipcpi.state_lock); + +        LOG_INFO("Unregistered %s.", name);          return 0;  } -static int ipcp_local_flow_alloc(pid_t         n_api, -                                 int           port_id, +static int ipcp_local_flow_alloc(int           fd,                                   char *        dst_name,                                   char *        src_ae_name,                                   enum qos_cube qos)  { -        int in_fd = -1;          int out_fd = -1; -        struct shm_ap_rbuff * rb; - -        LOG_INFO("Allocating flow to %s.", dst_name); +        LOG_DBG("Allocating flow to %s on fd %d.", dst_name, fd);          if (dst_name == NULL || src_ae_name == NULL)                  return -1;          /* This ipcpd has all QoS */ -        pthread_rwlock_rdlock(&_ipcp->state_lock); +        pthread_rwlock_rdlock(&ipcpi.state_lock); -        if (ipcp_get_state(_ipcp) != IPCP_ENROLLED) { -                pthread_rwlock_unlock(&_ipcp->state_lock); +        if (ipcp_get_state() != IPCP_ENROLLED) { +                pthread_rwlock_unlock(&ipcpi.state_lock);                  LOG_DBGF("Won't register with non-enrolled IPCP.");                  return -1; /* -ENOTENROLLED */          } -        rb = shm_ap_rbuff_open_s(n_api); -        if (rb == NULL) { -                pthread_rwlock_unlock(&_ipcp->state_lock); -                return -1; /* -ENORBUFF */ -        } - -        pthread_rwlock_wrlock(&_ap_instance->flows_lock); +        pthread_rwlock_wrlock(&local_data.lock); -        in_fd = bmp_allocate(_ap_instance->fds); -        if (!bmp_is_id_valid(_ap_instance->fds, in_fd)) { -                pthread_rwlock_unlock(&_ap_instance->flows_lock); -                pthread_rwlock_unlock(&_ipcp->state_lock); -                return -EMFILE; -        } +        out_fd = ipcp_flow_req_arr(getpid(), dst_name, src_ae_name); -        _ap_instance->flows[in_fd].port_id = port_id; -        _ap_instance->flows[in_fd].state   = FLOW_PENDING; -        _ap_instance->flows[in_fd].rb      = rb; +        local_data.in_out[fd]  = out_fd; +        local_data.in_out[out_fd] = fd; -        LOG_DBGF("Pending local flow with port_id %d.", port_id); +        pthread_rwlock_unlock(&local_data.lock); +        pthread_rwlock_unlock(&ipcpi.state_lock); -        /* reply to IRM */ -        port_id = ipcp_flow_req_arr(getpid(), -                                    dst_name, -                                    src_ae_name); - -        if (port_id < 0) { -                pthread_rwlock_unlock(&_ap_instance->flows_lock); -                pthread_rwlock_unlock(&_ipcp->state_lock); -                LOG_ERR("Could not get port id from IRMd"); -                /* shm_ap_rbuff_close(n_api); */ -                return -1; -        } - -        out_fd = bmp_allocate(_ap_instance->fds); -        if (!bmp_is_id_valid(_ap_instance->fds, out_fd)) { -                /* shm_ap_rbuff_close(n_api); */ -                pthread_rwlock_unlock(&_ap_instance->flows_lock); -                pthread_rwlock_unlock(&_ipcp->state_lock); -                return -1; /* -ENOMOREFDS */ -        } - -        _ap_instance->flows[out_fd].port_id = port_id; -        _ap_instance->flows[out_fd].rb      = NULL; -        _ap_instance->flows[out_fd].state   = FLOW_PENDING; - -        _ap_instance->in_out[in_fd]  = out_fd; -        _ap_instance->in_out[out_fd] = in_fd; - -        pthread_rwlock_unlock(&_ap_instance->flows_lock); -        pthread_rwlock_unlock(&_ipcp->state_lock); - -        LOG_DBGF("Pending local allocation request, port_id %d.", port_id); +        LOG_INFO("Pending local allocation request on fd %d.", fd);          return 0;  } -static int ipcp_local_flow_alloc_resp(pid_t n_api, -                                      int   port_id, -                                      int   response) +static int ipcp_local_flow_alloc_resp(int fd, int response)  { -        struct shm_ap_rbuff * rb; -        int in_fd = -1;          int out_fd = -1;          int ret = -1; +        LOG_DBG("Received response for fd %d: %d.", fd, response); +          if (response)                  return 0; -        pthread_rwlock_rdlock(&_ipcp->state_lock); - -        /* awaken pending flow */ - -        pthread_rwlock_wrlock(&_ap_instance->flows_lock); - -        in_fd = port_id_to_fd(port_id); -        if (in_fd < 0) { -                pthread_rwlock_unlock(&_ap_instance->flows_lock); -                pthread_rwlock_unlock(&_ipcp->state_lock); -                LOG_DBGF("Could not find flow with port_id %d.", port_id); -                return -1; -        } - -        if (_ap_instance->flows[in_fd].state != FLOW_PENDING) { -                pthread_rwlock_unlock(&_ap_instance->flows_lock); -                pthread_rwlock_unlock(&_ipcp->state_lock); -                LOG_DBGF("Flow was not pending."); -                return -1; -        } - -        rb = shm_ap_rbuff_open_s(n_api); -        if (rb == NULL) { -                LOG_ERR("Could not open N + 1 ringbuffer."); -                _ap_instance->flows[in_fd].state   = FLOW_NULL; -                _ap_instance->flows[in_fd].port_id = -1; -                _ap_instance->in_out[in_fd] = -1; -                pthread_rwlock_unlock(&_ap_instance->flows_lock); -                pthread_rwlock_unlock(&_ipcp->state_lock); -                return -1; -        } - -        _ap_instance->flows[in_fd].state = FLOW_ALLOCATED; -        _ap_instance->flows[in_fd].rb    = rb; - -        LOG_DBGF("Accepted flow, port_id %d on fd %d.", port_id, in_fd); +        pthread_rwlock_rdlock(&ipcpi.state_lock); -        out_fd = _ap_instance->in_out[in_fd]; +        out_fd = local_data.in_out[fd];          if (out_fd < 0) { -                pthread_rwlock_unlock(&_ap_instance->flows_lock); -                pthread_rwlock_unlock(&_ipcp->state_lock); -                LOG_DBGF("No pending local flow with port_id %d.", port_id); +                pthread_rwlock_unlock(&local_data.lock); +                pthread_rwlock_unlock(&ipcpi.state_lock);                  return -1;          } -        if (_ap_instance->flows[out_fd].state != FLOW_PENDING) { -                 /* FIXME: clean up other end */ -                pthread_rwlock_unlock(&_ap_instance->flows_lock); -                pthread_rwlock_unlock(&_ipcp->state_lock); -                LOG_DBGF("Flow was not pending."); -                return -1; -        } - -        _ap_instance->flows[out_fd].state = FLOW_ALLOCATED; - -        pthread_rwlock_unlock(&_ap_instance->flows_lock); -        pthread_rwlock_unlock(&_ipcp->state_lock); +        pthread_rwlock_unlock(&ipcpi.state_lock); -        if ((ret = ipcp_flow_alloc_reply(getpid(), -                                         _ap_instance->flows[out_fd].port_id, -                                         response)) < 0) { -                return -1; /* -EPIPE */ -        } +        if ((ret = ipcp_flow_alloc_reply(out_fd, response)) < 0) +                return -1; -        LOG_INFO("Flow allocation completed, port_ids (%d, %d).", -                 _ap_instance->flows[out_fd].port_id, -                 _ap_instance->flows[in_fd].port_id); +        LOG_INFO("Flow allocation completed, fds (%d, %d).", out_fd, fd);          return ret;  } -static int ipcp_local_flow_dealloc(int port_id) +static int ipcp_local_flow_dealloc(int fd)  { -        int fd = -1; -        struct shm_ap_rbuff * rb; - -        pthread_rwlock_rdlock(&_ipcp->state_lock); -        pthread_rwlock_wrlock(&_ap_instance->flows_lock); - -        fd = port_id_to_fd(port_id); -        if (fd < 0) { -                pthread_rwlock_unlock(&_ap_instance->flows_lock); -                pthread_rwlock_unlock(&_ipcp->state_lock); -                LOG_DBGF("Could not find flow with port_id %d.", port_id); -                return 0; -        } - -        bmp_release(_ap_instance->fds, fd); - -        if (_ap_instance->in_out[fd] != -1) -                _ap_instance->in_out[_ap_instance->in_out[fd]] = -1; - -        _ap_instance->in_out[fd] = -1; - -        _ap_instance->flows[fd].state   = FLOW_NULL; -        _ap_instance->flows[fd].port_id = -1; -        rb = _ap_instance->flows[fd].rb; -        _ap_instance->flows[fd].rb      = NULL; - -        pthread_rwlock_unlock(&_ap_instance->flows_lock); - -        if (rb != NULL) -                shm_ap_rbuff_close(rb); - -        pthread_rwlock_unlock(&_ipcp->state_lock); - -        LOG_DBGF("Flow with port_id %d deallocated.", port_id); - -        return 0; -} - -static struct ipcp * ipcp_local_create() -{ -        struct ipcp * i; -        struct ipcp_ops *  ops; +        int out_fd = -1; -        i = ipcp_instance_create(); -        if (i == NULL) -                return NULL; +        pthread_rwlock_rdlock(&ipcpi.state_lock); +        pthread_rwlock_wrlock(&local_data.lock); -        i->data = ipcp_data_create(); -        if (i->data == NULL) { -                free(i); -                return NULL; -        } +        out_fd = local_data.in_out[fd]; -        if (ipcp_data_init(i->data, THIS_TYPE) == NULL) { -                free(i->data); -                free(i); -                return NULL; +        if (out_fd != -1) { +                local_data.in_out[out_fd] = -1; +                flow_dealloc(out_fd);          } -        ops = malloc(sizeof(*ops)); -        if (ops == NULL) { -                free(i->data); -                free(i); -                return NULL; -        } +        local_data.in_out[fd] = -1; -        ops->ipcp_bootstrap       = ipcp_local_bootstrap; -        ops->ipcp_enroll          = NULL;                       /* shim */ -        ops->ipcp_name_reg        = ipcp_local_name_reg; -        ops->ipcp_name_unreg      = ipcp_local_name_unreg; -        ops->ipcp_flow_alloc      = ipcp_local_flow_alloc; -        ops->ipcp_flow_alloc_resp = ipcp_local_flow_alloc_resp; -        ops->ipcp_flow_dealloc    = ipcp_local_flow_dealloc; +        pthread_rwlock_unlock(&local_data.lock); +        pthread_rwlock_unlock(&ipcpi.state_lock); -        i->ops = ops; +        LOG_INFO("Flow with fd %d deallocated.", fd); -        i->state = IPCP_INIT; - -        return i; +        return 0;  } -#ifndef MAKE_CHECK +static struct ipcp_ops local_ops = { +        .ipcp_bootstrap       = ipcp_local_bootstrap, +        .ipcp_enroll          = NULL,                       /* shim */ +        .ipcp_name_reg        = ipcp_local_name_reg, +        .ipcp_name_unreg      = ipcp_local_name_unreg, +        .ipcp_flow_alloc      = ipcp_local_flow_alloc, +        .ipcp_flow_alloc_resp = ipcp_local_flow_alloc_resp, +        .ipcp_flow_dealloc    = ipcp_local_flow_dealloc +};  int main(int argc, char * argv[])  { @@ -571,7 +282,9 @@ int main(int argc, char * argv[])                  exit(EXIT_FAILURE);          } -        if (shim_ap_init() < 0) { +        local_data_init(); + +        if (ap_init(NULL) < 0) {                  close_logfile();                  exit(EXIT_FAILURE);          } @@ -591,17 +304,13 @@ int main(int argc, char * argv[])          sigaction(SIGHUP,  &sig_act, NULL);          sigaction(SIGPIPE, &sig_act, NULL); -        _ipcp = ipcp_local_create(); -        if (_ipcp == NULL) { -                LOG_ERR("Failed to create IPCP."); +        pthread_sigmask(SIG_BLOCK, &sigset, NULL); + +        if (ipcp_init(THIS_TYPE, &local_ops) < 0) {                  close_logfile();                  exit(EXIT_FAILURE);          } -        pthread_sigmask(SIG_BLOCK, &sigset, NULL); - -        pthread_create(&_ap_instance->mainloop, NULL, ipcp_main_loop, _ipcp); -          pthread_sigmask(SIG_UNBLOCK, &sigset, NULL);          if (ipcp_create_r(getpid())) { @@ -610,21 +319,16 @@ int main(int argc, char * argv[])                  exit(EXIT_FAILURE);          } -        pthread_join(_ap_instance->mainloop, NULL); - -        pthread_cancel(_ap_instance->sduloop); -        pthread_join(_ap_instance->sduloop, NULL); +        ipcp_fini(); -        shim_ap_fini(); +        pthread_cancel(local_data.sduloop); +        pthread_join(local_data.sduloop, NULL); -        ipcp_data_destroy(_ipcp->data); +        ap_fini(); -        free(_ipcp->ops); -        free(_ipcp); +        local_data_fini();          close_logfile();          exit(EXIT_SUCCESS);  } - -#endif /* MAKE_CHECK */ | 
