pool.c

/*
+----------------------------------------------------------------------+
  | Copyright (c) The PHP Group                                          |
  +----------------------------------------------------------------------+
  | This source file is subject to version 3.01 of the PHP license,      |
  | that is bundled with this package in the file LICENSE, and is        |
  | available through the world-wide-web at the following url:           |
  | https://www.php.net/license/3_01.txt                                 |
  | If you did not receive a copy of the PHP license and are unable to   |
  | obtain it through the world-wide-web, please send a note to          |
  | license@php.net so we can mail you a copy immediately.               |
  +----------------------------------------------------------------------+
  | Author: Edmond                                                       |
  +----------------------------------------------------------------------+
*/

#include "pool.h"
#include "php_async.h"
#include "exceptions.h"
#include "scheduler.h"
#include "coroutine.h"
#include "pool_arginfo.h"
#include "zend_exceptions.h"
#include "zend_interfaces.h"
#include "zend_enum.h"
#include "internal/zval_circular_buffer.h"

/**
 * Pool - Resource pool with automatic lifecycle management.
 *
 * Architecture (similar to Channel):
 * - Waiting queue for coroutines waiting for resources
 * - zend_async_resume_when registers with waker (so SUSPEND works)
 * - We control wake order via our queue
 * - On wake, coroutine retries and takes resource from pool
 */

#define METHOD(name) PHP_METHOD(Async_Pool, name)
#define THIS_POOL_OBJ ASYNC_POOL_OBJ_FROM_OBJ(Z_OBJ_P(ZEND_THIS))
#define THIS_POOL (THIS_POOL_OBJ->base)

#define THROW_IF_NO_POOL(obj) \
	if (UNEXPECTED((obj)->pool == NULL)) { \
		zend_throw_exception(async_ce_pool_exception, "Pool is not initialized", 0); \
		RETURN_THROWS(); \
	}

#define THROW_IF_CLOSED(pool) \
	if (UNEXPECTED(ZEND_ASYNC_POOL_IS_CLOSED(&(pool)->base))) { \
		zend_throw_exception(async_ce_pool_exception, "Pool is closed", 0); \
		RETURN_THROWS(); \
	}

/* Slots of a pool the engine builds hold what C put there rather than a zval: PDO's connection
 * pool keeps a raw pdo_dbh_t * in each one, and phpredis keeps a socket. Such a value cannot be
 * handed to PHP, and no PHP value can stand in for one on the way back — the C before_release
 * handler dereferences whatever it is given. */
#define THROW_IF_INTERNAL_RESOURCES(pool) \
	if (UNEXPECTED((pool)->base.handler_flags & ZEND_ASYNC_POOL_F_FACTORY_INTERNAL)) { \
		zend_throw_exception(async_ce_pool_exception, \
				"Pool resources are owned by the engine and cannot cross into PHP", 0); \
		RETURN_THROWS(); \
	}

#define ENSURE_COROUTINE_CONTEXT \
	if (UNEXPECTED(ZEND_ASYNC_CURRENT_COROUTINE == NULL)) { \
		async_scheduler_launch(); \
		if (UNEXPECTED(EG(exception) != NULL)) { \
			RETURN_THROWS(); \
		} \
	}

zend_class_entry *async_ce_pool = NULL;
zend_class_entry *async_ce_pool_exception = NULL;
static zend_object_handlers async_pool_handlers;

///////////////////////////////////////////////////////////////////////////////
// Queue operations (copied from channel.c)
///////////////////////////////////////////////////////////////////////////////

static void pool_queue_init(zend_async_callbacks_vector_t *queue)
{
	queue->data = NULL;
	queue->length = 0;
	queue->capacity = 0;
}

static void pool_queue_free(zend_async_callbacks_vector_t *queue)
{
	if (queue->data) {
		efree(queue->data);
		queue->data = NULL;
	}
}

static void pool_queue_push(zend_async_callbacks_vector_t *queue, zend_async_pool_waiter_t *waiter)
{
	if (queue->length >= queue->capacity) {
		uint32_t new_capacity = queue->capacity ? queue->capacity * 2 : 4;
		queue->data = erealloc(queue->data, new_capacity * sizeof(void *));
		queue->capacity = new_capacity;
	}
	waiter->callback.base.ref_count++;
	queue->data[queue->length++] = (zend_async_event_callback_t *) waiter;
}

/* Arrival order is the service order: swapping the last element into the hole
 * would serve the newest waiter first. */
static bool pool_queue_remove(zend_async_callbacks_vector_t *queue, zend_async_pool_waiter_t *waiter)
{
	for (uint32_t i = 0; i < queue->length; i++) {
		if (queue->data[i] == (zend_async_event_callback_t *) waiter) {
			queue->length--;
			memmove(queue->data + i,
					queue->data + i + 1,
					(queue->length - i) * sizeof(zend_async_event_callback_t *));
			return true;
		}
	}
	return false;
}

/* Ignores arrival order; only for callers that drain the whole queue. */
static zend_async_pool_waiter_t *pool_queue_pop_last(zend_async_callbacks_vector_t *queue)
{
	if (queue->length == 0) {
		return NULL;
	}
	return (zend_async_pool_waiter_t *) queue->data[--queue->length];
}

/* Oldest waiter with no reservation, or NULL when every waiter has one. */
static zend_async_pool_waiter_t *pool_queue_first_unreserved(zend_async_callbacks_vector_t *queue)
{
	for (uint32_t i = 0; i < queue->length; i++) {
		zend_async_pool_waiter_t *waiter = (zend_async_pool_waiter_t *) queue->data[i];
		if (!waiter->reserved) {
			return waiter;
		}
	}
	return NULL;
}

///////////////////////////////////////////////////////////////////////////////
// Resource creation/destruction
///////////////////////////////////////////////////////////////////////////////

static bool pool_create_resource(async_pool_t *pool, zval *result)
{
	zend_async_pool_t *base = &pool->base;

	/* Internal C handler */
	if (base->handler_flags & ZEND_ASYNC_POOL_F_FACTORY_INTERNAL) {
		if (base->factory.internal == NULL) {
			return false;
		}
		return base->factory.internal(base, result);
	}

	/* PHP callable */
	if (base->factory.fcall == NULL) {
		return false;
	}

	zval retval;
	ZVAL_UNDEF(&retval);

	base->factory.fcall->fci.retval = &retval;

	if (zend_call_function(&base->factory.fcall->fci, &base->factory.fcall->fci_cache) == FAILURE) {
		return false;
	}

	if (EG(exception)) {
		zval_ptr_dtor(&retval);
		return false;
	}

	ZVAL_COPY_VALUE(result, &retval);
	return true;
}

static void pool_destroy_resource(async_pool_t *pool, zval *resource)
{
	zend_async_pool_t *base = &pool->base;

	/* Internal C handler */
	if (base->handler_flags & ZEND_ASYNC_POOL_F_DESTRUCTOR_INTERNAL) {
		if (base->destructor.internal != NULL) {
			base->destructor.internal(base, resource);
		}
		zval_ptr_dtor(resource);
		return;
	}

	/* PHP callable */
	if (base->destructor.fcall != NULL) {
		zval retval;
		ZVAL_UNDEF(&retval);

		zval args[1];
		ZVAL_COPY(&args[0], resource);

		/* Use local copy of fci to avoid corrupting the stored structure */
		zend_fcall_info fci = base->destructor.fcall->fci;
		fci.retval = &retval;
		fci.param_count = 1;
		fci.params = args;

		zend_call_function(&fci, &base->destructor.fcall->fci_cache);

		zval_ptr_dtor(&args[0]);
		zval_ptr_dtor(&retval);
	}

	zval_ptr_dtor(resource);
}

