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;
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  $this->replicaOnly = $params['replicaOnly'] ?? false;
176  }
177 
186  private function getConnection( $shardIndex ) {
187  // Don't keep timing out trying to connect if the server is down
188  if (
189  isset( $this->connFailureErrors[$shardIndex] ) &&
190  ( $this->getCurrentTime() - $this->connFailureTimes[$shardIndex] ) < 60
191  ) {
192  throw $this->connFailureErrors[$shardIndex];
193  }
194 
195  if ( $shardIndex === self::SHARD_LOCAL ) {
196  $conn = $this->getConnectionViaLoadBalancer( $shardIndex );
197  } elseif ( $shardIndex === self::SHARD_GLOBAL ) {
198  $conn = $this->getConnectionViaLoadBalancer( $shardIndex );
199  } elseif ( is_int( $shardIndex ) ) {
200  if ( isset( $this->serverInfos[$shardIndex] ) ) {
201  $server = $this->serverInfos[$shardIndex];
202  $conn = $this->getConnectionFromServerInfo( $shardIndex, $server );
203  } else {
204  throw new UnexpectedValueException( "Invalid server index #$shardIndex" );
205  }
206  } else {
207  throw new UnexpectedValueException( "Invalid server index '$shardIndex'" );
208  }
209 
210  return $conn;
211  }
212 
218  private function getKeyLocation( $key ) {
219  if ( $this->serverTags ) {
220  // Striped array of database servers
221  if ( count( $this->serverTags ) == 1 ) {
222  $shardIndex = 0; // short-circuit
223  } else {
224  $sortedServers = $this->serverTags;
225  ArrayUtils::consistentHashSort( $sortedServers, $key );
226  reset( $sortedServers );
227  $shardIndex = key( $sortedServers );
228  }
229  } else {
230  // LoadBalancer based configuration
231  $shardIndex = ( strpos( $key, 'global:' ) === 0 && $this->globalKeyLb )
232  ? self::SHARD_GLOBAL
233  : self::SHARD_LOCAL;
234  }
235 
236  if ( $this->numTableShards > 1 ) {
237  $hash = hexdec( substr( md5( $key ), 0, 8 ) ) & 0x7fffffff;
238  $tableIndex = $hash % $this->numTableShards;
239  } else {
240  $tableIndex = null;
241  }
242 
243  return [ $shardIndex, $this->getTableNameByShard( $tableIndex ) ];
244  }
245 
251  private function getTableNameByShard( $index ) {
252  if ( $index !== null && $this->numTableShards > 1 ) {
253  $decimals = strlen( $this->numTableShards - 1 );
254 
255  return $this->tableName . sprintf( "%0{$decimals}d", $index );
256  }
257 
258  return $this->tableName;
259  }
260 
261  protected function doGet( $key, $flags = 0, &$casToken = null ) {
262  $getToken = ( $casToken === self::PASS_BY_REF );
263  $casToken = null;
264 
265  $blobs = $this->fetchBlobMulti( [ $key ] );
266  if ( array_key_exists( $key, $blobs ) ) {
267  $blob = $blobs[$key];
268  $result = $this->unserialize( $blob );
269  if ( $getToken && $blob !== false ) {
270  $casToken = $blob;
271  }
272  $valueSize = strlen( $blob );
273  } else {
274  $result = false;
275  $valueSize = false;
276  }
277 
278  $this->updateOpStats( self::METRIC_OP_GET, [ $key => [ null, $valueSize ] ] );
279 
280  return $result;
281  }
282 
283  protected function doGetMulti( array $keys, $flags = 0 ) {
284  $values = [];
285  $valueSizeByKey = [];
286 
287  $blobsByKey = $this->fetchBlobMulti( $keys );
288  foreach ( $keys as $key ) {
289  if ( array_key_exists( $key, $blobsByKey ) ) {
290  $blob = $blobsByKey[$key];
291  $value = $this->unserialize( $blob );
292  if ( $value !== false ) {
293  $values[$key] = $value;
294  }
295  $valueSize = strlen( $blob );
296  } else {
297  $valueSize = false;
298  }
299  $valueSizeByKey[$key] = [ null, $valueSize ];
300  }
301 
302  $this->updateOpStats( self::METRIC_OP_GET, $valueSizeByKey );
303 
304  return $values;
305  }
306 
307  private function fetchBlobMulti( array $keys ) {
308  $values = []; // array of (key => value)
309 
310  $keysByTableByShardIndex = [];
311  foreach ( $keys as $key ) {
312  list( $shardIndex, $tableName ) = $this->getKeyLocation( $key );
313  $keysByTableByShardIndex[$shardIndex][$tableName][] = $key;
314  }
315 
316  $dataRows = [];
317  foreach ( $keysByTableByShardIndex as $shardIndex => $serverKeys ) {
318  try {
319  $db = $this->getConnection( $shardIndex );
320  foreach ( $serverKeys as $tableName => $tableKeys ) {
321  $res = $db->select(
322  $tableName,
323  [ 'keyname', 'value', 'exptime' ],
324  [ 'keyname' => $tableKeys ],
325  __METHOD__,
326  // Approximate write-on-the-fly BagOStuff API via blocking.
327  // This approximation fails if a ROLLBACK happens (which is rare).
328  // We do not want to flush the TRX as that can break callers.
329  $db->trxLevel() ? [ 'LOCK IN SHARE MODE' ] : []
330  );
331  if ( $res === false ) {
332  continue;
333  }
334  foreach ( $res as $row ) {
335  $row->shardIndex = $shardIndex;
336  $row->tableName = $tableName;
337  $dataRows[$row->keyname] = $row;
338  }
339  }
340  } catch ( DBError $e ) {
341  $this->handleReadError( $e, $shardIndex );
342  }
343  }
344 
345  foreach ( $keys as $key ) {
346  if ( isset( $dataRows[$key] ) ) { // HIT?
347  $row = $dataRows[$key];
348  $this->debug( "get: retrieved data; expiry time is " . $row->exptime );
349  $db = null; // in case of connection failure
350  try {
351  $db = $this->getConnection( $row->shardIndex );
352  if ( $this->isExpired( $db, $row->exptime ) ) { // MISS
353  $this->debug( "get: key has expired" );
354  } else { // HIT
355  $values[$key] = $db->decodeBlob( $row->value );
356  }
357  } catch ( DBQueryError $e ) {
358  $this->handleWriteError( $e, $db, $row->shardIndex );
359  }
360  } else { // MISS
361  $this->debug( 'get: no matching rows' );
362  }
363  }
364 
365  return $values;
366  }
367 
368  protected function doSetMulti( array $data, $exptime = 0, $flags = 0 ) {
369  return $this->modifyMulti( $data, $exptime, $flags, self::$OP_SET );
370  }
371 
379  private function modifyMulti( array $data, $exptime, $flags, $op ) {
380  $keysByTableByShardIndex = [];
381  foreach ( $data as $key => $value ) {
382  list( $shardIndex, $tableName ) = $this->getKeyLocation( $key );
383  $keysByTableByShardIndex[$shardIndex][$tableName][] = $key;
384  }
385 
386  $exptime = $this->getExpirationAsTimestamp( $exptime );
387 
388  $result = true;
390  $silenceScope = $this->silenceTransactionProfiler();
391  foreach ( $keysByTableByShardIndex as $shardIndex => $serverKeys ) {
392  $db = null; // in case of connection failure
393  try {
394  $db = $this->getConnection( $shardIndex );
395  $this->occasionallyGarbageCollect( $db ); // expire old entries if any
396  $dbExpiry = $exptime ? $db->timestamp( $exptime ) : $this->getMaxDateTime( $db );
397  } catch ( DBError $e ) {
398  $this->handleWriteError( $e, $db, $shardIndex );
399  $result = false;
400  continue;
401  }
402 
403  foreach ( $serverKeys as $tableName => $tableKeys ) {
404  try {
405  $result = $this->updateTable(
406  $op,
407  $db,
408  $tableName,
409  $tableKeys,
410  $data,
411  $dbExpiry
412  ) && $result;
413  } catch ( DBError $e ) {
414  $this->handleWriteError( $e, $db, $shardIndex );
415  $result = false;
416  }
417 
418  }
419  }
420 
421  if ( $this->fieldHasFlags( $flags, self::WRITE_SYNC ) ) {
422  foreach ( $keysByTableByShardIndex as $shardIndex => $unused ) {
423  $result = $this->waitForReplication( $shardIndex ) && $result;
424  }
425  }
426 
427  return $result;
428  }
429 
441  private function updateTable( $op, $db, $table, $tableKeys, $data, $dbExpiry ) {
442  $success = true;
443 
444  if ( $op === self::$OP_ADD ) {
445  $valueSizesByKey = [];
446 
447  $rows = [];
448  foreach ( $tableKeys as $key ) {
449  $serialized = $this->getSerialized( $data[$key], $key );
450  $rows[] = [
451  'keyname' => $key,
452  'value' => $db->encodeBlob( $serialized ),
453  'exptime' => $dbExpiry
454  ];
455  $valueSizesByKey[$key] = [ strlen( $serialized ), null ];
456  }
457  $db->delete(
458  $table,
459  [
460  'keyname' => $tableKeys,
461  'exptime <= ' . $db->addQuotes( $db->timestamp() )
462  ],
463  __METHOD__
464  );
465  $db->insert( $table, $rows, __METHOD__, [ 'IGNORE' ] );
466 
467  $success = ( $db->affectedRows() == count( $rows ) );
468 
469  $this->updateOpStats( self::METRIC_OP_ADD, $valueSizesByKey );
470  } elseif ( $op === self::$OP_SET ) {
471  $valueSizesByKey = [];
472 
473  $rows = [];
474  foreach ( $tableKeys as $key ) {
475  $serialized = $this->getSerialized( $data[$key], $key );
476  $rows[] = [
477  'keyname' => $key,
478  'value' => $db->encodeBlob( $serialized ),
479  'exptime' => $dbExpiry
480  ];
481  $valueSizesByKey[$key] = [ strlen( $serialized ), null ];
482  }
483  $db->replace( $table, 'keyname', $rows, __METHOD__ );
484 
485  $this->updateOpStats( self::METRIC_OP_SET, $valueSizesByKey );
486  } elseif ( $op === self::$OP_DELETE ) {
487  $db->delete( $table, [ 'keyname' => $tableKeys ], __METHOD__ );
488 
489  $this->updateOpStats( self::METRIC_OP_DELETE, $tableKeys );
490  } elseif ( $op === self::$OP_TOUCH ) {
491  $db->update(
492  $table,
493  [ 'exptime' => $dbExpiry ],
494  [
495  'keyname' => $tableKeys,
496  'exptime > ' . $db->addQuotes( $db->timestamp() )
497  ],
498  __METHOD__
499  );
500 
501  $success = ( $db->affectedRows() == count( $tableKeys ) );
502 
503  $this->updateOpStats( self::METRIC_OP_CHANGE_TTL, $tableKeys );
504  } else {
505  throw new InvalidArgumentException( "Invalid operation '$op'" );
506  }
507 
508  return $success;
509  }
510 
511  protected function doSet( $key, $value, $exptime = 0, $flags = 0 ) {
512  return $this->modifyMulti( [ $key => $value ], $exptime, $flags, self::$OP_SET );
513  }
514 
515  protected function doAdd( $key, $value, $exptime = 0, $flags = 0 ) {
516  return $this->modifyMulti( [ $key => $value ], $exptime, $flags, self::$OP_ADD );
517  }
518 
519  protected function doCas( $casToken, $key, $value, $exptime = 0, $flags = 0 ) {
520  list( $shardIndex, $tableName ) = $this->getKeyLocation( $key );
521  $exptime = $this->getExpirationAsTimestamp( $exptime );
522  $serialized = $this->getSerialized( $value, $key );
523 
525  $silenceScope = $this->silenceTransactionProfiler();
526  $db = null; // in case of connection failure
527  try {
528  $db = $this->getConnection( $shardIndex );
529  // (T26425) use a replace if the db supports it instead of
530  // delete/insert to avoid clashes with conflicting keynames
531  $db->update(
532  $tableName,
533  [
534  'keyname' => $key,
535  'value' => $db->encodeBlob( $serialized ),
536  'exptime' => $exptime
537  ? $db->timestamp( $exptime )
538  : $this->getMaxDateTime( $db )
539  ],
540  [
541  'keyname' => $key,
542  'value' => $db->encodeBlob( $casToken ),
543  'exptime > ' . $db->addQuotes( $db->timestamp() )
544  ],
545  __METHOD__
546  );
547  } catch ( DBQueryError $e ) {
548  $this->handleWriteError( $e, $db, $shardIndex );
549 
550  return false;
551  }
552 
553  $success = (bool)$db->affectedRows();
554  if ( $this->fieldHasFlags( $flags, self::WRITE_SYNC ) ) {
555  $success = $this->waitForReplication( $shardIndex ) && $success;
556  }
557 
558  $this->updateOpStats( self::METRIC_OP_CAS, [ $key => [ strlen( $serialized ), null ] ] );
559 
560  return $success;
561  }
562 
563  protected function doDeleteMulti( array $keys, $flags = 0 ) {
564  return $this->modifyMulti(
565  array_fill_keys( $keys, null ),
566  0,
567  $flags,
568  self::$OP_DELETE
569  );
570  }
571 
572  protected function doDelete( $key, $flags = 0 ) {
573  return $this->modifyMulti( [ $key => null ], 0, $flags, self::$OP_DELETE );
574  }
575 
576  public function incr( $key, $step = 1, $flags = 0 ) {
577  list( $shardIndex, $tableName ) = $this->getKeyLocation( $key );
578 
579  $newCount = false;
581  $silenceScope = $this->silenceTransactionProfiler();
582  $db = null; // in case of connection failure
583  try {
584  $db = $this->getConnection( $shardIndex );
585  $encTimestamp = $db->addQuotes( $db->timestamp() );
586  $db->update(
587  $tableName,
588  [ 'value = value + ' . (int)$step ],
589  [ 'keyname' => $key, "exptime > $encTimestamp" ],
590  __METHOD__
591  );
592  if ( $db->affectedRows() > 0 ) {
593  $newValue = $db->selectField(
594  $tableName,
595  'value',
596  [ 'keyname' => $key, "exptime > $encTimestamp" ],
597  __METHOD__
598  );
599  if ( $this->isInteger( $newValue ) ) {
600  $newCount = (int)$newValue;
601  }
602  }
603  } catch ( DBError $e ) {
604  $this->handleWriteError( $e, $db, $shardIndex );
605  }
606 
607  $this->updateOpStats( $step >= 0 ? self::METRIC_OP_INCR : self::METRIC_OP_DECR, [ $key ] );
608 
609  return $newCount;
610  }
611 
612  public function decr( $key, $value = 1, $flags = 0 ) {
613  return $this->incr( $key, -$value, $flags );
614  }
615 
616  public function doChangeTTLMulti( array $keys, $exptime, $flags = 0 ) {
617  return $this->modifyMulti(
618  array_fill_keys( $keys, null ),
619  $exptime,
620  $flags,
621  self::$OP_TOUCH
622  );
623  }
624 
625  protected function doChangeTTL( $key, $exptime, $flags ) {
626  return $this->modifyMulti( [ $key => null ], $exptime, $flags, self::$OP_TOUCH );
627  }
628 
634  private function isExpired( IDatabase $db, $exptime ) {
635  return (
636  $exptime != $this->getMaxDateTime( $db ) &&
637  ConvertibleTimestamp::convert( TS_UNIX, $exptime ) < $this->getCurrentTime()
638  );
639  }
640 
645  private function getMaxDateTime( $db ) {
646  if ( (int)$this->getCurrentTime() > 0x7fffffff ) {
647  return $db->timestamp( 1 << 62 );
648  } else {
649  return $db->timestamp( 0x7fffffff );
650  }
651  }
652 
657  private function occasionallyGarbageCollect( IDatabase $db ) {
658  if (
659  // Random purging is enabled
660  $this->purgePeriod &&
661  // Only purge on one in every $this->purgePeriod writes
662  mt_rand( 0, $this->purgePeriod - 1 ) == 0 &&
663  // Avoid repeating the delete within a few seconds
664  ( $this->getCurrentTime() - $this->lastGarbageCollect ) > self::$GC_DELAY_SEC
665  ) {
666  $garbageCollector = function () use ( $db ) {
668  $db, $this->getCurrentTime(),
669  null,
670  $this->purgeLimit
671  );
672  $this->lastGarbageCollect = time();
673  };
674  if ( $this->asyncHandler ) {
675  $this->lastGarbageCollect = $this->getCurrentTime(); // avoid duplicate enqueues
676  ( $this->asyncHandler )( $garbageCollector );
677  } else {
678  $garbageCollector();
679  }
680  }
681  }
682 
683  public function expireAll() {
684  $this->deleteObjectsExpiringBefore( $this->getCurrentTime() );
685  }
686 
688  $timestamp,
689  callable $progress = null,
690  $limit = INF
691  ) {
693  $silenceScope = $this->silenceTransactionProfiler();
694 
695  $shardIndexes = $this->getServerShardIndexes();
696  shuffle( $shardIndexes );
697 
698  $ok = true;
699 
700  $keysDeletedCount = 0;
701  foreach ( $shardIndexes as $numServersDone => $shardIndex ) {
702  $db = null; // in case of connection failure
703  try {
704  $db = $this->getConnection( $shardIndex );
706  $db,
707  $timestamp,
708  $progress,
709  $limit,
710  $numServersDone,
711  $keysDeletedCount
712  );
713  } catch ( DBError $e ) {
714  $this->handleWriteError( $e, $db, $shardIndex );
715  $ok = false;
716  }
717  }
718 
719  return $ok;
720  }
721 
732  IDatabase $db,
733  $timestamp,
734  $progressCallback,
735  $limit,
736  $serversDoneCount = 0,
737  &$keysDeletedCount = 0
738  ) {
739  $cutoffUnix = ConvertibleTimestamp::convert( TS_UNIX, $timestamp );
740  $tableIndexes = range( 0, $this->numTableShards - 1 );
741  shuffle( $tableIndexes );
742 
743  foreach ( $tableIndexes as $numShardsDone => $tableIndex ) {
744  $continue = null; // last exptime
745  $lag = null; // purge lag
746  do {
747  $res = $db->select(
748  $this->getTableNameByShard( $tableIndex ),
749  [ 'keyname', 'exptime' ],
750  array_merge(
751  [ 'exptime < ' . $db->addQuotes( $db->timestamp( $cutoffUnix ) ) ],
752  $continue ? [ 'exptime >= ' . $db->addQuotes( $continue ) ] : []
753  ),
754  __METHOD__,
755  [ 'LIMIT' => min( $limit, 100 ), 'ORDER BY' => 'exptime' ]
756  );
757 
758  if ( $res->numRows() ) {
759  $row = $res->current();
760  if ( $lag === null ) {
761  $rowExpUnix = ConvertibleTimestamp::convert( TS_UNIX, $row->exptime );
762  $lag = max( $cutoffUnix - $rowExpUnix, 1 );
763  }
764 
765  $keys = [];
766  foreach ( $res as $row ) {
767  $keys[] = $row->keyname;
768  $continue = $row->exptime;
769  }
770 
771  $db->delete(
772  $this->getTableNameByShard( $tableIndex ),
773  [
774  'exptime < ' . $db->addQuotes( $db->timestamp( $cutoffUnix ) ),
775  'keyname' => $keys
776  ],
777  __METHOD__
778  );
779  $keysDeletedCount += $db->affectedRows();
780  }
781 
782  if ( is_callable( $progressCallback ) ) {
783  if ( $lag ) {
784  $continueUnix = ConvertibleTimestamp::convert( TS_UNIX, $continue );
785  $remainingLag = $cutoffUnix - $continueUnix;
786  $processedLag = max( $lag - $remainingLag, 0 );
787  $doneRatio =
788  ( $numShardsDone + $processedLag / $lag ) / $this->numTableShards;
789  } else {
790  $doneRatio = 1;
791  }
792 
793  $overallRatio = ( $doneRatio / $this->numServerShards )
794  + ( $serversDoneCount / $this->numServerShards );
795  call_user_func( $progressCallback, $overallRatio * 100 );
796  }
797  } while ( $res->numRows() && $keysDeletedCount < $limit );
798  }
799  }
800 
806  public function deleteAll() {
808  $silenceScope = $this->silenceTransactionProfiler();
809  foreach ( $this->getServerShardIndexes() as $shardIndex ) {
810  $db = null; // in case of connection failure
811  try {
812  $db = $this->getConnection( $shardIndex );
813  for ( $i = 0; $i < $this->numTableShards; $i++ ) {
814  $db->delete( $this->getTableNameByShard( $i ), '*', __METHOD__ );
815  }
816  } catch ( DBError $e ) {
817  $this->handleWriteError( $e, $db, $shardIndex );
818  return false;
819  }
820  }
821  return true;
822  }
823 
824  public function lock( $key, $timeout = 6, $expiry = 6, $rclass = '' ) {
825  // Avoid deadlocks and allow lock reentry if specified
826  if ( isset( $this->locks[$key] ) ) {
827  if ( $rclass != '' && $this->locks[$key]['class'] === $rclass ) {
828  ++$this->locks[$key]['depth'];
829  return true;
830  } else {
831  return false;
832  }
833  }
834 
835  list( $shardIndex ) = $this->getKeyLocation( $key );
836 
837  $db = null; // in case of connection failure
838  try {
839  $db = $this->getConnection( $shardIndex );
840  $ok = $db->lock( $key, __METHOD__, $timeout );
841  if ( $ok ) {
842  $this->locks[$key] = [ 'class' => $rclass, 'depth' => 1 ];
843  }
844 
845  $this->logger->warning(
846  __METHOD__ . " failed due to timeout for {key}.",
847  [ 'key' => $key, 'timeout' => $timeout ]
848  );
849 
850  return $ok;
851  } catch ( DBError $e ) {
852  $this->handleWriteError( $e, $db, $shardIndex );
853  $ok = false;
854  }
855 
856  return $ok;
857  }
858 
859  public function unlock( $key ) {
860  if ( !isset( $this->locks[$key] ) ) {
861  return false;
862  }
863 
864  if ( --$this->locks[$key]['depth'] <= 0 ) {
865  unset( $this->locks[$key] );
866 
867  list( $shardIndex ) = $this->getKeyLocation( $key );
868 
869  $db = null; // in case of connection failure
870  try {
871  $db = $this->getConnection( $shardIndex );
872  $ok = $db->unlock( $key, __METHOD__ );
873  if ( !$ok ) {
874  $this->logger->warning(
875  __METHOD__ . ' failed to release lock for {key}.',
876  [ 'key' => $key ]
877  );
878  }
879  } catch ( DBError $e ) {
880  $this->handleWriteError( $e, $db, $shardIndex );
881  $ok = false;
882  }
883 
884  return $ok;
885  }
886 
887  return true;
888  }
889 
898  public function makeKeyInternal( $keyspace, $components ) {
899  // SQL schema for 'objectcache' specifies keys as varchar(255). From that,
900  // subtract the number of characters we need for the keyspace and for
901  // the separator character needed for each argument. To handle some
902  // custom prefixes used by thing like WANObjectCache, limit to 205.
903  $keyspace = strtr( $keyspace, ' ', '_' );
904  $charsLeft = 205 - strlen( $keyspace ) - count( $components );
905  foreach ( $components as &$component ) {
906  $component = strtr( $component, [
907  ' ' => '_', // Avoid unnecessary misses from pre-1.35 code
908  ':' => '%3A',
909  ] );
910 
911  // 33 = 32 characters for the MD5 + 1 for the '#' prefix.
912  if ( $charsLeft > 33 && strlen( $component ) > $charsLeft ) {
913  $component = '#' . md5( $component );
914  }
915  $charsLeft -= strlen( $component );
916  }
917 
918  if ( $charsLeft < 0 ) {
919  return $keyspace . ':BagOStuff-long-key:##' . md5( implode( ':', $components ) );
920  }
921  return $keyspace . ':' . implode( ':', $components );
922  }
923 
932  protected function serialize( $data ) {
933  if ( is_int( $data ) ) {
934  return $data;
935  }
936 
937  $serial = serialize( $data );
938  if ( function_exists( 'gzdeflate' ) ) {
939  $serial = gzdeflate( $serial );
940  }
941 
942  return $serial;
943  }
944 
950  protected function unserialize( $serial ) {
951  if ( $this->isInteger( $serial ) ) {
952  return (int)$serial;
953  }
954 
955  if ( function_exists( 'gzinflate' ) ) {
956  AtEase::suppressWarnings();
957  $decomp = gzinflate( $serial );
958  AtEase::restoreWarnings();
959 
960  if ( $decomp !== false ) {
961  $serial = $decomp;
962  }
963  }
964 
965  return unserialize( $serial );
966  }
967 
973  private function getConnectionViaLoadBalancer( $shardIndex ) {
974  $lb = ( $shardIndex === self::SHARD_LOCAL ) ? $this->localKeyLb : $this->globalKeyLb;
975  if ( $lb->getServerAttributes( $lb->getWriterIndex() )[Database::ATTR_DB_LEVEL_LOCKING] ) {
976  // Use the main connection to avoid transaction deadlocks
977  $conn = $lb->getMaintenanceConnectionRef( DB_MASTER );
978  } else {
979  // If the RDBMs has row/table/page level locking, then use separate auto-commit
980  // connection to avoid needless contention and deadlocks.
981  $conn = $lb->getMaintenanceConnectionRef(
982  $this->replicaOnly ? DB_REPLICA : DB_MASTER, [],
983  false,
984  $lb::CONN_TRX_AUTOCOMMIT
985  );
986  }
987 
988  return $conn;
989  }
990 
997  private function getConnectionFromServerInfo( $shardIndex, array $server ) {
998  if ( !isset( $this->conns[$shardIndex] ) ) {
1000  $conn = Database::factory( $server['type'], array_merge(
1001  $server,
1002  [
1003  'flags' => ( $server['flags'] ?? 0 ) & ~IDatabase::DBO_TRX,
1004  'connLogger' => $this->logger,
1005  'queryLogger' => $this->logger
1006  ]
1007  ) );
1008  // Automatically create the objectcache table for sqlite as needed
1009  if ( $conn->getType() === 'sqlite' && !$conn->tableExists( 'objectcache', __METHOD__ ) ) {
1010  $this->initSqliteDatabase( $conn );
1011  }
1012  $this->conns[$shardIndex] = $conn;
1013  }
1014 
1015  return $this->conns[$shardIndex];
1016  }
1017 
1024  private function handleReadError( DBError $exception, $shardIndex ) {
1025  if ( $exception instanceof DBConnectionError ) {
1026  $this->markServerDown( $exception, $shardIndex );
1027  }
1028 
1029  $this->setAndLogDBError( $exception );
1030  }
1031 
1040  private function handleWriteError( DBError $exception, $db, $shardIndex ) {
1041  if ( !( $db instanceof IDatabase ) ) {
1042  $this->markServerDown( $exception, $shardIndex );
1043  }
1044 
1045  $this->setAndLogDBError( $exception );
1046  }
1047 
1051  private function setAndLogDBError( DBError $exception ) {
1052  $this->logger->error( "DBError: {$exception->getMessage()}" );
1053  if ( $exception instanceof DBConnectionError ) {
1054  $this->setLastError( BagOStuff::ERR_UNREACHABLE );
1055  $this->logger->debug( __METHOD__ . ": ignoring connection error" );
1056  } else {
1057  $this->setLastError( BagOStuff::ERR_UNEXPECTED );
1058  $this->logger->debug( __METHOD__ . ": ignoring query error" );
1059  }
1060  }
1061 
1068  private function markServerDown( DBError $exception, $shardIndex ) {
1069  unset( $this->conns[$shardIndex] ); // bug T103435
1070 
1071  $now = $this->getCurrentTime();
1072  if ( isset( $this->connFailureTimes[$shardIndex] ) ) {
1073  if ( $now - $this->connFailureTimes[$shardIndex] >= 60 ) {
1074  unset( $this->connFailureTimes[$shardIndex] );
1075  unset( $this->connFailureErrors[$shardIndex] );
1076  } else {
1077  $this->logger->debug( __METHOD__ . ": Server #$shardIndex already down" );
1078  return;
1079  }
1080  }
1081  $this->logger->info( __METHOD__ . ": Server #$shardIndex down until " . ( $now + 60 ) );
1082  $this->connFailureTimes[$shardIndex] = $now;
1083  $this->connFailureErrors[$shardIndex] = $exception;
1084  }
1085 
1090  private function initSqliteDatabase( IMaintainableDatabase $db ) {
1091  if ( $db->tableExists( 'objectcache', __METHOD__ ) ) {
1092  return;
1093  }
1094  // Use one table for SQLite; sharding does not seem to have much benefit
1095  $db->query( "PRAGMA journal_mode=WAL", __METHOD__ ); // this is permanent
1096  $db->startAtomic( __METHOD__ ); // atomic DDL
1097  try {
1098  $encTable = $db->tableName( 'objectcache' );
1099  $encExptimeIndex = $db->addIdentifierQuotes( $db->tablePrefix() . 'exptime' );
1100  $db->query(
1101  "CREATE TABLE $encTable (\n" .
1102  " keyname BLOB NOT NULL default '' PRIMARY KEY,\n" .
1103  " value BLOB,\n" .
1104  " exptime BLOB NOT NULL\n" .
1105  ")",
1106  __METHOD__
1107  );
1108  $db->query( "CREATE INDEX $encExptimeIndex ON $encTable (exptime)", __METHOD__ );
1109  $db->endAtomic( __METHOD__ );
1110  } catch ( DBError $e ) {
1111  $db->rollback( __METHOD__ );
1112  throw $e;
1113  }
1114  }
1115 
1119  public function createTables() {
1120  foreach ( $this->getServerShardIndexes() as $shardIndex ) {
1121  $db = $this->getConnection( $shardIndex );
1122  if ( in_array( $db->getType(), [ 'mysql', 'postgres' ], true ) ) {
1123  for ( $i = 0; $i < $this->numTableShards; $i++ ) {
1124  $encBaseTable = $db->tableName( 'objectcache' );
1125  $encShardTable = $db->tableName( $this->getTableNameByShard( $i ) );
1126  $db->query( "CREATE TABLE $encShardTable LIKE $encBaseTable", __METHOD__ );
1127  }
1128  }
1129  }
1130  }
1131 
1135  private function getServerShardIndexes() {
1136  if ( $this->serverTags ) {
1137  // Striped array of database servers
1138  $shardIndexes = range( 0, $this->numServerShards - 1 );
1139  } else {
1140  // LoadBalancer based configuration
1141  $shardIndexes = [];
1142  if ( $this->localKeyLb ) {
1143  $shardIndexes[] = self::SHARD_LOCAL;
1144  }
1145  if ( $this->globalKeyLb ) {
1146  $shardIndexes[] = self::SHARD_GLOBAL;
1147  }
1148  }
1149 
1150  return $shardIndexes;
1151  }
1152 
1159  private function waitForReplication( $shardIndex ) {
1160  if ( is_int( $shardIndex ) ) {
1161  return true; // striped only, no LoadBalancer
1162  }
1163 
1164  $lb = ( $shardIndex === self::SHARD_LOCAL ) ? $this->localKeyLb : $this->globalKeyLb;
1165  if ( !$lb->hasStreamingReplicaServers() ) {
1166  return true;
1167  }
1168 
1169  try {
1170  // Wait for any replica DBs to catch up
1171  $masterPos = $lb->getMasterPos();
1172  if ( !$masterPos ) {
1173  return true; // not applicable
1174  }
1175 
1176  $loop = new WaitConditionLoop(
1177  static function () use ( $lb, $masterPos ) {
1178  return $lb->waitForAll( $masterPos, 1 );
1179  },
1182  );
1183 
1184  return ( $loop->invoke() === $loop::CONDITION_REACHED );
1185  } catch ( DBError $e ) {
1186  $this->setAndLogDBError( $e );
1187 
1188  return false;
1189  }
1190  }
1191 
1197  private function silenceTransactionProfiler() {
1198  if ( $this->serverInfos ) {
1199  return null; // no TransactionProfiler injected anyway
1200  }
1201 
1202  $trxProfiler = Profiler::instance()->getTransactionProfiler();
1203  $oldSilenced = $trxProfiler->setSilenced( true );
1204  return new ScopedCallback( static function () use ( $trxProfiler, $oldSilenced ) {
1205  $trxProfiler->setSilenced( $oldSilenced );
1206  } );
1207  }
1208 }
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:770
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:283
SqlBagOStuff\doCas
doCas( $casToken, $key, $value, $exptime=0, $flags=0)
Check and set an item.
Definition: SqlBagOStuff.php:519
SqlBagOStuff\createTables
createTables()
Create the shard tables on all databases (e.g.
Definition: SqlBagOStuff.php:1119
Wikimedia\Rdbms\Database
Relational database abstraction object.
Definition: Database.php:50
MediumSpecificBagOStuff\isInteger
isInteger( $value)
Check if a value is an integer.
Definition: MediumSpecificBagOStuff.php:893
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:261
SqlBagOStuff\waitForReplication
waitForReplication( $shardIndex)
Wait for replica DBs to catch up to the master DB.
Definition: SqlBagOStuff.php:1159
SqlBagOStuff\makeKeyInternal
makeKeyInternal( $keyspace, $components)
Construct a cache key.
Definition: SqlBagOStuff.php:898
Wikimedia\Rdbms\IDatabase\tablePrefix
tablePrefix( $prefix=null)
Get/set the table prefix.
SqlBagOStuff\setAndLogDBError
setAndLogDBError(DBError $exception)
Definition: SqlBagOStuff.php:1051
MediumSpecificBagOStuff\debug
debug( $text)
Definition: MediumSpecificBagOStuff.php:1059
Profiler\instance
static instance()
Singleton.
Definition: Profiler.php:69
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:307
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:973
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:1090
$serialized
foreach( $res as $row) $serialized
Definition: testCompression.php:88
SqlBagOStuff\silenceTransactionProfiler
silenceTransactionProfiler()
Silence the transaction profiler until the return value falls out of scope.
Definition: SqlBagOStuff.php:1197
SqlBagOStuff\incr
incr( $key, $step=1, $flags=0)
Increase stored value of $key by $value while preserving its TTL.
Definition: SqlBagOStuff.php:576
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:441
$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 @newable.
Definition: DBError.php:32
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:511
SqlBagOStuff\$GC_DELAY_SEC
static int $GC_DELAY_SEC
Definition: SqlBagOStuff.php:75
MediumSpecificBagOStuff\$syncTimeout
int $syncTimeout
Seconds.
Definition: MediumSpecificBagOStuff.php:40
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:218
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:683
SqlBagOStuff\occasionallyGarbageCollect
occasionallyGarbageCollect(IDatabase $db)
Definition: SqlBagOStuff.php:657
SqlBagOStuff\unlock
unlock( $key)
Release an advisory lock on a key string.
Definition: SqlBagOStuff.php:859
SqlBagOStuff\deleteAll
deleteAll()
Delete content of shard tables in every server.
Definition: SqlBagOStuff.php:806
MediumSpecificBagOStuff\getExpirationAsTimestamp
getExpirationAsTimestamp( $exptime)
Convert an optionally relative timestamp to an absolute time.
Definition: MediumSpecificBagOStuff.php:853
SqlBagOStuff\$connFailureErrors
Exception[] $connFailureErrors
Map of (shard index => Exception)
Definition: SqlBagOStuff.php:72
SqlBagOStuff\doDeleteMulti
doDeleteMulti(array $keys, $flags=0)
Definition: SqlBagOStuff.php:563
$blob
$blob
Definition: testCompression.php:70
SqlBagOStuff\modifyMulti
modifyMulti(array $data, $exptime, $flags, $op)
Definition: SqlBagOStuff.php:379
BagOStuff\$asyncHandler
callable null $asyncHandler
Definition: BagOStuff.php:92
SqlBagOStuff\getConnectionFromServerInfo
getConnectionFromServerInfo( $shardIndex, array $server)
Definition: SqlBagOStuff.php:997
SqlBagOStuff\isExpired
isExpired(IDatabase $db, $exptime)
Definition: SqlBagOStuff.php:634
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:1024
MediumSpecificBagOStuff\updateOpStats
updateOpStats(string $op, array $keyInfo)
Definition: MediumSpecificBagOStuff.php:1070
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:932
SqlBagOStuff\handleWriteError
handleWriteError(DBError $exception, $db, $shardIndex)
Handle a DBQueryError which occurred during a write operation.
Definition: SqlBagOStuff.php:1040
SqlBagOStuff\$OP_SET
static string $OP_SET
Definition: SqlBagOStuff.php:78
Wikimedia\Rdbms\DBQueryError
@newable
Definition: DBQueryError.php:29
SqlBagOStuff\$globalKeyLb
ILoadBalancer null $globalKeyLb
Definition: SqlBagOStuff.php:46
SqlBagOStuff\getServerShardIndexes
getServerShardIndexes()
Definition: SqlBagOStuff.php:1135
SqlBagOStuff\doChangeTTL
doChangeTTL( $key, $exptime, $flags)
Definition: SqlBagOStuff.php:625
SqlBagOStuff\unserialize
unserialize( $serial)
Unserialize and, if necessary, decompress an object.
Definition: SqlBagOStuff.php:950
SqlBagOStuff\lock
lock( $key, $timeout=6, $expiry=6, $rclass='')
Acquire an advisory lock on a key string.
Definition: SqlBagOStuff.php:824
SqlBagOStuff\getConnection
getConnection( $shardIndex)
Get a connection to the specified database.
Definition: SqlBagOStuff.php:186
MediumSpecificBagOStuff\$busyCallbacks
callable[] $busyCallbacks
Definition: MediumSpecificBagOStuff.php:54
MediumSpecificBagOStuff\getSerialized
getSerialized( $value, $key)
Get the serialized form a value, using any applicable prepared value.
Definition: MediumSpecificBagOStuff.php:966
SqlBagOStuff\doChangeTTLMulti
doChangeTTLMulti(array $keys, $exptime, $flags=0)
Definition: SqlBagOStuff.php:616
SqlBagOStuff\markServerDown
markServerDown(DBError $exception, $shardIndex)
Mark a server down due to a DBConnectionError exception.
Definition: SqlBagOStuff.php:1068
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:731
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:612
SqlBagOStuff\doAdd
doAdd( $key, $value, $exptime=0, $flags=0)
Insert an item if it does not already exist.
Definition: SqlBagOStuff.php:515
SqlBagOStuff\$OP_ADD
static string $OP_ADD
Definition: SqlBagOStuff.php:80
BagOStuff\fieldHasFlags
fieldHasFlags( $field, $flags)
Definition: BagOStuff.php:593
SqlBagOStuff\getMaxDateTime
getMaxDateTime( $db)
Definition: SqlBagOStuff.php:645
Wikimedia\Rdbms\DBConnectionError
@newable
Definition: DBConnectionError.php:27
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:804
BagOStuff\$keyspace
string $keyspace
Default keyspace; used by makeKey()
Definition: BagOStuff.php:103
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:368
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:687
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:251
SqlBagOStuff\doDelete
doDelete( $key, $flags=0)
Delete an item.
Definition: SqlBagOStuff.php:572
Wikimedia\Rdbms\IDatabase\startAtomic
startAtomic( $fname=__METHOD__, $cancelable=self::ATOMIC_NOT_CANCELABLE)
Begin an atomic section of SQL statements.