Aaron Schulz has uploaded a new change for review.
https://gerrit.wikimedia.org/r/307329
Change subject: objectcache: allow for callbacks to mask SYNC_WRITE latency
......................................................................
objectcache: allow for callbacks to mask SYNC_WRITE latency
Also made RedisBagOStuff respect it and made a common
wait loop class.
Change-Id: I908222ad3788ebe330aa58831cda139da32becd8
---
M autoload.php
A includes/libs/WaitConditionLoop.php
M includes/libs/objectcache/BagOStuff.php
M includes/objectcache/RedisBagOStuff.php
M includes/objectcache/SqlBagOStuff.php
5 files changed, 252 insertions(+), 25 deletions(-)
git pull ssh://gerrit.wikimedia.org:29418/mediawiki/core
refs/changes/29/307329/1
diff --git a/autoload.php b/autoload.php
index 39102fd..0f9d2db 100644
--- a/autoload.php
+++ b/autoload.php
@@ -1498,6 +1498,7 @@
'VirtualRESTService' => __DIR__ .
'/includes/libs/virtualrest/VirtualRESTService.php',
'VirtualRESTServiceClient' => __DIR__ .
'/includes/libs/virtualrest/VirtualRESTServiceClient.php',
'WANObjectCache' => __DIR__ .
'/includes/libs/objectcache/WANObjectCache.php',
+ 'WaitConditionLoop' => __DIR__ . '/includes/libs/WaitConditionLoop.php',
'WantedCategoriesPage' => __DIR__ .
'/includes/specials/SpecialWantedcategories.php',
'WantedFilesPage' => __DIR__ .
'/includes/specials/SpecialWantedfiles.php',
'WantedPagesPage' => __DIR__ .
'/includes/specials/SpecialWantedpages.php',
diff --git a/includes/libs/WaitConditionLoop.php
b/includes/libs/WaitConditionLoop.php
new file mode 100644
index 0000000..1448e32
--- /dev/null
+++ b/includes/libs/WaitConditionLoop.php
@@ -0,0 +1,144 @@
+<?php
+/**
+ * Wait loop that reaches a condition or times out.
+ *
+ * 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
+ * @author Aaron Schulz
+ */
+
+/**
+ * Wait loop that reaches a condition or times out
+ * @since 1.29
+ */
+class WaitConditionLoop {
+ /** @var callable */
+ private $condition;
+ /** @var callable[] */
+ private $busyCallbacks = [];
+ /** @var float Seconds */
+ private $timeout;
+ /** @var float Seconds */
+ private $lastWaitTime;
+
+ /**
+ * Keep checking $condition() until the StatusValue it returns is
"good" or not "OK"
+ * per isGood() and isOK(), or until the timeout specified by $timeout
is reached
+ *
+ * @param callable $condition Callback that returns a StatusValue
+ * @param float $timeout Timeout in seconds
+ * @param array &$busyCallbacks List of callbacks to do useful work (by
reference)
+ */
+ public function __construct( callable $condition, $timeout = 5.0,
&$busyCallbacks = [] ) {
+ $this->condition = $condition;
+ $this->timeout = $timeout;
+ $this->busyCallbacks =& $busyCallbacks;
+ }
+
+ /**
+ * Invoke the loop
+ *
+ * The resulting status will fall into three possibilities:
+ * - If isGood(), then the condition was reached
+ * - If not isGood() but isOK(), then a timeout was reached
+ * - If not isOK(), then some serious error occured in the callback
+ *
+ * @return StatusValue
+ */
+ public function invoke() {
+ $elapsed = 0.0; // seconds
+ $sleepUs = 0; // microseconds to sleep each time
+ do {
+ $checkStartTime = microtime( true );
+ // Check if the condition is met yet
+ $realStart = microtime( true );
+ $cpuStart = $this->getCpuTime();
+ /** @var StatusValue $status */
+ $status = call_user_func( $this->condition );
+ $cpu = $this->getCpuTime() - $cpuStart;
+ $real = microtime( true ) - $realStart;
+ // Exit if the condition is reached
+ if ( $status->isGood() || !$status->isOK() ) {
+ $this->lastWaitTime = $elapsed;
+
+ return $status;
+ }
+ // Detect if the callback seems to block or if justs
burns CPU
+ $callbackBlocksViaIO = ( $real > 0.100 && $cpu <= $real
* .03 );
+ if ( $this->runBusyCallbacks( 1 ) == 0 &&
!$callbackBlocksViaIO ) {
+ // 10 queries = 10(10+100)/2 ms = 550ms, 14
queries = 1050ms
+ $sleepUs = min( $sleepUs + 10 * 1e3, 1e6 ); //
stop incrementing at ~1s
+ usleep( $sleepUs );
+ }
+ $checkEndTime = microtime( true );
+ // The max() protects against the clock getting set back
+ $elapsed += max( $checkEndTime - $checkStartTime, 0.010
);
+ } while ( $elapsed < $this->timeout );
+
+ $status->error( 'timeout' );
+ $this->lastWaitTime = $elapsed;
+
+ return $status;
+ }
+
+ /**
+ * @return float Seconds
+ */
+ public function getLastWaitTime() {
+ return $this->lastWaitTime;
+ }
+
+ /**
+ * @return float Returns 0.0 if not supported (Windows on PHP < 7)
+ */
+ private function getCpuTime() {
+ $time = 0.0;
+
+ if ( defined( 'HHVM_VERSION' ) && PHP_OS === 'Linux' ) {
+ $ru = getrusage( 2 /* RUSAGE_THREAD */ );
+ } else {
+ $ru = getrusage( 0 /* RUSAGE_SELF */ );
+ }
+ if ( $ru ) {
+ $time += $ru['ru_utime.tv_sec'] +
$ru['ru_utime.tv_usec'] / 1e6;
+ $time += $ru['ru_stime.tv_sec'] +
$ru['ru_stime.tv_usec'] / 1e6;
+ }
+
+ return $time;
+ }
+
+ /**
+ * Run callbacks to due useful things instead of blocking for stuff to
happen
+ *
+ * @param integer $maxRun Maximum number of tasks to run
+ * @return integer Number of callbacks run this time
+ * @since 1.28
+ */
+ protected function runBusyCallbacks( $maxRun = INF ) {
+ $runCount = 0;
+ foreach ( $this->busyCallbacks as $i => $workCallback ) {
+ if ( $runCount++ >= $maxRun ) {
+ break;
+ }
+ unset( $this->busyCallbacks[$i] ); // consume
+ $workCallback();
+ }
+
+ return $runCount;
+ }
+
+}
diff --git a/includes/libs/objectcache/BagOStuff.php
b/includes/libs/objectcache/BagOStuff.php
index 5472e83..616f07f 100644
--- a/includes/libs/objectcache/BagOStuff.php
+++ b/includes/libs/objectcache/BagOStuff.php
@@ -45,30 +45,28 @@
abstract class BagOStuff implements IExpiringStore, LoggerAwareInterface {
/** @var array[] Lock tracking */
protected $locks = [];
-
/** @var integer ERR_* class constant */
protected $lastError = self::ERR_NONE;
-
/** @var string */
protected $keyspace = 'local';
-
/** @var LoggerInterface */
protected $logger;
-
/** @var callback|null */
protected $asyncHandler;
+ /** @var integer Seconds */
+ protected $syncTimeout;
/** @var bool */
private $debugMode = false;
-
/** @var array */
private $duplicateKeyLookups = [];
-
/** @var bool */
private $reportDupes = false;
-
/** @var bool */
private $dupeTrackScheduled = false;
+
+ /** @var callable[] */
+ protected $busyCallbacks = [];
/** @var integer[] Map of (ATTR_* class constant => QOS_* class
constant) */
protected $attrMap = [];
@@ -86,6 +84,8 @@
const WRITE_SYNC = 1; // synchronously write to all locations for
replicated stores
const WRITE_CACHE_ONLY = 2; // Only change state of the in-memory cache
+ const SYNC_TIMEOUT_SEC = 3;
+
/**
* $params include:
* - logger: Psr\Log\LoggerInterface instance
@@ -94,6 +94,7 @@
* In CLI mode, it should run the task immediately.
* - reportDupes: Whether to emit warning log messages for all keys
that were
* requested more than once (requires an asyncHandler).
+ * - syncTimeout: How long to wait with WRITE_SYNC in seconds.
* @param array $params
*/
public function __construct( array $params = [] ) {
@@ -114,6 +115,10 @@
if ( !empty( $params['reportDupes'] ) && is_callable(
$this->asyncHandler ) ) {
$this->reportDupes = true;
}
+
+ $this->syncTimeout = isset( $params['syncTimeout'] )
+ ? $params['syncTimeout']
+ : self::SYNC_TIMEOUT_SEC;
}
/**
@@ -643,6 +648,19 @@
}
/**
+ * Let a callback be run to avoid wasting time on special blocking calls
+ *
+ * The result of the callback can be fetched by reapBusyCallback() using
+ * the request ID this returns. It will run the method if this had not
done so yet.
+ *
+ * @param callable $workCallback
+ * @since 1.28
+ */
+ public function addBusyCallback( callable $workCallback ) {
+ $this->busyCallbacks[] = $workCallback;
+ }
+
+ /**
* Modify a cache update operation array for EventRelayer::notify()
*
* This is used for relayed writes, e.g. for broadcasting a change
diff --git a/includes/objectcache/RedisBagOStuff.php
b/includes/objectcache/RedisBagOStuff.php
index c3e0c96..099c1eb 100644
--- a/includes/objectcache/RedisBagOStuff.php
+++ b/includes/objectcache/RedisBagOStuff.php
@@ -115,6 +115,9 @@
// No expiry, that is very different from zero
expiry in Redis
$result = $conn->set( $key, $this->serialize(
$value ) );
}
+ if ( ( $flags & self::WRITE_SYNC ) == self::WRITE_SYNC
) {
+ $result = $this->waitForReplication( $conn ) &&
$result;
+ }
} catch ( RedisException $e ) {
$result = false;
$this->handleException( $conn, $e );
@@ -410,7 +413,7 @@
* not. The safest response for us is to explicitly destroy the
connection
* object and let it be reopened during the next request.
* @param RedisConnRef $conn
- * @param Exception $e
+ * @param RedisException $e
*/
protected function handleException( RedisConnRef $conn, $e ) {
$this->setLastError( BagOStuff::ERR_UNEXPECTED );
@@ -418,6 +421,55 @@
}
/**
+ * @param RedisConnRef $conn
+ * @return bool
+ */
+ protected function waitForReplication( RedisConnRef $conn ) {
+ // http://redis.io/commands/info
+ $info = $conn->info( 'replication' );
+ if ( !isset( $info['slave0'] ) || !isset(
$info['master_repl_offset'] ) ) {
+ return true; // no slaves configured
+ }
+
+ $masterPos = (int)$info['master_repl_offset'];
+ $loop = new WaitConditionLoop(
+ function () use ( $conn, $masterPos ) {
+ $status = StatusValue::newGood();
+ if ( !$this->wasReplicated( $masterPos,
$conn->info( 'replication' ) ) ) {
+ $status->error( 'notsynced' );
+ }
+ return $status;
+ },
+ $this->syncTimeout,
+ $this->busyCallbacks
+ );
+
+ return $loop->invoke()->isGood();
+ }
+
+ /**
+ * @param integer $masterPos
+ * @param array $info Info from 'replication' group
+ * @return bool
+ */
+ private function wasReplicated( $masterPos, array $info ) {
+ foreach ( $info as $key => $value ) {
+ $m = [];
+ // https://github.com/antirez/redis/issues/2375
+ if ( preg_match( '/^slave\d+$/', $key )
+ && preg_match( '/offset=(\d+)/', $value, $m )
+ ) {
+ $slavePos = (int)$m[1];
+ if ( $slavePos < $masterPos ) {
+ return false;
+ }
+ }
+ }
+
+ return true;
+ }
+
+ /**
* Send information about a single request to the debug log
* @param string $method
* @param string $key
diff --git a/includes/objectcache/SqlBagOStuff.php
b/includes/objectcache/SqlBagOStuff.php
index 7a89991..68eeb34 100644
--- a/includes/objectcache/SqlBagOStuff.php
+++ b/includes/objectcache/SqlBagOStuff.php
@@ -377,7 +377,7 @@
public function set( $key, $value, $exptime = 0, $flags = 0 ) {
$ok = $this->setMulti( [ $key => $value ], $exptime );
if ( ( $flags & self::WRITE_SYNC ) == self::WRITE_SYNC ) {
- $ok = $ok && $this->waitForSlaves();
+ $ok = $this->waitForReplication() && $ok;
}
return $ok;
@@ -489,7 +489,7 @@
public function merge( $key, callable $callback, $exptime = 0,
$attempts = 10, $flags = 0 ) {
$ok = $this->mergeViaCas( $key, $callback, $exptime, $attempts
);
if ( ( $flags & self::WRITE_SYNC ) == self::WRITE_SYNC ) {
- $ok = $ok && $this->waitForSlaves();
+ $ok = $ok && $this->waitForReplication();
}
return $ok;
@@ -797,22 +797,34 @@
return !$this->serverInfos;
}
- protected function waitForSlaves() {
- if ( $this->usesMainDB() ) {
- $lb = $this->getSeparateMainLB()
- ?:
MediaWikiServices::getInstance()->getDBLoadBalancer();
- // Main LB is used; wait for any slaves to catch up
- try {
- $pos = $lb->getMasterPos();
- if ( $pos ) {
- return $lb->waitForAll( $pos, 3 );
- }
- } catch ( DBReplicationWaitError $e ) {
- return false;
- }
+ protected function waitForReplication() {
+ if ( !$this->usesMainDB() ) {
+ // Custom DB server list; probably doesn't use
replication
+ return true;
}
- // Custom DB server list; probably doesn't use replication
- return true;
+ $lb = $this->getSeparateMainLB()
+ ?:
MediaWikiServices::getInstance()->getDBLoadBalancer();
+
+ if ( $lb->getServerCount() <= 1 ) {
+ return true; // no slaves
+ }
+
+ // Main LB is used; wait for any slaves to catch up
+ $masterPos = $lb->getConnection( DB_MASTER )->getMasterPos();
+
+ $loop = new WaitConditionLoop(
+ function () use ( $lb, $masterPos ) {
+ $status = StatusValue::newGood();
+ if ( !$lb->waitForAll( $masterPos, 1 ) ) {
+ $status->error( 'not_synced' );
+ }
+ return $status;
+ },
+ $this->syncTimeout,
+ $this->busyCallbacks
+ );
+
+ return $loop->invoke()->isGood();
}
}
--
To view, visit https://gerrit.wikimedia.org/r/307329
To unsubscribe, visit https://gerrit.wikimedia.org/r/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I908222ad3788ebe330aa58831cda139da32becd8
Gerrit-PatchSet: 1
Gerrit-Project: mediawiki/core
Gerrit-Branch: master
Gerrit-Owner: Aaron Schulz <[email protected]>
_______________________________________________
MediaWiki-commits mailing list
[email protected]
https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits
