Do not start explicit transaction rounds for RecentChangesUpdateJob

[lhc/web/wiklou.git] / includes / jobqueue / JobRunner.php

diff --git a/includes/jobqueue/JobRunner.php b/includes/jobqueue/JobRunner.php
index 49b7a45..977fbda 100644 (file)
--- a/includes/jobqueue/JobRunner.php
+++ b/includes/jobqueue/JobRunner.php
@@ -119,6 +119,7 @@ class JobRunner implements LoggerAwareInterface {
                        $response['reached'] = 'none-possible';
                        return $response;
                }
+
                // Bail out if DB is in read-only mode
                if ( wfReadOnly() ) {
                        $response['reached'] = 'read-only';
@@ -126,6 +127,9 @@ class JobRunner implements LoggerAwareInterface {
                }
 
                $lbFactory = MediaWikiServices::getInstance()->getDBLoadBalancerFactory();
+               if ( $lbFactory->hasTransactionRound() ) {
+                       throw new LogicException( __METHOD__ . ' called with an active transaction round.' );
+               }
                // Bail out if there is too much DB lag.
                // This check should not block as we want to try other wiki queues.
                list( , $maxLag ) = $lbFactory->getMainLB( wfWikiID() )->getMaxLag();
@@ -134,9 +138,6 @@ class JobRunner implements LoggerAwareInterface {
                        return $response;
                }
 
-               // Flush any pending DB writes for sanity
-               $lbFactory->commitAll( __METHOD__ );
-
                // Catch huge single updates that lead to replica DB lag
                $trxProfiler = Profiler::instance()->getTransactionProfiler();
                $trxProfiler->setLogger( LoggerFactory::getInstance( 'DBPerformance' ) );
@@ -170,7 +171,6 @@ class JobRunner implements LoggerAwareInterface {
                        } else {
                                $job = $group->pop( $type ); // job from a single queue
                        }
-                       $lbFactory->commitMasterChanges( __METHOD__ ); // flush any JobQueueDB writes
 
                        if ( $job ) { // found a job
                                ++$jobsPopped;
@@ -193,7 +193,6 @@ class JobRunner implements LoggerAwareInterface {
                                $info = $this->executeJob( $job, $lbFactory, $stats, $popTime );
                                if ( $info['status'] !== false || !$job->allowRetries() ) {
                                        $group->ack( $job ); // succeeded or job cannot be retried
-                                       $lbFactory->commitMasterChanges( __METHOD__ ); // flush any JobQueueDB writes
                                }
 
                                // Back off of certain jobs for a while (for throttling and for errors)
@@ -281,7 +280,9 @@ class JobRunner implements LoggerAwareInterface {
        private function executeJob( Job $job, LBFactory $lbFactory, $stats, $popTime ) {
                $jType = $job->getType();
                $msg = $job->toString() . " STARTING";
-               $this->logger->debug( $msg );
+               $this->logger->debug( $msg, [
+                       'job_type' => $job->getType(),
+               ] );
                $this->debugCallback( $msg );
 
                // Run the job...
@@ -289,7 +290,9 @@ class JobRunner implements LoggerAwareInterface {
                $jobStartTime = microtime( true );
                try {
                        $fnameTrxOwner = get_class( $job ) . '::run'; // give run() outer scope
-                       $lbFactory->beginMasterChanges( $fnameTrxOwner );
+                       if ( !$job->hasExecutionFlag( $job::JOB_NO_EXPLICIT_TRX_ROUND ) ) {
+                               $lbFactory->beginMasterChanges( $fnameTrxOwner );
+                       }
                        $status = $job->run();
                        $error = $job->getLastError();
                        $this->commitMasterChanges( $lbFactory, $job, $fnameTrxOwner );
@@ -339,12 +342,23 @@ class JobRunner implements LoggerAwareInterface {
                }
 
                if ( $status === false ) {
+                       $msg = $job->toString() . " t={job_duration} error={job_error}";
+                       $this->logger->error( $msg, [
+                               'job_type' => $job->getType(),
+                               'job_duration' => $timeMs,
+                               'job_error' => $error,
+                       ] );
+
                        $msg = $job->toString() . " t=$timeMs error={$error}";
-                       $this->logger->error( $msg );
                        $this->debugCallback( $msg );
                } else {
+                       $msg = $job->toString() . " t={job_duration} good";
+                       $this->logger->info( $msg, [
+                               'job_type' => $job->getType(),
+                               'job_duration' => $timeMs,
+                       ] );
+
                        $msg = $job->toString() . " t=$timeMs good";
-                       $this->logger->info( $msg );
                        $this->debugCallback( $msg );
                }
 
@@ -488,9 +502,14 @@ class JobRunner implements LoggerAwareInterface {
                }
                $usedBytes = memory_get_usage();
                if ( $maxBytes && $usedBytes >= 0.95 * $maxBytes ) {
+                       $msg = "Detected excessive memory usage ({used_bytes}/{max_bytes}).";
+                       $this->logger->error( $msg, [
+                               'used_bytes' => $usedBytes,
+                               'max_bytes' => $maxBytes,
+                       ] );
+
                        $msg = "Detected excessive memory usage ($usedBytes/$maxBytes).";
                        $this->debugCallback( $msg );
-                       $this->logger->error( $msg );
 
                        return false;
                }
@@ -552,8 +571,14 @@ class JobRunner implements LoggerAwareInterface {
                }
 
                $ms = intval( 1000 * $time );
+
+               $msg = $job->toString() . " COMMIT ENQUEUED [{job_commit_write_ms}ms of writes]";
+               $this->logger->info( $msg, [
+                       'job_type' => $job->getType(),
+                       'job_commit_write_ms' => $ms,
+               ] );
+
                $msg = $job->toString() . " COMMIT ENQUEUED [{$ms}ms of writes]";
-               $this->logger->info( $msg );
                $this->debugCallback( $msg );
 
                // Wait for an exclusive lock to commit