/* release() borrows its argument and pool_destroy_resource drops a reference:
 * take one of our own, or the caller's variable is freed out from under it. */
static void pool_destroy_borrowed_resource(async_pool_t *pool, zval *resource)
{
	zval owned;
	ZVAL_COPY(&owned, resource);
	pool_destroy_resource(pool, &owned);
}

///////////////////////////////////////////////////////////////////////////////
// Callback helpers
///////////////////////////////////////////////////////////////////////////////

/* Call beforeAcquire callback, returns true if resource is OK */
static bool pool_call_before_acquire(async_pool_t *pool, zval *resource)
{
	zend_async_pool_t *base = &pool->base;

	/* Internal C handler */
	if (base->handler_flags & ZEND_ASYNC_POOL_F_BEFORE_ACQUIRE_INTERNAL) {
		if (base->before_acquire.internal == NULL) {
			return true;
		}
		return base->before_acquire.internal(base, resource);
	}

	/* PHP callable */
	if (base->before_acquire.fcall == NULL) {
		return true;
	}

	/* Park an in-flight exception: collect only the callback's own */
	zend_object *parked = NULL;
	zend_exception_save_fast(&EG(exception), &parked);

	zval retval;
	ZVAL_UNDEF(&retval);

	zval args[1];
	ZVAL_COPY(&args[0], resource);

	/* Use local copy of fci to avoid corrupting the stored structure */
	zend_fcall_info fci = base->before_acquire.fcall->fci;
	fci.retval = &retval;
	fci.param_count = 1;
	fci.params = args;

	zend_call_function(&fci, &base->before_acquire.fcall->fci_cache);

	zval_ptr_dtor(&args[0]);

	bool result = true;
	if (EG(exception)) {
		result = false;
		zend_clear_exception();
	} else if (Z_TYPE(retval) == IS_FALSE) {
		result = false;
	}

	zval_ptr_dtor(&retval);
	zend_exception_restore_fast(&EG(exception), &parked);
	return result;
}

/* Call beforeRelease callback, returns true if resource should be returned to pool */
static bool pool_call_before_release(async_pool_t *pool, zval *resource)
{
	zend_async_pool_t *base = &pool->base;

	/* Internal C handler */
	if (base->handler_flags & ZEND_ASYNC_POOL_F_BEFORE_RELEASE_INTERNAL) {
		if (base->before_release.internal == NULL) {
			return true;
		}
		return base->before_release.internal(base, resource);
	}

	/* PHP callable */
	if (base->before_release.fcall == NULL) {
		return true;
	}

	/* Park an in-flight exception: collect only the callback's own */
	zend_object *parked = NULL;
	zend_exception_save_fast(&EG(exception), &parked);

	zval retval;
	ZVAL_UNDEF(&retval);

	zval args[1];
	ZVAL_COPY(&args[0], resource);

	/* Use local copy of fci to avoid corrupting the stored structure */
	zend_fcall_info fci = base->before_release.fcall->fci;
	fci.retval = &retval;
	fci.param_count = 1;
	fci.params = args;

	zend_call_function(&fci, &base->before_release.fcall->fci_cache);

	zval_ptr_dtor(&args[0]);

	bool result = true;
	if (EG(exception)) {
		result = false;
		zend_clear_exception();
	} else if (Z_TYPE(retval) == IS_FALSE) {
		result = false;
	}

	zval_ptr_dtor(&retval);
	zend_exception_restore_fast(&EG(exception), &parked);
	return result;
}

/* Call strategy reportSuccess */
static void pool_strategy_report_success(async_pool_t *pool)
{
	zend_async_pool_t *base = &pool->base;

	/* Internal C strategy */
	if (base->handler_flags & ZEND_ASYNC_POOL_F_STRATEGY_INTERNAL) {
		if (base->strategy.internal && base->strategy.internal->report_success) {
			base->strategy.internal->report_success(base);
		}
		return;
	}

	/* PHP strategy */
	if (base->strategy.object == NULL) {
		return;
	}

	/* Park an in-flight exception: clear only the strategy's own */
	zend_object *parked = NULL;
	zend_exception_save_fast(&EG(exception), &parked);

	zval retval, source;
	ZVAL_UNDEF(&retval);

	if (base->wrapper) {
		ZVAL_OBJ(&source, base->wrapper);
	} else {
		ZVAL_NULL(&source);
	}

	zend_call_method_with_1_params(base->strategy.object, NULL, NULL, "reportSuccess", &retval, &source);

	if (EG(exception)) {
		zend_clear_exception();
	}

	zval_ptr_dtor(&retval);
	zend_exception_restore_fast(&EG(exception), &parked);
}

/* Call strategy reportFailure */
static void pool_strategy_report_failure(async_pool_t *pool, zend_object *error)
{
	zend_async_pool_t *base = &pool->base;

	/* Internal C strategy */
	if (base->handler_flags & ZEND_ASYNC_POOL_F_STRATEGY_INTERNAL) {
		if (base->strategy.internal && base->strategy.internal->report_failure) {
			base->strategy.internal->report_failure(base, error);
		}
		return;
	}

	/* PHP strategy */
	if (base->strategy.object == NULL) {
		return;
	}

	/* Park an in-flight exception: clear only the strategy's own */
	zend_object *parked = NULL;
	zend_exception_save_fast(&EG(exception), &parked);

	zval retval, source, error_zval;
	bool owns_error = false;
	ZVAL_UNDEF(&retval);

	if (base->wrapper) {
		ZVAL_OBJ(&source, base->wrapper);
	} else {
		ZVAL_NULL(&source);
	}

	if (error) {
		ZVAL_OBJ(&error_zval, error);
	} else {
		/* Create a generic exception if none provided. Construct it directly
		 * without going through zend_throw_exception(), which would transfer
		 * ownership to EG(exception) and then zend_clear_exception() would
		 * free the object — leaving a dangling pointer. */
		object_init_ex(&error_zval, zend_ce_exception);
		zval msg;
		ZVAL_STRING(&msg, "Resource validation failed");
		zend_update_property_ex(zend_ce_exception, Z_OBJ(error_zval),
			ZSTR_KNOWN(ZEND_STR_MESSAGE), &msg);
		zval_ptr_dtor(&msg);
		owns_error = true;
	}

	zend_call_method_with_2_params(base->strategy.object, NULL, NULL, "reportFailure", &retval, &source, &error_zval);

	if (EG(exception)) {
		zend_clear_exception();
	}

	zval_ptr_dtor(&retval);
	if (owns_error) {
		zval_ptr_dtor(&error_zval);
	}
	zend_exception_restore_fast(&EG(exception), &parked);
}

///////////////////////////////////////////////////////////////////////////////
// Wait/Wake operations (like channel)
///////////////////////////////////////////////////////////////////////////////

static bool pool_wake_waiter(async_pool_t *pool);

/**
 * Parks the current coroutine until the pool has an idle resource or free
 * capacity for it.
 *
 * Returns true when the pool reserved a place for this waiter, which only it
 * may take; false on timeout, cancellation and pool close. A caller that
 * returns without taking a reserved place must wake the next waiter.
 */
