Merged in dstaesse/ouroboros/be-fqueue (pull request #574)

dev: Revise fqueue API and add man pages

4 files changed, 35 insertions, 38 deletions

diff --git a/src/ipcpd/local/main.c b/src/ipcpd/local/main.c
index c6f88d78..aab82d25 100644
--- a/src/ipcpd/local/main.c
+++ b/src/ipcpd/local/main.c
@@ -51,7 +51,7 @@ struct {
         struct shim_data * shim_data;
 
         int                in_out[SYS_MAX_FLOWS];
-        flow_set_t *       flows;
+        fset_t *           flows;
         fqueue_t *         fq;
 
         pthread_rwlock_t   lock;
@@ -64,20 +64,20 @@ static int local_data_init(void)
         for (i = 0; i < SYS_MAX_FLOWS; ++i)
                 local_data.in_out[i] = -1;
 
-        local_data.flows = flow_set_create();
+        local_data.flows = fset_create();
         if (local_data.flows == NULL)
                 return -ENFILE;
 
         local_data.fq = fqueue_create();
         if (local_data.fq == NULL) {
-                flow_set_destroy(local_data.flows);
+                fset_destroy(local_data.flows);
                 return -ENOMEM;
         }
 
         local_data.shim_data = shim_data_create();
         if (local_data.shim_data == NULL) {
                 fqueue_destroy(local_data.fq);
-                flow_set_destroy(local_data.flows);
+                fset_destroy(local_data.flows);
                 return -ENOMEM;
         }
 
@@ -88,7 +88,7 @@ static int local_data_init(void)
 
 static void local_data_fini(void){
         shim_data_destroy(local_data.shim_data);
-        flow_set_destroy(local_data.flows);
+        fset_destroy(local_data.flows);
         fqueue_destroy(local_data.fq);
         pthread_rwlock_destroy(&local_data.lock);
 }
@@ -106,7 +106,7 @@ static void * ipcp_local_sdu_loop(void * o)
                 if (ipcp_get_state() != IPCP_OPERATIONAL)
                         return (void *) 1; /* -ENOTENROLLED */
 
-                flow_event_wait(local_data.flows, local_data.fq, &timeout);
+                fevent(local_data.flows, local_data.fq, &timeout);
 
                 while ((fd = fqueue_next(local_data.fq)) >= 0) {
                         pthread_rwlock_rdlock(&local_data.lock);
@@ -236,7 +236,7 @@ static int ipcp_local_flow_alloc(int             fd,
 
         pthread_mutex_unlock(&ipcpi.alloc_lock);
 
-        flow_set_add(local_data.flows, fd);
+        fset_add(local_data.flows, fd);
 
         log_info("Pending local allocation request on fd %d.", fd);
 
@@ -290,7 +290,7 @@ static int ipcp_local_flow_alloc_resp(int fd,
 
         pthread_rwlock_unlock(&local_data.lock);
 
-        flow_set_add(local_data.flows, fd);
+        fset_add(local_data.flows, fd);
 
         if ((ret = ipcp_flow_alloc_reply(out_fd, response)) < 0)
                 return -1;
@@ -308,7 +308,7 @@ static int ipcp_local_flow_dealloc(int fd)
 
         pthread_rwlock_wrlock(&local_data.lock);
 
-        flow_set_del(local_data.flows, fd);
+        fset_del(local_data.flows, fd);
 
         local_data.in_out[fd] = -1;
 
diff --git a/src/ipcpd/normal/sdu_sched.c b/src/ipcpd/normal/sdu_sched.c
index 10b0f02f..c7e799e2 100644
--- a/src/ipcpd/normal/sdu_sched.c
+++ b/src/ipcpd/normal/sdu_sched.c
@@ -38,9 +38,9 @@
 #define FD_UPDATE_TIMEOUT 10000 /* nanoseconds */
 
 struct sdu_sched {
-        flow_set_t * set[QOS_CUBE_MAX];
-        next_sdu_t   callback;
-        pthread_t    sdu_readers[IPCP_SCHED_THREADS];
+        fset_t *   set[QOS_CUBE_MAX];
+        next_sdu_t callback;
+        pthread_t  sdu_readers[IPCP_SCHED_THREADS];
 };
 
 static void cleanup_reader(void * o)
@@ -80,7 +80,7 @@ static void * sdu_reader(void * o)
                 /* FIXME: replace with scheduling policy call */
                 i = (i + 1) % QOS_CUBE_MAX;
 
-                ret = flow_event_wait(sched->set[i], fqs[i], &timeout);
+                ret = fevent(sched->set[i], fqs[i], &timeout);
                 if (ret == -ETIMEDOUT)
                         continue;
 
@@ -122,10 +122,10 @@ struct sdu_sched * sdu_sched_create(next_sdu_t callback)
         sdu_sched->callback = callback;
 
         for (i = 0; i < QOS_CUBE_MAX; ++i) {
-                sdu_sched->set[i] = flow_set_create();
+                sdu_sched->set[i] = fset_create();
                 if (sdu_sched->set[i] == NULL) {
                         for (j = 0; j < i; ++j)
-                                flow_set_destroy(sdu_sched->set[j]);
+                                fset_destroy(sdu_sched->set[j]);
                         goto fail_flow_set;
                 }
         }
@@ -162,7 +162,7 @@ void sdu_sched_destroy(struct sdu_sched * sdu_sched)
         }
 
         for (i = 0; i < QOS_CUBE_MAX; ++i)
-                flow_set_destroy(sdu_sched->set[i]);
+                fset_destroy(sdu_sched->set[i]);
 
         free(sdu_sched);
 }
@@ -175,7 +175,7 @@ void sdu_sched_add(struct sdu_sched * sdu_sched,
         assert(sdu_sched);
 
         ipcp_flow_get_qoscube(fd, &qc);
-        flow_set_add(sdu_sched->set[qc], fd);
+        fset_add(sdu_sched->set[qc], fd);
 }
 
 void sdu_sched_del(struct sdu_sched * sdu_sched,
@@ -186,5 +186,5 @@ void sdu_sched_del(struct sdu_sched * sdu_sched,
         assert(sdu_sched);
 
         ipcp_flow_get_qoscube(fd, &qc);
-        flow_set_del(sdu_sched->set[qc], fd);
+        fset_del(sdu_sched->set[qc], fd);
 }
diff --git a/src/ipcpd/shim-eth-llc/main.c b/src/ipcpd/shim-eth-llc/main.c
index e5f0cba8..be1e6423 100644
--- a/src/ipcpd/shim-eth-llc/main.c
+++ b/src/ipcpd/shim-eth-llc/main.c
@@ -143,7 +143,7 @@ struct {
 #endif /* HAVE_NETMAP */
 
         struct bmp *       saps;
-        flow_set_t *       np1_flows;
+        fset_t *           np1_flows;
         fqueue_t *         fq;
         int *              ef_to_fd;
         struct ef *        fd_to_ef;
@@ -180,7 +180,7 @@ static int eth_llc_data_init(void)
         if (eth_llc_data.saps == NULL)
                 goto fail_saps;
 
-        eth_llc_data.np1_flows = flow_set_create();
+        eth_llc_data.np1_flows = fset_create();
         if (eth_llc_data.np1_flows == NULL)
                 goto fail_np1_flows;
 
@@ -236,7 +236,7 @@ static int eth_llc_data_init(void)
  fail_shim_data:
         fqueue_destroy(eth_llc_data.fq);
  fail_fq:
-        flow_set_destroy(eth_llc_data.np1_flows);
+        fset_destroy(eth_llc_data.np1_flows);
  fail_np1_flows:
         bmp_destroy(eth_llc_data.saps);
  fail_saps:
@@ -261,7 +261,7 @@ void eth_llc_data_fini(void)
         pthread_rwlock_destroy(&eth_llc_data.flows_lock);
         shim_data_destroy(eth_llc_data.shim_data);
         fqueue_destroy(eth_llc_data.fq);
-        flow_set_destroy(eth_llc_data.np1_flows);
+        fset_destroy(eth_llc_data.np1_flows);
         bmp_destroy(eth_llc_data.saps);
         free(eth_llc_data.fd_to_ef);
         free(eth_llc_data.ef_to_fd);
@@ -751,10 +751,7 @@ static void * eth_llc_ipcp_sdu_writer(void * o)
 
         (void) o;
 
-        while (flow_event_wait(eth_llc_data.np1_flows,
-                               eth_llc_data.fq,
-                               &timeout)) {
-
+        while (fevent(eth_llc_data.np1_flows, eth_llc_data.fq, &timeout)) {
                 if (ipcp_get_state() != IPCP_OPERATIONAL)
                         return (void *) 0;
 
@@ -1120,7 +1117,7 @@ static int eth_llc_ipcp_flow_alloc(int             fd,
                 return -1;
         }
 
-        flow_set_add(eth_llc_data.np1_flows, fd);
+        fset_add(eth_llc_data.np1_flows, fd);
 
         log_dbg("Pending flow with fd %d on SAP %d.", fd, ssap);
 
@@ -1179,7 +1176,7 @@ static int eth_llc_ipcp_flow_alloc_resp(int fd,
                 return -1;
         }
 
-        flow_set_add(eth_llc_data.np1_flows, fd);
+        fset_add(eth_llc_data.np1_flows, fd);
 
         log_dbg("Accepted flow, fd %d, SAP %d.", fd, (uint8_t)ssap);
 
@@ -1195,7 +1192,7 @@ static int eth_llc_ipcp_flow_dealloc(int fd)
 
         pthread_rwlock_wrlock(&eth_llc_data.flows_lock);
 
-        flow_set_del(eth_llc_data.np1_flows, fd);
+        fset_del(eth_llc_data.np1_flows, fd);
 
         sap = eth_llc_data.fd_to_ef[fd].sap;
         memcpy(addr, eth_llc_data.fd_to_ef[fd].r_addr, MAC_SIZE);
diff --git a/src/ipcpd/shim-udp/main.c b/src/ipcpd/shim-udp/main.c
index b6f516bd..a94472b2 100644
--- a/src/ipcpd/shim-udp/main.c
+++ b/src/ipcpd/shim-udp/main.c
@@ -79,7 +79,7 @@ struct {
         struct sockaddr_in s_saddr;
         int                s_fd;
 
-        flow_set_t *       np1_flows;
+        fset_t *           np1_flows;
         fqueue_t *         fq;
         fd_set             flow_fd_s;
         /* bidir mappings of (n - 1) file descriptor to (n) flow descriptor */
@@ -110,20 +110,20 @@ static int udp_data_init(void)
 
         FD_ZERO(&udp_data.flow_fd_s);
 
-        udp_data.np1_flows = flow_set_create();
+        udp_data.np1_flows = fset_create();
         if (udp_data.np1_flows == NULL)
                 return -ENOMEM;
 
         udp_data.fq = fqueue_create();
         if (udp_data.fq == NULL) {
-                flow_set_destroy(udp_data.np1_flows);
+                fset_destroy(udp_data.np1_flows);
                 return -ENOMEM;
         }
 
         udp_data.shim_data = shim_data_create();
         if (udp_data.shim_data == NULL) {
                 fqueue_destroy(udp_data.fq);
-                flow_set_destroy(udp_data.np1_flows);
+                fset_destroy(udp_data.np1_flows);
                 return -ENOMEM;
         }
 
@@ -136,7 +136,7 @@ static int udp_data_init(void)
 
 static void udp_data_fini(void)
 {
-        flow_set_destroy(udp_data.np1_flows);
+        fset_destroy(udp_data.np1_flows);
         fqueue_destroy(udp_data.fq);
 
         shim_data_destroy(udp_data.shim_data);
@@ -518,7 +518,7 @@ static void * ipcp_udp_sdu_loop(void * o)
         (void) o;
 
         while (ipcp_get_state() == IPCP_OPERATIONAL) {
-                flow_event_wait(udp_data.np1_flows, udp_data.fq, &timeout);
+                fevent(udp_data.np1_flows, udp_data.fq, &timeout);
                 while ((fd = fqueue_next(udp_data.fq)) >= 0) {
                         if (ipcp_flow_read(fd, &sdb)) {
                                 log_err("Bad read from fd %d.", fd);
@@ -962,7 +962,7 @@ static int ipcp_udp_flow_alloc(int             fd,
         udp_data.fd_to_uf[fd].skfd = skfd;
         udp_data.uf_to_fd[skfd]    = fd;
 
-        flow_set_add(udp_data.np1_flows, fd);
+        fset_add(udp_data.np1_flows, fd);
 
         pthread_rwlock_unlock(&udp_data.flows_lock);
 
@@ -1038,7 +1038,7 @@ static int ipcp_udp_flow_alloc_resp(int fd,
 
         set_fd(skfd);
 
-        flow_set_add(udp_data.np1_flows, fd);
+        fset_add(udp_data.np1_flows, fd);
 
         pthread_rwlock_unlock(&udp_data.flows_lock);
 
@@ -1064,7 +1064,7 @@ static int ipcp_udp_flow_dealloc(int fd)
 
         pthread_rwlock_wrlock(&udp_data.flows_lock);
 
-        flow_set_del(udp_data.np1_flows, fd);
+        fset_del(udp_data.np1_flows, fd);
 
         skfd = udp_data.fd_to_uf[fd].skfd;