From 08aa4418c30cfc18ccc69a0f0f9cb9e17be6c196 Mon Sep 17 00:00:00 2001 From: Pierre Schmitz Date: Mon, 12 Aug 2013 09:28:15 +0200 Subject: Update to MediaWiki 1.21.1 --- maintenance/nextJobDB.php | 148 +++++++++++++++++----------------------------- 1 file changed, 53 insertions(+), 95 deletions(-) (limited to 'maintenance/nextJobDB.php') diff --git a/maintenance/nextJobDB.php b/maintenance/nextJobDB.php index e66e981b..1be5146e 100644 --- a/maintenance/nextJobDB.php +++ b/maintenance/nextJobDB.php @@ -18,7 +18,6 @@ * http://www.gnu.org/copyleft/gpl.html * * @file - * @todo Make this work on PostgreSQL and maybe other database servers * @ingroup Maintenance */ @@ -33,127 +32,86 @@ class nextJobDB extends Maintenance { public function __construct() { parent::__construct(); $this->mDescription = "Pick a database that has pending jobs"; - $this->addOption( 'type', "The type of job to search for", false, true ); + $this->addOption( 'type', "Search by job type", false, true ); + $this->addOption( 'types', "Space separated list of job types to search for", false, true ); } public function execute() { - global $wgMemc; - $type = $this->getOption( 'type', false ); + global $wgJobTypesExcludedFromDefaultQueue; - $memcKey = 'jobqueue:dbs:v2'; - $pendingDBs = $wgMemc->get( $memcKey ); - - // If the cache entry wasn't present, or in 1% of cases otherwise, - // regenerate the cache. - if ( !$pendingDBs || mt_rand( 0, 100 ) == 0 ) { - $pendingDBs = $this->getPendingDbs(); - $wgMemc->set( $memcKey, $pendingDBs, 300 ); + // job type required/picked + if ( $this->hasOption( 'types' ) ) { + $types = explode( ' ', $this->getOption( 'types' ) ); + } elseif ( $this->hasOption( 'type' ) ) { + $types = array( $this->getOption( 'type' ) ); + } else { + $types = false; } - if ( !$pendingDBs ) { - return; + // Handle any required periodic queue maintenance + $this->executeReadyPeriodicTasks(); + + // Get all the queues with jobs in them + $pendingDBs = JobQueueAggregator::singleton()->getAllReadyWikiQueues(); + if ( !count( $pendingDBs ) ) { + return; // no DBs with jobs or cache is both empty and locked } do { $again = false; - if ( $type === false ) { - $candidates = call_user_func_array( 'array_merge', $pendingDBs ); - } elseif ( isset( $pendingDBs[$type] ) ) { - $candidates = $pendingDBs[$type]; - } else { - $candidates = array(); - } - if ( !$candidates ) { - return; - } - - $candidates = array_values( $candidates ); - $db = $candidates[ mt_rand( 0, count( $candidates ) - 1 ) ]; - if ( !$this->checkJob( $type, $db ) ) { - // This job is not available in the current database. Remove it from - // the cache. - if ( $type === false ) { - foreach ( $pendingDBs as $type2 => $dbs ) { - $pendingDBs[$type2] = array_diff( $pendingDBs[$type2], array( $db ) ); + $candidates = array(); // list of (type, db) + // Flatten the tree of candidates into a flat list so that a random + // item can be selected, weighing each queue (type/db tuple) equally. + foreach ( $pendingDBs as $type => $dbs ) { + if ( + ( is_array( $types ) && in_array( $type, $types ) ) || + ( $types === false && !in_array( $type, $wgJobTypesExcludedFromDefaultQueue ) ) + ) { + foreach ( $dbs as $db ) { + $candidates[] = array( $type, $db ); } - } else { - $pendingDBs[$type] = array_diff( $pendingDBs[$type], array( $db ) ); } + } + if ( !count( $candidates ) ) { + return; // no jobs for this type + } - $wgMemc->set( $memcKey, $pendingDBs, 300 ); + list( $type, $db ) = $candidates[mt_rand( 0, count( $candidates ) - 1 )]; + if ( JobQueueGroup::singleton( $db )->isQueueDeprioritized( $type ) ) { + $pendingDBs[$type] = array_diff( $pendingDBs[$type], array( $db ) ); $again = true; } } while ( $again ); - $this->output( $db . "\n" ); - } - - /** - * Check if the specified database has a job of the specified type in it. - * The type may be false to indicate "all". - * @param $type string - * @param $dbName string - * @return bool - */ - function checkJob( $type, $dbName ) { - $lb = wfGetLB( $dbName ); - $db = $lb->getConnection( DB_MASTER, array(), $dbName ); - if ( $type === false ) { - $conds = Job::defaultQueueConditions( ); + if ( $this->hasOption( 'types' ) ) { + $this->output( $db . " " . $type . "\n" ); } else { - $conds = array( 'job_cmd' => $type ); + $this->output( $db . "\n" ); } - - - $exists = (bool) $db->selectField( 'job', '1', $conds, __METHOD__ ); - $lb->reuseConnection( $db ); - return $exists; } /** - * Get all databases that have a pending job - * @return array + * Do all ready periodic jobs for all databases every 5 minutes (and .1% of the time) + * @return integer */ - private function getPendingDbs() { - global $wgLocalDatabases; - $pendingDBs = array(); - # Cross-reference DBs by master DB server - $dbsByMaster = array(); - foreach ( $wgLocalDatabases as $db ) { - $lb = wfGetLB( $db ); - $dbsByMaster[$lb->getServerName( 0 )][] = $db; - } - - foreach ( $dbsByMaster as $dbs ) { - $dbConn = wfGetDB( DB_MASTER, array(), $dbs[0] ); - - # Padding row for MySQL bug - $pad = str_repeat( '-', 40 ); - $sql = "(SELECT '$pad' as db, '$pad' as job_cmd)"; - foreach ( $dbs as $wikiId ) { - if ( $sql != '' ) { - $sql .= ' UNION '; + private function executeReadyPeriodicTasks() { + global $wgLocalDatabases, $wgMemc; + + $count = 0; + $memcKey = 'jobqueue:periodic:lasttime'; + $timestamp = (int)$wgMemc->get( $memcKey ); // UNIX timestamp or 0 + if ( ( time() - $timestamp ) > 300 || mt_rand( 0, 999 ) == 0 ) { // 5 minutes + if ( $wgMemc->add( "$memcKey:rebuild", 1, 1800 ) ) { // lock + foreach ( $wgLocalDatabases as $db ) { + $count += JobQueueGroup::singleton( $db )->executeReadyPeriodicTasks(); } - - list( $dbName, $tablePrefix ) = wfSplitWikiID( $wikiId ); - $dbConn->tablePrefix( $tablePrefix ); - $jobTable = $dbConn->tableName( 'job' ); - - $sql .= "(SELECT DISTINCT '$wikiId' as db, job_cmd FROM $dbName.$jobTable GROUP BY job_cmd)"; - } - $res = $dbConn->query( $sql, __METHOD__ ); - $first = true; - foreach ( $res as $row ) { - if ( $first ) { - // discard padding row - $first = false; - continue; - } - $pendingDBs[$row->job_cmd][] = $row->db; + $wgMemc->set( $memcKey, time() ); + $wgMemc->delete( "$memcKey:rebuild" ); // unlock } } - return $pendingDBs; + + return $count; } } -- cgit v1.2.2