MediaWiki  master
RedisBagOStuff.php
Go to the documentation of this file.
1 <?php
35  protected $redisPool;
37  protected $servers;
39  protected $serverTagMap;
41  protected $automaticFailover;
42 
71  public function __construct( $params ) {
72  parent::__construct( $params );
73  $redisConf = [ 'serializer' => 'none' ]; // manage that in this class
74  foreach ( [ 'connectTimeout', 'persistent', 'password' ] as $opt ) {
75  if ( isset( $params[$opt] ) ) {
76  $redisConf[$opt] = $params[$opt];
77  }
78  }
79  $this->redisPool = RedisConnectionPool::singleton( $redisConf );
80 
81  $this->servers = $params['servers'];
82  foreach ( $this->servers as $key => $server ) {
83  $this->serverTagMap[is_int( $key ) ? $server : $key] = $server;
84  }
85 
86  $this->automaticFailover = $params['automaticFailover'] ?? true;
87 
88  // ...and uses rdb snapshots (redis.conf default)
90  }
91 
92  protected function doGet( $key, $flags = 0, &$casToken = null ) {
93  $getToken = ( $casToken === self::PASS_BY_REF );
94  $casToken = null;
95 
96  $conn = $this->getConnection( $key );
97  if ( !$conn ) {
98  return false;
99  }
100 
101  $e = null;
102  try {
103  $blob = $conn->get( $key );
104  if ( $getToken && $blob !== false ) {
105  $casToken = $blob;
106  }
107  $result = $this->unserialize( $blob );
108  $valueSize = strlen( $blob );
109  } catch ( RedisException $e ) {
110  $result = false;
111  $valueSize = false;
112  $this->handleException( $conn, $e );
113  }
114 
115  $this->logRequest( 'get', $key, $conn->getServer(), $e );
116 
117  $this->updateOpStats( self::METRIC_OP_GET, [ $key => [ 0, $valueSize ] ] );
118 
119  return $result;
120  }
121 
122  protected function doSet( $key, $value, $exptime = 0, $flags = 0 ) {
123  $conn = $this->getConnection( $key );
124  if ( !$conn ) {
125  return false;
126  }
127 
128  $ttl = $this->getExpirationAsTTL( $exptime );
129  $serialized = $this->getSerialized( $value, $key );
130  $valueSize = strlen( $serialized );
131 
132  $e = null;
133  try {
134  if ( $ttl ) {
135  $result = $conn->setex( $key, $ttl, $serialized );
136  } else {
137  $result = $conn->set( $key, $serialized );
138  }
139  } catch ( RedisException $e ) {
140  $result = false;
141  $this->handleException( $conn, $e );
142  }
143 
144  $this->logRequest( 'set', $key, $conn->getServer(), $e );
145 
146  $this->updateOpStats( self::METRIC_OP_SET, [ $key => [ $valueSize, 0 ] ] );
147 
148  return $result;
149  }
150 
151  protected function doDelete( $key, $flags = 0 ) {
152  $conn = $this->getConnection( $key );
153  if ( !$conn ) {
154  return false;
155  }
156 
157  $e = null;
158  try {
159  // Note that redis does not return false if the key was not there
160  $result = ( $conn->del( $key ) !== false );
161  } catch ( RedisException $e ) {
162  $result = false;
163  $this->handleException( $conn, $e );
164  }
165 
166  $this->logRequest( 'delete', $key, $conn->getServer(), $e );
167 
168  $this->updateOpStats( self::METRIC_OP_DELETE, [ $key ] );
169 
170  return $result;
171  }
172 
173  protected function doGetMulti( array $keys, $flags = 0 ) {
175  $conns = [];
176  $batches = [];
177  foreach ( $keys as $key ) {
178  $conn = $this->getConnection( $key );
179  if ( $conn ) {
180  $server = $conn->getServer();
181  $conns[$server] = $conn;
182  $batches[$server][] = $key;
183  }
184  }
185 
186  $blobsFound = [];
187  foreach ( $batches as $server => $batchKeys ) {
188  $conn = $conns[$server];
189 
190  $e = null;
191  try {
192  // Avoid mget() to reduce CPU hogging from a single request
193  $conn->multi( Redis::PIPELINE );
194  foreach ( $batchKeys as $key ) {
195  $conn->get( $key );
196  }
197  $batchResult = $conn->exec();
198  if ( $batchResult === false ) {
199  $this->logRequest( 'get', implode( ',', $batchKeys ), $server, true );
200  continue;
201  }
202 
203  foreach ( $batchResult as $i => $blob ) {
204  if ( $blob !== false ) {
205  $blobsFound[$batchKeys[$i]] = $blob;
206  }
207  }
208  } catch ( RedisException $e ) {
209  $this->handleException( $conn, $e );
210  }
211 
212  $this->logRequest( 'get', implode( ',', $batchKeys ), $server, $e );
213  }
214 
215  // Preserve the order of $keys
216  $result = [];
217  $valueSizesByKey = [];
218  foreach ( $keys as $key ) {
219  if ( array_key_exists( $key, $blobsFound ) ) {
220  $blob = $blobsFound[$key];
221  $value = $this->unserialize( $blob );
222  if ( $value !== false ) {
223  $result[$key] = $value;
224  }
225  $valueSize = strlen( $blob );
226  } else {
227  $valueSize = false;
228  }
229  $valueSizesByKey[$key] = [ 0, $valueSize ];
230  }
231 
232  $this->updateOpStats( self::METRIC_OP_GET, $valueSizesByKey );
233 
234  return $result;
235  }
236 
237  protected function doSetMulti( array $data, $exptime = 0, $flags = 0 ) {
238  $result = true;
239 
241  $conns = [];
242  $batches = [];
243  foreach ( $data as $key => $value ) {
244  $conn = $this->getConnection( $key );
245  if ( $conn ) {
246  $server = $conn->getServer();
247  $conns[$server] = $conn;
248  $batches[$server][] = $key;
249  } else {
250  $result = false;
251  }
252  }
253 
254  $ttl = $this->getExpirationAsTTL( $exptime );
255  $op = $ttl ? 'setex' : 'set';
256 
257  $valueSizesByKey = [];
258  foreach ( $batches as $server => $batchKeys ) {
259  $conn = $conns[$server];
260 
261  $e = null;
262  try {
263  // Avoid mset() to reduce CPU hogging from a single request
264  $conn->multi( Redis::PIPELINE );
265  foreach ( $batchKeys as $key ) {
266  $serialized = $this->getSerialized( $data[$key], $key );
267  if ( $ttl ) {
268  $conn->setex( $key, $ttl, $serialized );
269  } else {
270  $conn->set( $key, $serialized );
271  }
272  $valueSizesByKey[$key] = [ strlen( $serialized ), 0 ];
273  }
274  $batchResult = $conn->exec();
275  if ( $batchResult === false ) {
276  $result = false;
277  $this->logRequest( $op, implode( ',', $batchKeys ), $server, true );
278  continue;
279  }
280 
281  $result = $result && !in_array( false, $batchResult, true );
282  } catch ( RedisException $e ) {
283  $this->handleException( $conn, $e );
284  $result = false;
285  }
286 
287  $this->logRequest( $op, implode( ',', $batchKeys ), $server, $e );
288  }
289 
290  $this->updateOpStats( self::METRIC_OP_SET, $valueSizesByKey );
291 
292  return $result;
293  }
294 
295  protected function doDeleteMulti( array $keys, $flags = 0 ) {
296  $result = true;
297 
299  $conns = [];
300  $batches = [];
301  foreach ( $keys as $key ) {
302  $conn = $this->getConnection( $key );
303  if ( $conn ) {
304  $server = $conn->getServer();
305  $conns[$server] = $conn;
306  $batches[$server][] = $key;
307  } else {
308  $result = false;
309  }
310  }
311 
312  foreach ( $batches as $server => $batchKeys ) {
313  $conn = $conns[$server];
314 
315  $e = null;
316  try {
317  // Avoid delete() with array to reduce CPU hogging from a single request
318  $conn->multi( Redis::PIPELINE );
319  foreach ( $batchKeys as $key ) {
320  $conn->del( $key );
321  }
322  $batchResult = $conn->exec();
323  if ( $batchResult === false ) {
324  $result = false;
325  $this->logRequest( 'delete', implode( ',', $batchKeys ), $server, true );
326  continue;
327  }
328  // Note that redis does not return false if the key was not there
329  $result = $result && !in_array( false, $batchResult, true );
330  } catch ( RedisException $e ) {
331  $this->handleException( $conn, $e );
332  $result = false;
333  }
334 
335  $this->logRequest( 'delete', implode( ',', $batchKeys ), $server, $e );
336  }
337 
338  $this->updateOpStats( self::METRIC_OP_DELETE, array_values( $keys ) );
339 
340  return $result;
341  }
342 
343  public function doChangeTTLMulti( array $keys, $exptime, $flags = 0 ) {
344  $result = true;
345 
347  $conns = [];
348  $batches = [];
349  foreach ( $keys as $key ) {
350  $conn = $this->getConnection( $key );
351  if ( $conn ) {
352  $server = $conn->getServer();
353  $conns[$server] = $conn;
354  $batches[$server][] = $key;
355  } else {
356  $result = false;
357  }
358  }
359 
360  $relative = $this->isRelativeExpiration( $exptime );
361  $op = ( $exptime == self::TTL_INDEFINITE )
362  ? 'persist'
363  : ( $relative ? 'expire' : 'expireAt' );
364 
365  foreach ( $batches as $server => $batchKeys ) {
366  $conn = $conns[$server];
367 
368  $e = null;
369  try {
370  $conn->multi( Redis::PIPELINE );
371  foreach ( $batchKeys as $key ) {
372  if ( $exptime == self::TTL_INDEFINITE ) {
373  $conn->persist( $key );
374  } elseif ( $relative ) {
375  $conn->expire( $key, $this->getExpirationAsTTL( $exptime ) );
376  } else {
377  $conn->expireAt( $key, $this->getExpirationAsTimestamp( $exptime ) );
378  }
379  }
380  $batchResult = $conn->exec();
381  if ( $batchResult === false ) {
382  $result = false;
383  $this->logRequest( $op, implode( ',', $batchKeys ), $server, true );
384  continue;
385  }
386  $result = in_array( false, $batchResult, true ) ? false : $result;
387  } catch ( RedisException $e ) {
388  $this->handleException( $conn, $e );
389  $result = false;
390  }
391 
392  $this->logRequest( $op, implode( ',', $batchKeys ), $server, $e );
393  }
394 
395  $this->updateOpStats( self::METRIC_OP_CHANGE_TTL, array_values( $keys ) );
396 
397  return $result;
398  }
399 
400  protected function doAdd( $key, $value, $expiry = 0, $flags = 0 ) {
401  $conn = $this->getConnection( $key );
402  if ( !$conn ) {
403  return false;
404  }
405 
406  $ttl = $this->getExpirationAsTTL( $expiry );
407  $serialized = $this->getSerialized( $value, $key );
408  $valueSize = strlen( $serialized );
409 
410  try {
411  $result = $conn->set(
412  $key,
413  $serialized,
414  $ttl ? [ 'nx', 'ex' => $ttl ] : [ 'nx' ]
415  );
416  } catch ( RedisException $e ) {
417  $result = false;
418  $this->handleException( $conn, $e );
419  }
420 
421  $this->logRequest( 'add', $key, $conn->getServer(), $result );
422 
423  $this->updateOpStats( self::METRIC_OP_ADD, [ $key => [ $valueSize, 0 ] ] );
424 
425  return $result;
426  }
427 
428  public function incr( $key, $value = 1, $flags = 0 ) {
429  $conn = $this->getConnection( $key );
430  if ( !$conn ) {
431  return false;
432  }
433 
434  try {
435  if ( !$conn->exists( $key ) ) {
436  return false;
437  }
438  // @FIXME: on races, the key may have a 0 TTL
439  $result = $conn->incrBy( $key, $value );
440  } catch ( RedisException $e ) {
441  $result = false;
442  $this->handleException( $conn, $e );
443  }
444 
445  $this->logRequest( 'incr', $key, $conn->getServer(), $result );
446 
447  $this->updateOpStats( self::METRIC_OP_INCR, [ $key ] );
448 
449  return $result;
450  }
451 
452  public function decr( $key, $value = 1, $flags = 0 ) {
453  $conn = $this->getConnection( $key );
454  if ( !$conn ) {
455  return false;
456  }
457 
458  try {
459  if ( !$conn->exists( $key ) ) {
460  return false;
461  }
462  // @FIXME: on races, the key may have a 0 TTL
463  $result = $conn->decrBy( $key, $value );
464  } catch ( RedisException $e ) {
465  $result = false;
466  $this->handleException( $conn, $e );
467  }
468 
469  $this->logRequest( 'decr', $key, $conn->getServer(), $result );
470 
471  $this->updateOpStats( self::METRIC_OP_DECR, [ $key ] );
472 
473  return $result;
474  }
475 
476  protected function doIncrWithInit( $key, $exptime, $step, $init, $flags ) {
477  $conn = $this->getConnection( $key );
478  if ( !$conn ) {
479  return false;
480  }
481 
482  $ttl = $this->getExpirationAsTTL( $exptime );
483 
484  try {
485  if ( $init === $step && $exptime == self::TTL_INDEFINITE ) {
486  $newValue = $conn->incrBy( $key, $step );
487  } else {
488  $conn->multi( Redis::PIPELINE );
489  $conn->set(
490  $key,
491  (string)( $init - $step ),
492  $ttl ? [ 'nx', 'ex' => $ttl ] : [ 'nx' ]
493  );
494  $conn->incrBy( $key, $step );
495  $batchResult = $conn->exec();
496  $newValue = ( $batchResult === false ) ? false : $batchResult[1];
497  $this->logRequest( 'incrWithInit', $key, $conn->getServer(), $newValue === false );
498  }
499  } catch ( RedisException $e ) {
500  $newValue = false;
501  $this->handleException( $conn, $e );
502  }
503 
504  return $newValue;
505  }
506 
507  protected function doChangeTTL( $key, $exptime, $flags ) {
508  $conn = $this->getConnection( $key );
509  if ( !$conn ) {
510  return false;
511  }
512 
513  $relative = $this->isRelativeExpiration( $exptime );
514  try {
515  if ( $exptime == self::TTL_INDEFINITE ) {
516  $result = $conn->persist( $key );
517  $this->logRequest( 'persist', $key, $conn->getServer(), $result );
518  } elseif ( $relative ) {
519  $result = $conn->expire( $key, $this->getExpirationAsTTL( $exptime ) );
520  $this->logRequest( 'expire', $key, $conn->getServer(), $result );
521  } else {
522  $result = $conn->expireAt( $key, $this->getExpirationAsTimestamp( $exptime ) );
523  $this->logRequest( 'expireAt', $key, $conn->getServer(), $result );
524  }
525  } catch ( RedisException $e ) {
526  $result = false;
527  $this->handleException( $conn, $e );
528  }
529 
530  $this->updateOpStats( self::METRIC_OP_CHANGE_TTL, [ $key ] );
531 
532  return $result;
533  }
534 
539  protected function getConnection( $key ) {
540  $candidates = array_keys( $this->serverTagMap );
541 
542  if ( count( $this->servers ) > 1 ) {
543  ArrayUtils::consistentHashSort( $candidates, $key, '/' );
544  if ( !$this->automaticFailover ) {
545  $candidates = array_slice( $candidates, 0, 1 );
546  }
547  }
548 
549  while ( ( $tag = array_shift( $candidates ) ) !== null ) {
550  $server = $this->serverTagMap[$tag];
551  $conn = $this->redisPool->getConnection( $server, $this->logger );
552  if ( !$conn ) {
553  continue;
554  }
555 
556  // If automatic failover is enabled, check that the server's link
557  // to its master (if any) is up -- but only if there are other
558  // viable candidates left to consider. Also, getMasterLinkStatus()
559  // does not work with twemproxy, though $candidates will be empty
560  // by now in such cases.
561  if ( $this->automaticFailover && $candidates ) {
562  try {
564  $info = $conn->info();
565  if ( ( $info['master_link_status'] ?? null ) === 'down' ) {
566  // If the master cannot be reached, fail-over to the next server.
567  // If masters are in data-center A, and replica DBs in data-center B,
568  // this helps avoid the case were fail-over happens in A but not
569  // to the corresponding server in B (e.g. read/write mismatch).
570  continue;
571  }
572  } catch ( RedisException $e ) {
573  // Server is not accepting commands
574  $this->redisPool->handleError( $conn, $e );
575  continue;
576  }
577  }
578 
579  return $conn;
580  }
581 
583 
584  return null;
585  }
586 
591  protected function logError( $msg ) {
592  $this->logger->error( "Redis error: $msg" );
593  }
594 
603  protected function handleException( RedisConnRef $conn, RedisException $e ) {
605  $this->redisPool->handleError( $conn, $e );
606  }
607 
615  public function logRequest( $op, $keys, $server, $e = null ) {
616  $this->debug( "$op($keys) on $server: " . ( $e ? "failure" : "success" ) );
617  }
618 
619  public function makeKeyInternal( $keyspace, $components ) {
620  return $this->genericKeyFromComponents( $keyspace, ...$components );
621  }
622 
623  protected function convertGenericKey( $key ) {
624  return $key; // short-circuit; already uses "generic" keys
625  }
626 }
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
setLastError( $error)
Set the "last error" registry due to a problem encountered during an attempted operation.
Definition: BagOStuff.php:515
genericKeyFromComponents(... $components)
At a minimum, there must be a keyspace and collection name component.
Definition: BagOStuff.php:716
string $keyspace
Default keyspace; used by makeKey()
Definition: BagOStuff.php:104
Storage medium specific cache for storing items (e.g.
getExpirationAsTimestamp( $exptime)
Convert an optionally relative timestamp to an absolute time.
getSerialized( $value, $key)
Get the serialized form a value, using any applicable prepared value.
updateOpStats(string $op, array $keyInfo)
getExpirationAsTTL( $exptime)
Convert an optionally absolute expiry time to a relative time.
Redis-based caching module for redis server >= 2.6.12 and phpredis >= 2.2.4.
RedisConnectionPool $redisPool
incr( $key, $value=1, $flags=0)
Increase stored value of $key by $value while preserving its TTL.
handleException(RedisConnRef $conn, RedisException $e)
The redis extension throws an exception in response to various read, write and protocol errors.
convertGenericKey( $key)
Convert a "generic" reversible cache key into one for this cache.
array $servers
List of server names.
doSetMulti(array $data, $exptime=0, $flags=0)
makeKeyInternal( $keyspace, $components)
Make a cache key for the given keyspace and components.
doChangeTTL( $key, $exptime, $flags)
logRequest( $op, $keys, $server, $e=null)
Send information about a single request to the debug log.
doIncrWithInit( $key, $exptime, $step, $init, $flags)
doDelete( $key, $flags=0)
Delete an item.
logError( $msg)
Log a fatal error.
doAdd( $key, $value, $expiry=0, $flags=0)
Insert an item if it does not already exist.
__construct( $params)
Construct a RedisBagOStuff object.
doGet( $key, $flags=0, &$casToken=null)
array $serverTagMap
Map of (tag => server name)
doGetMulti(array $keys, $flags=0)
Get an associative array containing the item for each of the keys that have items.
doChangeTTLMulti(array $keys, $exptime, $flags=0)
doSet( $key, $value, $exptime=0, $flags=0)
Set an item.
doDeleteMulti(array $keys, $flags=0)
decr( $key, $value=1, $flags=0)
Decrease stored value of $key by $value while preserving its TTL.
Helper class to handle automatically marking connections as reusable (via RAII pattern)
static singleton(array $options)
const ERR_UNREACHABLE
Storage medium could not be reached to establish a connection.
const ATTR_DURABILITY
Durability of writes; see QOS_DURABILITY_* (higher means stronger)
const ERR_UNEXPECTED
Storage medium operation failed due to usage limitations or an I/O error.
const QOS_DURABILITY_DISK
Data is saved to disk and writes do not usually block on fsync()
foreach( $res as $row) $serialized