static bool pool_wait_for_resource(async_pool_t *pool, zend_long timeout_ms)
{
	zend_async_pool_t *base = &pool->base;
	zend_async_pool_waiter_t *waiter = ecalloc(1, sizeof(zend_async_pool_waiter_t));
	waiter->callback.base.callback = zend_async_waker_callback_resolve;
	waiter->callback.base.ref_count = 1;
	waiter->callback.coroutine = ZEND_ASYNC_CURRENT_COROUTINE;
	waiter->callback.event = &base->event;

	pool_queue_push(&pool->waiters, waiter);

	zend_async_resume_when(
			ZEND_ASYNC_CURRENT_COROUTINE, &base->event, false, zend_async_waker_callback_resolve, &waiter->callback);

	if (timeout_ms > 0) {
		zend_async_resume_when(ZEND_ASYNC_CURRENT_COROUTINE,
							   &ZEND_ASYNC_NEW_TIMER_EVENT(timeout_ms, false)->base,
							   true,
							   zend_async_waker_callback_timeout,
							   NULL);
	}

	ZEND_ASYNC_SUSPEND();

	const bool had_reservation = waiter->reserved;

	/* Cleanup after waking up */
	if (pool_queue_remove(&pool->waiters, waiter)) {
		/* Was still in queue (every path but pool close) - release queue's ref */
		ZEND_ASYNC_EVENT_CALLBACK_RELEASE(&waiter->callback.base);
	}

	if (had_reservation) {
		/* Spent by the caller's retry after we return, or handed on below. */
		ZEND_ASSERT(pool->reserved_count > 0 && "A reservation was consumed twice");
		pool->reserved_count--;
	}

	/* An unspent reservation would leave an idle resource or free capacity with
	 * no waiter assigned, and nothing else wakes the queue afterwards. */
	if (had_reservation && UNEXPECTED(EG(exception)) && !ZEND_ASYNC_POOL_IS_CLOSED(pool)) {
		pool_wake_waiter(pool);
	}

	/* Release our initial ref */
	ZEND_ASYNC_EVENT_CALLBACK_RELEASE(&waiter->callback.base);

	return had_reservation;
}

/**
 * Reserves an idle resource or free capacity for the oldest waiter without one
 * and queues its resumption. Returns false when every waiter already holds a
 * reservation, or when nobody waits.
 *
 * The waiter stays in the queue on purpose: until it runs, reserved_count is
 * what stops any other coroutine — the releasing one included — from taking
 * the place reserved for it.
 */
static bool pool_wake_waiter(async_pool_t *pool)
{
	zend_async_pool_waiter_t *waiter = pool_queue_first_unreserved(&pool->waiters);
	if (waiter == NULL) {
		return false;
	}

	/* The event callback stays registered; the waker's event cleanup drops it
	 * on resume. */
	ZEND_ASSERT(!waiter->reserved && "pool_queue_first_unreserved returned a reserved waiter");
	waiter->reserved = true;
	pool->reserved_count++;
	waiter->callback.base.callback(&pool->base.event, &waiter->callback.base, NULL, NULL);

	return true;
}

/* Drains the queue from the tail: order does not matter when every waiter gets
 * the same exception, and head removal would shift the rest each time. */
static void pool_wake_all_with_exception(async_pool_t *pool, zend_object *exception)
{
	zend_async_pool_t *base = &pool->base;
	zend_async_pool_waiter_t *waiter;
	while ((waiter = pool_queue_pop_last(&pool->waiters)) != NULL) {
		base->event.del_callback(&base->event, &waiter->callback.base);
		waiter->callback.base.callback(&base->event, &waiter->callback.base, NULL, exception);
		ZEND_ASYNC_EVENT_CALLBACK_RELEASE(&waiter->callback.base);
	}
}

///////////////////////////////////////////////////////////////////////////////
// Event handlers
///////////////////////////////////////////////////////////////////////////////

static bool pool_add_callback(zend_async_event_t *event, zend_async_event_callback_t *callback)
{
	return zend_async_callbacks_push(event, callback);
}

static bool pool_del_callback(zend_async_event_t *event, zend_async_event_callback_t *callback)
{
	return zend_async_callbacks_remove(event, callback);
}

static bool pool_dispose(zend_async_event_t *event)
{
	async_pool_t *pool = (async_pool_t *) event;
	zend_async_pool_t *base = &pool->base;

	/* Close if not already closed */
	if (!ZEND_ASYNC_POOL_IS_CLOSED(pool)) {
		zend_async_pool_close(pool);
	}

	/* Free callbacks */
	zend_async_callbacks_free(&base->event);
	pool_queue_free(&pool->waiters);

	/* Free circular buffer */
	circular_buffer_dtor(&pool->idle);

	/* Free fcall structures (only if not internal handlers) */
	if (!(base->handler_flags & ZEND_ASYNC_POOL_F_FACTORY_INTERNAL) && base->factory.fcall) {
		zend_fcall_release(base->factory.fcall);
	}
	if (!(base->handler_flags & ZEND_ASYNC_POOL_F_DESTRUCTOR_INTERNAL) && base->destructor.fcall) {
		zend_fcall_release(base->destructor.fcall);
	}
	if (!(base->handler_flags & ZEND_ASYNC_POOL_F_HEALTHCHECK_INTERNAL) && base->healthcheck.fcall) {
		zend_fcall_release(base->healthcheck.fcall);
	}
	if (!(base->handler_flags & ZEND_ASYNC_POOL_F_BEFORE_ACQUIRE_INTERNAL) && base->before_acquire.fcall) {
		zend_fcall_release(base->before_acquire.fcall);
	}
	if (!(base->handler_flags & ZEND_ASYNC_POOL_F_BEFORE_RELEASE_INTERNAL) && base->before_release.fcall) {
		zend_fcall_release(base->before_release.fcall);
	}

	/* Release strategy object */
	if (!(base->handler_flags & ZEND_ASYNC_POOL_F_STRATEGY_INTERNAL) && base->strategy.object) {
		OBJ_RELEASE(base->strategy.object);
		base->strategy.object = NULL;
	}

	efree(pool);
	return true;
}

static bool pool_start(zend_async_event_t *event)
{
	return true;
}

static bool pool_stop(zend_async_event_t *event)
{
	return true;
}

static zend_string *pool_info(zend_async_event_t *event)
{
	const async_pool_t *pool = (async_pool_t *) event;

	/* Without reserved, idle > 0 with coroutines queued reads as a lost wakeup. */
	return zend_strpprintf(0,
						   "Pool(idle=%zu, active=%u, reserved=%u, max=%u)",
						   circular_buffer_count(&pool->idle),
						   pool->active_count,
						   pool->reserved_count,
						   pool->base.max_size);
}

static void pool_event_init(async_pool_t *pool)
{
	zend_async_event_t *event = &pool->base.event;
	memset(event, 0, sizeof(zend_async_event_t));

	event->ref_count = 1;
	event->add_callback = pool_add_callback;
	event->del_callback = pool_del_callback;
	event->start = pool_start;
	event->stop = pool_stop;
	event->dispose = pool_dispose;
	event->info = pool_info;
}

///////////////////////////////////////////////////////////////////////////////
// Healthcheck timer
///////////////////////////////////////////////////////////////////////////////

