Skip to content

Commit a379978

Browse files
EdmondDantesEdmondDantes
committed
Merge branch 'true-async-api' into true-async-api-stable
2 parents 080ed45 + 67acaf0 commit a379978

File tree

2 files changed

+206
-27
lines changed

2 files changed

+206
-27
lines changed

‎Zend/zend_async_API.c

Lines changed: 110 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,18 @@ static void shutdown_stub(void) {}
4848

4949
static zend_array* get_coroutines_stub(void) { return NULL; }
5050

51+
static zend_future_t* future_create_stub(bool thread_safe, size_t extra_size)
52+
{
53+
ASYNC_THROW_ERROR("Async API is not enabled");
54+
return NULL;
55+
}
56+
57+
static zend_async_channel_t* channel_create_stub(size_t buffer_size, bool resizable, bool thread_safe, size_t extra_size)
58+
{
59+
ASYNC_THROW_ERROR("Async API is not enabled");
60+
return NULL;
61+
}
62+
5163
static void add_microtask_stub(zend_async_microtask_t *microtask) {}
5264

5365
static zend_array* get_awaiting_info_stub(zend_coroutine_t *coroutine) { return NULL; }
@@ -97,6 +109,8 @@ zend_async_get_coroutines_t zend_async_get_coroutines_fn = get_coroutines_stub;
97109
zend_async_add_microtask_t zend_async_add_microtask_fn = add_microtask_stub;
98110
zend_async_get_awaiting_info_t zend_async_get_awaiting_info_fn = get_awaiting_info_stub;
99111
zend_async_get_class_ce_t zend_async_get_class_ce_fn = get_class_ce;
112+
zend_async_future_create_t zend_async_future_create_fn = future_create_stub;
113+
zend_async_channel_create_t zend_async_channel_create_fn = channel_create_stub;
100114

