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,18 @@
+<?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;
+}

src/Pools/Adapter/Stack.php

@@ -0,0 +1,35 @@
+<?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);
+    }
+}

src/Pools/Adapter/Swoole.php

@@ -0,0 +1,46 @@
+<?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();
+    }
+}

src/Pools/Pool.php

@@ -3,6 +3,7 @@
 namespace Utopia\Pools;
 
 use Exception;
+use Utopia\Pools\Adapter as PoolAdapter;
 use Utopia\Telemetry\Adapter as Telemetry;
 use Utopia\Telemetry\Adapter\None as NoTelemetry;
 use Utopia\Telemetry\Gauge;
@@ -13,6 +14,7 @@
  */
 class Pool
 {
+    public const POP_TIMEOUT_IN_SECODNS = 3;
     /**
      * @var callable
      */
@@ -38,16 +40,18 @@ class Pool
      */
     protected int $retrySleep = 1; // seconds
 
-    /**
-     * @var array<Connection<TResource>|true>
-     */
-    protected array $pool = [];
+    protected PoolAdapter $pool;
 
     /**
      * @var array<string, Connection<TResource>>
      */
     protected array $active = [];
 
+    /**
+     * Total number of connections created
+     */
+    protected int $connectionsCreated = 0;
+
     private Gauge $telemetryOpenConnections;
     private Gauge $telemetryActiveConnections;
     private Gauge $telemetryIdleConnections;
@@ -58,14 +62,17 @@ class Pool
     private array $telemetryAttributes;
 
     /**
+     * @param PoolAdapter $adapter
      * @param string $name
      * @param int $size
      * @param callable(): TResource $init
      */
-    public function __construct(protected string $name, protected int $size, callable $init)
+    public function __construct(PoolAdapter $adapter, protected string $name, protected int $size, callable $init)
     {
         $this->init = $init;
-        $this->pool = array_fill(0, $this->size, true);
+        $this->pool = $adapter;
+        // Initialize empty channel (no pre-filling for lazy initialization)
+        $this->pool->fill($this->size, null);
         $this->setTelemetry(new NoTelemetry());
     }
 
@@ -223,48 +230,29 @@ public function pop(): Connection
         try {
             do {
                 $attempts++;
-                $connection = array_pop($this->pool);
+                // If pool is empty and size limit not reached, create new connection
+                if ($this->pool->count() === 0 && $this->connectionsCreated < $this->size) {
+                    $connection = $this->createConnection();
+                    $this->active[$connection->getID()] = $connection;
+                    return $connection;
+                }
 
-                if (is_null($connection)) {
+                $connection = $this->pool->pop(self::POP_TIMEOUT_IN_SECODNS);
+
+                if ($connection === false || $connection === null) {
                     if ($attempts >= $this->getRetryAttempts()) {
                         throw new Exception("Pool '{$this->name}' is empty (size {$this->size})");
                     }
 
                     $totalSleepTime += $this->getRetrySleep();
                     sleep($this->getRetrySleep());
                 } else {
-                    break;
-                }
-            } while ($attempts < $this->getRetryAttempts());
-
-            if ($connection === true) { // Pool has space, create connection
-                $attempts = 0;
-
-                do {
-                    try {
-                        $attempts++;
-                        $connection = new Connection(($this->init)());
-                        break; // leave loop if successful
-                    } catch (\Exception $e) {
-                        if ($attempts >= $this->getReconnectAttempts()) {
-                            throw new \Exception('Failed to create connection: ' . $e->getMessage());
-                        }
-                        $totalSleepTime += $this->getReconnectSleep();
-                        sleep($this->getReconnectSleep());
+                    if ($connection instanceof Connection) {
+                        $this->active[$connection->getID()] = $connection;
+                        return $connection;
                     }
-                } while ($attempts < $this->getReconnectAttempts());
-            }
-
-            if ($connection instanceof Connection) { // connection is available, return it
-                if (empty($connection->getID())) {
-                    $connection->setID($this->getName() . '-' . uniqid());
                 }
-
-                $connection->setPool($this);
-
-                $this->active[$connection->getID()] = $connection;
-                return $connection;
-            }
+            } while ($attempts < $this->getRetryAttempts());
 
             throw new Exception('Failed to get a connection from the pool');
         } finally {
@@ -273,14 +261,55 @@ public function pop(): Connection
         }
     }
 
+    /**
+     * Create a new connection
+     *
+     * @return Connection<TResource>
+     * @throws \Exception
+     */
+    protected function createConnection(): Connection
+    {
+        $this->connectionsCreated++;
+
+        $connection = null;
+        $attempts = 0;
+        do {
+            try {
+                $attempts++;
+                $connection = new Connection(($this->init)());
+                break;
+            } catch (\Exception $e) {
+                if ($attempts >= $this->getReconnectAttempts()) {
+                    $this->connectionsCreated--;
+                    throw new \Exception('Failed to create connection: ' . $e->getMessage());
+                }
+                sleep($this->getReconnectSleep());
+            }
+        } while ($attempts < $this->getReconnectAttempts());
+
+        if ($connection === null) {
+            $this->connectionsCreated--;
+            throw new \Exception('Failed to create connection');
+        }
+
+        if (empty($connection->getID())) {
+            $connection->setID($this->getName() . '-' . uniqid());
+        }
+
+        $connection->setPool($this);
+
+        return $connection;
+    }
+
     /**
      * @param Connection<TResource> $connection
      * @return $this<TResource>
      */
     public function push(Connection $connection): static
     {
         try {
-            $this->pool[] = $connection;
+            // Push the actual connection back to the pool
+            $this->pool->push($connection);
             unset($this->active[$connection->getID()]);
 
             return $this;
@@ -290,11 +319,14 @@ public function push(Connection $connection): static
     }
 
     /**
+     * Returns the number of available connections (idle + not yet created)
+     *
      * @return int
      */
     public function count(): int
     {
-        return count($this->pool);
+        // Available = idle connections in pool + connections not yet created
+        return $this->pool->count() + ($this->size - $this->connectionsCreated);
     }
 
     /**
@@ -323,14 +355,27 @@ public function destroy(?Connection $connection = null): static
     {
         try {
             if ($connection !== null) {
-                $this->pool[] = true;
+                $this->connectionsCreated--;
                 unset($this->active[$connection->getID()]);
+
+                // Create a new connection to maintain pool size
+                if ($this->connectionsCreated < $this->size) {
+                    $newConnection = $this->createConnection();
+                    $this->pool->push($newConnection);
+                }
+
                 return $this;
             }
 
             foreach ($this->active as $connection) {
-                $this->pool[] = true;
+                $this->connectionsCreated--;
                 unset($this->active[$connection->getID()]);
+
+                // Create a new connection to maintain pool size
+                if ($this->connectionsCreated < $this->size) {
+                    $newConnection = $this->createConnection();
+                    $this->pool->push($newConnection);
+                }
             }
 
             return $this;
@@ -344,26 +389,27 @@ public function destroy(?Connection $connection = null): static
      */
     public function isEmpty(): bool
     {
-        return empty($this->pool);
+        return $this->pool->count() === 0;
     }
 
     /**
      * @return bool
      */
     public function isFull(): bool
     {
-        return count($this->pool) === $this->size;
+        // Pool is full when all possible connections are available (idle or not created yet)
+        return count($this->active) === 0;
     }
 
     private function recordPoolTelemetry(): void
     {
-        // Connections get removed from $this->pool when they are active
         $activeConnections = count($this->active);
-        $existingConnections = count($this->pool);
-        $idleConnections = count(array_filter($this->pool, fn ($data) => $data instanceof Connection));
+        $idleConnections = $this->pool->count(); // Connections in the pool (idle)
+        $openConnections = $activeConnections + $idleConnections; // Total connections in use or available
+
         $this->telemetryActiveConnections->record($activeConnections, $this->telemetryAttributes);
         $this->telemetryIdleConnections->record($idleConnections, $this->telemetryAttributes);
-        $this->telemetryOpenConnections->record($activeConnections + $idleConnections, $this->telemetryAttributes);
-        $this->telemetryPoolCapacity->record($activeConnections + $existingConnections, $this->telemetryAttributes);
+        $this->telemetryOpenConnections->record($openConnections, $this->telemetryAttributes);
+        $this->telemetryPoolCapacity->record($activeConnections + $this->pool->count(), $this->telemetryAttributes);
     }
 }