From 860f077787aba9fcf35e0ff37c348f63b2ca8c4c Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Wed, 16 Nov 2022 12:50:37 +1300 Subject: [PATCH] Use separate registry entry for worker pools --- app/init.php | 172 ++++++++++++++++++++++++++++++++++++++++++++++- app/realtime.php | 24 +++---- composer.json | 2 +- composer.lock | 27 +++----- 4 files changed, 192 insertions(+), 33 deletions(-) diff --git a/app/init.php b/app/init.php index af87960325..ba7d2c7a03 100644 --- a/app/init.php +++ b/app/init.php @@ -497,7 +497,7 @@ $register->set('logger', function () { $adapter = new $classname($providerConfig); return new Logger($adapter); }); -$register->set('pools', function ($size = APP_DEFAULT_POOL_SIZE) { +$register->set('pools', function () { $group = new Group(); $fallbackForDB = AppwriteURL::unparse([ @@ -621,7 +621,175 @@ $register->set('pools', function ($size = APP_DEFAULT_POOL_SIZE) { break; } - $pool = new Pool($name, $size, function () use ($type, $resource, $dsn) { + $pool = new Pool($name, APP_DEFAULT_POOL_SIZE, function () use ($type, $resource, $dsn) { + // Get Adapter + $adapter = null; + + switch ($type) { + case 'database': + $adapter = match ($dsn->getScheme()) { + 'mariadb' => new MariaDB($resource()), + 'mysql' => new MySQL($resource()), + default => null + }; + + $adapter->setDefaultDatabase($dsn->getDatabase()); + + break; + case 'queue': + $adapter = $resource(); + break; + case 'pubsub': + $adapter = $resource(); + break; + case 'cache': + $adapter = match ($dsn->getScheme()) { + 'redis' => new RedisCache($resource()), + default => null + }; + break; + + default: + throw new Exception(Exception::GENERAL_SERVER_ERROR, "Server error: Missing adapter implementation."); + break; + } + + return $adapter; + }); + + $group->add($pool); + } + + Config::setParam('pools-' . $key, $config); + } + + return $group; +}); +$register->set('workerPools', function () { + $group = new Group(); + + $fallbackForDB = AppwriteURL::unparse([ + 'scheme' => 'mariadb', + 'host' => App::getEnv('_APP_DB_HOST', 'mariadb'), + 'port' => App::getEnv('_APP_DB_PORT', '3306'), + 'user' => App::getEnv('_APP_DB_USER', ''), + 'pass' => App::getEnv('_APP_DB_PASS', ''), + ]); + $fallbackForRedis = AppwriteURL::unparse([ + 'scheme' => 'redis', + 'host' => App::getEnv('_APP_REDIS_HOST', 'redis'), + 'port' => App::getEnv('_APP_REDIS_PORT', '6379'), + 'user' => App::getEnv('_APP_REDIS_USER', ''), + 'pass' => App::getEnv('_APP_REDIS_PASS', ''), + ]); + + $connections = [ + 'console' => [ + 'type' => 'database', + 'dsns' => App::getEnv('_APP_CONNECTIONS_DB_CONSOLE', $fallbackForDB), + 'multiple' => false, + 'schemes' => ['mariadb', 'mysql'], + ], + 'database' => [ + 'type' => 'database', + 'dsns' => App::getEnv('_APP_CONNECTIONS_DB_PROJECT', $fallbackForDB), + 'multiple' => true, + 'schemes' => ['mariadb', 'mysql'], + ], + 'queue' => [ + 'type' => 'queue', + 'dsns' => App::getEnv('_APP_CONNECTIONS_QUEUE', $fallbackForRedis), + 'multiple' => false, + 'schemes' => ['redis'], + ], + 'pubsub' => [ + 'type' => 'pubsub', + 'dsns' => App::getEnv('_APP_CONNECTIONS_PUBSUB', $fallbackForRedis), + 'multiple' => false, + 'schemes' => ['redis'], + ], + 'cache' => [ + 'type' => 'cache', + 'dsns' => App::getEnv('_APP_CONNECTIONS_CACHE', $fallbackForRedis), + 'multiple' => true, + 'schemes' => ['redis'], + ], + ]; + + foreach ($connections as $key => $connection) { + $type = $connection['type'] ?? ''; + $dsns = $connection['dsns'] ?? ''; + $multipe = $connection['multiple'] ?? false; + $schemes = $connection['schemes'] ?? []; + $config = []; + $dsns = explode(',', $connection['dsns'] ?? ''); + + foreach ($dsns as &$dsn) { + $dsn = explode('=', $dsn); + $name = ($multipe) ? $key . '_' . $dsn[0] : $key; + $dsn = $dsn[1] ?? ''; + $config[] = $name; + + if (empty($dsn)) { + //throw new Exception(Exception::GENERAL_SERVER_ERROR, "Missing value for DSN connection in {$key}"); + continue; + } + + $dsn = new DSN($dsn); + $dsnHost = $dsn->getHost(); + $dsnPort = $dsn->getPort(); + $dsnUser = $dsn->getUser(); + $dsnPass = $dsn->getPassword(); + $dsnScheme = $dsn->getScheme(); + $dsnDatabase = $dsn->getDatabase(); + + if (!in_array($dsnScheme, $schemes)) { + throw new Exception(Exception::GENERAL_SERVER_ERROR, "Invalid console database scheme"); + } + + /** + * Get Resource + * + * Creation could be reused accross connection types like database, cache, queue, etc. + * + * Resource assignment to an adapter will happen below. + */ + + switch ($dsnScheme) { + case 'mysql': + case 'mariadb': + $resource = function () use ($dsnHost, $dsnPort, $dsnUser, $dsnPass, $dsnDatabase) { + return new PDOProxy(function () use ($dsnHost, $dsnPort, $dsnUser, $dsnPass, $dsnDatabase) { + return new PDO("mysql:host={$dsnHost};port={$dsnPort};dbname={$dsnDatabase};charset=utf8mb4", $dsnUser, $dsnPass, array( + PDO::ATTR_TIMEOUT => 3, // Seconds + PDO::ATTR_PERSISTENT => true, + PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC, + PDO::ATTR_ERRMODE => App::isDevelopment() ? PDO::ERRMODE_WARNING : PDO::ERRMODE_SILENT, // If in production mode, warnings are not displayed + PDO::ATTR_EMULATE_PREPARES => true, + PDO::ATTR_STRINGIFY_FETCHES => true + )); + }); + }; + break; + case 'redis': + $resource = function () use ($dsnHost, $dsnPort, $dsnPass) { + $redis = new Redis(); + @$redis->pconnect($dsnHost, (int)$dsnPort); + if ($dsnPass) { + $redis->auth($dsnPass); + } + $redis->setOption(Redis::OPT_READ_TIMEOUT, -1); + + return $redis; + }; + break; + + default: + throw new Exception(Exception::GENERAL_SERVER_ERROR, "Invalid scheme"); + break; + } + + $pool = new Pool($name, getWorkerPoolSize(), function () use ($type, $resource, $dsn) { // Get Adapter $adapter = null; diff --git a/app/realtime.php b/app/realtime.php index 25c0fee858..22c74168dc 100644 --- a/app/realtime.php +++ b/app/realtime.php @@ -37,7 +37,7 @@ function getConsoleDB(): Database global $register; /** @var \Utopia\Pools\Group $pools */ - $pools = $register->get('pools', args: [getWorkerPoolSize()]); + $pools = $register->get('workerPools'); $dbAdapter = $pools ->get('console') @@ -57,7 +57,7 @@ function getProjectDB(Document $project): Database global $register; /** @var \Utopia\Pools\Group $pools */ - $pools = $register->get('pools', args: [getWorkerPoolSize()]); + $pools = $register->get('workerPools'); if ($project->isEmpty() || $project->getId() === 'console') { return getConsoleDB(); @@ -79,7 +79,7 @@ function getCache(): Cache { global $register; - $pools = $register->get('pools'); /** @var \Utopia\Pools\Group $pools */ + $pools = $register->get('workerPools'); /** @var \Utopia\Pools\Group $pools */ $list = Config::getParam('pools-cache', []); $adapters = []; @@ -187,7 +187,7 @@ $server->onStart(function () use ($stats, $register, $containerId, &$statsDocume sleep(DATABASE_RECONNECT_SLEEP); } } while (true); - $register->get('pools')->reclaim(); + $register->get('workerPools')->reclaim(); }); /** @@ -213,7 +213,7 @@ $server->onStart(function () use ($stats, $register, $containerId, &$statsDocume } catch (\Throwable $th) { call_user_func($logError, $th, "updateWorkerDocument"); } finally { - $register->get('pools')->reclaim(); + $register->get('workerPools')->reclaim(); } }); }); @@ -274,7 +274,7 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats, ])); } - $register->get('pools')->reclaim(); + $register->get('workerPools')->reclaim(); } /** * Sending test message for SDK E2E tests every 5 seconds. @@ -309,7 +309,7 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats, } $start = time(); - $redis = $register->get('pools')->get('pubsub')->pop()->getResource(); /** @var Redis $redis */ + $redis = $register->get('workerPools')->get('pubsub')->pop()->getResource(); /** @var Redis $redis */ $redis->setOption(Redis::OPT_READ_TIMEOUT, -1); if ($redis->ping(true)) { @@ -338,7 +338,7 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats, $realtime->subscribe($projectId, $connection, $roles, $realtime->connections[$connection]['channels']); - $register->get('pools')->reclaim(); + $register->get('workerPools')->reclaim(); } } @@ -370,7 +370,7 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats, sleep(DATABASE_RECONNECT_SLEEP); continue; } finally { - $register->get('pools')->reclaim(); + $register->get('workerPools')->reclaim(); } } @@ -384,7 +384,7 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, Console::info("Connection open (user: {$connection})"); - App::setResource('pools', fn() => $register->get('pools')); + App::setResource('pools', fn() => $register->get('workerPools')); App::setResource('request', fn() => $request); App::setResource('response', fn() => $response); @@ -480,7 +480,7 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, Console::error('[Error] Message: ' . $response['data']['message']); } } finally { - $register->get('pools')->reclaim(); + $register->get('workerPools')->reclaim(); } }); @@ -575,7 +575,7 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re $server->close($connection, $th->getCode()); } } finally { - $register->get('pools')->reclaim(); + $register->get('workerPools')->reclaim(); } }); diff --git a/composer.json b/composer.json index 36aed0d96b..3dd374b8e1 100644 --- a/composer.json +++ b/composer.json @@ -59,7 +59,7 @@ "utopia-php/platform": "0.3.*", "utopia-php/pools": "0.4.*", "utopia-php/preloader": "0.2.*", - "utopia-php/registry": "dev-feat-allow-params as 0.5.0", + "utopia-php/registry": "0.5.0", "utopia-php/storage": "0.11.*", "utopia-php/swoole": "0.5.*", "utopia-php/websocket": "0.1.0", diff --git a/composer.lock b/composer.lock index f042279e4f..2e39916a0c 100644 --- a/composer.lock +++ b/composer.lock @@ -4,7 +4,7 @@ "Read more about it at https://getcomposer.org/doc/01-basic-usage.md#installing-dependencies", "This file is @generated automatically" ], - "content-hash": "b125039c64ae4cbe0d2a1b57322d0ebe", + "content-hash": "bf3e2ed6ee8e49ab74af97b368b89a63", "packages": [ { "name": "adhocore/jwt", @@ -2359,16 +2359,16 @@ }, { "name": "utopia-php/registry", - "version": "dev-feat-allow-params", + "version": "0.5.0", "source": { "type": "git", "url": "https://github.com/utopia-php/registry.git", - "reference": "6c571f8f4127094b3af8909d1b595fd6b937255d" + "reference": "bedc4ed54527b2803e6dfdccc39449f98522b70d" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/utopia-php/registry/zipball/6c571f8f4127094b3af8909d1b595fd6b937255d", - "reference": "6c571f8f4127094b3af8909d1b595fd6b937255d", + "url": "https://api.github.com/repos/utopia-php/registry/zipball/bedc4ed54527b2803e6dfdccc39449f98522b70d", + "reference": "bedc4ed54527b2803e6dfdccc39449f98522b70d", "shasum": "" }, "require": { @@ -2405,9 +2405,9 @@ ], "support": { "issues": "https://github.com/utopia-php/registry/issues", - "source": "https://github.com/utopia-php/registry/tree/feat-allow-params" + "source": "https://github.com/utopia-php/registry/tree/0.5.0" }, - "time": "2022-11-09T02:23:35+00:00" + "time": "2021-03-10T10:45:22+00:00" }, { "name": "utopia-php/storage", @@ -5215,18 +5215,9 @@ "time": "2022-09-28T08:42:51+00:00" } ], - "aliases": [ - { - "package": "utopia-php/registry", - "version": "dev-feat-allow-params", - "alias": "0.5.0", - "alias_normalized": "0.5.0.0" - } - ], + "aliases": [], "minimum-stability": "stable", - "stability-flags": { - "utopia-php/registry": 20 - }, + "stability-flags": [], "prefer-stable": false, "prefer-lowest": false, "platform": {