lib: Demultiplex the fast path

The fast path will now use an incoming ring buffer per flow per
process. This necessitated the development of a new method for the
asynchronous io call, which is now based on an event queue system for
scalability (fqueue). The ipcpd's and tools have been updated to this
API.

9 files changed, 144 insertions, 94 deletions

diff --git a/include/ouroboros/CMakeLists.txt b/include/ouroboros/CMakeLists.txt
index f24857ed..41feb65e 100644
--- a/include/ouroboros/CMakeLists.txt
+++ b/include/ouroboros/CMakeLists.txt
@@ -7,11 +7,11 @@ set(HEADER_FILES
   dev.h
   errno.h
   fcntl.h
+  fqueue.h
   irm.h
   irm_config.h
   nsm.h
-  qos.h
-  select.h)
+  qos.h)
 
 install(FILES ${HEADER_FILES} DESTINATION usr/include/ouroboros)
 
diff --git a/include/ouroboros/config.h.in b/include/ouroboros/config.h.in
index 143ae7c8..a9d65aec 100644
--- a/include/ouroboros/config.h.in
+++ b/include/ouroboros/config.h.in
@@ -36,6 +36,7 @@
 #define IPCP_NORMAL_EXEC       "@IPCP_NORMAL_TARGET@"
 #define IPCP_LOCAL_EXEC        "@IPCP_LOCAL_TARGET@"
 #define AP_MAX_FLOWS           256
+#define AP_MAX_FQUEUES         64
 #define SHM_RDRB_BLOCK_SIZE    sysconf(_SC_PAGESIZE)
 #define SHM_RDRB_MULTI_BLOCK
 #define SHM_RDRB_PREFIX        "/ouroboros.rdrb."
@@ -43,7 +44,8 @@
 #define SHM_BUFFER_SIZE        (1 << 14)
 #define DU_BUFF_HEADSPACE      128
 #define DU_BUFF_TAILSPACE      0
-#define SHM_AP_RBUFF_PREFIX    "/ouroboros.rbuff."
+#define SHM_RBUFF_PREFIX       "/ouroboros.rbuff."
+#define SHM_FLOW_SET_PREFIX    "/ouroboros.sets."
 #define IRMD_MAX_FLOWS         4096
 #define IRMD_THREADPOOL_SIZE   5
 #define LOG_DIR                "/@LOG_DIR@/"
diff --git a/include/ouroboros/select.h b/include/ouroboros/fqueue.h
index de309b8d..943d6510 100644
--- a/include/ouroboros/select.h
+++ b/include/ouroboros/fqueue.h
@@ -1,7 +1,7 @@
 /*
  * Ouroboros - Copyright (C) 2016
  *
- * A select call for flows
+ * Flow queues
  *
  *    Dimitri Staessens <[email protected]>
  *    Sander Vrijders   <[email protected]>
@@ -21,32 +21,42 @@
  * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  */
 
-#ifndef OUROBOROS_SELECT_H
-#define OUROBOROS_SELECT_H
+#ifndef OUROBOROS_FQUEUE_H
+#define OUROBOROS_FQUEUE_H
 
 #include <stdbool.h>
 #include <time.h>
 
 struct flow_set;
 
+struct fqueue;
+
 typedef struct flow_set flow_set_t;
+typedef struct fqueue fqueue_t;
 
 flow_set_t * flow_set_create();
 
 void         flow_set_destroy(flow_set_t * set);
 
+fqueue_t *   fqueue_create();
+
+void         fqueue_destroy(struct fqueue * fq);
+
 void         flow_set_zero(flow_set_t * set);
 
