[JobQueue] Added support for using an external DB.

* Added a new ScopedCallback class to handle reuseConnection() easily.

Change-Id: Icfab280f0173c2cb1ac44cc0bd45e9202600687b

diff --git a/includes/AutoLoader.php b/includes/AutoLoader.php
index fd34590..5c23c4d 100644 (file)
--- a/includes/AutoLoader.php
+++ b/includes/AutoLoader.php
@@ -217,6 +217,7 @@ $wgAutoloadLocalClasses = array(
        'Sanitizer' => 'includes/Sanitizer.php',
        'DataUpdate' => 'includes/DataUpdate.php',
        'SqlDataUpdate' => 'includes/SqlDataUpdate.php',
+       'ScopedCallback' => 'includes/ScopedCallback.php',
        'ScopedPHPTimeout' => 'includes/ScopedPHPTimeout.php',
        'SiteConfiguration' => 'includes/SiteConfiguration.php',
        'SiteStats' => 'includes/SiteStats.php',

diff --git a/includes/ScopedCallback.php b/includes/ScopedCallback.php
new file mode 100644 (file)
index 0000000..1d5b26b
--- /dev/null
+++ b/includes/ScopedCallback.php
@@ -0,0 +1,40 @@
+<?php
+/**
+ * This file deals with RAII style scoped callbacks.
+ *
+ * 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
+ */
+
+/**
+ * Class for asserting that a callback happens when an dummy object leaves scope
+ */
+class ScopedCallback {
+       /** @var Closure */
+       protected $callback;
+
+       /**
+        * @param $callback Closure
+        */
+       public function __construct( Closure $callback ) {
+               $this->callback = $callback;
+       }
+
+       function __destruct() {
+               call_user_func( $this->callback );
+       }
+}

diff --git a/includes/job/JobQueueDB.php b/includes/job/JobQueueDB.php
index e4972eb..7f4300c 100644 (file)
--- a/includes/job/JobQueueDB.php
+++ b/includes/job/JobQueueDB.php
@@ -35,6 +35,21 @@ class JobQueueDB extends JobQueue {
        const MAX_JOB_RANDOM  = 2147483647; // integer; 2^31 - 1, used for job_random
        const MAX_OFFSET      = 255; // integer; maximum number of rows to skip
 
+       protected $cluster = false; // string; name of an external DB cluster
+
+       /**
+        * Additional parameters include:
+        *   - cluster : The name of an external cluster registered via LBFactory.
+        *               If not specified, the primary DB cluster for the wiki will be used.
+        *               This can be overridden with a custom cluster so that DB handles will
+        *               be retrieved via LBFactory::getExternalLB() and getConnection().
+        * @param $params array
+        */
+       protected function __construct( array $params ) {
+               parent::__construct( $params );
+               $this->cluster = isset( $params['cluster'] ) ? $params['cluster'] : false;
+       }
+
        /**
         * @see JobQueue::doIsEmpty()
         * @return bool
@@ -51,7 +66,8 @@ class JobQueueDB extends JobQueue {
                        return false;
                }
 
-               $found = $this->getSlaveDB()->selectField( // unclaimed job
+               list( $dbr, $scope ) = $this->getSlaveDB();
+               $found = $dbr->selectField( // unclaimed job
                        'job', '1', array( 'job_cmd' => $this->type, 'job_token' => '' ), __METHOD__
                );
                $wgMemc->add( $key, $found ? 'false' : 'true', self::CACHE_TTL_LONG );
@@ -116,7 +132,7 @@ class JobQueueDB extends JobQueue {
         */
        protected function doBatchPush( array $jobs, $flags ) {
                if ( count( $jobs ) ) {
-                       $dbw = $this->getMasterDB();
+                       list( $dbw, $scope ) = $this->getMasterDB();
 
                        $rowSet = array(); // (sha1 => job) map for jobs that are de-duplicated
                        $rowList = array(); // list of jobs for jobs that are are not de-duplicated
@@ -135,7 +151,7 @@ class JobQueueDB extends JobQueue {
                        $ttl    = self::CACHE_TTL_LONG;
 
                        $dbw->onTransactionIdle(
-                               function() use ( $dbw, $rowSet, $rowList, $atomic, $key, $ttl
+                               function() use ( $dbw, $rowSet, $rowList, $atomic, $key, $ttl, $scope
                        ) {
                                global $wgMemc;
 
@@ -193,7 +209,7 @@ class JobQueueDB extends JobQueue {
                        return false; // queue is empty
                }
 
-               $dbw = $this->getMasterDB();
+               list( $dbw, $scope ) = $this->getMasterDB();
                $dbw->commit( __METHOD__, 'flush' ); // flush existing transaction
 
                $uuid = wfRandomString( 32 ); // pop attempt
@@ -249,7 +265,7 @@ class JobQueueDB extends JobQueue {
        protected function claimRandom( $uuid, $rand, $gte ) {
                global $wgMemc;
 
-               $dbw = $this->getMasterDB();
+               list( $dbw, $scope ) = $this->getMasterDB();
                // Check cache to see if the queue has <= OFFSET items
                $tinyQueue = $wgMemc->get( $this->getCacheKey( 'small' ) );
 
@@ -325,7 +341,7 @@ class JobQueueDB extends JobQueue {
         * @return Row|false
         */
        protected function claimOldest( $uuid ) {
-               $dbw  = $this->getMasterDB();
+               list( $dbw, $scope ) = $this->getMasterDB();
 
                $row = false; // the row acquired
                do {
@@ -386,7 +402,7 @@ class JobQueueDB extends JobQueue {
         */
        protected function recycleStaleJobs() {
                $now   = time();
-               $dbw   = $this->getMasterDB();
+               list( $dbw, $scope ) = $this->getMasterDB();
                $count = 0; // affected rows
 
                if ( !$dbw->lock( "jobqueue-recycle-{$this->type}", __METHOD__, 1 ) ) {
@@ -459,7 +475,7 @@ class JobQueueDB extends JobQueue {
                        throw new MWException( "Job of type '{$job->getType()}' has no ID." );
                }
 
-               $dbw = $this->getMasterDB();
+               list( $dbw, $scope ) = $this->getMasterDB();
                $dbw->commit( __METHOD__, 'flush' ); // flush existing transaction
 
                // Delete a row with a single DELETE without holding row locks over RTTs...
@@ -488,7 +504,8 @@ class JobQueueDB extends JobQueue {
                // deferred till "transaction idle", do the same here, so that the ordering is
                // maintained. Having only the de-duplication registration succeed would cause
                // jobs to become no-ops without any actual jobs that made them redundant.
-               $this->getMasterDB()->onTransactionIdle( function() use ( $params, $key ) {
+               list( $dbw, $scope ) = $this->getMasterDB();
+               $dbw->onTransactionIdle( function() use ( $params, $key, $scope ) {
                        global $wgMemc;
 
                        $timestamp = $wgMemc->get( $key ); // current last timestamp of this job
@@ -536,17 +553,34 @@ class JobQueueDB extends JobQueue {
        }
 
        /**
-        * @return DatabaseBase
+        * @return Array (DatabaseBase, ScopedCallback)
         */
        protected function getSlaveDB() {
-               return wfGetDB( DB_SLAVE, array(), $this->wiki );
+               return $this->getDB( DB_SLAVE );
        }
 
        /**
-        * @return DatabaseBase
+        * @return Array (DatabaseBase, ScopedCallback)
         */
        protected function getMasterDB() {
-               return wfGetDB( DB_MASTER, array(), $this->wiki );
+               return $this->getDB( DB_MASTER );
+       }
+
+       /**
+        * @param $index integer (DB_SLAVE/DB_MASTER)
+        * @return Array (DatabaseBase, ScopedCallback)
+        */
+       protected function getDB( $index ) {
+               $lb = ( $this->cluster !== false )
+                       ? wfGetLBFactory()->getExternalLB( $this->cluster, $this->wiki )
+                       : wfGetLB( $this->wiki );
+               $conn = $lb->getConnection( $index );
+               return array(
+                       $conn,
+                       new ScopedCallback( function() use ( $lb, $conn ) {
+                               $lb->reuseConnection( $conn );
+                       } )
+               );
        }
 
        /**
@@ -554,7 +588,7 @@ class JobQueueDB extends JobQueue {
         * @return array
         */
        protected function insertFields( Job $job ) {
-               $dbw = $this->getMasterDB();
+               list( $dbw, $scope ) = $this->getMasterDB();
                return array(
                        // Fields that describe the nature of the job
                        'job_cmd'       => $job->getType(),