Acquire/Release Event Loop

.github/workflows/proof-alarm.yml

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

include/aws/io/event_loop.h

@@ -46,7 +46,6 @@ 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);
 };
 

include/aws/io/private/event_loop_impl.h

@@ -75,6 +75,7 @@ struct aws_event_loop {
     uint64_t latest_tick_start;
     size_t current_tick_latency_sum;
     struct aws_atomic_var next_flush_time;
+    struct aws_event_loop_group *base_elg;
     void *impl_data;
 };
 
@@ -322,12 +323,18 @@ AWS_IO_API
 void aws_event_loop_free_io_event_resources(struct aws_event_loop *event_loop, struct aws_io_handle *handle);
 
 /**
- * Retrieves the aws_event_loop_group that is the parent of the aws_event_loop. This is only supported when using a
- * dispatch queue event loop as they are async and their sockets need to retain a refcount on the elg to keep it alive
- * and insure it has not been asyncronously destroyed before anything that needs it.
+ * "acquire a reference to the event loop" by acquiring a reference to the base event loop group it belongs to.
+ * This is used to ensure that the event loop alive. Return null if the event loop does not belong to a group,
+ * otherwise, return the event loop.
  */
 AWS_IO_API
-void *get_base_event_loop_group(struct aws_event_loop *event_loop);
+struct aws_event_loop *aws_event_loop_acquire(struct aws_event_loop *event_loop);
+
+/**
+ * "release a reference to the event loop" by releasing a reference to the base event loop group it belongs to.
+ */
+AWS_IO_API
+void aws_event_loop_release(struct aws_event_loop *event_loop);
 
 AWS_IO_API
 struct aws_event_loop_group *aws_event_loop_group_new_internal(

source/bsd/kqueue_event_loop.c

@@ -49,15 +49,7 @@ static int s_subscribe_to_io_events(
     void *user_data);
 static int s_unsubscribe_from_io_events(struct aws_event_loop *event_loop, struct aws_io_handle *handle);
 static void s_free_io_event_resources(void *user_data);
-static void *s_get_base_event_loop_group(struct aws_event_loop *event_loop) {
-    (void)event_loop;
-    AWS_LOGF_ERROR(
-        AWS_LS_IO_EVENT_LOOP,
-        "id=%p: get_base_event_loop_group() is not supported using KQueue Event Loops",
-        (void *)event_loop);
-    aws_raise_error(AWS_ERROR_PLATFORM_NOT_SUPPORTED);
-    return NULL;
-}
+
 static bool s_is_event_thread(struct aws_event_loop *event_loop);
 
 static void aws_event_loop_thread(void *user_data);
@@ -148,7 +140,6 @@ struct aws_event_loop_vtable s_kqueue_vtable = {
     .subscribe_to_io_events = s_subscribe_to_io_events,
     .unsubscribe_from_io_events = s_unsubscribe_from_io_events,
     .free_io_event_resources = s_free_io_event_resources,
-    .get_base_event_loop_group = s_get_base_event_loop_group,
     .is_on_callers_thread = s_is_event_thread,
 };
 
@@ -275,6 +266,8 @@ struct aws_event_loop *aws_event_loop_new_with_kqueue(
 
     event_loop->vtable = &s_kqueue_vtable;
 
+    event_loop->base_elg = options->parent_elg;
+
     /* success */
     return event_loop;
 

source/darwin/dispatch_queue_event_loop.c

@@ -64,7 +64,6 @@ static void s_free_io_event_resources(void *user_data) {
     /* No io event resources to free */
     (void)user_data;
 }
-static void *s_get_base_event_loop_group(struct aws_event_loop *event_loop);
 static bool s_is_on_callers_thread(struct aws_event_loop *event_loop);
 
 static struct aws_event_loop_vtable s_vtable = {
@@ -80,7 +79,6 @@ static struct aws_event_loop_vtable s_vtable = {
     .subscribe_to_io_events = s_subscribe_to_io_events,
     .unsubscribe_from_io_events = s_unsubscribe_from_io_events,
     .free_io_event_resources = s_free_io_event_resources,
-    .get_base_event_loop_group = s_get_base_event_loop_group,
     .is_on_callers_thread = s_is_on_callers_thread,
 };
 
@@ -254,7 +252,7 @@ struct aws_event_loop *aws_event_loop_new_with_dispatch_queue(
     dispatch_loop->allocator = alloc;
     loop->impl_data = dispatch_loop;
     dispatch_loop->base_loop = loop;
-    dispatch_loop->base_elg = options->parent_elg;
+    dispatch_loop->base_loop->base_elg = options->parent_elg;
     dispatch_loop->synced_data.execution_state = AWS_DLES_SUSPENDED;
     aws_ref_count_init(&dispatch_loop->ref_count, dispatch_loop, s_dispatch_event_loop_on_zero_ref_count);
 
@@ -738,17 +736,6 @@ static int s_connect_to_io_completion_port(struct aws_event_loop *event_loop, st
     return AWS_OP_SUCCESS;
 }
 
-/*
- * Because dispatch queue is async we may need to acquire a refcount of the parent event loop group to prevent
- * the event loop or dispatch loop from being cleaned out from underneath something that needs it. We expose the
- * base elg so anything that needs to insure the event loops and dispatch loops don't get prematurely cleaned can
- * hold a refcount.
- */
-static void *s_get_base_event_loop_group(struct aws_event_loop *event_loop) {
-    struct aws_dispatch_loop *dispatch_loop = event_loop->impl_data;
-    return dispatch_loop->base_elg;
-}
-
 /*
  * We use aws_thread_id_equal with syched_data.current_thread_id and synced_data.is_executing to determine
  * if operation is being executed on the same dispatch queue thread.

source/darwin/dispatch_queue_event_loop_private.h

@@ -35,7 +35,6 @@ struct aws_dispatch_loop {
     dispatch_queue_t dispatch_queue;
     struct aws_task_scheduler scheduler;
     struct aws_event_loop *base_loop;
-    struct aws_event_loop_group *base_elg;
 
     struct aws_ref_count ref_count;
 

source/darwin/nw_socket.c

@@ -420,11 +420,16 @@ static void s_set_event_loop(struct aws_socket *aws_socket, struct aws_event_loo
     nw_socket->event_loop = event_loop;
 
     AWS_LOGF_DEBUG(
-        AWS_LS_IO_SOCKET,
-        "id=%p nw_socket=%p: s_set_event_loop: socket acquire event loop group.",
-        (void *)aws_socket,
-        (void *)nw_socket);
-    aws_event_loop_group_acquire(get_base_event_loop_group(event_loop));
+        AWS_LS_IO_SOCKET, "id=%p nw_socket=%p: nw_socket set event loop.", (void *)aws_socket, (void *)nw_socket);
+
+    if (!aws_event_loop_acquire(event_loop)) {
+        AWS_LOGF_ERROR(
+            AWS_LS_IO_SOCKET,
+            "id=%p nw_socket=%p: failed to acquire event loop group.",
+            (void *)aws_socket,
+            (void *)nw_socket);
+        AWS_FATAL_ASSERT(0);
+    }
 }
 
 static void s_release_event_loop(struct nw_socket *nw_socket) {
@@ -433,7 +438,7 @@ static void s_release_event_loop(struct nw_socket *nw_socket) {
             AWS_LS_IO_SOCKET, "nw_socket=%p: s_release_event_loop: socket has not event loop.", (void *)nw_socket);
         return;
     }
-    aws_event_loop_group_release(get_base_event_loop_group(nw_socket->event_loop));
+    aws_event_loop_release(nw_socket->event_loop);
     AWS_LOGF_DEBUG(
         AWS_LS_IO_SOCKET, "nw_socket=%p: s_release_event_loop: socket release event loop group.", (void *)nw_socket);
     nw_socket->event_loop = NULL;

source/event_loop.c

@@ -655,9 +655,15 @@ void aws_event_loop_free_io_event_resources(struct aws_event_loop *event_loop, s
     event_loop->vtable->free_io_event_resources(handle->additional_data);
 }
 
-void *get_base_event_loop_group(struct aws_event_loop *event_loop) {
-    AWS_ASSERT(event_loop && event_loop->vtable->get_base_event_loop_group);
-    return event_loop->vtable->get_base_event_loop_group(event_loop);
+struct aws_event_loop *aws_event_loop_acquire(struct aws_event_loop *event_loop) {
+    AWS_ASSERT(event_loop->base_elg);
+    aws_event_loop_group_acquire(event_loop->base_elg);
+    return event_loop;
+}
+
+void aws_event_loop_release(struct aws_event_loop *event_loop) {
+    AWS_ASSERT(event_loop->base_elg);
+    aws_event_loop_group_release(event_loop->base_elg);
 }
 
 bool aws_event_loop_thread_is_callers_thread(struct aws_event_loop *event_loop) {

source/linux/epoll_event_loop.c

@@ -68,15 +68,6 @@ static int s_subscribe_to_io_events(
     void *user_data);
 static int s_unsubscribe_from_io_events(struct aws_event_loop *event_loop, struct aws_io_handle *handle);
 static void s_free_io_event_resources(void *user_data);
-static void *s_get_base_event_loop_group(struct aws_event_loop *event_loop) {
-    (void)event_loop;
-    AWS_LOGF_ERROR(
-        AWS_LS_IO_EVENT_LOOP,
-        "id=%p: get_base_event_loop_group() is not supported using Epoll Event Loops",
-        (void *)event_loop);
-    aws_raise_error(AWS_ERROR_PLATFORM_NOT_SUPPORTED);
-    return NULL;
-}
 static bool s_is_on_callers_thread(struct aws_event_loop *event_loop);
 
 static void aws_event_loop_thread(void *args);
@@ -94,7 +85,6 @@ static struct aws_event_loop_vtable s_vtable = {
     .subscribe_to_io_events = s_subscribe_to_io_events,
     .unsubscribe_from_io_events = s_unsubscribe_from_io_events,
     .free_io_event_resources = s_free_io_event_resources,
-    .get_base_event_loop_group = s_get_base_event_loop_group,
     .is_on_callers_thread = s_is_on_callers_thread,
 };
 
@@ -218,6 +208,7 @@ struct aws_event_loop *aws_event_loop_new_with_epoll(
 
     loop->impl_data = epoll_loop;
     loop->vtable = &s_vtable;
+    loop->base_elg = options->parent_elg;
 
     return loop;
 

source/windows/iocp/iocp_event_loop.c

@@ -124,15 +124,6 @@ static int s_subscribe_to_io_events(
 }
 static int s_unsubscribe_from_io_events(struct aws_event_loop *event_loop, struct aws_io_handle *handle);
 static void s_free_io_event_resources(void *user_data);
-static void *s_get_base_event_loop_group(struct aws_event_loop *event_loop) {
-    (void)event_loop;
-    AWS_LOGF_ERROR(
-        AWS_LS_IO_EVENT_LOOP,
-        "id=%p: get_base_event_loop_group() is not supported using IOCP Event Loops",
-        (void *)event_loop);
-    aws_raise_error(AWS_ERROR_PLATFORM_NOT_SUPPORTED);
-    return NULL;
-}
 static void aws_event_loop_thread(void *user_data);
 
 void aws_overlapped_init(
@@ -169,7 +160,6 @@ struct aws_event_loop_vtable s_iocp_vtable = {
     .subscribe_to_io_events = s_subscribe_to_io_events,
     .unsubscribe_from_io_events = s_unsubscribe_from_io_events,
     .free_io_event_resources = s_free_io_event_resources,
-    .get_base_event_loop_group = s_get_base_event_loop_group,
     .is_on_callers_thread = s_is_event_thread,
 };
 
@@ -273,6 +263,8 @@ struct aws_event_loop *aws_event_loop_new_with_iocp(
 
     event_loop->vtable = &s_iocp_vtable;
 
+    event_loop->base_elg = options->parent_elg;
+
     return event_loop;
 
 clean_up:

tests/vcc/new_destroy.c

@@ -184,6 +184,7 @@ struct aws_event_loop *aws_event_loop_new_with_epoll(
 
     loop->impl_data = epoll_loop;
     loop->vtable = &s_vtable;
+    loop->base_elg = options->parent_elg;
 
     _(wrap(&epoll_loop->task_pre_queue.head))
     _(wrap(&epoll_loop->task_pre_queue.tail))