/* Call healthcheck, returns true if resource is healthy */
static bool pool_call_healthcheck(async_pool_t *pool, zval *resource)
{
	zend_async_pool_t *base = &pool->base;

	/* Internal C handler */
	if (base->handler_flags & ZEND_ASYNC_POOL_F_HEALTHCHECK_INTERNAL) {
		if (base->healthcheck.internal == NULL) {
			return true;
		}
		return base->healthcheck.internal(base, resource);
	}

	/* PHP callable */
	if (base->healthcheck.fcall == NULL) {
		return true;
	}

	zval retval;
	ZVAL_UNDEF(&retval);

	zval args[1];
	ZVAL_COPY(&args[0], resource);

	/* Use local copy of fci to avoid corrupting the stored structure */
	zend_fcall_info fci = base->healthcheck.fcall->fci;
	fci.retval = &retval;
	fci.param_count = 1;
	fci.params = args;

	zend_call_function(&fci, &base->healthcheck.fcall->fci_cache);

	zval_ptr_dtor(&args[0]);

	bool is_healthy = true;
	if (EG(exception)) {
		is_healthy = false;
		zend_clear_exception();
	} else if (Z_TYPE(retval) == IS_FALSE) {
		is_healthy = false;
	}

	zval_ptr_dtor(&retval);
	return is_healthy;
}

static void pool_healthcheck_callback_dispose(zend_async_event_callback_t *callback, zend_async_event_t *event)
{
	/* No-op: callback is embedded in async_pool_t, not heap-allocated */
}

static void pool_healthcheck_timer_callback(zend_async_event_t *timer_event,
											zend_async_event_callback_t *callback,
											void *result,
											zend_object *exception)
{
	async_pool_t *pool = (async_pool_t *) ((char *) callback - offsetof(async_pool_t, healthcheck_callback));
	zend_async_pool_t *base = &pool->base;

	if (ZEND_ASYNC_POOL_IS_CLOSED(pool)) {
		return;
	}

	/* Check if healthcheck is set */
	bool has_healthcheck = (base->handler_flags & ZEND_ASYNC_POOL_F_HEALTHCHECK_INTERNAL)
			? (base->healthcheck.internal != NULL)
			: (base->healthcheck.fcall != NULL);

	if (!has_healthcheck) {
		return;
	}

	/* Process all idle resources */
	size_t count = circular_buffer_count(&pool->idle);

	for (size_t i = 0; i < count; i++) {
		zval resource;
		if (zval_circular_buffer_pop(&pool->idle, &resource) != SUCCESS) {
			break;
		}

		/* During the hook the resource is neither idle nor active, and the hook is
		 * PHP code that may inspect the pool: count it as active. */
		pool->active_count++;

		bool is_healthy = pool_call_healthcheck(pool, &resource);

		pool->active_count--;

		if (is_healthy) {
			/* Closed while we checked: close already drained idle. */
			if (ZEND_ASYNC_POOL_IS_CLOSED(pool)) {
				pool_destroy_resource(pool, &resource);
				break;
			}

			/* Resource is healthy - return to buffer */
			zval_circular_buffer_push(&pool->idle, &resource, true);
			zval_ptr_dtor(&resource);
		} else {
			/* Resource is dead - destroy it */
			pool_destroy_resource(pool, &resource);

			if (UNEXPECTED(EG(exception))) {
				break;
			}

			/* Create new one to maintain min_size */
			uint32_t total = ASYNC_POOL_TOTAL(pool);
			if (total < base->min_size) {
				zval new_resource;
				if (pool_create_resource(pool, &new_resource)) {
					zval_circular_buffer_push(&pool->idle, &new_resource, true);
					zval_ptr_dtor(&new_resource);
					continue;
				}

				if (UNEXPECTED(EG(exception))) {
					break;
				}
			}
		}
	}

	/* Destroying a resource frees capacity and no release follows to report it.
	 * Placed after the loop to cover the exits a throwing hook takes. */
	pool_wake_waiter(pool);
}

static void pool_start_healthcheck_timer(async_pool_t *pool)
{
	zend_async_pool_t *base = &pool->base;

	if (pool->healthcheck_interval_ms == 0) {
		return;
	}

	/* Check if healthcheck is set */
	const bool has_healthcheck = (base->handler_flags & ZEND_ASYNC_POOL_F_HEALTHCHECK_INTERNAL)
			? (base->healthcheck.internal != NULL)
			: (base->healthcheck.fcall != NULL);

	if (false == has_healthcheck) {
		return;
	}

	pool->healthcheck_timer = ZEND_ASYNC_NEW_TIMER_EVENT(pool->healthcheck_interval_ms, true);

	if (pool->healthcheck_timer == NULL) {
		return;
	}

	/* The healthcheck timer is a background heartbeat, not application work.
	 * Mark it HIDDEN so it does not count toward active_event_count — otherwise
	 * an idle pool keeps the reactor loop alive and blocks a graceful worker
	 * shutdown (e.g. on hot-reload), tripping the scheduler's loop-alive assert. */
	ZEND_ASYNC_EVENT_SET_HIDDEN(&pool->healthcheck_timer->base);

	/* Use inline callback embedded in pool structure */
	pool->healthcheck_callback.ref_count = 1;
	pool->healthcheck_callback.callback = pool_healthcheck_timer_callback;
	pool->healthcheck_callback.dispose = pool_healthcheck_callback_dispose;

	pool->healthcheck_timer->base.add_callback(&pool->healthcheck_timer->base, &pool->healthcheck_callback);
	pool->healthcheck_timer->base.start(&pool->healthcheck_timer->base);
}

static void pool_stop_healthcheck_timer(async_pool_t *pool)
{
	if (pool->healthcheck_timer != NULL) {
		pool->healthcheck_timer->base.stop(&pool->healthcheck_timer->base);
		ZEND_ASYNC_EVENT_RELEASE(&pool->healthcheck_timer->base);
		pool->healthcheck_timer = NULL;
	}
}

///////////////////////////////////////////////////////////////////////////////
// C API implementation
///////////////////////////////////////////////////////////////////////////////

/* Create pool with PHP callables (fcall) */
async_pool_t *zend_async_pool_create(zend_fcall_t *factory,
									 zend_fcall_t *destructor,
									 zend_fcall_t *healthcheck,
									 zend_fcall_t *before_acquire,
									 zend_fcall_t *before_release,
									 uint32_t min_size,
									 uint32_t max_size,
									 uint32_t healthcheck_interval_ms)
{
	async_pool_t *pool = ecalloc(1, sizeof(async_pool_t));
	zend_async_pool_t *base = &pool->base;

	/* Event init */
	pool_event_init(pool);

	/* Callbacks - all fcall (PHP callables) */
	base->handler_flags = 0;
	base->factory.fcall = factory;
	base->destructor.fcall = destructor;
	base->healthcheck.fcall = healthcheck;
	base->before_acquire.fcall = before_acquire;
	base->before_release.fcall = before_release;

	/* Config */
	base->min_size = min_size;
	base->max_size = max_size;
	pool->healthcheck_interval_ms = healthcheck_interval_ms;
	pool->active_count = 0;

	/* Init idle buffer */
	circular_buffer_ctor(&pool->idle, min_size > 0 ? min_size : 4, sizeof(zval), &zend_std_persistent_allocator);

	/* Init waiters queue */
	pool_queue_init(&pool->waiters);

	/* Pre-warm: create min_size resources */
	for (uint32_t i = 0; i < min_size; i++) {
		zval resource;
		if (pool_create_resource(pool, &resource)) {
			zval_circular_buffer_push(&pool->idle, &resource, true);
			zval_ptr_dtor(&resource);
		} else if (UNEXPECTED(EG(exception))) {
			/* Stop on first error */
			break;
		}
	}

	/* Start healthcheck timer */
	pool_start_healthcheck_timer(pool);

	return pool;
}

