Refactoring pools for concurrent operations

composer.json

@@ -30,12 +30,14 @@
     "require-dev": {
         "phpunit/phpunit": "11.*",
         "laravel/pint": "1.*",
-        "phpstan/phpstan": "1.*"
+        "phpstan/phpstan": "1.*",
+        "swoole/ide-helper": "5.1.2"
     },
     "suggests": {
         "ext-mongodb": "Needed to support MongoDB database pools",
         "ext-redis": "Needed to support Redis cache pools",
-        "ext-pdo": "Needed to support MariaDB, MySQL or SQLite database pools"
+        "ext-pdo": "Needed to support MariaDB, MySQL or SQLite database pools",
+        "ext-swoole" : "Needed to support Swoole based pool adapter"
     },
     "config": {
         "platform": {

src/Pools/Adapter.php

@@ -0,0 +1,27 @@
+<?php
+
+namespace Utopia\Pools;
+
+abstract class Adapter
+{
+    abstract public function initialize(int $size): static;
+
+    abstract public function push(mixed $connection): static;
+
+    /**
+     * @param int $timeout
+     * @return mixed
+     */
+    abstract public function pop(int $timeout): mixed;
+
+    abstract public function count(): int;
+
+    /**
+     * Execute a callback with lock protection if the adapter supports it
+     *
+     * @param callable $callback
+     * @param int $timeout Timeout in seconds
+     * @return mixed
+     */
+    abstract public function synchronized(callable $callback, int $timeout): mixed;
+}

src/Pools/Adapter/Stack.php

@@ -0,0 +1,68 @@
+<?php
+
+namespace Utopia\Pools\Adapter;
+
+use Utopia\Pools\Adapter;
+
+class Stack extends Adapter
+{
+    /** @var array<mixed> $pool */
+    protected array $pool = [];
+
+    /**
+     * Initialize the stack-based pool.
+     *
+     * Note:
+     * - `$size` is accepted for API compatibility with other pool adapters.
+     * - The stack adapter does NOT enforce capacity limits.
+     * - `$size` is ignored because the pool is backed by a simple array.
+     *
+     * @param int $size Ignored by the stack adapter.
+     */
+    public function initialize(int $size): static
+    {
+        $this->pool = [];
+        return $this;
+    }
+
+    public function push(mixed $connection): static
+    {
+        // Push connection to pool
+        $this->pool[] = $connection;
+        return $this;
+    }
+
+    /**
+     * Pop an item from the stack.
+     *
+     * Note: The stack adapter does not support blocking operations.
+     * The `$timeout` parameter is ignored.
+     *
+     * @param int $timeout Ignored by the stack adapter.
+     * @return mixed|null Returns the popped item, or null if the stack is empty.
+     */
+    public function pop(int $timeout): mixed
+    {
+        return array_pop($this->pool);
+    }
+
+    public function count(): int
+    {
+        return count($this->pool);
+    }
+
+    /**
+     * Executes the callback without acquiring a lock.
+     *
+     * This implementation does not provide mutual exclusion.
+     * The `$timeout` parameter is ignored.
+     *
+     * @param callable $callback Callback to execute.
+     * @param int $timeout Ignored.
+     * @return mixed The value returned by the callback.
+     */
+    public function synchronized(callable $callback, int $timeout): mixed
+    {
+        return $callback();
+    }
+}

src/Pools/Adapter/Swoole.php

@@ -0,0 +1,75 @@
+<?php
+
+namespace Utopia\Pools\Adapter;
+
+use Utopia\Pools\Adapter;
+use Swoole\Coroutine\Channel;
+use Swoole\Lock;
+
+class Swoole extends Adapter
+{
+    protected Channel $pool;
+
+    /** @var Lock $lock */
+    protected Lock $lock;
+    public function initialize(int $size): static
+    {
+        $this->pool = new Channel($size);
+
+        $this->lock = new Lock(SWOOLE_MUTEX);
+
+        return $this;
+    }
+
+    public function push(mixed $connection): static
+    {
+        // Push connection to channel
+        $this->pool->push($connection);
+        return $this;
+    }
+
+    /**
+     * Pop an item from the pool.
+     *
+     * @param int $timeout Timeout in seconds. Use 0 for non-blocking pop.
+     * @return mixed|false Returns the pooled value, or false if the pool is empty
+     *                     or the timeout expires.
+     */
+    public function pop(int $timeout): mixed
+    {
+        return $this->pool->pop($timeout);
+    }
+
+    public function count(): int
+    {
+        $length = $this->pool->length();
+        return is_int($length) ? $length : 0;
+    }
+
+    /**
+     * Executes a callback while holding a lock.
+     *
+     * The lock is acquired before invoking the callback and is always released
+     * afterward, even if the callback throws an exception.
+     *
+     * @param callable $callback Callback to execute within the critical section.
+     * @param int $timeout Maximum time (in seconds) to wait for the lock.
+     * @return mixed The value returned by the callback.
+     *
+     * @throws \RuntimeException If the lock cannot be acquired within the timeout.
+    */
+    public function synchronized(callable $callback, int $timeout): mixed
+    {
+        $acquired = $this->lock->lockwait($timeout);
+
+        if (!$acquired) {
+            throw new \RuntimeException("Failed to acquire lock within {$timeout} seconds");
+        }
+
+        try {
+            return $callback();
+        } finally {
+            $this->lock->unlock();
+        }
+    }
+}