#53: Add separate queue for resumed coroutines to improve event cleanup stability

        The previous clean_events_for_resumed_coroutines algorithm was unstable due to
        complex tracking of circular buffer changes during coroutine additions and
        buffer reallocations. This commit introduces a dedicated resumed_coroutines
        queue that provides a more reliable approach.

        Changes:
        - Add resumed_coroutines circular buffer to module globals
        - Initialize resumed_coroutines queue in PHP_RINIT_FUNCTION
        - Track resumed coroutines in async_scheduler_coroutine_enqueue() and async_coroutine_resume()
        - Replace complex clean_events_for_resumed_coroutines() with simple process_resumed_coroutines()
        - Update scheduler_next_tick() and fiber_entry() to use new queue-based approach
        - Add proper cleanup in both engine_shutdown() and async_scheduler_dtor() to prevent memory leaks

        Benefits:
        - Eliminates race conditions from buffer reallocation tracking
        - Simplifies logic by checking queue contents instead of head pointer changes
        - Improves code readability and maintainability
        - Fixes memory leak of 8192 bytes in circular buffer allocation

Author: EdmondDantes

async.c

@@ -962,6 +962,7 @@ PHP_RINIT_FUNCTION(async) /* {{{ */
 	ZEND_ASYNC_INITIALIZE;
 	circular_buffer_ctor(&ASYNC_G(microtasks), 64, sizeof(zend_async_microtask_t *), &zend_std_allocator);
 	circular_buffer_ctor(&ASYNC_G(coroutine_queue), 128, sizeof(zend_coroutine_t *), &zend_std_allocator);
+	circular_buffer_ctor(&ASYNC_G(resumed_coroutines), 64, sizeof(zend_coroutine_t *), &zend_std_allocator);
 	zend_hash_init(&ASYNC_G(coroutines), 128, NULL, NULL, 0);
 
 	ASYNC_G(reactor_started) = false;

async_API.c

@@ -208,6 +208,7 @@ static void engine_shutdown(void)
 
 	circular_buffer_dtor(&ASYNC_G(microtasks));
 	circular_buffer_dtor(&ASYNC_G(coroutine_queue));
