1 files changed, 67 insertions, 24 deletions

diff --git a/src/lib/dev.c b/src/lib/dev.c
index 1478d0bb..ab869509 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -140,10 +140,10 @@ struct {
         struct port *         ports;
         struct list_head      flow_list;
 
-        pthread_t             mon;
-
         pthread_t             tx;
+        pthread_t             rx;
         size_t                n_frcti;
+        fset_t *              frct_set;
 
         pthread_rwlock_t      lock;
 } ai;
@@ -252,7 +252,7 @@ static int proc_announce(char * prog)
 #include "crypt.c"
 #include "frct.c"
 
-void * frct_tx(void * o)
+void * flow_tx(void * o)
 {
         struct timespec tic = {0, TICTIME};
 
@@ -327,28 +327,52 @@ static void _flow_keepalive(struct flow * flow)
         }
 }
 
-void * monitor(void * o)
+static void handle_keepalives(void)
+{
+        struct list_head * p;
+        struct list_head * h;
+
+        pthread_rwlock_rdlock(&ai.lock);
+
+        list_for_each_safe(p, h, &ai.flow_list) {
+                struct flow * flow;
+                flow = list_entry(p, struct flow, next);
+                _flow_keepalive(flow);
+        }
+
+        pthread_rwlock_unlock(&ai.lock);
+}
+
+static void __cleanup_fqueue_destroy(void * fq)
+{
+        fqueue_destroy((fqueue_t *) fq);
+}
+
+void * flow_rx(void * o)
 {
         struct timespec tic = {0, TICTIME};
+        int             ret;
+        struct fqueue * fq;
 
         (void) o;
 
-        while (true) {
-                struct list_head * p;
-                struct list_head * h;
+        fq = fqueue_create();
 
-                pthread_rwlock_rdlock(&ai.lock);
+        pthread_cleanup_push(__cleanup_fqueue_destroy, fq);
 
-                list_for_each_safe(p, h, &ai.flow_list) {
-                        struct flow * flow  = list_entry(p, struct flow, next);
-                        _flow_keepalive(flow);
+        /* fevent will filter all FRCT packets for us */
+        while ((ret = fevent(ai.frct_set, fq, &tic)) != 0) {
+                if (ret == -ETIMEDOUT) {
+                        handle_keepalives();
+                        continue;
                 }
 
-                pthread_rwlock_unlock(&ai.lock);
-
-                nanosleep(&tic, NULL);
+                while (fqueue_next(fq) >= 0)
+                        ; /* no need to act */
         }
 
+        pthread_cleanup_pop(true);
+
         return (void *) 0;
 }
 
@@ -371,6 +395,8 @@ static void flow_fini(int fd)
                         pthread_join(ai.tx, NULL);
                 }
 
+                shm_flow_set_del(ai.fqset, 0, ai.flows[fd].flow_id);
+
                 frcti_destroy(ai.flows[fd].frcti);
         }
 
@@ -462,9 +488,12 @@ static int flow_init(int       flow_id,
                 if (flow->frcti == NULL)
                         goto fail_frcti;
 
+                if (shm_flow_set_add(ai.fqset, 0, flow_id))
+                        goto fail_flow_set_add;
+
                 ++ai.n_frcti;
                 if (ai.n_frcti == 1 &&
-                    pthread_create(&ai.tx, NULL, frct_tx, NULL) < 0)
+                    pthread_create(&ai.tx, NULL, flow_tx, NULL) < 0)
                         goto fail_tx_thread;
         }
 
@@ -479,6 +508,8 @@ static int flow_init(int       flow_id,
         return fd;
 
  fail_tx_thread:
+        shm_flow_set_del(ai.fqset, 0, flow_id);
+ fail_flow_set_add:
         frcti_destroy(flow->frcti);
  fail_frcti:
         crypt_fini(flow->ctx);
@@ -582,6 +613,10 @@ static void init(int     argc,
         if (ai.fqset == NULL)
                 goto fail_fqset;
 
+        ai.frct_set = fset_create();
+        if (ai.frct_set == NULL || ai.frct_set->idx != 0)
+                goto fail_frct_set;
+
         if (timerwheel_init() < 0)
                 goto fail_timerwheel;
 
@@ -593,7 +628,7 @@ static void init(int     argc,
                         goto fail_rib_init;
         }
 #endif
-        if (pthread_create(&ai.mon, NULL, monitor, NULL) < 0)
+        if (pthread_create(&ai.rx, NULL, flow_rx, NULL) < 0)
                 goto fail_monitor;
 
         return;
@@ -605,6 +640,8 @@ static void init(int     argc,
 #endif
         timerwheel_fini();
  fail_timerwheel:
+        fset_destroy(ai.frct_set);
+ fail_frct_set:
         shm_flow_set_close(ai.fqset);
  fail_fqset:
         pthread_rwlock_destroy(&ai.lock);
@@ -640,8 +677,10 @@ static void fini(void)
         if (ai.fds == NULL)
                 return;
 
-        pthread_cancel(ai.mon);
-        pthread_join(ai.mon, NULL);
+        pthread_cancel(ai.rx);
+        pthread_join(ai.rx, NULL);
+
+        fset_destroy(ai.frct_set);
 
         pthread_rwlock_wrlock(&ai.lock);
 
@@ -1510,6 +1549,9 @@ int fset_add(struct flow_set * set,
                 goto fail;
         }
 
+        if (flow->frcti != NULL)
+                shm_flow_set_del(ai.fqset, 0, ai.flows[fd].flow_id);
+
         ret = shm_flow_set_add(ai.fqset, set->idx, ai.flows[fd].flow_id);
         if (ret < 0)
                 goto fail;
@@ -1541,6 +1583,9 @@ void fset_del(struct flow_set * set,
         if (flow->flow_id >= 0)
                 shm_flow_set_del(ai.fqset, set->idx, flow->flow_id);
 
+        if (flow->frcti != NULL)
+                shm_flow_set_add(ai.fqset, 0, ai.flows[fd].flow_id);
+
         pthread_rwlock_unlock(&ai.lock);
 }
 
@@ -1628,12 +1673,10 @@ int fqueue_next(struct fqueue * fq)
         if (fq->fqsize == 0 || fq->next == fq->fqsize)
                 return -EPERM;
 
-        pthread_rwlock_rdlock(&ai.lock);
-
-        if (fq->next != 0 && fqueue_filter(fq) == 0) {
-                pthread_rwlock_unlock(&ai.lock);
+        if (fq->next != 0 && fqueue_filter(fq) == 0)
                 return -EPERM;
-        }
+
+        pthread_rwlock_rdlock(&ai.lock);
 
         fd = ai.ports[fq->fqueue[fq->next]].fd;
 
@@ -1687,7 +1730,7 @@ ssize_t fevent(struct flow_set *       set,
                 ret = fqueue_filter(fq);
         }
 
-        assert(ret);
+        assert(ret != 0);
 
         return 1;
 }