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

Timer::tick

Browse Source
This commit is contained in:
shimon
2022-10-27 12:51:53 +03:00
parent b7d7f43aef
commit d7d98d9866
8 changed files with 156 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/controllers/api/edge.php
+4 -12
View File
@@ -35,21 +35,13 @@ App::post('/v1/edge/sync')
->param('keys', '', new ArrayList(new Text(100), 1000), 'Cache keys. an array containing alphanumerical cache keys')
->inject('request')
->inject('response')
->inject('register')
->action(function (array $keys, Request $request, Response $response, Registry $register) {
->action(function (array $keys, Request $request, Response $response) {
if (empty($keys)) {
//if (empty($keys)) {
throw new Exception(Exception::KEY_NOT_FOUND);
}
//}
$pools = $register->get('pools');
$queue = $pools
->get('queue')
->pop()
->getResource()
;
$client = new SyncIn('syncIn', new QueueRedis(fn() => $queue));
$client = new SyncIn('syncIn', new QueueRedis(App::getEnv('_APP_REDIS_HOST', 'redis'), App::getEnv('_APP_REDIS_PORT', '6379')));
$client->enqueue(['value' => ['keys' => $keys]]);
app/init.php
+18 -17
View File
@@ -533,12 +533,12 @@ $register->set('pools', function () {
'multiple' => true,
'schemes' => ['mariadb', 'mysql'],
],
'queue' => [
'type' => 'queue',
'dsns' => App::getEnv('_APP_CONNECTIONS_QUEUE', $fallbackForRedis),
'multiple' => false,
'schemes' => ['redis'],
],
// 'queue' => [
// 'type' => 'queue',
// 'dsns' => App::getEnv('_APP_CONNECTIONS_QUEUE', $fallbackForRedis),
// 'multiple' => false,
// 'schemes' => ['redis'],
// ],
'pubsub' => [
'type' => 'pubsub',
'dsns' => App::getEnv('_APP_CONNECTIONS_PUBSUB', $fallbackForRedis),
@@ -641,9 +641,11 @@ $register->set('pools', function () {
$adapter->setDefaultDatabase($dsn->getDatabase());
break;
case 'pubsub':
case 'queue':
$adapter = $resource();
break;
$adapter = $resource();
// case 'queue':
// $adapter = $resource();
// break;
case 'cache':
$adapter = match ($dsn->getScheme()) {
'redis' => new RedisCache($resource()),
@@ -1025,12 +1027,11 @@ App::setResource('console', function () {
]);
}, []);
App::setResource('queue', function (Group $pools) {
return $pools
->get('queue')
->pop()
->getResource();
}, ['pools']);
//App::setResource('queue', function (Group $pools) {
// $pools->get('queue')
// ->pop()
// ->getResource();
//}, ['pools']);
App::setResource('dbForProject', function (Group $pools, Database $dbForConsole, Cache $cache, Document $project) {
if ($project->isEmpty() || $project->getId() === 'console') {
@@ -1062,7 +1063,7 @@ App::setResource('dbForConsole', function (Group $pools, Cache $cache) {
return $database;
}, ['pools', 'cache']);
App::setResource('cache', function (Group $pools, $queue) {
App::setResource('cache', function (Group $pools) {
$list = Config::getParam('pools-cache', []);
$adapters = [];
@@ -1075,7 +1076,7 @@ App::setResource('cache', function (Group $pools, $queue) {
}
$cache = new Cache(new Sharding($adapters));
$client = new SyncOut('syncOut', new QueueRedis(fn() => $queue));
$client = new SyncOut('syncOut', new QueueRedis(App::getEnv('_APP_REDIS_HOST', 'redis'), App::getEnv('_APP_REDIS_PORT', '6379')));
$cache->on(cache::EVENT_SAVE, function ($key) use ($client) {
$client
@@ -1088,7 +1089,7 @@ App::setResource('cache', function (Group $pools, $queue) {
});
return $cache;
}, ['pools', 'queue']);
}, ['pools']);
App::setResource('deviceLocal', function () {
return new Local();
app/tasks/sync-edge.php
+8 -16
View File
@@ -7,7 +7,7 @@ use Utopia\App;
use Utopia\CLI\Console;
use Utopia\Database\DateTime;
use Utopia\Database\Query;
use Utopia\Queue\Client as SyncIn;
use Utopia\Queue\Client as SyncOut;
use Utopia\Queue\Connection\Redis as QueueRedis;
$cli
@@ -15,14 +15,14 @@ $cli
->desc('Schedules edge sync tasks')
->action(function () use ($register) {
Console::title('Syncs edges V1');
Console::success(APP_NAME . ' Syncs cloud process v1 has started');
Console::success(APP_NAME . ' Syncs Edge process v1 has started');
$interval = (int) App::getEnv('_APP_SYNC_EDGE_INTERVAL', '180');
Console::loop(function () use ($interval, $register) {
$database = getConsoleDB();
$time = DateTime::now();
$region = App::getEnv('_APP_REGION', 'nyc1');
Console::info("[{$time}] Notifying workers with cloud tasks every {$interval} seconds");
Console::info("[{$time}] Notifying workers with edges tasks every {$interval} seconds");
$time = DateTime::now();
$chunks = $database->find('syncs', [
@@ -31,22 +31,14 @@ $cli
]);
if (count($chunks) > 0) {
Console::info("[{$time}] Found " . \count($chunks) . " cache key chunks to purge.");
$pools = $register->get('pools');
$queue = $pools
->get('queue')
->pop()
->getResource()
;
$client = new SyncIn('syncIn', new QueueRedis(fn() => $queue));
foreach ($chunks as $chunk) {
$client = new SyncOut('syncOut', new QueueRedis(App::getEnv('_APP_REDIS_HOST', 'redis'), App::getEnv('_APP_REDIS_PORT', '6379')));
foreach ($chunks as $counter => $chunk) {
Console::info("[{$time}] Sending chunk .($counter+1). ot of " . count($chunks) . " to {$chunk->getAttribute('target')}");
$client
->enqueue([
'value' => [
'region' => $chunk->getAttribute('regionDest'),
'chunk' => $chunk->getAttribute('keys')
'region' => $chunk->getAttribute('target'),
'keys' => $chunk->getAttribute('keys')
]
]);
app/worker.php
+46
View File
@@ -45,3 +45,49 @@ Server::setResource('cache', function (Registry $register) {
return new Cache(new Sharding($adapters));
}, ['register']);
/**
* Get console database
* @return Database
*/
function getConsoleDB(): Database
{
global $register;
$pools = $register->get('pools'); /** @var \Utopia\Pools\Group $pools */
$dbAdapter = $pools
->get('console')
->pop()
->getResource()
;
$database = new Database($dbAdapter, getCache());
$database->setNamespace('console');
return $database;
}
/**
* Get Cache
* @return Cache
*/
function getCache(): Cache
{
global $register;
$pools = $register->get('pools'); /** @var \Utopia\Pools\Group $pools */
$list = Config::getParam('pools-cache', []);
$adapters = [];
foreach ($list as $value) {
$adapters[] = $pools
->get($value)
->pop()
->getResource()
;
}
return new Cache(new Sharding($adapters));
}
app/workers/sync-In.php
+2 -9
View File
@@ -10,15 +10,8 @@ use Utopia\Queue\Message;
global $register;
$pools = $register->get('pools');
$queue = $pools
->get('queue')
->pop()
->getResource()
;
$connection = new Queue\Connection\Redis(fn() => $queue);
$adapter = new Queue\Adapter\Swoole($connection, 1, 'syncIn');
$connection = new Queue\Connection\Redis(App::getEnv('_APP_REDIS_HOST', 'redis'), App::getEnv('_APP_REDIS_PORT', '6379'));
$adapter = new Queue\Adapter\Swoole($connection, 2, 'syncIn');
$server = new Queue\Server($adapter);
$server->job()
app/workers/sync-out.php
+59 -55
View File
@@ -7,30 +7,39 @@ use Appwrite\Utopia\Response;
use Swoole\Runtime;
use Swoole\Timer;
use Utopia\App;
use Utopia\CLI\Console;
use Utopia\Config\Config;
use Utopia\Database\Database;
use Utopia\Database\Document;
use Utopia\Database\Exception\Authorization;
use Utopia\Database\Exception\Structure;
use Utopia\Queue;
use Utopia\Queue\Message;
global $register;
$keys = [];
$counter = 0;
$regions = array_filter(
Config::getParam('regions', []),
fn ($region) => App::getEnv('_APP_REGION', 'nyc1') !== $region
&& $region !== 'default',
ARRAY_FILTER_USE_KEY
);
const SUBMITION_INTERVAL = 20;
const MAX_KEY_COUNT = 10;
$stack = [
'regions' => $regions,
'keys' => [],
];
$failures = [];
const MAX_KEY_COUNT = 2;
const MAX_CURL_SEND_ATTEMPTS = 4;
/**
* @param string $url
* @param string $token
* @param array $keys
* @param array $stack
* @return array
*/
function send(string $url, string $token, array $keys): array
function send(string $url, string $token, array $stack): array
{
$payload = [];
$ch = curl_init($url);
@@ -41,7 +50,7 @@ function send(string $url, string $token, array $keys): array
curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
curl_setopt($ch, CURLOPT_TIMEOUT, 5);
curl_setopt($ch, CURLOPT_CUSTOMREQUEST, 'POST');
curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($keys));
curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($stack));
for ($attempts = 0; $attempts < MAX_CURL_SEND_ATTEMPTS; $attempts++) {
$response = curl_exec($ch);
@@ -68,24 +77,20 @@ function send(string $url, string $token, array $keys): array
* @throws Structure
* @throws Exception
*/
function call($database, $regions, $keys): void
function call($database, $regions, $stack): void
{
$jwt = new JWT(App::getEnv('_APP_OPENSSL_KEY_V1'), 'HS256', 600, 10);
$token = $jwt->encode([]);
foreach ($regions as $code => $region) {
var_dump('Sending request to ' . $code . '...............');
$response = send($region['domain'] . '/v1/edge/sync', $token, ['keys' => $keys]);
var_dump([
'keys' => $keys,
'response' => $response
]);
Console::info("Sending request to {$code}");
$response = send($region['domain'] . '/v1/edge/sync', $token, ['keys' => $stack]);
if ($response['status'] !== Response::STATUS_CODE_OK) {
$database->createDocument('syncs', new Document([
'region' => App::getEnv('_APP_REGION', 'nyc1'),
'target' => $code,
'keys' => $keys,
'keys' => $stack,
'status' => $response['status'],
'payload' => $response['payload'],
]));
@@ -93,53 +98,35 @@ function call($database, $regions, $keys): void
}
}
$pools = $register->get('pools');
$queue = $pools
->get('queue')
->pop()
->getResource()
;
$connection = new Queue\Connection\Redis(fn() => $queue);
$adapter = new Queue\Adapter\Swoole($connection, 2, 'syncOut');
$connection = new Queue\Connection\Redis(App::getEnv('_APP_REDIS_HOST', 'redis'), App::getEnv('_APP_REDIS_PORT', '6379'));
$adapter = new Queue\Adapter\Swoole($connection, 1, 'syncOut');
$server = new Queue\Server($adapter);
$server->job()
->inject('message')
->inject('dbForConsole')
->action(function (Message $message, Database $dbForConsole) use (&$keys, &$counter) {
->action(function (Message $message) use (&$stack, &$failures) {
$payload = $message->getPayload()['value'];
$regions = Config::getParam('regions', true);
$regions = array_filter(
$regions,
fn ($region) => App::getEnv('_APP_REGION', 'nyc1') !== $region
&& $region !== 'default',
ARRAY_FILTER_USE_KEY
);
$payload = $message->getPayload()['value'] ?? [];
if (!empty($payload['region'])) {
if (!empty($payload['keys'])) {
$regions = array_filter(
$regions,
fn ($region) => $payload['region'] === $region,
Config::getParam('regions', []),
fn ($region) => $payload['region'] === $region,
ARRAY_FILTER_USE_KEY
);
$failures[] = [
'regions' => $regions,
'keys' => $payload['keys']
];
}
if (!empty($payload['chunk'])) {
call($dbForConsole, $regions, $payload['chunk']);
return;
if (!empty($payload['key'])) {
if (!in_array($payload['key'], $stack['keys'] ?? [])) {
$stack['keys'][] = $payload['key'];
}
}
$keys[$payload['key']] = null;
// if (count($keys) >= MAX_KEY_COUNT || ($counter + SUBMITION_INTERVAL) < time()) {
// call($dbForConsole, $regions, array_keys($keys));
// $counter = time();
// $keys = [];
// }
});
Runtime::enableCoroutine(SWOOLE_HOOK_ALL);
@@ -153,11 +140,28 @@ $server
});
$server
->workerStart(function () {
Timer::tick(1000, function () {
var_dump(date('m/d/Y H:i:s', time()));
->workerStart(function () use (&$stack, &$failures) {
Timer::tick(10000, function () use (&$stack, &$failures) {
if (empty($stack['keys']) && count($failures) === 0) {
Console::info("Stack is empty");
return;
}
if (count($failures) > 0) {
$i = 0;
while ($i < count($failures)) {
$failure = array_shift($failures);
call(getConsoleDB(), $failure['regions'], $failure['keys']);
$i++;
}
return;
}
$chunk = array_slice($stack['keys'], 0, MAX_KEY_COUNT);
array_splice($stack['keys'], 0, MAX_KEY_COUNT);
call(getConsoleDB(), $stack['regions'], $chunk);
$chunk = [];
});
echo "Out region [" . App::getEnv('_APP_REGION', 'nyc1') . "] cache purging worker Started" . PHP_EOL;
})
->start();
composer.json
+1 -1
View File
@@ -61,7 +61,7 @@
"utopia-php/websocket": "0.1.0",
"utopia-php/image": "0.5.*",
"utopia-php/orchestration": "0.6.*",
"utopia-php/queue": "dev-refactor-redis-client as 0.3.0",
"utopia-php/queue": "0.3.0",
"utopia-php/pools": "0.1.*",
"resque/php-resque": "1.3.6",
"matomo/device-detector": "6.0.0",
composer.lock
Generated
+18 -25
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": "0b5f0fd08db65ca4c2e2f1a6d33cd299",
"content-hash": "d2d51b9a3dcbb3542243d1a68a078ece",
"packages": [
{
"name": "adhocore/jwt",
@@ -693,16 +693,16 @@
},
{
"name": "guzzlehttp/psr7",
"version": "2.4.1",
"version": "2.4.2",
"source": {
"type": "git",
"url": "https://github.com/guzzle/psr7.git",
"reference": "69568e4293f4fa993f3b0e51c9723e1e17c41379"
"reference": "3148458748274be1546f8f2809a6c09fe66f44aa"
},
"dist": {
"type": "zip",
"url": "https://api.github.com/repos/guzzle/psr7/zipball/69568e4293f4fa993f3b0e51c9723e1e17c41379",
"reference": "69568e4293f4fa993f3b0e51c9723e1e17c41379",
"url": "https://api.github.com/repos/guzzle/psr7/zipball/3148458748274be1546f8f2809a6c09fe66f44aa",
"reference": "3148458748274be1546f8f2809a6c09fe66f44aa",
"shasum": ""
},
"require": {
@@ -792,7 +792,7 @@
],
"support": {
"issues": "https://github.com/guzzle/psr7/issues",
"source": "https://github.com/guzzle/psr7/tree/2.4.1"
"source": "https://github.com/guzzle/psr7/tree/2.4.2"
},
"funding": [
{
@@ -808,7 +808,7 @@
"type": "tidelift"
}
],
"time": "2022-08-28T14:45:39+00:00"
"time": "2022-10-25T13:49:28+00:00"
},
{
"name": "influxdb/influxdb-php",
@@ -1901,12 +1901,12 @@
"source": {
"type": "git",
"url": "https://github.com/utopia-php/cache.git",
"reference": "99e7085eb229d0f0159a4f2107ea5ea123f7b32b"
"reference": "d0cd43b679fa23dc720c3bb9dff4c47c0eb4790d"
},
"dist": {
"type": "zip",
"url": "https://api.github.com/repos/utopia-php/cache/zipball/99e7085eb229d0f0159a4f2107ea5ea123f7b32b",
"reference": "99e7085eb229d0f0159a4f2107ea5ea123f7b32b",
"url": "https://api.github.com/repos/utopia-php/cache/zipball/d0cd43b679fa23dc720c3bb9dff4c47c0eb4790d",
"reference": "d0cd43b679fa23dc720c3bb9dff4c47c0eb4790d",
"shasum": ""
},
"require": {
@@ -1942,7 +1942,7 @@
"issues": "https://github.com/utopia-php/cache/issues",
"source": "https://github.com/utopia-php/cache/tree/feat-redis-sync"
},
"time": "2022-10-18T06:58:42+00:00"
"time": "2022-10-26T08:25:43+00:00"
},
{
"name": "utopia-php/cli",
@@ -2537,16 +2537,16 @@
},
{
"name": "utopia-php/queue",
"version": "dev-refactor-redis-client",
"version": "0.3.0",
"source": {
"type": "git",
"url": "https://github.com/utopia-php/queue.git",
"reference": "3e49bcbfc343792ea4899d60b24e7b6b334e5910"
"reference": "42b132c6f2431b726c2bc629c386921e4934b863"
},
"dist": {
"type": "zip",
"url": "https://api.github.com/repos/utopia-php/queue/zipball/3e49bcbfc343792ea4899d60b24e7b6b334e5910",
"reference": "3e49bcbfc343792ea4899d60b24e7b6b334e5910",
"url": "https://api.github.com/repos/utopia-php/queue/zipball/42b132c6f2431b726c2bc629c386921e4934b863",
"reference": "42b132c6f2431b726c2bc629c386921e4934b863",
"shasum": ""
},
"require": {
@@ -2592,9 +2592,9 @@
],
"support": {
"issues": "https://github.com/utopia-php/queue/issues",
"source": "https://github.com/utopia-php/queue/tree/refactor-redis-client"
"source": "https://github.com/utopia-php/queue/tree/0.3.0"
},
"time": "2022-10-24T14:29:47+00:00"
"time": "2022-10-19T13:22:07+00:00"
},
{
"name": "utopia-php/registry",
@@ -5469,19 +5469,12 @@
"version": "dev-feat-update-cache-lib",
"alias": "0.26.1",
"alias_normalized": "0.26.1.0"
},
{
"package": "utopia-php/queue",
"version": "dev-refactor-redis-client",
"alias": "0.3.0",
"alias_normalized": "0.3.0.0"
}
],
"minimum-stability": "stable",
"stability-flags": {
"utopia-php/cache": 20,
"utopia-php/database": 20,
"utopia-php/queue": 20
"utopia-php/database": 20
},
"prefer-stable": false,
"prefer-lowest": false,