Merged in dstaesse/ouroboros/be-new-api (pull request #439)

lib: Revise flow allocation API

8 files changed, 36 insertions, 86 deletions

diff --git a/src/tools/cbr/cbr_client.c b/src/tools/cbr/cbr_client.c
index 16ade13d..5ec1d560 100644
--- a/src/tools/cbr/cbr_client.c
+++ b/src/tools/cbr/cbr_client.c
@@ -63,7 +63,6 @@ int client_main(char * server,
         struct sigaction sig_act;
 
         int fd = 0;
-        int result = 0;
         char buf[size];
         long seqnr = 0;
         long gap = size * 8.0 * (BILLION / (double) rate);
@@ -90,19 +89,12 @@ int client_main(char * server,
         printf("Client started, duration %d, rate %lu b/s, size %d B.\n",
                duration, rate, size);
 
-        fd = flow_alloc(server, NULL);
+        fd = flow_alloc(server, NULL, NULL);
         if (fd < 0) {
                 printf("Failed to allocate flow.\n");
                 return -1;
         }
 
-        result = flow_alloc_res(fd);
-        if (result < 0) {
-                printf("Flow allocation refused.\n");
-                flow_dealloc(fd);
-                return -1;
-        }
-
         clock_gettime(CLOCK_REALTIME, &start);
         if (!flood) {
                 while (!stop) {
diff --git a/src/tools/cbr/cbr_server.c b/src/tools/cbr/cbr_server.c
index 9198858c..1a963a64 100644
--- a/src/tools/cbr/cbr_server.c
+++ b/src/tools/cbr/cbr_server.c
@@ -146,6 +146,8 @@ static void * worker(void * o)
 
                 pthread_mutex_lock(&fds_lock);
                 fds_count--;
+
+                pthread_cond_signal(&fds_signal);
                 pthread_mutex_unlock(&fds_lock);
         }
 
@@ -154,8 +156,7 @@ static void * worker(void * o)
 
 static void * listener(void * o)
 {
-        int client_fd = 0;
-        int response = 0;
+        int fd = 0;
         qosspec_t qs;
 
         (void) o;
@@ -164,8 +165,19 @@ static void * listener(void * o)
                server_settings.interval, server_settings.timeout);
 
         while (true) {
-                client_fd = flow_accept(&qs);
-                if (client_fd < 0) {
+                pthread_mutex_lock(&fds_lock);
+                pthread_cleanup_push((void(*)(void *)) pthread_mutex_unlock,
+                                     (void *) &fds_lock);
+
+                while (fds_count == THREADS_SIZE) {
+                        printf("Can't accept any more flows, waiting.\n");
+                        pthread_cond_wait(&fds_signal, &fds_lock);
+                }
+
+                pthread_cleanup_pop(true);
+
+                fd = flow_accept(&qs, NULL);
+                if (fd < 0) {
                         printf("Failed to accept flow.\n");
                         break;
                 }
@@ -174,26 +186,12 @@ static void * listener(void * o)
 
                 pthread_mutex_lock(&fds_lock);
 
-                response = (fds_count < THREADS_SIZE) ? 0 : -1;
-
-                if (flow_alloc_resp(client_fd, response)) {
-                        printf("Failed to give an allocate response.\n");
-                        flow_dealloc(client_fd);
-                        pthread_mutex_unlock(&fds_lock);
-                        continue;
-                }
-
-                if (response) {
-                        printf("Can't accept any more flows, denying.\n");
-                        continue;
-                }
-
                 fds_count++;
                 fds_index = (fds_index + 1) % THREADS_SIZE;
-                fds[fds_index] = client_fd;
+                fds[fds_index] = fd;
 
-                pthread_mutex_unlock(&fds_lock);
                 pthread_cond_signal(&fds_signal);
+                pthread_mutex_unlock(&fds_lock);
         }
 
         return 0;
diff --git a/src/tools/echo/echo_client.c b/src/tools/echo/echo_client.c
index f84de73a..5ec2051f 100644
--- a/src/tools/echo/echo_client.c
+++ b/src/tools/echo/echo_client.c
@@ -26,25 +26,17 @@
 int client_main(void)
 {
         int fd = 0;
-        int result = 0;
         char buf[BUF_SIZE];
         char * message  = "Client says hi!";
         ssize_t count = 0;
 
-        fd = flow_alloc("echo", NULL);
+        fd = flow_alloc("echo", NULL, NULL);
         if (fd < 0) {
                 printf("Failed to allocate flow.\n");
                 return -1;
         }
 
-        result = flow_alloc_res(fd);
-        if (result < 0) {
-                printf("Flow allocation refused.\n");
-                flow_dealloc(fd);
-                return -1;
-        }
-
-        if (flow_write(fd, message, strlen(message) + 1) == -1) {
+        if (flow_write(fd, message, strlen(message) + 1) < 0) {
                 printf("Failed to write SDU.\n");
                 flow_dealloc(fd);
                 return -1;
diff --git a/src/tools/echo/echo_server.c b/src/tools/echo/echo_server.c
index aa136485..771155f4 100644
--- a/src/tools/echo/echo_server.c
+++ b/src/tools/echo/echo_server.c
@@ -37,7 +37,7 @@ void shutdown_server(int signo)
 
 int server_main(void)
 {
-        int    client_fd = 0;
+        int    fd = 0;
         char   buf[BUF_SIZE];
         ssize_t count = 0;
         qosspec_t qs;
@@ -51,36 +51,30 @@ int server_main(void)
         }
 
         while (true) {
-                client_fd = flow_accept(&qs);
-                if (client_fd < 0) {
+                fd = flow_accept(&qs, NULL);
+                if (fd < 0) {
                         printf("Failed to accept flow.\n");
                         break;
                 }
 
                 printf("New flow.\n");
 
-                if (flow_alloc_resp(client_fd, 0)) {
-                        printf("Failed to give an allocate response.\n");
-                        flow_dealloc(client_fd);
-                        continue;
-                }
-
-                count = flow_read(client_fd, &buf, BUF_SIZE);
+                count = flow_read(fd, &buf, BUF_SIZE);
                 if (count < 0) {
                         printf("Failed to read SDU.\n");
-                        flow_dealloc(client_fd);
+                        flow_dealloc(fd);
                         continue;
                 }
 
                 printf("Message from client is %.*s.\n", (int) count, buf);
 
-                if (flow_write(client_fd, buf, count) == -1) {
+                if (flow_write(fd, buf, count) == -1) {
                         printf("Failed to write SDU.\n");
-                        flow_dealloc(client_fd);
+                        flow_dealloc(fd);
                         continue;
                 }
 
-                flow_dealloc(client_fd);
+                flow_dealloc(fd);
         }
 
         return 0;
diff --git a/src/tools/operf/operf_client.c b/src/tools/operf/operf_client.c
index d2f08ef4..7827b62b 100644
--- a/src/tools/operf/operf_client.c
+++ b/src/tools/operf/operf_client.c
@@ -182,18 +182,12 @@ int client_main(void)
         client.sent = 0;
         client.rcvd = 0;
 
-        fd = flow_alloc(client.s_apn, NULL);
+        fd = flow_alloc(client.s_apn, NULL, NULL);
         if (fd < 0) {
                 printf("Failed to allocate flow.\n");
                 return -1;
         }
 
-        if (flow_alloc_res(fd)) {
-                printf("Flow allocation refused.\n");
-                flow_dealloc(fd);
-                return -1;
-        }
-
         clock_gettime(CLOCK_REALTIME, &tic);
 
         pthread_create(&client.reader_pt, NULL, reader, &fd);
diff --git a/src/tools/operf/operf_server.c b/src/tools/operf/operf_server.c
index 3665d4cc..b17a4f7b 100644
--- a/src/tools/operf/operf_server.c
+++ b/src/tools/operf/operf_server.c
@@ -108,7 +108,7 @@ void * accept_thread(void * o)
         printf("Ouroboros perf server started.\n");
 
         while (true) {
-                fd = flow_accept(&qs);
+                fd = flow_accept(&qs, NULL);
                 if (fd < 0) {
                         printf("Failed to accept flow.\n");
                         break;
@@ -116,12 +116,6 @@ void * accept_thread(void * o)
 
                 printf("New flow %d.\n", fd);
 
-                if (flow_alloc_resp(fd, 0)) {
-                        printf("Failed to give an allocate response.\n");
-                        flow_dealloc(fd);
-                        continue;
-                }
-
                 clock_gettime(CLOCK_REALTIME, &now);
 
                 pthread_mutex_lock(&server.lock);
diff --git a/src/tools/oping/oping_client.c b/src/tools/oping/oping_client.c
index a91a126c..77a08db7 100644
--- a/src/tools/oping/oping_client.c
+++ b/src/tools/oping/oping_client.c
@@ -176,7 +176,6 @@ static int client_init(void)
         client.rtt_m2 = 0;
 
         pthread_mutex_init(&client.lock, NULL);
-        pthread_mutex_lock(&client.lock);
 
         return 0;
 }
@@ -213,21 +212,13 @@ int client_main(void)
                 return -1;
         }
 
-        fd = flow_alloc(client.s_apn, NULL);
+        fd = flow_alloc(client.s_apn, NULL, NULL);
         if (fd < 0) {
                 printf("Failed to allocate flow.\n");
-                return -1;
-        }
-
-        if (flow_alloc_res(fd)) {
-                printf("Flow allocation refused.\n");
-                flow_dealloc(fd);
                 client_fini();
                 return -1;
         }
 
-        pthread_mutex_unlock(&client.lock);
-
         clock_gettime(CLOCK_REALTIME, &tic);
 
         pthread_create(&client.reader_pt, NULL, reader, &fd);
diff --git a/src/tools/oping/oping_server.c b/src/tools/oping/oping_server.c
index e20e236d..44a301ba 100644
--- a/src/tools/oping/oping_server.c
+++ b/src/tools/oping/oping_server.c
@@ -57,6 +57,7 @@ void * cleaner_thread(void * o)
                 for (i = 0; i < OPING_MAX_FLOWS; ++i)
                         if (flow_set_has(server.flows, i) &&
                             ts_diff_ms(&server.times[i], &now) > deadline_ms) {
+                                printf("Flow %d timed out.\n", i);
                                 flow_set_del(server.flows, i);
                                 flow_dealloc(i);
                         }
@@ -110,8 +111,8 @@ void * server_thread(void *o)
 
 void * accept_thread(void * o)
 {
-        int fd = 0;
-        struct timespec now = {0, 0};
+        int fd;
+        struct timespec now;
         qosspec_t qs;
 
         (void) o;
@@ -119,7 +120,7 @@ void * accept_thread(void * o)
         printf("Ouroboros ping server started.\n");
 
         while (true) {
-                fd = flow_accept(&qs);
+                fd = flow_accept(&qs, NULL);
                 if (fd < 0) {
                         printf("Failed to accept flow.\n");
                         break;
@@ -127,12 +128,6 @@ void * accept_thread(void * o)
 
                 printf("New flow %d.\n", fd);
 
-                if (flow_alloc_resp(fd, 0)) {
-                        printf("Failed to give an allocate response.\n");
-                        flow_dealloc(fd);
-                        continue;
-                }
-
                 clock_gettime(CLOCK_REALTIME, &now);
 
                 pthread_mutex_lock(&server.lock);