Big job batching

CHANGELOG-WIP.md

@@ -66,6 +66,7 @@
 - Element source definitions can now include a `defaultSourcePath` key.
 - Element custom field validation now respects the list of attributes passed to `validate()`.
 - Improving IDE autocompletion for chained query param calls. ([#12656](https://github.com/craftcms/cms/pull/12656))
+- Added `craft\base\Batchable`.
 - Added `craft\base\Element::cpRevisionsUrl()`.
 - Added `craft\base\Element::indexElements()`.
 - Added `craft\base\ElementInterface::findSource()`.
@@ -81,6 +82,7 @@
 - Added `craft\console\ControllerTrait::success()`.
 - Added `craft\console\ControllerTrait::tip()`.
 - Added `craft\console\ControllerTrait::warning()`.
+- Added `craft\db\QueryBatcher`.
 - Added `craft\debug\DumpPanel`.
 - Added `craft\elements\conditions\assets\ViewableConditionRule`. ([#12266](https://github.com/craftcms/cms/pull/12266))
 - Added `craft\elements\conditions\entries\ViewableConditionRule`. ([#12266](https://github.com/craftcms/cms/pull/12266))
@@ -101,6 +103,7 @@
 - Added `craft\models\ImageTransform::$upscale`.
 - Added `craft\models\VolumeFolder::getHasChildren()`.
 - Added `craft\models\VolumeFolder::setHasChildren()`.
+- Added `craft\queue\BaseBatchableJob`. ([#12638](https://github.com/craftcms/cms/pull/12638))
 - Added `craft\queue\jobs\GenerateImageTransform`. ([#12340](https://github.com/craftcms/cms/pull/12340))
 - Added `craft\services\Assets::createFolderQuery()`.
 - Added `craft\services\Assets::foldersExist()`.
@@ -143,6 +146,7 @@
 ### System
 - Improved element deletion performance. ([#12223](https://github.com/craftcms/cms/pull/12223))
 - Improved queue performance. ([#12274](https://github.com/craftcms/cms/issues/12274), [#12340](https://github.com/craftcms/cms/pull/12340))
+- “Applying new propagation method to elements”, “Propagating [element type]”, and “Resaving [element type]” queue jobs are now splt up into batches of up to 100 items. ([#12638](https://github.com/craftcms/cms/pull/12638))
 - Assets’ alternative text values are now included as search keywords.
 - Updated LitEmoji to v4. ([#12226](https://github.com/craftcms/cms/discussions/12226))
 - Fixed a database deadlock error that could occur when updating a relation or structure position for an element that was simultaneously being saved. ([#9905](https://github.com/craftcms/cms/issues/9905))

src/base/Batchable.php

@@ -0,0 +1,29 @@
+<?php
+/**
+ * @link https://craftcms.com/
+ * @copyright Copyright (c) Pixel & Tonic, Inc.
+ * @license https://craftcms.github.io/license/
+ */
+
+namespace craft\base;
+
+use Countable;
+
+/**
+ * Batchable defines the common interface to be implemented by classes that
+ * provide items which can be counted and accessed in slices.
+ *
+ * @author Pixel & Tonic, Inc. <[email protected]>
+ * @since 3.4.0
+ */
+interface Batchable extends Countable
+{
+    /**
+     * Returns a slice of the items
+     *
+     * @param int $offset
+     * @param int $limit
+     * @return iterable
+     */
+    public function getSlice(int $offset, int $limit): iterable;
+}

src/db/QueryBatcher.php

@@ -0,0 +1,85 @@
+<?php
+/**
+ * @link https://craftcms.com/
+ * @copyright Copyright (c) Pixel & Tonic, Inc.
+ * @license https://craftcms.github.io/license/
+ */
+
+namespace craft\db;
+
+use craft\base\Batchable;
+use yii\db\Connection as YiiConnection;
+use yii\db\Query as YiiQuery;
+use yii\db\QueryInterface;
+
+/**
+ * QueryBatcher provides a [[Batchable]] wrapper for a given [[QueryInterface]] object.
+ *
+ * @author Pixel & Tonic, Inc. <[email protected]>
+ * @since 4.4.0
+ */
+class QueryBatcher implements Batchable
+{
+    /**
+     * Constructor
+     *
+     * :::warning
+     * The query should have [[QueryInterface::orderBy()|`orderBy`]] set on it, ideally to the table’s primary key
+     * column. That will ensure that the rows returned in result batches are consecutive.
+     * :::
+     *
+     * @param QueryInterface $query
+     * @param YiiConnection|null $db
+     */
+    public function __construct(
+        private QueryInterface $query,
+        private ?YiiConnection $db = null,
+    ) {
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function count(): int
+    {
+        try {
+            return $this->query->count(db: $this->db);
+        } catch (QueryAbortedException) {
+            return 0;
+        }
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function getSlice(int $offset, int $limit): iterable
+    {
+        /** @var YiiQuery $query */
+        $query = $this->query;
+
+        if (is_int($query->limit)) {
+            // Don't go passed the query's limit
+            if ($offset >= $query->limit) {
+                return [];
+            }
+            $limit = min($limit, $query->limit - $offset);
+        }
+
+        $queryOffset = $query->offset;
+        $queryLimit = $query->limit;
+
+        try {
+            $slice = $query
+                ->offset((is_int($queryOffset) ? $queryOffset : 0) + $offset)
+                ->limit($limit)
+                ->all();
+        } catch (QueryAbortedException) {
+            $slice = [];
+        }
+
+        $query->offset($queryOffset);
+        $query->limit($queryLimit);
+
+        return $slice;
+    }
+}

src/helpers/Queue.php

@@ -8,6 +8,7 @@
 namespace craft\helpers;
 
 use Craft;
+use craft\queue\BaseBatchedJob;
 use yii\base\NotSupportedException;
 use yii\queue\JobInterface;
 use yii\queue\Queue as BaseQueue;
@@ -41,6 +42,12 @@ public static function push(
             $queue = Craft::$app->getQueue();
         }
 
+        if ($job instanceof BaseBatchedJob) {
+            // Keep track of the priority and TTR in case there will be additional jobs
+            $job->priority = $priority;
+            $job->ttr = $ttr;
+        }
+
         try {
             return $queue
                 ->priority($priority)

src/queue/BaseBatchedJob.php

@@ -0,0 +1,175 @@
+<?php
+/**
+ * @link https://craftcms.com/
+ * @copyright Copyright (c) Pixel & Tonic, Inc.
+ * @license https://craftcms.github.io/license/
+ */
+
+namespace craft\queue;
+
+use Craft;
+use craft\base\Batchable;
+use craft\helpers\ConfigHelper;
+use craft\helpers\Queue as QueueHelper;
+use craft\i18n\Translation;
+
+/**
+ * BaseBatchedJob is the base class for large jobs that may need to spawn
+ * additional jobs to complete the workload.
+ *
+ * ::: warning
+ * Batched jobs should *always* be pushed to the queue using [[QueueHelper::push()]],
+ * so the `priority` and `ttr` settings can be maintained for additional spawned jobs.
+ * :::
+ *
+ * :::warning
+ * Spawned jobs are cloned from the current job, so any public properties that are set to objects which aren’t
+ * `serialize()`-friendly should be excluded via `__sleep()`, and any private/protected properties will need
+ * to be reset to their default values via `__wakeup()` to avoid uninitialized property errors.
+ * :::
+ *
+ * @author Pixel & Tonic, Inc. <[email protected]>
+ * @since 4.4.0
+ */
+abstract class BaseBatchedJob extends BaseJob
+{
+    /**
+     * @var int The number of items that should be processed in a single batch
+     */
+    public int $batchSize = 100;
+
+    /**
+     * @var int The index of the current batch (starting with `0`)
+     */
+    public int $batchIndex = 0;
+
+    /**
+     * @var int The offset to start fetching items by.
+     */
+    public int $itemOffset = 0;
+
+    /**
+     * @var int|null The job’s priority
+     */
+    public ?int $priority = null;
+
+    /**
+     * @var int|null The job’s TTR
+     */
+    public ?int $ttr = null;
+
+    private ?Batchable $_data = null;
+    private ?int $_totalItems = null;
+
+    public function __sleep(): array
+    {
+        return array_keys(Craft::getObjectVars($this));
+    }
+
+    /**
+     * Loads the batchable data.
+     *
+     * @return Batchable
+     */
+    abstract protected function loadData(): Batchable;
+
+    /**
+     * Returns the batchable data.
+     *
+     * @return Batchable
+     */
+    final protected function data(): Batchable
+    {
+        if (!isset($this->_data)) {
+            $this->_data = $this->loadData();
+        }
+        return $this->_data;
+    }
+
+    /**
+     * Returns the total number of items across all the batches.
+     *
+     * @return int
+     */
+    final protected function totalItems(): int
+    {
+        if (!isset($this->_totalItems)) {
+            $this->_totalItems = $this->data()->count();
+        }
+        return $this->_totalItems;
+    }
+
+    /**
+     * Returns the total number of batches.
+     *
+     * @return int
+     */
+    final protected function totalBatches(): int
+    {
+        return (int)ceil($this->totalItems() / $this->batchSize);
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function execute($queue): void
+    {
+        $items = $this->data()->getSlice($this->itemOffset, $this->batchSize);
+        $totalInBatch = is_array($items) ? count($items) : iterator_count($items);
+
+        $memoryLimit = ConfigHelper::sizeInBytes(ini_get('memory_limit'));
+        $startMemory = $memoryLimit != -1 ? memory_get_usage() : null;
+
+        $i = 0;
+
+        foreach ($items as $item) {
+            $this->setProgress($queue, $i / $totalInBatch, Translation::prep('app', '{step, number} of {total, number}', [
+                'step' => $this->itemOffset + 1,
+                'total' => $this->totalItems(),
+            ]));
+            $this->processItem($item);
+            $this->itemOffset++;
+            $i++;
+
+            // Make sure we're not getting uncomfortably close to the memory limit, every 10 items
+            if ($startMemory !== null && $i % 10 === 0) {
+                $memory = memory_get_usage();
+                $avgMemory = ($memory - $startMemory) / $i;
+                if ($memory + ($avgMemory * 15) > $memoryLimit) {
+                    break;
+                }
+            }
+        }
+
+        // Spawn another job if there are more items
+        if ($this->itemOffset < $this->totalItems()) {
+            $nextJob = clone $this;
+            $nextJob->batchIndex++;
+            QueueHelper::push($nextJob, $this->priority, 0, $this->ttr, $queue);
+        }
+    }
+
+    /**
+     * Processes an item.
+     *
+     * @param mixed $item
+     */
+    abstract protected function processItem(mixed $item);
+
+    /**
+     * @inheritdoc
+     */
+    final public function getDescription(): ?string
+    {
+        $description = $this->description ?? $this->defaultDescription();
+        $totalBatches = $this->totalBatches();
+        if ($totalBatches <= 1) {
+            return $description;
+        }
+        return Craft::t('app', '{description} (batch {index, number} of {total, number})', [
+            'description' => Translation::translate($description),
+            'index' => $this->batchIndex + 1,
+            'total' => $totalBatches,
+        ]);
+    }
+}

src/queue/BaseJob.php

@@ -37,6 +37,12 @@ abstract class BaseJob extends BaseObject implements JobInterface
      */
     private ?string $_progressLabel = null;
 
+    public function __wakeup(): void
+    {
+        $this->_progress = 0;
+        $this->_progressLabel = null;
+    }
+
     /**
      * @inheritdoc
      */