dev: Revise fqueue API and add man pages

25 files changed, 339 insertions, 110 deletions

diff --git a/doc/man/CMakeLists.txt b/doc/man/CMakeLists.txt
index eadd385e..280252ed 100644
--- a/doc/man/CMakeLists.txt
+++ b/doc/man/CMakeLists.txt
@@ -7,6 +7,18 @@ set(MAN_NAMES
   flow_dealloc.3
   flow_read.3
   flow_write.3
+  fqueue.3
+  fqueue_create.3
+  fqueue_destroy.3
+  fqueue_next.3
+  fevent.3
+  fset.3
+  fset_create.3
+  fset_destroy.3
+  fset_zero.3
+  fset_add.3
+  fset_del.3
+  fset_has.3
   )
 
 macro(INSTALL_MAN __mans)
diff --git a/doc/man/fevent.3 b/doc/man/fevent.3
new file mode 100644
index 00000000..0e1f5f07
--- /dev/null
+++ b/doc/man/fevent.3
@@ -0,0 +1 @@
+.so fqueue.3
diff --git a/doc/man/flow_alloc.3 b/doc/man/flow_alloc.3
index dd0fe7ea..662c46ec 100644
--- a/doc/man/flow_alloc.3
+++ b/doc/man/flow_alloc.3
@@ -31,7 +31,7 @@ to support Inter-Process Communication (IPC). Such a collection of
 allocated system and network resources is referred to as a flow. A
 flow has a certain Quality of Service (QoS) associated with it.
 
-The \fB flow_accept\fR() function blocks the calling thread waiting
+The \fBflow_accept\fR() function blocks the calling thread waiting
 for an incoming request to allocate a flow. If \fBqosspec_t * \fIqs\fR
 is not NULL, the value of \fIqs\fR will be updated to reflect the
 actual QoS provided by the IPC facility for the accepted flow. Which
