Daniel Kinzler has uploaded a new change for review.
https://gerrit.wikimedia.org/r/253898
Change subject: ChangeDispatcher should use locks on the local DB.
......................................................................
ChangeDispatcher should use locks on the local DB.
Previously, we places global locks on the respective client DB.
In order to avoid opening database connections, we should indetead
manage the locks for all clients on the repo database.
Note: while this will likely solve the "many connections" part of T118162,
it does nothing to clarify the original cause of that issue.
Bug: T118162
Change-Id: Ibea2f574de6398f805cbc2276d6a93d9780e4b9f
---
M repo/includes/store/ChangeDispatchCoordinator.php
M repo/includes/store/sql/SqlChangeDispatchCoordinator.php
M repo/maintenance/dispatchChanges.php
M repo/tests/phpunit/includes/store/sql/SqlChangeDispatchCoordinatorTest.php
4 files changed, 96 insertions(+), 148 deletions(-)
git pull ssh://gerrit.wikimedia.org:29418/mediawiki/extensions/Wikibase
refs/changes/98/253898/1
diff --git a/repo/includes/store/ChangeDispatchCoordinator.php
b/repo/includes/store/ChangeDispatchCoordinator.php
index 12e10ed..ce1211d 100644
--- a/repo/includes/store/ChangeDispatchCoordinator.php
+++ b/repo/includes/store/ChangeDispatchCoordinator.php
@@ -45,8 +45,11 @@
/**
* Initializes the dispatch table by injecting dummy records for all
target wikis
* that are in the configuration but not yet in the dispatch table.
+ *
+ * @param string[] $clientWikiDBs Associative array mapping client wiki
IDs to
+ * client wiki (logical) database names.
*/
- public function initState();
+ public function initState( $clientWikiDBs );
/**
* Attempt to lock the given target wiki. If it can't be locked because
diff --git a/repo/includes/store/sql/SqlChangeDispatchCoordinator.php
b/repo/includes/store/sql/SqlChangeDispatchCoordinator.php
index b4defc3..28f4d58 100644
--- a/repo/includes/store/sql/SqlChangeDispatchCoordinator.php
+++ b/repo/includes/store/sql/SqlChangeDispatchCoordinator.php
@@ -9,6 +9,7 @@
use Wikibase\Lib\Reporting\MessageReporter;
use Wikibase\Lib\Reporting\NullMessageReporter;
use Wikibase\Store\ChangeDispatchCoordinator;
+use Wikimedia\Assert\Assert;
/**
* SQL based implementation of ChangeDispatchCoordinator;
@@ -41,7 +42,7 @@
private $releaseClientLockOverride = null;
/**
- * @var callable Override for !$db->lockIsFree
+ * @var callable Override for $db->lockIsFree
*/
private $isClientLockUsedOverride = null;
@@ -90,23 +91,17 @@
private $messageReporter;
/**
- * @var string The logical name of the repository's database
+ * @var string|false The logical name of the repository's database
*/
private $repoDB;
/**
- * @var string[] Logical names of local client wiki databases, provided
as a mapping of
- * global site ID to database name for each client wiki.
+ * @param string|false $repoDB
*/
- private $clientWikis;
+ public function __construct( $repoDB ) {
+ Assert::parameterType( 'string|boolean', $repoDB, '$repoDB' );
- /**
- * @param string $repoDB
- * @param string[] $clientWikis Mapping of site IDs to database names.
- */
- public function __construct( $repoDB, array $clientWikis ) {
$this->repoDB = $repoDB;
- $this->clientWikis = $clientWikis;
$this->messageReporter = new NullMessageReporter();
}
@@ -284,15 +279,6 @@
}
/**
- * @param string|bool $wikiDB: the logical name of the client wiki's
database.
- *
- * @return LoadBalancer $wikiDB's database load balancer.
- */
- private function getClientLB( $wikiDB ) {
- return wfGetLB( $wikiDB );
- }
-
- /**
* @return DatabaseBase A connection to the repo's master database
*/
private function getRepoMaster() {
@@ -300,27 +286,10 @@
}
/**
- * @param string|bool $wikiDB: the logical name of the client wiki's
database.
- *
- * @return DatabaseBase A connection to $wikiDB's master database
- */
- private function getClientMaster( $wikiDB ) {
- return $this->getClientLB( $wikiDB )->getConnection( DB_MASTER,
array(), $wikiDB );
- }
-
- /**
* @param DatabaseBase $db: the repo database connection to release for
re-use.
*/
private function releaseRepoMaster( DatabaseBase $db ) {
$this->getRepoLB()->reuseConnection( $db );
- }
-
- /**
- * @param string|bool $wikiDB: the logical name of the client wiki's
database.
- * @param DatabaseBase $db: the client database connection to release
for re-use.
- */
- private function releaseClientMaster( $wikiDB, DatabaseBase $db ) {
- $this->getClientLB( $wikiDB )->reuseConnection( $db );
}
/**
@@ -433,8 +402,13 @@
/**
* Initializes the dispatch table by injecting dummy records for all
target wikis
* that are in the configuration but not yet in the dispatch table.
+ *
+ * @param string[] $clientWikiDBs Associative array mapping client wiki
IDs to
+ * client wiki (logical) database names.
+ *
+ * @throws \DBUnexpectedError
*/
- public function initState() {
+ public function initState( $clientWikiDBs ) {
$db = $this->getRepoMaster();
$trackedSiteIds = $db->selectFieldValues(
@@ -444,7 +418,7 @@
__METHOD__
);
- $untracked = array_diff_key( $this->clientWikis, array_flip(
$trackedSiteIds ) );
+ $untracked = array_diff_key( $clientWikiDBs, array_flip(
$trackedSiteIds ) );
foreach ( $untracked as $siteID => $wikiDB ) {
$state = array(
@@ -483,13 +457,6 @@
* @see selectClient()
*/
public function lockClient( $siteID ) {
- if ( !isset( $this->clientWikis[ $siteID ] ) ) {
- throw new MWException( "Wiki not configured: $siteID; "
- ."consider removing it from the " .
$this->stateTable );
- }
-
- $wikiDB = $this->clientWikis[ $siteID ];
-
$this->trace( "Trying $siteID" );
// start transaction
@@ -519,9 +486,11 @@
$state = get_object_vars( $state );
}
+ $lock = $this->getClientLockName( $siteID );
+
if ( $state['chd_lock'] !== null ) {
// bail out if another dispatcher instance is
holding a lock for that wiki
- if ( $this->isClientLockUsed( $wikiDB,
$state['chd_lock'] ) ) {
+ if ( $this->isClientLockUsed( $db, $lock ) ) {
$this->trace( "$siteID is already being
handled by another process."
. " (lock: " .
$state['chd_lock'] . ")" );
@@ -531,23 +500,23 @@
}
}
- $lock = $this->getClientLock( $wikiDB );
+ $ok = $this->engageClientLock( $db, $lock );
- if ( $lock === false ) {
+ if ( !$ok ) {
// This really shouldn't happen, since we
already checked if another process has a lock.
// The write lock we are holding on the
wb_changes_dispatch table should be preventing
// any race conditions.
// However, another process may still hold the
lock if it grabbed it without locking
// wb_changes_dispatch, or if it didn't record
the lock in wb_changes_dispatch.
- $this->trace( "Warning: Failed to acquire lock
on $wikiDB for site $siteID!" );
+ $this->trace( "Warning: Failed to acquire lock
$lock for site $siteID!" );
$db->rollback( __METHOD__ );
$this->releaseRepoMaster( $db );
return false;
}
- $this->trace( "Locked client $siteID" );
+ $this->trace( "Locked client $siteID with $lock" );
$state['chd_lock'] = $lock;
$state['chd_touched'] = wfTimestamp( TS_MW,
$this->now() ); // XXX: use DB time
@@ -568,7 +537,7 @@
$db->commit( __METHOD__ );
$this->releaseRepoMaster( $db );
- $this->trace( "Locked $wikiDB for site $siteID at
{$state['chd_seen']}." );
+ $this->trace( "Locked site $siteID at {$state['chd_seen']}." );
unset( $state['chd_disabled'] ); // don't mess with this.
@@ -594,7 +563,7 @@
$db->begin( __METHOD__ );
try {
- $this->releaseClientLock( $wikiDB, $state['chd_lock'] );
+ $this->releaseClientLock( $db, $state['chd_lock'] );
$state['chd_lock'] = null;
$state['chd_touched'] = wfTimestamp( TS_MW,
$this->now() );
@@ -622,12 +591,12 @@
/**
* Determines the name of the global lock that should be used to lock
the given client.
*
- * @param string $wikiDB: The logical database name of the wiki to lock
+ * @param string $siteID: The site ID of the wiki to lock
*
* @return string the lock name to use.
*/
- private function getClientLockName( $wikiDB ) {
- return "$wikiDB.WikiBase.dispatchChanges";
+ private function getClientLockName( $siteID ) {
+ return "WikiBase.dispatchChanges.$siteID";
}
/**
@@ -635,49 +604,17 @@
*
* The lock is acquired on the client wiki's master DB.
*
- * @param string $wikiDB The logical database name of the wiki
for which to grab a lock.
- * @param string|null $lockName The name of the lock to acquire. If
not given, getClientLockName()
- * will be used to generate an appropriate name.
- *
- * @return string|bool The lock name if the lock was acquired, false
otherwise.
- */
- private function getClientLock( $wikiDB, $lockName = null ) {
- $this->trace( "Trying to get client lock for $wikiDB" );
-
- if ( $lockName === null ) {
- $lockName = $this->getClientLockName( $wikiDB );
- $this->trace( "Lock name not defined for $wikiDB. Got
lock name $lockName." );
- }
-
- $this->trace( "Trying to get client master for $wikiDB" );
- $ok = $this->engageClientLock( $wikiDB, $lockName );
-
- $msg = $ok ? "Set lock for $wikiDB" : "Failed to set lock for
$wikiDB";
- $this->trace( $msg );
-
- return $ok ? $lockName : false;
- }
-
- /**
- * Tries to acquire a global lock on the given client wiki.
- *
- * The lock is acquired on the client wiki's master DB.
- *
- * @param string $wikiDB The logical database name of the wiki for
which to release the lock.
+ * @param DatabaseBase $db The database connection to work on.
* @param string $lock The name of the lock to release.
*
* @return bool whether the lock was released successfully.
*/
- private function engageClientLock( $wikiDB, $lock ) {
+ private function engageClientLock( DatabaseBase $db, $lock ) {
if ( isset( $this->engageClientLockOverride ) ) {
- return call_user_func( $this->engageClientLockOverride,
$wikiDB, $lock );
+ return call_user_func( $this->engageClientLockOverride,
$db, $lock );
}
- $db = $this->getClientMaster( $wikiDB );
- $ok = $db->lock( $lock, __METHOD__ );
- $this->releaseClientMaster( $wikiDB, $db );
-
- return $ok;
+ return $db->lock( $lock, __METHOD__ );
}
/**
@@ -685,21 +622,17 @@
*
* The lock is released on the client wiki's master DB.
*
- * @param string $wikiDB The logical database name of the wiki for
which to release the lock.
+ * @param DatabaseBase $db The database connection to work on.
* @param string $lock The name of the lock to release.
*
* @return bool whether the lock was released successfully.
*/
- private function releaseClientLock( $wikiDB, $lock ) {
+ private function releaseClientLock( DatabaseBase $db, $lock ) {
if ( isset( $this->releaseClientLockOverride ) ) {
- return call_user_func(
$this->releaseClientLockOverride, $wikiDB, $lock );
+ return call_user_func(
$this->releaseClientLockOverride, $db, $lock );
}
- $db = $this->getClientMaster( $wikiDB );
- $ok = $db->unlock( $lock, __METHOD__ );
- $this->releaseClientMaster( $wikiDB, $db );
-
- return $ok;
+ return $db->unlock( $lock, __METHOD__ );
}
/**
@@ -707,21 +640,17 @@
*
* The lock is checked on the client wiki's master DB.
*
- * @param string $wikiDB The logical database name of the wiki for
which to check the lock.
+ * @param DatabaseBase $db The database connection to work on.
* @param string $lock The name of the lock to check.
*
* @return bool true if the given lock is currently held by another
process, false otherwise.
*/
- private function isClientLockUsed( $wikiDB, $lock ) {
+ private function isClientLockUsed( DatabaseBase $db, $lock ) {
if ( isset( $this->isClientLockUsedOverride ) ) {
- return call_user_func( $this->isClientLockUsedOverride,
$wikiDB, $lock );
+ return call_user_func( $this->isClientLockUsedOverride,
$db, $lock );
}
- $db = $this->getClientMaster( $wikiDB );
- $free = $db->lockIsFree( $lock, __METHOD__ );
- $this->releaseClientMaster( $wikiDB, $db );
-
- return !$free;
+ return $db->lockIsFree( $lock, __METHOD__ );
}
private function warn( $message ) {
diff --git a/repo/maintenance/dispatchChanges.php
b/repo/maintenance/dispatchChanges.php
index 1fcf394..9520660 100644
--- a/repo/maintenance/dispatchChanges.php
+++ b/repo/maintenance/dispatchChanges.php
@@ -64,17 +64,39 @@
}
/**
+ * @param SettingsArray $settings
+ *
+ * @return string[] A mapping of client wiki site IDs to logical
database names.
+ */
+ private function getClientWikis( SettingsArray $settings ) {
+ $clientWikis = $settings->getSetting( 'localClientDatabases' );
+
+ // make sure we have a mapping from siteId to database name in
clientWikis:
+ foreach ( $clientWikis as $siteID => $dbName ) {
+ if ( is_int( $siteID ) ) {
+ unset( $clientWikis[$siteID] );
+ $clientWikis[$dbName] = $dbName;
+ }
+ }
+
+ return $clientWikis;
+ }
+
+ /**
* Initializes members from command line options and configuration
settings.
*
+ * @param string[] $clientWikis A mapping of client wiki site IDs to
logical database names.
* @param ChangeLookup $changeLookup
* @param SettingsArray $settings
*
* @return ChangeDispatcher
- * @throws MWException
*/
- private function newChangeDispatcher( ChangeLookup $changeLookup,
SettingsArray $settings ) {
+ private function newChangeDispatcher(
+ array $clientWikis,
+ ChangeLookup $changeLookup,
+ SettingsArray $settings
+ ) {
$repoDB = $settings->getSetting( 'changesDatabase' );
- $clientWikis = $settings->getSetting( 'localClientDatabases' );
$batchChunkFactor = $settings->getSetting(
'dispatchBatchChunkFactor' );
$batchCacheFactor = $settings->getSetting(
'dispatchBatchCacheFactor' );
$subscriptionLookupMode = $settings->getSetting(
'subscriptionLookupMode' );
@@ -90,19 +112,6 @@
$cacheChunkSize = $batchSize * $batchChunkFactor;
$cacheSize = $cacheChunkSize * $batchCacheFactor;
$changesCache = new ChunkCache( $changeLookup, $cacheChunkSize,
$cacheSize );
-
- // make sure we have a mapping from siteId to database name in
clientWikis:
- foreach ( $clientWikis as $siteID => $dbName ) {
- if ( is_int( $siteID ) ) {
- unset( $clientWikis[$siteID] );
- $clientWikis[$dbName] = $dbName;
- }
- }
-
- if ( empty( $clientWikis ) ) {
- throw new MWException( "No client wikis configured!
Please set \$wgWBRepoSettings['localClientDatabases']." );
- }
-
$reporter = new ObservableMessageReporter();
$self = $this; // PHP 5.3...
@@ -112,7 +121,7 @@
}
);
- $coordinator = new SqlChangeDispatchCoordinator( $repoDB,
$clientWikis );
+ $coordinator = new SqlChangeDispatchCoordinator( $repoDB );
$coordinator->setMessageReporter( $reporter );
$coordinator->setBatchSize( $batchSize );
$coordinator->setDispatchInterval( $dispatchInterval );
@@ -158,12 +167,19 @@
$delay = (int)$this->getOption( 'idle-delay', 10 );
$wikibaseRepo = WikibaseRepo::getDefaultInstance();
+ $clientWikis = $this->getClientWikis(
$wikibaseRepo->getSettings() );
+
+ if ( empty( $clientWikis ) ) {
+ throw new MWException( "No client wikis configured!
Please set \$wgWBRepoSettings['localClientDatabases']." );
+ }
+
$dispatcher = $this->newChangeDispatcher(
+ $clientWikis,
$wikibaseRepo->getStore()->getChangeLookup(),
$wikibaseRepo->getSettings()
);
- $dispatcher->getDispatchCoordinator()->initState();
+ $dispatcher->getDispatchCoordinator()->initState( $clientWikis
);
$passes = $maxPasses === PHP_INT_MAX ? "unlimited" : $maxPasses;
$time = $maxTime === PHP_INT_MAX ? "unlimited" : $maxTime;
diff --git
a/repo/tests/phpunit/includes/store/sql/SqlChangeDispatchCoordinatorTest.php
b/repo/tests/phpunit/includes/store/sql/SqlChangeDispatchCoordinatorTest.php
index f11e7a1..3fb5d34 100644
--- a/repo/tests/phpunit/includes/store/sql/SqlChangeDispatchCoordinatorTest.php
+++ b/repo/tests/phpunit/includes/store/sql/SqlChangeDispatchCoordinatorTest.php
@@ -26,15 +26,7 @@
}
private function getCoordinator() {
- $clientWikis = array(
- 'dewiki' => 'dewikidb',
- 'enwiki' => 'enwikidb',
- 'nlwiki' => 'nlwikidb',
- 'ruwiki' => 'ruwikidb',
- 'zhwiki' => 'zhwikidb',
- );
-
- $coordinator = new SqlChangeDispatchCoordinator( false,
$clientWikis );
+ $coordinator = new SqlChangeDispatchCoordinator( false );
$coordinator->setBatchSize( 3 );
$coordinator->setRandomness( 3 );
@@ -51,15 +43,15 @@
return wfTimestamp( TS_UNIX, '20140303000000' );
} );
- $coordinator->setIsClientLockUsedOverride( function( $wikiDB,
$lockName ) {
- return $wikiDB === 'zhwikidb';
+ $coordinator->setIsClientLockUsedOverride( function( $db,
$lockName ) {
+ return $lockName === 'WikiBase.dispatchChanges.zhwiki';
} );
- $coordinator->setEngageClientLockOverride( function( $wikiDB ) {
- return $wikiDB !== 'zhwikidb';
+ $coordinator->setEngageClientLockOverride( function( $db,
$lockName ) {
+ return $lockName !== 'WikiBase.dispatchChanges.zhwiki';
} );
- $coordinator->setReleaseClientLockOverride( function( $wikiDB )
{
+ $coordinator->setReleaseClientLockOverride( function( $db,
$lockName ) {
return true;
} );
@@ -103,7 +95,15 @@
public function testInitState() {
$coordinator = $this->getCoordinator();
- $coordinator->initState();
+ $clientWikis = array(
+ 'dewiki' => 'dewikidb',
+ 'enwiki' => 'enwikidb',
+ 'nlwiki' => 'nlwikidb',
+ 'ruwiki' => 'ruwikidb',
+ 'zhwiki' => 'zhwikidb',
+ );
+
+ $coordinator->initState( $clientWikis );
$rows = $this->fetchChangesDispatchRows();
@@ -183,7 +183,7 @@
'chd_db' => 'dewikidb',
'chd_seen' => '0',
'chd_touched' => '20140303000000',
- 'chd_lock' =>
"dewikidb.WikiBase.dispatchChanges",
+ 'chd_lock' => "WikiBase.dispatchChanges.dewiki",
'chd_disabled' => '0',
),
array(
@@ -204,7 +204,7 @@
'chd_db' => 'dewikidb',
'chd_seen' => '0',
'chd_touched' => '20140101000055',
- 'chd_lock' =>
"dewikidb.WikiBase.dispatchChanges",
+ 'chd_lock' => "WikiBase.dispatchChanges.dewiki",
'chd_disabled' => '0',
),
array(
@@ -224,7 +224,7 @@
'chd_db' => 'dewikidb',
'chd_seen' => 23,
'chd_touched' => '20140101000055',
- 'chd_lock' => "dewiki.WikiBase.dispatchChanges",
+ 'chd_lock' => "WikiBase.dispatchChanges.dewiki",
'chd_disabled' => '0',
);
@@ -378,7 +378,7 @@
'chd_db' => 'nlwikidb',
'chd_seen' => '7',
'chd_touched' => '20140303000000',
- 'chd_lock' =>
'nlwikidb.WikiBase.dispatchChanges',
+ 'chd_lock' =>
'WikiBase.dispatchChanges.nlwiki',
)
),
'locked or disabled' => array(
@@ -392,7 +392,7 @@
'chd_db' => 'dewikidb',
'chd_seen' => '0',
'chd_touched' => '20140303000000',
- 'chd_lock' =>
'dewikidb.WikiBase.dispatchChanges',
+ 'chd_lock' =>
'WikiBase.dispatchChanges.dewiki',
)
),
'broken lock' => array(
@@ -402,7 +402,7 @@
'chd_db' => 'enwikidb',
'chd_seen' => '0',
'chd_touched' => '20140303000000',
- 'chd_lock' =>
'enwikidb.WikiBase.dispatchChanges',
+ 'chd_lock' =>
'WikiBase.dispatchChanges.enwiki',
)
),
'no pending changed' => array(
--
To view, visit https://gerrit.wikimedia.org/r/253898
To unsubscribe, visit https://gerrit.wikimedia.org/r/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Ibea2f574de6398f805cbc2276d6a93d9780e4b9f
Gerrit-PatchSet: 1
Gerrit-Project: mediawiki/extensions/Wikibase
Gerrit-Branch: master
Gerrit-Owner: Daniel Kinzler <[email protected]>
_______________________________________________
MediaWiki-commits mailing list
[email protected]
https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits
