3 files changed, 26 insertions, 17 deletions

diff --git a/src/tools/oping/oping.c b/src/tools/oping/oping.c
index b476b33a..801f79b5 100644
--- a/src/tools/oping/oping.c
+++ b/src/tools/oping/oping.c
@@ -54,6 +54,7 @@ struct c {
         double rtt_m2;
 
         flow_set_t * flows;
+        fqueue_t *   fq;
 
         /* needs locking */
         struct timespec * times;
@@ -66,6 +67,7 @@ struct c {
 struct s {
         struct timespec   times[OPING_MAX_FLOWS];
         flow_set_t *      flows;
+        fqueue_t *        fq;
         pthread_mutex_t   lock;
 
         pthread_t cleaner_pt;
diff --git a/src/tools/oping/oping_client.c b/src/tools/oping/oping_client.c
index 9f49a1df..85cb2880 100644
--- a/src/tools/oping/oping_client.c
+++ b/src/tools/oping/oping_client.c
@@ -64,16 +64,14 @@ void * reader(void * o)
         int msg_len = 0;
         double ms = 0;
         double d = 0;
-        fqueue_t * fq = fqueue_create();
-        if (fq == NULL)
-                return (void *) 1;
 
         (void) o;
 
         /* FIXME: use flow timeout option once we have it */
         while (client.rcvd != client.count
-               && flow_event_wait(client.flows, fq, &timeout) != -ETIMEDOUT) {
-                while ((fd = fqueue_next(fq)) >= 0) {
+               && (flow_event_wait(client.flows, client.fq, &timeout)
+                   != -ETIMEDOUT)) {
+                while ((fd = fqueue_next(client.fq)) >= 0) {
                         msg_len = flow_read(fd, buf, OPING_BUF_SIZE);
                         if (msg_len < 0)
                                 continue;
@@ -175,7 +173,13 @@ int client_main(void)
 
         client.flows = flow_set_create();
         if (client.flows == NULL)
-                return 0;
+                return -1;
+
+        client.fq = fqueue_create();
+        if (client.fq == NULL) {
+                flow_set_destroy(client.flows);
+                return -1;
+        }
 
         fd = flow_alloc(client.s_apn, NULL, NULL);
         if (fd < 0) {
@@ -251,6 +255,8 @@ int client_main(void)
 
         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);
 
diff --git a/src/tools/oping/oping_server.c b/src/tools/oping/oping_server.c
index bcd47f9a..720e71b6 100644
--- a/src/tools/oping/oping_server.c
+++ b/src/tools/oping/oping_server.c
@@ -75,23 +75,15 @@ void * server_thread(void *o)
         struct timespec now = {0, 0};
         struct timespec timeout = {0, 100 * MILLION};
         int fd;
-        fqueue_t * fq = fqueue_create();
-        if (fq == NULL)
-                return (void *) 1;
 
         (void) o;
 
         while (true) {
-                int ret = flow_event_wait(server.flows, fq, &timeout);
-                if (ret == -ETIMEDOUT)
+                if (flow_event_wait(server.flows, server.fq, &timeout)
+                    == -ETIMEDOUT)
                         continue;
 
-                if (ret < 0) {
-                        printf("Event error.\n");
-                        break;
-                }
-
-                while ((fd = fqueue_next(fq)) >= 0) {
+                while ((fd = fqueue_next(server.fq)) >= 0) {
                         msg_len = flow_read(fd, buf, OPING_BUF_SIZE);
                         if (msg_len < 0)
                                 continue;
@@ -176,6 +168,12 @@ int server_main(void)
         if (server.flows == NULL)
                 return 0;
 
+        server.fq = fqueue_create();
+        if (server.fq == NULL) {
+                flow_set_destroy(server.flows);
+                return -1;
+        }
+
         pthread_create(&server.cleaner_pt, NULL, cleaner_thread, NULL);
         pthread_create(&server.accept_pt, NULL, accept_thread, NULL);
         pthread_create(&server.server_pt, NULL, server_thread, NULL);
@@ -185,6 +183,9 @@ int server_main(void)
         pthread_cancel(server.server_pt);
         pthread_cancel(server.cleaner_pt);
 
+        flow_set_destroy(server.flows);
+        fqueue_destroy(server.fq);
+
         pthread_join(server.server_pt, NULL);
         pthread_join(server.cleaner_pt, NULL);