core: support configurable event loop mechanism

CMakeLists.txt

@@ -165,6 +165,22 @@ option(FLB_CONFIG_YAML         "Enable YAML config format"    Yes)
 # List of plugins available and defaults for each option
 include(cmake/plugins_options.cmake)
 
+
+# Event loop backend (advanced users only)
+# ----------------------------------------
+# The following options are for advanced users only, they are used to enable
+# specific event loop backends. By default, the build system will try to
+# detect the best backend for the current Operating System. If you want to
+# force a specific backend, you can use ONE of the following options.
+#
+# Note that these options will enable the proper option for Monkey library.
+option(FLB_EVENT_LOOP_EPOLL     "Enable epoll(2) event loop backend"  No)
+option(FLB_EVENT_LOOP_POLL      "Enable poll(2) event loop backend"   No)
+option(FLB_EVENT_LOOP_KQUEUE    "Enable kqueue(2) event loop backend" No)
+option(FLB_EVENT_LOOP_SELECT    "Enable select(2) event loop backend" No)
+option(FLB_EVENT_LOOP_LIBEVENT  "Enable libevent event loop backend"  No)
+
+
 if(DEFINED FLB_NIGHTLY_BUILD AND NOT "${FLB_NIGHTLY_BUILD}" STREQUAL "")
   FLB_DEFINITION_VAL(FLB_NIGHTLY_BUILD ${FLB_NIGHTLY_BUILD})
 endif()
@@ -501,12 +517,38 @@ if(FLB_TESTS_RUNTIME)
   FLB_OPTION(FLB_HTTP_SERVER  ON)
 endif()
 
-# MK Core
+# Monkey Core Library
 macro(MK_SET_OPTION option value)
   set(${option} ${value} CACHE INTERNAL "" FORCE)
 endmacro()
-MK_SET_OPTION(MK_SYSTEM_MALLOC  ON)
-MK_SET_OPTION(MK_DEBUG          ON)
+MK_SET_OPTION(MK_SYSTEM_MALLOC    ON)
+MK_SET_OPTION(MK_DEBUG            ON)
+
+# Monkey backend event loop
+if (FLB_EVENT_LOOP_EPOLL)
+  MK_SET_OPTION(MK_EVENT_LOOP_EPOLL ON)
+  FLB_DEFINITION(FLB_EVENT_LOOP_EPOLL)
+  message(STATUS "FLB Event loop backend > epoll(2)")
+elseif (FLB_EVENT_LOOP_POLL)
+  MK_SET_OPTION(MK_EVENT_LOOP_POLL ON)
+  FLB_DEFINITION(FLB_EVENT_LOOP_POLL)
+  message(STATUS "FLB Event loop backend > poll(2)")
+elseif (FLB_EVENT_LOOP_KQUEUE)
+  MK_SET_OPTION(MK_EVENT_LOOP_KQUEUE ON)
+  FLB_DEFINITION(FLB_EVENT_LOOP_KQUEUE)
+  message(STATUS "FLB Event loop backend > kqueue(2)")
+elseif (FLB_EVENT_LOOP_SELECT)
+  MK_SET_OPTION(MK_EVENT_LOOP_SELECT ON)
+  FLB_DEFINITION(FLB_EVENT_LOOP_SELECT)
+  message(STATUS "FLB Event loop backend > select(2)")
+elseif (FLB_EVENT_LOOP_LIBEVENT)
+  MK_SET_OPTION(MK_EVENT_LOOP_LIBEVENT ON)
+  FLB_DEFINITION(FLB_EVENT_LOOP_LIBEVENT)
+  message(STATUS "FLB Event loop backend > libevent")
+else()
+  FLB_DEFINITION(FLB_EVENT_LOOP_AUTO_DISCOVERY)
+  message(STATUS "FLB Event loop backend > auto discovery (Monkey library)")
+endif()
 
 # Build Monkey HTTP Server
 if(FLB_HTTP_SERVER)

lib/monkey/CMakeLists.txt

@@ -71,6 +71,19 @@ option(MK_WITHOUT_BIN          "Do not build binary"      No)
 option(MK_WITHOUT_CONF         "Skip configuration files" No)
 option(MK_STATIC_LIB_MODE      "Static library mode"      No)
 
