MediaWiki  master
SqlBagOStuff.php
Go to the documentation of this file.
1 <?php
24 use Wikimedia\AtEase\AtEase;
25 use Wikimedia\ObjectFactory;
33 use Wikimedia\ScopedCallback;
34 use Wikimedia\Timestamp\ConvertibleTimestamp;
35 use Wikimedia\WaitConditionLoop;
36 
44  protected $localKeyLb;
46  protected $globalKeyLb;
47 
49  protected $serverInfos = [];
51  protected $serverTags = [];
53  protected $numServerShards;
55  protected $lastGarbageCollect = 0;
57  protected $purgePeriod = 10;
59  protected $purgeLimit = 100;
61  protected $numTableShards = 1;
63  protected $tableName = 'objectcache';
65  protected $replicaOnly = false;
66 
68  protected $conns;
70  protected $connFailureTimes = [];
72  protected $connFailureErrors = [];
73 
75  private static $GC_DELAY_SEC = 1;
76 
78  private static $OP_SET = 'set';
80  private static $OP_ADD = 'add';
82  private static $OP_TOUCH = 'touch';
84  private static $OP_DELETE = 'delete';
85 
87  private const SHARD_LOCAL = 'local';
89  private const SHARD_GLOBAL = 'global';
90 
127  public function __construct( $params ) {
128  parent::__construct( $params );
129 
130  $this->attrMap[self::ATTR_EMULATION] = self::QOS_EMULATION_SQL;
131 
132  if ( isset( $params['servers'] ) || isset( $params['server'] ) ) {
133  $index = 0;
134  foreach ( ( $params['servers'] ?? [ $params['server'] ] ) as $tag => $info ) {
135  $this->serverInfos[$index] = $info;
136  $this->serverTags[$index] = is_string( $tag ) ? $tag : "#$index";
137  ++$index;
138  }
139  // Horizontal partitioning by key hash (if any)
140  $this->numServerShards = count( $this->serverInfos );
141  $this->attrMap[self::ATTR_SYNCWRITES] = self::QOS_SYNCWRITES_NONE;
142  } else {
143  if ( isset( $params['localKeyLB'] ) ) {
144  $this->localKeyLb = ( $params['localKeyLB'] instanceof ILoadBalancer )
145  ? $params['localKeyLB']
146  : ObjectFactory::getObjectFromSpec( $params['localKeyLB'] );
147  }
148  if ( isset( $params['globalKeyLB'] ) ) {
149  $this->globalKeyLb = ( $params['globalKeyLB'] instanceof ILoadBalancer )
150  ? $params['globalKeyLB']
151  : ObjectFactory::getObjectFromSpec( $params['globalKeyLB'] );
152  }
153  $this->localKeyLb = $this->localKeyLb ?: $this->globalKeyLb;
154  if ( !$this->localKeyLb ) {
155  throw new InvalidArgumentException(
156  "Config requires 'server', 'servers', or 'localKeyLB'/'globalKeyLB'"
157  );
158  }
159  // Verticle partitioning by global vs local keys (if any)
160  $this->numServerShards = ( $this->localKeyLb === $this->globalKeyLb ) ? 1 : 2;
161  $this->attrMap[self::ATTR_SYNCWRITES] = self::QOS_SYNCWRITES_BE;
162  }
163  if ( isset( $params['purgePeriod'] ) ) {
164  $this->purgePeriod = intval( $params['purgePeriod'] );
165  }
166  if ( isset( $params['purgeLimit'] ) ) {
167  $this->purgeLimit = intval( $params['purgeLimit'] );
168  }
169  if ( isset( $params['tableName'] ) ) {
170  $this->tableName = $params['tableName'];
171  }
172  if ( isset( $params['shards'] ) ) {
173  $this->numTableShards = intval( $params['shards'] );
174  }
175  // Backwards-compatibility for < 1.34
176  $this->replicaOnly = $params['replicaOnly'] ?? ( $params['slaveOnly'] ?? false );
177  }
178 
187  private function getConnection( $shardIndex ) {
188  // Don't keep timing out trying to connect if the server is down
189  if (
190  isset( $this->connFailureErrors[$shardIndex] ) &&
191  ( $this->getCurrentTime() - $this->connFailureTimes[$shardIndex] ) < 60
192  ) {
193  throw $this->connFailureErrors[$shardIndex];
194  }
195 
196  if ( $shardIndex === self::SHARD_LOCAL ) {
197  $conn = $this->getConnectionViaLoadBalancer( $shardIndex );
198  } elseif ( $shardIndex === self::SHARD_GLOBAL ) {
199  $conn = $this->getConnectionViaLoadBalancer( $shardIndex );
200  } elseif ( is_int( $shardIndex ) ) {
201  if ( isset( $this->serverInfos[$shardIndex] ) ) {
202  $server = $this->serverInfos[$shardIndex];
203  $conn = $this->getConnectionFromServerInfo( $shardIndex, $server );
204  } else {
205  throw new UnexpectedValueException( "Invalid server index #$shardIndex" );
206  }
207  } else {
208  throw new UnexpectedValueException( "Invalid server index '$shardIndex'" );
209  }
210 
211  return $conn;
212  }
213 
219  private function getKeyLocation( $key ) {
220  if ( $this->serverTags ) {
221  // Striped array of database servers
222  if ( count( $this->serverTags ) == 1 ) {
223  $shardIndex = 0; // short-circuit
224  } else {
225  $sortedServers = $this->serverTags;
226  ArrayUtils::consistentHashSort( $sortedServers, $key );
227  reset( $sortedServers );
228  $shardIndex = key( $sortedServers );
229  }
230  } else {
231  // LoadBalancer based configuration
232  $shardIndex = ( strpos( $key, 'global:' ) === 0 && $this->globalKeyLb )
233  ? self::SHARD_GLOBAL
234  : self::SHARD_LOCAL;
235  }
236 
237  if ( $this->numTableShards > 1 ) {
238  $hash = hexdec( substr( md5( $key ), 0, 8 ) ) & 0x7fffffff;
239  $tableIndex = $hash % $this->numTableShards;
240  } else {
241  $tableIndex = null;
242  }
243 
244  return [ $shardIndex, $this->getTableNameByShard( $tableIndex ) ];
245  }
246 
252  private function getTableNameByShard( $index ) {
253  if ( $index !== null && $this->numTableShards > 1 ) {
254  $decimals = strlen( $this->numTableShards - 1 );
255 
256  return $this->tableName . sprintf( "%0{$decimals}d", $index );
257  }
258 
259  return $this->tableName;
260  }
261 
262  protected function doGet( $key, $flags = 0, &$casToken = null ) {
263  $casToken = null;
264 
265  $blobs = $this->fetchBlobMulti( [ $key ] );
266  if ( array_key_exists( $key, $blobs ) ) {
267  $blob = $blobs[$key];
268  $value = $this->unserialize( $blob );
269 
270  $casToken = ( $value !== false ) ? $blob : null;
271 
272  return $value;
273  }
274 
275  return false;
276  }
277 
278  protected function doGetMulti( array $keys, $flags = 0 ) {
279  $values = [];
280 
281  $blobs = $this->fetchBlobMulti( $keys );
282  foreach ( $blobs as $key => $blob ) {
283  $values[$key] = $this->unserialize( $blob );
284  }
285 
286  return $values;
287  }
288 
289  private function fetchBlobMulti( array $keys ) {
290  $values = []; // array of (key => value)
291 
292  $keysByTableByShardIndex = [];
293  foreach ( $keys as $key ) {
294  list( $shardIndex, $tableName ) = $this->getKeyLocation( $key );
295  $keysByTableByShardIndex[$shardIndex][$tableName][] = $key;
296  }
297 
298  $dataRows = [];
299  foreach ( $keysByTableByShardIndex as $shardIndex => $serverKeys ) {
300  try {
301  $db = $this->getConnection( $shardIndex );
302  foreach ( $serverKeys as $tableName => $tableKeys ) {
303  $res = $db->select(
304  $tableName,
305  [ 'keyname', 'value', 'exptime' ],
306  [ 'keyname' => $tableKeys ],
307  __METHOD__,
308  // Approximate write-on-the-fly BagOStuff API via blocking.
309  // This approximation fails if a ROLLBACK happens (which is rare).
310  // We do not want to flush the TRX as that can break callers.
311  $db->trxLevel() ? [ 'LOCK IN SHARE MODE' ] : []
312  );
313  if ( $res === false ) {
314  continue;
315  }
316  foreach ( $res as $row ) {
317  $row->shardIndex = $shardIndex;
318  $row->tableName = $tableName;
319  $dataRows[$row->keyname] = $row;
320  }
321  }
322  } catch ( DBError $e ) {
323  $this->handleReadError( $e, $shardIndex );
324  }
325  }
326 
327  foreach ( $keys as $key ) {
328  if ( isset( $dataRows[$key] ) ) { // HIT?
329  $row = $dataRows[$key];
330  $this->debug( "get: retrieved data; expiry time is " . $row->exptime );
331  $db = null; // in case of connection failure
332  try {
333  $db = $this->getConnection( $row->shardIndex );
334  if ( $this->isExpired( $db, $row->exptime ) ) { // MISS
335  $this->debug( "get: key has expired" );
336  } else { // HIT
337  $values[$key] = $db->decodeBlob( $row->value );
338  }
339  } catch ( DBQueryError $e ) {
340  $this->handleWriteError( $e, $db, $row->shardIndex );
341  }
342  } else { // MISS
343  $this->debug( 'get: no matching rows' );
344  }
345  }
346 
347  return $values;
348  }
349 
350  protected function doSetMulti( array $data, $exptime = 0, $flags = 0 ) {
351  return $this->modifyMulti( $data, $exptime, $flags, self::$OP_SET );
352  }
353 
361  private function modifyMulti( array $data, $exptime, $flags, $op ) {
362  $keysByTableByShardIndex = [];
363  foreach ( $data as $key => $value ) {
364  list( $shardIndex, $tableName ) = $this->getKeyLocation( $key );
365  $keysByTableByShardIndex[$shardIndex][$tableName][] = $key;
366  }
367 
368  $exptime = $this->getExpirationAsTimestamp( $exptime );
369 
370  $result = true;
372  $silenceScope = $this->silenceTransactionProfiler();
373  foreach ( $keysByTableByShardIndex as $shardIndex => $serverKeys ) {
374  $db = null; // in case of connection failure
375  try {
376  $db = $this->getConnection( $shardIndex );
377  $this->occasionallyGarbageCollect( $db ); // expire old entries if any
378  $dbExpiry = $exptime ? $db->timestamp( $exptime ) : $this->getMaxDateTime( $db );
379  } catch ( DBError $e ) {
380  $this->handleWriteError( $e, $db, $shardIndex );
381  $result = false;
382  continue;
383  }
384 
385  foreach ( $serverKeys as $tableName => $tableKeys ) {
386  try {
387  $result = $this->updateTable(
388  $op,
389  $db,
390  $tableName,
391  $tableKeys,
392  $data,
393  $dbExpiry
394  ) && $result;
395  } catch ( DBError $e ) {
396  $this->handleWriteError( $e, $db, $shardIndex );
397  $result = false;
398  }
399 
400  }
401  }
402 
403  if ( $this->fieldHasFlags( $flags, self::WRITE_SYNC ) ) {
404  foreach ( $keysByTableByShardIndex as $shardIndex => $unused ) {
405  $result = $this->waitForReplication( $shardIndex ) && $result;
406  }
407  }
408 
409  return $result;
410  }
411 
423  private function updateTable( $op, $db, $table, $tableKeys, $data, $dbExpiry ) {
424  $success = true;
425 
426  if ( $op === self::$OP_ADD ) {
427  $rows = [];
428  foreach ( $tableKeys as $key ) {
429  $rows[] = [
430  'keyname' => $key,
431  'value' => $db->encodeBlob( $this->serialize( $data[$key] ) ),
432  'exptime' => $dbExpiry
433  ];
434  }
435  $db->delete(
436  $table,
437  [
438  'keyname' => $tableKeys,
439  'exptime <= ' . $db->addQuotes( $db->timestamp() )
440  ],
441  __METHOD__
442  );
443  $db->insert( $table, $rows, __METHOD__, [ 'IGNORE' ] );
444 
445  $success = ( $db->affectedRows() == count( $rows ) );
446  } elseif ( $op === self::$OP_SET ) {
447  $rows = [];
448  foreach ( $tableKeys as $key ) {
449  $rows[] = [
450  'keyname' => $key,
451  'value' => $db->encodeBlob( $this->serialize( $data[$key] ) ),
452  'exptime' => $dbExpiry
453  ];
454  }
455  $db->replace( $table, 'keyname', $rows, __METHOD__ );
456  } elseif ( $op === self::$OP_DELETE ) {
457  $db->delete( $table, [ 'keyname' => $tableKeys ], __METHOD__ );
458  } elseif ( $op === self::$OP_TOUCH ) {
459  $db->update(
460  $table,
461  [ 'exptime' => $dbExpiry ],
462  [
463  'keyname' => $tableKeys,
464  'exptime > ' . $db->addQuotes( $db->timestamp() )
465  ],
466  __METHOD__
467  );
468 
469  $success = ( $db->affectedRows() == count( $tableKeys ) );
470  } else {
471  throw new InvalidArgumentException( "Invalid operation '$op'" );
472  }
473 
474  return $success;
475  }
476 
477  protected function doSet( $key, $value, $exptime = 0, $flags = 0 ) {
478  return $this->modifyMulti( [ $key => $value ], $exptime, $flags, self::$OP_SET );
479  }
480 
481  protected function doAdd( $key, $value, $exptime = 0, $flags = 0 ) {
482  return $this->modifyMulti( [ $key => $value ], $exptime, $flags, self::$OP_ADD );
483  }
484 
485  protected function doCas( $casToken, $key, $value, $exptime = 0, $flags = 0 ) {
486  list( $shardIndex, $tableName ) = $this->getKeyLocation( $key );
487  $exptime = $this->getExpirationAsTimestamp( $exptime );
488 
490  $silenceScope = $this->silenceTransactionProfiler();
491  $db = null; // in case of connection failure
492  try {
493  $db = $this->getConnection( $shardIndex );
494  // (T26425) use a replace if the db supports it instead of
495  // delete/insert to avoid clashes with conflicting keynames
496  $db->update(
497  $tableName,
498  [
499  'keyname' => $key,
500  'value' => $db->encodeBlob( $this->serialize( $value ) ),
501  'exptime' => $exptime
502  ? $db->timestamp( $exptime )
503  : $this->getMaxDateTime( $db )
504  ],
505  [
506  'keyname' => $key,
507  'value' => $db->encodeBlob( $casToken ),
508  'exptime > ' . $db->addQuotes( $db->timestamp() )
509  ],
510  __METHOD__
511  );
512  } catch ( DBQueryError $e ) {
513  $this->handleWriteError( $e, $db, $shardIndex );
514 
515  return false;
516  }
517 
518  $success = (bool)$db->affectedRows();
519  if ( $this->fieldHasFlags( $flags, self::WRITE_SYNC ) ) {
520  $success = $this->waitForReplication( $shardIndex ) && $success;
521  }
522 
523  return $success;
524  }
525 
526  protected function doDeleteMulti( array $keys, $flags = 0 ) {
527  return $this->modifyMulti(
528  array_fill_keys( $keys, null ),
529  0,
530  $flags,
531  self::$OP_DELETE
532  );
533  }
534 
535  protected function doDelete( $key, $flags = 0 ) {
536  return $this->modifyMulti( [ $key => null ], 0, $flags, self::$OP_DELETE );
537  }
538 
539  public function incr( $key, $step = 1, $flags = 0 ) {
540  list( $shardIndex, $tableName ) = $this->getKeyLocation( $key );
541 
542  $newCount = false;
544  $silenceScope = $this->silenceTransactionProfiler();
545  $db = null; // in case of connection failure
546  try {
547  $db = $this->getConnection( $shardIndex );
548  $encTimestamp = $db->addQuotes( $db->timestamp() );
549  $db->update(
550  $tableName,
551  [ 'value = value + ' . (int)$step ],
552  [ 'keyname' => $key, "exptime > $encTimestamp" ],
553  __METHOD__
554  );
555  if ( $db->affectedRows() > 0 ) {
556  $newValue = $db->selectField(
557  $tableName,
558  'value',
559  [ 'keyname' => $key, "exptime > $encTimestamp" ],
560  __METHOD__
561  );
562  if ( $this->isInteger( $newValue ) ) {
563  $newCount = (int)$newValue;
564  }
565  }
566  } catch ( DBError $e ) {
567  $this->handleWriteError( $e, $db, $shardIndex );
568  }
569 
570  return $newCount;
571  }
572 
573  public function decr( $key, $value = 1, $flags = 0 ) {
574  return $this->incr( $key, -$value, $flags );
575  }
576 
577  public function changeTTLMulti( array $keys, $exptime, $flags = 0 ) {
578  return $this->modifyMulti(
579  array_fill_keys( $keys, null ),
580  $exptime,
581  $flags,
582  self::$OP_TOUCH
583  );
584  }
585 
586  protected function doChangeTTL( $key, $exptime, $flags ) {
587  return $this->modifyMulti( [ $key => null ], $exptime, $flags, self::$OP_TOUCH );
588  }
589 
595  private function isExpired( IDatabase $db, $exptime ) {
596  return (
597  $exptime != $this->getMaxDateTime( $db ) &&
598  ConvertibleTimestamp::convert( TS_UNIX, $exptime ) < $this->getCurrentTime()
599  );
600  }
601 
606  private function getMaxDateTime( $db ) {
607  if ( (int)$this->getCurrentTime() > 0x7fffffff ) {
608  return $db->timestamp( 1 << 62 );
609  } else {
610  return $db->timestamp( 0x7fffffff );
611  }
612  }
613 
618  private function occasionallyGarbageCollect( IDatabase $db ) {
619  if (
620  // Random purging is enabled
621  $this->purgePeriod &&
622  // Only purge on one in every $this->purgePeriod writes
623  mt_rand( 0, $this->purgePeriod - 1 ) == 0 &&
624  // Avoid repeating the delete within a few seconds
625  ( $this->getCurrentTime() - $this->lastGarbageCollect ) > self::$GC_DELAY_SEC
626  ) {
627  $garbageCollector = function () use ( $db ) {
629  $db, $this->getCurrentTime(),
630  null,
631  $this->purgeLimit
632  );
633  $this->lastGarbageCollect = time();
634  };
635  if ( $this->asyncHandler ) {
636  $this->lastGarbageCollect = $this->getCurrentTime(); // avoid duplicate enqueues
637  ( $this->asyncHandler )( $garbageCollector );
638  } else {
639  $garbageCollector();
640  }
641  }
642  }
643 
644  public function expireAll() {
645  $this->deleteObjectsExpiringBefore( $this->getCurrentTime() );
646  }
647 
649  $timestamp,
650  callable $progress = null,
651  $limit = INF
652  ) {
654  $silenceScope = $this->silenceTransactionProfiler();
655 
656  $shardIndexes = $this->getServerShardIndexes();
657  shuffle( $shardIndexes );
658 
659  $ok = true;
660 
661  $keysDeletedCount = 0;
662  foreach ( $shardIndexes as $numServersDone => $shardIndex ) {
663  $db = null; // in case of connection failure
664  try {
665  $db = $this->getConnection( $shardIndex );
667  $db,
668  $timestamp,
669  $progress,
670  $limit,
671  $numServersDone,
672  $keysDeletedCount
673  );
674  } catch ( DBError $e ) {
675  $this->handleWriteError( $e, $db, $shardIndex );
676  $ok = false;
677  }
678  }
679 
680  return $ok;
681  }
682 
693  IDatabase $db,
694  $timestamp,
695  $progressCallback,
696  $limit,
697  $serversDoneCount = 0,
698  &$keysDeletedCount = 0
699  ) {
700  $cutoffUnix = ConvertibleTimestamp::convert( TS_UNIX, $timestamp );
701  $tableIndexes = range( 0, $this->numTableShards - 1 );
702  shuffle( $tableIndexes );
703 
704  foreach ( $tableIndexes as $numShardsDone => $tableIndex ) {
705  $continue = null; // last exptime
706  $lag = null; // purge lag
707  do {
708  $res = $db->select(
709  $this->getTableNameByShard( $tableIndex ),
710  [ 'keyname', 'exptime' ],
711  array_merge(
712  [ 'exptime < ' . $db->addQuotes( $db->timestamp( $cutoffUnix ) ) ],
713  $continue ? [ 'exptime >= ' . $db->addQuotes( $continue ) ] : []
714  ),
715  __METHOD__,
716  [ 'LIMIT' => min( $limit, 100 ), 'ORDER BY' => 'exptime' ]
717  );
718 
719  if ( $res->numRows() ) {
720  $row = $res->current();
721  if ( $lag === null ) {
722  $rowExpUnix = ConvertibleTimestamp::convert( TS_UNIX, $row->exptime );
723  $lag = max( $cutoffUnix - $rowExpUnix, 1 );
724  }
725 
726  $keys = [];
727  foreach ( $res as $row ) {
728  $keys[] = $row->keyname;
729  $continue = $row->exptime;
730  }
731 
732  $db->delete(
733  $this->getTableNameByShard( $tableIndex ),
734  [
735  'exptime < ' . $db->addQuotes( $db->timestamp( $cutoffUnix ) ),
736  'keyname' => $keys
737  ],
738  __METHOD__
739  );
740  $keysDeletedCount += $db->affectedRows();
741  }
742 
743  if ( is_callable( $progressCallback ) ) {
744  if ( $lag ) {
745  $continueUnix = ConvertibleTimestamp::convert( TS_UNIX, $continue );
746  $remainingLag = $cutoffUnix - $continueUnix;
747  $processedLag = max( $lag - $remainingLag, 0 );
748  $doneRatio =
749  ( $numShardsDone + $processedLag / $lag ) / $this->numTableShards;
750  } else {
751  $doneRatio = 1;
752  }
753 
754  $overallRatio = ( $doneRatio / $this->numServerShards )
755  + ( $serversDoneCount / $this->numServerShards );
756  call_user_func( $progressCallback, $overallRatio * 100 );
757  }
758  } while ( $res->numRows() && $keysDeletedCount < $limit );
759  }
760  }
761 
767  public function deleteAll() {
769  $silenceScope = $this->silenceTransactionProfiler();
770  foreach ( $this->getServerShardIndexes() as $shardIndex ) {
771  $db = null; // in case of connection failure
772  try {
773  $db = $this->getConnection( $shardIndex );
774  for ( $i = 0; $i < $this->numTableShards; $i++ ) {
775  $db->delete( $this->getTableNameByShard( $i ), '*', __METHOD__ );
776  }
777  } catch ( DBError $e ) {
778  $this->handleWriteError( $e, $db, $shardIndex );
779  return false;
780  }
781  }
782  return true;
783  }
784 
785  public function lock( $key, $timeout = 6, $expiry = 6, $rclass = '' ) {
786  // Avoid deadlocks and allow lock reentry if specified
787  if ( isset( $this->locks[$key] ) ) {
788  if ( $rclass != '' && $this->locks[$key]['class'] === $rclass ) {
789  ++$this->locks[$key]['depth'];
790  return true;
791  } else {
792  return false;
793  }
794  }
795 
796  list( $shardIndex ) = $this->getKeyLocation( $key );
797 
798  $db = null; // in case of connection failure
799  try {
800  $db = $this->getConnection( $shardIndex );
801  $ok = $db->lock( $key, __METHOD__, $timeout );
802  if ( $ok ) {
803  $this->locks[$key] = [ 'class' => $rclass, 'depth' => 1 ];
804  }
805 
806  $this->logger->warning(
807  __METHOD__ . " failed due to timeout for {key}.",
808  [ 'key' => $key, 'timeout' => $timeout ]
809  );
810 
811  return $ok;
812  } catch ( DBError $e ) {
813  $this->handleWriteError( $e, $db, $shardIndex );
814  $ok = false;
815  }
816 
817  return $ok;
818  }
819 
820  public function unlock( $key ) {
821  if ( !isset( $this->locks[$key] ) ) {
822  return false;
823  }
824 
825  if ( --$this->locks[$key]['depth'] <= 0 ) {
826  unset( $this->locks[$key] );
827 
828  list( $shardIndex ) = $this->getKeyLocation( $key );
829 
830  $db = null; // in case of connection failure
831  try {
832  $db = $this->getConnection( $shardIndex );
833  $ok = $db->unlock( $key, __METHOD__ );
834  if ( !$ok ) {
835  $this->logger->warning(
836  __METHOD__ . ' failed to release lock for {key}.',
837  [ 'key' => $key ]
838  );
839  }
840  } catch ( DBError $e ) {
841  $this->handleWriteError( $e, $db, $shardIndex );
842  $ok = false;
843  }
844 
845  return $ok;
846  }
847 
848  return true;
849  }
850 
859  public function makeKeyInternal( $keyspace, $args ) {
860  // SQL schema for 'objectcache' specifies keys as varchar(255). From that,
861  // subtract the number of characters we need for the keyspace and for
862  // the separator character needed for each argument. To handle some
863  // custom prefixes used by thing like WANObjectCache, limit to 205.
864  $keyspace = strtr( $keyspace, ' ', '_' );
865  $charsLeft = 205 - strlen( $keyspace ) - count( $args );
866  foreach ( $args as &$arg ) {
867  $arg = strtr( $arg, [
868  ' ' => '_', // Avoid unnecessary misses from pre-1.35 code
869  ':' => '%3A',
870  ] );
871 
872  // 33 = 32 characters for the MD5 + 1 for the '#' prefix.
873  if ( $charsLeft > 33 && strlen( $arg ) > $charsLeft ) {
874  $arg = '#' . md5( $arg );
875  }
876  $charsLeft -= strlen( $arg );
877  }
878 
879  if ( $charsLeft < 0 ) {
880  return $keyspace . ':BagOStuff-long-key:##' . md5( implode( ':', $args ) );
881  }
882  return $keyspace . ':' . implode( ':', $args );
883  }
884 
893  protected function serialize( $data ) {
894  if ( is_int( $data ) ) {
895  return $data;
896  }
897 
898  $serial = serialize( $data );
899  if ( function_exists( 'gzdeflate' ) ) {
900  $serial = gzdeflate( $serial );
901  }
902 
903  return $serial;
904  }
905 
911  protected function unserialize( $serial ) {
912  if ( $this->isInteger( $serial ) ) {
913  return (int)$serial;
914  }
915 
916  if ( function_exists( 'gzinflate' ) ) {
917  AtEase::suppressWarnings();
918  $decomp = gzinflate( $serial );
919  AtEase::restoreWarnings();
920 
921  if ( $decomp !== false ) {
922  $serial = $decomp;
923  }
924  }
925 
926  return unserialize( $serial );
927  }
928 
934  private function getConnectionViaLoadBalancer( $shardIndex ) {
935  $lb = ( $shardIndex === self::SHARD_LOCAL ) ? $this->localKeyLb : $this->globalKeyLb;
936  if ( $lb->getServerAttributes( $lb->getWriterIndex() )[Database::ATTR_DB_LEVEL_LOCKING] ) {
937  // Use the main connection to avoid transaction deadlocks
938  $conn = $lb->getMaintenanceConnectionRef( DB_MASTER );
939  } else {
940  // If the RDBMs has row/table/page level locking, then use separate auto-commit
941  // connection to avoid needless contention and deadlocks.
942  $conn = $lb->getMaintenanceConnectionRef(
943  $this->replicaOnly ? DB_REPLICA : DB_MASTER, [],
944  false,
945  $lb::CONN_TRX_AUTOCOMMIT
946  );
947  }
948 
949  return $conn;
950  }
951 
958  private function getConnectionFromServerInfo( $shardIndex, array $server ) {
959  if ( !isset( $this->conns[$shardIndex] ) ) {
961  $conn = Database::factory( $server['type'], array_merge(
962  $server,
963  [
964  'flags' => ( $server['flags'] ?? 0 ) & ~IDatabase::DBO_TRX,
965  'connLogger' => $this->logger,
966  'queryLogger' => $this->logger
967  ]
968  ) );
969  // Automatically create the objectcache table for sqlite as needed
970  if ( $conn->getType() === 'sqlite' && !$conn->tableExists( 'objectcache' ) ) {
971  $this->initSqliteDatabase( $conn );
972  }
973  $this->conns[$shardIndex] = $conn;
974  }
975 
976  return $this->conns[$shardIndex];
977  }
978 
985  private function handleReadError( DBError $exception, $shardIndex ) {
986  if ( $exception instanceof DBConnectionError ) {
987  $this->markServerDown( $exception, $shardIndex );
988  }
989 
990  $this->setAndLogDBError( $exception );
991  }
992 
1001  private function handleWriteError( DBError $exception, $db, $shardIndex ) {
1002  if ( !( $db instanceof IDatabase ) ) {
1003  $this->markServerDown( $exception, $shardIndex );
1004  }
1005 
1006  $this->setAndLogDBError( $exception );
1007  }
1008 
1012  private function setAndLogDBError( DBError $exception ) {
1013  $this->logger->error( "DBError: {$exception->getMessage()}" );
1014  if ( $exception instanceof DBConnectionError ) {
1015  $this->setLastError( BagOStuff::ERR_UNREACHABLE );
1016  $this->logger->debug( __METHOD__ . ": ignoring connection error" );
1017  } else {
1018  $this->setLastError( BagOStuff::ERR_UNEXPECTED );
1019  $this->logger->debug( __METHOD__ . ": ignoring query error" );
1020  }
1021  }
1022 
1029  private function markServerDown( DBError $exception, $shardIndex ) {
1030  unset( $this->conns[$shardIndex] ); // bug T103435
1031 
1032  $now = $this->getCurrentTime();
1033  if ( isset( $this->connFailureTimes[$shardIndex] ) ) {
1034  if ( $now - $this->connFailureTimes[$shardIndex] >= 60 ) {
1035  unset( $this->connFailureTimes[$shardIndex] );
1036  unset( $this->connFailureErrors[$shardIndex] );
1037  } else {
1038  $this->logger->debug( __METHOD__ . ": Server #$shardIndex already down" );
1039  return;
1040  }
1041  }
1042  $this->logger->info( __METHOD__ . ": Server #$shardIndex down until " . ( $now + 60 ) );
1043  $this->connFailureTimes[$shardIndex] = $now;
1044  $this->connFailureErrors[$shardIndex] = $exception;
1045  }
1046 
1051  private function initSqliteDatabase( IMaintainableDatabase $db ) {
1052  if ( $db->tableExists( 'objectcache' ) ) {
1053  return;
1054  }
1055  // Use one table for SQLite; sharding does not seem to have much benefit
1056  $db->query( "PRAGMA journal_mode=WAL" ); // this is permanent
1057  $db->startAtomic( __METHOD__ ); // atomic DDL
1058  try {
1059  $encTable = $db->tableName( 'objectcache' );
1060  $encExptimeIndex = $db->addIdentifierQuotes( $db->tablePrefix() . 'exptime' );
1061  $db->query(
1062  "CREATE TABLE $encTable (\n" .
1063  " keyname BLOB NOT NULL default '' PRIMARY KEY,\n" .
1064  " value BLOB,\n" .
1065  " exptime TEXT\n" .
1066  ")",
1067  __METHOD__
1068  );
1069  $db->query( "CREATE INDEX $encExptimeIndex ON $encTable (exptime)" );
1070  $db->endAtomic( __METHOD__ );
1071  } catch ( DBError $e ) {
1072  $db->rollback( __METHOD__ );
1073  throw $e;
1074  }
1075  }
1076 
1080  public function createTables() {
1081  foreach ( $this->getServerShardIndexes() as $shardIndex ) {
1082  $db = $this->getConnection( $shardIndex );
1083  if ( in_array( $db->getType(), [ 'mysql', 'postgres' ], true ) ) {
1084  for ( $i = 0; $i < $this->numTableShards; $i++ ) {
1085  $encBaseTable = $db->tableName( 'objectcache' );
1086  $encShardTable = $db->tableName( $this->getTableNameByShard( $i ) );
1087  $db->query( "CREATE TABLE $encShardTable LIKE $encBaseTable" );
1088  }
1089  }
1090  }
1091  }
1092 
1096  private function getServerShardIndexes() {
1097  if ( $this->serverTags ) {
1098  // Striped array of database servers
1099  $shardIndexes = range( 0, $this->numServerShards - 1 );
1100  } else {
1101  // LoadBalancer based configuration
1102  $shardIndexes = [];
1103  if ( $this->localKeyLb ) {
1104  $shardIndexes[] = self::SHARD_LOCAL;
1105  }
1106  if ( $this->globalKeyLb ) {
1107  $shardIndexes[] = self::SHARD_GLOBAL;
1108  }
1109  }
1110 
1111  return $shardIndexes;
1112  }
1113 
1120  private function waitForReplication( $shardIndex ) {
1121  if ( is_int( $shardIndex ) ) {
1122  return true; // striped only, no LoadBalancer
1123  }
1124 
1125  $lb = ( $shardIndex === self::SHARD_LOCAL ) ? $this->localKeyLb : $this->globalKeyLb;
1126  if ( !$lb->hasStreamingReplicaServers() ) {
1127  return true;
1128  }
1129 
1130  try {
1131  // Wait for any replica DBs to catch up
1132  $masterPos = $lb->getMasterPos();
1133  if ( !$masterPos ) {
1134  return true; // not applicable
1135  }
1136 
1137  $loop = new WaitConditionLoop(
1138  function () use ( $lb, $masterPos ) {
1139  return $lb->waitForAll( $masterPos, 1 );
1140  },
1143  );
1144 
1145  return ( $loop->invoke() === $loop::CONDITION_REACHED );
1146  } catch ( DBError $e ) {
1147  $this->setAndLogDBError( $e );
1148 
1149  return false;
1150  }
1151  }
1152 
1158  private function silenceTransactionProfiler() {
1159  if ( $this->serverInfos ) {
1160  return null; // no TransactionProfiler injected anyway
1161  }
1162 
1163  $trxProfiler = Profiler::instance()->getTransactionProfiler();
1164  $oldSilenced = $trxProfiler->setSilenced( true );
1165  return new ScopedCallback( function () use ( $trxProfiler, $oldSilenced ) {
1166  $trxProfiler->setSilenced( $oldSilenced );
1167  } );
1168  }
1169 }
SqlBagOStuff\$purgeLimit
int $purgeLimit
Definition: SqlBagOStuff.php:59
SqlBagOStuff\__construct
__construct( $params)
Constructor.
Definition: SqlBagOStuff.php:127
MediumSpecificBagOStuff\setLastError
setLastError( $err)
Set the "last error" registry.
Definition: MediumSpecificBagOStuff.php:755
SqlBagOStuff\doGetMulti
doGetMulti(array $keys, $flags=0)
Get an associative array containing the item for each of the keys that have items.
Definition: SqlBagOStuff.php:278
SqlBagOStuff\doCas
doCas( $casToken, $key, $value, $exptime=0, $flags=0)
Check and set an item.
Definition: SqlBagOStuff.php:485
SqlBagOStuff\createTables
createTables()
Create the shard tables on all databases (e.g.
Definition: SqlBagOStuff.php:1080
Wikimedia\Rdbms\Database
Relational database abstraction object.
Definition: Database.php:49
MediumSpecificBagOStuff\isInteger
isInteger( $value)
Check if a value is an integer.
Definition: MediumSpecificBagOStuff.php:878
SqlBagOStuff\changeTTLMulti
changeTTLMulti(array $keys, $exptime, $flags=0)
Change the expiration of multiple keys that exist.
Definition: SqlBagOStuff.php:577
ArrayUtils\consistentHashSort
static consistentHashSort(&$array, $key, $separator="\000")
Sort the given array in a pseudo-random order which depends only on the given key and each element va...
Definition: ArrayUtils.php:49
Wikimedia\Rdbms\IDatabase\affectedRows
affectedRows()
Get the number of rows affected by the last write query.
SqlBagOStuff\doGet
doGet( $key, $flags=0, &$casToken=null)
Definition: SqlBagOStuff.php:262
SqlBagOStuff\waitForReplication
waitForReplication( $shardIndex)
Wait for replica DBs to catch up to the master DB.
Definition: SqlBagOStuff.php:1120
Wikimedia\Rdbms\IDatabase\tablePrefix
tablePrefix( $prefix=null)
Get/set the table prefix.
MediumSpecificBagOStuff\$keyspace
string $keyspace
Definition: MediumSpecificBagOStuff.php:40
SqlBagOStuff\setAndLogDBError
setAndLogDBError(DBError $exception)
Definition: SqlBagOStuff.php:1012
MediumSpecificBagOStuff\debug
debug( $text)
Definition: MediumSpecificBagOStuff.php:1061
Profiler\instance
static instance()
Singleton.
Definition: Profiler.php:63
Wikimedia\Rdbms\IDatabase\rollback
rollback( $fname=__METHOD__, $flush=self::FLUSHING_ONE)
Rollback a transaction previously started using begin()
SqlBagOStuff\fetchBlobMulti
fetchBlobMulti(array $keys)
Definition: SqlBagOStuff.php:289
SqlBagOStuff\$numServerShards
int $numServerShards
Number of database servers shards (e.g.
Definition: SqlBagOStuff.php:53
SqlBagOStuff\$serverInfos
array[] $serverInfos
(server index => server config)
Definition: SqlBagOStuff.php:49
SqlBagOStuff\getConnectionViaLoadBalancer
getConnectionViaLoadBalancer( $shardIndex)
Definition: SqlBagOStuff.php:934
SqlBagOStuff\$numTableShards
int $numTableShards
Definition: SqlBagOStuff.php:61
Wikimedia\Rdbms\IDatabase\endAtomic
endAtomic( $fname=__METHOD__)
Ends an atomic section of SQL statements.
SqlBagOStuff\initSqliteDatabase
initSqliteDatabase(IMaintainableDatabase $db)
Definition: SqlBagOStuff.php:1051
SqlBagOStuff\silenceTransactionProfiler
silenceTransactionProfiler()
Silence the transaction profiler until the return value falls out of scope.
Definition: SqlBagOStuff.php:1158
SqlBagOStuff\incr
incr( $key, $step=1, $flags=0)
Increase stored value of $key by $value while preserving its TTL.
Definition: SqlBagOStuff.php:539
Wikimedia\Rdbms\IMaintainableDatabase\tableName
tableName( $name, $format='quoted')
Format a table name ready for use in constructing an SQL query.
SqlBagOStuff\$OP_TOUCH
static string $OP_TOUCH
Definition: SqlBagOStuff.php:82
SqlBagOStuff\$connFailureTimes
float[] $connFailureTimes
Map of (shard index => UNIX timestamps)
Definition: SqlBagOStuff.php:70
SqlBagOStuff\updateTable
updateTable( $op, $db, $table, $tableKeys, $data, $dbExpiry)
Definition: SqlBagOStuff.php:423
$success
$success
Definition: NoLocalSettings.php:42
SqlBagOStuff\$localKeyLb
ILoadBalancer null $localKeyLb
Definition: SqlBagOStuff.php:44
$res
$res
Definition: testCompression.php:57
Wikimedia\Rdbms\DBError
Database error base class.
Definition: DBError.php:30
DBO_TRX
const DBO_TRX
Definition: defines.php:12
Wikimedia\Rdbms\IDatabase\tableExists
tableExists( $table, $fname=__METHOD__)
Query whether a given table exists.
Wikimedia\Rdbms\IDatabase
Basic database interface for live and lazy-loaded relation database handles.
Definition: IDatabase.php:38
SqlBagOStuff\doSet
doSet( $key, $value, $exptime=0, $flags=0)
Set an item.
Definition: SqlBagOStuff.php:477
SqlBagOStuff\$GC_DELAY_SEC
static int $GC_DELAY_SEC
Definition: SqlBagOStuff.php:75
MediumSpecificBagOStuff\$syncTimeout
int $syncTimeout
Seconds.
Definition: MediumSpecificBagOStuff.php:42
SqlBagOStuff\$lastGarbageCollect
int $lastGarbageCollect
UNIX timestamp.
Definition: SqlBagOStuff.php:55
SqlBagOStuff\getKeyLocation
getKeyLocation( $key)
Get the server index and table name for a given key.
Definition: SqlBagOStuff.php:219
Wikimedia\Rdbms\IDatabase\timestamp
timestamp( $ts=0)
Convert a timestamp in one of the formats accepted by ConvertibleTimestamp to the format used for ins...
SqlBagOStuff\expireAll
expireAll()
Definition: SqlBagOStuff.php:644
SqlBagOStuff\occasionallyGarbageCollect
occasionallyGarbageCollect(IDatabase $db)
Definition: SqlBagOStuff.php:618
SqlBagOStuff\unlock
unlock( $key)
Release an advisory lock on a key string.
Definition: SqlBagOStuff.php:820
SqlBagOStuff\deleteAll
deleteAll()
Delete content of shard tables in every server.
Definition: SqlBagOStuff.php:767
MediumSpecificBagOStuff\getExpirationAsTimestamp
getExpirationAsTimestamp( $exptime)
Convert an optionally relative timestamp to an absolute time.
Definition: MediumSpecificBagOStuff.php:838
SqlBagOStuff\$connFailureErrors
Exception[] $connFailureErrors
Map of (shard index => Exception)
Definition: SqlBagOStuff.php:72
SqlBagOStuff\doDeleteMulti
doDeleteMulti(array $keys, $flags=0)
Definition: SqlBagOStuff.php:526
$blob
$blob
Definition: testCompression.php:70
SqlBagOStuff\modifyMulti
modifyMulti(array $data, $exptime, $flags, $op)
Definition: SqlBagOStuff.php:361
BagOStuff\$asyncHandler
callable null $asyncHandler
Definition: BagOStuff.php:75
SqlBagOStuff\getConnectionFromServerInfo
getConnectionFromServerInfo( $shardIndex, array $server)
Definition: SqlBagOStuff.php:958
$args
if( $line===false) $args
Definition: mcc.php:124
SqlBagOStuff\isExpired
isExpired(IDatabase $db, $exptime)
Definition: SqlBagOStuff.php:595
SqlBagOStuff\$tableName
string $tableName
Definition: SqlBagOStuff.php:63
MediumSpecificBagOStuff
Storage medium specific cache for storing items (e.g.
Definition: MediumSpecificBagOStuff.php:34
Wikimedia\Rdbms\IDatabase\query
query( $sql, $fname=__METHOD__, $flags=0)
Run an SQL query and return the result.
DB_REPLICA
const DB_REPLICA
Definition: defines.php:25
SqlBagOStuff\handleReadError
handleReadError(DBError $exception, $shardIndex)
Handle a DBError which occurred during a read operation.
Definition: SqlBagOStuff.php:985
DB_MASTER
const DB_MASTER
Definition: defines.php:26
SqlBagOStuff\serialize
serialize( $data)
Serialize an object and, if possible, compress the representation.
Definition: SqlBagOStuff.php:893
SqlBagOStuff\handleWriteError
handleWriteError(DBError $exception, $db, $shardIndex)
Handle a DBQueryError which occurred during a write operation.
Definition: SqlBagOStuff.php:1001
SqlBagOStuff\$OP_SET
static string $OP_SET
Definition: SqlBagOStuff.php:78
Wikimedia\Rdbms\DBQueryError
Definition: DBQueryError.php:27
SqlBagOStuff\$globalKeyLb
ILoadBalancer null $globalKeyLb
Definition: SqlBagOStuff.php:46
SqlBagOStuff\getServerShardIndexes
getServerShardIndexes()
Definition: SqlBagOStuff.php:1096
SqlBagOStuff\doChangeTTL
doChangeTTL( $key, $exptime, $flags)
Definition: SqlBagOStuff.php:586
SqlBagOStuff\unserialize
unserialize( $serial)
Unserialize and, if necessary, decompress an object.
Definition: SqlBagOStuff.php:911
SqlBagOStuff\lock
lock( $key, $timeout=6, $expiry=6, $rclass='')
Acquire an advisory lock on a key string.
Definition: SqlBagOStuff.php:785
SqlBagOStuff\makeKeyInternal
makeKeyInternal( $keyspace, $args)
Construct a cache key.
Definition: SqlBagOStuff.php:859
SqlBagOStuff\getConnection
getConnection( $shardIndex)
Get a connection to the specified database.
Definition: SqlBagOStuff.php:187
MediumSpecificBagOStuff\$busyCallbacks
callable[] $busyCallbacks
Definition: MediumSpecificBagOStuff.php:56
SqlBagOStuff\markServerDown
markServerDown(DBError $exception, $shardIndex)
Mark a server down due to a DBConnectionError exception.
Definition: SqlBagOStuff.php:1029
SqlBagOStuff
Class to store objects in the database.
Definition: SqlBagOStuff.php:42
SqlBagOStuff\deleteServerObjectsExpiringBefore
deleteServerObjectsExpiringBefore(IDatabase $db, $timestamp, $progressCallback, $limit, $serversDoneCount=0, &$keysDeletedCount=0)
Definition: SqlBagOStuff.php:692
SqlBagOStuff\$conns
IMaintainableDatabase[] $conns
Map of (shard index => DB handle)
Definition: SqlBagOStuff.php:68
Wikimedia\Rdbms\IDatabase\addQuotes
addQuotes( $s)
Escape and quote a raw value string for use in a SQL query.
SqlBagOStuff\decr
decr( $key, $value=1, $flags=0)
Decrease stored value of $key by $value while preserving its TTL.
Definition: SqlBagOStuff.php:573
SqlBagOStuff\doAdd
doAdd( $key, $value, $exptime=0, $flags=0)
Insert an item if it does not already exist.
Definition: SqlBagOStuff.php:481
SqlBagOStuff\$OP_ADD
static string $OP_ADD
Definition: SqlBagOStuff.php:80
BagOStuff\fieldHasFlags
fieldHasFlags( $field, $flags)
Definition: BagOStuff.php:508
SqlBagOStuff\getMaxDateTime
getMaxDateTime( $db)
Definition: SqlBagOStuff.php:606
Wikimedia\Rdbms\DBConnectionError
Definition: DBConnectionError.php:26
Wikimedia\Rdbms\IDatabase\getType
getType()
Get the type of the DBMS (e.g.
$keys
$keys
Definition: testCompression.php:72
Wikimedia\Rdbms\IDatabase\select
select( $table, $vars, $conds='', $fname=__METHOD__, $options=[], $join_conds=[])
Execute a SELECT query constructed using the various parameters provided.
BagOStuff\getCurrentTime
getCurrentTime()
Definition: BagOStuff.php:565
SqlBagOStuff\$serverTags
string[] $serverTags
(server index => tag/host name)
Definition: SqlBagOStuff.php:51
SqlBagOStuff\$replicaOnly
bool $replicaOnly
Definition: SqlBagOStuff.php:65
Wikimedia\Rdbms\IDatabase\addIdentifierQuotes
addIdentifierQuotes( $s)
Escape a SQL identifier (e.g.
SqlBagOStuff\$OP_DELETE
static string $OP_DELETE
Definition: SqlBagOStuff.php:84
SqlBagOStuff\doSetMulti
doSetMulti(array $data, $exptime=0, $flags=0)
Definition: SqlBagOStuff.php:350
Wikimedia\Rdbms\IMaintainableDatabase
Advanced database interface for IDatabase handles that include maintenance methods.
Definition: IMaintainableDatabase.php:38
SqlBagOStuff\$purgePeriod
int $purgePeriod
Definition: SqlBagOStuff.php:57
SqlBagOStuff\deleteObjectsExpiringBefore
deleteObjectsExpiringBefore( $timestamp, callable $progress=null, $limit=INF)
Delete all objects expiring before a certain date.
Definition: SqlBagOStuff.php:648
Wikimedia\Rdbms\IDatabase\delete
delete( $table, $conds, $fname=__METHOD__)
Delete all rows in a table that match a condition.
Wikimedia\Rdbms\ILoadBalancer
Database cluster connection, tracking, load balancing, and transaction manager interface.
Definition: ILoadBalancer.php:81
SqlBagOStuff\getTableNameByShard
getTableNameByShard( $index)
Get the table name for a given shard index.
Definition: SqlBagOStuff.php:252
SqlBagOStuff\doDelete
doDelete( $key, $flags=0)
Delete an item.
Definition: SqlBagOStuff.php:535
Wikimedia\Rdbms\IDatabase\startAtomic
startAtomic( $fname=__METHOD__, $cancelable=self::ATOMIC_NOT_CANCELABLE)
Begin an atomic section of SQL statements.