return;
}
- $fnameEffective = $fname;
// The transaction owner and any caller with the empty transaction ticket can commit
// so that getEmptyTransactionTicket() callers don't risk seeing DBTransactionError.
if ( $this->trxRoundId !== false && $fname !== $this->trxRoundId ) {
$this->trxLogger->info( "$fname: committing on behalf of {$this->trxRoundId}." );
$fnameEffective = $this->trxRoundId;
+ } else {
+ $fnameEffective = $fname;
}
$this->commitMasterChanges( $fnameEffective );
// If a nested caller committed on behalf of $fname, start another empty $fname
// transaction, leaving the caller with the same empty transaction state as before.
if ( $fnameEffective !== $fname ) {
- $this->beginMasterChanges( $fname );
+ $this->beginMasterChanges( $fnameEffective );
}
}
$this->getLazyConnectionRef( DB_MASTER, [], $db->getWikiID() )
);
$db->setTransactionProfiler( $this->trxProfiler );
- if ( $this->trxRoundId !== false ) {
- $this->applyTransactionRoundFlags( $db );
- }
if ( $server['serverIndex'] === $this->getWriterIndex() ) {
+ if ( $this->trxRoundId !== false ) {
+ $this->applyTransactionRoundFlags( $db );
+ }
foreach ( $this->trxRecurringCallbacks as $name => $callback ) {
$db->setTransactionListener( $name, $callback );
}
$backoffs = $this->syncBackoffDeltas( $backoffs, $backoffDeltas, $wait );
}
- $info = $this->executeJob( $job, $stats, $popTime );
+ $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
/**
* @param Job $job
+ * @param LBFactory $lbFactory
* @param StatsdDataFactory $stats
* @param float $popTime
* @return array Map of status/error/timeMs
*/
- private function executeJob( Job $job, $stats, $popTime ) {
+ private function executeJob( Job $job, LBFactory $lbFactory, $stats, $popTime ) {
$jType = $job->getType();
$msg = $job->toString() . " STARTING";
$this->logger->debug( $msg );
$this->debugCallback( $msg );
- $lbFactory = MediaWikiServices::getInstance()->getDBLoadBalancerFactory();
// Run the job...
$rssStart = $this->getMaxRssKb();
$jobStartTime = microtime( true );
try {
+ $fnameTrxOwner = get_class( $job ) . '::run'; // give run() outer scope
+ $lbFactory->beginMasterChanges( $fnameTrxOwner );
$status = $job->run();
$error = $job->getLastError();
- $this->commitMasterChanges( $lbFactory, $job );
-
+ $this->commitMasterChanges( $lbFactory, $job, $fnameTrxOwner );
+ // Run any deferred update tasks; doUpdates() manages transactions itself
DeferredUpdates::doUpdates();
- $this->commitMasterChanges( $lbFactory, $job );
} catch ( Exception $e ) {
MWExceptionHandler::rollbackMasterChangesAndLog( $e );
$status = false;
*
* @param LBFactory $lbFactory
* @param Job $job
+ * @param string $fnameTrxOwner
* @throws DBError
*/
- private function commitMasterChanges( LBFactory $lbFactory, Job $job ) {
+ private function commitMasterChanges( LBFactory $lbFactory, Job $job, $fnameTrxOwner ) {
global $wgJobSerialCommitThreshold;
$lb = $lbFactory->getMainLB( wfWikiID() );
}
if ( !$dbwSerial ) {
- $lbFactory->commitMasterChanges( __METHOD__ );
+ $lbFactory->commitMasterChanges( $fnameTrxOwner );
return;
}
}
// Actually commit the DB master changes
- $lbFactory->commitMasterChanges( __METHOD__ );
+ $lbFactory->commitMasterChanges( $fnameTrxOwner );
// Release the lock
$dbwSerial->unlock( 'jobrunner-serial-commit', __METHOD__ );
}
public function execute() {
- global $wgCommandLineMode;
-
if ( $this->hasOption( 'procs' ) ) {
$procs = intval( $this->getOption( 'procs' ) );
if ( $procs < 1 || $procs > 1000 ) {
$outputJSON = ( $this->getOption( 'result' ) === 'json' );
$wait = $this->hasOption( 'wait' );
- // Enable DBO_TRX for atomicity; JobRunner manages transactions
- // and works well in web server mode already (@TODO: this is a hack)
- $wgCommandLineMode = false;
-
$runner = new JobRunner( LoggerFactory::getInstance( 'runJobs' ) );
if ( !$outputJSON ) {
$runner->setDebugHandler( [ $this, 'debugInternal' ] );
sleep( 1 );
}
-
- $wgCommandLineMode = true;
}
/**