'JobQueueAggregatorRedis' => __DIR__ . '/includes/jobqueue/aggregator/JobQueueAggregatorRedis.php',
'JobQueueConnectionError' => __DIR__ . '/includes/jobqueue/JobQueue.php',
'JobQueueDB' => __DIR__ . '/includes/jobqueue/JobQueueDB.php',
+ 'JobQueueEnqueueUpdate' => __DIR__ . '/includes/deferred/JobQueueEnqueueUpdate.php',
'JobQueueError' => __DIR__ . '/includes/jobqueue/JobQueue.php',
'JobQueueFederated' => __DIR__ . '/includes/jobqueue/JobQueueFederated.php',
'JobQueueGroup' => __DIR__ . '/includes/jobqueue/JobQueueGroup.php',
__METHOD__
);
- // Important: this must be the last deferred update added (T100085, T154425)
- DeferredUpdates::addCallableUpdate( [ JobQueueGroup::class, 'pushLazyJobs' ] );
-
// Do any deferred jobs; preferring to run them now if a client will not wait on them
DeferredUpdates::doUpdates( $blocksHttpClient ? 'enqueue' : 'run' );
--- /dev/null
+<?php
+/**
+ * Handler for triggering the enqueuing of lazy-pushed jobs
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+ * http://www.gnu.org/copyleft/gpl.html
+ *
+ * @file
+ */
+
+use Wikimedia\Assert\Assert;
+
+/**
+ * Enqueue lazy-pushed jobs that have accumulated from JobQueueGroup
+ *
+ * @ingroup JobQueue
+ * @since 1.33
+ */
+class JobQueueEnqueueUpdate implements DeferrableUpdate, MergeableUpdate {
+ /** @var array[] Map of (domain ID => IJobSpecification[]) */
+ private $jobsByDomain;
+
+ /**
+ * @param string $domain DB domain ID
+ * @param IJobSpecification[] $jobs
+ */
+ public function __construct( $domain, array $jobs ) {
+ $this->jobsByDomain[$domain] = $jobs;
+ }
+
+ public function merge( MergeableUpdate $update ) {
+ /** @var JobQueueEnqueueUpdate $update */
+ Assert::parameterType( __CLASS__, $update, '$update' );
+
+ foreach ( $update->jobsByDomain as $domain => $jobs ) {
+ $this->jobsByDomain[$domain] = $this->jobsByDomain[$domain] ?? [];
+ $this->jobsByDomain[$domain] = array_merge( $this->jobsByDomain[$domain], $jobs );
+ }
+ }
+
+ public function doUpdate() {
+ foreach ( $this->jobsByDomain as $domain => $jobs ) {
+ $group = JobQueueGroup::singleton( $domain );
+ try {
+ $group->push( $jobs );
+ } catch ( Exception $e ) {
+ // Get in as many jobs as possible and let other post-send updates happen
+ MWExceptionHandler::logException( $e );
+ }
+ }
+ }
+}
/** @var array Map of (bucket => (queue => JobQueue, types => list of types) */
protected $coalescedQueues;
- /** @var Job[] */
- protected $bufferedJobs = [];
-
const TYPE_DEFAULT = 1; // integer; jobs popped by default
const TYPE_ANY = 2; // integer; any job
// Throw errors now instead of on push(), when other jobs may be buffered
$this->assertValidJobs( $jobs );
- $this->bufferedJobs = array_merge( $this->bufferedJobs, $jobs );
+ DeferredUpdates::addUpdate( new JobQueueEnqueueUpdate( $this->wiki, $jobs ) );
}
/**
*
* @return void
* @since 1.26
+ * @deprecated Since 1.33 Not needed anymore
*/
public static function pushLazyJobs() {
- foreach ( self::$instances as $group ) {
- try {
- $group->push( $group->bufferedJobs );
- $group->bufferedJobs = [];
- } catch ( Exception $e ) {
- // Get in as many jobs as possible and let other post-send updates happen
- MWExceptionHandler::logException( $e );
- }
- }
+ wfDeprecated( __METHOD__, '1.33' );
}
/**
}
}
}
-
- function __destruct() {
- $n = count( $this->bufferedJobs );
- if ( $n > 0 ) {
- $type = implode( ', ', array_unique( array_map( 'get_class', $this->bufferedJobs ) ) );
- trigger_error( __METHOD__ . ": $n buffered job(s) of type(s) $type never inserted." );
- }
- }
}
$status = $job->run();
$error = $job->getLastError();
$this->commitMasterChanges( $lbFactory, $job, $fnameTrxOwner );
- // Important: this must be the last deferred update added (T100085, T154425)
- DeferredUpdates::addCallableUpdate( [ JobQueueGroup::class, 'pushLazyJobs' ] );
// Run any deferred update tasks; doUpdates() manages transactions itself
DeferredUpdates::doUpdates();
} catch ( Exception $e ) {