diff --git a/doc/man/fqueue.3 b/doc/man/fqueue.3
new file mode 100644
index 00000000..611d54c7
--- /dev/null
+++ b/doc/man/fqueue.3
@@ -0,0 +1,104 @@
+.\" Ouroboros man pages (C) 2017
+.\" Dimitri Staessens <dimitri.staessens@ugent.be>
+.\" Sander Vrijders <sander.vrijders@ugent.be>
+
+.TH FQUEUE 3 2017-08-29 GNU "Ouroboros Programmer's Manual"
+
+.SH NAME
+
+fqueue_create, fqueue_destroy, fqueue_next, fevent \- I/O multiplexing
+on flows
+
+.SH SYNOPSIS
+
+.B #include <ouroboros/fqueue.h>
+
+\fBfqueue_t * fqueue_create(void);\fR
+
+\fBvoid fqueue_destroy(fqueue_t * \fIfq\fB);
+
+\fBint fqueue_next(fqueue_t * \fIfq\fB);
+
+\fBint fevent(fset_t * \fIset\fB, fqueue_t * \fIfq\fB,
+const struct timespec * \fItimeo\fB);
+
+Compile and link with \fI-louroboros\fR.
+
+.SH DESCRIPTION
+
+These calls are used to
+
+The \fBfqueue_create\fR() function creates an fqueue_t structure which
+is an event queue that stores events that occured within a certain
+\fBfset_t\fR.
+
+The \fBfqueue_destroy\fR() function frees any resources associated with
+an \fBfqueue_t\fR \fIfq\fR.
+
+The \fBfqueue_next\fR() function retrieves the next event (a \fIflow
+descriptor\fR) that is ready within the event queue \fIfq\fR.
+
+The \fBfevent\fR() function retrieves all events that occured on any
+\fIflow descriptor\fR within \fIset\fR and returns them in the event
+queue \fBfq\fR. If a \fBstruct timespec *\fI timeo\fR can be provided,
+it will return either with new events or when \fItimeo\fR has expired.
+If \fItimeo\fR is NULL, the call will block indefinitely until an
+event occurs.
+
+.SH RETURN VALUE
+
+On success, \fBfqueue_create\fR() returns a pointer to an
+\fBfqueue_t\fR.
+
+\fBfqueue_destroy\fR() has no return value.
+
+On success, \fBfevent\fR() returns the number of events that occured in \fIset\fR.
+
+.SH ERRORS
+
+\fBfqueue_create\fR() returns NULL when insufficient resources
+were available to create the \fBfqueue_t\fR.
+
+\fBfqueue_next\fR() and \fBfevent\fR() can return
+
+.B -EINVAL
+An invalid argument was passed (\fIfq\fR or \fIset\fR was \fINULL\fR).
+
+In addition, \fBfqueue_next\fR() can return
+
+.B -EPERM
+No more fds available in \fIfq\fR.
+
+and \fBfevent\fR() can return
+
+.B -ETIMEDOUT
+the interval set int \fItimeo\tR expired before any event in \fIset\fR
+occured.
+
+.SH ATTRIBUTES
+
+For an explanation of the terms used in this section, see \fBattributes\fR(7).
+
+.TS
+box, tab(&);
+LB|LB|LB
+L|L|L.
+Interface & Attribute & Value
+_
+\fBfqueue_create\fR() & Thread safety & MT-Safe
+_
+\fBfqueue_destroy\fR() & Thread safety & MT-Safe
+_
+\fBfqueue_next\fR() & Thread safety & MT-Safe
+_
+\fBfevent\fR() & Thread safety & MT-Safe
+.TE
+
+.SH SEE ALSO
+
+.BR fset_create "(3), " fset_destroy "(3), " \
+fset_add "(3), " fset_del "(3), " fset_has "(3), " fset_zero (3)
+
+.SH COLOPHON
+This page is part of the Ouroboros project, found at
+https://bitbucket.org/ouroboros-rina/ouroboros
diff --git a/doc/man/fqueue_create.3 b/doc/man/fqueue_create.3
new file mode 100644
index 00000000..0e1f5f07
--- /dev/null
+++ b/doc/man/fqueue_create.3
@@ -0,0 +1 @@
+.so fqueue.3
diff --git a/doc/man/fqueue_destroy.3 b/doc/man/fqueue_destroy.3
new file mode 100644
index 00000000..0e1f5f07
--- /dev/null
+++ b/doc/man/fqueue_destroy.3
@@ -0,0 +1 @@
+.so fqueue.3
diff --git a/doc/man/fqueue_next.3 b/doc/man/fqueue_next.3
new file mode 100644
index 00000000..0e1f5f07
--- /dev/null
+++ b/doc/man/fqueue_next.3
@@ -0,0 +1 @@
+.so fqueue.3
diff --git a/doc/man/fset.3 b/doc/man/fset.3
new file mode 100644
index 00000000..b4f3a1b0
--- /dev/null
+++ b/doc/man/fset.3
@@ -0,0 +1,107 @@
+.\" Ouroboros man pages (C) 2017
+.\" Dimitri Staessens <dimitri.staessens@ugent.be>
+.\" Sander Vrijders <sander.vrijders@ugent.be>
+
+.TH FSET 3 2017-04-10 GNU "Ouroboros Programmer's Manual"
+
+.SH NAME
+
+fset_create, fset_destroy, fset_zero, fset_add, fset_del, fset_has \-
+manipulation of a set of flow descriptors
+
+.SH SYNOPSIS
+
+.B #include <ouroboros/fqueue.h>
+
+\fBfset_t * fset_create(void);
+
+\fBvoid fset_destroy(fset_t * \fIset\fB);
+
+\fBvoid fset_zero(fset_t * \fIset\fB);
+
+\fBint fset_add(fset_t * \fIset\fB, int \fIfd\fB);
+
+\fBint fset_del(fset_t * \fIset\fB, int \fIfd\fB);
+
+\fBbool fset_has(fset_t * \fIset\fB, int \fIfd\fB);
+
+Compile and link with \fI-louroboros\fR.
+
+.SH DESCRIPTION
+
+These calls are used to manage a set of \fIflow descriptors\fR.
+
+The \fBfset_create\fR() function creates and fset_t struct which holds
+a set of flow descriptors.
+
+The \fBfset_destroy\fR() function frees any resources associated with
+an \fBfset_t \fIset\fR.
+
+The \fBset_zero\fR() function removes all flow descriptors from an
+\fBfset_t \fIset\fR.
+
+The \fBset_add\fR() function adds a flow descriptor \fIfd\fR to the
+\fBfset_t \fIset\fR. A flow descriptor can be in at most one fset at a
+time.
+
+The \fBfset_del\fR() function removes a flow descriptor \fIfd\fR from
+the \fBfset_t \fIset\fR. \fBfset_del\fR() ignores invalid input.
+
+The \fBfset_has\fR() function checks whether a flow descriptor \fIfd\fR is
+an element of the \fBfset_t \fIset\fR.
+
+.SH RETURN VALUE
+
+On success, \fBfset_create\fR() returns a pointer to an \fBfset_t\fB.
+
+\fBfset_destroy\fR(), \fBset_zero\fR() and \fBfset_del\fR() have no return value.
+
+\fBfset_add\fR() returns 0 on success or an error code.
+
+\fBfset_has\fR() returns true when \fIfd\fR is in the set, false if it
+is not or on invalid input.
+
+.SH ERRORS
+
+\fBfset_create\fR() returns NULL when insufficient resources
+were available to create the \fBfset_t\fR.
+
+\fBfset_add\fR() can return the following errors:
+
+.B -EINVAL
+An invalid argument was passed (\fIset\fR was NULL or \fIfd\fR was
+invalid).
+
+.B -EPERM
+The passed flow descriptor \fIfd\fR was already in another \fBfset_t\fR.
+
+.SH ATTRIBUTES
+
+For an explanation of the terms used in this section, see \fBattributes\fR(7).
+
+.TS
+box, tab(&);
+LB|LB|LB
+L|L|L.
+Interface & Attribute & Value
+_
+\fBfset_create\fR() & Thread safety & MT-Safe
+_
+\fBfset_destroy\fR() & Thread safety & MT-Safe
+_
+\fBfset_zero\fR() & Thread safety & MT-Safe
+_
+\fBfset_add\fR() & Thread safety & MT-Safe
+_
+\fBfset_del\fR() & Thread safety & MT-Safe
+_
+\fBfset_has\fR() & Thread safety & MT-Safe
+.TE
+
+.SH SEE ALSO
+
+.BR fqueue "(3), " fevent (3)
+
+.SH COLOPHON
+This page is part of the Ouroboros project, found at
+https://bitbucket.org/ouroboros-rina/ouroboros
diff --git a/doc/man/fset_add.3 b/doc/man/fset_add.3
new file mode 100644
index 00000000..9fe22cd3
--- /dev/null
+++ b/doc/man/fset_add.3
@@ -0,0 +1 @@
+.so fset.3
diff --git a/doc/man/fset_create.3 b/doc/man/fset_create.3
new file mode 100644
index 00000000..9fe22cd3
--- /dev/null
+++ b/doc/man/fset_create.3
@@ -0,0 +1 @@
+.so fset.3
diff --git a/doc/man/fset_del.3 b/doc/man/fset_del.3
new file mode 100644
index 00000000..9fe22cd3
--- /dev/null
+++ b/doc/man/fset_del.3
@@ -0,0 +1 @@
+.so fset.3
diff --git a/doc/man/fset_destroy.3 b/doc/man/fset_destroy.3
new file mode 100644
index 00000000..9fe22cd3
--- /dev/null
+++ b/doc/man/fset_destroy.3
@@ -0,0 +1 @@
+.so fset.3
diff --git a/doc/man/fset_has.3 b/doc/man/fset_has.3
new file mode 100644
index 00000000..9fe22cd3
--- /dev/null
+++ b/doc/man/fset_has.3
@@ -0,0 +1 @@
+.so fset.3
diff --git a/doc/man/fset_zero.3 b/doc/man/fset_zero.3
new file mode 100644
index 00000000..9fe22cd3
--- /dev/null
+++ b/doc/man/fset_zero.3
@@ -0,0 +1 @@
+.so fset.3
diff --git a/include/ouroboros/fqueue.h b/include/ouroboros/fqueue.h
index 716d0216..48b5266e 100644
--- a/include/ouroboros/fqueue.h
+++ b/include/ouroboros/fqueue.h
@@ -32,35 +32,35 @@ struct flow_set;
 
 struct fqueue;
 
