Skip to content

Commit ae8cd99

Browse files
paxucluspaxuclus
authored
Merge pull request #29 from netlogix/feat/workers-on-react-processes
2 parents 931632b + 27f8714 commit ae8cd99

21 files changed

Lines changed: 935 additions & 544 deletions

‎Classes/AsScheduledJob/CopyToScheduler.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -69,7 +69,7 @@ public function execute(JoinPointInterface $joinPoint): bool
6969
$schedulingInformation = $job->getSchedulingInformation();
7070
switch (true) {
7171
case $schedulingInformation instanceof SchedulingInformation:
72-
$scheduledJob = new ScheduledJob(
72+
$scheduledJob = ScheduledJob::createNew(
7373
$job,
7474
SchedulingInformation::QUEUE_NAME,
7575
$schedulingInformation->dueDate,

‎Classes/Command/SchedulerCommandController.php‎

Lines changed: 127 additions & 84 deletions
Original file line numberDiff line numberDiff line change
@@ -4,42 +4,27 @@
44

55
namespace Netlogix\JobQueue\Scheduled\Command;
66

7-
use Doctrine\DBAL\Exception\ConnectionLost;
8-
use Doctrine\ORM\EntityManagerInterface;
7+
use Doctrine\DBAL\Types\Types;
98
use Flowpack\JobQueue\Common\Job\JobManager;
10-
use Flowpack\JobQueue\Common\Queue\FakeQueue;
11-
use Flowpack\JobQueue\Common\Queue\Message;
12-
use Neos\Flow\Annotations as Flow;
139
use Neos\Flow\Cli\CommandController;
1410
use Neos\Flow\Log\ThrowableStorageInterface;
15-
use Netlogix\JobQueue\Scheduled\AsScheduledJob\SchedulingInformation;
11+
use Netlogix\JobQueue\Pool\Pool;
1612
use Netlogix\JobQueue\Scheduled\Domain\Model\ScheduledJob;
1713
use Netlogix\JobQueue\Scheduled\Domain\SchedulingCoordinator;
1814
use Netlogix\JobQueue\Scheduled\Domain\Scheduler;
15+
use Netlogix\JobQueue\Scheduled\Service\Connection;
1916

20-
/**
21-
* @Flow\Scope("singleton")
22-
*/
2317
class SchedulerCommandController extends CommandController
2418
{
2519
private const TEN_MINUTES_IN_SECONDS = 600;
2620

27-
protected $stopPollingAfter = self::TEN_MINUTES_IN_SECONDS;
21+
protected Scheduler $scheduler;
2822

29-
/**
30-
* @var Scheduler
31-
*/
32-
protected $scheduler;
23+
protected JobManager $jobManager;
3324

34-
/**
35-
* @var JobManager
36-
*/
37-
protected $jobManager;
25+
protected ThrowableStorageInterface $throwableStorage;
3826

39-
/**
40-
* @var ThrowableStorageInterface
41-
*/
42-
protected $throwableStorage;
27+
protected Connection $dbal;
4328

4429
public function injectScheduler(Scheduler $scheduler): void
4530
{
@@ -56,93 +41,151 @@ public function injectThrowableStorageInterface(ThrowableStorageInterface $throw
5641
$this->throwableStorage = $throwableStorage;
5742
}
5843

59-
public function injectEntityManager(EntityManagerInterface $entityManager): void
44+
public function injectConnection(Connection $connection): void
6045
{
61-
/*
62-
* Find a better way to keep connections open. This might only work for MySQL.
63-
*/
64-
$entityManager
65-
->getConnection()
66-
->exec('SET SESSION wait_timeout = 3600;');
46+
$this->dbal = $connection;
6747
}
6848

6949
/**
70-
* Fetch due jobs and schedule them for execution.
50+
* Reset stale jobs that have not changed for too long.
51+
*
52+
* @param string $groupName Free jobs in this group only
53+
* @param int $minutes Count jobs as stale if their last activity was more than these many minutes ago
7154
*/
72-
public function queueDueJobsCommand(string $groupName): void
73-
{
74-
$this->queueDueJobs($groupName);
55+
public function resetStaleJobsCommand(
56+
string $groupName,
57+
int $minutes = 10
58+
): void {
59+
$tableName = ScheduledJob::TABLE_NAME;
60+
$freed = $this->dbal->executeQuery(
61+
sql: <<<MySQL
62+
UPDATE {$tableName}
63+
SET running = 0,
64+
claimed = '',
65+
incarnation = incarnation + 1
66+
WHERE running = 1
67+
AND claimed NOT LIKE 'failed(%)'
68+
AND groupname = :groupName
69+
AND activity < NOW() - INTERVAL :minutes MINUTE
70+
MySQL,
71+
params: [
72+
'groupName' => $groupName,
73+
'minutes' => max($minutes, 1),
74+
],
75+
types: [
76+
'groupName' => Types::STRING,
77+
'minutes' => Types::SMALLINT,
78+
],
79+
)->rowCount();
80+
81+
if ($freed) {
82+
$this->outputLine('Freed ' . $freed . ' stale jobs.');
83+
}
7584
}
7685

7786
/**
7887
* Fetch due jobs and schedule them, then wait and retry.
7988
* This is probably not the best way of polling for changes
89+
*
90+
* @param string $groupName Handle jobs in this group only
91+
* @param bool $outputResults Write child process output to the console
92+
* @param int $parallel Number of jobs to handle in parallel
93+
* @param int $preforkSize Number of jobs to already boot up without having a job waiting
94+
* @param int $stopPollingAfter Stop polling after this many seconds
95+
* @param float $pollingIntervalInSeconds How often to check for new jobs in seconds
8096
*/
81-
public function pollForIncomingJobsCommand(string $groupName): void
82-
{
83-
$startTime = time();
84-
$endTime = $startTime + $this->stopPollingAfter;
85-
86-
while (true) {
87-
$numberOfHandledJobs = $this->queueDueJobs($groupName, $endTime);
88-
if (time() >= $endTime) {
89-
return;
90-
}
91-
if ($numberOfHandledJobs === 0) {
92-
sleep(1);
93-
}
94-
}
97+
public function pollForIncomingJobsCommand(
98+
string $groupName,
99+
bool $outputResults = false,
100+
int $parallel = 1,
101+
int $preforkSize = 0,
102+
int $stopPollingAfter = self::TEN_MINUTES_IN_SECONDS,
103+
float $pollingIntervalInSeconds = 0.1
104+
): void {
105+
$parallel = max($parallel, 1);
106+
Pool::create(
107+
outputResults: $outputResults,
108+
preforkSize: $preforkSize
109+
)
110+
->runLoop(function (Pool $pool) use ($groupName, $parallel, $stopPollingAfter, $pollingIntervalInSeconds, &$queueDueJobs): void {
111+
// Check for new jobs in the database and schedule as much as the pool has capacity for
112+
$queueDueJobs = $pool->eventLoop->addPeriodicTimer(
113+
interval: $pollingIntervalInSeconds,
114+
callback: function () use ($pool, $groupName, $parallel) {
115+
$this->queueDueJobs(pool: $pool, groupName: $groupName, parallel: $parallel);
116+
}
117+
);
118+
119+
// Keep the database connection alive
120+
$ping = $pool->eventLoop->addPeriodicTimer(
121+
interval: 30,
122+
callback: function () use (&$ping) {
123+
$this->scheduler->ping();
124+
}
125+
);
126+
127+
// Once the timeout is reached, wait until the final jobs are done and stop the loop
128+
if ($stopPollingAfter) {
129+
$pool->eventLoop->addTimer(
130+
interval: $stopPollingAfter,
131+
callback: function () use ($pool, $queueDueJobs, $ping) {
132+
$pool->eventLoop->cancelTimer($queueDueJobs);
133+
$checkForPoolToClear = $pool->eventLoop->addPeriodicTimer(
134+
interval: 1,
135+
callback: function () use ($pool, $ping, &$checkForPoolToClear) {
136+
if (count($pool) === 0) {
137+
$pool->eventLoop->cancelTimer($ping);
138+
$pool->eventLoop->cancelTimer($checkForPoolToClear);
139+
}
140+
}
141+
);
142+
}
143+
);
144+
}
145+
});
95146
}
96147

97148
/**
149+
* @param Pool $pool A pool of subprocesses waiting for new jobs to execute
98150
* @param string $groupName Handle only jobs of this group
99-
* @param int $endTime Stop handling new jobs once this time is reached
151+
* @param int $parallel Number of jobs to handle in parallel
100152
* @return int Number of handled jobs
101-
* @throws ConnectionLost
102153
*/
103-
protected function queueDueJobs(string $groupName, int $endTime): int
154+
protected function queueDueJobs(Pool $pool, string $groupName, int $parallel): int
104155
{
105156
$numberOfHandledJobs = 0;
106157
$retry = new SchedulingCoordinator($this->scheduler);
107158

108-
while ($next = $this->scheduler->next($groupName)) {
109-
try {
110-
if ($next->getQueueName() === SchedulingInformation::QUEUE_NAME) {
111-
$this->executeLocally($next);
112-
} else {
113-
$this->executeInQueue($next);
114-
}
115-
$numberOfHandledJobs++;
116-
$this->scheduler->release($next);
117-
} catch(ConnectionLost $e) {
118-
// Assuming we're running as supervisor process: Kill and restart
119-
// The task in question might be stuck because they are claimed and we cannot release them
120-
throw $e;
121-
} catch (\Throwable $throwable) {
122-
$this->throwableStorage->logThrowable($throwable);
123-
$retry->markJobForRescheduling($next);
124-
}
125-
if (time() >= $endTime) {
159+
while (count($pool) < $parallel) {
160+
$next = $this->scheduler->next($groupName);
161+
162+
if (!$next) {
126163
return $numberOfHandledJobs;
127164
}
128-
}
129165

130-
$retry->scheduleAll();
131-
return $numberOfHandledJobs;
132-
}
166+
$numberOfHandledJobs++;
133167

134-
protected function executeLocally(ScheduledJob $scheduledJob): void
135-
{
136-
$job = $scheduledJob->getJob();
137-
$message = new Message('3429a80d-1c21-433d-8d9f-82468b53fb2b', $job, 0);
138-
$job->execute(new FakeQueue(SchedulingInformation::QUEUE_NAME), $message);
139-
}
168+
$process = $pool->runPayload(payload: $next->getSerializedJob(), queueName: $next->getQueueName());
140169

141-
protected function executeInQueue(ScheduledJob $next): void
142-
{
143-
$this->jobManager->queue(
144-
$next->getQueueName(),
145-
$next->getJob()
146-
);
170+
$ping = $pool->eventLoop->addPeriodicTimer(
171+
interval: 1,
172+
callback: function () use ($process, $next) {
173+
if ($process->isRunning()) {
174+
$this->scheduler->activity($next);
175+
}
176+
}
177+
);
178+
179+
$process->on(Pool::EVENT_EXIT, function () use ($pool, $retry, $ping, &$numberOfHandledJobs) {
180+
$pool->eventLoop->cancelTimer($ping);
181+
$numberOfHandledJobs--;
182+
if ($numberOfHandledJobs === 0) {
183+
($numberOfHandledJobs === 0) && $retry->scheduleAll();
184+
}
185+
});
186+
$process->on(Pool::EVENT_SUCCESS, fn () => $this->scheduler->release($next));
187+
$process->on(Pool::EVENT_ERROR, fn () => $retry->markJobForRescheduling($next));
188+
}
189+
return $numberOfHandledJobs;
147190
}
148191
}

0 commit comments

Comments
 (0)