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  $value = $this->unserialize( $blob );
269  if ( $getToken && $value !== false ) {
270  $casToken = $blob;
271  }
272 
273  return $value;
274  }
275 
276  return false;
277  }
278 
279  protected function doGetMulti( array $keys, $flags = 0 ) {
280  $values = [];
281 
282  $blobs = $this->fetchBlobMulti( $keys );
283  foreach ( $blobs as $key => $blob ) {
284  $values[$key] = $this->unserialize( $blob );
285  }
286 
287  return $values;
288  }
289 
290  private function fetchBlobMulti( array $keys ) {
291  $values = []; // array of (key => value)
292 
293  $keysByTableByShardIndex = [];
294  foreach ( $keys as $key ) {
295  list( $shardIndex, $tableName ) = $this->getKeyLocation( $key );
296  $keysByTableByShardIndex[$shardIndex][$tableName][] = $key;
297  }
298 
299  $dataRows = [];
300  foreach ( $keysByTableByShardIndex as $shardIndex => $serverKeys ) {
301  try {
302  $db = $this->getConnection( $shardIndex );
303  foreach ( $serverKeys as $tableName => $tableKeys ) {
304  $res = $db->select(
305  $tableName,
306  [ 'keyname', 'value', 'exptime' ],
307  [ 'keyname' => $tableKeys ],
308  __METHOD__,
309  // Approximate write-on-the-fly BagOStuff API via blocking.
310  // This approximation fails if a ROLLBACK happens (which is rare).
311  // We do not want to flush the TRX as that can break callers.
312  $db->trxLevel() ? [ 'LOCK IN SHARE MODE' ] : []
313  );
314  if ( $res === false ) {
315  continue;
316  }
317  foreach ( $res as $row ) {
318  $row->shardIndex = $shardIndex;
319  $row->tableName = $tableName;
320  $dataRows[$row->keyname] = $row;
321  }
322  }
323  } catch ( DBError $e ) {
324  $this->handleReadError( $e, $shardIndex );
325  }
326  }
327 
328  foreach ( $keys as $key ) {
329  if ( isset( $dataRows[$key] ) ) { // HIT?
330  $row = $dataRows[$key];
331  $this->debug( "get: retrieved data; expiry time is " . $row->exptime );
332  $db = null; // in case of connection failure
333  try {
334  $db = $this->getConnection( $row->shardIndex );
335  if ( $this->isExpired( $db, $row->exptime ) ) { // MISS
336  $this->debug( "get: key has expired" );
337  } else { // HIT
338  $values[$key] = $db->decodeBlob( $row->value );
339  }
340  } catch ( DBQueryError $e ) {
341  $this->handleWriteError( $e, $db, $row->shardIndex );
342  }
343  } else { // MISS
344  $this->debug( 'get: no matching rows' );
345  }
346  }
347 
348  return $values;
349  }
350 
351  protected function doSetMulti( array $data, $exptime = 0, $flags = 0 ) {
352  return $this->modifyMulti( $data, $exptime, $flags, self::$OP_SET );
353  }
354 
362  private function modifyMulti( array $data, $exptime, $flags, $op ) {
363  $keysByTableByShardIndex = [];
364  foreach ( $data as $key => $value ) {
365  list( $shardIndex, $tableName ) = $this->getKeyLocation( $key );
366  $keysByTableByShardIndex[$shardIndex][$tableName][] = $key;
367  }
368 
369  $exptime = $this->getExpirationAsTimestamp( $exptime );
370 
371  $result = true;
373  $silenceScope = $this->silenceTransactionProfiler();
374  foreach ( $keysByTableByShardIndex as $shardIndex => $serverKeys ) {
375  $db = null; // in case of connection failure
376  try {
377  $db = $this->getConnection( $shardIndex );
378  $this->occasionallyGarbageCollect( $db ); // expire old entries if any
379  $dbExpiry = $exptime ? $db->timestamp( $exptime ) : $this->getMaxDateTime( $db );
380  } catch ( DBError $e ) {
381  $this->handleWriteError( $e, $db, $shardIndex );
382  $result = false;
383  continue;
384  }
385 
386  foreach ( $serverKeys as $tableName => $tableKeys ) {
387  try {
388  $result = $this->updateTable(
389  $op,
390  $db,
391  $tableName,
392  $tableKeys,
393  $data,
394  $dbExpiry
395  ) && $result;
396  } catch ( DBError $e ) {
397  $this->handleWriteError( $e, $db, $shardIndex );
398  $result = false;
399  }
400 
401  }
402  }
403 
404  if ( $this->fieldHasFlags( $flags, self::WRITE_SYNC ) ) {
405  foreach ( $keysByTableByShardIndex as $shardIndex => $unused ) {
406  $result = $this->waitForReplication( $shardIndex ) && $result;
407  }
408  }
409 
410  return $result;
411  }
412 
424  private function updateTable( $op, $db, $table, $tableKeys, $data, $dbExpiry ) {
425  $success = true;
426 
427  if ( $op === self::$OP_ADD ) {
428  $rows = [];
429  foreach ( $tableKeys as $key ) {
430  $rows[] = [
431  'keyname' => $key,
432  'value' => $db->encodeBlob( $this->serialize( $data[$key] ) ),
433  'exptime' => $dbExpiry
434  ];
435  }
436  $db->delete(
437  $table,
438  [
439  'keyname' => $tableKeys,
440  'exptime <= ' . $db->addQuotes( $db->timestamp() )
441  ],
442  __METHOD__
443  );
444  $db->insert( $table, $rows, __METHOD__, [ 'IGNORE' ] );
445 
446  $success = ( $db->affectedRows() == count( $rows ) );
447  } elseif ( $op === self::$OP_SET ) {
448  $rows = [];
449  foreach ( $tableKeys as $key ) {
450  $rows[] = [
451  'keyname' => $key,
452  'value' => $db->encodeBlob( $this->serialize( $data[$key] ) ),
453  'exptime' => $dbExpiry
454  ];
455  }
456  $db->replace( $table, 'keyname', $rows, __METHOD__ );
457  } elseif ( $op === self::$OP_DELETE ) {
458  $db->delete( $table, [ 'keyname' => $tableKeys ], __METHOD__ );
459  } elseif ( $op === self::$OP_TOUCH ) {
460  $db->update(
461  $table,
462  [ 'exptime' => $dbExpiry ],
463  [
464  'keyname' => $tableKeys,
465  'exptime > ' . $db->addQuotes( $db->timestamp() )
466  ],
467  __METHOD__
468  );
469 
470  $success = ( $db->affectedRows() == count( $tableKeys ) );
471  } else {
472  throw new InvalidArgumentException( "Invalid operation '$op'" );
473  }
474 
475  return $success;
476  }
477 
478  protected function doSet( $key, $value, $exptime = 0, $flags = 0 ) {
479  return $this->modifyMulti( [ $key => $value ], $exptime, $flags, self::$OP_SET );
480  }
481 
482  protected function doAdd( $key, $value, $exptime = 0, $flags = 0 ) {
483  return $this->modifyMulti( [ $key => $value ], $exptime, $flags, self::$OP_ADD );
484  }
485 
486  protected function doCas( $casToken, $key, $value, $exptime = 0, $flags = 0 ) {
487  list( $shardIndex, $tableName ) = $this->getKeyLocation( $key );
488  $exptime = $this->getExpirationAsTimestamp( $exptime );
489 
491  $silenceScope = $this->silenceTransactionProfiler();
492  $db = null; // in case of connection failure
493  try {
494  $db = $this->getConnection( $shardIndex );
495  // (T26425) use a replace if the db supports it instead of
496  // delete/insert to avoid clashes with conflicting keynames
497  $db->update(
498  $tableName,
499  [
500  'keyname' => $key,
501  'value' => $db->encodeBlob( $this->serialize( $value ) ),
502  'exptime' => $exptime
503  ? $db->timestamp( $exptime )
504  : $this->getMaxDateTime( $db )
505  ],
506  [
507  'keyname' => $key,
508  'value' => $db->encodeBlob( $casToken ),
509  'exptime > ' . $db->addQuotes( $db->timestamp() )
510  ],
511  __METHOD__
512  );
513  } catch ( DBQueryError $e ) {
514  $this->handleWriteError( $e, $db, $shardIndex );
515 
516  return false;
517  }
518 
519  $success = (bool)$db->affectedRows();
520  if ( $this->fieldHasFlags( $flags, self::WRITE_SYNC ) ) {
521  $success = $this->waitForReplication( $shardIndex ) && $success;
522  }
523 
524  return $success;
525  }
526 
527  protected function doDeleteMulti( array $keys, $flags = 0 ) {
528  return $this->modifyMulti(
529  array_fill_keys( $keys, null ),
530  0,
531  $flags,
532  self::$OP_DELETE
533  );
534  }
535 
536  protected function doDelete( $key, $flags = 0 ) {
537  return $this->modifyMulti( [ $key => null ], 0, $flags, self::$OP_DELETE );
538  }
539 
540  public function incr( $key, $step = 1, $flags = 0 ) {
541  list( $shardIndex, $tableName ) = $this->getKeyLocation( $key );
542 
543  $newCount = false;
545  $silenceScope = $this->silenceTransactionProfiler();
546  $db = null; // in case of connection failure
547  try {
548  $db = $this->getConnection( $shardIndex );
549  $encTimestamp = $db->addQuotes( $db->timestamp() );
550  $db->update(
551  $tableName,
552  [ 'value = value + ' . (int)$step ],
553  [ 'keyname' => $key, "exptime > $encTimestamp" ],
554  __METHOD__
555  );
556  if ( $db->affectedRows() > 0 ) {
557  $newValue = $db->selectField(
558  $tableName,
559  'value',
560  [ 'keyname' => $key, "exptime > $encTimestamp" ],
561  __METHOD__
562  );
563  if ( $this->isInteger( $newValue ) ) {
564  $newCount = (int)$newValue;
565  }
566  }
567  } catch ( DBError $e ) {
568  $this->handleWriteError( $e, $db, $shardIndex );
569  }
570 
571  return $newCount;
572  }
573 
574  public function decr( $key, $value = 1, $flags = 0 ) {
575  return $this->incr( $key, -$value, $flags );
576  }
577 
578  public function changeTTLMulti( array $keys, $exptime, $flags = 0 ) {
579  return $this->modifyMulti(
580  array_fill_keys( $keys, null ),
581  $exptime,
582  $flags,
583  self::$OP_TOUCH
584  );
585  }
586 
587  protected function doChangeTTL( $key, $exptime, $flags ) {
588  return $this->modifyMulti( [ $key => null ], $exptime, $flags, self::$OP_TOUCH );
589  }
590 
596  private function isExpired( IDatabase $db, $exptime ) {
597  return (
598  $exptime != $this->getMaxDateTime( $db ) &&
599  ConvertibleTimestamp::convert( TS_UNIX, $exptime ) < $this->getCurrentTime()
600  );
601  }
602 
607  private function getMaxDateTime( $db ) {
608  if ( (int)$this->getCurrentTime() > 0x7fffffff ) {
609  return $db->timestamp( 1 << 62 );
610  } else {
611  return $db->timestamp( 0x7fffffff );
612  }
613  }
614 
619  private function occasionallyGarbageCollect( IDatabase $db ) {
620  if (
621  // Random purging is enabled
622  $this->purgePeriod &&
623  // Only purge on one in every $this->purgePeriod writes
624  mt_rand( 0, $this->purgePeriod - 1 ) == 0 &&
625  // Avoid repeating the delete within a few seconds
626  ( $this->getCurrentTime() - $this->lastGarbageCollect ) > self::$GC_DELAY_SEC
627  ) {
628  $garbageCollector = function () use ( $db ) {
630  $db, $this->getCurrentTime(),
631  null,
632  $this->purgeLimit
633  );
634  $this->lastGarbageCollect = time();
635  };
636  if ( $this->asyncHandler ) {
637  $this->lastGarbageCollect = $this->getCurrentTime(); // avoid duplicate enqueues
638  ( $this->asyncHandler )( $garbageCollector );
639  } else {
640  $garbageCollector();
641  }
642  }
643  }
644 
645  public function expireAll() {
646  $this->deleteObjectsExpiringBefore( $this->getCurrentTime() );
647  }
648 
650  $timestamp,
651  callable $progress = null,
652  $limit = INF
653  ) {
655  $silenceScope = $this->silenceTransactionProfiler();
656 
657  $shardIndexes = $this->getServerShardIndexes();
658  shuffle( $shardIndexes );
659 
660  $ok = true;
661 
662  $keysDeletedCount = 0;
663  foreach ( $shardIndexes as $numServersDone => $shardIndex ) {
664  $db = null; // in case of connection failure
665  try {
666  $db = $this->getConnection( $shardIndex );
668  $db,
669  $timestamp,
670  $progress,
671  $limit,
672  $numServersDone,
673  $keysDeletedCount
674  );
675  } catch ( DBError $e ) {
676  $this->handleWriteError( $e, $db, $shardIndex );
677  $ok = false;
678  }
679  }
680 
681  return $ok;
682  }
683 
694  IDatabase $db,
695  $timestamp,
696  $progressCallback,
697  $limit,
698  $serversDoneCount = 0,
699  &$keysDeletedCount = 0
700  ) {
701  $cutoffUnix = ConvertibleTimestamp::convert( TS_UNIX, $timestamp );
702  $tableIndexes = range( 0, $this->numTableShards - 1 );
703  shuffle( $tableIndexes );
704 
705  foreach ( $tableIndexes as $numShardsDone => $tableIndex ) {
706  $continue = null; // last exptime
707  $lag = null; // purge lag
708  do {
709  $res = $db->select(
710  $this->getTableNameByShard( $tableIndex ),
711  [ 'keyname', 'exptime' ],
712  array_merge(
713  [ 'exptime < ' . $db->addQuotes( $db->timestamp( $cutoffUnix ) ) ],
714  $continue ? [ 'exptime >= ' . $db->addQuotes( $continue ) ] : []
715  ),
716  __METHOD__,
717  [ 'LIMIT' => min( $limit, 100 ), 'ORDER BY' => 'exptime' ]
718  );
719 
720  if ( $res->numRows() ) {
721  $row = $res->current();
722  if ( $lag === null ) {
723  $rowExpUnix = ConvertibleTimestamp::convert( TS_UNIX, $row->exptime );
724  $lag = max( $cutoffUnix - $rowExpUnix, 1 );
725  }
726 
727  $keys = [];
728  foreach ( $res as $row ) {
729  $keys[] = $row->keyname;
730  $continue = $row->exptime;
731  }
732 
733  $db->delete(
734  $this->getTableNameByShard( $tableIndex ),
735  [
736  'exptime < ' . $db->addQuotes( $db->timestamp( $cutoffUnix ) ),
737  'keyname' => $keys
738  ],
739  __METHOD__
740  );
741  $keysDeletedCount += $db->affectedRows();
742  }
743 
744  if ( is_callable( $progressCallback ) ) {
745  if ( $lag ) {
746  $continueUnix = ConvertibleTimestamp::convert( TS_UNIX, $continue );
747  $remainingLag = $cutoffUnix - $continueUnix;
748  $processedLag = max( $lag - $remainingLag, 0 );
749  $doneRatio =
750  ( $numShardsDone + $processedLag / $lag ) / $this->numTableShards;
751  } else {
752  $doneRatio = 1;
753  }
754 
755  $overallRatio = ( $doneRatio / $this->numServerShards )
756  + ( $serversDoneCount / $this->numServerShards );
757  call_user_func( $progressCallback, $overallRatio * 100 );
758  }
759  } while ( $res->numRows() && $keysDeletedCount < $limit );
760  }
761  }
762 
768  public function deleteAll() {
770  $silenceScope = $this->silenceTransactionProfiler();
771  foreach ( $this->getServerShardIndexes() as $shardIndex ) {
772  $db = null; // in case of connection failure
773  try {
774  $db = $this->getConnection( $shardIndex );
775  for ( $i = 0; $i < $this->numTableShards; $i++ ) {
776  $db->delete( $this->getTableNameByShard( $i ), '*', __METHOD__ );
777  }
778  } catch ( DBError $e ) {
779  $this->handleWriteError( $e, $db, $shardIndex );
780  return false;
781  }
782  }
783  return true;
784  }
785 
786  public function lock( $key, $timeout = 6, $expiry = 6, $rclass = '' ) {
787  // Avoid deadlocks and allow lock reentry if specified
788  if ( isset( $this->locks[$key] ) ) {
789  if ( $rclass != '' && $this->locks[$key]['class'] === $rclass ) {
790  ++$this->locks[$key]['depth'];
791  return true;
792  } else {
793  return false;
794  }
795  }
796 
797  list( $shardIndex ) = $this->getKeyLocation( $key );
798 
799  $db = null; // in case of connection failure
800  try {
801  $db = $this->getConnection( $shardIndex );
802  $ok = $db->lock( $key, __METHOD__, $timeout );
803  if ( $ok ) {
804  $this->locks[$key] = [ 'class' => $rclass, 'depth' => 1 ];
805  }
806 
807  $this->logger->warning(
808  __METHOD__ . " failed due to timeout for {key}.",
809  [ 'key' => $key, 'timeout' => $timeout ]
810  );
811 
812  return $ok;
813  } catch ( DBError $e ) {
814  $this->handleWriteError( $e, $db, $shardIndex );
815  $ok = false;
816  }
817 
818  return $ok;
819  }
820 
821  public function unlock( $key ) {
822  if ( !isset( $this->locks[$key] ) ) {
823  return false;
824  }
825 
826  if ( --$this->locks[$key]['depth'] <= 0 ) {
827  unset( $this->locks[$key] );
828 
829  list( $shardIndex ) = $this->getKeyLocation( $key );
830 
831  $db = null; // in case of connection failure
832  try {
833  $db = $this->getConnection( $shardIndex );
834  $ok = $db->unlock( $key, __METHOD__ );
835  if ( !$ok ) {
836  $this->logger->warning(
837  __METHOD__ . ' failed to release lock for {key}.',
838  [ 'key' => $key ]
839  );
840  }
841  } catch ( DBError $e ) {
842  $this->handleWriteError( $e, $db, $shardIndex );
843  $ok = false;
844  }
845 
846  return $ok;
847  }
848 
849  return true;
850  }
851 
860  public function makeKeyInternal( $keyspace, $args ) {
861  // SQL schema for 'objectcache' specifies keys as varchar(255). From that,
862  // subtract the number of characters we need for the keyspace and for
863  // the separator character needed for each argument. To handle some
864  // custom prefixes used by thing like WANObjectCache, limit to 205.
865  $keyspace = strtr( $keyspace, ' ', '_' );
866  $charsLeft = 205 - strlen( $keyspace ) - count( $args );
867  foreach ( $args as &$arg ) {
868  $arg = strtr( $arg, [
869  ' ' => '_', // Avoid unnecessary misses from pre-1.35 code
870  ':' => '%3A',
871  ] );
872 
873  // 33 = 32 characters for the MD5 + 1 for the '#' prefix.
874  if ( $charsLeft > 33 && strlen( $arg ) > $charsLeft ) {
875  $arg = '#' . md5( $arg );
876  }
877  $charsLeft -= strlen( $arg );
878  }
879 
880  if ( $charsLeft < 0 ) {
881  return $keyspace . ':BagOStuff-long-key:##' . md5( implode( ':', $args ) );
882  }
883  return $keyspace . ':' . implode( ':', $args );
884  }
885 
894  protected function serialize( $data ) {
895  if ( is_int( $data ) ) {
896  return $data;
897  }
898 
899  $serial = serialize( $data );
900  if ( function_exists( 'gzdeflate' ) ) {
901  $serial = gzdeflate( $serial );
902  }
903 
904  return $serial;
905  }
906 
912  protected function unserialize( $serial ) {
913  if ( $this->isInteger( $serial ) ) {
914  return (int)$serial;
915  }
916 
917  if ( function_exists( 'gzinflate' ) ) {
918  AtEase::suppressWarnings();
919  $decomp = gzinflate( $serial );
920  AtEase::restoreWarnings();
921 
922  if ( $decomp !== false ) {
923  $serial = $decomp;
924  }
925  }
926 
927  return unserialize( $serial );
928  }
929 
935  private function getConnectionViaLoadBalancer( $shardIndex ) {
936  $lb = ( $shardIndex === self::SHARD_LOCAL ) ? $this->localKeyLb : $this->globalKeyLb;
937  if ( $lb->getServerAttributes( $lb->getWriterIndex() )[Database::ATTR_DB_LEVEL_LOCKING] ) {
938  // Use the main connection to avoid transaction deadlocks
939  $conn = $lb->getMaintenanceConnectionRef( DB_MASTER );
940  } else {
941  // If the RDBMs has row/table/page level locking, then use separate auto-commit
942  // connection to avoid needless contention and deadlocks.
943  $conn = $lb->getMaintenanceConnectionRef(
944  $this->replicaOnly ? DB_REPLICA : DB_MASTER, [],
945  false,
946  $lb::CONN_TRX_AUTOCOMMIT
947  );
948  }
949 
950  return $conn;
951  }
952 
959  private function getConnectionFromServerInfo( $shardIndex, array $server ) {
960  if ( !isset( $this->conns[$shardIndex] ) ) {
962  $conn = Database::factory( $server['type'], array_merge(
963  $server,
964  [
965  'flags' => ( $server['flags'] ?? 0 ) & ~IDatabase::DBO_TRX,
966  'connLogger' => $this->logger,
967  'queryLogger' => $this->logger
968  ]
969  ) );
970  // Automatically create the objectcache table for sqlite as needed
971  if ( $conn->getType() === 'sqlite' && !$conn->tableExists( 'objectcache', __METHOD__ ) ) {
972  $this->initSqliteDatabase( $conn );
973  }
974  $this->conns[$shardIndex] = $conn;
975  }
976 
977  return $this->conns[$shardIndex];
978  }
979 
986  private function handleReadError( DBError $exception, $shardIndex ) {
987  if ( $exception instanceof DBConnectionError ) {
988  $this->markServerDown( $exception, $shardIndex );
989  }
990 
991  $this->setAndLogDBError( $exception );
992  }
993 
1002  private function handleWriteError( DBError $exception, $db, $shardIndex ) {
1003  if ( !( $db instanceof IDatabase ) ) {
1004  $this->markServerDown( $exception, $shardIndex );
1005  }
1006 
1007  $this->setAndLogDBError( $exception );
1008  }
1009 
1013  private function setAndLogDBError( DBError $exception ) {
1014  $this->logger->error( "DBError: {$exception->getMessage()}" );
1015  if ( $exception instanceof DBConnectionError ) {
1016  $this->setLastError( BagOStuff::ERR_UNREACHABLE );
1017  $this->logger->debug( __METHOD__ . ": ignoring connection error" );
1018  } else {
1019  $this->setLastError( BagOStuff::ERR_UNEXPECTED );
1020  $this->logger->debug( __METHOD__ . ": ignoring query error" );
1021  }
1022  }
1023 
1030  private function markServerDown( DBError $exception, $shardIndex ) {
1031  unset( $this->conns[$shardIndex] ); // bug T103435
1032 
1033  $now = $this->getCurrentTime();
1034  if ( isset( $this->connFailureTimes[$shardIndex] ) ) {
1035  if ( $now - $this->connFailureTimes[$shardIndex] >= 60 ) {
1036  unset( $this->connFailureTimes[$shardIndex] );
1037  unset( $this->connFailureErrors[$shardIndex] );
1038  } else {
1039  $this->logger->debug( __METHOD__ . ": Server #$shardIndex already down" );
1040  return;
1041  }
1042  }
1043  $this->logger->info( __METHOD__ . ": Server #$shardIndex down until " . ( $now + 60 ) );
1044  $this->connFailureTimes[$shardIndex] = $now;
1045  $this->connFailureErrors[$shardIndex] = $exception;
1046  }
1047 
1052  private function initSqliteDatabase( IMaintainableDatabase $db ) {
1053  if ( $db->tableExists( 'objectcache', __METHOD__ ) ) {
1054  return;
1055  }
1056  // Use one table for SQLite; sharding does not seem to have much benefit
1057  $db->query( "PRAGMA journal_mode=WAL", __METHOD__ ); // this is permanent
1058  $db->startAtomic( __METHOD__ ); // atomic DDL
1059  try {
1060  $encTable = $db->tableName( 'objectcache' );
1061  $encExptimeIndex = $db->addIdentifierQuotes( $db->tablePrefix() . 'exptime' );
1062  $db->query(
1063  "CREATE TABLE $encTable (\n" .
1064  " keyname BLOB NOT NULL default '' PRIMARY KEY,\n" .
1065  " value BLOB,\n" .
1066  " exptime TEXT\n" .
1067  ")",
1068  __METHOD__
1069  );
1070  $db->query( "CREATE INDEX $encExptimeIndex ON $encTable (exptime)", __METHOD__ );
1071  $db->endAtomic( __METHOD__ );
1072  } catch ( DBError $e ) {
1073  $db->rollback( __METHOD__ );
1074  throw $e;
1075  }
1076  }
1077 
1081  public function createTables() {
1082  foreach ( $this->getServerShardIndexes() as $shardIndex ) {
1083  $db = $this->getConnection( $shardIndex );
1084  if ( in_array( $db->getType(), [ 'mysql', 'postgres' ], true ) ) {
1085  for ( $i = 0; $i < $this->numTableShards; $i++ ) {
1086  $encBaseTable = $db->tableName( 'objectcache' );
1087  $encShardTable = $db->tableName( $this->getTableNameByShard( $i ) );
1088  $db->query( "CREATE TABLE $encShardTable LIKE $encBaseTable", __METHOD__ );
1089  }
1090  }
1091  }
1092  }
1093 
1097  private function getServerShardIndexes() {
1098  if ( $this->serverTags ) {
1099  // Striped array of database servers
1100  $shardIndexes = range( 0, $this->numServerShards - 1 );
1101  } else {
1102  // LoadBalancer based configuration
1103  $shardIndexes = [];
1104  if ( $this->localKeyLb ) {
1105  $shardIndexes[] = self::SHARD_LOCAL;
1106  }
1107  if ( $this->globalKeyLb ) {
1108  $shardIndexes[] = self::SHARD_GLOBAL;
1109  }
1110  }
1111 
1112  return $shardIndexes;
1113  }
1114 
1121  private function waitForReplication( $shardIndex ) {
1122  if ( is_int( $shardIndex ) ) {
1123  return true; // striped only, no LoadBalancer
1124  }
1125 
1126  $lb = ( $shardIndex === self::SHARD_LOCAL ) ? $this->localKeyLb : $this->globalKeyLb;
1127  if ( !$lb->hasStreamingReplicaServers() ) {
1128  return true;
1129  }
1130 
1131  try {
1132  // Wait for any replica DBs to catch up
1133  $masterPos = $lb->getMasterPos();
1134  if ( !$masterPos ) {
1135  return true; // not applicable
1136  }
1137 
1138  $loop = new WaitConditionLoop(
1139  function () use ( $lb, $masterPos ) {
1140  return $lb->waitForAll( $masterPos, 1 );
1141  },
1144  );
1145 
1146  return ( $loop->invoke() === $loop::CONDITION_REACHED );
1147  } catch ( DBError $e ) {
1148  $this->setAndLogDBError( $e );
1149 
1150  return false;
1151  }
1152  }
1153 
1159  private function silenceTransactionProfiler() {
1160  if ( $this->serverInfos ) {
1161  return null; // no TransactionProfiler injected anyway
1162  }
1163 
1164  $trxProfiler = Profiler::instance()->getTransactionProfiler();
1165  $oldSilenced = $trxProfiler->setSilenced( true );
1166  return new ScopedCallback( function () use ( $trxProfiler, $oldSilenced ) {
1167  $trxProfiler->setSilenced( $oldSilenced );
1168  } );
1169  }
1170 }
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:758
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:279
SqlBagOStuff\doCas
doCas( $casToken, $key, $value, $exptime=0, $flags=0)
Check and set an item.
Definition: SqlBagOStuff.php:486
SqlBagOStuff\createTables
createTables()
Create the shard tables on all databases (e.g.
Definition: SqlBagOStuff.php:1081
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:881
SqlBagOStuff\changeTTLMulti
changeTTLMulti(array $keys, $exptime, $flags=0)
Change the expiration of multiple keys that exist.
Definition: SqlBagOStuff.php:578
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:1121
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:1013
MediumSpecificBagOStuff\debug
debug( $text)
Definition: MediumSpecificBagOStuff.php:1064
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:290
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:935
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:1052
SqlBagOStuff\silenceTransactionProfiler
silenceTransactionProfiler()
Silence the transaction profiler until the return value falls out of scope.
Definition: SqlBagOStuff.php:1159
SqlBagOStuff\incr
incr( $key, $step=1, $flags=0)
Increase stored value of $key by $value while preserving its TTL.
Definition: SqlBagOStuff.php:540
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:424
$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 Stable to extend.
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:478
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: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:645
SqlBagOStuff\occasionallyGarbageCollect
occasionallyGarbageCollect(IDatabase $db)
Definition: SqlBagOStuff.php:619
SqlBagOStuff\unlock
unlock( $key)
Release an advisory lock on a key string.
Definition: SqlBagOStuff.php:821
SqlBagOStuff\deleteAll
deleteAll()
Delete content of shard tables in every server.
Definition: SqlBagOStuff.php:768
MediumSpecificBagOStuff\getExpirationAsTimestamp
getExpirationAsTimestamp( $exptime)
Convert an optionally relative timestamp to an absolute time.
Definition: MediumSpecificBagOStuff.php:841
SqlBagOStuff\$connFailureErrors
Exception[] $connFailureErrors
Map of (shard index => Exception)
Definition: SqlBagOStuff.php:72
SqlBagOStuff\doDeleteMulti
doDeleteMulti(array $keys, $flags=0)
Definition: SqlBagOStuff.php:527
$blob
$blob
Definition: testCompression.php:70
SqlBagOStuff\modifyMulti
modifyMulti(array $data, $exptime, $flags, $op)
Definition: SqlBagOStuff.php:362
BagOStuff\$asyncHandler
callable null $asyncHandler
Definition: BagOStuff.php:76
SqlBagOStuff\getConnectionFromServerInfo
getConnectionFromServerInfo( $shardIndex, array $server)
Definition: SqlBagOStuff.php:959
$args
if( $line===false) $args
Definition: mcc.php:124
SqlBagOStuff\isExpired
isExpired(IDatabase $db, $exptime)
Definition: SqlBagOStuff.php:596
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:986
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:894
SqlBagOStuff\handleWriteError
handleWriteError(DBError $exception, $db, $shardIndex)
Handle a DBQueryError which occurred during a write operation.
Definition: SqlBagOStuff.php:1002
SqlBagOStuff\$OP_SET
static string $OP_SET
Definition: SqlBagOStuff.php:78
Wikimedia\Rdbms\DBQueryError
Definition: DBQueryError.php:29
SqlBagOStuff\$globalKeyLb
ILoadBalancer null $globalKeyLb
Definition: SqlBagOStuff.php:46
SqlBagOStuff\getServerShardIndexes
getServerShardIndexes()
Definition: SqlBagOStuff.php:1097
SqlBagOStuff\doChangeTTL
doChangeTTL( $key, $exptime, $flags)
Definition: SqlBagOStuff.php:587
SqlBagOStuff\unserialize
unserialize( $serial)
Unserialize and, if necessary, decompress an object.
Definition: SqlBagOStuff.php:912
SqlBagOStuff\lock
lock( $key, $timeout=6, $expiry=6, $rclass='')
Acquire an advisory lock on a key string.
Definition: SqlBagOStuff.php:786
SqlBagOStuff\makeKeyInternal
makeKeyInternal( $keyspace, $args)
Construct a cache key.
Definition: SqlBagOStuff.php:860
SqlBagOStuff\getConnection
getConnection( $shardIndex)
Get a connection to the specified database.
Definition: SqlBagOStuff.php:186
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:1030
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:693
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:574
SqlBagOStuff\doAdd
doAdd( $key, $value, $exptime=0, $flags=0)
Insert an item if it does not already exist.
Definition: SqlBagOStuff.php:482
SqlBagOStuff\$OP_ADD
static string $OP_ADD
Definition: SqlBagOStuff.php:80
BagOStuff\fieldHasFlags
fieldHasFlags( $field, $flags)
Definition: BagOStuff.php:509
SqlBagOStuff\getMaxDateTime
getMaxDateTime( $db)
Definition: SqlBagOStuff.php:607
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:566
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:351
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:649
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:536
Wikimedia\Rdbms\IDatabase\startAtomic
startAtomic( $fname=__METHOD__, $cancelable=self::ATOMIC_NOT_CANCELABLE)
Begin an atomic section of SQL statements.