+	circular_buffer_dtor(&ASYNC_G(resumed_coroutines));
 	zend_hash_destroy(&ASYNC_G(coroutines));
 
 	if (ASYNC_G(root_context) != NULL) {

coroutine.c

@@ -700,6 +700,9 @@ void async_coroutine_resume(zend_coroutine_t *coroutine, zend_object *error, con
 	}
 
 	coroutine->waker->status = ZEND_ASYNC_WAKER_QUEUED;
+
+	// Add to resumed_coroutines queue for event cleanup
+	circular_buffer_push(&ASYNC_G(resumed_coroutines), &coroutine, true);
 }
 
 void async_coroutine_cancel(zend_coroutine_t *zend_coroutine,

php_async.h

@@ -73,6 +73,8 @@ ZEND_BEGIN_MODULE_GLOBALS(async)
 circular_buffer_t microtasks;
 /* Queue of coroutine_queue */
 circular_buffer_t coroutine_queue;
+/* Queue of resumed coroutines for event cleanup */
+circular_buffer_t resumed_coroutines;
 /* List of coroutines  */
 HashTable coroutines;
 /* The transfer structure is used to return to the main execution context. */

scheduler.c

@@ -302,34 +302,19 @@ static zend_always_inline void return_to_main(zend_fiber_transfer *transfer)
 /// COROUTINE QUEUE MANAGEMENT
 ///////////////////////////////////////////////////////////
 
-static zend_always_inline void clean_events_for_resumed_coroutines(const circular_buffer_t *queue, const void *previous_data, size_t previous_count, size_t previous_head)
+static zend_always_inline void process_resumed_coroutines(void)
 {
-	const size_t new_head = queue->head;
-	const size_t mask = queue->capacity - 1;
-	const size_t item_size = queue->item_size;
-	size_t current;
-
-	// Check if reallocation occurred
-	if (queue->data != previous_data) {
-		// After reallocation: tail should be 0, old elements at [0, previous_count)
-		ZEND_ASSERT(queue->tail == 0 && "After reallocation tail should be 0");
-		current = previous_count;
-	} else {
-		// No reallocation: use original head
-		current = previous_head;
-	}
-
-	while (current != new_head) {
-		zend_coroutine_t *coroutine = *(zend_coroutine_t**)((char*)queue->data + current * item_size);
+	circular_buffer_t *resumed_queue = &ASYNC_G(resumed_coroutines);
+	zend_coroutine_t *coroutine = NULL;
 
-		if (coroutine != NULL && coroutine->waker != NULL) {
+	while (circular_buffer_pop_ptr(resumed_queue, (void**)&coroutine) == SUCCESS) {
+		if (EXPECTED(coroutine != NULL && coroutine->waker != NULL)) {
 			ZEND_ASYNC_WAKER_CLEAN_EVENTS(coroutine->waker);
 		}
-
-		current = (current + 1) & mask;
 	}
 }
 
+
 static zend_always_inline async_coroutine_t *next_coroutine(void)
 {
 	async_coroutine_t *coroutine;
@@ -705,6 +690,7 @@ static void async_scheduler_dtor(void)
 	OBJ_RELEASE(&async_coroutine->std);
 
 	zval_c_buffer_cleanup(&ASYNC_G(coroutine_queue));
+	zval_c_buffer_cleanup(&ASYNC_G(resumed_coroutines));
 	zval_c_buffer_cleanup(&ASYNC_G(microtasks));
 
 	zval *current;
@@ -1056,6 +1042,9 @@ void async_scheduler_coroutine_enqueue(zend_coroutine_t *coroutine)
 			async_throw_error("Failed to enqueue coroutine");
 		} else {
 			coroutine->waker->status = ZEND_ASYNC_WAKER_QUEUED;
+
+			// Add to resumed_coroutines queue for event cleanup
+			circular_buffer_push_ptr_with_resize(&ASYNC_G(resumed_coroutines), coroutine);
 		}
 
 		//
@@ -1087,14 +1076,10 @@ static zend_always_inline void scheduler_next_tick(void)
 		ASYNC_G(last_reactor_tick) = current_time;
 		const circular_buffer_t * queue = &ASYNC_G(coroutine_queue);
 
-		const void *previous_data = queue->data;
-		const size_t previous_count = circular_buffer_count(queue);
-		const size_t previous_head = queue->head;
-
 		has_handles = ZEND_ASYNC_REACTOR_EXECUTE(circular_buffer_is_not_empty(queue));
 
-		if (previous_head != queue->head) {
-			clean_events_for_resumed_coroutines(queue, previous_data, previous_count, previous_head);
+		if (circular_buffer_is_not_empty(&ASYNC_G(resumed_coroutines))) {
+			process_resumed_coroutines();
 		}
 
 		TRY_HANDLE_SUSPEND_EXCEPTION();
@@ -1317,16 +1302,11 @@ ZEND_STACK_ALIGNED void fiber_entry(zend_fiber_transfer *transfer)
 			execute_microtasks();
 			TRY_HANDLE_EXCEPTION();
 
-			const void *previous_data = coroutine_queue->data;
-			const size_t previous_count = circular_buffer_count(coroutine_queue);
-			const size_t previous_head = coroutine_queue->head;
-
-			has_next_coroutine = previous_count > 0;
-
+			has_next_coroutine = circular_buffer_count(coroutine_queue) > 0;
 			has_handles = ZEND_ASYNC_REACTOR_EXECUTE(has_next_coroutine);
 
-			if (previous_head != coroutine_queue->head) {
-				clean_events_for_resumed_coroutines(coroutine_queue, previous_data, previous_count, previous_head);
+			if (circular_buffer_is_not_empty(&ASYNC_G(resumed_coroutines))) {
+				process_resumed_coroutines();
 			}
 
 			TRY_HANDLE_EXCEPTION();