+
+# Event loop backend
+# ------------------
+# Options to force a specific event loop mechanism. If all of them are
+# turned off, the system will do an automatic selection.
+#
+# Check on mk_core/CMakeLists.txt for more details.
+option(MK_EVENT_LOOP_SELECT    "Use select(2) event loop" No)
+option(MK_EVENT_LOOP_POLL      "Use poll(2) event loop"   No)
+option(MK_EVENT_LOOP_KQUEUE    "Use kqueue(2) event loop" No)
+option(MK_EVENT_LOOP_EPOLL     "Use epoll(2) event loop"  No)
+option(MK_EVENT_LOOP_LIBEVENT  "Use libevent event loop"  No)
+
 # If building just for a "library" mode, disable plugins
 if(MK_LIB_ONLY)
   set(MK_PLUGIN_AUTH       No)
@@ -279,7 +292,16 @@ include_directories(include/monkey/)
 include_directories(${CMAKE_CURRENT_BINARY_DIR}/include/)
 include_directories(${CMAKE_CURRENT_BINARY_DIR}/include/monkey/)
 
-if (CMAKE_SYSTEM_NAME MATCHES "Windows")
+# Check if we need to build libevent by using auto discovery mechanism
+if (CMAKE_SYSTEM_NAME MATCHES "Windows" AND
+    (
+      NOT MK_EVENT_LOOP_SELECT AND
+      NOT MK_EVENT_LOOP_POLL AND
+      NOT MK_EVENT_LOOP_KQUEUE AND
+      NOT MK_EVENT_LOOP_EPOLL
+    ) OR MK_EVENT_LOOP_LIBEVENT)
+
+  set(MK_EVENT_LOOP_LIBEVENT Yes)
   include_directories(mk_core/deps/libevent/include)
   include_directories("${PROJECT_BINARY_DIR}/mk_core/deps/libevent/include/")
 endif()

lib/monkey/include/monkey/mk_core/mk_core_info.h.in

@@ -23,4 +23,5 @@
 /* General flags set by CMakeLists.txt */
 @MK_CORE_BUILD_FLAGS@
 
+
 #endif

lib/monkey/include/monkey/mk_core/mk_event.h

@@ -70,14 +70,27 @@
 
 #define MK_EVENT_IS_REGISTERED(event) ((event->status & MK_EVENT_REGISTERED) != 0)
 
-#if defined(_WIN32)
-    #include "mk_event_libevent.h"
-#elif defined(MK_HAVE_EVENT_SELECT)
+#if defined(MK_EVENT_LOOP_SELECT)
     #include "mk_event_select.h"
-#elif defined(__linux__) && !defined(LINUX_KQUEUE)
+#elif defined(MK_EVENT_LOOP_POLL)
+    #include "mk_event_poll.h"
+#elif defined(MK_EVENT_LOOP_KQUEUE)
+    #include "mk_event_kqueue.h"
+#elif defined(MK_EVENT_LOOP_EPOLL)
     #include "mk_event_epoll.h"
+#elif defined(MK_EVENT_LOOP_LIBEVENT)
+    #include "mk_event_libevent.h"
 #else
-    #include "mk_event_kqueue.h"
+    /* do our best based on the operating system */
+    #if defined(__linux__)
+        #include "mk_event_epoll.h"
+    #elif defined(__FreeBSD__) || defined(__APPLE__) || defined(__DragonFly__) || defined(__OpenBSD__)
+        #include "mk_event_kqueue.h"
+    #elif defined(_WIN32)
+        #include "mk_event_libevent.h"
+    #else
+        #include "mk_event_select.h"
+    #endif
 #endif
 
 #if defined(_WIN32)

lib/monkey/include/monkey/mk_core/mk_event_select.h

@@ -53,7 +53,7 @@ struct mk_event_ctx {
     for (__i = 0;                                                            \
          __i < evl->n_events;                                                \
          __i++,                                                              \
-             event = ((__i < evl->n_events) ? __ctx->fired[__i].data : NULL) \
+             event = ((__i < evl->n_events) ? &__ctx->fired[__i] : NULL) \
          )
 
 #endif

lib/monkey/mk_core/CMakeLists.txt

@@ -15,6 +15,7 @@ include_directories(../include/monkey/)
 macro(MK_DEFINITION var)
   add_definitions(-D${var})
   set(MK_CORE_BUILD_FLAGS "${MK_CORE_BUILD_FLAGS}#ifndef ${var}\n#define ${var}\n#endif\n")
+  set(MK_CORE_INFO_DEFS "${MK_CORE_INFO_DEFS} ${var}")
 endmacro()
 
 # Set threading system
@@ -26,6 +27,9 @@ if (CMAKE_SYSTEM_NAME MATCHES "Windows")
     )
   add_subdirectory(deps/)
 else()