/* Create pool with internal C handlers */
async_pool_t *zend_async_pool_create_internal(zend_async_pool_factory_fn factory,
											  zend_async_pool_destructor_fn destructor,
											  zend_async_pool_healthcheck_fn healthcheck,
											  zend_async_pool_before_acquire_fn before_acquire,
											  zend_async_pool_before_release_fn before_release,
											  uint32_t min_size,
											  uint32_t max_size,
											  uint32_t healthcheck_interval_ms)
{
	async_pool_t *pool = ecalloc(1, sizeof(async_pool_t));
	zend_async_pool_t *base = &pool->base;

	/* Event init */
	pool_event_init(pool);

	/* Callbacks - all internal C handlers */
	base->handler_flags = ZEND_ASYNC_POOL_F_FACTORY_INTERNAL | ZEND_ASYNC_POOL_F_DESTRUCTOR_INTERNAL |
			ZEND_ASYNC_POOL_F_HEALTHCHECK_INTERNAL | ZEND_ASYNC_POOL_F_BEFORE_ACQUIRE_INTERNAL |
			ZEND_ASYNC_POOL_F_BEFORE_RELEASE_INTERNAL;

	base->factory.internal = factory;
	base->destructor.internal = destructor;
	base->healthcheck.internal = healthcheck;
	base->before_acquire.internal = before_acquire;
	base->before_release.internal = before_release;

	/* Config */
	base->min_size = min_size;
	base->max_size = max_size;
	pool->healthcheck_interval_ms = healthcheck_interval_ms;
	pool->active_count = 0;

	/* Init idle buffer */
	circular_buffer_ctor(&pool->idle, min_size > 0 ? min_size : 4, sizeof(zval), &zend_std_persistent_allocator);

	/* Init waiters queue */
	pool_queue_init(&pool->waiters);

	/* Pre-warm: create min_size resources */
	for (uint32_t i = 0; i < min_size; i++) {
		zval resource;
		if (pool_create_resource(pool, &resource)) {
			zval_circular_buffer_push(&pool->idle, &resource, true);
			zval_ptr_dtor(&resource);
		} else if (EG(exception)) {
			/* Stop on first error */
			break;
		}
	}

	/* Start healthcheck timer */
	pool_start_healthcheck_timer(pool);

	return pool;
}

bool zend_async_pool_acquire(async_pool_t *pool, zval *result, zend_long timeout_ms)
{
	const zend_async_pool_t *base = &pool->base;
	/* Set when the last wakeup reserved a resource or capacity for this one. */
	bool has_reservation = false;

retry:
	/* 1. Closed? */
	if (ZEND_ASYNC_POOL_IS_CLOSED(pool)) {
		zend_throw_exception(async_ce_pool_exception, "Pool is closed", 0);
		goto give_up;
	}

	/* 2. Circuit breaker check */
	if (base->circuit_state == ZEND_ASYNC_CIRCUIT_STATE_INACTIVE) {
		async_throw_service_unavailable("Service is unavailable (circuit breaker is open)");
		goto give_up;
	}

	/* Take only what is reserved for nobody, or what is reserved for us. Without
	 * the rule a coroutine that releases and re-acquires takes back the slot its
	 * own release just handed to a parked waiter. */
	const bool may_serve_self = has_reservation || ASYNC_POOL_UNRESERVED(pool) > 0;

	/* 2. Have idle resource? */
	if (!circular_buffer_is_empty(&pool->idle) && may_serve_self) {
		zval resource;
		zval_circular_buffer_pop(&pool->idle, &resource);
		has_reservation = false;

		/* Reserve before beforeAcquire: it may yield, and until then the
		 * resource is in neither idle nor active_count. */
		pool->active_count++;

		/* beforeAcquire check (if set) */
		if (!pool_call_before_acquire(pool, &resource)) {
			/* Check failed - destroy and try next */
			pool->active_count--;
			pool_destroy_resource(pool, &resource);
			if (UNEXPECTED(EG(exception))) {
				/* Capacity freed: no other path wakes a waiter for it. */
				pool_wake_waiter(pool);
				return false;
			}
			goto retry;
		}

		ZVAL_COPY_VALUE(result, &resource);
		return true;
	}

	/* 3. Can create new? Reserve slot BEFORE factory call — factory may yield,
	 * allowing other coroutines to pass the same check concurrently. Capacity is
	 * reserved for waiters the same way an idle resource is. */
	const uint32_t total = ASYNC_POOL_TOTAL(pool);
	if (total < base->max_size && may_serve_self) {
		has_reservation = false;
		pool->active_count++;
		if (pool_create_resource(pool, result)) {
			return true;
		}
		pool->active_count--;

		/* Factory failed — wake one parked waiter so it gets a chance to retry
		 * (and propagate its own failure). Without this, waiters that entered
		 * while we held the slot reservation deadlock: no idle, no release,
		 * nothing to wake them. Each wakeup cascades to the next on failure. */
		pool_wake_waiter(pool);

		if (UNEXPECTED(EG(exception))) {
			return false;
		}

		/* Factory returned false without setting an exception — synthesize one
		 * so the caller fails fast instead of falling through to wait forever
		 * (nothing in flight will wake us either). */
		zend_throw_exception(async_ce_pool_exception, "Failed to create pool resource", 0);
		return false;
	}

	/* 4. Wait - like channel */
	has_reservation = pool_wait_for_resource(pool, timeout_ms);

	if (UNEXPECTED(EG(exception))) {
		return false;
	}

	goto retry;

give_up:
	/* Leaving without spending the reservation: hand it back to the queue. */
	if (has_reservation) {
		pool_wake_waiter(pool);
	}

	return false;
}

bool zend_async_pool_try_acquire(async_pool_t *pool, zval *result)
{
	zend_async_pool_t *base = &pool->base;

retry:
	if (ZEND_ASYNC_POOL_IS_CLOSED(pool)) {
		return false;
	}

	/* Circuit breaker check */
	if (base->circuit_state == ZEND_ASYNC_CIRCUIT_STATE_INACTIVE) {
		return false;
	}

	/* Same rule as the blocking path: never take a reserved place. */
	const bool may_serve_self = ASYNC_POOL_UNRESERVED(pool) > 0;

	/* Have idle? */
	if (!circular_buffer_is_empty(&pool->idle) && may_serve_self) {
		zval resource;
		zval_circular_buffer_pop(&pool->idle, &resource);

		/* Reserve before a hook that may yield. */
		pool->active_count++;

		/* beforeAcquire check */
		if (!pool_call_before_acquire(pool, &resource)) {
			pool->active_count--;
			pool_destroy_resource(pool, &resource);
			if (UNEXPECTED(EG(exception))) {
				pool_wake_waiter(pool);
				return false;
			}
			goto retry;
		}

		ZVAL_COPY_VALUE(result, &resource);
		return true;
	}

	/* Can create new? Reserve slot before factory (may yield). */
	const uint32_t total = ASYNC_POOL_TOTAL(pool);
	if (total < base->max_size && may_serve_self) {
		pool->active_count++;
		if (pool_create_resource(pool, result)) {
			return true;
		}
		pool->active_count--;

		/* Nothing else reports the freed capacity to the queue — same reason as
		 * in zend_async_pool_acquire. */
		pool_wake_waiter(pool);
	}

	/* No resource available */
	return false;
}

