Refactoring pools for concurrent operations

composer.json

@@ -25,17 +25,20 @@
     },
     "require": {
         "php": ">=8.4",
-        "utopia-php/telemetry": "0.1.*"
+        "utopia-php/telemetry": "0.1.*",
+        "ext-swoole": "*"
     },
     "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 fill(int $size, mixed $value): 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
+     *
+     * @param callable $callback
+     * @param float $timeout Timeout in seconds
+     * @return mixed
+     */
+    abstract public function withLock(callable $callback, int $timeout): mixed;
+}

src/Pools/Adapter/Stack.php

@@ -0,0 +1,40 @@
+<?php
+
+namespace Utopia\Pools\Adapter;
+
+use Utopia\Pools\Adapter;
+
+class Stack extends Adapter
+{
+    /** @var array<mixed> $pool */
+    protected array $pool = [];
+
+    public function fill(int $size, mixed $value): static
+    {
+        // Initialize empty pool (no pre-filling)
+        $this->pool = [];
+        return $this;
+    }
+
+    public function push(mixed $connection): static
+    {
+        // Push connection to pool
+        $this->pool[] = $connection;
+        return $this;
+    }
+
+    public function pop(int $timeout): mixed
+    {
+        return array_pop($this->pool);
+    }
+
+    public function count(): int
+    {
+        return count($this->pool);
+    }
+
+    public function withLock(callable $callback, int $timeout): mixed
+    {
+        return $callback();
+    }
+}

src/Pools/Adapter/Swoole.php

@@ -0,0 +1,62 @@
+<?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 fill(int $size, mixed $value): static
+    {
+        // Create empty channel with capacity (no pre-filling)
+        $this->pool = new Channel($size);
+
+        // Initialize lock for thread-safe operations
+        $this->lock = new Lock(SWOOLE_MUTEX);
+
+        return $this;
+    }
+
+    public function push(mixed $connection): static
+    {
+        // Push connection to channel
+        $this->pool->push($connection);
+        return $this;
+    }
+
+    public function pop(int $timeout): mixed
+    {
+        $result = $this->pool->pop($timeout);
+
+        // if pool is empty or timeout occured => result will be false
+        return $result;
+    }
+
+
+    public function count(): int
+    {
+        return (int) $this->pool->length();
+    }
+
+    public function withLock(callable $callback, int $timeout): mixed
+    {
+        // Acquire lock for thread-safe operations with timeout to prevent deadlock
+        $acquired = $this->lock->lockwait($timeout);
+
+        if (!$acquired) {
+            throw new \RuntimeException("Failed to acquire lock within {$timeout} seconds");
+        }
+
+        try {
+            return $callback();
+        } finally {
+            $this->lock->unlock();
+        }
+    }
+}