Revert "Reclaim only used connections for workers"
This reverts commit fefb60557d.
# Conflicts:
# app/worker.php
This commit is contained in:
Jake Barnby
parent
9a3f6e7f71
commit
3859f037a1
|
|
@ -16,7 +16,6 @@ use Appwrite\Event\Phone;
|
|||
use Appwrite\Event\Usage;
|
||||
use Appwrite\Event\UsageDump;
|
||||
use Appwrite\Platform\Appwrite;
|
||||
use Appwrite\Utopia\Pools\Connections;
|
||||
use Swoole\Runtime;
|
||||
use Utopia\App;
|
||||
use Utopia\Cache\Adapter\Sharding;
|
||||
|
|
@ -37,32 +36,25 @@ use Utopia\Pools\Group;
|
|||
use Utopia\Queue\Connection;
|
||||
|
||||
Authorization::disable();
|
||||
Runtime::enableCoroutine();
|
||||
Runtime::enableCoroutine(SWOOLE_HOOK_ALL);
|
||||
|
||||
Server::setResource('register', fn () => $register);
|
||||
|
||||
Server::setResource('connections', function () {
|
||||
return new Connections();
|
||||
});
|
||||
|
||||
Server::setResource('dbForConsole', function (Cache $cache, Registry $register, Connections $connections) {
|
||||
Server::setResource('dbForConsole', function (Cache $cache, Registry $register) {
|
||||
$pools = $register->get('pools');
|
||||
|
||||
$connection = $pools
|
||||
$database = $pools
|
||||
->get('console')
|
||||
->pop();
|
||||
|
||||
$connections->add($connection);
|
||||
|
||||
$database = $connection->getResource();
|
||||
->pop()
|
||||
->getResource()
|
||||
;
|
||||
|
||||
$adapter = new Database($database, $cache);
|
||||
$adapter->setNamespace('_console');
|
||||
|
||||
return $adapter;
|
||||
}, ['cache', 'register', 'connections']);
|
||||
}, ['cache', 'register']);
|
||||
|
||||
Server::setResource('dbForProject', function (Cache $cache, Registry $register, Message $message, Database $dbForConsole, Connections $connections) {
|
||||
Server::setResource('dbForProject', function (Cache $cache, Registry $register, Message $message, Database $dbForConsole) {
|
||||
$payload = $message->getPayload() ?? [];
|
||||
$project = new Document($payload['project'] ?? []);
|
||||
|
||||
|
|
@ -71,19 +63,16 @@ Server::setResource('dbForProject', function (Cache $cache, Registry $register,
|
|||
}
|
||||
|
||||
$pools = $register->get('pools');
|
||||
|
||||
$connection = $pools
|
||||
$database = $pools
|
||||
->get($project->getAttribute('database'))
|
||||
->pop();
|
||||
|
||||
$database = $connection->getResource();
|
||||
|
||||
$connections->add($connection);
|
||||
->pop()
|
||||
->getResource()
|
||||
;
|
||||
|
||||
$adapter = new Database($database, $cache);
|
||||
$adapter->setNamespace('_' . $project->getInternalId());
|
||||
return $adapter;
|
||||
}, ['cache', 'register', 'message', 'dbForConsole', 'connections']);
|
||||
}, ['cache', 'register', 'message', 'dbForConsole']);
|
||||
|
||||
Server::setResource('project', function (Message $message, Database $dbForConsole) {
|
||||
$payload = $message->getPayload() ?? [];
|
||||
|
|
@ -92,14 +81,14 @@ Server::setResource('project', function (Message $message, Database $dbForConsol
|
|||
if ($project->getId() === 'console') {
|
||||
return $project;
|
||||
}
|
||||
|
||||
return $dbForConsole->getDocument('projects', $project->getId());
|
||||
;
|
||||
}, ['message', 'dbForConsole']);
|
||||
|
||||
Server::setResource('getProjectDB', function (Group $pools, Database $dbForConsole, Cache $cache, Connections $connections) {
|
||||
Server::setResource('getProjectDB', function (Group $pools, Database $dbForConsole, $cache) {
|
||||
$databases = []; // TODO: @Meldiron This should probably be responsibility of utopia-php/pools
|
||||
|
||||
return function (Document $project) use ($pools, $dbForConsole, $cache, $connections, &$databases): Database {
|
||||
return function (Document $project) use ($pools, $dbForConsole, $cache, &$databases): Database {
|
||||
if ($project->isEmpty() || $project->getId() === 'console') {
|
||||
return $dbForConsole;
|
||||
}
|
||||
|
|
@ -112,13 +101,10 @@ Server::setResource('getProjectDB', function (Group $pools, Database $dbForConso
|
|||
return $database;
|
||||
}
|
||||
|
||||
$dbConnection = $pools
|
||||
$dbAdapter = $pools
|
||||
->get($databaseName)
|
||||
->pop();
|
||||
|
||||
$dbAdapter = $dbConnection->getResource();
|
||||
|
||||
$connections->add($dbConnection);
|
||||
->pop()
|
||||
->getResource();
|
||||
|
||||
$database = new Database($dbAdapter, $cache);
|
||||
|
||||
|
|
@ -128,7 +114,7 @@ Server::setResource('getProjectDB', function (Group $pools, Database $dbForConso
|
|||
|
||||
return $database;
|
||||
};
|
||||
}, ['pools', 'dbForConsole', 'cache', 'connections']);
|
||||
}, ['pools', 'dbForConsole', 'cache']);
|
||||
|
||||
Server::setResource('abuseRetention', function () {
|
||||
return DateTime::addSeconds(new \DateTime(), -1 * App::getEnv('_APP_MAINTENANCE_RETENTION_ABUSE', 86400));
|
||||
|
|
@ -142,104 +128,82 @@ Server::setResource('executionRetention', function () {
|
|||
return DateTime::addSeconds(new \DateTime(), -1 * App::getEnv('_APP_MAINTENANCE_RETENTION_EXECUTION', 1209600));
|
||||
});
|
||||
|
||||
Server::setResource('cache', function (Registry $register, Connections $connections) {
|
||||
Server::setResource('cache', function (Registry $register) {
|
||||
$pools = $register->get('pools');
|
||||
$list = Config::getParam('pools-cache', []);
|
||||
$adapters = [];
|
||||
|
||||
foreach ($list as $value) {
|
||||
$connection = $pools
|
||||
$adapters[] = $pools
|
||||
->get($value)
|
||||
->pop();
|
||||
|
||||
$connections->add($connection);
|
||||
|
||||
$adapters[] = $connection->getResource();
|
||||
->pop()
|
||||
->getResource()
|
||||
;
|
||||
}
|
||||
|
||||
return new Cache(new Sharding($adapters));
|
||||
}, ['register', 'connections']);
|
||||
|
||||
}, ['register']);
|
||||
Server::setResource('log', fn() => new Log());
|
||||
|
||||
Server::setResource('queueForUsage', function (Connection $queue) {
|
||||
return new Usage($queue);
|
||||
}, ['queue']);
|
||||
|
||||
Server::setResource('queueForUsageDump', function (Connection $queue) {
|
||||
return new UsageDump($queue);
|
||||
}, ['queue']);
|
||||
|
||||
Server::setResource('queue', function (Group $pools) {
|
||||
return $pools->get('queue')->pop()->getResource();
|
||||
}, ['pools']);
|
||||
|
||||
Server::setResource('queueForDatabase', function (Connection $queue) {
|
||||
return new EventDatabase($queue);
|
||||
}, ['queue']);
|
||||
|
||||
Server::setResource('queueForMessaging', function (Connection $queue) {
|
||||
return new Phone($queue);
|
||||
}, ['queue']);
|
||||
|
||||
Server::setResource('queueForMails', function (Connection $queue) {
|
||||
return new Mail($queue);
|
||||
}, ['queue']);
|
||||
|
||||
Server::setResource('queueForBuilds', function (Connection $queue) {
|
||||
return new Build($queue);
|
||||
}, ['queue']);
|
||||
|
||||
Server::setResource('queueForDeletes', function (Connection $queue) {
|
||||
return new Delete($queue);
|
||||
}, ['queue']);
|
||||
|
||||
Server::setResource('queueForEvents', function (Connection $queue) {
|
||||
return new Event($queue);
|
||||
}, ['queue']);
|
||||
|
||||
Server::setResource('queueForAudits', function (Connection $queue) {
|
||||
return new Audit($queue);
|
||||
}, ['queue']);
|
||||
|
||||
Server::setResource('queueForFunctions', function (Connection $queue) {
|
||||
return new Func($queue);
|
||||
}, ['queue']);
|
||||
|
||||
Server::setResource('queueForCertificates', function (Connection $queue) {
|
||||
return new Certificate($queue);
|
||||
}, ['queue']);
|
||||
|
||||
Server::setResource('queueForMigrations', function (Connection $queue) {
|
||||
return new Migration($queue);
|
||||
}, ['queue']);
|
||||
|
||||
Server::setResource('logger', function (Registry $register) {
|
||||
return $register->get('logger');
|
||||
}, ['register']);
|
||||
|
||||
Server::setResource('pools', function (Registry $register) {
|
||||
return $register->get('pools');
|
||||
}, ['register']);
|
||||
|
||||
Server::setResource('getFunctionsDevice', function () {
|
||||
return function (string $projectId) {
|
||||
return getDevice(APP_STORAGE_FUNCTIONS . '/app-' . $projectId);
|
||||
};
|
||||
});
|
||||
|
||||
Server::setResource('getFilesDevice', function () {
|
||||
return function (string $projectId) {
|
||||
return getDevice(APP_STORAGE_UPLOADS . '/app-' . $projectId);
|
||||
};
|
||||
});
|
||||
|
||||
Server::setResource('getBuildsDevice', function () {
|
||||
return function (string $projectId) {
|
||||
return getDevice(APP_STORAGE_BUILDS . '/app-' . $projectId);
|
||||
};
|
||||
});
|
||||
|
||||
Server::setResource('getCacheDevice', function () {
|
||||
return function (string $projectId) {
|
||||
return getDevice(APP_STORAGE_CACHE . '/app-' . $projectId);
|
||||
|
|
@ -290,9 +254,9 @@ $worker = $platform->getWorker();
|
|||
|
||||
$worker
|
||||
->shutdown()
|
||||
->inject('connections')
|
||||
->action(function (Connections $connections) {
|
||||
$connections->reclaim();
|
||||
->inject('pools')
|
||||
->action(function (Group $pools) {
|
||||
$pools->reclaim();
|
||||
});
|
||||
|
||||
$worker
|
||||
|
|
@ -301,10 +265,7 @@ $worker
|
|||
->inject('logger')
|
||||
->inject('log')
|
||||
->inject('project')
|
||||
->inject('connections')
|
||||
->action(function (Throwable $error, ?Logger $logger, Log $log, Document $project, Connections $connections) {
|
||||
$connections->reclaim();
|
||||
|
||||
->action(function (Throwable $error, ?Logger $logger, Log $log, Document $project) {
|
||||
$version = App::getEnv('_APP_VERSION', 'UNKNOWN');
|
||||
|
||||
if ($error instanceof PDOException) {
|
||||
|
|
|
|||
Loading…
Reference in a new issue