async: scheduler slots are process-wide, registered once

The C slots are set once per process; per-thread state lives in the
bridge's handlers container. The bridge no longer unregisters at RSHUTDOWN
(it was yanking the slots from under other ZTS threads mid-request) and
skips slot registration when its own module already owns them, so a later
request just re-binds its handlers and launches.

Core: register/unregister run under a mutex (two threads' first
registrations raced the check-then-set), and the module name is copied
into process memory - the bridge was passing a request-local string.

Author: EdmondDantes

Zend/zend_async_API.c

@@ -35,8 +35,12 @@ static void internal_globals_dtor(zend_async_globals_t *globals)
 	(void) globals;
 }
 
+static MUTEX_T scheduler_mutex = NULL;
+
 void zend_async_globals_ctor(void)
 {
+	scheduler_mutex = tsrm_mutex_alloc();
+
 #ifdef ZTS
 	ts_allocate_fast_id(&zend_async_globals_id, &zend_async_globals_offset,
 			sizeof(zend_async_globals_t), (ts_allocate_ctor) internal_globals_ctor,
@@ -506,12 +510,17 @@ ZEND_API bool zend_async_scheduler_register(
 		return false;
 	}
 
+	tsrm_mutex_lock(scheduler_mutex);
+
 	/* A scheduler is registered once per process. */
 	if (scheduler_module_name != NULL) {
+		const char *owner = scheduler_module_name;
+
+		tsrm_mutex_unlock(scheduler_mutex);
 		zend_error(E_CORE_WARNING,
 				"The module %s cannot register an Async scheduler: %s already did",
 				module,
-				scheduler_module_name);
+				owner);
 		return false;
 	}
 
@@ -599,18 +608,28 @@ ZEND_API bool zend_async_scheduler_register(
 		zend_async_coroutine_get_awaiting_info_fn = api->get_awaiting_info;
 	}
 
-	scheduler_module_name = module;
+	/* The caller's string may be request-local: keep a process copy. */
+	scheduler_module_name = pestrdup(module, 1);
+
+	tsrm_mutex_unlock(scheduler_mutex);
+
 	ZEND_ASYNC_INITIALIZE;
 
 	return true;
 }
 
-/* Withdraw the registration and reset every slot to its default. Safe per
- * request (the PHP bridge registers per request); the process-wide state —
- * the internal-context key registry — is NOT touched here. */
+/* Withdraw the registration and reset every slot to its default. The slots
+ * are process-wide: this runs at process shutdown, or when a just-registered
+ * scheduler fails to launch — never per request. The internal-context key
+ * registry is NOT touched here. */
 ZEND_API void zend_async_scheduler_unregister(void)
 {
-	scheduler_module_name = NULL;
+	tsrm_mutex_lock(scheduler_mutex);
+
+	if (scheduler_module_name != NULL) {
+		pefree((char *) scheduler_module_name, 1);
+		scheduler_module_name = NULL;
+	}
 
 	zend_async_new_coroutine_fn = NULL;
 	zend_async_gc_new_coroutine_fn = NULL;
@@ -633,13 +652,20 @@ ZEND_API void zend_async_scheduler_unregister(void)
 	zend_async_coroutine_add_awaiting_info_fn = NULL;
 	zend_async_coroutine_remove_awaiting_info_fn = NULL;
 	zend_async_coroutine_get_awaiting_info_fn = NULL;
+
+	tsrm_mutex_unlock(scheduler_mutex);
 }
 
 void zend_async_api_shutdown(void)
 {
 	zend_async_globals_dtor();
 	zend_async_scheduler_unregister();
 	internal_context_keys_shutdown();
+
+	if (scheduler_mutex != NULL) {
+		tsrm_mutex_free(scheduler_mutex);
+		scheduler_mutex = NULL;
+	}
 }
 
 ZEND_API bool zend_async_is_enabled(void)

ext/async_scheduler_hook/async_scheduler_hook.c

@@ -1110,18 +1110,37 @@ ZEND_METHOD(Async_SchedulerHook, register)
 	ASH_G(module) = zend_string_copy(module);
 	ASH_G(active) = true;
 
-	zend_async_scheduler_api_t api;
-	php_async_build_api(&api);
+	/* The C slots are process-wide and set once; per-thread state is the
+	 * handlers container above. A later request (any thread) only re-binds
+	 * its handlers and launches. */
+	const char *owner = zend_async_get_scheduler_module();
+	bool registered_now = false;
 
-	if (!zend_async_scheduler_register(ZSTR_VAL(module), &api)) {
+	if (owner == NULL) {
+		zend_async_scheduler_api_t api;
+		php_async_build_api(&api);
+
+		registered_now = zend_async_scheduler_register(ZSTR_VAL(module), &api);
+
+		if (!registered_now) {
+			/* Lost a registration race — acceptable if to ourselves. */
+			owner = zend_async_get_scheduler_module();
+		}
+	}
+
+	if (!registered_now && (owner == NULL || strcmp(owner, ZSTR_VAL(module)) != 0)) {
 		php_async_handlers_reset();
 		zend_throw_error(NULL, "Async\\SchedulerHook::register(): the engine refused the scheduler");
 		RETURN_THROWS();
 	}
 
 	if (!ZEND_ASYNC_SCHEDULER_LAUNCH()) {
 		php_async_handlers_reset();
-		zend_async_scheduler_unregister();
+
+		if (registered_now) {
+			zend_async_scheduler_unregister();
+		}
+
 		ZEND_ASYNC_DEACTIVATE;
 		RETURN_THROWS();
 	}
@@ -1372,10 +1391,6 @@ PHP_RSHUTDOWN_FUNCTION(async_scheduler_hook)
 {
 	if (ASH_G(active)) {
 		php_async_handlers_reset();
-
-		/* The hooks died with this request; free the registration so the
-		 * next request in this process can register a scheduler again. */
-		zend_async_scheduler_unregister();
 		ZEND_ASYNC_DEACTIVATE;
 	}