-void         flow_set_add(flow_set_t * set,
+int          flow_set_add(flow_set_t * set,
                           int          fd);
 
-void         flow_set_del(flow_set_t * set,
+bool         flow_set_has(flow_set_t * set,
                           int          fd);
 
-bool         flow_set_has(flow_set_t * set,
+void         flow_set_del(flow_set_t * set,
                           int          fd);
 
-int          flow_select(flow_set_t *            set,
-                         const struct timespec * timeout);
+int          fqueue_next(fqueue_t * fq);
+
+int          flow_event_wait(flow_set_t *            set,
+                             fqueue_t *              fq,
+                             const struct timespec * timeout);
 
 #endif /* OUROBOROS_SELECT_H */
diff --git a/include/ouroboros/ipcp-dev.h b/include/ouroboros/ipcp-dev.h
index 9343aeaa..3ab05bd7 100644
--- a/include/ouroboros/ipcp-dev.h
+++ b/include/ouroboros/ipcp-dev.h
@@ -47,7 +47,4 @@ int  ipcp_flow_write(int                  fd,
 
 void ipcp_flow_del(struct shm_du_buff * sdb);
 
-/* returns flow descriptor and du buff */
-int  ipcp_read_shim(struct shm_du_buff ** sdb);
-
 #endif /* OUROBOROS_IPCP_DEV_H */
diff --git a/include/ouroboros/local-dev.h b/include/ouroboros/local-dev.h
index 77ff47e9..30f440b1 100644
--- a/include/ouroboros/local-dev.h
+++ b/include/ouroboros/local-dev.h
@@ -20,14 +20,12 @@
  * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  */
 
-#include <ouroboros/shm_ap_rbuff.h>
-
 #ifndef OUROBOROS_LOCAL_DEV_H
 #define OUROBOROS_LOCAL_DEV_H
 
-struct rb_entry * local_flow_read(int fd);
+ssize_t local_flow_read(int fd);
 
-int               local_flow_write(int               fd,
-                                   struct rb_entry * e);
+int     local_flow_write(int     fd,
+                         ssize_t idx);
 
 #endif /* OUROBOROS_LOCAL_DEV_H */
diff --git a/include/ouroboros/shm_ap_rbuff.h b/include/ouroboros/shm_ap_rbuff.h
deleted file mode 100644
index 453e4bf8..00000000
--- a/include/ouroboros/shm_ap_rbuff.h
+++ /dev/null
@@ -1,73 +0,0 @@
-/*
- * Ouroboros - Copyright (C) 2016
- *
- * Ring buffer for application processes
- *
- *    Dimitri Staessens <[email protected]>
- *    Sander Vrijders   <[email protected]>
- *
- * This program is free software; you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation; either version 2 of the License, or
- * (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program; if not, write to the Free Software
- * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
- */
-
-#ifndef OUROBOROS_SHM_AP_RBUFF_H
-#define OUROBOROS_SHM_AP_RBUFF_H
-
-#include <ouroboros/select.h>
-#include <sys/types.h>
-#include <sys/time.h>
-#include <stdbool.h>
-
-struct shm_ap_rbuff;
-
-struct rb_entry {
-        ssize_t index;
-        int     port_id;
-};
-
-struct shm_ap_rbuff * shm_ap_rbuff_create();
-
-struct shm_ap_rbuff * shm_ap_rbuff_open(pid_t api);
-
-void                  shm_ap_rbuff_close(struct shm_ap_rbuff * rb);
-
-void                  shm_ap_rbuff_destroy(struct shm_ap_rbuff * rb);
-
-void                  shm_ap_rbuff_open_port(struct shm_ap_rbuff * rb,
-                                             int                   port_id);
-
-int                   shm_ap_rbuff_close_port(struct shm_ap_rbuff * rb,
-                                              int                   port_id);
-
-int                   shm_ap_rbuff_write(struct shm_ap_rbuff * rb,
-                                         struct rb_entry *     e);
-
-struct rb_entry *     shm_ap_rbuff_read(struct shm_ap_rbuff * rb);
-
-int                   shm_ap_rbuff_pop_idx(struct shm_ap_rbuff * rb);
-
-int                   shm_ap_rbuff_peek_b(struct shm_ap_rbuff *   rb,
-                                          bool *                  set,
-                                          const struct timespec * timeout);
-
-ssize_t               shm_ap_rbuff_read_port(struct shm_ap_rbuff * rb,
-                                             int                   port_id);
-
-ssize_t               shm_ap_rbuff_read_port_b(struct shm_ap_rbuff *   rb,
-                                               int                     port_id,
-                                               const struct timespec * timeout);
-
-void                  shm_ap_rbuff_reset(struct shm_ap_rbuff * rb);
-
-#endif /* OUROBOROS_SHM_AP_RBUFF_H */
diff --git a/include/ouroboros/shm_flow_set.h b/include/ouroboros/shm_flow_set.h
new file mode 100644
index 00000000..32db5d36
--- /dev/null
+++ b/include/ouroboros/shm_flow_set.h
@@ -0,0 +1,63 @@
+/*
+ * Ouroboros - Copyright (C) 2016
+ *
+ * Management of flow_sets for fqueue
+ *
+ *    Dimitri Staessens <[email protected]>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+#ifndef OUROBOROS_SHM_FLOW_SET_H
+#define OUROBOROS_SHM_FLOW_SET_H
+
+#include <ouroboros/fqueue.h>
+
+#include <sys/time.h>
+
+struct shm_flow_set;
+
+struct shm_flow_set * shm_flow_set_create();
+
+void                  shm_flow_set_destroy(struct shm_flow_set * set);
+
+struct shm_flow_set * shm_flow_set_open(pid_t api);
+
+void                  shm_flow_set_close(struct shm_flow_set * set);
+
+void                  shm_flow_set_zero(struct shm_flow_set * shm_set,
+                                        ssize_t               idx);
+
+int                   shm_flow_set_add(struct shm_flow_set * shm_set,
+                                       ssize_t               idx,
+                                       int                   port_id);
+
+int                   shm_flow_set_has(struct shm_flow_set * shm_set,
+                                       ssize_t               idx,
+                                       int                   port_id);
+
+void                  shm_flow_set_del(struct shm_flow_set * shm_set,
+                                       ssize_t               idx,
+                                       int                   port_id);
+
+void                  shm_flow_set_notify(struct shm_flow_set * set,
+                                          int                   port_id);
+
+int                   shm_flow_set_wait(const struct shm_flow_set * shm_set,
+                                        ssize_t                     idx,
+                                        int *                       fqueue,
+                                        const struct timespec *     timeout);
+
+#endif /* OUROBOROS_SHM_FLOW_SET_H */
diff --git a/include/ouroboros/shm_rbuff.h b/include/ouroboros/shm_rbuff.h
new file mode 100644
index 00000000..03660b88
--- /dev/null
+++ b/include/ouroboros/shm_rbuff.h
@@ -0,0 +1,53 @@
+/*
+ * Ouroboros - Copyright (C) 2016
+ *
+ * Ring buffer for incoming SDUs
+ *
+ *    Dimitri Staessens <[email protected]>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+#ifndef OUROBOROS_SHM_RBUFF_H
+#define OUROBOROS_SHM_RBUFF_H
+
+#include <sys/types.h>
+#include <sys/time.h>
+
+struct shm_rbuff;
+
+struct shm_rbuff * shm_rbuff_create(int port_id);
+
+struct shm_rbuff * shm_rbuff_open(pid_t api, int port_id);
+
+void               shm_rbuff_close(struct shm_rbuff * rb);
+
+void               shm_rbuff_destroy(struct shm_rbuff * rb);
+
+int                shm_rbuff_block(struct shm_rbuff * rb);
+
+void               shm_rbuff_unblock(struct shm_rbuff * rb);
+
+int                shm_rbuff_write(struct shm_rbuff * rb,
+                                   ssize_t            idx);
+
+ssize_t            shm_rbuff_read(struct shm_rbuff * rb);
+
+ssize_t            shm_rbuff_read_b(struct shm_rbuff *      rb,
+                                    const struct timespec * timeout);
+
+void               shm_rbuff_reset(struct shm_rbuff * rb);
+
+#endif /* OUROBOROS_SHM_RBUFF_H */
diff --git a/include/ouroboros/wrap/ouroboros.i b/include/ouroboros/wrap/ouroboros.i
index 394b505a..26cc6076 100644
--- a/include/ouroboros/wrap/ouroboros.i
+++ b/include/ouroboros/wrap/ouroboros.i
@@ -26,11 +26,11 @@
 #include "ouroboros/dev.h"
 #include "ouroboros/errno.h"
 #include "ouroboros/fcntl.h"
+#include "ouroboros/fqueue.h"
 #include "ouroboros/irm.h"
 #include "ouroboros/irm_config.h"
 #include "ouroboros/nsm.h"
 #include "ouroboros/qos.h"
-#include "ouroboros/select.h"
 %}
 
 typedef int pid_t;
@@ -39,8 +39,8 @@ typedef int pid_t;
 %include "ouroboros/dev.h"
 %include "ouroboros/errno.h"
 %include "ouroboros/fcntl.h"
+%include "ouroboros/fqueue.h"
 %include "ouroboros/irm.h"
 %include "ouroboros/irm_config.h"
 %include "ouroboros/nsm.h"
 %include "ouroboros/qos.h"
-%include "ouroboros/select.h"