-typedef struct flow_set flow_set_t;
+typedef struct flow_set fset_t;
 typedef struct fqueue fqueue_t;
 
 __BEGIN_DECLS
 
-flow_set_t * flow_set_create(void);
+fset_t *   fset_create(void);
 
-void         flow_set_destroy(flow_set_t * set);
+void       fset_destroy(fset_t * set);
 
-fqueue_t *   fqueue_create(void);
+fqueue_t * fqueue_create(void);
 
-void         fqueue_destroy(struct fqueue * fq);
+void       fqueue_destroy(struct fqueue * fq);
 
-void         flow_set_zero(flow_set_t * set);
+void       fset_zero(fset_t * set);
 
-int          flow_set_add(flow_set_t * set,
-                          int          fd);
+int        fset_add(fset_t * set,
+                    int      fd);
 
-bool         flow_set_has(const flow_set_t * set,
-                          int                fd);
+bool       fset_has(const fset_t * set,
+                    int            fd);
 
-void         flow_set_del(flow_set_t * set,
-                          int          fd);
+void       fset_del(fset_t * set,
+                    int      fd);
 
-int          fqueue_next(fqueue_t * fq);
+int        fqueue_next(fqueue_t * fq);
 
-int          flow_event_wait(flow_set_t *            set,
-                             fqueue_t *              fq,
-                             const struct timespec * timeout);
+int        fevent(fset_t *                set,
+                  fqueue_t *              fq,
+                  const struct timespec * timeo);
 
 __END_DECLS
 
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;
 