void zend_async_pool_release(async_pool_t *pool, zval *resource)
{
	/* beforeRelease callback (if set) */
	/* Returns false = resource is broken, destroy it */
	const bool reusable = pool_call_before_release(pool, resource);

	/* After the hook, which may yield. Exactly one decrement. */
	pool->active_count--;

	if (!reusable) {
		/* Report failure to strategy */
		pool_strategy_report_failure(pool, NULL);
		pool_destroy_borrowed_resource(pool, resource);
		/* Slot is now free (total dropped) but no idle resource was returned.
		 * Wake one waiter so it can try to create a fresh resource instead of
		 * parking forever — without this the pool deadlocks when every
		 * connection in flight fails (e.g. backend is permanently down). */
		pool_wake_waiter(pool);
		return;
	}

	/* Report success to strategy */
	pool_strategy_report_success(pool);

	/* Closed? Destroy resource */
	if (ZEND_ASYNC_POOL_IS_CLOSED(pool)) {
		pool_destroy_borrowed_resource(pool, resource);
		return;
	}

	/* Buffer first: the woken waiter must find the resource in place, and until
	 * it runs the reservation keeps everyone else off. */
	zval_circular_buffer_push(&pool->idle, resource, true);
	pool_wake_waiter(pool);
}

void zend_async_pool_close(async_pool_t *pool)
{
	zend_async_pool_t *base = &pool->base;

	if (ZEND_ASYNC_POOL_IS_CLOSED(pool)) {
		return;
	}

	/* Park an in-flight exception (a dtor can close the pool mid-unwind):
	 * the destroy loop below must collect only what resource dtors threw. */
	zend_object **exception_ptr = &EG(exception);
	zend_object *parked_exception = NULL;
	zend_object **parked_exception_ptr = &parked_exception;
	zend_exception_save_fast(exception_ptr, parked_exception_ptr);

	ZEND_ASYNC_EVENT_SET_CLOSED(&base->event);

	/* Stop healthcheck timer */
	pool_stop_healthcheck_timer(pool);

	/* async_new_exception() answers NULL once EG(active) is cleared, and a pool the engine
	 * frees itself is closed after that point. A clean shutdown leaves no waiter to reject:
	 * the scheduler destroys every coroutine before the request ends. A bailout can leave
	 * one, and its coroutine is already unwound, so the rejection has nobody to reach. */
	zend_object *ex = async_new_exception(async_ce_pool_exception, "Pool is closed");

	if (EXPECTED(ex != NULL)) {
		pool_wake_all_with_exception(pool, ex);
		OBJ_RELEASE(ex);
	}

	/* Destroy all idle resources — must destroy all even if destructor throws */
	zval resource;
	zend_object *first_exception = NULL;
	while (!circular_buffer_is_empty(&pool->idle)) {
		if (zval_circular_buffer_pop(&pool->idle, &resource) == SUCCESS) {

			pool_destroy_resource(pool, &resource);

			if (UNEXPECTED(EG(exception))) {
				zend_object *exception = EG(exception);
				GC_ADDREF(exception);
				zend_clear_exception();

				if (first_exception != NULL) {
					zend_exception_set_previous(exception, first_exception);
				}

				first_exception = exception;
			}
		}
	}

	if (first_exception != NULL) {
		zval ex_zval;
		ZVAL_OBJ(&ex_zval, first_exception);
		zend_throw_exception_object(&ex_zval);
	}

	zend_exception_restore_fast(exception_ptr, parked_exception_ptr);
}

void zend_async_pool_destroy(async_pool_t *pool)
{
	zend_async_pool_t *base = &pool->base;

	/* Close if not already closed */
	if (!ZEND_ASYNC_POOL_IS_CLOSED(pool)) {
		zend_async_pool_close(pool);
	}

	/* Free callbacks */
	zend_async_callbacks_free(&base->event);
	pool_queue_free(&pool->waiters);

	/* Free circular buffer */
	circular_buffer_dtor(&pool->idle);

	/* Free fcall structures (only if not internal handlers) */
	if (!(base->handler_flags & ZEND_ASYNC_POOL_F_FACTORY_INTERNAL) && base->factory.fcall) {
		zend_fcall_release(base->factory.fcall);
	}
	if (!(base->handler_flags & ZEND_ASYNC_POOL_F_DESTRUCTOR_INTERNAL) && base->destructor.fcall) {
		zend_fcall_release(base->destructor.fcall);
	}
	if (!(base->handler_flags & ZEND_ASYNC_POOL_F_HEALTHCHECK_INTERNAL) && base->healthcheck.fcall) {
		zend_fcall_release(base->healthcheck.fcall);
	}
	if (!(base->handler_flags & ZEND_ASYNC_POOL_F_BEFORE_ACQUIRE_INTERNAL) && base->before_acquire.fcall) {
		zend_fcall_release(base->before_acquire.fcall);
	}
	if (!(base->handler_flags & ZEND_ASYNC_POOL_F_BEFORE_RELEASE_INTERNAL) && base->before_release.fcall) {
		zend_fcall_release(base->before_release.fcall);
	}

	/* Release strategy object */
	if (!(base->handler_flags & ZEND_ASYNC_POOL_F_STRATEGY_INTERNAL) && base->strategy.object) {
		OBJ_RELEASE(base->strategy.object);
		base->strategy.object = NULL;
	}

	efree(pool);
}

uint32_t zend_async_pool_count(async_pool_t *pool)
{
	return ASYNC_POOL_TOTAL(pool);
}

uint32_t zend_async_pool_idle_count(async_pool_t *pool)
{
	return circular_buffer_count(&pool->idle);
}

uint32_t zend_async_pool_active_count(async_pool_t *pool)
{
	return pool->active_count;
}

///////////////////////////////////////////////////////////////////////////////
// PHP object handlers
///////////////////////////////////////////////////////////////////////////////

static zend_object *async_pool_create_object(zend_class_entry *ce)
{
	async_pool_obj_t *obj = zend_object_alloc(sizeof(async_pool_obj_t), ce);

	zend_object_std_init(&obj->std, ce);
	obj->std.handlers = &async_pool_handlers;
	obj->pool = NULL;

	return &obj->std;
}

/* Create PHP object wrapper for existing pool (for C API) */
zend_object *async_pool_create_object_for_pool(async_pool_t *pool)
{
	zend_object *obj = async_pool_create_object(async_ce_pool);
	async_pool_obj_t *pool_obj = ASYNC_POOL_OBJ_FROM_OBJ(obj);
	pool_obj->pool = pool;

	/* Wrapper holds a reference to the pool */
	ZEND_ASYNC_EVENT_ADD_REF(&pool->base.event);

	/* Set wrapper reference for strategy callbacks */
	pool->base.wrapper = obj;

	return obj;
}

