MediaWiki  master
NameTableStore.php
Go to the documentation of this file.
1 <?php
21 namespace MediaWiki\Storage;
22 
23 use Exception;
24 use IExpiringStore;
25 use Psr\Log\LoggerInterface;
26 use WANObjectCache;
27 use Wikimedia\Assert\Assert;
31 use Wikimedia\RequestTimeout\TimeoutException;
32 
38 
40  private $loadBalancer;
41 
43  private $cache;
44 
46  private $logger;
47 
49  private $tableCache = null;
50 
52  private $domain;
53 
55  private $cacheTTL;
56 
58  private $table;
60  private $idField;
62  private $nameField;
66  private $insertCallback;
67 
86  public function __construct(
87  ILoadBalancer $dbLoadBalancer,
89  LoggerInterface $logger,
90  $table,
91  $idField,
92  $nameField,
93  callable $normalizationCallback = null,
94  $dbDomain = false,
95  callable $insertCallback = null
96  ) {
97  $this->loadBalancer = $dbLoadBalancer;
98  $this->cache = $cache;
99  $this->logger = $logger;
100  $this->table = $table;
101  $this->idField = $idField;
102  $this->nameField = $nameField;
103  $this->normalizationCallback = $normalizationCallback;
104  $this->domain = $dbDomain;
105  $this->cacheTTL = IExpiringStore::TTL_MONTH;
106  $this->insertCallback = $insertCallback;
107  }
108 
115  private function getDBConnection( $index, $flags = 0 ) {
116  return $this->loadBalancer->getConnectionRef( $index, [], $this->domain, $flags );
117  }
118 
127  private function getCacheKey() {
128  return $this->cache->makeGlobalKey(
129  'NameTableSqlStore',
130  $this->table,
131  $this->loadBalancer->resolveDomainID( $this->domain )
132  );
133  }
134 
139  private function normalizeName( $name ) {
140  if ( $this->normalizationCallback === null ) {
141  return $name;
142  }
143  return call_user_func( $this->normalizationCallback, $name );
144  }
145 
163  public function acquireId( string $name ) {
164  $name = $this->normalizeName( $name );
165 
167  $searchResult = array_search( $name, $table, true );
168  if ( $searchResult === false ) {
169  $id = $this->store( $name );
170  if ( $id === null ) {
171  // RACE: $name was already in the db, probably just inserted, so load from primary DB.
172  // Use DBO_TRX to avoid missing inserts due to other threads or REPEATABLE-READs.
174 
175  $searchResult = array_search( $name, $table, true );
176  if ( $searchResult === false ) {
177  // Insert failed due to IGNORE flag, but DB_PRIMARY didn't give us the data
178  $m = "No insert possible but primary DB didn't give us a record for " .
179  "'{$name}' in '{$this->table}'";
180  $this->logger->error( $m );
181  throw new NameTableAccessException( $m );
182  }
183  } else {
184  if ( isset( $table[$id] ) ) {
185  // This can happen when a transaction is rolled back and acquireId is called in
186  // an onTransactionResolution() callback, which gets executed before retryStore()
187  // has a chance to run. The right thing to do in this case is to discard the old
188  // value. According to the contract of acquireId, the caller should not have
189  // used it outside the transaction, so it should not be persisted anywhere after
190  // the rollback.
191  $m = "Got ID $id for '$name' from insert"
192  . " into '{$this->table}', but ID $id was previously associated with"
193  . " the name '{$table[$id]}'. Overriding the old value, which presumably"
194  . " has been removed from the database due to a transaction rollback.";
195 
196  $this->logger->warning( $m );
197  }
198 
199  $table[$id] = $name;
200  $searchResult = $id;
201 
202  // As store returned an ID we know we inserted so delete from WAN cache
203  $dbw = $this->getDBConnection( DB_PRIMARY );
204  $dbw->onTransactionPreCommitOrIdle( function () {
205  $this->cache->delete( $this->getCacheKey() );
206  }, __METHOD__ );
207  }
208  $this->tableCache = $table;
209  }
210 
211  return $searchResult;
212  }
213 
226  public function reloadMap( $connFlags = 0 ) {
227  if ( $connFlags !== 0 && defined( 'MW_PHPUNIT_TEST' ) ) {
228  // HACK: We can't use $connFlags while doing PHPUnit tests, because the
229  // fake database tables are bound to a single connection.
230  $connFlags = 0;
231  }
232 
233  $dbw = $this->getDBConnection( DB_PRIMARY, $connFlags );
234  $this->tableCache = $this->loadTable( $dbw );
235  $dbw->onTransactionPreCommitOrIdle( function () {
236  $this->cache->reap( $this->getCacheKey(), INF );
237  }, __METHOD__ );
238 
239  return $this->tableCache;
240  }
241 
252  public function getId( string $name ) {
253  $name = $this->normalizeName( $name );
254 
256  $searchResult = array_search( $name, $table, true );
257 
258  if ( $searchResult !== false ) {
259  return $searchResult;
260  }
261 
262  throw NameTableAccessException::newFromDetails( $this->table, 'name', $name );
263  }
264 
276  public function getName( int $id ) {
278  if ( array_key_exists( $id, $table ) ) {
279  return $table[$id];
280  }
281  $fname = __METHOD__;
282 
283  $table = $this->cache->getWithSetCallback(
284  $this->getCacheKey(),
285  $this->cacheTTL,
286  function ( $oldValue, &$ttl, &$setOpts ) use ( $id, $fname ) {
287  // Check if cached value is up-to-date enough to have $id
288  if ( is_array( $oldValue ) && array_key_exists( $id, $oldValue ) ) {
289  // Completely leave the cache key alone
290  $ttl = WANObjectCache::TTL_UNCACHEABLE;
291  // Use the old value
292  return $oldValue;
293  }
294  // Regenerate from replica DB, and primary DB if needed
295  foreach ( [ DB_REPLICA, DB_PRIMARY ] as $source ) {
296  // Log a fallback to primary
297  if ( $source === DB_PRIMARY ) {
298  $this->logger->info(
299  $fname . ' falling back to primary select from ' .
300  $this->table . ' with id ' . $id
301  );
302  }
303  $db = $this->getDBConnection( $source );
304  $cacheSetOpts = Database::getCacheSetOptions( $db );
305  $table = $this->loadTable( $db );
306  if ( array_key_exists( $id, $table ) ) {
307  break; // found it
308  }
309  }
310  // Use the value from last source checked
311  $setOpts += $cacheSetOpts;
312 
313  return $table;
314  },
315  [ 'minAsOf' => INF ] // force callback run
316  );
317 
318  $this->tableCache = $table;
319 
320  if ( array_key_exists( $id, $table ) ) {
321  return $table[$id];
322  }
323 
324  throw NameTableAccessException::newFromDetails( $this->table, 'id', $id );
325  }
326 
334  public function getMap() {
335  return $this->getTableFromCachesOrReplica();
336  }
337 
341  private function getTableFromCachesOrReplica() {
342  if ( $this->tableCache !== null ) {
343  return $this->tableCache;
344  }
345 
346  $table = $this->cache->getWithSetCallback(
347  $this->getCacheKey(),
348  $this->cacheTTL,
349  function ( $oldValue, &$ttl, &$setOpts ) {
350  $dbr = $this->getDBConnection( DB_REPLICA );
351  $setOpts += Database::getCacheSetOptions( $dbr );
352  return $this->loadTable( $dbr );
353  }
354  );
355 
356  $this->tableCache = $table;
357 
358  return $table;
359  }
360 
368  private function loadTable( IDatabase $db ) {
369  $result = $db->select(
370  $this->table,
371  [
372  'id' => $this->idField,
373  'name' => $this->nameField
374  ],
375  [],
376  __METHOD__,
377  [ 'ORDER BY' => 'id' ]
378  );
379 
380  $assocArray = [];
381  foreach ( $result as $row ) {
382  $assocArray[$row->id] = $row->name;
383  }
384 
385  return $assocArray;
386  }
387 
394  private function store( string $name ) {
395  Assert::parameter( $name !== '', '$name', 'should not be an empty string' );
396  // Note: this is only called internally so normalization of $name has already occurred.
397 
398  $dbw = $this->getDBConnection( DB_PRIMARY );
399 
400  $id = null;
401  $dbw->doAtomicSection(
402  __METHOD__,
403  function ( IDatabase $unused, $fname )
404  use ( $name, &$id, $dbw ) {
405  // NOTE: use IDatabase from the parent scope here, not the function parameter.
406  // If $dbw is a wrapper around the actual DB, we need to call the wrapper here,
407  // not the inner instance.
408  $dbw->insert(
409  $this->table,
410  $this->getFieldsToStore( $name ),
411  $fname,
412  [ 'IGNORE' ]
413  );
414 
415  if ( $dbw->affectedRows() === 0 ) {
416  $this->logger->info(
417  'Tried to insert name into table ' . $this->table . ', but value already existed.'
418  );
419 
420  return;
421  }
422 
423  $id = $dbw->insertId();
424 
425  // Any open transaction may still be rolled back. If that happens, we have to re-try the
426  // insertion and restore a consistent state of the cached table.
427  $dbw->onAtomicSectionCancel(
428  function ( $trigger, IDatabase $unused ) use ( $name, $id, $dbw ) {
429  $this->retryStore( $dbw, $name, $id );
430  },
431  $fname );
432  },
433  IDatabase::ATOMIC_CANCELABLE
434  );
435 
436  return $id;
437  }
438 
447  private function retryStore( IDatabase $dbw, $name, $id ) {
448  // NOTE: in the closure below, use the IDatabase from the original method call,
449  // not the one passed to the closure as a parameter.
450  // If $dbw is a wrapper around the actual DB, we need to call the wrapper,
451  // not the inner instance.
452 
453  try {
454  $dbw->doAtomicSection(
455  __METHOD__,
456  function ( IDatabase $unused, $fname ) use ( $name, $id, $dbw ) {
457  // Try to insert a row with the ID we originally got.
458  // If that fails (because of a key conflict), we will just try to get another ID again later.
459  $dbw->insert(
460  $this->table,
461  $this->getFieldsToStore( $name, $id ),
462  $fname
463  );
464 
465  // Make sure we re-load the map in case this gets rolled back again.
466  // We could re-try once more, but that bears the risk of an infinite loop.
467  // So let's just give up on the ID.
468  $dbw->onAtomicSectionCancel(
469  function ( $trigger, IDatabase $unused ) {
470  $this->logger->warning(
471  'Re-insertion of name into table ' . $this->table
472  . ' was rolled back. Giving up and reloading the cache.'
473  );
475  },
476  $fname
477  );
478 
479  $this->logger->info(
480  'Re-insert name into table ' . $this->table . ' after failed transaction.'
481  );
482  },
483  IDatabase::ATOMIC_CANCELABLE
484  );
485  } catch ( TimeoutException $e ) {
486  throw $e;
487  } catch ( Exception $ex ) {
488  $this->logger->error(
489  'Re-insertion of name into table ' . $this->table . ' failed: ' . $ex->getMessage()
490  );
491  } finally {
492  // NOTE: we reload regardless of whether the above insert succeeded. There is
493  // only three possibilities: the insert succeeded, so the new map will have
494  // the desired $id/$name mapping. Or the insert failed because another
495  // process already inserted that same $id/$name mapping, in which case the
496  // new map will also have it. Or another process grabbed the desired ID for
497  // another name, or the database refuses to insert the given ID into the
498  // auto increment field - in that case, the new map will not have a mapping
499  // for $name (or has a different mapping for $name). In that last case, we can
500  // only hope that the ID produced within the failed transaction has not been
501  // used outside that transaction.
502 
504  }
505  }
506 
512  private function getFieldsToStore( $name, $id = null ) {
513  $fields = [];
514 
515  $fields[$this->nameField] = $name;
516 
517  if ( $id !== null ) {
518  $fields[$this->idField] = $id;
519  }
520 
521  if ( $this->insertCallback !== null ) {
522  $fields = call_user_func( $this->insertCallback, $fields );
523  }
524  return $fields;
525  }
526 
527 }
Exception representing a failure to look up a row from a name table.
static newFromDetails( $tableName, $accessType, $accessValue)
acquireId(string $name)
Acquire the id of the given name.
store(string $name)
Stores the given name in the DB, returning the ID when an insert occurs.
getId(string $name)
Get the id of the given name.
getMap()
Get the whole table, in no particular order as a map of ids to names.
getName(int $id)
Get the name of the given id.
retryStore(IDatabase $dbw, $name, $id)
After the initial insertion got rolled back, this can be used to try the insertion again,...
loadTable(IDatabase $db)
Gets the table from the db.
getCacheKey()
Gets the cache key for names.
reloadMap( $connFlags=0)
Reloads the name table from the primary database, and purges the WAN cache entry.
__construct(ILoadBalancer $dbLoadBalancer, WANObjectCache $cache, LoggerInterface $logger, $table, $idField, $nameField, callable $normalizationCallback=null, $dbDomain=false, callable $insertCallback=null)
Multi-datacenter aware caching interface.
static getCacheSetOptions(?IDatabase ... $dbs)
Merge the result of getSessionLagStatus() for several DBs using the most pessimistic values to estima...
Definition: Database.php:4295
Generic interface providing TTL constants for lightweight expiring object stores.
Basic database interface for live and lazy-loaded relation database handles.
Definition: IDatabase.php:39
doAtomicSection( $fname, callable $callback, $cancelable=self::ATOMIC_NOT_CANCELABLE)
Perform an atomic section of reversible SQL statements from a callback.
select( $table, $vars, $conds='', $fname=__METHOD__, $options=[], $join_conds=[])
Execute a SELECT query constructed using the various parameters provided.
onAtomicSectionCancel(callable $callback, $fname=__METHOD__)
Run a callback when the atomic section is cancelled.
insert( $table, $rows, $fname=__METHOD__, $options=[])
Insert row(s) into a table, in the provided order.
Database cluster connection, tracking, load balancing, and transaction manager interface.
const CONN_TRX_AUTOCOMMIT
DB handle should have DBO_TRX disabled and the caller will leave it as such.
$source
const DB_REPLICA
Definition: defines.php:26
const DB_PRIMARY
Definition: defines.php:28