Grand dispatch queue

.github/workflows/ci.yml

@@ -158,12 +158,15 @@ jobs:
 
   macos:
     runs-on: macos-14 # latest
+    strategy:
+      matrix:
+        eventloop: ["-DAWS_USE_DISPATCH_QUEUE=ON", "-DAWS_USE_DISPATCH_QUEUE=OFF"]
     steps:
     - name: Build ${{ env.PACKAGE_NAME }} + consumers
       run: |
         python3 -c "from urllib.request import urlretrieve; urlretrieve('${{ env.BUILDER_HOST }}/${{ env.BUILDER_SOURCE }}/${{ env.BUILDER_VERSION }}/builder.pyz?run=${{ env.RUN }}', 'builder')"
         chmod a+x builder
-        ./builder build -p ${{ env.PACKAGE_NAME }}
+        ./builder build -p ${{ env.PACKAGE_NAME }} --cmake-extra=${{ matrix.eventloop }}
 
   macos-x64:
     runs-on: macos-14-large # latest
@@ -176,12 +179,15 @@ jobs:
 
   macos-debug:
     runs-on: macos-14 # latest
+    strategy:
+      matrix:
+        eventloop: ["-DAWS_USE_DISPATCH_QUEUE=ON", "-DAWS_USE_DISPATCH_QUEUE=OFF"]
     steps:
     - name: Build ${{ env.PACKAGE_NAME }} + consumers
       run: |
         python3 -c "from urllib.request import urlretrieve; urlretrieve('${{ env.BUILDER_HOST }}/${{ env.BUILDER_SOURCE }}/${{ env.BUILDER_VERSION }}/builder.pyz?run=${{ env.RUN }}', 'builder')"
         chmod a+x builder
-        ./builder build -p ${{ env.PACKAGE_NAME }} --config Debug
+        ./builder build -p ${{ env.PACKAGE_NAME }} --cmake-extra=${{ matrix.eventloop }} --config Debug
 
   freebsd:
     runs-on: ubuntu-22.04  # latest

CMakeLists.txt

@@ -48,7 +48,6 @@ file(GLOB AWS_IO_TESTING_HEADERS
         "include/aws/testing/*.h"
         )
 
-
 file(GLOB AWS_IO_PRIV_HEADERS
         "include/aws/io/private/*.h"
         )
@@ -79,7 +78,7 @@ if (WIN32)
                 )
          list(APPEND AWS_IO_OS_SRC ${AWS_IO_IOCP_SRC})
 
-         set(EVENT_LOOP_DEFINE "IO_COMPLETION_PORTS")
+         set(EVENT_LOOP_DEFINES "AWS_USE_IO_COMPLETION_PORTS")
     endif ()
 
     if (MSVC)
@@ -106,7 +105,7 @@ elseif (CMAKE_SYSTEM_NAME STREQUAL "Linux" OR CMAKE_SYSTEM_NAME STREQUAL "Androi
             )
     set(PLATFORM_LIBS "")
 
-    set(EVENT_LOOP_DEFINE "EPOLL")
+    set(EVENT_LOOP_DEFINES "-DAWS_USE_EPOLL")
     set(USE_S2N ON)
 
 elseif (APPLE)
@@ -117,17 +116,33 @@ elseif (APPLE)
     file(GLOB AWS_IO_OS_SRC
             "source/bsd/*.c"
             "source/posix/*.c"
-            "source/darwin/*.c"
+            "source/darwin/darwin_pki_utils.c"
+            "source/darwin/secure_transport_tls_channel_handler.c"
             )
 
     find_library(SECURITY_LIB Security)
     if (NOT SECURITY_LIB)
         message(FATAL_ERROR "Security framework not found")
     endif ()
 
+    find_library(NETWORK_LIB Network)
+    if (NOT NETWORK_LIB)
+        message(FATAL_ERROR "Network framework not found")
+    endif ()
+
     #No choice on TLS for apple, darwinssl will always be used.