static void async_pool_free_object(zend_object *object)
{
	async_pool_obj_t *obj = ASYNC_POOL_OBJ_FROM_OBJ(object);

	if (obj->pool != NULL) {
		ZEND_ASYNC_EVENT_RELEASE(&obj->pool->base.event);
		obj->pool = NULL;
	}

	zend_object_std_dtor(object);
}

static HashTable *async_pool_get_gc(zend_object *object, zval **table, int *num)
{
	async_pool_obj_t *obj = ASYNC_POOL_OBJ_FROM_OBJ(object);

	if (obj->pool == NULL) {
		*table = NULL;
		*num = 0;
		return NULL;
	}

	zend_get_gc_buffer *buf = zend_get_gc_buffer_create();

	/* GC for idle resources */
	circular_buffer_t *cb = &obj->pool->idle;
	size_t idx = cb->tail;
	while (idx != cb->head) {
		zval *val = (zval *) ((char *) cb->data + idx * cb->item_size);
		zend_get_gc_buffer_add_zval(buf, val);
		idx = (idx + 1) & (cb->capacity - 1);
	}

	/* GC for the PHP callable handlers. Without these, any reference cycle
	 * that passes through factory/destructor/healthcheck/beforeAcquire/
	 * beforeRelease — e.g. a factory closure that captures an object which
	 * owns the pool — is invisible to the cycle collector and leaks. */
	zend_async_pool_t *base = &obj->pool->base;

	if (!(base->handler_flags & ZEND_ASYNC_POOL_F_FACTORY_INTERNAL) && base->factory.fcall) {
		zend_get_gc_buffer_add_zval(buf, &base->factory.fcall->fci.function_name);
	}

	if (!(base->handler_flags & ZEND_ASYNC_POOL_F_DESTRUCTOR_INTERNAL) && base->destructor.fcall) {
		zend_get_gc_buffer_add_zval(buf, &base->destructor.fcall->fci.function_name);
	}

	if (!(base->handler_flags & ZEND_ASYNC_POOL_F_HEALTHCHECK_INTERNAL) && base->healthcheck.fcall) {
		zend_get_gc_buffer_add_zval(buf, &base->healthcheck.fcall->fci.function_name);
	}

	if (!(base->handler_flags & ZEND_ASYNC_POOL_F_BEFORE_ACQUIRE_INTERNAL) && base->before_acquire.fcall) {
		zend_get_gc_buffer_add_zval(buf, &base->before_acquire.fcall->fci.function_name);
	}

	if (!(base->handler_flags & ZEND_ASYNC_POOL_F_BEFORE_RELEASE_INTERNAL) && base->before_release.fcall) {
		zend_get_gc_buffer_add_zval(buf, &base->before_release.fcall->fci.function_name);
	}

	/* GC for the circuit breaker strategy object (cycles through it otherwise
	 * leak the same way — Pool::setCircuitBreakerStrategy holds a ref). */
	if (!(base->handler_flags & ZEND_ASYNC_POOL_F_STRATEGY_INTERNAL) && base->strategy.object != NULL) {
		zend_get_gc_buffer_add_obj(buf, base->strategy.object);
	}

	zend_get_gc_buffer_use(buf, table, num);
	return NULL;
}

///////////////////////////////////////////////////////////////////////////////
// PHP Methods
///////////////////////////////////////////////////////////////////////////////

METHOD(__construct)
{
	zend_fcall_info fci_factory = { 0 }, fci_destructor = { 0 }, fci_healthcheck = { 0 };
	zend_fcall_info fci_before_acquire = { 0 }, fci_before_release = { 0 };
	zend_fcall_info_cache fcc_factory = { 0 }, fcc_destructor = { 0 }, fcc_healthcheck = { 0 };
	zend_fcall_info_cache fcc_before_acquire = { 0 }, fcc_before_release = { 0 };
	zend_long min = 0;
	zend_long max = 10;
	zend_long healthcheck_interval = 0;

	ZEND_PARSE_PARAMETERS_START(1, 8)
	Z_PARAM_FUNC(fci_factory, fcc_factory)
	Z_PARAM_OPTIONAL
	Z_PARAM_FUNC_OR_NULL(fci_destructor, fcc_destructor)
	Z_PARAM_FUNC_OR_NULL(fci_healthcheck, fcc_healthcheck)
	Z_PARAM_FUNC_OR_NULL(fci_before_acquire, fcc_before_acquire)
	Z_PARAM_FUNC_OR_NULL(fci_before_release, fcc_before_release)
	Z_PARAM_LONG(min)
	Z_PARAM_LONG(max)
	Z_PARAM_LONG(healthcheck_interval)
	ZEND_PARSE_PARAMETERS_END();

	/* An object carries one pool for its life. Nothing releases a replaced pool, so the first
	 * one is left with its resources and no owner, and PDO::getPool() caches this wrapper, so
	 * the handle's own object would describe a pool it never opened. */
	if (UNEXPECTED(THIS_POOL_OBJ->pool != NULL)) {
		zend_throw_exception(async_ce_pool_exception, "Pool is already initialized", 0);
		RETURN_THROWS();
	}

	if (min < 0) {
		zend_argument_value_error(6, "must be >= 0");
		RETURN_THROWS();
	}
	if (max < 1) {
		zend_argument_value_error(7, "must be >= 1");
		RETURN_THROWS();
	}
	if (min > max) {
		zend_argument_value_error(6, "must be <= max");
		RETURN_THROWS();
	}
	if (healthcheck_interval < 0) {
		zend_argument_value_error(8, "must be >= 0");
		RETURN_THROWS();
	}

	/* Copy fcalls */
	zend_fcall_t *factory = NULL;
	zend_fcall_t *destructor = NULL;
	zend_fcall_t *healthcheck = NULL;
	zend_fcall_t *before_acquire = NULL;
	zend_fcall_t *before_release = NULL;

	factory = ecalloc(1, sizeof(zend_fcall_t));
	factory->fci = fci_factory;
	factory->fci_cache = fcc_factory;
	ZVAL_COPY(&factory->fci.function_name, &fci_factory.function_name);

	if (ZEND_FCI_INITIALIZED(fci_destructor)) {
		destructor = ecalloc(1, sizeof(zend_fcall_t));
		destructor->fci = fci_destructor;
		destructor->fci_cache = fcc_destructor;
		ZVAL_COPY(&destructor->fci.function_name, &fci_destructor.function_name);
	}

	if (ZEND_FCI_INITIALIZED(fci_healthcheck)) {
		healthcheck = ecalloc(1, sizeof(zend_fcall_t));
		healthcheck->fci = fci_healthcheck;
		healthcheck->fci_cache = fcc_healthcheck;
		ZVAL_COPY(&healthcheck->fci.function_name, &fci_healthcheck.function_name);
	}

	if (ZEND_FCI_INITIALIZED(fci_before_acquire)) {
		before_acquire = ecalloc(1, sizeof(zend_fcall_t));
		before_acquire->fci = fci_before_acquire;
		before_acquire->fci_cache = fcc_before_acquire;
		ZVAL_COPY(&before_acquire->fci.function_name, &fci_before_acquire.function_name);
	}

	if (ZEND_FCI_INITIALIZED(fci_before_release)) {
		before_release = ecalloc(1, sizeof(zend_fcall_t));
		before_release->fci = fci_before_release;
		before_release->fci_cache = fcc_before_release;
		ZVAL_COPY(&before_release->fci.function_name, &fci_before_release.function_name);
	}

	async_pool_obj_t *obj = THIS_POOL_OBJ;
	obj->pool = zend_async_pool_create(factory,
									   destructor,
									   healthcheck,
									   before_acquire,
									   before_release,
									   (uint32_t) min,
									   (uint32_t) max,
									   (uint32_t) healthcheck_interval);

	if (EG(exception)) {
		RETURN_THROWS();
	}

	/* Set wrapper for strategy callbacks */
	obj->pool->base.wrapper = &obj->std;
}

