lib: dev: Provide a set of fds to flow_select

The flow_select call now takes as a parameter a flow_set_t, which
specifies a set of flow descriptors that will unblock the select call
when an SDU for one of them arrives. The select call has been moved to
its own header.

3 files changed, 23 insertions, 14 deletions

diff --git a/src/tools/oping/oping.c b/src/tools/oping/oping.c
index 2871e79..7d2edf3 100644
--- a/src/tools/oping/oping.c
+++ b/src/tools/oping/oping.c
@@ -23,6 +23,9 @@
 
 #define _POSIX_C_SOURCE 199506L
 
+#include <ouroboros/select.h>
+#include <ouroboros/dev.h>
+
 #include <stdio.h>
 #include <string.h>
 #include <pthread.h>
@@ -59,9 +62,9 @@ struct c {
 } client;
 
 struct s {
-        struct timespec times[OPING_MAX_FLOWS];
-        bool            flows[OPING_MAX_FLOWS];
-        pthread_mutex_t lock;
+        struct timespec   times[OPING_MAX_FLOWS];
+        struct flow_set * flows;
+        pthread_mutex_t   lock;
 
         pthread_t cleaner_pt;
         pthread_t accept_pt;
diff --git a/src/tools/oping/oping_client.c b/src/tools/oping/oping_client.c
index 6e1fbc5..3a25498 100644
--- a/src/tools/oping/oping_client.c
+++ b/src/tools/oping/oping_client.c
@@ -65,7 +65,7 @@ void * reader(void * o)
 
         /* FIXME: use flow timeout option once we have it */
         while(client.rcvd != client.count &&
-              (fd = flow_select(&timeout)) != -ETIMEDOUT) {
+              (fd = flow_select(NULL, &timeout)) != -ETIMEDOUT) {
                 flow_cntl(fd, FLOW_F_SETFL, FLOW_O_NONBLOCK);
                 while (!((msg_len = flow_read(fd, buf, OPING_BUF_SIZE)) < 0)) {
                         if (msg_len < 0)
diff --git a/src/tools/oping/oping_server.c b/src/tools/oping/oping_server.c
index 7761110..9c7b1be 100644
--- a/src/tools/oping/oping_server.c
+++ b/src/tools/oping/oping_server.c
@@ -21,8 +21,6 @@
  * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  */
 
-#include <ouroboros/dev.h>
-
 #ifdef __FreeBSD__
 #define __XSI_VISIBLE 500
 #endif
@@ -53,9 +51,9 @@ void * cleaner_thread(void * o)
                 clock_gettime(CLOCK_REALTIME, &now);
                 pthread_mutex_lock(&server.lock);
                 for (i = 0; i < OPING_MAX_FLOWS; ++i)
-                        if (server.flows[i] &&
+                        if (flow_set_has(server.flows, i) &&
                             ts_diff_ms(&server.times[i], &now) > deadline_ms) {
-                                server.flows[i] = false;
+                                flow_set_del(server.flows, i);
                                 flow_dealloc(i);
                         }
 
@@ -70,10 +68,16 @@ void * server_thread(void *o)
         int msg_len = 0;
         struct oping_msg * msg = (struct oping_msg *) buf;
         struct timespec now = {0, 0};
+        struct timespec timeout = {0, 100 * MILLION};
 
         while (true) {
-
-                int fd = flow_select(NULL);
+                int fd = flow_select(server.flows, &timeout);
+                if (fd == -ETIMEDOUT)
+                        continue;
+                if (fd < 0) {
+                        printf("Failed to get active fd.\n");
+                        continue;
+                }
                 while (!((msg_len = flow_read(fd, buf, OPING_BUF_SIZE)) < 0)) {
                         if (msg_len < 0)
                                 continue;
@@ -126,7 +130,7 @@ void * accept_thread(void * o)
                 clock_gettime(CLOCK_REALTIME, &now);
 
                 pthread_mutex_lock(&server.lock);
-                server.flows[fd] = true;
+                flow_set_add(server.flows, fd);
                 server.times[fd] = now;
                 pthread_mutex_unlock(&server.lock);
 
@@ -139,7 +143,6 @@ void * accept_thread(void * o)
 int server_main()
 {
         struct sigaction sig_act;
-        int i = 0;
 
         memset(&sig_act, 0, sizeof sig_act);
         sig_act.sa_sigaction = &shutdown_server;
@@ -153,8 +156,11 @@ int server_main()
                 return -1;
         }
 
-        for (i = 0; i < OPING_MAX_FLOWS; ++i)
-                server.flows[i] = false;
+        server.flows = flow_set_create();
+        if (server.flows == NULL)
+                return 0;
+
+        flow_set_zero(server.flows);
 
         pthread_create(&server.cleaner_pt, NULL, cleaner_thread, NULL);
         pthread_create(&server.accept_pt, NULL, accept_thread, NULL);