From 90e31c04bc76afd8777b5e01a9bafc8cc0404977 Mon Sep 17 00:00:00 2001 From: Yurii Kuznietsov Date: Sun, 15 Jun 2025 12:39:43 +0300 Subject: [PATCH] Job sub-queue (#3403) * cleanup * job m0 queue --- application/Espo/Core/Job/JobManager.php | 7 + application/Espo/Core/Job/QueueName.php | 14 +- application/Espo/Core/Job/QueueProcessor.php | 9 +- .../Espo/Core/Job/QueueProcessor/Params.php | 35 +++ .../Espo/Core/Job/QueueProcessor/Picker.php | 134 +++++++++ application/Espo/Core/Job/QueueUtil.php | 7 +- .../Core/Job/QueueProcessor/PickerTest.php | 255 ++++++++++++++++++ 7 files changed, 450 insertions(+), 11 deletions(-) create mode 100644 application/Espo/Core/Job/QueueProcessor/Picker.php create mode 100644 tests/unit/Espo/Core/Job/QueueProcessor/PickerTest.php diff --git a/application/Espo/Core/Job/JobManager.php b/application/Espo/Core/Job/JobManager.php index 2bfdabf1e7..b4409616af 100644 --- a/application/Espo/Core/Job/JobManager.php +++ b/application/Espo/Core/Job/JobManager.php @@ -123,6 +123,13 @@ class JobManager ->withUseProcessPool($this->useProcessPool) ->withLimit($limit); + $subQueueParams = [ + $params->withWeight(0.5), + $params->withQueue(QueueName::M0)->withWeight(0.5), + ]; + + $params = $params->withSubQueueParamsList($subQueueParams); + $this->queueProcessor->process($params); } diff --git a/application/Espo/Core/Job/QueueName.php b/application/Espo/Core/Job/QueueName.php index ef0c2ad52c..bf6871d7f4 100644 --- a/application/Espo/Core/Job/QueueName.php +++ b/application/Espo/Core/Job/QueueName.php @@ -32,17 +32,25 @@ namespace Espo\Core\Job; class QueueName { /** - * Executes as soon as possible. + * Executes as soon as possible. Non-parallel. */ public const Q0 = 'q0'; /** - * Executes every minute. + * Executes every minute. Non-parallel. */ public const Q1 = 'q1'; /** - * Executes as soon as possible. For email processing. + * Executes as soon as possible. For email processing. Non-parallel. */ public const E0 = 'e0'; + + /** + * Executes in the main queue pool in parallel. Along with jobs without specified queue. + * A portion is always picked for a queue iteration, even if there are no-queue + * jobs ordered before. E.g. if the portion size is 100, and there are 200 empty-queue + * jobs and 5 m0 jobs, 95 and 5 will be picked respectfully. + */ + const M0 = 'm0'; } diff --git a/application/Espo/Core/Job/QueueProcessor.php b/application/Espo/Core/Job/QueueProcessor.php index a29adf8ea6..e90e5b1a0f 100644 --- a/application/Espo/Core/Job/QueueProcessor.php +++ b/application/Espo/Core/Job/QueueProcessor.php @@ -29,6 +29,7 @@ namespace Espo\Core\Job; +use Espo\Core\Job\QueueProcessor\Picker; use Espo\Entities\Job as JobEntity; use Espo\Core\Job\QueueProcessor\Params; use Espo\Core\ORM\EntityManager; @@ -46,6 +47,7 @@ class QueueProcessor private JobRunner $jobRunner, private AsyncPoolFactory $asyncPoolFactory, private EntityManager $entityManager, + private Picker $picker, ConfigDataProvider $configDataProvider ) { $this->noTableLocking = $configDataProvider->noTableLocking(); @@ -54,12 +56,9 @@ class QueueProcessor public function process(Params $params): void { $pool = $params->useProcessPool() ? - $this->asyncPoolFactory->create() : - null; + $this->asyncPoolFactory->create() : null; - $pendingJobList = $this->queueUtil->getPendingJobList($params); - - foreach ($pendingJobList as $job) { + foreach ($this->picker->pick($params) as $job) { $this->processJob($params, $job, $pool); } diff --git a/application/Espo/Core/Job/QueueProcessor/Params.php b/application/Espo/Core/Job/QueueProcessor/Params.php index f9522a7cb7..fdc4568b25 100644 --- a/application/Espo/Core/Job/QueueProcessor/Params.php +++ b/application/Espo/Core/Job/QueueProcessor/Params.php @@ -36,6 +36,9 @@ class Params private ?string $queue = null; private ?string $group = null; private int $limit = 0; + /** @var ?Params[] */ + private ?array $subQueueParamsList = null; + private float $weight = 1.0; public function withUseProcessPool(bool $useProcessPool): self { @@ -77,6 +80,25 @@ class Params return $obj; } + public function withWeight(float $weight): self + { + $obj = clone $this; + $obj->weight = $weight; + + return $obj; + } + + /** + * @param ?Params[] $subQueueParamsList + */ + public function withSubQueueParamsList(?array $subQueueParamsList): self + { + $obj = clone $this; + $obj->subQueueParamsList = $subQueueParamsList; + + return $obj; + } + public function useProcessPool(): bool { return $this->useProcessPool; @@ -102,6 +124,19 @@ class Params return $this->limit; } + public function getWeight(): float + { + return $this->weight; + } + + /** + * @return ?Params[] + */ + public function getSubQueueParamsList(): ?array + { + return $this->subQueueParamsList; + } + public static function create(): self { return new self(); diff --git a/application/Espo/Core/Job/QueueProcessor/Picker.php b/application/Espo/Core/Job/QueueProcessor/Picker.php new file mode 100644 index 0000000000..97ed945b30 --- /dev/null +++ b/application/Espo/Core/Job/QueueProcessor/Picker.php @@ -0,0 +1,134 @@ +. + * + * The interactive user interfaces in modified source and object code versions + * of this program must display Appropriate Legal Notices, as required under + * Section 5 of the GNU Affero General Public License version 3. + * + * In accordance with Section 7(b) of the GNU Affero General Public License version 3, + * these Appropriate Legal Notices must retain the display of the "EspoCRM" word. + ************************************************************************/ + +namespace Espo\Core\Job\QueueProcessor; + +use Espo\Core\Job\QueueUtil; +use Espo\Entities\Job; +use RuntimeException; + +/** + * Picks jobs for a portion distributing by weights if needed. + */ +class Picker +{ + public function __construct( + private QueueUtil $queueUtil, + ) {} + + /** + * @param Params $params + * @return iterable + */ + public function pick(Params $params): iterable + { + $paramsList = $params->getSubQueueParamsList(); + + if (!$paramsList) { + return $this->queueUtil->getPendingJobs($params); + } + + $groups = []; + + foreach ($paramsList as $itemParams) { + $groups[] = iterator_to_array($this->queueUtil->getPendingJobs($itemParams)); + } + + return $this->pickJobsRecursively($paramsList, $groups, $params->getLimit()); + } + + /** + * @param Params[] $paramsList, + * @param Job[][] $groups + * @return Job[] + */ + private function pickJobsRecursively( + array $paramsList, + array $groups, + int $limit, + ): array { + + $totalWeight = array_reduce($paramsList, fn ($c, $it) => $c + $it->getWeight(), 0.0); + + /** @var Job[][] $leftovers */ + $leftovers = []; + $output = []; + + foreach ($paramsList as $i => $itemParams) { + if (!array_key_exists($i, $groups)) { + throw new RuntimeException(); + } + + $jobs = $groups[$i]; + $weight = $itemParams->getWeight(); + + $portion = (int) round($weight / $totalWeight * $limit); + + $pickedJobs = []; + + while (count($pickedJobs) < $portion) { + if (count($jobs) === 0) { + break; + } + + $pickedJobs[] = array_shift($jobs); + } + + $output = array_merge($output, $pickedJobs); + + $leftovers[] = $jobs; + } + + $left = $limit - count($output); + $leftoverCount = array_reduce($leftovers, fn ($c, $it) => $c + count($it), 0); + + if ($left && $leftoverCount) { + foreach ($leftovers as $i => $jobs) { + if (count($jobs) === 0) { + unset($leftovers[$i]); + unset($paramsList[$i]); + } + } + + $leftovers = array_values($leftovers); + $paramsList = array_values($paramsList); + + $rest = $this->pickJobsRecursively( + paramsList: $paramsList, + groups: $leftovers, + limit: $leftoverCount, + ); + + $output = array_merge($output, $rest); + $output = array_slice($output, 0, $limit); + } + + return $output; + } +} diff --git a/application/Espo/Core/Job/QueueUtil.php b/application/Espo/Core/Job/QueueUtil.php index 7ba184592c..49f8a2a4d6 100644 --- a/application/Espo/Core/Job/QueueUtil.php +++ b/application/Espo/Core/Job/QueueUtil.php @@ -29,6 +29,7 @@ namespace Espo\Core\Job; +use Countable; use Espo\Core\Job\QueueProcessor\Params; use Espo\Core\ORM\EntityManager; use Espo\Core\Utils\Config; @@ -36,9 +37,9 @@ use Espo\Core\Utils\DateTime as DateTimeUtil; use Espo\Core\Utils\System; use Espo\Core\Job\Job\Status; use Espo\Entities\Job as JobEntity; -use Espo\ORM\Collection; use DateTime; +use Espo\ORM\Collection; use Espo\ORM\Name\Attribute; use Exception; use LogicException; @@ -73,9 +74,9 @@ class QueueUtil } /** - * @return Collection + * @return Collection&Countable */ - public function getPendingJobList(Params $params): Collection + public function getPendingJobs(Params $params): Collection { $queue = $params->getQueue(); $group = $params->getGroup(); diff --git a/tests/unit/Espo/Core/Job/QueueProcessor/PickerTest.php b/tests/unit/Espo/Core/Job/QueueProcessor/PickerTest.php new file mode 100644 index 0000000000..a4273c7918 --- /dev/null +++ b/tests/unit/Espo/Core/Job/QueueProcessor/PickerTest.php @@ -0,0 +1,255 @@ +. + * + * The interactive user interfaces in modified source and object code versions + * of this program must display Appropriate Legal Notices, as required under + * Section 5 of the GNU Affero General Public License version 3. + * + * In accordance with Section 7(b) of the GNU Affero General Public License version 3, + * these Appropriate Legal Notices must retain the display of the "EspoCRM" word. + ************************************************************************/ + +namespace tests\unit\Espo\Core\Job\QueueProcessor; + +use Countable; +use Espo\Core\Job\QueueProcessor\Params; +use Espo\Core\Job\QueueProcessor\Picker; +use Espo\Core\Job\QueueUtil; +use Espo\Entities\Job; +use Espo\ORM\Collection; +use Espo\ORM\EntityCollection; + +use PHPUnit\Framework\TestCase; + +class PickerTest extends TestCase +{ + /** + * @param string[] $ids + * @return Collection&Countable + */ + private function createCollection(array $ids): Collection&Countable + { + $output = new EntityCollection(); + + foreach ($ids as $id) { + $job = $this->createMock(Job::class); + + $job->expects($this->any()) + ->method('getId') + ->willReturn($id); + + $output[] = $job; + } + + return $output; + } + + /** + * @param string[] $ids + * @param iterable $pick + */ + private function assertPick(array $ids, iterable $pick): void + { + $pick = iterator_to_array($pick); + + $this->assertSameSize($ids, $pick); + + foreach ($pick as $i => $job) { + $this->assertEquals($ids[$i], $job->getId()); + } + } + + /** + * @param float[] $weights + * @param string[][] $idsGroups + * @param string[] $idsExpected + */ + private function prepareAndTest(int $limit, array $weights, array $idsGroups, array $idsExpected): void + { + $util = $this->createMock(QueueUtil::class); + + $params = Params::create()->withLimit($limit); + + $paramList = []; + + foreach ($weights as $i => $weight) { + $paramList[] = $params + ->withWeight($weight) + ->withQueue('m' . $i); + } + + + $params = $params->withSubQueueParamsList($paramList); + + $picker = new Picker($util); + + $collections = []; + + foreach ($idsGroups as $ids) { + $collections[] = $this->createCollection($ids); + } + + $util->expects($this->any()) + ->method('getPendingJobs') + ->willReturnOnConsecutiveCalls(...$collections); + + $pick = $picker->pick($params); + + $this->assertPick($idsExpected, $pick); + } + + public function testPick0(): void + { + $this->prepareAndTest( + limit: 10, + weights: [1.0], + idsGroups: [ + ['a0', 'a1', 'a2', 'a3', 'a4', 'a5'], + ], + idsExpected: + ['a0', 'a1', 'a2', 'a3', 'a4', 'a5'], + ); + } + + public function testPick1(): void + { + $this->prepareAndTest( + limit: 10, + weights: [0.5, 0.5], + idsGroups: [ + ['a0', 'a1', 'a2', 'a3', 'a4', 'a5'], + ['b0', 'b1', 'b2', 'b3', 'b4', 'b5'], + ], + idsExpected: array_merge( + ['a0', 'a1', 'a2', 'a3', 'a4'], + ['b0', 'b1', 'b2', 'b3', 'b4'], + ), + ); + } + + public function testPick2(): void + { + $this->prepareAndTest( + limit: 10, + weights: [0.5, 0.5], + idsGroups: [ + ['a0', 'a1', 'a2'], + ['b0', 'b1', 'b2', 'b3', 'b4', 'b5', 'b6', 'b7'], + ], + idsExpected: array_merge( + ['a0', 'a1', 'a2'], + ['b0', 'b1', 'b2', 'b3', 'b4', 'b5', 'b6'], + ), + ); + } + + public function testPick3(): void + { + $this->prepareAndTest( + limit: 10, + weights: [0.5, 0.5], + idsGroups: [ + ['a0', 'a1', 'a2', 'a4', 'a5', 'a6', 'a7'], + ['b0', 'b1', 'b2'], + ], + idsExpected: + ['a0', 'a1', 'a2', 'a4', 'a5', 'b0', 'b1', 'b2', 'a6', 'a7'], + ); + } + + public function testPick4(): void + { + $this->prepareAndTest( + limit: 10, + weights: [0.5, 0.5], + idsGroups: [ + ['a0', 'a1', 'a2'], + ['b0', 'b1', 'b2'], + ], + idsExpected: + ['a0', 'a1', 'a2', 'b0', 'b1', 'b2'], + ); + } + + public function testPick5(): void + { + $this->prepareAndTest( + limit: 15, + weights: [0.5, 0.5], + idsGroups: [ + ['a0', 'a1', 'a2'], + ['b0', 'b1', 'b2', 'b3', 'b4', 'b5', 'b6', 'b7', 'b8', 'b9', 'b10', 'b11', 'b12'], + ], + idsExpected: array_merge( + ['a0', 'a1', 'a2'], + ['b0', 'b1', 'b2', 'b3', 'b4', 'b5', 'b6', 'b7', 'b8', 'b9', 'b10', 'b11'], + ), + ); + } + + public function testPick6(): void + { + $this->prepareAndTest( + limit: 10, + weights: [0.5, 0.5], + idsGroups: [ + [], + ['b0', 'b1', 'b2'], + ], + idsExpected: + [ 'b0', 'b1', 'b2'], + ); + } + + public function testPick7(): void + { + $this->prepareAndTest( + limit: 10, + weights: [0.5, 0.5], + idsGroups: [ + ['a0', 'a1', 'a2', 'a3', 'a4', 'a5', 'a6'], + ['b0', 'b1', 'b2', 'b3', 'b4', 'b5'], + ], + idsExpected: array_merge( + ['a0', 'a1', 'a2', 'a3', 'a4'], + ['b0', 'b1', 'b2', 'b3', 'b4'], + ), + ); + } + + public function testPick8(): void + { + $this->prepareAndTest( + limit: 10, + weights: [0.5, 0.3, 0.2], + idsGroups: [ + ['a0', 'a1', 'a2', 'a3', 'a4', 'a5', 'a6'], + ['b0', 'b1', 'b2', 'b3', 'b4', 'b5'], + ['c0', 'c1', 'c2', 'c3', 'c4', 'c5'], + ], + idsExpected: array_merge( + ['a0', 'a1', 'a2', 'a3', 'a4'], + ['b0', 'b1', 'b2'], + ['c0', 'c1'], + ), + ); + } +}