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

lib: Revise flow allocation API

2 files changed, 19 insertions, 29 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;