diff --git a/src/lib/cdap.c b/src/lib/cdap.c
index 5ed86ad1..7d2feae3 100644
--- a/src/lib/cdap.c
+++ b/src/lib/cdap.c
@@ -52,7 +52,7 @@ struct fd_el {
 };
 
 struct cdap {
-        flow_set_t *     set;
+        fset_t *         set;
         fqueue_t *       fq;
 
         bool             proc;
@@ -304,7 +304,7 @@ static void * sdu_reader(void * o)
         ssize_t len;
         buffer_t data;
 
-        while (flow_event_wait(instance->set, instance->fq, NULL)) {
+        while (fevent(instance->set, instance->fq, NULL)) {
                 int fd;
                 set_proc(instance, true);
                 fd = fqueue_next(instance->fq);
@@ -440,7 +440,7 @@ struct cdap * cdap_create()
         if (instance->ids == NULL)
                 goto fail_bmp_create;
 
-        instance->set = flow_set_create();
+        instance->set = fset_create();
         if (instance->set == NULL)
                 goto fail_set_create;
 
@@ -463,7 +463,7 @@ struct cdap * cdap_create()
  fail_pthread_create:
         fqueue_destroy(instance->fq);
  fail_fqueue_create:
-        flow_set_destroy(instance->set);
+        fset_destroy(instance->set);
  fail_set_create:
         bmp_destroy(instance->ids);
  fail_bmp_create:
@@ -499,7 +499,7 @@ int cdap_destroy(struct cdap * instance)
 
         fqueue_destroy(instance->fq);
 
-        flow_set_destroy(instance->set);
+        fset_destroy(instance->set);
 
         pthread_cond_destroy(&instance->cond);
         pthread_mutex_destroy(&instance->mtx);
@@ -553,7 +553,7 @@ int cdap_add_flow(struct cdap * instance,
 
         pthread_rwlock_wrlock(&instance->flows_lock);
 
-        if (flow_set_add(instance->set, fd)) {
+        if (fset_add(instance->set, fd)) {
                 pthread_rwlock_unlock(&instance->flows_lock);
                 free(e);
                 return -1;
@@ -579,7 +579,7 @@ int cdap_del_flow(struct cdap * instance,
 
         pthread_rwlock_wrlock(&instance->flows_lock);
 
-        flow_set_del(instance->set, fd);
+        fset_del(instance->set, fd);
 
         list_for_each_safe(p, h, &instance->flows) {
                 struct fd_el * e = list_entry(p, struct fd_el, next);
diff --git a/src/lib/dev.c b/src/lib/dev.c
index cfcfdf81..d5044fb3 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -926,7 +926,7 @@ int flow_get_timeout(int               fd,
 {
         int ret = 0;
 
-        if (fd < 0 || fd >= AP_MAX_FLOWS || timeo == NULL)
+        if (fd < 0 || fd > AP_MAX_FLOWS || timeo == NULL)
                 return -EINVAL;
 
         pthread_rwlock_wrlock(&ai.lock);
@@ -949,7 +949,7 @@ int flow_get_timeout(int               fd,
 int flow_set_timeout(int                     fd,
                      const struct timespec * timeo)
 {
-        if (fd < 0 || fd >= AP_MAX_FLOWS)
+        if (fd < 0 || fd > AP_MAX_FLOWS)
                 return -EINVAL;
 
         pthread_rwlock_wrlock(&ai.lock);
@@ -974,7 +974,7 @@ int flow_set_timeout(int                     fd,
 int flow_get_qosspec(int         fd,
                      qosspec_t * qs)
 {
-        if (fd < 0 || fd >= AP_MAX_FLOWS || qs == NULL)
+        if (fd < 0 || fd > AP_MAX_FLOWS || qs == NULL)
                 return -EINVAL;
 
         pthread_rwlock_wrlock(&ai.lock);
@@ -1000,7 +1000,7 @@ ssize_t flow_write(int          fd,
         if (buf == NULL)
                 return 0;
 
-        if (fd < 0 || fd >= AP_MAX_FLOWS)
+        if (fd < 0 || fd > AP_MAX_FLOWS)
                 return -EBADF;
 
         pthread_rwlock_rdlock(&ai.lock);
@@ -1067,7 +1067,7 @@ ssize_t flow_read(int    fd,
         uint8_t * sdu;
         bool      used;
 
-        if (fd < 0 || fd >= AP_MAX_FLOWS)
+        if (fd < 0 || fd > AP_MAX_FLOWS)
                 return -EBADF;
 
         pthread_rwlock_rdlock(&ai.lock);
@@ -1104,7 +1104,7 @@ ssize_t flow_read(int    fd,
 
 /* fqueue functions. */
 
-struct flow_set * flow_set_create()
+struct flow_set * fset_create()
 {
         struct flow_set * set = malloc(sizeof(*set));
         if (set == NULL)
@@ -1126,12 +1126,12 @@ struct flow_set * flow_set_create()
         return set;
 }
 
-void flow_set_destroy(struct flow_set * set)
+void fset_destroy(struct flow_set * set)
 {
         if (set == NULL)
                 return;
 
-        flow_set_zero(set);
+        fset_zero(set);
 
         pthread_rwlock_wrlock(&ai.lock);
 
@@ -1163,7 +1163,7 @@ void fqueue_destroy(struct fqueue * fq)
         free(fq);
 }
 
-void flow_set_zero(struct flow_set * set)
+void fset_zero(struct flow_set * set)
 {
         if (set == NULL)
                 return;
@@ -1171,14 +1171,14 @@ void flow_set_zero(struct flow_set * set)
         shm_flow_set_zero(ai.fqset, set->idx);
 }
 
-int flow_set_add(struct flow_set * set,
-                 int               fd)
+int fset_add(struct flow_set * set,
+             int               fd)
 {
         int ret;
         size_t sdus;
         size_t i;
 
-        if (set == NULL)
+        if (set == NULL || fd < 0 || fd > AP_MAX_FLOWS)
                 return -EINVAL;
 
         pthread_rwlock_wrlock(&ai.lock);
@@ -1194,10 +1194,10 @@ int flow_set_add(struct flow_set * set,
         return ret;
 }
 
-void flow_set_del(struct flow_set * set,
-                  int               fd)
+void fset_del(struct flow_set * set,
+              int               fd)
 {
-        if (set == NULL)
+        if (set == NULL || fd < 0 || fd > AP_MAX_FLOWS)
                 return;
 
         pthread_rwlock_wrlock(&ai.lock);
@@ -1208,8 +1208,8 @@ void flow_set_del(struct flow_set * set,
         pthread_rwlock_unlock(&ai.lock);
 }
 
-bool flow_set_has(const struct flow_set * set,
-                  int                     fd)
+bool fset_has(const struct flow_set * set,
+              int                     fd)
 {
         bool ret = false;
 
@@ -1254,9 +1254,9 @@ int fqueue_next(struct fqueue * fq)
         return fd;
 }
 
-int flow_event_wait(struct flow_set *       set,
-                    struct fqueue *         fq,
-                    const struct timespec * timeout)
+int fevent(struct flow_set *       set,
+           struct fqueue *         fq,
+           const struct timespec * timeo)
 {
         ssize_t           ret;
         struct timespec   abstime;
@@ -1270,9 +1270,9 @@ int flow_event_wait(struct flow_set *       set,
 
         assert(!fq->next);
 
-        if (timeout != NULL) {
+        if (timeo != NULL) {
                 clock_gettime(PTHREAD_COND_CLOCK, &abstime);
-                ts_add(&abstime, timeout, &abstime);
+                ts_add(&abstime, timeo, &abstime);
                 t = &abstime;
         }
 
@@ -1541,7 +1541,7 @@ void ipcp_flow_fini(int fd)
 int ipcp_flow_get_qoscube(int         fd,
                           qoscube_t * cube)
 {
-        if (fd < 0 || fd >= AP_MAX_FLOWS || cube == NULL)
+        if (fd < 0 || fd > AP_MAX_FLOWS || cube == NULL)
                 return -EINVAL;
 
         pthread_rwlock_wrlock(&ai.lock);
diff --git a/src/tools/operf/operf.c b/src/tools/operf/operf.c
index e54fbc6b..62adcdb7 100644
--- a/src/tools/operf/operf.c
+++ b/src/tools/operf/operf.c
@@ -52,13 +52,13 @@ struct c {
 } client;
 
 struct s {
-        struct timespec  times[OPERF_MAX_FLOWS];
-        flow_set_t *     flows;
-        fqueue_t *       fq;
-        pthread_mutex_t  lock;
+        struct timespec times[OPERF_MAX_FLOWS];
+        fset_t *        flows;
+        fqueue_t *      fq;
+        pthread_mutex_t lock;
 
-        uint8_t          buffer[OPERF_BUF_SIZE];
-        ssize_t          timeout;
+        uint8_t buffer[OPERF_BUF_SIZE];
+        ssize_t timeout;
 
         pthread_t cleaner_pt;
         pthread_t accept_pt;
diff --git a/src/tools/operf/operf_server.c b/src/tools/operf/operf_server.c
index c016ad63..858fbb90 100644
--- a/src/tools/operf/operf_server.c
+++ b/src/tools/operf/operf_server.c
@@ -50,11 +50,11 @@ void * cleaner_thread(void * o)
                 clock_gettime(CLOCK_REALTIME, &now);
                 pthread_mutex_lock(&server.lock);
                 for (i = 0; i < OPERF_MAX_FLOWS; ++i)
-                        if (flow_set_has(server.flows, i) &&
+                        if (fset_has(server.flows, i) &&
                             ts_diff_ms(&server.times[i], &now)
                             > server.timeout) {
                                 printf("Flow %d timed out.\n", i);
-                                flow_set_del(server.flows, i);
+                                fset_del(server.flows, i);
                                 flow_dealloc(i);
                         }
 
@@ -72,7 +72,7 @@ void * server_thread(void *o)
 
         (void) o;
 
-        while (flow_event_wait(server.flows, server.fq, &timeout))
+        while (fevent(server.flows, server.fq, &timeout))
                 while ((fd = fqueue_next(server.fq)) >= 0) {
                         msg_len = flow_read(fd, server.buffer, OPERF_BUF_SIZE);
                         if (msg_len < 0)
@@ -115,7 +115,7 @@ void * accept_thread(void * o)
                 clock_gettime(CLOCK_REALTIME, &now);
 
                 pthread_mutex_lock(&server.lock);
-                flow_set_add(server.flows, fd);
+                fset_add(server.flows, fd);
                 server.times[fd] = now;
                 pthread_mutex_unlock(&server.lock);
         }
@@ -139,13 +139,13 @@ int server_main(void)
                 return -1;
         }
 
-        server.flows = flow_set_create();
+        server.flows = fset_create();
         if (server.flows == NULL)
                 return 0;
 
         server.fq = fqueue_create();
         if (server.fq == NULL) {
-                flow_set_destroy(server.flows);
+                fset_destroy(server.flows);
                 return -1;
         }
 
@@ -158,7 +158,7 @@ int server_main(void)
         pthread_cancel(server.server_pt);
         pthread_cancel(server.cleaner_pt);
 
-        flow_set_destroy(server.flows);
+        fset_destroy(server.flows);
         fqueue_destroy(server.fq);
 
         pthread_join(server.server_pt, NULL);
diff --git a/src/tools/oping/oping.c b/src/tools/oping/oping.c
index 13132815..bca4f80b 100644
--- a/src/tools/oping/oping.c
+++ b/src/tools/oping/oping.c
@@ -63,10 +63,10 @@ struct c {
 } client;
 
 struct s {
-        struct timespec   times[OPING_MAX_FLOWS];
-        flow_set_t *      flows;
-        fqueue_t *        fq;
-        pthread_mutex_t   lock;
+        struct timespec times[OPING_MAX_FLOWS];
+        fset_t *        flows;
+        fqueue_t *      fq;
+        pthread_mutex_t lock;
 
         pthread_t cleaner_pt;
         pthread_t accept_pt;
diff --git a/src/tools/oping/oping_server.c b/src/tools/oping/oping_server.c
index 57d1fd7c..6a535caf 100644
--- a/src/tools/oping/oping_server.c
+++ b/src/tools/oping/oping_server.c
@@ -51,10 +51,10 @@ void * cleaner_thread(void * o)
                 clock_gettime(CLOCK_REALTIME, &now);
                 pthread_mutex_lock(&server.lock);
                 for (i = 0; i < OPING_MAX_FLOWS; ++i)
-                        if (flow_set_has(server.flows, i) &&
+                        if (fset_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);
+                                fset_del(server.flows, i);
                                 flow_dealloc(i);
                         }
 
@@ -75,8 +75,7 @@ void * server_thread(void *o)
         (void) o;
 
         while (true) {
-                if (flow_event_wait(server.flows, server.fq, &timeout)
-                    == -ETIMEDOUT)
+                if (fevent(server.flows, server.fq, &timeout) == -ETIMEDOUT)
                         continue;
 
                 while ((fd = fqueue_next(server.fq)) >= 0) {
@@ -127,7 +126,7 @@ void * accept_thread(void * o)
                 clock_gettime(CLOCK_REALTIME, &now);
 
                 pthread_mutex_lock(&server.lock);
-                flow_set_add(server.flows, fd);
+                fset_add(server.flows, fd);
                 server.times[fd] = now;
                 pthread_mutex_unlock(&server.lock);
 
@@ -153,13 +152,13 @@ int server_main(void)
                 return -1;
         }
 
-        server.flows = flow_set_create();
+        server.flows = fset_create();
         if (server.flows == NULL)
                 return 0;
 
         server.fq = fqueue_create();
         if (server.fq == NULL) {
-                flow_set_destroy(server.flows);
+                fset_destroy(server.flows);
                 return -1;
         }
 
@@ -172,7 +171,7 @@ int server_main(void)
         pthread_cancel(server.server_pt);
         pthread_cancel(server.cleaner_pt);
 
-        flow_set_destroy(server.flows);
+        fset_destroy(server.flows);
         fqueue_destroy(server.fq);
 
         pthread_join(server.server_pt, NULL);