101115
static zend_atomic_bool reactor_lock = {0};
102116
static char * reactor_module_name = NULL;
@@ -428,14 +442,46 @@ ZEND_API zend_coroutine_event_callback_t * zend_async_coroutine_callback_new(
428442
/* Waker API */
429443
//////////////////////////////////////////////////////////////////////
430444

445+
static zend_always_inline zend_async_waker_trigger_t *waker_trigger_create(zend_async_event_t *event, uint32_t initial_capacity)
446+
{
447+
size_t total_size = sizeof(zend_async_waker_trigger_t) + (initial_capacity - 1) * sizeof(zend_async_event_callback_t *);
448+
zend_async_waker_trigger_t *trigger = (zend_async_waker_trigger_t *)emalloc(total_size);
449+
450+
trigger->length = 0;
451+
trigger->capacity = initial_capacity;
452+
trigger->event = event;
453+
454+
return trigger;
455+
}
456+
457+
static zend_always_inline zend_async_waker_trigger_t *waker_trigger_add_callback(zend_async_waker_trigger_t *trigger, zend_async_event_callback_t *callback)
458+
{
459+
if (trigger->length >= trigger->capacity) {
460+
uint32_t new_capacity = trigger->capacity * 2;
461+
size_t total_size = sizeof(zend_async_waker_trigger_t) + (new_capacity - 1) * sizeof(zend_async_event_callback_t *);
462+
463+
zend_async_waker_trigger_t *new_trigger = (zend_async_waker_trigger_t *)erealloc(trigger, total_size);
464+
new_trigger->capacity = new_capacity;
465+
trigger = new_trigger;
466+
}
467+
468+
trigger->data[trigger->length++] = callback;
469+
return trigger;
470+
}
471+
431472
static void waker_events_dtor(zval *item)
432473
{
433474
zend_async_waker_trigger_t * trigger = Z_PTR_P(item);
434475
zend_async_event_t *event = trigger->event;
435476
trigger->event = NULL;
436477

437478
if (event != NULL) {
438-
event->del_callback(event, trigger->callback);
479+
// Remove all callbacks from the event
480+
for (uint32_t i = 0; i < trigger->length; i++) {
481+
if (trigger->data[i] != NULL) {
482+
event->del_callback(event, trigger->data[i]);
483+
}
484+
}
439485
//
440486
// At this point, we explicitly stop the event because it is no longer being listened to by our handlers.
441487
// However, this does not mean the object is destroyed—it may remain in memory if something still holds a reference to it.
@@ -444,6 +490,7 @@ static void waker_events_dtor(zval *item)
444490
ZEND_ASYNC_EVENT_RELEASE(event);
445491
}
446492

493+
// Free the entire trigger (includes flexible array member)
447494
efree(trigger);
448495
}
449496

@@ -558,11 +605,29 @@ void coroutine_event_callback_dispose(zend_async_event_callback_t *callback, zen
558605
zend_async_waker_t * waker = coroutine->waker;
559606

560607
if (event != NULL && waker != NULL) {
561-
// remove the event from the waker
562-
zend_hash_index_del(&waker->events, (zend_ulong)event);
563-
564-
if (waker->triggered_events != NULL) {
565-
zend_hash_index_del(waker->triggered_events, (zend_ulong)event);
608+
// Find the trigger for this event
609+
zval *trigger_zval = zend_hash_index_find(&waker->events, (zend_ulong)event);
610+
611+
if (trigger_zval != NULL) {
612+
zend_async_waker_trigger_t *trigger = Z_PTR_P(trigger_zval);
613+
614+
// Remove only this specific callback from the trigger
615+
for (uint32_t i = 0; i < trigger->length; i++) {
616+
if (trigger->data[i] == callback) {
617+
// Move last element to current position (O(1) removal)
618+
trigger->data[i] = trigger->data[--trigger->length];
619+
break;
620+
}
621+
}
622+
623+
// If no more callbacks in trigger, remove the entire event
624+
if (trigger->length == 0) {
625+
zend_hash_index_del(&waker->events, (zend_ulong)event);
626+
627+
if (waker->triggered_events != NULL) {
628+
zend_hash_index_del(waker->triggered_events, (zend_ulong)event);
629+
}
630+
}
566631
}
567632
}
568633
}
@@ -603,6 +668,12 @@ ZEND_API void zend_async_resume_when(
603668
zend_coroutine_event_callback_t *event_callback
604669
)
605670
{
671+
ZEND_ASSERT(EG(exception) == NULL && "Cannot resume when there is an active exception in the engine.");
672+
673+
if (UNEXPECTED(EG(exception))) {
674+
return;
675+
}
676+
606677
bool locally_allocated_callback = false;
607678

608679
if (UNEXPECTED(ZEND_ASYNC_EVENT_IS_CLOSED(event))) {
@@ -655,29 +726,45 @@ ZEND_API void zend_async_resume_when(
655726
}
656727

657728
if (EXPECTED(coroutine->waker != NULL)) {
658-
zend_async_waker_trigger_t *trigger = emalloc(sizeof(zend_async_waker_trigger_t));
659-
trigger->event = event;
660-
trigger->callback = &event_callback->base;
729+
zval *trigger_zval = zend_hash_index_find(&coroutine->waker->events, (zend_ulong)event);
730+
zend_async_waker_trigger_t *trigger;
731+
732+
if (UNEXPECTED(trigger_zval != NULL)) {
733+
// Event already exists, add callback to existing trigger
734+
trigger = Z_PTR_P(trigger_zval);
735+
trigger = waker_trigger_add_callback(trigger, &event_callback->base);
736+
// Update the hash table entry with potentially new pointer after realloc
737+
Z_PTR_P(trigger_zval) = trigger;
738+
} else {
739+
// New event, create new trigger
740+
trigger = waker_trigger_create(event, 1);
741+
trigger = waker_trigger_add_callback(trigger, &event_callback->base);
661742

662-
if (UNEXPECTED(zend_hash_index_add_ptr(&coroutine->waker->events, (zend_ulong)event, trigger) == NULL)) {
663-
efree(trigger);
743+
if (UNEXPECTED(zend_hash_index_add_ptr(&coroutine->waker->events, (zend_ulong)event, trigger) == NULL)) {
744+
// This should not happen with new events, but handle gracefully
745+
efree(trigger);
664746

665-
if (locally_allocated_callback) {
666-
event_callback->base.dispose(&event_callback->base, event);
667-
}
747+
event_callback->coroutine = NULL;
748+
event->del_callback(event, &event_callback->base);
668749

669-
if (trans_event) {
670-
event->dispose(event);
671-
}
750+
event_callback->coroutine = NULL;
751+
event->del_callback(event, &event_callback->base);
672752

673-
zend_throw_error(NULL, "Failed to add event to the waker: maybe event already exists");
753+
if (locally_allocated_callback) {
754+
event_callback->base.dispose(&event_callback->base, event);
755+
}
674756

675-
return;
676-
}
677-
}
757+
if (trans_event) {
758+
event->dispose(event);
759+
}
678760

679-
if (false == trans_event) {
680-
ZEND_ASYNC_EVENT_ADD_REF(event);
761+
zend_throw_error(NULL, "Failed to add event to the waker");
762+
} else {
763+
if (false == trans_event) {
764+
ZEND_ASYNC_EVENT_ADD_REF(event);
765+
}
766+
}
767+
}
681768
}
682769
}
683770

‎Zend/zend_async_API.h

Lines changed: 96 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -19,9 +19,9 @@
1919
#include "zend_fibers.h"
2020
#include "zend_globals.h"
2121

22-
#define ZEND_ASYNC_API "TrueAsync API v0.3.0"
22+
#define ZEND_ASYNC_API "TrueAsync API v0.4.0"
2323
#define ZEND_ASYNC_API_VERSION_MAJOR 0
24-
#define ZEND_ASYNC_API_VERSION_MINOR 3
24+
#define ZEND_ASYNC_API_VERSION_MINOR 4
2525
#define ZEND_ASYNC_API_VERSION_PATCH 0
2626

2727
#define ZEND_ASYNC_API_VERSION_NUMBER \
@@ -119,6 +119,16 @@ typedef enum
119119
* zend_coroutine_t is a Basic data structure that represents a coroutine in the Zend Engine.
120120
*/
121121
typedef struct _zend_coroutine_s zend_coroutine_t;
122+
123+
/**
124+
* zend_future_t is a data structure that represents a future result container.
125+
*/
126+
typedef struct _zend_future_s zend_future_t;
127+
128+
/**
129+
* zend_async_channel_t is a data structure that represents a communication channel.
130+
*/
131+
typedef struct _zend_async_channel_s zend_async_channel_t;
122132
typedef struct _zend_async_context_s zend_async_context_t;
123133
typedef struct _zend_async_waker_s zend_async_waker_t;
124134
typedef struct _zend_async_microtask_s zend_async_microtask_t;
@@ -127,6 +137,10 @@ typedef struct _zend_async_iterator_s zend_async_iterator_t;
127137
typedef struct _zend_fcall_s zend_fcall_t;
128138
typedef void (*zend_coroutine_entry_t)(void);
129139

140+
/* Channel method function types */
141+
typedef bool (*zend_channel_send_t)(zend_async_channel_t *channel, zval *value);
142+
typedef bool (*zend_channel_receive_t)(zend_async_channel_t *channel, zval *result);
143+
130144
/* Coroutine Switch Handlers */
131145
typedef struct _zend_coroutine_switch_handler_s zend_coroutine_switch_handler_t;
132146
typedef struct _zend_coroutine_switch_handlers_vector_s zend_coroutine_switch_handlers_vector_t;
@@ -203,6 +217,8 @@ typedef zend_array* (*zend_async_get_coroutines_t)(void);
203217
typedef void (*zend_async_add_microtask_t)(zend_async_microtask_t *microtask);
204218
typedef zend_array* (*zend_async_get_awaiting_info_t)(zend_coroutine_t * coroutine);
205219
typedef zend_class_entry* (*zend_async_get_class_ce_t)(zend_async_class type);
220+
typedef zend_future_t* (*zend_async_future_create_t)(bool thread_safe, size_t extra_size);
221+
typedef zend_async_channel_t* (*zend_async_channel_create_t)(size_t buffer_size, bool resizable, bool thread_safe, size_t extra_size);
206222

207223
typedef void (*zend_async_reactor_startup_t)(void);
208224
typedef void (*zend_async_reactor_shutdown_t)(void);
@@ -402,8 +418,10 @@ struct _zend_coroutine_event_callback_s {
402418
};
403419

404420
struct _zend_async_waker_trigger_s {
405-
zend_async_event_t *event;
406-
zend_async_event_callback_t *callback;
421+
uint32_t length; /* current number of callbacks */
422+
uint32_t capacity; /* allocated slots in the array */
423+
zend_async_event_t *event;
424+
zend_async_event_callback_t *data[1]; /* flexible array member */
407425
};
408426

409427
/* Dynamic array of async event callbacks with single iterator protection */
@@ -1049,6 +1067,70 @@ struct _zend_async_context_s {
10491067
zend_async_context_dispose_t dispose;
10501068
};
10511069

1070+
///////////////////////////////////////////////////////////////
1071+
/// Future
1072+
///////////////////////////////////////////////////////////////
1073+
1074+
/**
1075+
* zend_future_t structure represents a future result container.
1076+
* It inherits from zend_async_event_t to participate in the event system.
1077+
*/
1078+
struct _zend_future_s {
1079+
zend_async_event_t event; /* Event inheritance (first member) */
1080+
zval result; /* Result value */
1081+
zend_object *exception; /* Exception object (NULL = no error) */
1082+
/* Created file and line number */
1083+
uint32_t lineno;
1084+
uint32_t completed_lineno;
1085+
/* Completed file and line number */
1086+
zend_string *filename;
1087+
zend_string *completed_filename;
1088+
};
1089+
1090+
#define ZEND_FUTURE_F_THREAD_SAFE (1u << 10)
1091+
#define ZEND_FUTURE_F_IGNORED (1u << 11)
1092+
1093+
#define ZEND_FUTURE_IS_COMPLETED(future) (((future)->event.flags & ZEND_ASYNC_EVENT_F_CLOSED) != 0)
1094+
1095+
#define ZEND_FUTURE_SET_THREAD_SAFE(future) ((future)->event.flags |= ZEND_FUTURE_F_THREAD_SAFE)
1096+
#define ZEND_FUTURE_SET_IGNORED(future) ((future)->event.flags |= ZEND_FUTURE_F_IGNORED)
1097+
1098+
1099+
#define ZEND_FUTURE_COMPLETE(future, result) do { \
1100+
if(ZEND_ASYNC_EVENT_IS_CLOSED(&(future)->event)) { \
1101+
break; \
1102+
} \
1103+
ZVAL_COPY(&(future)->result, (result)); \
1104+
(future)->event.stop(&(future)->event); \
1105+
} while (0)
1106+
1107+
#define ZEND_FUTURE_REJECT(future, error) do { \
1108+
if(ZEND_ASYNC_EVENT_IS_CLOSED(&(future)->event)) { \
1109+
break; \
1110+
} \
1111+
(future)->exception = error; \
1112+
GC_ADDREF(error); \
1113+
(future)->event.stop(&(future)->event); \
1114+
} while (0)
1115+
1116+
///////////////////////////////////////////////////////////////
1117+
/// Channel
1118+
///////////////////////////////////////////////////////////////
1119+
1120+
/**
1121+
* zend_async_channel_t structure represents a communication channel.
1122+
* It inherits from zend_async_event_t to participate in the event system.
1123+
*/
1124+
struct _zend_async_channel_s {
1125+
zend_async_event_t event; /* Event inheritance (first member) */
1126+
/* Channel-specific method pointers */
1127+
zend_channel_send_t send; /* Send method */
1128+
zend_channel_receive_t receive; /* Receive method */
1129+
};
1130+
1131+
#define ZEND_ASYNC_CHANNEL_F_THREAD_SAFE (1u << 10)
1132+
1133+
10521134
///////////////////////////////////////////////////////////////
10531135
/// Global Macros
10541136
///////////////////////////////////////////////////////////////
@@ -1177,6 +1259,8 @@ ZEND_API extern zend_async_get_coroutines_t zend_async_get_coroutines_fn;
11771259
ZEND_API extern zend_async_add_microtask_t zend_async_add_microtask_fn;
11781260
ZEND_API extern zend_async_get_awaiting_info_t zend_async_get_awaiting_info_fn;
11791261
ZEND_API extern zend_async_get_class_ce_t zend_async_get_class_ce_fn;
1262+
ZEND_API extern zend_async_future_create_t zend_async_future_create_fn;
1263+
ZEND_API extern zend_async_channel_create_t zend_async_channel_create_fn;
11801264

11811265
/* Iterator API */
11821266
ZEND_API extern zend_async_new_iterator_t zend_async_new_iterator_fn;
@@ -1361,6 +1445,14 @@ ZEND_API void zend_async_add_main_coroutine_start_handler(
13611445

13621446
ZEND_API void zend_async_call_main_coroutine_start_handlers(zend_coroutine_t *main_coroutine);
13631447

1448+
/* Future API Functions */
1449+
#define ZEND_ASYNC_NEW_FUTURE(thread_safe) zend_async_future_create_fn(thread_safe, 0)
1450+
#define ZEND_ASYNC_NEW_FUTURE_EX(thread_safe, extra_size) zend_async_future_create_fn(thread_safe, extra_size)
1451+
1452+
/* Channel API Functions */
1453+
#define ZEND_ASYNC_NEW_CHANNEL(buffer_size, resizable, thread_safe) zend_async_channel_create_fn(buffer_size, resizable, thread_safe, 0)
1454+
#define ZEND_ASYNC_NEW_CHANNEL_EX(buffer_size, resizable, thread_safe, extra_size) zend_async_channel_create_fn(buffer_size, resizable, thread_safe, extra_size)
1455+
13641456
END_EXTERN_C()
13651457

13661458
#define ZEND_ASYNC_IS_ENABLED() zend_async_is_enabled()

0 commit comments

Comments
 (0)