Skip to content

Commit 1db31ba

Browse files
loks0nloks0n
committed
chore: add workerstop
1 parent eed8b7c commit 1db31ba

20 files changed

Lines changed: 359 additions & 247 deletions

File tree

‎pint.json‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,6 @@
11
{
2-
"preset": "psr12"
2+
"preset": "psr12",
3+
"rules": {
4+
"single_quote": true
5+
}
36
}

‎src/Queue/Adapter.php‎

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -40,10 +40,4 @@ abstract public function workerStart(callable $callback): self;
4040
* @return self
4141
*/
4242
abstract public function workerStop(callable $callback): self;
43-
44-
/**
45-
* Returns the native server object from the Adapter.
46-
* @return mixed
47-
*/
48-
abstract public function getNative(): mixed;
4943
}

‎src/Queue/Adapter/Swoole.php‎

Lines changed: 67 additions & 64 deletions
Original file line numberDiff line numberDiff line change
@@ -2,101 +2,104 @@
22

33
namespace Utopia\Queue\Adapter;
44

5-
use Swoole\Constant;
6-
use Swoole\Process\Pool;
7-
use Utopia\CLI\Console;
5+
use Swoole\Coroutine;
6+
use Swoole\Process;
87
use Utopia\Queue\Adapter;
98
use Utopia\Queue\Consumer;
109

1110
class Swoole extends Adapter
1211
{
13-
protected Pool $pool;
14-
15-
/** @var callable */
16-
private $onStop;
17-
18-
public function __construct(Consumer $consumer, int $workerNum, string $queue, string $namespace = 'utopia-queue')
19-
{
12+
/** @var Process[] */
13+
protected array $workers = [];
14+
protected bool $running = false;
15+
16+
/** @var callable[] */
17+
protected array $onWorkerStart = [];
18+
19+
/** @var callable[] */
20+
protected array $onWorkerStop = [];
21+
22+
public function __construct(
23+
Consumer $consumer,
24+
int $workerNum,
25+
string $queue,
26+
string $namespace = 'utopia-queue',
27+
) {
2028
parent::__construct($workerNum, $queue, $namespace);
21-
2229
$this->consumer = $consumer;
23-
$this->pool = new Pool($workerNum);
2430
}
2531

2632
public function start(): self
2733
{
28-
$this->pool->set(['enable_coroutine' => true]);
34+
$this->running = true;
2935

30-
// Register signal handlers in the main process before starting pool
31-
if (extension_loaded('pcntl')) {
32-
pcntl_signal(SIGTERM, function () {
33-
Console::info("[Swoole] Received SIGTERM, initiating graceful shutdown...");
34-
$this->stop();
35-
});
36+
for ($i = 0; $i < $this->workerNum; $i++) {
37+
$this->spawnWorker($i);
38+
}
3639

37-
pcntl_signal(SIGINT, function () {
38-
Console::info("[Swoole] Received SIGINT, initiating graceful shutdown...");
39-
$this->stop();
40-
});
40+
Coroutine::set(['hook_flags' => SWOOLE_HOOK_ALL]);
4141

42-
// Enable async signals
43-
pcntl_async_signals(true);
44-
} else {
45-
Console::warning("[Swoole] pcntl extension is not loaded, worker will not shutdown gracefully.");
46-
}
42+
Coroutine\run(function () {
43+
Process::signal(SIGTERM, fn () => $this->stop());
44+
Process::signal(SIGINT, fn () => $this->stop());
45+
Process::signal(SIGCHLD, fn () => $this->reap());
46+
47+
while ($this->running && \count($this->workers) > 0) {
48+
Coroutine::sleep(1);
49+
}
50+
});
4751

48-
$this->pool->start();
4952
return $this;
5053
}
5154

52-
public function stop(): self
55+
protected function spawnWorker(int $workerId): void
5356
{
54-
if ($this->onStop) {
55-
call_user_func($this->onStop);
56-
}
57+
$process = new Process(function () use ($workerId) {
58+
Coroutine::set(['hook_flags' => SWOOLE_HOOK_ALL]);
5759

58-
Console::info("[Swoole] Shutting down process pool...");
59-
$this->pool->shutdown();
60-
Console::success("[Swoole] Process pool stopped.");
61-
return $this;
62-
}
60+
Coroutine\run(function () use ($workerId) {
61+
Process::signal(SIGTERM, fn () => $this->consumer->close());
6362

64-
public function workerStart(callable $callback): self
65-
{
66-
$this->pool->on(Constant::EVENT_WORKER_START, function (Pool $pool, string $workerId) use ($callback) {
67-
// Register signal handlers in each worker process for graceful shutdown
68-
if (extension_loaded('pcntl')) {
69-
pcntl_signal(SIGTERM, function () use ($workerId) {
70-
Console::info("[Worker] Worker {$workerId} received SIGTERM, closing consumer...");
71-
$this->consumer->close();
72-
});
73-
74-
pcntl_signal(SIGINT, function () use ($workerId) {
75-
Console::info("[Worker] Worker {$workerId} received SIGINT, closing consumer...");
76-
$this->consumer->close();
77-
});
78-
79-
pcntl_async_signals(true);
63+
foreach ($this->onWorkerStart as $callback) {
64+
$callback((string)$workerId);
65+
}
66+
});
67+
68+
// onWorkerStop runs after Coroutine::run() exits
69+
foreach ($this->onWorkerStop as $callback) {
70+
$callback((string)$workerId);
8071
}
72+
}, false, 0, false);
8173

82-
call_user_func($callback, $workerId);
83-
});
74+
$pid = $process->start();
75+
$this->workers[$pid] = $process;
76+
}
8477

85-
return $this;
78+
protected function reap(): void
79+
{
80+
while (($ret = Process::wait(false)) !== false) {
81+
unset($this->workers[$ret['pid']]);
82+
}
8683
}
8784

88-
public function workerStop(callable $callback): self
85+
public function stop(): self
8986
{
90-
$this->onStop = $callback;
91-
$this->pool->on(Constant::EVENT_WORKER_STOP, function (Pool $pool, string $workerId) use ($callback) {
92-
call_user_func($callback, $workerId);
93-
});
87+
$this->running = false;
88+
foreach ($this->workers as $pid => $process) {
89+
Process::kill($pid, SIGTERM);
90+
}
91+
return $this;
92+
}
9493

94+
public function workerStart(callable $callback): self
95+
{
96+
$this->onWorkerStart[] = $callback;
9597
return $this;
9698
}
9799

98-
public function getNative(): Pool
100+
public function workerStop(callable $callback): self
99101
{
100-
return $this->pool;
102+
$this->onWorkerStop[] = $callback;
103+
return $this;
101104
}
102105
}

‎src/Queue/Adapter/Workerman.php‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -47,9 +47,4 @@ public function workerStop(callable $callback): self
4747

4848
return $this;
4949
}
50-
51-
public function getNative(): Worker
52-
{
53-
return $this->worker;
54-
}
5550
}

