22 use Liuggio\StatsdClient\Factory\StatsdDataFactoryInterface;
23 use Psr\Log\LoggerAwareInterface;
24 use Psr\Log\LoggerInterface;
25 use Psr\Log\NullLogger;
162 private $keyHighByteSize;
164 private $keyHighUplinkBps;
170 private $callbackDepth = 0;
172 private $warmupCache = [];
174 private $warmupKeyMisses = 0;
177 private $wallClockOverride;
180 private const MAX_COMMIT_DELAY = 3;
182 private const MAX_READ_LAG = 7;
184 public const HOLDOFF_TTL = self::MAX_COMMIT_DELAY + self::MAX_READ_LAG + 1;
187 private const LOW_TTL = 60;
192 private const HOT_TTR = 900;
194 private const AGE_NEW = 60;
197 private const TSE_NONE = -1;
207 public const MIN_TIMESTAMP_NONE = 0.0;
210 private const PC_PRIMARY =
'primary:1000';
216 private const SCHEME_HASH_TAG = 1;
218 private const SCHEME_HASH_STOP = 2;
221 private const CHECK_KEY_TTL = self::TTL_YEAR;
223 private const INTERIM_KEY_TTL = 2;
226 private const LOCK_TTL = 10;
228 private const COOLOFF_TTL = 2;
230 private const RAMPUP_TTL = 30;
233 private const TINY_NEGATIVE = -0.000001;
235 private const TINY_POSTIVE = 0.000001;
238 private const RECENT_SET_LOW_MS = 50;
240 private const RECENT_SET_HIGH_MS = 100;
243 private const GENERATION_HIGH_SEC = 0.2;
246 private const PURGE_TIME = 0;
248 private const PURGE_HOLDOFF = 1;
251 private const VERSION = 1;
267 private const RES_VALUE = 0;
269 private const RES_VERSION = 1;
271 private const RES_AS_OF = 2;
273 private const RES_TTL = 3;
275 private const RES_TOMB_AS_OF = 4;
277 private const RES_CHECK_AS_OF = 5;
279 private const RES_TOUCH_AS_OF = 6;
281 private const RES_CUR_TTL = 7;
284 private const FLD_FORMAT_VERSION = 0;
286 private const FLD_VALUE = 1;
288 private const FLD_TTL = 2;
290 private const FLD_TIME = 3;
292 private const FLD_FLAGS = 4;
294 private const FLD_VALUE_VERSION = 5;
295 private const FLD_GENERATION_TIME = 6;
298 private const TYPE_VALUE =
'v';
300 private const TYPE_TIMESTAMP =
't';
302 private const TYPE_MUTEX =
'm';
304 private const TYPE_INTERIM =
'i';
306 private const TYPE_COOLOFF =
'c';
309 private const PURGE_VAL_PREFIX =
'PURGED';
348 $this->cache = $params[
'cache'];
349 $this->broadcastRoute = $params[
'broadcastRoutingPrefix'] ??
null;
350 $this->epoch = $params[
'epoch'] ?? 0;
351 $this->secret = $params[
'secret'] ?? (string)$this->epoch;
352 if ( ( $params[
'coalesceScheme'] ??
'' ) ===
'hash_tag' ) {
356 $this->coalesceScheme = self::SCHEME_HASH_TAG;
359 $this->coalesceScheme = self::SCHEME_HASH_STOP;
362 $this->keyHighQps = $params[
'keyHighQps'] ?? 100;
363 $this->keyHighByteSize = $params[
'keyHighByteSize'] ?? ( 128 * 1024 );
364 $this->keyHighUplinkBps = $params[
'keyHighUplinkBps'] ?? ( 1e9 / 8 / 100 );
366 $this->
setLogger( $params[
'logger'] ??
new NullLogger() );
368 $this->asyncHandler = $params[
'asyncHandler'] ??
null;
370 $this->missLog = array_fill( 0, 10, [
'', 0.0 ] );
372 $this->cache->registerWrapperInfoForStats(
375 [ __CLASS__,
'getCollectionFromSisterKey' ]
450 final public function get( $key, &$curTTL =
null, array $checkKeys = [], &$info = [] ) {
457 $curTTL =
$res[self::RES_CUR_TTL];
459 ?
$res[self::RES_AS_OF]
461 self::KEY_VERSION =>
$res[self::RES_VERSION],
462 self::KEY_AS_OF =>
$res[self::RES_AS_OF],
463 self::KEY_TTL =>
$res[self::RES_TTL],
464 self::KEY_CUR_TTL =>
$res[self::RES_CUR_TTL],
465 self::KEY_TOMB_AS_OF =>
$res[self::RES_TOMB_AS_OF],
466 self::KEY_CHECK_AS_OF =>
$res[self::RES_CHECK_AS_OF]
469 if ( $curTTL ===
null || $curTTL <= 0 ) {
471 unset( $this->missLog[array_key_first( $this->missLog )] );
475 return $res[self::RES_VALUE];
505 array $checkKeys = [],
516 $resByKey = $this->
fetchKeys( $keys, $checkKeys );
517 foreach ( $resByKey as $key =>
$res ) {
518 if (
$res[self::RES_VALUE] !==
false ) {
519 $valuesByKey[$key] =
$res[self::RES_VALUE];
522 if (
$res[self::RES_CUR_TTL] !==
null ) {
523 $curTTLs[$key] =
$res[self::RES_CUR_TTL];
525 $info[$key] = $legacyInfo
526 ?
$res[self::RES_AS_OF]
528 self::KEY_VERSION =>
$res[self::RES_VERSION],
529 self::KEY_AS_OF =>
$res[self::RES_AS_OF],
530 self::KEY_TTL =>
$res[self::RES_TTL],
531 self::KEY_CUR_TTL =>
$res[self::RES_CUR_TTL],
532 self::KEY_TOMB_AS_OF =>
$res[self::RES_TOMB_AS_OF],
533 self::KEY_CHECK_AS_OF =>
$res[self::RES_CHECK_AS_OF]
554 protected function fetchKeys( array
$keys, array $checkKeys, $touchedCb =
null ) {
560 $valueSisterKeys = [];
562 $checkSisterKeysForAll = [];
564 $checkSisterKeysByKey = [];
566 foreach (
$keys as $key ) {
567 $sisterKey = $this->makeSisterKey( $key, self::TYPE_VALUE );
568 $allSisterKeys[] = $sisterKey;
569 $valueSisterKeys[] = $sisterKey;
572 foreach ( $checkKeys as $i => $checkKeyOrKeyGroup ) {
574 if ( is_int( $i ) ) {
576 $sisterKey = $this->makeSisterKey( $checkKeyOrKeyGroup, self::TYPE_TIMESTAMP );
577 $allSisterKeys[] = $sisterKey;
578 $checkSisterKeysForAll[] = $sisterKey;
581 foreach ( (array)$checkKeyOrKeyGroup as $checkKey ) {
582 $sisterKey = $this->makeSisterKey( $checkKey, self::TYPE_TIMESTAMP );
583 $allSisterKeys[] = $sisterKey;
584 $checkSisterKeysByKey[$i][] = $sisterKey;
589 if ( $this->warmupCache ) {
591 $wrappedBySisterKey = $this->warmupCache;
592 $sisterKeysMissing = array_diff( $allSisterKeys, array_keys( $wrappedBySisterKey ) );
593 if ( $sisterKeysMissing ) {
594 $this->warmupKeyMisses += count( $sisterKeysMissing );
595 $wrappedBySisterKey += $this->cache->getMulti( $sisterKeysMissing );
599 $wrappedBySisterKey = $this->cache->getMulti( $allSisterKeys );
606 $ckPurgesForAll = $this->processCheckKeys(
607 $checkSisterKeysForAll,
613 foreach ( $checkSisterKeysByKey as $keyWithCheckKeys => $checkKeysForKey ) {
614 $ckPurgesByKey[$keyWithCheckKeys] = $this->processCheckKeys(
623 foreach ( $valueSisterKeys as $valueSisterKey ) {
625 $key = current(
$keys );
628 if ( array_key_exists( $valueSisterKey, $wrappedBySisterKey ) ) {
630 $wrapped = $wrappedBySisterKey[$valueSisterKey];
636 $res = $this->unwrap( $wrapped, $now );
637 $value =
$res[self::RES_VALUE];
639 foreach ( array_merge( $ckPurgesForAll, $ckPurgesByKey[$key] ?? [] ) as $ckPurge ) {
640 $res[self::RES_CHECK_AS_OF] = max(
641 $ckPurge[self::PURGE_TIME],
642 $res[self::RES_CHECK_AS_OF]
645 $holdoffDeadline = $ckPurge[self::PURGE_TIME] + $ckPurge[self::PURGE_HOLDOFF];
647 if ( $value !==
false && $holdoffDeadline >=
$res[self::RES_AS_OF] ) {
649 $ago = min( $ckPurge[self::PURGE_TIME] - $now, self::TINY_NEGATIVE );
651 $res[self::RES_CUR_TTL] = min(
$res[self::RES_CUR_TTL], $ago );
655 if ( $touchedCb !==
null && $value !==
false ) {
656 $touched = $touchedCb( $value );
657 if ( $touched !==
null && $touched >=
$res[self::RES_AS_OF] ) {
658 $res[self::RES_CUR_TTL] = min(
659 $res[self::RES_CUR_TTL],
660 $res[self::RES_AS_OF] - $touched,
668 $res[self::RES_TOUCH_AS_OF] = max(
$res[self::RES_TOUCH_AS_OF], $touched );
670 $resByKey[$key] =
$res;
682 private function processCheckKeys(
683 array $checkSisterKeys,
684 array $wrappedBySisterKey,
689 foreach ( $checkSisterKeys as $timeKey ) {
690 $purge = isset( $wrappedBySisterKey[$timeKey] )
691 ? $this->parsePurgeValue( $wrappedBySisterKey[$timeKey] )
694 if ( $purge ===
null ) {
695 $wrapped = $this->makeCheckPurgeValue( $now, self::HOLDOFF_TTL, $purge );
700 $this->cache::WRITE_BACKGROUND
793 final public function set( $key, $value, $ttl = self::TTL_INDEFINITE, array $opts = [] ) {
794 $kClass = $this->determineKeyClassForStats( $key );
796 $ok = $this->setMainValue(
800 $opts[
'version'] ??
null,
801 $opts[
'walltime'] ??
null,
803 $opts[
'since'] ??
null,
804 $opts[
'pending'] ??
false,
805 $opts[
'lockTSE'] ?? self::TSE_NONE,
806 $opts[
'staleTTL'] ?? self::STALE_TTL_NONE,
807 $opts[
'segmentable'] ??
false,
808 $opts[
'creating'] ??
false
811 $this->stats->increment(
"wanobjectcache.$kClass.set." . ( $ok ?
'ok' :
'error' ) );
831 private function setMainValue(
839 bool $dataPendingCommit,
852 $walltime ??= $this->timeSinceLoggedMiss( $key, $now );
853 $dataSnapshotLag = ( $dataReadSince !== null ) ? max( 0, $now - $dataReadSince ) : 0;
854 $dataCombinedLag = $dataReplicaLag + $dataSnapshotLag;
861 if ( $dataPendingCommit ) {
863 $mitigated =
'pending writes';
865 $mitigationTTL = self::TTL_UNCACHEABLE;
866 } elseif ( $dataSnapshotLag > self::MAX_READ_LAG ) {
868 $pregenSnapshotLag = ( $walltime !== null ) ? ( $dataSnapshotLag - $walltime ) : 0;
869 if ( ( $pregenSnapshotLag + self::GENERATION_HIGH_SEC ) > self::MAX_READ_LAG ) {
871 $mitigated =
'snapshot lag (late generation)';
873 $mitigationTTL = self::TTL_UNCACHEABLE;
876 $mitigated =
'snapshot lag (high generation time)';
880 } elseif ( $dataReplicaLag ===
false || $dataReplicaLag > self::MAX_READ_LAG ) {
882 $mitigated =
'replication lag';
885 } elseif ( $dataCombinedLag > self::MAX_READ_LAG ) {
886 $pregenCombinedLag = ( $walltime !== null ) ? ( $dataCombinedLag - $walltime ) : 0;
888 if ( ( $pregenCombinedLag + self::GENERATION_HIGH_SEC ) > self::MAX_READ_LAG ) {
890 $mitigated =
'read lag (late generation)';
892 $mitigationTTL = self::TTL_UNCACHEABLE;
895 $mitigated =
'read lag (high generation time)';
903 $mitigationTTL =
null;
906 if ( $mitigationTTL === self::TTL_UNCACHEABLE ) {
907 $this->logger->warning(
908 "Rejected set() for {cachekey} due to $mitigated.",
911 'lag' => $dataReplicaLag,
912 'age' => $dataSnapshotLag,
913 'walltime' => $walltime
924 if ( $mitigationTTL !==
null ) {
926 if ( $lockTSE >= 0 ) {
928 $logicalTTL = min( $ttl ?: INF, $mitigationTTL );
931 $ttl = min( $ttl ?: INF, $mitigationTTL );
934 $this->logger->warning(
935 "Lowered set() TTL for {cachekey} due to $mitigated.",
938 'lag' => $dataReplicaLag,
939 'age' => $dataSnapshotLag,
940 'walltime' => $walltime
946 $wrapped = $this->wrap( $value, $logicalTTL ?: $ttl, $version, $now, $walltime );
947 $storeTTL = $ttl + $staleTTL;
949 $flags = $this->cache::WRITE_BACKGROUND;
950 if ( $segmentable ) {
951 $flags |= $this->cache::WRITE_ALLOW_SEGMENTS;
955 $ok = $this->cache->add(
956 $this->makeSisterKey( $key, self::TYPE_VALUE ),
962 $ok = $this->cache->merge(
963 $this->makeSisterKey( $key, self::TYPE_VALUE ),
964 static function (
$cache, $key, $cWrapped ) use ( $wrapped ) {
966 return ( is_string( $cWrapped ) ) ?
false : $wrapped;
969 $this->cache::MAX_CONFLICTS_ONE,
1043 $valueSisterKey = $this->makeSisterKey( $key, self::TYPE_VALUE );
1059 $purge = $this->makeTombstonePurgeValue( $now );
1063 $kClass = $this->determineKeyClassForStats( $key );
1064 $this->stats->increment(
"wanobjectcache.$kClass.delete." . ( $ok ?
'ok' :
'error' ) );
1154 $checkSisterKeysByKey = [];
1155 foreach (
$keys as $key ) {
1156 $checkSisterKeysByKey[$key] = $this->makeSisterKey( $key, self::TYPE_TIMESTAMP );
1159 $wrappedBySisterKey = $this->cache->getMulti( $checkSisterKeysByKey );
1160 $wrappedBySisterKey += array_fill_keys( $checkSisterKeysByKey,
false );
1164 foreach ( $checkSisterKeysByKey as $key => $checkSisterKey ) {
1165 $purge = $this->parsePurgeValue( $wrappedBySisterKey[$checkSisterKey] );
1166 if ( $purge ===
null ) {
1167 $wrapped = $this->makeCheckPurgeValue( $now, self::HOLDOFF_TTL, $purge );
1171 self::CHECK_KEY_TTL,
1172 $this->cache::WRITE_BACKGROUND
1176 $times[$key] = $purge[self::PURGE_TIME];
1216 $checkSisterKey = $this->makeSisterKey( $key, self::TYPE_TIMESTAMP );
1219 $purge = $this->makeCheckPurgeValue( $now, $holdoff );
1222 $kClass = $this->determineKeyClassForStats( $key );
1223 $this->stats->increment(
"wanobjectcache.$kClass.ck_touch." . ( $ok ?
'ok' :
'error' ) );
1256 $checkSisterKey = $this->makeSisterKey( $key, self::TYPE_TIMESTAMP );
1259 $kClass = $this->determineKeyClassForStats( $key );
1260 $this->stats->increment(
"wanobjectcache.$kClass.ck_reset." . ( $ok ?
'ok' :
'error' ) );
1567 $key, $ttl, $callback, array $opts = [], array $cbParams = []
1569 $version = $opts[
'version'] ??
null;
1570 $pcTTL = $opts[
'pcTTL'] ?? self::TTL_UNCACHEABLE;
1571 $pCache = ( $pcTTL >= 0 )
1572 ? $this->getProcessCache( $opts[
'pcGroup'] ?? self::PC_PRIMARY )
1578 if ( $pCache && $this->callbackDepth == 0 ) {
1579 $cached = $pCache->get( $key, $pcTTL,
false );
1580 if ( $cached !==
false ) {
1581 $this->logger->debug(
"getWithSetCallback($key): process cache hit" );
1586 [ $value, $valueVersion, $curAsOf ] = $this->fetchOrRegenerate( $key, $ttl, $callback, $opts, $cbParams );
1587 if ( $valueVersion !== $version ) {
1591 $this->logger->debug(
"getWithSetCallback($key): using variant key" );
1592 [ $value ] = $this->fetchOrRegenerate(
1593 $this->
makeGlobalKey(
'WANCache-key-variant', md5( $key ), (
string)$version ),
1596 [
'version' =>
null,
'minAsOf' => $curAsOf ] + $opts,
1602 if ( $pCache && $value !==
false ) {
1603 $pCache->set( $key, $value );
1625 private function fetchOrRegenerate( $key, $ttl, $callback, array $opts, array $cbParams ) {
1626 $checkKeys = $opts[
'checkKeys'] ?? [];
1628 $minAsOf = $opts[
'minAsOf'] ?? self::MIN_TIMESTAMP_NONE;
1629 $hotTTR = $opts[
'hotTTR'] ?? self::HOT_TTR;
1630 $lowTTL = $opts[
'lowTTL'] ?? min( self::LOW_TTL, $ttl );
1631 $ageNew = $opts[
'ageNew'] ?? self::AGE_NEW;
1632 $touchedCb = $opts[
'touchedCallback'] ??
null;
1635 $kClass = $this->determineKeyClassForStats( $key );
1638 $curState = $this->
fetchKeys( [ $key ], $checkKeys, $touchedCb )[$key];
1639 $curValue = $curState[self::RES_VALUE];
1641 if ( $this->isAcceptablyFreshValue( $curState, $graceTTL, $minAsOf ) ) {
1643 $this->stats->timing(
1644 "wanobjectcache.$kClass.hit.good",
1648 return [ $curValue, $curState[self::RES_VERSION], $curState[self::RES_AS_OF] ];
1649 } elseif ( $this->scheduleAsyncRefresh( $key, $ttl, $callback, $opts, $cbParams ) ) {
1650 $this->logger->debug(
"fetchOrRegenerate($key): hit with async refresh" );
1651 $this->stats->timing(
1652 "wanobjectcache.$kClass.hit.refresh",
1656 return [ $curValue, $curState[self::RES_VERSION], $curState[self::RES_AS_OF] ];
1658 $this->logger->debug(
"fetchOrRegenerate($key): hit with sync refresh" );
1662 $isKeyTombstoned = ( $curState[self::RES_TOMB_AS_OF] !== null );
1664 if ( $isKeyTombstoned ) {
1665 $volState = $this->getInterimValue( $key, $minAsOf, $startTime, $touchedCb );
1666 $volValue = $volState[self::RES_VALUE];
1668 $volState = $curState;
1669 $volValue = $curValue;
1677 $lastPurgeTime = max(
1679 $volState[self::RES_TOUCH_AS_OF],
1680 $curState[self::RES_TOMB_AS_OF],
1681 $curState[self::RES_CHECK_AS_OF]
1683 $safeMinAsOf = max( $minAsOf, $lastPurgeTime + self::TINY_POSTIVE );
1684 if ( $this->isExtremelyNewValue( $volState, $safeMinAsOf, $startTime ) ) {
1685 $this->logger->debug(
"fetchOrRegenerate($key): volatile hit" );
1686 $this->stats->timing(
1687 "wanobjectcache.$kClass.hit.volatile",
1691 return [ $volValue, $volState[self::RES_VERSION], $curState[self::RES_AS_OF] ];
1694 $lockTSE = $opts[
'lockTSE'] ?? self::TSE_NONE;
1695 $busyValue = $opts[
'busyValue'] ??
null;
1697 $segmentable = $opts[
'segmentable'] ??
false;
1698 $version = $opts[
'version'] ??
null;
1701 $useRegenerationLock =
1712 $curState[self::RES_CUR_TTL] !==
null &&
1713 $curState[self::RES_CUR_TTL] <= 0 &&
1714 abs( $curState[self::RES_CUR_TTL] ) <= $lockTSE
1718 ( $busyValue !==
null && $volValue ===
false );
1723 $hasLock = $useRegenerationLock && $this->claimStampedeLock( $key );
1724 if ( $useRegenerationLock && !$hasLock ) {
1727 if ( $this->
isValid( $volValue, $volState[self::RES_AS_OF], $minAsOf ) ) {
1728 $this->logger->debug(
"fetchOrRegenerate($key): returning stale value" );
1729 $this->stats->timing(
1730 "wanobjectcache.$kClass.hit.stale",
1734 return [ $volValue, $volState[self::RES_VERSION], $curState[self::RES_AS_OF] ];
1735 } elseif ( $busyValue !==
null ) {
1736 $miss = is_infinite( $minAsOf ) ?
'renew' :
'miss';
1737 $this->logger->debug(
"fetchOrRegenerate($key): busy $miss" );
1738 $this->stats->timing(
1739 "wanobjectcache.$kClass.$miss.busy",
1742 $placeholderValue = $this->resolveBusyValue( $busyValue );
1744 return [ $placeholderValue, $version, $curState[self::RES_AS_OF] ];
1751 ++$this->callbackDepth;
1756 ( $curState[self::RES_VERSION] === $version ) ? $curValue :
false,
1759 ( $curState[self::RES_VERSION] === $version ) ? $curState[self::RES_AS_OF] :
null,
1763 --$this->callbackDepth;
1768 $elapsed = max( $postCallbackTime - $startTime, 0.0 );
1771 $walltime = max( $postCallbackTime - $preCallbackTime, 0.0 );
1772 $this->stats->timing(
"wanobjectcache.$kClass.regen_walltime", 1e3 * $walltime );
1777 ( $value !==
false && $ttl >= 0 ) &&
1779 ( !$useRegenerationLock || $hasLock || $isKeyTombstoned ) &&
1781 $this->checkAndSetCooloff( $key, $kClass, $value, $elapsed, $hasLock )
1784 if ( $isKeyTombstoned ) {
1785 $this->setInterimValue(
1794 $this->setMainValue(
1801 $setOpts[
'lag'] ?? 0,
1803 $setOpts[
'since'] ?? $preCallbackTime,
1805 $setOpts[
'pending'] ??
false,
1809 ( $curValue ===
false )
1814 $this->yieldStampedeLock( $key, $hasLock );
1816 $miss = is_infinite( $minAsOf ) ?
'renew' :
'miss';
1817 $this->logger->debug(
"fetchOrRegenerate($key): $miss, new value computed" );
1818 $this->stats->timing(
1819 "wanobjectcache.$kClass.$miss.compute",
1823 return [ $value, $version, $curState[self::RES_AS_OF] ];
1830 private function claimStampedeLock( $key ) {
1831 $checkSisterKey = $this->makeSisterKey( $key, self::TYPE_MUTEX );
1833 return $this->cache->add( $checkSisterKey, 1, self::LOCK_TTL );
1840 private function yieldStampedeLock( $key, $hasLock ) {
1842 $checkSisterKey = $this->makeSisterKey( $key, self::TYPE_MUTEX );
1843 $this->cache->delete( $checkSisterKey, $this->cache::WRITE_BACKGROUND );
1857 private function makeSisterKeys( array $baseKeys,
string $type,
string $route =
null ) {
1859 foreach ( $baseKeys as $baseKey ) {
1860 $sisterKeys[] = $this->makeSisterKey( $baseKey,
$type, $route );
1876 private function makeSisterKey(
string $baseKey,
string $typeChar,
string $route =
null ) {
1877 if ( $this->coalesceScheme === self::SCHEME_HASH_STOP ) {
1879 $sisterKey =
'WANCache:' . $baseKey .
'|#|' . $typeChar;
1882 $sisterKey =
'WANCache:{' . $baseKey .
'}:' . $typeChar;
1885 if ( $route !==
null ) {
1886 $sisterKey = $this->
prependRoute( $sisterKey, $route );
1899 if ( substr( $sisterKey, -4 ) ===
'|#|v' ) {
1901 $collection = substr( $sisterKey, 9, strcspn( $sisterKey,
':|', 9 ) );
1902 } elseif ( substr( $sisterKey, -3 ) ===
'}:v' ) {
1904 $collection = substr( $sisterKey, 10, strcspn( $sisterKey,
':}', 10 ) );
1906 $collection =
'internal';
1924 private function isExtremelyNewValue(
$res, $minAsOf, $now ) {
1925 if (
$res[self::RES_VALUE] ===
false ||
$res[self::RES_AS_OF] < $minAsOf ) {
1929 $age = $now -
$res[self::RES_AS_OF];
1931 return ( $age < mt_rand( self::RECENT_SET_LOW_MS, self::RECENT_SET_HIGH_MS ) / 1e3 );
1955 private function checkAndSetCooloff( $key, $kClass, $value, $elapsed, $hasLock ) {
1956 if ( is_scalar( $value ) ) {
1958 $hypotheticalSize = strlen( (
string)$value );
1961 $hypotheticalSize = $this->keyHighByteSize;
1973 $missesPerSecForHighQPS = ( min( $elapsed, 1 ) * $this->keyHighQps );
1985 if ( ( $missesPerSecForHighQPS * $hypotheticalSize ) >= $this->keyHighUplinkBps ) {
1986 $cooloffSisterKey = $this->makeSisterKey( $key, self::TYPE_COOLOFF );
1987 $watchPoint = $this->cache->watchErrors();
1989 !$this->cache->add( $cooloffSisterKey, 1, self::COOLOFF_TTL ) &&
1991 $this->cache->getLastError( $watchPoint ) === self::ERR_NONE
1993 $this->logger->debug(
"checkAndSetCooloff($key): bounced; {$elapsed}s" );
1994 $this->stats->increment(
"wanobjectcache.$kClass.cooloff_bounce" );
2002 $this->stats->timing(
"wanobjectcache.$kClass.regen_set_delay", 1e3 * $elapsed );
2003 $this->stats->updateCount(
"wanobjectcache.$kClass.regen_set_bytes", $hypotheticalSize );
2017 private function getInterimValue( $key, $minAsOf, $now, $touchedCb ) {
2019 $interimSisterKey = $this->makeSisterKey( $key, self::TYPE_INTERIM );
2020 $wrapped = $this->cache->get( $interimSisterKey );
2021 $res = $this->unwrap( $wrapped, $now );
2022 if (
$res[self::RES_VALUE] !==
false &&
$res[self::RES_AS_OF] >= $minAsOf ) {
2023 if ( $touchedCb !==
null ) {
2026 $res[self::RES_TOUCH_AS_OF] = max(
2027 $touchedCb(
$res[self::RES_VALUE] ),
2028 $res[self::RES_TOUCH_AS_OF]
2036 return $this->unwrap(
false, $now );
2048 private function setInterimValue(
2057 $ttl = max( self::INTERIM_KEY_TTL, (
int)$ttl );
2060 $wrapped = $this->wrap( $value, $ttl, $version, $now, $walltime );
2062 $flags = $this->cache::WRITE_BACKGROUND;
2063 if ( $segmentable ) {
2064 $flags |= $this->cache::WRITE_ALLOW_SEGMENTS;
2067 return $this->cache->set(
2068 $this->makeSisterKey( $key, self::TYPE_INTERIM ),
2079 private function resolveBusyValue( $busyValue ) {
2080 return ( $busyValue instanceof Closure ) ? $busyValue() : $busyValue;
2149 ArrayIterator $keyedIds, $ttl, callable $callback, array $opts = []
2152 $this->warmupCache = $this->fetchWrappedValuesForWarmupCache(
2153 $this->getNonProcessCachedMultiKeys( $keyedIds, $opts ),
2154 $opts[
'checkKeys'] ?? []
2156 $this->warmupKeyMisses = 0;
2162 $proxyCb =
static function ( $oldValue, &$ttl, &$setOpts, $oldAsOf, $params )
2165 return $callback( $params[
'id'], $oldValue, $ttl, $setOpts, $oldAsOf );
2170 foreach ( $keyedIds as $key => $id ) {
2180 $this->warmupCache = [];
2252 ArrayIterator $keyedIds, $ttl, callable $callback, array $opts = []
2254 $checkKeys = $opts[
'checkKeys'] ?? [];
2255 $minAsOf = $opts[
'minAsOf'] ?? self::MIN_TIMESTAMP_NONE;
2258 unset( $opts[
'lockTSE'] );
2259 unset( $opts[
'busyValue'] );
2262 $keysByIdGet = $this->getNonProcessCachedMultiKeys( $keyedIds, $opts );
2263 $this->warmupCache = $this->fetchWrappedValuesForWarmupCache( $keysByIdGet, $checkKeys );
2264 $this->warmupKeyMisses = 0;
2270 $resByKey = $this->
fetchKeys( $keysByIdGet, $checkKeys );
2271 foreach ( $keysByIdGet as $id => $key ) {
2272 $res = $resByKey[$key];
2274 $res[self::RES_VALUE] ===
false ||
2275 $res[self::RES_CUR_TTL] < 0 ||
2276 $res[self::RES_AS_OF] < $minAsOf
2284 $newTTLsById = array_fill_keys( $idsRegen, $ttl );
2285 $newValsById = $idsRegen ? $callback( $idsRegen, $newTTLsById, $newSetOpts ) : [];
2287 $method = __METHOD__;
2292 $proxyCb =
function ( $oldValue, &$ttl, &$setOpts, $oldAsOf, $params )
2293 use ( $callback, $newValsById, $newTTLsById, $newSetOpts, $method )
2295 $id = $params[
'id'];
2297 if ( array_key_exists( $id, $newValsById ) ) {
2299 $newValue = $newValsById[$id];
2300 $ttl = $newTTLsById[$id];
2301 $setOpts = $newSetOpts;
2305 $ttls = [ $id => $ttl ];
2306 $result = $callback( [ $id ], $ttls, $setOpts );
2307 if ( !isset( $result[$id] ) ) {
2309 $this->logger->warning(
2310 $method .
' failed due to {id} not set in result {result}', [
2312 'result' => json_encode( $result )
2315 $newValue = $result[$id];
2324 foreach ( $keyedIds as $key => $id ) {
2334 $this->warmupCache = [];
2351 return $this->cache->makeGlobalKey( ...func_get_args() );
2364 public function makeKey( $collection, ...$components ) {
2366 return $this->cache->makeKey( ...func_get_args() );
2377 return hash_hmac(
'sha256', $component, $this->secret );
2433 foreach ( $ids as $id ) {
2435 if ( strlen( $id ) > 64 ) {
2436 $this->logger->warning( __METHOD__ .
": long ID '$id'; use hash256()" );
2438 $key = $keyCallback( $id, $this );
2440 if ( !isset( $idByKey[$key] ) ) {
2441 $idByKey[$key] = $id;
2442 } elseif ( (
string)$id !== (
string)$idByKey[$key] ) {
2443 throw new UnexpectedValueException(
2444 "Cache key collision; IDs ('$id','{$idByKey[$key]}') map to '$key'"
2449 return new ArrayIterator( $idByKey );
2488 if ( count( $ids ) !== count(
$res ) ) {
2491 $ids = array_keys( array_fill_keys( $ids,
true ) );
2492 if ( count( $ids ) !== count(
$res ) ) {
2493 throw new UnexpectedValueException(
"Multi-key result does not match ID list" );
2497 return array_combine( $ids,
$res );
2507 return $this->cache->watchErrors();
2528 $code = $this->cache->getLastError( $watchPoint );
2546 $this->cache->clearLastError();
2555 $this->processCaches = [];
2588 return $this->cache->getQoS( $flag );
2654 public function adaptiveTTL( $mtime, $maxTTL, $minTTL = 30, $factor = 0.2 ) {
2656 $mtime = (int)$mtime;
2657 if ( $mtime <= 0 ) {
2664 return (
int)min( $maxTTL, max( $minTTL, $factor * $age ) );
2674 return $this->warmupKeyMisses;
2692 if ( $this->broadcastRoute !==
null ) {
2693 $routeKey = $this->
prependRoute( $sisterKey, $this->broadcastRoute );
2695 $routeKey = $sisterKey;
2698 return $this->cache->set(
2702 $this->cache::WRITE_BACKGROUND
2715 if ( $this->broadcastRoute !==
null ) {
2716 $routeKey = $this->
prependRoute( $sisterKey, $this->broadcastRoute );
2718 $routeKey = $sisterKey;
2721 return $this->cache->delete( $routeKey, $this->cache::WRITE_BACKGROUND );
2730 if ( $sisterKey[0] ===
'/' ) {
2731 throw new RuntimeException(
"Sister key '$sisterKey' already contains a route." );
2734 return $route . $sisterKey;
2748 private function scheduleAsyncRefresh( $key, $ttl, $callback, array $opts, array $cbParams ) {
2749 if ( !$this->asyncHandler ) {
2757 $func(
function () use ( $key, $ttl, $callback, $opts, $cbParams ) {
2758 $opts[
'minAsOf'] = INF;
2760 $this->fetchOrRegenerate( $key, $ttl, $callback, $opts, $cbParams );
2761 }
catch ( Exception $e ) {
2763 $this->logger->error(
'Async refresh failed for {key}', [
2783 private function isAcceptablyFreshValue(
$res, $graceTTL, $minAsOf ) {
2784 if ( !$this->
isValid( $res[self::RES_VALUE],
$res[self::RES_AS_OF], $minAsOf ) ) {
2789 $curTTL =
$res[self::RES_CUR_TTL];
2790 if ( $curTTL > 0 ) {
2796 $curGraceTTL = $graceTTL + $curTTL;
2798 return ( $curGraceTTL > 0 )
2816 $curTTL =
$res[self::RES_CUR_TTL];
2817 $logicalTTL =
$res[self::RES_TTL];
2818 $asOf =
$res[self::RES_AS_OF];
2842 if ( $ageNew < 0 || $timeTillRefresh <= 0 ) {
2846 $age = $now - $asOf;
2847 $timeOld = $age - $ageNew;
2848 if ( $timeOld <= 0 ) {
2852 $popularHitsPerSec = 1;
2856 $refreshWindowSec = max( $timeTillRefresh - $ageNew - self::RAMPUP_TTL / 2, 1 );
2860 $chance = 1 / ( $popularHitsPerSec * $refreshWindowSec );
2862 $chance *= ( $timeOld <= self::RAMPUP_TTL ) ? $timeOld / self::RAMPUP_TTL : 1;
2864 return ( mt_rand( 1, 1000000000 ) <= 1000000000 * $chance );
2886 if ( $lowTTL <= 0 ) {
2891 $effectiveLowTTL = min( $lowTTL, $logicalTTL ?: INF );
2894 $timeOld = $effectiveLowTTL - $curTTL;
2895 if ( $timeOld <= 0 || $timeOld >= $effectiveLowTTL ) {
2900 $ttrRatio = $timeOld / $effectiveLowTTL;
2903 $chance = $ttrRatio ** 4;
2905 return ( mt_rand( 1, 1000000000 ) <= 1000000000 * $chance );
2916 protected function isValid( $value, $asOf, $minAsOf ) {
2917 return ( $value !==
false && $asOf >= $minAsOf );
2928 private function wrap( $value, $ttl, $version, $now, $walltime ) {
2932 self::FLD_FORMAT_VERSION => self::VERSION,
2933 self::FLD_VALUE => $value,
2934 self::FLD_TTL => $ttl,
2935 self::FLD_TIME => $now
2937 if ( $version !==
null ) {
2938 $wrapped[self::FLD_VALUE_VERSION] = $version;
2958 private function unwrap( $wrapped, $now ) {
2962 self::RES_VALUE =>
false,
2963 self::RES_VERSION =>
null,
2964 self::RES_AS_OF =>
null,
2965 self::RES_TTL =>
null,
2966 self::RES_TOMB_AS_OF =>
null,
2968 self::RES_CHECK_AS_OF =>
null,
2969 self::RES_TOUCH_AS_OF =>
null,
2970 self::RES_CUR_TTL => null
2973 if ( is_array( $wrapped ) ) {
2976 ( $wrapped[self::FLD_FORMAT_VERSION] ??
null ) === self::VERSION &&
2977 $wrapped[self::FLD_TIME] >= $this->epoch
2979 if ( $wrapped[self::FLD_TTL] > 0 ) {
2981 $age = $now - $wrapped[self::FLD_TIME];
2982 $curTTL = max( $wrapped[self::FLD_TTL] - $age, 0.0 );
2987 $res[self::RES_VALUE] = $wrapped[self::FLD_VALUE];
2988 $res[self::RES_VERSION] = $wrapped[self::FLD_VALUE_VERSION] ??
null;
2989 $res[self::RES_AS_OF] = $wrapped[self::FLD_TIME];
2990 $res[self::RES_CUR_TTL] = $curTTL;
2991 $res[self::RES_TTL] = $wrapped[self::FLD_TTL];
2995 $purge = $this->parsePurgeValue( $wrapped );
2996 if ( $purge !==
null ) {
2998 $curTTL = min( $purge[self::PURGE_TIME] - $now, self::TINY_NEGATIVE );
2999 $res[self::RES_CUR_TTL] = $curTTL;
3000 $res[self::RES_TOMB_AS_OF] = $purge[self::PURGE_TIME];
3011 private function determineKeyClassForStats( $key ) {
3012 $parts = explode(
':', $key, 3 );
3015 return strtr( $parts[1] ?? $parts[0],
'.',
'_' );
3026 private function parsePurgeValue( $value ) {
3027 if ( !is_string( $value ) ) {
3031 $segments = explode(
':', $value, 3 );
3032 $prefix = $segments[0];
3033 if ( $prefix !== self::PURGE_VAL_PREFIX ) {
3038 $timestamp = (float)$segments[1];
3040 $holdoff = isset( $segments[2] ) ? (int)$segments[2] : self::HOLDOFF_TTL;
3042 if ( $timestamp < $this->epoch ) {
3047 return [ self::PURGE_TIME => $timestamp, self::PURGE_HOLDOFF => $holdoff ];
3054 private function makeTombstonePurgeValue(
float $timestamp ) {
3055 return self::PURGE_VAL_PREFIX .
':' . (int)$timestamp;
3064 private function makeCheckPurgeValue(
float $timestamp,
int $holdoff, array &$purge =
null ) {
3065 $normalizedTime = (int)$timestamp;
3067 $purge = [ self::PURGE_TIME => (float)$normalizedTime, self::PURGE_HOLDOFF => $holdoff ];
3069 return self::PURGE_VAL_PREFIX .
":$normalizedTime:$holdoff";
3076 private function getProcessCache( $group ) {
3077 if ( !isset( $this->processCaches[$group] ) ) {
3078 [ , $size ] = explode(
':', $group );
3079 $this->processCaches[$group] =
new MapCacheLRU( (
int)$size );
3080 if ( $this->wallClockOverride !==
null ) {
3081 $this->processCaches[$group]->setMockTime( $this->wallClockOverride );
3085 return $this->processCaches[$group];
3093 private function getNonProcessCachedMultiKeys( ArrayIterator
$keys, array $opts ) {
3094 $pcTTL = $opts[
'pcTTL'] ?? self::TTL_UNCACHEABLE;
3097 if ( $pcTTL > 0 && $this->callbackDepth == 0 ) {
3098 $pCache = $this->getProcessCache( $opts[
'pcGroup'] ?? self::PC_PRIMARY );
3099 foreach (
$keys as $key => $id ) {
3100 if ( !$pCache->has( $key, $pcTTL ) ) {
3101 $keysMissing[$id] = $key;
3106 return $keysMissing;
3115 private function fetchWrappedValuesForWarmupCache( array
$keys, array $checkKeys ) {
3121 $sisterKeys = $this->makeSisterKeys(
$keys, self::TYPE_VALUE );
3123 foreach ( $checkKeys as $i => $checkKeyOrKeyGroup ) {
3125 if ( is_int( $i ) ) {
3127 $sisterKeys[] = $this->makeSisterKey( $checkKeyOrKeyGroup, self::TYPE_TIMESTAMP );
3130 foreach ( (array)$checkKeyOrKeyGroup as $checkKey ) {
3131 $sisterKeys[] = $this->makeSisterKey( $checkKey, self::TYPE_TIMESTAMP );
3136 $wrappedBySisterKey = $this->cache->getMulti( $sisterKeys );
3137 $wrappedBySisterKey += array_fill_keys( $sisterKeys,
false );
3139 return $wrappedBySisterKey;
3147 private function timeSinceLoggedMiss( $key, $now ) {
3148 for ( end( $this->missLog ); $miss = current( $this->missLog ); prev( $this->missLog ) ) {
3149 if ( $miss[0] === $key ) {
3150 return ( $now - $miss[1] );
3162 return $this->wallClockOverride ?: microtime(
true );
3170 $this->wallClockOverride =& $time;
3171 $this->cache->setMockTime( $time );
3172 foreach ( $this->processCaches as $pCache ) {
3173 $pCache->setMockTime( $time );
A BagOStuff object with no objects in it.
Handles a simple LRU key/value map with a maximum number of entries.
Multi-datacenter aware caching interface.
makeGlobalKey( $collection,... $components)
Make a cache key for the global keyspace and given components.
const HOLDOFF_TTL
Seconds to tombstone keys on delete() and to treat keys as volatile after purges.
const KEY_VERSION
Version number attribute for a key; keep value for b/c (< 1.36)
__construct(array $params)
isValid( $value, $asOf, $minAsOf)
Check that a wrapper value exists and has an acceptable age.
worthRefreshPopular( $asOf, $ageNew, $timeTillRefresh, $now)
Check if a key is due for randomized regeneration due to its popularity.
multiRemap(array $ids, array $res)
Get an (ID => value) map from (i) a non-unique list of entity IDs, and (ii) the list of corresponding...
relayVolatilePurge(string $sisterKey, string $purgeValue, int $ttl)
Set a sister key to a purge value in all datacenters.
prependRoute(string $sisterKey, string $route)
touchCheckKey( $key, $holdoff=self::HOLDOFF_TTL)
Increase the last-purge timestamp of a "check" key in all datacenters.
adaptiveTTL( $mtime, $maxTTL, $minTTL=30, $factor=0.2)
Get a TTL that is higher for objects that have not changed recently.
const GRACE_TTL_NONE
Idiom for set()/getWithSetCallback() meaning "no post-expiration grace period".
BagOStuff $cache
The local datacenter cache.
fetchKeys(array $keys, array $checkKeys, $touchedCb=null)
Fetch the value and key metadata of several keys from cache.
getWithSetCallback( $key, $ttl, $callback, array $opts=[], array $cbParams=[])
Method to fetch/regenerate a cache key.
getCheckKeyTime( $key)
Fetch the value of a timestamp "check" key.
getMulti(array $keys, &$curTTLs=[], array $checkKeys=[], &$info=[])
Fetch the value of several keys from cache.
getMultiWithUnionSetCallback(ArrayIterator $keyedIds, $ttl, callable $callback, array $opts=[])
Method to fetch/regenerate multiple cache keys at once.
relayNonVolatilePurge(string $sisterKey)
Remove a sister key from all datacenters.
getMultiWithSetCallback(ArrayIterator $keyedIds, $ttl, callable $callback, array $opts=[])
Method to fetch multiple cache keys at once with regeneration.
makeMultiKeys(array $ids, $keyCallback)
Get an iterator of (cache key => entity ID) for a list of entity IDs.
static newEmpty()
Get an instance that wraps EmptyBagOStuff.
int $coalesceScheme
Scheme to use for key coalescing (Hash Tags or Hash Stops)
isLotteryRefreshDue( $res, $lowTTL, $ageNew, $hotTTR, $now)
Check if a key is due for randomized regeneration due to near-expiration/popularity.
watchErrors()
Get a "watch point" token that can be used to get the "last error" to occur after now.
bool $useInterimHoldOffCaching
Whether to use "interim" caching while keys are tombstoned.
worthRefreshExpiring( $curTTL, $logicalTTL, $lowTTL)
Check if a key is nearing expiration and thus due for randomized regeneration.
const HOLDOFF_TTL_NONE
Idiom for delete()/touchCheckKey() meaning "no hold-off period".
useInterimHoldOffCaching( $enabled)
Enable or disable the use of brief caching for tombstoned keys.
StatsdDataFactoryInterface $stats
clearProcessCache()
Clear the in-process caches; useful for testing.
const KEY_AS_OF
Generation completion timestamp attribute for a key; keep value for b/c (< 1.36)
getLastError( $watchPoint=0)
Get the "last error" registry.
makeKey( $collection,... $components)
Make a cache key using the "global" keyspace for the given components.
float $epoch
Unix timestamp of the oldest possible valid values.
callable null $asyncHandler
Function that takes a WAN cache callback and runs it later.
string null $broadcastRoute
Routing prefix for operations that should be broadcasted to all data centers.
setLogger(LoggerInterface $logger)
static getCollectionFromSisterKey(string $sisterKey)
const PASS_BY_REF
Idiom for get()/getMulti() to return extra information by reference.
const KEY_CHECK_AS_OF
Highest "check" key timestamp for a key; keep value for b/c (< 1.36)
clearLastError()
Clear the "last error" registry.
const STALE_TTL_NONE
Idiom for set()/getWithSetCallback() meaning "no post-expiration persistence".
MapCacheLRU[] $processCaches
Map of group PHP instance caches.
string $secret
Stable secret used for hashing long strings into key components.
resetCheckKey( $key)
Clear the last-purge timestamp of a "check" key in all datacenters.
const KEY_TOMB_AS_OF
Tomstone timestamp attribute for a key; keep value for b/c (< 1.36)
const KEY_CUR_TTL
Remaining TTL attribute for a key; keep value for b/c (< 1.36)
const TTL_LAGGED
Max TTL, in seconds, to store keys when a data source has high replication lag.
hash256( $component)
Hash a possibly long string into a suitable component for makeKey()/makeGlobalKey()
getMultiCheckKeyTime(array $keys)
Fetch the values of each timestamp "check" key.
const KEY_TTL
Logical TTL attribute for a key.
Generic interface for object stores with key encoding methods.