MediaWiki  master
SqlBagOStuff.php
Go to the documentation of this file.
1 <?php
36 
44  protected $serverInfos;
46  protected $serverTags;
48  protected $numServerShards;
50  protected $lastGarbageCollect = 0;
52  protected $purgePeriod = 10;
54  protected $purgeLimit = 100;
56  protected $numTableShards = 1;
58  protected $tableName = 'objectcache';
60  protected $replicaOnly = false;
61 
63  protected $separateMainLB;
65  protected $conns;
67  protected $connFailureTimes = [];
69  protected $connFailureErrors = [];
70 
72  private static $GC_DELAY_SEC = 1;
73 
75  private static $OP_SET = 'set';
77  private static $OP_ADD = 'add';
79  private static $OP_TOUCH = 'touch';
81  private static $OP_DELETE = 'delete';
82 
122  public function __construct( $params ) {
123  parent::__construct( $params );
124 
125  $this->attrMap[self::ATTR_EMULATION] = self::QOS_EMULATION_SQL;
126  $this->attrMap[self::ATTR_SYNCWRITES] = self::QOS_SYNCWRITES_NONE;
127 
128  if ( isset( $params['servers'] ) ) {
129  $this->serverInfos = [];
130  $this->serverTags = [];
131  $this->numServerShards = count( $params['servers'] );
132  $index = 0;
133  foreach ( $params['servers'] as $tag => $info ) {
134  $this->serverInfos[$index] = $info;
135  if ( is_string( $tag ) ) {
136  $this->serverTags[$index] = $tag;
137  } else {
138  $this->serverTags[$index] = $info['host'] ?? "#$index";
139  }
140  ++$index;
141  }
142  } elseif ( isset( $params['server'] ) ) {
143  $this->serverInfos = [ $params['server'] ];
144  $this->numServerShards = count( $this->serverInfos );
145  } else {
146  // Default to using the main wiki's database servers
147  $this->serverInfos = [];
148  $this->numServerShards = 1;
149  $this->attrMap[self::ATTR_SYNCWRITES] = self::QOS_SYNCWRITES_BE;
150  }
151  if ( isset( $params['purgePeriod'] ) ) {
152  $this->purgePeriod = intval( $params['purgePeriod'] );
153  }
154  if ( isset( $params['purgeLimit'] ) ) {
155  $this->purgeLimit = intval( $params['purgeLimit'] );
156  }
157  if ( isset( $params['tableName'] ) ) {
158  $this->tableName = $params['tableName'];
159  }
160  if ( isset( $params['shards'] ) ) {
161  $this->numTableShards = intval( $params['shards'] );
162  }
163  // Backwards-compatibility for < 1.34
164  $this->replicaOnly = $params['replicaOnly'] ?? ( $params['slaveOnly'] ?? false );
165  }
166 
174  private function getConnection( $shardIndex ) {
175  if ( $shardIndex >= $this->numServerShards ) {
176  throw new MWException( __METHOD__ . ": Invalid server index \"$shardIndex\"" );
177  }
178 
179  # Don't keep timing out trying to connect for each call if the DB is down
180  if (
181  isset( $this->connFailureErrors[$shardIndex] ) &&
182  ( $this->getCurrentTime() - $this->connFailureTimes[$shardIndex] ) < 60
183  ) {
184  throw $this->connFailureErrors[$shardIndex];
185  }
186 
187  if ( $this->serverInfos ) {
188  if ( !isset( $this->conns[$shardIndex] ) ) {
189  // Use custom database defined by server connection info
190  $info = $this->serverInfos[$shardIndex];
191  $type = $info['type'] ?? 'mysql';
192  $host = $info['host'] ?? '[unknown]';
193  $this->logger->debug( __CLASS__ . ": connecting to $host" );
194  $conn = Database::factory( $type, $info );
195  $conn->clearFlag( DBO_TRX ); // auto-commit mode
196  $this->conns[$shardIndex] = $conn;
197  // Automatically create the objectcache table for sqlite as needed
198  if ( $conn->getType() === 'sqlite' ) {
199  $this->initSqliteDatabase( $conn );
200  }
201  }
202  $conn = $this->conns[$shardIndex];
203  } else {
204  // Use the main LB database
205  $lb = MediaWikiServices::getInstance()->getDBLoadBalancer();
206  $index = $this->replicaOnly ? DB_REPLICA : DB_MASTER;
207  // If the RDBMS has row-level locking, use the autocommit connection to avoid
208  // contention and deadlocks. Do not do this if it only has DB-level locking since
209  // that would just cause deadlocks.
210  $attribs = $lb->getServerAttributes( $lb->getWriterIndex() );
211  $flags = $attribs[Database::ATTR_DB_LEVEL_LOCKING] ? 0 : $lb::CONN_TRX_AUTOCOMMIT;
212  $conn = $lb->getMaintenanceConnectionRef( $index, [], false, $flags );
213  }
214 
215  $this->logger->debug( sprintf( "Connection %s will be used for SqlBagOStuff", $conn ) );
216 
217  return $conn;
218  }
219 
225  private function getTableByKey( $key ) {
226  if ( $this->numTableShards > 1 ) {
227  $hash = hexdec( substr( md5( $key ), 0, 8 ) ) & 0x7fffffff;
228  $tableIndex = $hash % $this->numTableShards;
229  } else {
230  $tableIndex = 0;
231  }
232  if ( $this->numServerShards > 1 ) {
233  $sortedServers = $this->serverTags;
234  ArrayUtils::consistentHashSort( $sortedServers, $key );
235  reset( $sortedServers );
236  $shardIndex = key( $sortedServers );
237  } else {
238  $shardIndex = 0;
239  }
240  return [ $shardIndex, $this->getTableNameByShard( $tableIndex ) ];
241  }
242 
248  private function getTableNameByShard( $index ) {
249  if ( $this->numTableShards > 1 ) {
250  $decimals = strlen( $this->numTableShards - 1 );
251  return $this->tableName .
252  sprintf( "%0{$decimals}d", $index );
253  } else {
254  return $this->tableName;
255  }
256  }
257 
258  protected function doGet( $key, $flags = 0, &$casToken = null ) {
259  $casToken = null;
260 
261  $blobs = $this->fetchBlobMulti( [ $key ] );
262  if ( array_key_exists( $key, $blobs ) ) {
263  $blob = $blobs[$key];
264  $value = $this->unserialize( $blob );
265 
266  $casToken = ( $value !== false ) ? $blob : null;
267 
268  return $value;
269  }
270 
271  return false;
272  }
273 
274  protected function doGetMulti( array $keys, $flags = 0 ) {
275  $values = [];
276 
277  $blobs = $this->fetchBlobMulti( $keys );
278  foreach ( $blobs as $key => $blob ) {
279  $values[$key] = $this->unserialize( $blob );
280  }
281 
282  return $values;
283  }
284 
285  private function fetchBlobMulti( array $keys, $flags = 0 ) {
286  $values = []; // array of (key => value)
287 
288  $keysByTable = [];
289  foreach ( $keys as $key ) {
290  list( $shardIndex, $tableName ) = $this->getTableByKey( $key );
291  $keysByTable[$shardIndex][$tableName][] = $key;
292  }
293 
294  $dataRows = [];
295  foreach ( $keysByTable as $shardIndex => $serverKeys ) {
296  try {
297  $db = $this->getConnection( $shardIndex );
298  foreach ( $serverKeys as $tableName => $tableKeys ) {
299  $res = $db->select( $tableName,
300  [ 'keyname', 'value', 'exptime' ],
301  [ 'keyname' => $tableKeys ],
302  __METHOD__,
303  // Approximate write-on-the-fly BagOStuff API via blocking.
304  // This approximation fails if a ROLLBACK happens (which is rare).
305  // We do not want to flush the TRX as that can break callers.
306  $db->trxLevel() ? [ 'LOCK IN SHARE MODE' ] : []
307  );
308  if ( $res === false ) {
309  continue;
310  }
311  foreach ( $res as $row ) {
312  $row->shardIndex = $shardIndex;
313  $row->tableName = $tableName;
314  $dataRows[$row->keyname] = $row;
315  }
316  }
317  } catch ( DBError $e ) {
318  $this->handleReadError( $e, $shardIndex );
319  }
320  }
321 
322  foreach ( $keys as $key ) {
323  if ( isset( $dataRows[$key] ) ) { // HIT?
324  $row = $dataRows[$key];
325  $this->debug( "get: retrieved data; expiry time is " . $row->exptime );
326  $db = null; // in case of connection failure
327  try {
328  $db = $this->getConnection( $row->shardIndex );
329  if ( $this->isExpired( $db, $row->exptime ) ) { // MISS
330  $this->debug( "get: key has expired" );
331  } else { // HIT
332  $values[$key] = $db->decodeBlob( $row->value );
333  }
334  } catch ( DBQueryError $e ) {
335  $this->handleWriteError( $e, $db, $row->shardIndex );
336  }
337  } else { // MISS
338  $this->debug( 'get: no matching rows' );
339  }
340  }
341 
342  return $values;
343  }
344 
345  protected function doSetMulti( array $data, $exptime = 0, $flags = 0 ) {
346  return $this->modifyMulti( $data, $exptime, $flags, self::$OP_SET );
347  }
348 
356  private function modifyMulti( array $data, $exptime, $flags, $op ) {
357  $keysByTable = [];
358  foreach ( $data as $key => $value ) {
359  list( $shardIndex, $tableName ) = $this->getTableByKey( $key );
360  $keysByTable[$shardIndex][$tableName][] = $key;
361  }
362 
363  $exptime = $this->getExpirationAsTimestamp( $exptime );
364 
365  $result = true;
367  $silenceScope = $this->silenceTransactionProfiler();
368  foreach ( $keysByTable as $shardIndex => $serverKeys ) {
369  $db = null; // in case of connection failure
370  try {
371  $db = $this->getConnection( $shardIndex );
372  $this->occasionallyGarbageCollect( $db ); // expire old entries if any
373  $dbExpiry = $exptime ? $db->timestamp( $exptime ) : $this->getMaxDateTime( $db );
374  } catch ( DBError $e ) {
375  $this->handleWriteError( $e, $db, $shardIndex );
376  $result = false;
377  continue;
378  }
379 
380  foreach ( $serverKeys as $tableName => $tableKeys ) {
381  try {
382  $result = $this->updateTableKeys(
383  $op,
384  $db,
385  $tableName,
386  $tableKeys,
387  $data,
388  $dbExpiry
389  ) && $result;
390  } catch ( DBError $e ) {
391  $this->handleWriteError( $e, $db, $shardIndex );
392  $result = false;
393  }
394 
395  }
396  }
397 
398  if ( $this->fieldHasFlags( $flags, self::WRITE_SYNC ) ) {
399  $result = $this->waitForReplication() && $result;
400  }
401 
402  return $result;
403  }
404 
416  private function updateTableKeys( $op, $db, $table, $tableKeys, $data, $dbExpiry ) {
417  $success = true;
418 
419  if ( $op === self::$OP_ADD ) {
420  $rows = [];
421  foreach ( $tableKeys as $key ) {
422  $rows[] = [
423  'keyname' => $key,
424  'value' => $db->encodeBlob( $this->serialize( $data[$key] ) ),
425  'exptime' => $dbExpiry
426  ];
427  }
428  $db->delete(
429  $table,
430  [
431  'keyname' => $tableKeys,
432  'exptime <= ' . $db->addQuotes( $db->timestamp() )
433  ],
434  __METHOD__
435  );
436  $db->insert( $table, $rows, __METHOD__, [ 'IGNORE' ] );
437 
438  $success = ( $db->affectedRows() == count( $rows ) );
439  } elseif ( $op === self::$OP_SET ) {
440  $rows = [];
441  foreach ( $tableKeys as $key ) {
442  $rows[] = [
443  'keyname' => $key,
444  'value' => $db->encodeBlob( $this->serialize( $data[$key] ) ),
445  'exptime' => $dbExpiry
446  ];
447  }
448  $db->replace( $table, [ 'keyname' ], $rows, __METHOD__ );
449  } elseif ( $op === self::$OP_DELETE ) {
450  $db->delete( $table, [ 'keyname' => $tableKeys ], __METHOD__ );
451  } elseif ( $op === self::$OP_TOUCH ) {
452  $db->update(
453  $table,
454  [ 'exptime' => $dbExpiry ],
455  [
456  'keyname' => $tableKeys,
457  'exptime > ' . $db->addQuotes( $db->timestamp() )
458  ],
459  __METHOD__
460  );
461 
462  $success = ( $db->affectedRows() == count( $tableKeys ) );
463  } else {
464  throw new InvalidArgumentException( "Invalid operation '$op'" );
465  }
466 
467  return $success;
468  }
469 
470  protected function doSet( $key, $value, $exptime = 0, $flags = 0 ) {
471  return $this->modifyMulti( [ $key => $value ], $exptime, $flags, self::$OP_SET );
472  }
473 
474  protected function doAdd( $key, $value, $exptime = 0, $flags = 0 ) {
475  return $this->modifyMulti( [ $key => $value ], $exptime, $flags, self::$OP_ADD );
476  }
477 
478  protected function doCas( $casToken, $key, $value, $exptime = 0, $flags = 0 ) {
479  list( $shardIndex, $tableName ) = $this->getTableByKey( $key );
480  $exptime = $this->getExpirationAsTimestamp( $exptime );
481 
483  $silenceScope = $this->silenceTransactionProfiler();
484  $db = null; // in case of connection failure
485  try {
486  $db = $this->getConnection( $shardIndex );
487  // (T26425) use a replace if the db supports it instead of
488  // delete/insert to avoid clashes with conflicting keynames
489  $db->update(
490  $tableName,
491  [
492  'keyname' => $key,
493  'value' => $db->encodeBlob( $this->serialize( $value ) ),
494  'exptime' => $exptime
495  ? $db->timestamp( $exptime )
496  : $this->getMaxDateTime( $db )
497  ],
498  [
499  'keyname' => $key,
500  'value' => $db->encodeBlob( $casToken ),
501  'exptime > ' . $db->addQuotes( $db->timestamp() )
502  ],
503  __METHOD__
504  );
505  } catch ( DBQueryError $e ) {
506  $this->handleWriteError( $e, $db, $shardIndex );
507 
508  return false;
509  }
510 
511  $success = (bool)$db->affectedRows();
512  if ( $this->fieldHasFlags( $flags, self::WRITE_SYNC ) ) {
513  $success = $this->waitForReplication() && $success;
514  }
515 
516  return $success;
517  }
518 
519  protected function doDeleteMulti( array $keys, $flags = 0 ) {
520  return $this->modifyMulti(
521  array_fill_keys( $keys, null ),
522  0,
523  $flags,
524  self::$OP_DELETE
525  );
526  }
527 
528  protected function doDelete( $key, $flags = 0 ) {
529  return $this->modifyMulti( [ $key => null ], 0, $flags, self::$OP_DELETE );
530  }
531 
532  public function incr( $key, $step = 1, $flags = 0 ) {
533  list( $shardIndex, $tableName ) = $this->getTableByKey( $key );
534 
535  $newCount = false;
537  $silenceScope = $this->silenceTransactionProfiler();
538  $db = null; // in case of connection failure
539  try {
540  $db = $this->getConnection( $shardIndex );
541  $encTimestamp = $db->addQuotes( $db->timestamp() );
542  $db->update(
543  $tableName,
544  [ 'value = value + ' . (int)$step ],
545  [ 'keyname' => $key, "exptime > $encTimestamp" ],
546  __METHOD__
547  );
548  if ( $db->affectedRows() > 0 ) {
549  $newValue = $db->selectField(
550  $tableName,
551  'value',
552  [ 'keyname' => $key, "exptime > $encTimestamp" ],
553  __METHOD__
554  );
555  if ( $this->isInteger( $newValue ) ) {
556  $newCount = (int)$newValue;
557  }
558  }
559  } catch ( DBError $e ) {
560  $this->handleWriteError( $e, $db, $shardIndex );
561  }
562 
563  return $newCount;
564  }
565 
566  public function decr( $key, $value = 1, $flags = 0 ) {
567  return $this->incr( $key, -$value, $flags );
568  }
569 
570  public function changeTTLMulti( array $keys, $exptime, $flags = 0 ) {
571  return $this->modifyMulti(
572  array_fill_keys( $keys, null ),
573  $exptime,
574  $flags,
575  self::$OP_TOUCH
576  );
577  }
578 
579  protected function doChangeTTL( $key, $exptime, $flags ) {
580  return $this->modifyMulti( [ $key => null ], $exptime, $flags, self::$OP_TOUCH );
581  }
582 
588  private function isExpired( IDatabase $db, $exptime ) {
589  return (
590  $exptime != $this->getMaxDateTime( $db ) &&
591  ConvertibleTimestamp::convert( TS_UNIX, $exptime ) < $this->getCurrentTime()
592  );
593  }
594 
599  private function getMaxDateTime( $db ) {
600  if ( (int)$this->getCurrentTime() > 0x7fffffff ) {
601  return $db->timestamp( 1 << 62 );
602  } else {
603  return $db->timestamp( 0x7fffffff );
604  }
605  }
606 
611  private function occasionallyGarbageCollect( IDatabase $db ) {
612  if (
613  // Random purging is enabled
614  $this->purgePeriod &&
615  // Only purge on one in every $this->purgePeriod writes
616  mt_rand( 0, $this->purgePeriod - 1 ) == 0 &&
617  // Avoid repeating the delete within a few seconds
618  ( $this->getCurrentTime() - $this->lastGarbageCollect ) > self::$GC_DELAY_SEC
619  ) {
620  $garbageCollector = function () use ( $db ) {
622  $db, $this->getCurrentTime(),
623  null,
624  $this->purgeLimit
625  );
626  $this->lastGarbageCollect = time();
627  };
628  if ( $this->asyncHandler ) {
629  $this->lastGarbageCollect = $this->getCurrentTime(); // avoid duplicate enqueues
630  ( $this->asyncHandler )( $garbageCollector );
631  } else {
632  $garbageCollector();
633  }
634  }
635  }
636 
637  public function expireAll() {
638  $this->deleteObjectsExpiringBefore( $this->getCurrentTime() );
639  }
640 
642  $timestamp,
643  callable $progress = null,
644  $limit = INF
645  ) {
647  $silenceScope = $this->silenceTransactionProfiler();
648 
649  $shardIndexes = range( 0, $this->numServerShards - 1 );
650  shuffle( $shardIndexes );
651 
652  $ok = true;
653 
654  $keysDeletedCount = 0;
655  foreach ( $shardIndexes as $numServersDone => $shardIndex ) {
656  $db = null; // in case of connection failure
657  try {
658  $db = $this->getConnection( $shardIndex );
660  $db,
661  $timestamp,
662  $progress,
663  $limit,
664  $numServersDone,
665  $keysDeletedCount
666  );
667  } catch ( DBError $e ) {
668  $this->handleWriteError( $e, $db, $shardIndex );
669  $ok = false;
670  }
671  }
672 
673  return $ok;
674  }
675 
686  IDatabase $db,
687  $timestamp,
688  $progressCallback,
689  $limit,
690  $serversDoneCount = 0,
691  &$keysDeletedCount = 0
692  ) {
693  $cutoffUnix = ConvertibleTimestamp::convert( TS_UNIX, $timestamp );
694  $shardIndexes = range( 0, $this->numTableShards - 1 );
695  shuffle( $shardIndexes );
696 
697  foreach ( $shardIndexes as $numShardsDone => $shardIndex ) {
698  $continue = null; // last exptime
699  $lag = null; // purge lag
700  do {
701  $res = $db->select(
702  $this->getTableNameByShard( $shardIndex ),
703  [ 'keyname', 'exptime' ],
704  array_merge(
705  [ 'exptime < ' . $db->addQuotes( $db->timestamp( $cutoffUnix ) ) ],
706  $continue ? [ 'exptime >= ' . $db->addQuotes( $continue ) ] : []
707  ),
708  __METHOD__,
709  [ 'LIMIT' => min( $limit, 100 ), 'ORDER BY' => 'exptime' ]
710  );
711 
712  if ( $res->numRows() ) {
713  $row = $res->current();
714  if ( $lag === null ) {
715  $rowExpUnix = ConvertibleTimestamp::convert( TS_UNIX, $row->exptime );
716  $lag = max( $cutoffUnix - $rowExpUnix, 1 );
717  }
718 
719  $keys = [];
720  foreach ( $res as $row ) {
721  $keys[] = $row->keyname;
722  $continue = $row->exptime;
723  }
724 
725  $db->delete(
726  $this->getTableNameByShard( $shardIndex ),
727  [
728  'exptime < ' . $db->addQuotes( $db->timestamp( $cutoffUnix ) ),
729  'keyname' => $keys
730  ],
731  __METHOD__
732  );
733  $keysDeletedCount += $db->affectedRows();
734  }
735 
736  if ( is_callable( $progressCallback ) ) {
737  if ( $lag ) {
738  $continueUnix = ConvertibleTimestamp::convert( TS_UNIX, $continue );
739  $remainingLag = $cutoffUnix - $continueUnix;
740  $processedLag = max( $lag - $remainingLag, 0 );
741  $doneRatio = ( $numShardsDone + $processedLag / $lag ) / $this->numTableShards;
742  } else {
743  $doneRatio = 1;
744  }
745 
746  $overallRatio = ( $doneRatio / $this->numServerShards )
747  + ( $serversDoneCount / $this->numServerShards );
748  call_user_func( $progressCallback, $overallRatio * 100 );
749  }
750  } while ( $res->numRows() && $keysDeletedCount < $limit );
751  }
752  }
753 
759  public function deleteAll() {
761  $silenceScope = $this->silenceTransactionProfiler();
762  for ( $shardIndex = 0; $shardIndex < $this->numServerShards; $shardIndex++ ) {
763  $db = null; // in case of connection failure
764  try {
765  $db = $this->getConnection( $shardIndex );
766  for ( $i = 0; $i < $this->numTableShards; $i++ ) {
767  $db->delete( $this->getTableNameByShard( $i ), '*', __METHOD__ );
768  }
769  } catch ( DBError $e ) {
770  $this->handleWriteError( $e, $db, $shardIndex );
771  return false;
772  }
773  }
774  return true;
775  }
776 
777  public function lock( $key, $timeout = 6, $expiry = 6, $rclass = '' ) {
778  // Avoid deadlocks and allow lock reentry if specified
779  if ( isset( $this->locks[$key] ) ) {
780  if ( $rclass != '' && $this->locks[$key]['class'] === $rclass ) {
781  ++$this->locks[$key]['depth'];
782  return true;
783  } else {
784  return false;
785  }
786  }
787 
788  list( $shardIndex ) = $this->getTableByKey( $key );
789 
790  $db = null; // in case of connection failure
791  try {
792  $db = $this->getConnection( $shardIndex );
793  $ok = $db->lock( $key, __METHOD__, $timeout );
794  if ( $ok ) {
795  $this->locks[$key] = [ 'class' => $rclass, 'depth' => 1 ];
796  }
797 
798  $this->logger->warning(
799  __METHOD__ . " failed due to timeout for {key}.",
800  [ 'key' => $key, 'timeout' => $timeout ]
801  );
802 
803  return $ok;
804  } catch ( DBError $e ) {
805  $this->handleWriteError( $e, $db, $shardIndex );
806  $ok = false;
807  }
808 
809  return $ok;
810  }
811 
812  public function unlock( $key ) {
813  if ( !isset( $this->locks[$key] ) ) {
814  return false;
815  }
816 
817  if ( --$this->locks[$key]['depth'] <= 0 ) {
818  unset( $this->locks[$key] );
819 
820  list( $shardIndex ) = $this->getTableByKey( $key );
821 
822  $db = null; // in case of connection failure
823  try {
824  $db = $this->getConnection( $shardIndex );
825  $ok = $db->unlock( $key, __METHOD__ );
826  if ( !$ok ) {
827  $this->logger->warning(
828  __METHOD__ . ' failed to release lock for {key}.',
829  [ 'key' => $key ]
830  );
831  }
832  } catch ( DBError $e ) {
833  $this->handleWriteError( $e, $db, $shardIndex );
834  $ok = false;
835  }
836 
837  return $ok;
838  }
839 
840  return true;
841  }
842 
851  protected function serialize( $data ) {
852  if ( is_int( $data ) ) {
853  return $data;
854  }
855 
856  $serial = serialize( $data );
857  if ( function_exists( 'gzdeflate' ) ) {
858  $serial = gzdeflate( $serial );
859  }
860 
861  return $serial;
862  }
863 
869  protected function unserialize( $serial ) {
870  if ( $this->isInteger( $serial ) ) {
871  return (int)$serial;
872  }
873 
874  if ( function_exists( 'gzinflate' ) ) {
875  AtEase::suppressWarnings();
876  $decomp = gzinflate( $serial );
877  AtEase::restoreWarnings();
878 
879  if ( $decomp !== false ) {
880  $serial = $decomp;
881  }
882  }
883 
884  return unserialize( $serial );
885  }
886 
893  private function handleReadError( DBError $exception, $shardIndex ) {
894  if ( $exception instanceof DBConnectionError ) {
895  $this->markServerDown( $exception, $shardIndex );
896  }
897 
898  $this->setAndLogDBError( $exception );
899  }
900 
909  private function handleWriteError( DBError $exception, $db, $shardIndex ) {
910  if ( !( $db instanceof IDatabase ) ) {
911  $this->markServerDown( $exception, $shardIndex );
912  }
913 
914  $this->setAndLogDBError( $exception );
915  }
916 
920  private function setAndLogDBError( DBError $exception ) {
921  $this->logger->error( "DBError: {$exception->getMessage()}" );
922  if ( $exception instanceof DBConnectionError ) {
924  $this->logger->debug( __METHOD__ . ": ignoring connection error" );
925  } else {
927  $this->logger->debug( __METHOD__ . ": ignoring query error" );
928  }
929  }
930 
937  private function markServerDown( DBError $exception, $shardIndex ) {
938  unset( $this->conns[$shardIndex] ); // bug T103435
939 
940  $now = $this->getCurrentTime();
941  if ( isset( $this->connFailureTimes[$shardIndex] ) ) {
942  if ( $now - $this->connFailureTimes[$shardIndex] >= 60 ) {
943  unset( $this->connFailureTimes[$shardIndex] );
944  unset( $this->connFailureErrors[$shardIndex] );
945  } else {
946  $this->logger->debug( __METHOD__ . ": Server #$shardIndex already down" );
947  return;
948  }
949  }
950  $this->logger->info( __METHOD__ . ": Server #$shardIndex down until " . ( $now + 60 ) );
951  $this->connFailureTimes[$shardIndex] = $now;
952  $this->connFailureErrors[$shardIndex] = $exception;
953  }
954 
959  private function initSqliteDatabase( IMaintainableDatabase $db ) {
960  if ( $db->tableExists( 'objectcache' ) ) {
961  return;
962  }
963  // Use one table for SQLite; sharding does not seem to have much benefit
964  $db->query( "PRAGMA journal_mode=WAL" ); // this is permanent
965  $db->startAtomic( __METHOD__ ); // atomic DDL
966  try {
967  $encTable = $db->tableName( 'objectcache' );
968  $encExptimeIndex = $db->addIdentifierQuotes( $db->tablePrefix() . 'exptime' );
969  $db->query(
970  "CREATE TABLE $encTable (\n" .
971  " keyname BLOB NOT NULL default '' PRIMARY KEY,\n" .
972  " value BLOB,\n" .
973  " exptime TEXT\n" .
974  ")",
975  __METHOD__
976  );
977  $db->query( "CREATE INDEX $encExptimeIndex ON $encTable (exptime)" );
978  $db->endAtomic( __METHOD__ );
979  } catch ( DBError $e ) {
980  $db->rollback( __METHOD__ );
981  throw $e;
982  }
983  }
984 
988  public function createTables() {
989  for ( $shardIndex = 0; $shardIndex < $this->numServerShards; $shardIndex++ ) {
990  $db = $this->getConnection( $shardIndex );
991  if ( in_array( $db->getType(), [ 'mysql', 'postgres' ], true ) ) {
992  for ( $i = 0; $i < $this->numTableShards; $i++ ) {
993  $encBaseTable = $db->tableName( 'objectcache' );
994  $encShardTable = $db->tableName( $this->getTableNameByShard( $i ) );
995  $db->query( "CREATE TABLE $encShardTable LIKE $encBaseTable" );
996  }
997  }
998  }
999  }
1000 
1004  private function usesMainDB() {
1005  return !$this->serverInfos;
1006  }
1007 
1008  private function waitForReplication() {
1009  if ( !$this->usesMainDB() ) {
1010  // Custom DB server list; probably doesn't use replication
1011  return true;
1012  }
1013 
1014  $lb = MediaWikiServices::getInstance()->getDBLoadBalancer();
1015  if ( $lb->getServerCount() <= 1 ) {
1016  return true; // no replica DBs
1017  }
1018 
1019  // Main LB is used; wait for any replica DBs to catch up
1020  try {
1021  $masterPos = $lb->getMasterPos();
1022  if ( !$masterPos ) {
1023  return true; // not applicable
1024  }
1025 
1026  $loop = new WaitConditionLoop(
1027  function () use ( $lb, $masterPos ) {
1028  return $lb->waitForAll( $masterPos, 1 );
1029  },
1032  );
1033 
1034  return ( $loop->invoke() === $loop::CONDITION_REACHED );
1035  } catch ( DBError $e ) {
1036  $this->setAndLogDBError( $e );
1037 
1038  return false;
1039  }
1040  }
1041 
1047  private function silenceTransactionProfiler() {
1048  if ( !$this->usesMainDB() ) {
1049  // Custom DB is configured which either has no TransactionProfiler injected,
1050  // or has one specific for cache use, which we shouldn't silence
1051  return null;
1052  }
1053 
1054  $trxProfiler = Profiler::instance()->getTransactionProfiler();
1055  $oldSilenced = $trxProfiler->setSilenced( true );
1056  return new ScopedCallback( function () use ( $trxProfiler, $oldSilenced ) {
1057  $trxProfiler->setSilenced( $oldSilenced );
1058  } );
1059  }
1060 }
static string $OP_DELETE
modifyMulti(array $data, $exptime, $flags, $op)
unserialize( $serial)
Unserialize and, if necessary, decompress an object.
addIdentifierQuotes( $s)
Escape a SQL identifier (e.g.
affectedRows()
Get the number of rows affected by the last write query.
$success
doSet( $key, $value, $exptime=0, $flags=0)
array [] $serverInfos
(server index => server config)
static instance()
Singleton.
Definition: Profiler.php:63
changeTTLMulti(array $keys, $exptime, $flags=0)
static string $OP_SET
deleteServerObjectsExpiringBefore(IDatabase $db, $timestamp, $progressCallback, $limit, $serversDoneCount=0, &$keysDeletedCount=0)
lock( $key, $timeout=6, $expiry=6, $rclass='')
endAtomic( $fname=__METHOD__)
Ends an atomic section of SQL statements.
isExpired(IDatabase $db, $exptime)
int $lastGarbageCollect
UNIX timestamp.
const DB_MASTER
Definition: defines.php:26
initSqliteDatabase(IMaintainableDatabase $db)
timestamp( $ts=0)
Convert a timestamp in one of the formats accepted by ConvertibleTimestamp to the format used for ins...
getCurrentTime()
Definition: BagOStuff.php:523
rollback( $fname=__METHOD__, $flush=self::FLUSHING_ONE)
Rollback a transaction previously started using begin()
getMaxDateTime( $db)
string [] $serverTags
(server index => tag/host name)
fetchBlobMulti(array $keys, $flags=0)
string $tableName
doAdd( $key, $value, $exptime=0, $flags=0)
getConnection( $shardIndex)
Get a connection to the specified database.
createTables()
Create the shard tables on all databases (e.g.
doChangeTTL( $key, $exptime, $flags)
occasionallyGarbageCollect(IDatabase $db)
deleteAll()
Delete content of shard tables in every server.
int $numServerShards
Storage medium specific cache for storing items (e.g.
doGet( $key, $flags=0, &$casToken=null)
const DBO_TRX
Definition: defines.php:12
getExpirationAsTimestamp( $exptime)
Convert an optionally relative timestamp to an absolute time.
doGetMulti(array $keys, $flags=0)
static string $OP_TOUCH
LoadBalancer null $separateMainLB
static int $GC_DELAY_SEC
doSetMulti(array $data, $exptime=0, $flags=0)
Basic database interface for live and lazy-loaded relation database handles.
Definition: IDatabase.php:38
array $connFailureTimes
UNIX timestamps.
startAtomic( $fname=__METHOD__, $cancelable=self::ATOMIC_NOT_CANCELABLE)
Begin an atomic section of SQL statements.
decr( $key, $value=1, $flags=0)
Advanced database interface for IDatabase handles that include maintenance methods.
static consistentHashSort(&$array, $key, $separator="\0")
Sort the given array in a pseudo-random order which depends only on the given key and each element va...
Definition: ArrayUtils.php:49
setLastError( $err)
Set the "last error" registry.
getTableNameByShard( $index)
Get the table name for a given shard index.
select( $table, $vars, $conds='', $fname=__METHOD__, $options=[], $join_conds=[])
Execute a SELECT query constructed using the various parameters provided.
updateTableKeys( $op, $db, $table, $tableKeys, $data, $dbExpiry)
handleWriteError(DBError $exception, $db, $shardIndex)
Handle a DBQueryError which occurred during a write operation.
doCas( $casToken, $key, $value, $exptime=0, $flags=0)
doDelete( $key, $flags=0)
getTableByKey( $key)
Get the server index and table name for a given key.
markServerDown(DBError $exception, $shardIndex)
Mark a server down due to a DBConnectionError exception.
doDeleteMulti(array $keys, $flags=0)
const DB_REPLICA
Definition: defines.php:25
Class to store objects in the database.
isInteger( $value)
Check if a value is an integer.
query( $sql, $fname=__METHOD__, $flags=0)
Run an SQL query and return the result.
delete( $table, $conds, $fname=__METHOD__)
DELETE query wrapper.
deleteObjectsExpiringBefore( $timestamp, callable $progress=null, $limit=INF)
silenceTransactionProfiler()
Silence the transaction profiler until the return value falls out of scope.
addQuotes( $s)
Escape and quote a raw value string for use in a SQL query.
tablePrefix( $prefix=null)
Get/set the table prefix.
incr( $key, $step=1, $flags=0)
return true
Definition: router.php:92
Database error base class.
Definition: DBError.php:30
array $connFailureErrors
Exceptions.
callable null $asyncHandler
Definition: BagOStuff.php:68
tableName( $name, $format='quoted')
Format a table name ready for use in constructing an SQL query.
setAndLogDBError(DBError $exception)
tableExists( $table, $fname=__METHOD__)
Query whether a given table exists.
fieldHasFlags( $field, $flags)
Definition: BagOStuff.php:493
__construct( $params)
Constructor.
serialize( $data)
Serialize an object and, if possible, compress the representation.
static string $OP_ADD
handleReadError(DBError $exception, $shardIndex)
Handle a DBError which occurred during a read operation.