‎src/Queue/Broker/AMQP.php‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -111,7 +111,7 @@ public function consume(Queue $queue, callable $messageCallback, callable $succe
111111
$channel->exchange_declare("{$queue->namespace}.failed", AMQPExchangeType::TOPIC, durable: true, auto_delete: false, arguments: new AMQPTable($this->exchangeArguments));
112112

113113
// 2. Declare the working queue and configure the DLX for receiving rejected messages.
114-
$channel->queue_declare($queue->name, durable: true, auto_delete: false, arguments: new AMQPTable(array_merge($this->queueArguments, ["x-dead-letter-exchange" => "{$queue->namespace}.failed"])));
114+
$channel->queue_declare($queue->name, durable: true, auto_delete: false, arguments: new AMQPTable(array_merge($this->queueArguments, ['x-dead-letter-exchange' => "{$queue->namespace}.failed"])));
115115
$channel->queue_bind($queue->name, $queue->namespace, routing_key: $queue->name);
116116

117117
// 3. Declare the dead-letter-queue and bind it to the DLX.
@@ -162,7 +162,7 @@ public function getQueueSize(Queue $queue, bool $failedJobs = false): int
162162
{
163163
$queueName = $queue->name;
164164
if ($failedJobs) {
165-
$queueName = $queueName . ".failed";
165+
$queueName = $queueName . '.failed';
166166
}
167167

168168
$client = new Client();

‎src/Queue/Broker/Pool.php‎

Lines changed: 15 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -30,26 +30,36 @@ public function getQueueSize(Queue $queue, bool $failedJobs = false): int
3030
return $this->delegatePublish(__FUNCTION__, \func_get_args());
3131
}
3232

33-
public function consume(Queue $queue, callable $messageCallback, callable $successCallback, callable $errorCallback): void
34-
{
33+
public function consume(
34+
Queue $queue,
35+
callable $messageCallback,
36+
callable $successCallback,
37+
callable $errorCallback,
38+
): void {
3539
$this->delegateConsumer(__FUNCTION__, \func_get_args());
3640
}
3741

3842
public function close(): void
3943
{
40-
$this->delegateConsumer(__FUNCTION__, \func_get_args());
44+
// TODO: Implement closing all connections in the pool
4145
}
4246

4347
protected function delegatePublish(string $method, array $args): mixed
4448
{
45-
return $this->publisher?->use(function (Publisher $adapter) use ($method, $args) {
49+
return $this->publisher?->use(function (Publisher $adapter) use (
50+
$method,
51+
$args,
52+
) {
4653
return $adapter->$method(...$args);
4754
});
4855
}
4956

5057
protected function delegateConsumer(string $method, array $args): mixed
5158
{
52-
return $this->consumer?->use(function (Consumer $adapter) use ($method, $args) {
59+
return $this->consumer?->use(function (Consumer $adapter) use (
60+
$method,
61+
$args,
62+
) {
5363
return $adapter->$method(...$args);
5464
});
5565
}

‎src/Queue/Broker/Redis.php‎

Lines changed: 14 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,8 @@
1010

1111
class Redis implements Publisher, Consumer
1212
{
13+
private const int POP_TIMEOUT = 2;
14+
1315
private bool $closed = false;
1416

1517
public function __construct(private readonly Connection $connection)
@@ -22,7 +24,16 @@ public function consume(Queue $queue, callable $messageCallback, callable $succe
2224
/**
2325
* Waiting for next Job.
2426
*/
25-
$nextMessage = $this->connection->rightPopArray("{$queue->namespace}.queue.{$queue->name}", 5);
27+
try {
28+
$nextMessage = $this->connection->rightPopArray("{$queue->namespace}.queue.{$queue->name}", self::POP_TIMEOUT);
29+
30+
} catch (\RedisException $e) {
31+
if ($this->closed) {
32+
break;
33+
}
34+
35+
throw $e;
36+
}
2637

2738
if (!$nextMessage) {
2839
continue;
@@ -92,6 +103,7 @@ public function consume(Queue $queue, callable $messageCallback, callable $succe
92103
public function close(): void
93104
{
94105
$this->closed = true;
106+
$this->connection->close();
95107
}
96108

97109
public function enqueue(Queue $queue, array $payload): bool
@@ -115,7 +127,7 @@ public function retry(Queue $queue, ?int $limit = null): void
115127
$processed = 0;
116128

117129
while (true) {
118-
$pid = $this->connection->rightPop("{$queue->namespace}.failed.{$queue->name}", 5);
130+
$pid = $this->connection->rightPop("{$queue->namespace}.failed.{$queue->name}", self::POP_TIMEOUT);
119131

120132
// No more jobs to retry
121133
if ($pid === false) {

‎src/Queue/Connection.php‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,4 +25,5 @@ public function setArray(string $key, array $value): bool;
2525
public function increment(string $key): int;
2626
public function decrement(string $key): int;
2727
public function ping(): bool;
28+
public function close(): void;
2829
}

‎src/Queue/Connection/Redis.php‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -169,6 +169,12 @@ public function ping(): bool
169169
}
170170
}
171171

172+
public function close(): void
173+
{
174+
$this->redis?->close();
175+
$this->redis = null;
176+
}
177+
172178
protected function getRedis(): \Redis
173179
{
174180
if ($this->redis) {

‎src/Queue/Connection/RedisCluster.php‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -166,6 +166,12 @@ public function ping(): bool
166166
}
167167
}
168168

169+
public function close(): void
170+
{
171+
$this->redis?->close();
172+
$this->redis = null;
173+
}
174+
169175
protected function getRedis(): \RedisCluster
170176
{
171177
if ($this->redis) {

0 commit comments

Comments
 (0)