+  if (MK_EVENT_LOOP_LIBEVENT)
+    add_subdirectory(deps/)
+  endif()
   MK_DEFINITION(MK_THREADS_POSIX)
 endif()
 
@@ -86,23 +90,78 @@ if (HAVE_UNISTD_H)
   MK_DEFINITION(MK_HAVE_UNISTD_H)
 endif()
 
-# Lookup event-loop mechanism: do we need to fallback to select(2) ?
+# Check for a valid event loop mechanism
+
+# select(2)
 check_c_source_compiles("
-  #include <sys/epoll.h>
+  #include <stdio.h>
+  #include <sys/select.h>
   int main() {
-     return epoll_create(1);
-  }" HAVE_EPOLL)
+     fd_set rfds;
+     struct timeval tv;
+     return select(1, &rfds, NULL, NULL, &tv);
+  }" HAVE_SELECT)
+
+check_c_source_compiles("
+  #include <poll.h>
+  int main() {
+     struct pollfd fds[1];
+     fds[0].fd = 0;
+     fds[0].events = POLLIN;
+     return poll(fds, 1, 1);
+  }" HAVE_POLL)
 
 check_c_source_compiles("
   #include <sys/event.h>
   int main() {
      return kqueue();
   }" HAVE_KQUEUE)
 
+check_c_source_compiles("
+  #include <sys/epoll.h>
+  int main() {
+     return epoll_create(1);
+  }" HAVE_EPOLL)
+
+if (MK_EVENT_LOOP_SELECT)
+  if (NOT HAVE_SELECT)
+    message(FATAL_ERROR "Event loop backend > select(2) not available")
+  else()
+    message(STATUS "Event loop backend > select(2)")
+    MK_DEFINITION(MK_EVENT_LOOP_SELECT)
+  endif()
+endif()
+
+if (MK_EVENT_LOOP_POLL)
+  if (NOT HAVE_POLL)
+    message(FATAL_ERROR "Event loop backend > poll(2) not available")
+  else()
+    message(STATUS "Event loop backend > poll(2)")
+    MK_DEFINITION(MK_EVENT_LOOP_POLL)
+  endif()
+endif()
 
-if ((NOT HAVE_EPOLL AND NOT HAVE_KQUEUE) OR MK_USE_EVENT_SELECT)
-  message(STATUS "Event loop backend > select(2)")
-  MK_DEFINITION(MK_HAVE_EVENT_SELECT)
+if (MK_EVENT_LOOP_KQUEUE)
+  if (NOT HAVE_KQUEUE)
+    message(FATAL_ERROR "Event loop backend > kqueue(2) not available")
+  else()
+    message(STATUS "Event loop backend > kqueue(2)")
+    MK_DEFINITION(MK_EVENT_LOOP_KQUEUE)
+  endif()
+endif()
+
+if (MK_EVENT_LOOP_EPOLL)
+  if (NOT HAVE_EPOLL)
+    message(FATAL_ERROR "Event loop backend > epoll(2) not available")
+  else()
+    message(STATUS "Event loop backend > epoll(2)")
+    MK_DEFINITION(MK_EVENT_LOOP_EPOLL)
+  endif()
+endif()
+
+if (MK_EVENT_LOOP_LIBEVENT)
+    message(STATUS "Event loop backend > libevent")
+    MK_DEFINITION(MK_EVENT_LOOP_LIBEVENT)
 endif()
 
 # Validate timerfd_create()
@@ -145,12 +204,12 @@ endif()
 
 configure_file(
   "${CMAKE_CURRENT_SOURCE_DIR}/../include/monkey/mk_core/mk_core_info.h.in"
-  "${PROJECT_BINARY_DIR}/include/monkey/mk_core/mk_core_info.h" 
+  "${PROJECT_BINARY_DIR}/include/monkey/mk_core/mk_core_info.h"
   )
 
 add_library(mk_core STATIC ${src})
 target_link_libraries(mk_core ${CMAKE_THREAD_LIBS_INIT})
 
-if (CMAKE_SYSTEM_NAME MATCHES "Windows")
+if (MK_EVENT_LOOP_LIBEVENT)
   target_link_libraries(mk_core event)
 endif()

lib/monkey/mk_core/deps/CMakeLists.txt

@@ -1,13 +1,13 @@
-macro(FORCE_OPTION option value)
-  set(${option} ${value} CACHE "" INTERNAL FORCE)
+macro(FORCE_OPTION option value type)
+  set(${option} ${value} CACHE ${type} INTERNAL FORCE)
 endmacro()
 
-FORCE_OPTION(EVENT__DISABLE_THREAD_SUPPORT  ON)
-FORCE_OPTION(EVENT__DISABLE_OPENSSL         ON)
-FORCE_OPTION(EVENT__DISABLE_BENCHMARK       ON)
-FORCE_OPTION(EVENT__DISABLE_TESTS           ON)
-FORCE_OPTION(EVENT__DISABLE_REGRESS         ON)
-FORCE_OPTION(EVENT__DISABLE_SAMPLES         ON)
-FORCE_OPTION(EVENT__BUILD_SHARED_LIBRARIES  OFF)
+FORCE_OPTION(EVENT__DISABLE_THREAD_SUPPORT  ON   BOOL)
+FORCE_OPTION(EVENT__DISABLE_OPENSSL         ON   BOOL)
+FORCE_OPTION(EVENT__DISABLE_BENCHMARK       ON   BOOL)
+FORCE_OPTION(EVENT__DISABLE_TESTS           ON   BOOL)
+FORCE_OPTION(EVENT__DISABLE_REGRESS         ON   BOOL)
+FORCE_OPTION(EVENT__DISABLE_SAMPLES         ON   BOOL)
+FORCE_OPTION(EVENT__BUILD_SHARED_LIBRARIES  OFF  BOOL)
 
 add_subdirectory(libevent)

lib/monkey/mk_core/mk_event.c

@@ -28,14 +28,27 @@
 #include <mk_core/mk_utils.h>
 #include <mk_core/mk_event.h>
 
-#if defined(_WIN32)
-    #include "mk_event_libevent.c"
-#elif defined(MK_HAVE_EVENT_SELECT)
+#if defined(MK_EVENT_LOOP_SELECT)
     #include "mk_event_select.c"
-#elif defined(__linux__) && !defined(LINUX_KQUEUE)
+#elif defined(MK_EVENT_LOOP_POLL)
+    #include "mk_event_poll.c"
+#elif defined(MK_EVENT_LOOP_KQUEUE)
+    #include "mk_event_kqueue.c"
+#elif defined(MK_EVENT_LOOP_EPOLL)
     #include "mk_event_epoll.c"
+#elif defined(MK_EVENT_LOOP_LIBEVENT)
+    #include "mk_event_libevent.c"
 #else
-    #include "mk_event_kqueue.c"
+    /* do our best based on the operating system */
+    #if defined(__linux__)
+        #include "mk_event_epoll.c"
+    #elif defined(__FreeBSD__) || defined(__APPLE__) || defined(__DragonFly__) || defined(__OpenBSD__)
+        #include "mk_event_kqueue.c"
+    #elif defined(_WIN32)
+        #include "mk_event_libevent.c"
+    #else
+        #include "mk_event_select.c"
+    #endif
 #endif
 
 /* Initialize backend */

src/flb_output_thread.c

@@ -249,7 +249,8 @@ static void output_thread(void *data)
     while (running) {
         mk_event_wait(th_ins->evl);
         flb_event_priority_live_foreach(event, th_ins->evl_bktq, th_ins->evl,
-                                      FLB_ENGINE_LOOP_MAX_ITER) {
+                                         FLB_ENGINE_LOOP_MAX_ITER) {
+
             /*
              * FIXME
              * -----
@@ -274,7 +275,6 @@ static void output_thread(void *data)
                     flb_errno();
                     continue;
                 }
-
                 /*
                  * If the address receives 0xdeadbeef, means the thread must
                  * be terminated.
@@ -285,16 +285,17 @@ static void output_thread(void *data)
                                  thread_id);
                     continue;
                 }
-
-                /* Start the co-routine with the flush callback */
-                out_flush = flb_output_flush_create(task,
-                                                    task->i_ins,
-                                                    th_ins->ins,
-                                                    th_ins->config);
-                if (!out_flush) {
-                    continue;
+                else {
+                    /* Start the co-routine with the flush callback */
+                    out_flush = flb_output_flush_create(task,
+                                                        task->i_ins,
+                                                        th_ins->ins,
+                                                        th_ins->config);
+                    if (!out_flush) {
+                        continue;
+                    }
+                    flb_coro_resume(out_flush->coro);
                 }
-                flb_coro_resume(out_flush->coro);
             }
             else if (event->type == FLB_ENGINE_EV_CUSTOM) {
                 event->handler(event);
@@ -330,6 +331,7 @@ static void output_thread(void *data)
 
         /* Destroy upstream connections from the 'pending destroy list' */
         flb_upstream_conn_pending_destroy_list(&th_ins->upstreams);
+
         flb_sched_timer_cleanup(sched);
 
         /* Check if we should stop the event loop */