Merged in dstaesse/ouroboros/be-dealloc (pull request #290)

lib: Clean up flow if alloc_res fails

2 files changed, 129 insertions, 81 deletions

diff --git a/src/lib/dev.c b/src/lib/dev.c
index a0c47403..3d527943 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -189,6 +189,37 @@ static int api_announce(char * ap_name)
         return ret;
 }
 
+static void reset_flow(int fd)
+{
+        assert (!(fd < 0));
+
+        if (ai.flows[fd].port_id != -1)
+                port_destroy(&ai.ports[ai.flows[fd].port_id]);
+
+        ai.flows[fd].port_id = -1;
+        if (ai.flows[fd].rx_rb != NULL) {
+                shm_rbuff_close(ai.flows[fd].rx_rb);
+                ai.flows[fd].rx_rb = NULL;
+        }
+        if (ai.flows[fd].tx_rb != NULL) {
+                shm_rbuff_close(ai.flows[fd].tx_rb);
+                ai.flows[fd].tx_rb = NULL;
+        }
+
+        if (ai.flows[fd].set != NULL) {
+                shm_flow_set_close(ai.flows[fd].set);
+                ai.flows[fd].set = NULL;
+        }
+
+        ai.flows[fd].oflags = 0;
+        ai.flows[fd].api = -1;
+
+        if (ai.flows[fd].timeout != NULL) {
+                free(ai.flows[fd].timeout);
+                ai.flows[fd].timeout = NULL;
+        }
+}
+
 int ap_init(char * ap_name)
 {
         int i = 0;
@@ -288,13 +319,8 @@ void ap_fini()
                         ssize_t idx;
                         while ((idx = shm_rbuff_read(ai.flows[i].rx_rb)) >= 0)
                                 shm_rdrbuff_remove(ai.rdrb, idx);
-                        shm_rbuff_close(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);
+                reset_flow(i);
         }
 
         for (i = 0; i < IRMD_MAX_FLOWS; ++i) {
@@ -315,7 +341,6 @@ void ap_fini()
         pthread_rwlock_destroy(&ai.data_lock);
 }
 