-    list(APPEND PLATFORM_LIBS "-framework Security")
-    set(EVENT_LOOP_DEFINE "KQUEUE")
+    list(APPEND PLATFORM_LIBS "-framework Security -framework Network")
+
+     if(AWS_USE_DISPATCH_QUEUE OR IOS)
+        set(EVENT_LOOP_DEFINES "-DAWS_USE_DISPATCH_QUEUE" )
+        message("use dispatch queue")
+        file(GLOB AWS_IO_DISPATCH_QUEUE_SRC
+        "source/darwin/dispatch_queue_event_loop.c"
+        )
+        list(APPEND AWS_IO_OS_SRC ${AWS_IO_DISPATCH_QUEUE_SRC})
+     else ()
+         set(EVENT_LOOP_DEFINES "-DAWS_USE_KQUEUE")
+    endif()
 
 elseif (CMAKE_SYSTEM_NAME STREQUAL "FreeBSD" OR CMAKE_SYSTEM_NAME STREQUAL "NetBSD" OR CMAKE_SYSTEM_NAME STREQUAL "OpenBSD")
     file(GLOB AWS_IO_OS_HEADERS
@@ -138,7 +153,7 @@ elseif (CMAKE_SYSTEM_NAME STREQUAL "FreeBSD" OR CMAKE_SYSTEM_NAME STREQUAL "NetB
             "source/posix/*.c"
             )
 
-    set(EVENT_LOOP_DEFINE "KQUEUE")
+    set(EVENT_LOOP_DEFINES "-DAWS_USE_KQUEUE")
     set(USE_S2N ON)
 
 endif()
@@ -191,7 +206,7 @@ aws_add_sanitizers(${PROJECT_NAME})
 # We are not ABI stable yet
 set_target_properties(${PROJECT_NAME} PROPERTIES VERSION 1.0.0)
 
-target_compile_definitions(${PROJECT_NAME} PUBLIC "-DAWS_USE_${EVENT_LOOP_DEFINE}")
+target_compile_definitions(${PROJECT_NAME} PUBLIC "${EVENT_LOOP_DEFINES}")
 
 if (BYO_CRYPTO)
     target_compile_definitions(${PROJECT_NAME} PUBLIC "-DBYO_CRYPTO")

include/aws/io/event_loop.h

@@ -70,15 +70,19 @@ struct aws_overlapped {
     void *user_data;
 };
 
-#else /* !AWS_USE_IO_COMPLETION_PORTS */
+#endif /* AWS_USE_IO_COMPLETION_PORTS */
 
 typedef void(aws_event_loop_on_event_fn)(
     struct aws_event_loop *event_loop,
     struct aws_io_handle *handle,
     int events,
     void *user_data);
 
-#endif /* AWS_USE_IO_COMPLETION_PORTS */
+enum aws_event_loop_style {
+    AWS_EVENT_LOOP_STYLE_UNDEFINED = 0,
+    AWS_EVENT_LOOP_STYLE_POLL_BASED = 1,
+    AWS_EVENT_LOOP_STYLE_COMPLETION_PORT_BASED = 2,
+};
 
 struct aws_event_loop_vtable {
     void (*destroy)(struct aws_event_loop *event_loop);
@@ -88,16 +92,16 @@ struct aws_event_loop_vtable {
     void (*schedule_task_now)(struct aws_event_loop *event_loop, struct aws_task *task);
     void (*schedule_task_future)(struct aws_event_loop *event_loop, struct aws_task *task, uint64_t run_at_nanos);
     void (*cancel_task)(struct aws_event_loop *event_loop, struct aws_task *task);
-#if AWS_USE_IO_COMPLETION_PORTS
-    int (*connect_to_io_completion_port)(struct aws_event_loop *event_loop, struct aws_io_handle *handle);
-#else
-    int (*subscribe_to_io_events)(
-        struct aws_event_loop *event_loop,
-        struct aws_io_handle *handle,
-        int events,
-        aws_event_loop_on_event_fn *on_event,
-        void *user_data);
-#endif
+    union {
+        int (*connect_to_completion_port)(struct aws_event_loop *event_loop, struct aws_io_handle *handle);
+        int (*subscribe_to_io_events)(
+            struct aws_event_loop *event_loop,
+            struct aws_io_handle *handle,
+            int events,
+            aws_event_loop_on_event_fn *on_event,
+            void *user_data);
+    } register_style;
+    enum aws_event_loop_style event_loop_style;
     int (*unsubscribe_from_io_events)(struct aws_event_loop *event_loop, struct aws_io_handle *handle);
     void (*free_io_event_resources)(void *user_data);
     bool (*is_on_callers_thread)(struct aws_event_loop *event_loop);
@@ -140,6 +144,21 @@ struct aws_event_loop_group {
     struct aws_shutdown_callback_options shutdown_options;
 };
 
+typedef struct aws_event_loop *(aws_new_system_event_loop_fn)(struct aws_allocator *alloc,
+                                                              const struct aws_event_loop_options *options);
+
+struct aws_event_loop_configuration {
+    enum aws_event_loop_style style;
+    aws_new_system_event_loop_fn *event_loop_new_fn;
+    const char *name;
+    bool is_default;
+};
+
+struct aws_event_loop_configuration_group {
+    size_t configuration_count;
+    const struct aws_event_loop_configuration *configurations;
+};
+
 AWS_EXTERN_C_BEGIN
 
 #ifdef AWS_USE_IO_COMPLETION_PORTS
@@ -166,6 +185,10 @@ AWS_IO_API
 struct _OVERLAPPED *aws_overlapped_to_windows_overlapped(struct aws_overlapped *overlapped);
 #endif /* AWS_USE_IO_COMPLETION_PORTS */
 
+/* Get available event-loop configurations, this will return each available event-loop implementation for the current
+ * running system */
+AWS_IO_API const struct aws_event_loop_configuration_group *aws_event_loop_get_available_configurations(void);
+
 /**
  * Creates an instance of the default event loop implementation for the current architecture and operating system.
  */
@@ -181,6 +204,38 @@ struct aws_event_loop *aws_event_loop_new_default_with_options(
     struct aws_allocator *alloc,
     const struct aws_event_loop_options *options);
 
+// DEBUG WIP We should expose or condense all these def specific function APIs and not make them
+// defined specific. Consolidation of them should work and branched logic within due to all the
+// arguments being the same. Let's move away from different API based on framework and instead
+// raise an unsupported platform error or simply use branching in implementation.
+#ifdef AWS_USE_IO_COMPLETION_PORTS
+AWS_IO_API
+struct aws_event_loop *aws_event_loop_new_iocp_with_options(
+    struct aws_allocator *alloc,
+    const struct aws_event_loop_options *options);
+#endif /* AWS_USE_IO_COMPLETION_PORTS */
+
+#ifdef AWS_USE_DISPATCH_QUEUE
+AWS_IO_API
+struct aws_event_loop *aws_event_loop_new_dispatch_queue_with_options(
+    struct aws_allocator *alloc,
+    const struct aws_event_loop_options *options);
+#endif /* AWS_USE_DISPATCH_QUEUE */
+
+#ifdef AWS_USE_KQUEUE
+AWS_IO_API
+struct aws_event_loop *aws_event_loop_new_kqueue_with_options(
+    struct aws_allocator *alloc,
+    const struct aws_event_loop_options *options);
+#endif /* AWS_USE_KQUEUE */
+
+#ifdef AWS_USE_EPOLL
+AWS_IO_API
+struct aws_event_loop *aws_event_loop_new_epoll_with_options(
+    struct aws_allocator *alloc,
+    const struct aws_event_loop_options *options);
+#endif /* AWS_USE_EPOLL */
+
 /**
  * Invokes the destroy() fn for the event loop implementation.
  * If the event loop is still in a running state, this function will block waiting on the event loop to shutdown.
@@ -319,8 +374,6 @@ void aws_event_loop_schedule_task_future(
 AWS_IO_API
 void aws_event_loop_cancel_task(struct aws_event_loop *event_loop, struct aws_task *task);
 
-#if AWS_USE_IO_COMPLETION_PORTS
-
 /**
  * Associates an aws_io_handle with the event loop's I/O Completion Port.
  *
@@ -332,11 +385,7 @@ void aws_event_loop_cancel_task(struct aws_event_loop *event_loop, struct aws_ta
  * A handle may only be connected to one event loop in its lifetime.
  */
 AWS_IO_API
-int aws_event_loop_connect_handle_to_io_completion_port(
-    struct aws_event_loop *event_loop,
-    struct aws_io_handle *handle);
-
-#else /* !AWS_USE_IO_COMPLETION_PORTS */
+int aws_event_loop_connect_handle_to_completion_port(struct aws_event_loop *event_loop, struct aws_io_handle *handle);
 
 /**
  * Subscribes on_event to events on the event-loop for handle. events is a bitwise concatenation of the events that were
@@ -353,8 +402,6 @@ int aws_event_loop_subscribe_to_io_events(
     aws_event_loop_on_event_fn *on_event,
     void *user_data);
 
-#endif /* AWS_USE_IO_COMPLETION_PORTS */
-
 /**
  * Unsubscribes handle from event-loop notifications.
  * This function is not thread safe and should be called inside the event-loop's thread.
@@ -399,6 +446,13 @@ struct aws_event_loop_group *aws_event_loop_group_new(
     void *new_loop_user_data,
     const struct aws_shutdown_callback_options *shutdown_options);
 
+AWS_IO_API
+struct aws_event_loop_group *aws_event_loop_group_new_from_config(
+    struct aws_allocator *allocator,
+    const struct aws_event_loop_configuration *config,
+    uint16_t max_threads,
+    const struct aws_shutdown_callback_options *shutdown_options);
+
 /** Creates an event loop group, with clock, number of loops to manage, the function to call for creating a new
  * event loop, and also pins all loops to hw threads on the same cpu_group (e.g. NUMA nodes). Note:
  * If el_count exceeds the number of hw threads in the cpu_group it will be ignored on the assumption that if you
@@ -456,6 +510,9 @@ struct aws_event_loop_group *aws_event_loop_group_acquire(struct aws_event_loop_
 AWS_IO_API
 void aws_event_loop_group_release(struct aws_event_loop_group *el_group);
 
+AWS_IO_API
+enum aws_event_loop_style aws_event_loop_group_get_style(struct aws_event_loop_group *el_group);
+
 AWS_IO_API
 struct aws_event_loop *aws_event_loop_group_get_loop_at(struct aws_event_loop_group *el_group, size_t index);
 

include/aws/io/io.h

@@ -14,12 +14,24 @@ AWS_PUSH_SANE_WARNING_LEVEL
 
 #define AWS_C_IO_PACKAGE_ID 1
 
+struct aws_io_handle;
+
+#ifdef AWS_USE_DISPATCH_QUEUE
+typedef void aws_io_set_queue_on_handle_fn(struct aws_io_handle *handle, void *queue);
+typedef void aws_io_clear_queue_on_handle_fn(struct aws_io_handle *handle);
+#endif /* AWS_USE_DISPATCH_QUEUE */
+
 struct aws_io_handle {
     union {
         int fd;
+        /* on Apple systems, handle is of type nw_connection_t. On Windows, it's a SOCKET handle. */
         void *handle;
     } data;
     void *additional_data;
+#ifdef AWS_USE_DISPATCH_QUEUE
+    aws_io_set_queue_on_handle_fn *set_queue;
+    aws_io_clear_queue_on_handle_fn *clear_queue;
+#endif /* AWS_USE_DISPATCH_QUEUE */
 };
 
 enum aws_io_message_type {

include/aws/io/private/dispatch_queue.h

@@ -0,0 +1,63 @@
+#ifndef AWS_IO_PRIVATE_DISPATCH_QUEUE_H
+#define AWS_IO_PRIVATE_DISPATCH_QUEUE_H
+/**
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#include <Security/Security.h>
+#include <aws/common/mutex.h>
+#include <aws/common/thread.h>
+#include <aws/io/tls_channel_handler.h>
+#include <dispatch/dispatch.h>
+
+struct secure_transport_ctx {
+    struct aws_tls_ctx ctx;
+    CFAllocatorRef wrapped_allocator;
+    CFArrayRef certs;
+    SecIdentityRef secitem_identity;
+    CFArrayRef ca_cert;
+    enum aws_tls_versions minimum_version;
+    struct aws_string *alpn_list;
+    bool verify_peer;
+};
+
+struct dispatch_scheduling_state {
+    // Let's us skip processing an iteration task if one is already in the middle
+    // of executing
+    bool is_executing_iteration;
+
+    // List<scheduled_service_entry> in sorted order by timestamp
+    //
+    // When we go to schedule a new iteration, we check here first to see
+    // if our scheduling attempt is redundant
+    struct aws_linked_list scheduled_services;
+};
+
+struct dispatch_loop {
+    struct aws_allocator *allocator;
+    struct aws_ref_count ref_count;
+    dispatch_queue_t dispatch_queue;
+    struct aws_task_scheduler scheduler;
+    struct aws_linked_list local_cross_thread_tasks;
+
+    // Apple dispatch queue uses the id string to identify the dispatch queue
+    struct aws_string *dispatch_queue_id;
+
+    struct {
+        struct dispatch_scheduling_state scheduling_state;
+        struct aws_linked_list cross_thread_tasks;
+        struct aws_mutex lock;
+        bool suspended;
+        // `is_executing` flag and `current_thread_id` together are used to identify the excuting
+        // thread id for dispatch queue. See `static bool s_is_on_callers_thread(struct aws_event_loop *event_loop)`
+        // for details.
+        bool is_executing;
+        aws_thread_id_t current_thread_id;
+    } synced_data;
+
+    bool wakeup_schedule_needed;
+    bool is_destroying;
+};
+
+#endif /* #ifndef AWS_IO_PRIVATE_DISPATCH_QUEUE_H */

source/bsd/kqueue_event_loop.c

@@ -124,14 +124,15 @@ struct aws_event_loop_vtable s_kqueue_vtable = {
     .wait_for_stop_completion = s_wait_for_stop_completion,
     .schedule_task_now = s_schedule_task_now,
     .schedule_task_future = s_schedule_task_future,
-    .subscribe_to_io_events = s_subscribe_to_io_events,
+    .register_style.subscribe_to_io_events = s_subscribe_to_io_events,
+    .event_loop_style = AWS_EVENT_LOOP_STYLE_POLL_BASED,
     .cancel_task = s_cancel_task,
     .unsubscribe_from_io_events = s_unsubscribe_from_io_events,
     .free_io_event_resources = s_free_io_event_resources,
     .is_on_callers_thread = s_is_event_thread,
 };
 
-struct aws_event_loop *aws_event_loop_new_default_with_options(
+struct aws_event_loop *aws_event_loop_new_kqueue_with_options(
     struct aws_allocator *alloc,
     const struct aws_event_loop_options *options) {
     AWS_ASSERT(alloc);