METHOD(acquire)
{
	zend_long timeout = 0;

	ZEND_PARSE_PARAMETERS_START(0, 1)
	Z_PARAM_OPTIONAL
	Z_PARAM_LONG(timeout)
	ZEND_PARSE_PARAMETERS_END();

	ENSURE_COROUTINE_CONTEXT

	async_pool_obj_t *obj = THIS_POOL_OBJ;

	THROW_IF_NO_POOL(obj)

	THROW_IF_INTERNAL_RESOURCES(obj->pool)

	if (!zend_async_pool_acquire(obj->pool, return_value, timeout)) {
		RETURN_THROWS();
	}
}

METHOD(tryAcquire)
{
	ZEND_PARSE_PARAMETERS_NONE();

	async_pool_obj_t *obj = THIS_POOL_OBJ;

	if (obj->pool == NULL) {
		RETURN_NULL();
	}

	THROW_IF_INTERNAL_RESOURCES(obj->pool)

	if (!zend_async_pool_try_acquire(obj->pool, return_value)) {
		RETURN_NULL();
	}
}

METHOD(release)
{
	zval *resource;

	ZEND_PARSE_PARAMETERS_START(1, 1)
	Z_PARAM_ZVAL(resource)
	ZEND_PARSE_PARAMETERS_END();

	async_pool_obj_t *obj = THIS_POOL_OBJ;

	THROW_IF_NO_POOL(obj)

	THROW_IF_INTERNAL_RESOURCES(obj->pool)

	zend_async_pool_release(obj->pool, resource);
}

METHOD(close)
{
	ZEND_PARSE_PARAMETERS_NONE();

	async_pool_obj_t *obj = THIS_POOL_OBJ;

	if (obj->pool != NULL) {
		zend_async_pool_close(obj->pool);
	}
}

METHOD(isClosed)
{
	ZEND_PARSE_PARAMETERS_NONE();

	async_pool_obj_t *obj = THIS_POOL_OBJ;

	if (obj->pool == NULL) {
		RETURN_TRUE;
	}

	RETURN_BOOL(ZEND_ASYNC_POOL_IS_CLOSED(obj->pool));
}

METHOD(count)
{
	ZEND_PARSE_PARAMETERS_NONE();

	async_pool_obj_t *obj = THIS_POOL_OBJ;

	if (obj->pool == NULL) {
		RETURN_LONG(0);
	}

	RETURN_LONG(zend_async_pool_count(obj->pool));
}

METHOD(idleCount)
{
	ZEND_PARSE_PARAMETERS_NONE();

	async_pool_obj_t *obj = THIS_POOL_OBJ;

	if (obj->pool == NULL) {
		RETURN_LONG(0);
	}

	RETURN_LONG(zend_async_pool_idle_count(obj->pool));
}

METHOD(activeCount)
{
	ZEND_PARSE_PARAMETERS_NONE();

	async_pool_obj_t *obj = THIS_POOL_OBJ;

	if (obj->pool == NULL) {
		RETURN_LONG(0);
	}

	RETURN_LONG(zend_async_pool_active_count(obj->pool));
}

///////////////////////////////////////////////////////////////////////////////
// CircuitBreaker interface implementation
///////////////////////////////////////////////////////////////////////////////

METHOD(setCircuitBreakerStrategy)
{
	zend_object *strategy_obj = NULL;

	ZEND_PARSE_PARAMETERS_START(1, 1)
	Z_PARAM_OBJ_OF_CLASS_OR_NULL(strategy_obj, async_ce_circuit_breaker_strategy)
	ZEND_PARSE_PARAMETERS_END();

	async_pool_obj_t *obj = THIS_POOL_OBJ;

	THROW_IF_NO_POOL(obj)

	zend_async_pool_t *base = &obj->pool->base;

	/* Release old strategy if any */
	if (base->strategy.object != NULL) {
		OBJ_RELEASE(base->strategy.object);
		base->strategy.object = NULL;
	}

	/* Set new strategy */
	if (strategy_obj != NULL) {
		GC_ADDREF(strategy_obj);
		base->strategy.object = strategy_obj;
	}
}

METHOD(getState)
{
	ZEND_PARSE_PARAMETERS_NONE();

	async_pool_obj_t *obj = THIS_POOL_OBJ;

	THROW_IF_NO_POOL(obj)

	zend_async_pool_t *base = &obj->pool->base;

	zend_object *case_obj = NULL;
	switch (base->circuit_state) {
		case ZEND_ASYNC_CIRCUIT_STATE_ACTIVE:
			case_obj = zend_enum_get_case_cstr(async_ce_circuit_breaker_state, "ACTIVE");
			break;
		case ZEND_ASYNC_CIRCUIT_STATE_INACTIVE:
			case_obj = zend_enum_get_case_cstr(async_ce_circuit_breaker_state, "INACTIVE");
			break;
		case ZEND_ASYNC_CIRCUIT_STATE_RECOVERING:
			case_obj = zend_enum_get_case_cstr(async_ce_circuit_breaker_state, "RECOVERING");
			break;
	}

	if (case_obj != NULL) {
		RETURN_OBJ_COPY(case_obj);
	}

	RETURN_NULL();
}

METHOD(activate)
{
	ZEND_PARSE_PARAMETERS_NONE();

	async_pool_obj_t *obj = THIS_POOL_OBJ;

	THROW_IF_NO_POOL(obj)

	obj->pool->base.circuit_state = ZEND_ASYNC_CIRCUIT_STATE_ACTIVE;
}

METHOD(deactivate)
{
	ZEND_PARSE_PARAMETERS_NONE();

	async_pool_obj_t *obj = THIS_POOL_OBJ;

	THROW_IF_NO_POOL(obj)

	obj->pool->base.circuit_state = ZEND_ASYNC_CIRCUIT_STATE_INACTIVE;
}

METHOD(recover)
{
	ZEND_PARSE_PARAMETERS_NONE();

	async_pool_obj_t *obj = THIS_POOL_OBJ;

	THROW_IF_NO_POOL(obj)

	obj->pool->base.circuit_state = ZEND_ASYNC_CIRCUIT_STATE_RECOVERING;
}

///////////////////////////////////////////////////////////////////////////////
// Registration
///////////////////////////////////////////////////////////////////////////////

void async_register_pool_ce(void)
{
	async_ce_pool_exception = register_class_Async_PoolException(async_ce_async_exception);

	async_ce_pool = register_class_Async_Pool(zend_ce_countable, async_ce_circuit_breaker);
	async_ce_pool->create_object = async_pool_create_object;

	memcpy(&async_pool_handlers, &std_object_handlers, sizeof(zend_object_handlers));
	async_pool_handlers.offset = offsetof(async_pool_obj_t, std);
	async_pool_handlers.get_gc = async_pool_get_gc;
	async_pool_handlers.free_obj = async_pool_free_object;
	async_pool_handlers.clone_obj = NULL;
}