diff --git a/application/Espo/Core/Job/AbstractQueueJob.php b/application/Espo/Core/Job/AbstractQueueJob.php index 980d5986eb..b42b4b7317 100644 --- a/application/Espo/Core/Job/AbstractQueueJob.php +++ b/application/Espo/Core/Job/AbstractQueueJob.php @@ -29,11 +29,16 @@ namespace Espo\Core\Job; +use Espo\Core\Utils\DateTime; + use Espo\ORM\EntityManager; +use Espo\Entities\Job as JobEntity; + +use DateTimeImmutable; use RuntimeException; -abstract class AbstractQueueJob implements Job +abstract class AbstractQueueJob implements JobPreperable { protected $queue = null; @@ -43,6 +48,8 @@ abstract class AbstractQueueJob implements Job private $entityManager; + private const SHIFT_PERIOD = '5 seconds'; + public function __construct( JobManager $jobManager, QueuePortionNumberProvider $portionNumberProvider, @@ -61,6 +68,74 @@ abstract class AbstractQueueJob implements Job $limit = $this->portionNumberProvider->get($this->queue); - $this->jobManager->processQueue($this->queue, $limit); + $group = $data->get('group'); + + $this->jobManager->processQueue($this->queue, $group, $limit); + } + + public function prepare(ScheduledJobData $data, DateTimeImmutable $executeTime): void + { + $groupList = []; + + $shiftPeriod = self::SHIFT_PERIOD; + + $query = $this->entityManager + ->getQueryBuilder() + ->select('group') + ->from(JobEntity::ENTITY_TYPE) + ->where([ + 'status' => JobStatus::PENDING, + 'queue' => $this->queue, + 'executeTime<=' => $executeTime + ->modify($shiftPeriod) + ->format(DateTime::SYSTEM_DATE_TIME_FORMAT), + ]) + ->groupBy('group') + ->build(); + + $sth = $this->entityManager->getQueryExecutor()->execute($query); + + while ($row = $sth->fetch()) { + $groupList[] = $row['group'] ?? null; + } + + if (!count($groupList)) { + return; + } + + foreach ($groupList as $group) { + $existingJob = $this->entityManager + ->getRDBRepository(JobEntity::ENTITY_TYPE) + ->select('id') + ->where([ + 'scheduledJobId' => $data->getId(), + 'targetGroup' => $group, + 'status' => [ + JobStatus::RUNNING, + JobStatus::READY, + ] + ]) + ->findOne(); + + if ($existingJob) { + continue; + } + + $name = $data->getName(); + + if ($group) { + $name .= ' :: ' . $group; + } + + $this->entityManager->createEntity(JobEntity::ENTITY_TYPE, [ + 'scheduledJobId' => $data->getId(), + 'executeTime' => $executeTime->format(DateTime::SYSTEM_DATE_TIME_FORMAT), + 'name' => $data->getName(), + 'data' => [ + 'group' => $group, + ], + 'targetGroup' => $group, + ]); + } } } diff --git a/application/Espo/Core/Job/JobManager.php b/application/Espo/Core/Job/JobManager.php index a1b6f80a14..9857a31645 100644 --- a/application/Espo/Core/Job/JobManager.php +++ b/application/Espo/Core/Job/JobManager.php @@ -164,11 +164,12 @@ class JobManager /** * Process pending jobs from a specific queue. Jobs within a queue are processed one by one. */ - public function processQueue(string $queue, int $limit): void + public function processQueue(string $queue, ?string $group, int $limit): void { $params = QueueProcessorParams ::create() ->withQueue($queue) + ->withGroup($group) ->withLimit($limit) ->withUseProcessPool(false) ->withNoLock(true); diff --git a/application/Espo/Core/Job/MetadataProvider.php b/application/Espo/Core/Job/MetadataProvider.php new file mode 100644 index 0000000000..3cb0d3e66d --- /dev/null +++ b/application/Espo/Core/Job/MetadataProvider.php @@ -0,0 +1,62 @@ +metadata = $metadata; + } + + /** + * @return string[] + */ + public function getPreparableJobNameList(): array + { + $list = []; + + $items = $this->metadata->get(['app', 'scheduledJobs']) ?? []; + + foreach ($items as $name => $item) { + $isPreparable = $item['isPreparable'] ?? false; + + if ($isPreparable) { + $list[] = $name; + } + } + + return $list; + } +} \ No newline at end of file diff --git a/application/Espo/Core/Job/QueueProcessor.php b/application/Espo/Core/Job/QueueProcessor.php index a2c619cff0..d402362440 100644 --- a/application/Espo/Core/Job/QueueProcessor.php +++ b/application/Espo/Core/Job/QueueProcessor.php @@ -77,6 +77,7 @@ class QueueProcessor $pendingJobList = $this->queueUtil->getPendingJobList( $this->params->getQueue(), + $this->params->getGroup(), $this->params->getLimit() ); diff --git a/application/Espo/Core/Job/QueueProcessorParams.php b/application/Espo/Core/Job/QueueProcessorParams.php index f71d747d30..608c58528a 100644 --- a/application/Espo/Core/Job/QueueProcessorParams.php +++ b/application/Espo/Core/Job/QueueProcessorParams.php @@ -37,6 +37,8 @@ class QueueProcessorParams private $queue = null; + private $group = null; + private $limit = 0; public function withUseProcessPool(bool $useProcessPool): self @@ -66,6 +68,15 @@ class QueueProcessorParams return $obj; } + public function withGroup(?string $group): self + { + $obj = clone $this; + + $obj->group = $group; + + return $obj; + } + public function withLimit(int $limit): self { $obj = clone $this; @@ -90,6 +101,11 @@ class QueueProcessorParams return $this->queue; } + public function getGroup(): ?string + { + return $this->group; + } + public function getLimit(): int { return $this->limit; diff --git a/application/Espo/Core/Job/QueueUtil.php b/application/Espo/Core/Job/QueueUtil.php index 77c34d5828..38098f634e 100644 --- a/application/Espo/Core/Job/QueueUtil.php +++ b/application/Espo/Core/Job/QueueUtil.php @@ -50,15 +50,22 @@ class QueueUtil private $scheduleUtil; + private $metadataProvider; + private const NOT_EXISTING_PROCESS_PERIOD = 300; private const READY_NOT_STARTED_PERIOD = 60; - public function __construct(Config $config, EntityManager $entityManager, ScheduleUtil $scheduleUtil) - { + public function __construct( + Config $config, + EntityManager $entityManager, + ScheduleUtil $scheduleUtil, + MetadataProvider $metadataProvider + ) { $this->config = $config; $this->entityManager = $entityManager; $this->scheduleUtil = $scheduleUtil; + $this->metadataProvider = $metadataProvider; } public function isJobPending(string $id): bool @@ -79,7 +86,10 @@ class QueueUtil return $job->get('status') === JobStatus::PENDING; } - public function getPendingJobList(?string $queue = null, int $limit = 0): Collection + /** + * @return JobEntity[] + */ + public function getPendingJobList(?string $queue = null, ?string $group = null, int $limit = 0): Collection { $builder = $this->entityManager ->getRDBRepository(JobEntity::ENTITY_TYPE) @@ -92,6 +102,7 @@ class QueueUtil 'targetType', 'methodName', 'serviceName', + 'className', 'job', 'data', ]) @@ -99,6 +110,7 @@ class QueueUtil 'status' => JobStatus::PENDING, 'executeTime<=' => DateTimeUtil::getSystemNowString(), 'queue' => $queue, + 'group' => $group, ]) ->order('number'); @@ -137,15 +149,16 @@ class QueueUtil $list = []; $jobList = $this->entityManager - ->getRepository(JobEntity::ENTITY_TYPE) + ->getRDBRepository(JobEntity::ENTITY_TYPE) ->select(['scheduledJobId']) + ->leftJoin('scheduledJob') ->where([ 'status' => [ JobStatus::RUNNING, JobStatus::READY, ], 'scheduledJobId!=' => null, - 'targetId=' => null, + 'scheduledJob.job!=' => $this->metadataProvider->getPreparableJobNameList(), ]) ->order('executeTime') ->find(); @@ -363,10 +376,12 @@ class QueueUtil $duplicateJobList = $this->entityManager ->getRepository(JobEntity::ENTITY_TYPE) ->select(['scheduledJobId']) + ->leftJoin('scheduledJob') ->where([ 'scheduledJobId!=' => null, 'status' => JobStatus::PENDING, 'executeTime<=' => DateTimeUtil::getSystemNowString(), + 'scheduledJob.job!=' => $this->metadataProvider->getPreparableJobNameList(), 'targetId' => null, ]) ->groupBy(['scheduledJobId']) diff --git a/application/Espo/Entities/Job.php b/application/Espo/Entities/Job.php index 0e0baf2f70..610ec1b49a 100644 --- a/application/Espo/Entities/Job.php +++ b/application/Espo/Entities/Job.php @@ -77,6 +77,22 @@ class Job extends Entity return $this->get('targetId'); } + /** + * Get a group. + */ + public function getGroup(): ?string + { + return $this->get('group'); + } + + /** + * Get a queue. + */ + public function getQueue(): ?string + { + return $this->get('queue'); + } + /** * Get data. */ diff --git a/application/Espo/Resources/metadata/app/metadata.json b/application/Espo/Resources/metadata/app/metadata.json index df7292486c..58b98a2d34 100644 --- a/application/Espo/Resources/metadata/app/metadata.json +++ b/application/Espo/Resources/metadata/app/metadata.json @@ -21,6 +21,7 @@ ["app", "massActions", "__ANY__", "implementationClassName"], ["app", "actions", "__ANY__", "implementationClassName"], ["app", "fieldProcessing"], + ["app", "scheduledJobs"], ["selectDefs"], ["recordDefs"], ["aclDefs"], diff --git a/application/Espo/Resources/metadata/app/scheduledJobs.json b/application/Espo/Resources/metadata/app/scheduledJobs.json new file mode 100644 index 0000000000..738cea84b1 --- /dev/null +++ b/application/Espo/Resources/metadata/app/scheduledJobs.json @@ -0,0 +1,17 @@ +{ + "ProcessJobQueueQ0": { + "isPreparable": true + }, + "ProcessJobQueueQ1": { + "isPreparable": true + }, + "ProcessJobQueueE0": { + "isPreparable": true + }, + "CheckEmailAccounts": { + "isPreparable": true + }, + "CheckInboundEmails": { + "isPreparable": true + } +} diff --git a/application/Espo/Resources/metadata/entityDefs/Job.json b/application/Espo/Resources/metadata/entityDefs/Job.json index 1c152fe610..3b1f22fbc0 100644 --- a/application/Espo/Resources/metadata/entityDefs/Job.json +++ b/application/Espo/Resources/metadata/entityDefs/Job.json @@ -14,7 +14,8 @@ "Failed": "danger", "Running": "warning", "Ready": "warning" - } + }, + "maxLength": 16 }, "executeTime": { "type": "datetime", @@ -64,6 +65,16 @@ "maxLength": 36, "default": null }, + "group": { + "type": "varchar", + "maxLength": 64, + "default": null + }, + "targetGroup": { + "type": "varchar", + "maxLength": 64, + "default": null + }, "startedAt": { "type": "datetime", "hasSeconds": true @@ -117,6 +128,12 @@ }, "status": { "columns": ["status", "deleted"] + }, + "statusScheduledJobId": { + "columns": ["status", "scheduledJobId"] + }, + "queueGroupExecuteTime": { + "columns": ["queue", "status", "group", "executeTime"] } } } diff --git a/tests/integration/Espo/Core/Job/JobTest.php b/tests/integration/Espo/Core/Job/JobTest.php index 0fdb4233db..73e205ad22 100644 --- a/tests/integration/Espo/Core/Job/JobTest.php +++ b/tests/integration/Espo/Core/Job/JobTest.php @@ -58,20 +58,86 @@ class JobTest extends \tests\integration\Core\BaseTestCase $this->entityManager = $this->getContainer()->get('entityManager'); } - public function testProcessQueue(): void + public function testProcessQueueNoGroup(): void { $job = $this->entityManager->createEntity('Job', [ 'job' => 'Dummy', 'queue' => 'q0', ]); - $this->jobManager->processQueue('q0', 10); + $this->jobManager->processQueue('q0', null, 10); $jobReloaded = $this->entityManager->getEntity('Job', $job->id); $this->assertEquals(JobStatus::SUCCESS, $jobReloaded->getStatus()); } + public function testProcessQueueGroupAll(): void + { + $job1 = $this->entityManager->createEntity('Job', [ + 'job' => 'Dummy', + 'queue' => 'q0', + ]); + + $job2 = $this->entityManager->createEntity('Job', [ + 'job' => 'Dummy', + 'queue' => 'q0', + 'group' => 'group-1', + ]); + + $job3 = $this->entityManager->createEntity('Job', [ + 'job' => 'Dummy', + 'queue' => 'q0', + 'group' => 'group-1', + ]); + + $this->jobManager->process(); + + $job1Reloaded = $this->entityManager->getEntity('Job', $job1->getId()); + $job2Reloaded = $this->entityManager->getEntity('Job', $job2->getId()); + $job3Reloaded = $this->entityManager->getEntity('Job', $job3->getId()); + + $this->assertEquals(JobStatus::SUCCESS, $job1Reloaded->getStatus()); + $this->assertEquals(JobStatus::SUCCESS, $job2Reloaded->getStatus()); + $this->assertEquals(JobStatus::SUCCESS, $job3Reloaded->getStatus()); + } + + public function testProcessQueueGroupSeparate(): void + { + $job1 = $this->entityManager->createEntity('Job', [ + 'job' => 'Dummy', + 'queue' => 'q0', + ]); + + $job2 = $this->entityManager->createEntity('Job', [ + 'job' => 'Dummy', + 'queue' => 'q0', + 'group' => 'group-1', + ]); + + $job3 = $this->entityManager->createEntity('Job', [ + 'job' => 'Dummy', + 'queue' => 'q0', + 'group' => 'group-1', + ]); + + $this->jobManager->processQueue('q0', 'group-1', 100); + + $job1Reloaded = $this->entityManager->getEntity('Job', $job1->getId()); + $job2Reloaded = $this->entityManager->getEntity('Job', $job2->getId()); + $job3Reloaded = $this->entityManager->getEntity('Job', $job3->getId()); + + $this->assertEquals(JobStatus::PENDING, $job1Reloaded->getStatus()); + $this->assertEquals(JobStatus::SUCCESS, $job2Reloaded->getStatus()); + $this->assertEquals(JobStatus::SUCCESS, $job3Reloaded->getStatus()); + + $this->jobManager->processQueue('q0', null, 100); + + $job1Reloaded2 = $this->entityManager->getEntity('Job', $job1->getId()); + + $this->assertEquals(JobStatus::SUCCESS, $job1Reloaded2->getStatus()); + } + public function testRunJobById(): void { $job = $this->entityManager->createEntity('Job', [ diff --git a/tests/unit/Espo/Core/Job/MetadataProviderTest.php b/tests/unit/Espo/Core/Job/MetadataProviderTest.php new file mode 100644 index 0000000000..451463f9a1 --- /dev/null +++ b/tests/unit/Espo/Core/Job/MetadataProviderTest.php @@ -0,0 +1,70 @@ +metadata = $this->createMock(Metadata::class); + } + + public function testGetPreparableJobNameList() + { + $this->metadata + ->method('get') + ->with(['app', 'scheduledJobs']) + ->willReturn([ + 'Test1' => [ + 'isPreparable' => true, + ], + 'Test2' => [ + 'isPreparable' => true, + ], + 'Test3' => [], + 'Test4' => [ + 'isPreparable' => false, + ], + ]); + + $provider = new MetadataProvider($this->metadata); + + $list = $provider->getPreparableJobNameList(); + + $this->assertEquals(['Test1', 'Test2'], $list); + } +} diff --git a/tests/unit/Espo/Core/Job/QueueProcessorParamsTest.php b/tests/unit/Espo/Core/Job/QueueProcessorParamsTest.php index 191809bf6e..edd068851a 100644 --- a/tests/unit/Espo/Core/Job/QueueProcessorParamsTest.php +++ b/tests/unit/Espo/Core/Job/QueueProcessorParamsTest.php @@ -60,11 +60,14 @@ class QueueProcessorParamsTest extends \PHPUnit\Framework\TestCase ->withLimit(10) ->withUseProcessPool(true) ->withNoLock(true) + ->withGroup('group-0') ->withQueue('q0'); $this->assertTrue($params->useProcessPool()); $this->assertTrue($params->noLock()); $this->assertEquals('q0', $params->getQueue()); + + $this->assertEquals('group-0', $params->getGroup()); } }