Grand dispatch queue (#661)

Co-authored-by: Zhihui Xia <[email protected]>
Co-authored-by: Bret Ambrose <[email protected]>
Co-authored-by: Bret Ambrose <[email protected]>
Co-authored-by: Michael Graeb <[email protected]>

Author: 5 people

.github/workflows/ci.yml

@@ -232,7 +232,8 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        eventloop: ["kqueue"] # TODO: Add "dispatch_queue" when apple network framework is implemented.
+        eventloop: ["kqueue", "dispatch_queue"]
+        sanitizers: [",thread", ",address,undefined"]
     steps:
     - uses: aws-actions/configure-aws-credentials@v4
       with:
@@ -242,7 +243,7 @@ jobs:
       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 }} --cmake-extra=-DAWS_USE_APPLE_NETWORK_FRAMEWORK=${{ matrix.eventloop == 'dispatch_queue' && 'ON' || 'OFF' }}
+        ./builder build -p ${{ env.PACKAGE_NAME }} --cmake-extra=-DAWS_USE_APPLE_NETWORK_FRAMEWORK=${{ matrix.eventloop == 'dispatch_queue' && 'ON' || 'OFF' }} --cmake-extra=-DENABLE_SANITIZERS=ON --cmake-extra=-DSANITIZERS="${{ matrix.sanitizers }}"
 
   macos-x64:
     runs-on: macos-14-large # latest
@@ -262,7 +263,8 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        eventloop: ["kqueue"] # TODO: Add "-DAWS_USE_APPLE_NETWORK_FRAMEWORK=ON" when apple network framework is implemented.
+        eventloop: ["kqueue", "dispatch_queue"]
+        sanitizers: [",thread", ",address,undefined"]
     steps:
     - uses: aws-actions/configure-aws-credentials@v4
       with:
@@ -272,8 +274,25 @@ jobs:
       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 }} --cmake-extra=-DAWS_USE_APPLE_NETWORK_FRAMEWORK=${{ matrix.eventloop == 'dispatch_queue' && 'ON' || 'OFF' }} --config Debug
+        ./builder build -p ${{ env.PACKAGE_NAME }} --cmake-extra=-DAWS_USE_APPLE_NETWORK_FRAMEWORK=${{ matrix.eventloop == 'dispatch_queue' && 'ON' || 'OFF' }} --cmake-extra=-DENABLE_SANITIZERS=ON --cmake-extra=-DSANITIZERS="${{ matrix.sanitizers }}" --config Debug
 
+  macos-secitem:
+    runs-on: macos-14 # latest
+    strategy:
+      fail-fast: false
+      matrix:
+        sanitizers: [",thread", ",address,undefined"]
+    steps:
+    - uses: aws-actions/configure-aws-credentials@v4
+      with:
+        role-to-assume: ${{ env.CRT_CI_ROLE }}
+        aws-region: ${{ env.AWS_DEFAULT_REGION }}
+    - 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 }} --cmake-extra=-DAWS_USE_SECITEM=ON --cmake-extra=-DAWS_USE_APPLE_NETWORK_FRAMEWORK=ON --cmake-extra=-DENABLE_SANITIZERS=ON --cmake-extra=-DSANITIZERS="${{ matrix.sanitizers }}"
+  
   freebsd:
     runs-on: ubuntu-24.04  # latest
     steps:

.github/workflows/proof-alarm.yml

@@ -16,7 +16,7 @@ jobs:
     - name: Check
       run: |
         TMPFILE=$(mktemp)
-        echo "fb906f599051ed940f141b7d11de0db1 source/linux/epoll_event_loop.c" > $TMPFILE
+        echo "e857a2e5f72ab77a94e56372d89abf99 source/linux/epoll_event_loop.c" > $TMPFILE
         md5sum --check $TMPFILE
 
     # No further steps if successful

CMakeLists.txt

@@ -110,8 +110,10 @@ elseif (APPLE)
         list(APPEND EVENT_LOOP_DEFINES "DISPATCH_QUEUE")
     endif ()
 