-
 int flow_accept(char ** ae_name, struct qos_spec * qos)
 {
         irm_msg_t msg = IRM_MSG__INIT;
@@ -355,6 +380,7 @@ int flow_accept(char ** ae_name, struct qos_spec * qos)
 
         ai.flows[fd].rx_rb = shm_rbuff_open(ai.api, recv_msg->port_id);
         if (ai.flows[fd].rx_rb == NULL) {
+                reset_flow(fd);
                 bmp_release(ai.fds, fd);
                 pthread_rwlock_unlock(&ai.flows_lock);
                 pthread_rwlock_unlock(&ai.data_lock);
@@ -364,9 +390,8 @@ int flow_accept(char ** ae_name, struct qos_spec * qos)
 
         ai.flows[fd].set = shm_flow_set_open(recv_msg->api);
         if (ai.flows[fd].set == NULL) {
+                reset_flow(fd);
                 bmp_release(ai.fds, fd);
-                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);
                 irm_msg__free_unpacked(recv_msg, NULL);
@@ -377,9 +402,7 @@ int flow_accept(char ** ae_name, struct qos_spec * qos)
         if (ae_name != NULL) {
                 *ae_name = strdup(recv_msg->ae_name);
                 if (*ae_name == NULL) {
-                        shm_rbuff_close(ai.flows[fd].tx_rb);
-                        shm_rbuff_close(ai.flows[fd].tx_rb);
-                        shm_flow_set_close(ai.flows[fd].set);
+                        reset_flow(fd);
                         bmp_release(ai.fds, fd);
                         pthread_rwlock_unlock(&ai.flows_lock);
                         pthread_rwlock_unlock(&ai.data_lock);
@@ -452,6 +475,7 @@ int flow_alloc_resp(int fd, int response)
         ai.flows[fd].tx_rb = shm_rbuff_open(ai.flows[fd].api,
                                             ai.flows[fd].port_id);
         if (ai.flows[fd].tx_rb == NULL) {
+                reset_flow(fd);
                 pthread_rwlock_unlock(&ai.flows_lock);
                 pthread_rwlock_unlock(&ai.data_lock);
                 return -1;
@@ -516,6 +540,7 @@ int flow_alloc(char * dst_name, char * src_ae_name, struct qos_spec * qos)
         ai.flows[fd].api     = recv_msg->api;
         ai.flows[fd].rx_rb   = shm_rbuff_open(ai.api, recv_msg->port_id);
         if (ai.flows[fd].rx_rb == NULL) {
+                reset_flow(fd);
                 bmp_release(ai.fds, fd);
                 pthread_rwlock_unlock(&ai.flows_lock);
                 pthread_rwlock_unlock(&ai.data_lock);
@@ -567,7 +592,7 @@ int flow_alloc_res(int fd)
 
         ai.flows[fd].set = shm_flow_set_open(ai.flows[fd].api);
         if (ai.flows[fd].set == NULL) {
-                shm_rbuff_close(ai.flows[fd].tx_rb);
+                reset_flow(fd);
                 pthread_rwlock_unlock(&ai.flows_lock);
                 pthread_rwlock_unlock(&ai.data_lock);
                 return -1;
@@ -577,19 +602,28 @@ int flow_alloc_res(int fd)
         pthread_rwlock_unlock(&ai.data_lock);
 
         recv_msg = send_recv_irm_msg_b(&msg);
-        if (recv_msg == NULL) {
-                return -1;
-        }
+        if (recv_msg == NULL)
+                result = -1;
 
         if (!recv_msg->has_result) {
                 irm_msg__free_unpacked(recv_msg, NULL);
-                return -1;
+                result = -1;
         }
 
         result = recv_msg->result;
 
         irm_msg__free_unpacked(recv_msg, NULL);
 
+        if (result) {
+                pthread_rwlock_rdlock(&ai.data_lock);
+                pthread_rwlock_wrlock(&ai.flows_lock);
+
+                reset_flow(fd);
+
+                pthread_rwlock_unlock(&ai.flows_lock);
+                pthread_rwlock_unlock(&ai.data_lock);
+        }
+
         return result;
 }
 
@@ -598,6 +632,9 @@ int flow_dealloc(int fd)
         irm_msg_t msg = IRM_MSG__INIT;
         irm_msg_t * recv_msg = NULL;
 
+        if (fd < 0)
+                return -EINVAL;
+
         msg.code         = IRM_MSG_CODE__IRM_FLOW_DEALLOC;
         msg.has_port_id  = true;
         msg.has_api      = true;
@@ -607,9 +644,10 @@ int flow_dealloc(int fd)
         pthread_rwlock_wrlock(&ai.flows_lock);
 
         if (ai.flows[fd].port_id < 0) {
+                bmp_release(ai.fds, fd);
                 pthread_rwlock_unlock(&ai.flows_lock);
                 pthread_rwlock_unlock(&ai.data_lock);
-                return -ENOTALLOC;
+                return 0;
         }
 
         msg.port_id = ai.flows[fd].port_id;
@@ -632,22 +670,7 @@ int flow_dealloc(int fd)
         pthread_rwlock_rdlock(&ai.data_lock);
         pthread_rwlock_wrlock(&ai.flows_lock);
 
-        port_destroy(&ai.ports[msg.port_id]);
-
-        ai.flows[fd].port_id = -1;
-        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;
-        ai.flows[fd].oflags = 0;
-        ai.flows[fd].api = -1;
-        shm_flow_set_close(ai.flows[fd].set);
-        if (ai.flows[fd].timeout != NULL) {
-                free(ai.flows[fd].timeout);
-                ai.flows[fd].timeout = NULL;
-        }
-
-        bmp_release(ai.fds, fd);
+        reset_flow(fd);
 
         pthread_rwlock_unlock(&ai.flows_lock);
         pthread_rwlock_unlock(&ai.data_lock);
@@ -865,7 +888,8 @@ struct fqueue * fqueue_create()
 void fqueue_destroy(struct fqueue * fq)
 {
         if (fq == NULL)
-                return
+                return;
+
         free(fq);
 }
 
@@ -1003,7 +1027,7 @@ int np1_flow_alloc(pid_t n_api, int 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);
+                reset_flow(fd);
                 pthread_rwlock_unlock(&ai.flows_lock);
                 pthread_rwlock_unlock(&ai.data_lock);
                 return -1;
@@ -1056,8 +1080,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_close(ai.flows[fd].rx_rb);
+                reset_flow(fd);
                 port_destroy(&ai.ports[port_id]);
                 pthread_rwlock_unlock(&ai.flows_lock);
                 pthread_rwlock_unlock(&ai.data_lock);
@@ -1066,10 +1089,7 @@ int np1_flow_resp(pid_t n_api, int 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_close(ai.flows[fd].rx_rb);
-                port_destroy(&ai.ports[port_id]);
+                reset_flow(fd);
                 pthread_rwlock_unlock(&ai.flows_lock);
                 pthread_rwlock_unlock(&ai.data_lock);
                 return -1;
@@ -1156,8 +1176,7 @@ int ipcp_flow_req_arr(pid_t  api, char * dst_name, char * src_ae_name)
 
         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]);
+                reset_flow(fd);
                 pthread_rwlock_unlock(&ai.flows_lock);
                 pthread_rwlock_unlock(&ai.data_lock);
                 return -1;
@@ -1209,6 +1228,7 @@ int ipcp_flow_alloc_reply(int fd, int response)
         ai.flows[fd].tx_rb = shm_rbuff_open(ai.flows[fd].api,
                                             ai.flows[fd].port_id);
         if (ai.flows[fd].tx_rb == NULL) {
+                reset_flow(fd);
                 pthread_rwlock_unlock(&ai.flows_lock);
                 pthread_rwlock_unlock(&ai.data_lock);
                 return -1;
@@ -1216,6 +1236,7 @@ int ipcp_flow_alloc_reply(int fd, int response)
 
         ai.flows[fd].set = shm_flow_set_open(ai.flows[fd].api);
         if (ai.flows[fd].set == NULL) {
+                reset_flow(fd);
                 pthread_rwlock_unlock(&ai.flows_lock);
                 pthread_rwlock_unlock(&ai.data_lock);
                 return -1;
diff --git a/src/tools/oping/oping_client.c b/src/tools/oping/oping_client.c
index 85cb2880..1ce96a75 100644
--- a/src/tools/oping/oping_client.c
+++ b/src/tools/oping/oping_client.c
@@ -162,6 +162,51 @@ void * writer(void * o)
         return (void *) 0;
 }
 
+static int client_init(void)
+{
+        client.flows = flow_set_create();
+        if (client.flows == NULL)
+                return -ENOMEM;
+
+        client.fq = fqueue_create();
+        if (client.fq == NULL) {
+                flow_set_destroy(client.flows);
+                return -ENOMEM;
+        }
+
+        client.times = malloc(sizeof(struct timespec) * client.count);
+        if (client.times == NULL) {
+                flow_set_destroy(client.flows);
+                fqueue_destroy(client.fq);
+                pthread_mutex_unlock(&client.lock);
+                return -ENOMEM;
+        }
+
+        client.sent = 0;
+        client.rcvd = 0;
+        client.rtt_min = FLT_MAX;
+        client.rtt_max = 0;
+        client.rtt_avg = 0;
+        client.rtt_m2 = 0;
+
+        pthread_mutex_init(&client.lock, NULL);
+        pthread_mutex_lock(&client.lock);
+
+        return 0;
+}
+
+void client_fini(void)
+{
+        if (client.flows != NULL)
+                flow_set_destroy(client.flows);
+
+        if (client.fq != NULL)
+                fqueue_destroy(client.fq);
+
+        if (client.times != NULL)
+                free(client.times);
+}
+
 int client_main(void)
 {
         struct sigaction sig_act;
@@ -171,18 +216,27 @@ int client_main(void)
 
         int fd;
 
-        client.flows = flow_set_create();
-        if (client.flows == NULL)
+        memset(&sig_act, 0, sizeof sig_act);
+        sig_act.sa_sigaction = &shutdown_client;
+        sig_act.sa_flags = 0;
+
+        if (sigaction(SIGINT,  &sig_act, NULL) ||
+            sigaction(SIGTERM, &sig_act, NULL) ||
+            sigaction(SIGHUP,  &sig_act, NULL) ||
+            sigaction(SIGPIPE, &sig_act, NULL)) {
+                printf("Failed to install sighandler.\n");
                 return -1;
+        }
 
-        client.fq = fqueue_create();
-        if (client.fq == NULL) {
-                flow_set_destroy(client.flows);
+        if (client_init()) {
+                printf("Failed to initialize client.\n");
                 return -1;
         }
 
         fd = flow_alloc(client.s_apn, NULL, NULL);
         if (fd < 0) {
+                flow_set_destroy(client.flows);
+                fqueue_destroy(client.fq);
                 printf("Failed to allocate flow.\n");
                 return -1;
         }
@@ -191,36 +245,12 @@ int client_main(void)
 
         if (flow_alloc_res(fd)) {
                 printf("Flow allocation refused.\n");
+                flow_set_del(client.flows, fd);
                 flow_dealloc(fd);
+                client_fini();
                 return -1;
         }
 
-        memset(&sig_act, 0, sizeof sig_act);
-        sig_act.sa_sigaction = &shutdown_client;
-        sig_act.sa_flags = 0;
-
-        if (sigaction(SIGINT,  &sig_act, NULL) ||
-            sigaction(SIGTERM, &sig_act, NULL) ||
-            sigaction(SIGHUP,  &sig_act, NULL) ||
-            sigaction(SIGPIPE, &sig_act, NULL)) {
-                printf("Failed to install sighandler.\n");
-                return -1;
-        }
-
-        pthread_mutex_init(&client.lock, NULL);
-        pthread_mutex_lock(&client.lock);
-        client.sent = 0;
-        client.rcvd = 0;
-        client.rtt_min = FLT_MAX;
-        client.rtt_max = 0;
-        client.rtt_avg = 0;
-        client.rtt_m2 = 0;
-        client.times = malloc(sizeof(struct timespec) * client.count);
-        if (client.times == NULL) {
-                pthread_mutex_unlock(&client.lock);
-                return -ENOMEM;
-        }
-
         pthread_mutex_unlock(&client.lock);
 
         clock_gettime(CLOCK_REALTIME, &tic);
@@ -253,14 +283,11 @@ int client_main(void)
                         printf("NaN ms\n");
         }
 
-        pthread_mutex_lock(&client.lock);
-        free(client.times);
-        flow_set_destroy(client.flows);
-        fqueue_destroy(client.fq);
-        pthread_mutex_unlock(&client.lock);
-        pthread_mutex_destroy(&client.lock);
+        flow_set_del(client.flows, fd);
 
         flow_dealloc(fd);
 
+        client_fini();
+
         return 0;
 }