Appwrite/appwrite
1
0
Fork 0
mirror of https://github.com/appwrite/appwrite.git synced 2026-05-26 13:51:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use separate registry entry for worker pools

Browse Source
This commit is contained in:
Jake Barnby
2022-11-16 12:50:37 +13:00
parent dc6c8d9b9a
commit 860f077787
4 changed files with 192 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/init.php
+170 -2
View File
@@ -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;
app/realtime.php
+12 -12
View File
@@ -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();
}
});
composer.json
+1 -1
View File
@@ -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",
composer.lock
Generated
+9 -18
View File
@@ -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": {