-   # Enable KQUEUE on APPLE platforms
-    list(APPEND EVENT_LOOP_DEFINES "KQUEUE")
+    # Enable KQUEUE on MacOS only if AWS_USE_SECITEM is not declared. SecItem requires Dispatch Queue.
+    if (${CMAKE_SYSTEM_NAME} MATCHES "Darwin" AND NOT DEFINED AWS_USE_SECITEM)
+        list(APPEND EVENT_LOOP_DEFINES "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
@@ -182,6 +184,10 @@ foreach(EVENT_LOOP_DEFINE IN LISTS EVENT_LOOP_DEFINES)
     target_compile_definitions(${PROJECT_NAME} PUBLIC "-DAWS_ENABLE_${EVENT_LOOP_DEFINE}")
 endforeach()
 
+if (AWS_USE_SECITEM)
+    target_compile_definitions(${PROJECT_NAME} PUBLIC "-DAWS_USE_SECITEM")
+endif()
+
 if (BYO_CRYPTO)
     target_compile_definitions(${PROJECT_NAME} PUBLIC "-DBYO_CRYPTO")
 endif()
@@ -202,6 +208,10 @@ if (AWS_USE_APPLE_NETWORK_FRAMEWORK)
     target_compile_definitions(${PROJECT_NAME} PUBLIC "-DAWS_USE_APPLE_NETWORK_FRAMEWORK")
 endif()
 
+if (AWS_USE_APPLE_DISPATCH_QUEUE)
+    target_compile_definitions(${PROJECT_NAME} PUBLIC "-DAWS_USE_APPLE_DISPATCH_QUEUE")
+endif()
+
 target_include_directories(${PROJECT_NAME} PUBLIC
         $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/include>
         $<INSTALL_INTERFACE:include>)

README.md

@@ -151,7 +151,8 @@ Core to Async-IO is the event-loop. We provide an implementation for most platfo
 Platform | Implementation
 --- | ---
 Linux | Edge-Triggered Epoll
-BSD Variants and Apple Devices | KQueue
+BSD Variants | KQueue
+Apple Devices | KQueue or Apple Dispatch Queue
 Windows | IOCP (IO Completion Ports)
 
 Also, you can always implement your own as well.
@@ -645,7 +646,7 @@ All exported functions, simply shim into the v-table and return.
 
 We include a cross-platform API for sockets. We support TCP and UDP using IPv4 and IPv6, and Unix Domain sockets. On Windows,
 we use Named Pipes to support the functionality of Unix Domain sockets. On Windows, this is implemented with winsock2, and on
-all unix platforms we use the posix API.
+all unix platforms we use the posix API. We also provides options to use Apple Network Framework on Apple.
 
 Upon a connection being established, the new socket (either as the result of a `connect()` or `start_accept()` call)
 will not be attached to any event loops. It is your responsibility to register it with an event loop to begin receiving
@@ -715,47 +716,53 @@ upon completion of asynchronous operations. If you are using UDP or LOCAL, `conn
 
 Shuts down any pending operations on the socket, and cleans up state. The socket object can be re initialized after this operation.
 
-    int aws_socket_connect(struct aws_socket *socket, struct aws_socket_endpoint *remote_endpoint);
+    int aws_socket_set_cleanup_complete_callback(struct aws_socket *socket, aws_socket_on_shutdown_complete_fn fn, void *user_data);
 
-Connects to a remote endpoint. In UDP, this simply binds the socket to a remote address for use with `aws_socket_write()`,
-and if the operation is successful, the socket can immediately be used for write operations.
+Sets the clean up completion callback. The callback will be invoked if `aws_socket_clean_up()` finish to clean up the socket resources. It is safe to release the socket memory after this callback is invoked.
 
-In TCP, this will function will not block. If the return value is successful, then you must wait on the `on_connection_established()`
-callback to be invoked before using the socket.
+    int aws_socket_connect(struct aws_socket *socket, const struct aws_socket_endpoint *remote_endpoint, struct aws_event_loop *event_loop, aws_socket_on_connection_result_fn *on_connection_result, void *user_data);
+
+Connects to a remote endpoint. In TCP and all Apple Network Framework connections (regardless it is UDP, TCP or LOCAL), when the connection succeed, you still must wait on  the `on_connection_result()` callback to be invoked before using the socket.
+
+In UDP, this simply binds the socket to a remote address for use with `aws_socket_write()`, and if the operation is successful, 
+the socket can immediately be used for write operations.
 
 For LOCAL (Unix Domain Sockets or Named Pipes), the socket will be immediately ready for use upon a successful return.
 
     int aws_socket_bind(struct aws_socket *socket, struct aws_socket_endpoint *local_endpoint);
 
-Binds the socket to a local address. In UDP mode, the socket is ready for `aws_socket_read()` operations. In connection oriented
-modes, you still must call `aws_socket_listen()` and `aws_socket_start_accept()` before using the socket.
+Binds the socket to a local address. In UDP mode, the socket is ready for `aws_socket_read()` operations. In connection oriented 
+modes or if you are using Apple Network Framework (regardless it is UDP or TCP), you still must call `aws_socket_listen()` and 
+`aws_socket_start_accept()` before using the socket.
 
     int aws_socket_listen(struct aws_socket *socket, int backlog_size);
 
-TCP and LOCAL only. Sets up the socket to listen on the address bound to in `aws_socket_bind()`.
+TCP, LOCAL, and Apple Network Framework only. Sets up the socket to listen on the address bound to in `aws_socket_bind()`.
 
-    int aws_socket_start_accept(struct aws_socket *socket);
+    int aws_socket_start_accept(struct aws_socket *socket, struct aws_event_loop *accept_loop, struct aws_socket_listener_options options);
 
-TCP and LOCAL only. The socket will begin accepting new connections. This is an asynchronous operation. New connections will
-arrive via the `on_incoming_connection()` callback.
+TCP, LOCAL, and Apple Network Framework only. The socket will begin accepting new connections. This is an asynchronous operation. `on_accept_start()` will be invoked when the listener is ready to accept new connection. New connections will arrive via the `on_accept_result()` callback. 
 
     int aws_socket_stop_accept(struct aws_socket *socket);
 
-TCP and LOCAL only. The socket will shutdown the listener. It is safe to call `aws_socket_start_accept()` again after this
-operation.
+TCP, LOCAL, and Apple Network Framework only. The socket will shutdown the listener. It is safe to call `aws_socket_start_accept()` 
+again after this operation.
 
     int aws_socket_close(struct aws_socket *socket);
 
 Calls `close()` on the socket and unregisters all io operations from the event loop.
 
+    int aws_socket_set_close_complete_callback(struct aws_socket *socket, aws_socket_on_shutdown_complete_fn fn, void *user_data);
+
+Sets the close completion callback. The callback will be invoked if `aws_socket_close()` finish to process all the I/O events and close the socket.
+
     struct aws_io_handle *aws_socket_get_io_handle(struct aws_socket *socket);
 
 Fetches the underlying io handle for use in event loop registrations and channel handlers.
 
     int aws_socket_set_options(struct aws_socket *socket, struct aws_socket_options *options);
 
-Sets new socket options on the underlying socket. This is mainly useful in context of accepting a new connection via:
-`on_incoming_connection()`.
+Sets new socket options on the underlying socket.
 
     int aws_socket_read(struct aws_socket *socket, struct aws_byte_buf *buffer, size_t *amount_read);
 

include/aws/io/channel_bootstrap.h

@@ -132,6 +132,14 @@ typedef void(aws_server_bootstrap_on_accept_channel_shutdown_fn)(
     struct aws_channel *channel,
     void *user_data);
 
+/**
+ * This function is only used for async listener (Apple Network Framework in this case).
+ * Once the server listener socket is finished setup and starting listening, this fuction
+ * will be invoked.
+ */
+typedef void(
+    aws_server_bootstrap_on_listener_setup_fn)(struct aws_server_bootstrap *bootstrap, int error_code, void *user_data);
+
 /**
  * Once the server listener socket is finished destroying, and all the existing connections are closed, this fuction
  * will be invoked.
@@ -210,6 +218,7 @@ struct aws_server_socket_channel_bootstrap_options {
     uint32_t port;
     const struct aws_socket_options *socket_options;
     const struct aws_tls_connection_options *tls_options;
+    aws_server_bootstrap_on_listener_setup_fn *setup_callback;
     aws_server_bootstrap_on_accept_channel_setup_fn *incoming_callback;
     aws_server_bootstrap_on_accept_channel_shutdown_fn *shutdown_callback;
     aws_server_bootstrap_on_server_listener_destroy_fn *destroy_callback;
@@ -288,6 +297,10 @@ AWS_IO_API int aws_server_bootstrap_set_alpn_callback(
  * shutting down. Immediately after the `shutdown_callback` returns, the channel is cleaned up automatically. All
  * callbacks are invoked the thread of the event-loop that the listening socket is assigned to
  *
+ * `setup_callback`. If set, the callback will be asynchronously invoked when the listener is ready for use. For Apple
+ * Network Framework, the listener is not usable until the callback is invoked. If the listener creation failed
+ * (return NULL), the `setup_callback` will not be invoked.
+ *
  * Upon shutdown of your application, you'll want to call `aws_server_bootstrap_destroy_socket_listener` with the return
  * value from this function.
  *

include/aws/io/event_loop.h

@@ -29,7 +29,8 @@ typedef void(aws_event_loop_on_event_fn)(
  * @internal
  */
 struct aws_event_loop_vtable {
-    void (*destroy)(struct aws_event_loop *event_loop);
+    void (*start_destroy)(struct aws_event_loop *event_loop);
+    void (*complete_destroy)(struct aws_event_loop *event_loop);
     int (*run)(struct aws_event_loop *event_loop);
     int (*stop)(struct aws_event_loop *event_loop);
     int (*wait_for_stop_completion)(struct aws_event_loop *event_loop);
@@ -45,6 +46,7 @@ struct aws_event_loop_vtable {
         void *user_data);
     int (*unsubscribe_from_io_events)(struct aws_event_loop *event_loop, struct aws_io_handle *handle);
     void (*free_io_event_resources)(void *user_data);
+    void *(*get_base_event_loop_group)(struct aws_event_loop *event_loop);
     bool (*is_on_callers_thread)(struct aws_event_loop *event_loop);
 };
 
@@ -245,15 +247,34 @@ void aws_event_loop_clean_up_base(struct aws_event_loop *event_loop);
 /**
  * @internal - Don't use outside of testing.
  *
- * Invokes the destroy() fn for the event loop implementation.
+ * Destroys an event loop implementation.
  * If the event loop is still in a running state, this function will block waiting on the event loop to shutdown.
- * If you do not want this function to block, call aws_event_loop_stop() manually first.
  * If the event loop is shared by multiple threads then destroy must be called by exactly one thread. All other threads
  * must ensure their API calls to the event loop happen-before the call to destroy.
+ *
+ * Internally, this calls aws_event_loop_start_destroy() followed by aws_event_loop_complete_destroy()
  */
 AWS_IO_API
 void aws_event_loop_destroy(struct aws_event_loop *event_loop);
 
+/**
+ * @internal
+ *
+ * Signals an event loop to begin its destruction process.  If an event loop's implementation of this API does anything,
+ * it must be quick and non-blocking.  Most event loop implementations have an empty implementation for this function.
+ */
+AWS_IO_API
+void aws_event_loop_start_destroy(struct aws_event_loop *event_loop);
+
+/**
+ * @internal
+ *
+ * Waits for an event loop to complete its destruction process.  aws_event_loop_start_destroy() must have been called
+ * previously for this function to not deadlock.
+ */
+AWS_IO_API
+void aws_event_loop_complete_destroy(struct aws_event_loop *event_loop);
+
 AWS_EXTERN_C_END
 
 AWS_POP_SANE_WARNING_LEVEL

include/aws/io/io.h

@@ -14,12 +14,17 @@ AWS_PUSH_SANE_WARNING_LEVEL
 
 #define AWS_C_IO_PACKAGE_ID 1
 
+struct aws_io_handle;
+typedef void aws_io_set_queue_on_handle_fn(struct aws_io_handle *handle, void *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;
+    aws_io_set_queue_on_handle_fn *set_queue;
 };
 
 enum aws_io_message_type {
@@ -94,13 +99,6 @@ enum aws_io_errors {
     AWS_IO_CHANNEL_READ_WOULD_EXCEED_WINDOW,
     AWS_IO_EVENT_LOOP_ALREADY_ASSIGNED,
     AWS_IO_EVENT_LOOP_SHUTDOWN,
-    AWS_IO_TLS_ERROR_NEGOTIATION_FAILURE,
-    AWS_IO_TLS_ERROR_NOT_NEGOTIATED,
-    AWS_IO_TLS_ERROR_WRITE_FAILURE,
-    AWS_IO_TLS_ERROR_ALERT_RECEIVED,
-    AWS_IO_TLS_CTX_ERROR,
-    AWS_IO_TLS_VERSION_UNSUPPORTED,
-    AWS_IO_TLS_CIPHER_PREF_UNSUPPORTED,
     AWS_IO_MISSING_ALPN_MESSAGE,
     AWS_IO_UNHANDLED_ALPN_PROTOCOL_MESSAGE,
     AWS_IO_FILE_VALIDATION_FAILURE,
@@ -123,6 +121,7 @@ enum aws_io_errors {
     AWS_IO_SOCKET_INVALID_ADDRESS,
     AWS_IO_SOCKET_ILLEGAL_OPERATION_FOR_STATE,
     AWS_IO_SOCKET_CONNECT_ABORTED,
+    AWS_IO_SOCKET_MISSING_EVENT_LOOP,
     AWS_IO_DNS_QUERY_FAILED,
     AWS_IO_DNS_INVALID_NAME,
     AWS_IO_DNS_NO_ADDRESS_FOR_HOST,
@@ -132,12 +131,35 @@ enum aws_io_errors {
     DEPRECATED_AWS_IO_INVALID_FILE_HANDLE,
     AWS_IO_SHARED_LIBRARY_LOAD_FAILURE,
     AWS_IO_SHARED_LIBRARY_FIND_SYMBOL_FAILURE,
-    AWS_IO_TLS_NEGOTIATION_TIMEOUT,
-    AWS_IO_TLS_ALERT_NOT_GRACEFUL,
     AWS_IO_MAX_RETRIES_EXCEEDED,
     AWS_IO_RETRY_PERMISSION_DENIED,
+
+    AWS_IO_TLS_ERROR_NEGOTIATION_FAILURE,
+    AWS_IO_TLS_ERROR_NOT_NEGOTIATED,
+    AWS_IO_TLS_ERROR_WRITE_FAILURE,
+    AWS_IO_TLS_ERROR_ALERT_RECEIVED,
+    AWS_IO_TLS_CTX_ERROR,
+    AWS_IO_TLS_VERSION_UNSUPPORTED,
+    AWS_IO_TLS_CIPHER_PREF_UNSUPPORTED,
+    AWS_IO_TLS_NEGOTIATION_TIMEOUT,
+    AWS_IO_TLS_ALERT_NOT_GRACEFUL,
     AWS_IO_TLS_DIGEST_ALGORITHM_UNSUPPORTED,
     AWS_IO_TLS_SIGNATURE_ALGORITHM_UNSUPPORTED,
+    AWS_IO_TLS_ERROR_READ_FAILURE,
+    AWS_IO_TLS_UNKNOWN_ROOT_CERTIFICATE,
+    AWS_IO_TLS_NO_ROOT_CERTIFICATE_FOUND,
+    AWS_IO_TLS_CERTIFICATE_EXPIRED,
+    AWS_IO_TLS_CERTIFICATE_NOT_YET_VALID,
+    AWS_IO_TLS_BAD_CERTIFICATE,
+    AWS_IO_TLS_PEER_CERTIFICATE_EXPIRED,
+    AWS_IO_TLS_BAD_PEER_CERTIFICATE,
+    AWS_IO_TLS_PEER_CERTIFICATE_REVOKED,
+    AWS_IO_TLS_PEER_CERTIFICATE_UNKNOWN,
+    AWS_IO_TLS_INTERNAL_ERROR,
+    AWS_IO_TLS_CLOSED_GRACEFUL,
+    AWS_IO_TLS_CLOSED_ABORT,
+    AWS_IO_TLS_INVALID_CERTIFICATE_CHAIN,
+    AWS_IO_TLS_HOST_NAME_MISSMATCH,
 
     AWS_ERROR_PKCS11_VERSION_UNSUPPORTED,
     AWS_ERROR_PKCS11_TOKEN_NOT_FOUND,
@@ -250,8 +272,6 @@ enum aws_io_errors {
     AWS_IO_STREAM_SEEK_UNSUPPORTED,
     AWS_IO_STREAM_GET_LENGTH_UNSUPPORTED,
 
-    AWS_IO_TLS_ERROR_READ_FAILURE,
-
     AWS_ERROR_PEM_MALFORMED,
 
     AWS_IO_ERROR_END_RANGE = AWS_ERROR_ENUM_END_RANGE(AWS_C_IO_PACKAGE_ID),