WANObjectCache.php

Go to the documentation of this file.

<?php
use Liuggio\StatsdClient\Factory\StatsdDataFactoryInterface;
use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;

class WANObjectCache implements IExpiringStore, LoggerAwareInterface {
    protected $cache;
    protected $processCaches = [];
    protected $mcrouterAware;
    protected $region;
    protected $cluster;
    protected $logger;
    protected $stats;
    protected $useInterimHoldOffCaching = true;
    protected $asyncHandler;
    protected $epoch;

    private $callbackDepth = 0;
    private $warmupCache = [];
    private $warmupKeyMisses = 0;

    private $wallClockOverride;

    const MAX_COMMIT_DELAY = 3;
    const MAX_READ_LAG = 7;
    const HOLDOFF_TTL = 11; // MAX_COMMIT_DELAY + MAX_READ_LAG + 1

    const CHECK_KEY_TTL = self::TTL_YEAR;
    const INTERIM_KEY_TTL = 1;

    const LOCK_TTL = 10;
    const COOLOFF_TTL = 1;
    const LOW_TTL = 30;

    const AGE_NEW = 60;
    const HOT_TTR = 900;
    const HIT_RATE_HIGH = 1;
    const RAMPUP_TTL = 30;

    const TTL_UNCACHEABLE = -1;
    const TSE_NONE = -1;
    const TTL_LAGGED = 30;
    const HOLDOFF_NONE = 0;
    const STALE_TTL_NONE = 0;
    const GRACE_TTL_NONE = 0;

    const MIN_TIMESTAMP_NONE = 0.0;

    const TINY_NEGATIVE = -0.000001;
    const TINY_POSTIVE = 0.000001;

    const SET_DELAY_HIGH_MS = 50;
    const RECENT_SET_LOW_MS = 50;
    const RECENT_SET_HIGH_MS = 100;

    const PASS_BY_REF = -1;

    const VERSION = 1;

    const FLD_VERSION = 0; // key to cache version number
    const FLD_VALUE = 1; // key to the cached value
    const FLD_TTL = 2; // key to the original TTL
    const FLD_TIME = 3; // key to the cache time
    const FLD_FLAGS = 4; // key to the flags bitfield (reserved number)
    const FLD_HOLDOFF = 5; // key to any hold-off TTL

    const VALUE_KEY_PREFIX = 'WANCache:v:';
    const INTERIM_KEY_PREFIX = 'WANCache:i:';
    const TIME_KEY_PREFIX = 'WANCache:t:';
    const MUTEX_KEY_PREFIX = 'WANCache:m:';
    const COOLOFF_KEY_PREFIX = 'WANCache:c:';

    const PURGE_VAL_PREFIX = 'PURGED:';

    const VFLD_DATA = 'WOC:d'; // key to the value of versioned data
    const VFLD_VERSION = 'WOC:v'; // key to the version of the value present

    const PC_PRIMARY = 'primary:1000'; // process cache name and max key count

    public function __construct( array $params ) {
        $this->cache = $params['cache'];
        $this->region = $params['region'] ?? 'main';
        $this->cluster = $params['cluster'] ?? 'wan-main';
        $this->mcrouterAware = !empty( $params['mcrouterAware'] );
        $this->epoch = $params['epoch'] ?? 1.0;

        $this->setLogger( $params['logger'] ?? new NullLogger() );
        $this->stats = $params['stats'] ?? new NullStatsdDataFactory();
        $this->asyncHandler = $params['asyncHandler'] ?? null;
    }

    public function setLogger( LoggerInterface $logger ) {
        $this->logger = $logger;
    }

    public static function newEmpty() {
        return new static( [ 'cache' => new EmptyBagOStuff() ] );
    }

    final public function get(
        $key, &$curTTL = null, array $checkKeys = [], &$info = null
    ) {
        $curTTLs = self::PASS_BY_REF;
        $infoByKey = self::PASS_BY_REF;
        $values = $this->getMulti( [ $key ], $curTTLs, $checkKeys, $infoByKey );
        $curTTL = $curTTLs[$key] ?? null;
        if ( $info === self::PASS_BY_REF ) {
            $info = [
                'asOf' => $infoByKey[$key]['asOf'] ?? null,
                'tombAsOf' => $infoByKey[$key]['tombAsOf'] ?? null,
                'lastCKPurge' => $infoByKey[$key]['lastCKPurge'] ?? null
            ];
        } else {
            $info = $infoByKey[$key]['asOf'] ?? null; // b/c
        }

        return $values[$key] ?? false;
    }

    final public function getMulti(
        array $keys,
        &$curTTLs = [],
        array $checkKeys = [],
        &$info = null
    ) {
        $result = [];
        $curTTLs = [];
        $infoByKey = [];

        $vPrefixLen = strlen( self::VALUE_KEY_PREFIX );
        $valueKeys = self::prefixCacheKeys( $keys, self::VALUE_KEY_PREFIX );

        $checkKeysForAll = [];
        $checkKeysByKey = [];
        $checkKeysFlat = [];
        foreach ( $checkKeys as $i => $checkKeyGroup ) {
            $prefixed = self::prefixCacheKeys( (array)$checkKeyGroup, self::TIME_KEY_PREFIX );
            $checkKeysFlat = array_merge( $checkKeysFlat, $prefixed );
            // Are these check keys for a specific cache key, or for all keys being fetched?
            if ( is_int( $i ) ) {
                $checkKeysForAll = array_merge( $checkKeysForAll, $prefixed );
            } else {
                $checkKeysByKey[$i] = $prefixed;
            }
        }

        // Fetch all of the raw values
        $keysGet = array_merge( $valueKeys, $checkKeysFlat );
        if ( $this->warmupCache ) {
            $wrappedValues = array_intersect_key( $this->warmupCache, array_flip( $keysGet ) );
            $keysGet = array_diff( $keysGet, array_keys( $wrappedValues ) ); // keys left to fetch
            $this->warmupKeyMisses += count( $keysGet );
        } else {
            $wrappedValues = [];
        }
        if ( $keysGet ) {
            $wrappedValues += $this->cache->getMulti( $keysGet );
        }
        // Time used to compare/init "check" keys (derived after getMulti() to be pessimistic)
        $now = $this->getCurrentTime();

        // Collect timestamps from all "check" keys
        $purgeValuesForAll = $this->processCheckKeys( $checkKeysForAll, $wrappedValues, $now );
        $purgeValuesByKey = [];
        foreach ( $checkKeysByKey as $cacheKey => $checks ) {
            $purgeValuesByKey[$cacheKey] =
                $this->processCheckKeys( $checks, $wrappedValues, $now );
        }

        // Get the main cache value for each key and validate them
        foreach ( $valueKeys as $vKey ) {
            $key = substr( $vKey, $vPrefixLen ); // unprefix
            list( $value, $curTTL, $asOf, $tombAsOf ) = isset( $wrappedValues[$vKey] )
                ? $this->unwrap( $wrappedValues[$vKey], $now )
                : [ false, null, null, null ]; // not found
            // Force dependent keys to be seen as stale for a while after purging
            // to reduce race conditions involving stale data getting cached
            $purgeValues = $purgeValuesForAll;
            if ( isset( $purgeValuesByKey[$key] ) ) {
                $purgeValues = array_merge( $purgeValues, $purgeValuesByKey[$key] );
            }

            $lastCKPurge = null; // timestamp of the highest check key
            foreach ( $purgeValues as $purge ) {
                $lastCKPurge = max( $purge[self::FLD_TIME], $lastCKPurge );
                $safeTimestamp = $purge[self::FLD_TIME] + $purge[self::FLD_HOLDOFF];
                if ( $value !== false && $safeTimestamp >= $asOf ) {
                    // How long ago this value was invalidated by *this* check key
                    $ago = min( $purge[self::FLD_TIME] - $now, self::TINY_NEGATIVE );
                    // How long ago this value was invalidated by *any* known check key
                    $curTTL = min( $curTTL, $ago );
                }
            }

            if ( $value !== false ) {
                $result[$key] = $value;
            }
            if ( $curTTL !== null ) {
                $curTTLs[$key] = $curTTL;
            }

            $infoByKey[$key] = ( $info === self::PASS_BY_REF )
                ? [ 'asOf' => $asOf, 'tombAsOf' => $tombAsOf, 'lastCKPurge' => $lastCKPurge ]
                : $asOf; // b/c
        }

        $info = $infoByKey;

        return $result;
    }

    private function processCheckKeys( array $timeKeys, array $wrappedValues, $now ) {
        $purgeValues = [];
        foreach ( $timeKeys as $timeKey ) {
            $purge = isset( $wrappedValues[$timeKey] )
                ? $this->parsePurgeValue( $wrappedValues[$timeKey] )
                : false;
            if ( $purge === false ) {
                // Key is not set or malformed; regenerate
                $newVal = $this->makePurgeValue( $now, self::HOLDOFF_TTL );
                $this->cache->add( $timeKey, $newVal, self::CHECK_KEY_TTL );
                $purge = $this->parsePurgeValue( $newVal );
            }
            $purgeValues[] = $purge;
        }
        return $purgeValues;
    }

    final public function set( $key, $value, $ttl = self::TTL_INDEFINITE, array $opts = [] ) {
        $now = $this->getCurrentTime();
        $lockTSE = $opts['lockTSE'] ?? self::TSE_NONE;
        $staleTTL = $opts['staleTTL'] ?? self::STALE_TTL_NONE;
        $age = isset( $opts['since'] ) ? max( 0, $now - $opts['since'] ) : 0;
        $creating = $opts['creating'] ?? false;
        $lag = $opts['lag'] ?? 0;

        // Do not cache potentially uncommitted data as it might get rolled back
        if ( !empty( $opts['pending'] ) ) {
            $this->logger->info(
                'Rejected set() for {cachekey} due to pending writes.',
                [ 'cachekey' => $key ]
            );

            return true; // no-op the write for being unsafe
        }

        $logicalTTL = null; // logical TTL override
        // Check if there's a risk of writing stale data after the purge tombstone expired
        if ( $lag === false || ( $lag + $age ) > self::MAX_READ_LAG ) {
            // Case A: any long-running transaction
            if ( $age > self::MAX_READ_LAG ) {
                if ( $lockTSE >= 0 ) {
                    // Store value as *almost* stale to avoid cache and mutex stampedes
                    $logicalTTL = self::TTL_SECOND;
                    $this->logger->info(
                        'Lowered set() TTL for {cachekey} due to snapshot lag.',
                        [ 'cachekey' => $key, 'lag' => $lag, 'age' => $age ]
                    );
                } else {
                    $this->logger->info(
                        'Rejected set() for {cachekey} due to snapshot lag.',
                        [ 'cachekey' => $key, 'lag' => $lag, 'age' => $age ]
                    );

                    return true; // no-op the write for being unsafe
                }
            // Case B: high replication lag; lower TTL instead of ignoring all set()s
            } elseif ( $lag === false || $lag > self::MAX_READ_LAG ) {
                if ( $lockTSE >= 0 ) {
                    $logicalTTL = min( $ttl ?: INF, self::TTL_LAGGED );
                } else {
                    $ttl = min( $ttl ?: INF, self::TTL_LAGGED );
                }
                $this->logger->warning(
                    'Lowered set() TTL for {cachekey} due to replication lag.',
                    [ 'cachekey' => $key, 'lag' => $lag, 'age' => $age ]
                );
            // Case C: medium length request with medium replication lag
            } elseif ( $lockTSE >= 0 ) {
                // Store value as *almost* stale to avoid cache and mutex stampedes
                $logicalTTL = self::TTL_SECOND;
                $this->logger->info(
                    'Lowered set() TTL for {cachekey} due to high read lag.',
                    [ 'cachekey' => $key, 'lag' => $lag, 'age' => $age ]
                );
            } else {
                $this->logger->info(
                    'Rejected set() for {cachekey} due to high read lag.',
                    [ 'cachekey' => $key, 'lag' => $lag, 'age' => $age ]
                );

                return true; // no-op the write for being unsafe
            }
        }

        // Wrap that value with time/TTL/version metadata
        $wrapped = $this->wrap( $value, $logicalTTL ?: $ttl, $now );
        $storeTTL = $ttl + $staleTTL;

        if ( $creating ) {
            $ok = $this->cache->add( self::VALUE_KEY_PREFIX . $key, $wrapped, $storeTTL );
        } else {
            $ok = $this->cache->merge(
                self::VALUE_KEY_PREFIX . $key,
                function ( $cache, $key, $cWrapped ) use ( $wrapped ) {
                    // A string value means that it is a tombstone; do nothing in that case
                    return ( is_string( $cWrapped ) ) ? false : $wrapped;
                },
                $storeTTL,
                1 // 1 attempt
            );
        }

        return $ok;
    }

    final public function delete( $key, $ttl = self::HOLDOFF_TTL ) {
        if ( $ttl <= 0 ) {
            // Publish the purge to all datacenters
            $ok = $this->relayDelete( self::VALUE_KEY_PREFIX . $key );
        } else {
            // Publish the purge to all datacenters
            $ok = $this->relayPurge( self::VALUE_KEY_PREFIX . $key, $ttl, self::HOLDOFF_NONE );
        }

        $kClass = $this->determineKeyClassForStats( $key );
        $this->stats->increment( "wanobjectcache.$kClass.delete." . ( $ok ? 'ok' : 'error' ) );

        return $ok;
    }

    final public function getCheckKeyTime( $key ) {
        return $this->getMultiCheckKeyTime( [ $key ] )[$key];
    }

    final public function getMultiCheckKeyTime( array $keys ) {
        $rawKeys = [];
        foreach ( $keys as $key ) {
            $rawKeys[$key] = self::TIME_KEY_PREFIX . $key;
        }

        $rawValues = $this->cache->getMulti( $rawKeys );
        $rawValues += array_fill_keys( $rawKeys, false );

        $times = [];
        foreach ( $rawKeys as $key => $rawKey ) {
            $purge = $this->parsePurgeValue( $rawValues[$rawKey] );
            if ( $purge !== false ) {
                $time = $purge[self::FLD_TIME];
            } else {
                // Casting assures identical floats for the next getCheckKeyTime() calls
                $now = (string)$this->getCurrentTime();
                $this->cache->add(
                    $rawKey,
                    $this->makePurgeValue( $now, self::HOLDOFF_TTL ),
                    self::CHECK_KEY_TTL
                );
                $time = (float)$now;
            }

            $times[$key] = $time;
        }

        return $times;
    }

    final public function touchCheckKey( $key, $holdoff = self::HOLDOFF_TTL ) {
        // Publish the purge to all datacenters
        $ok = $this->relayPurge( self::TIME_KEY_PREFIX . $key, self::CHECK_KEY_TTL, $holdoff );

        $kClass = $this->determineKeyClassForStats( $key );
        $this->stats->increment( "wanobjectcache.$kClass.ck_touch." . ( $ok ? 'ok' : 'error' ) );

        return $ok;
    }

    final public function resetCheckKey( $key ) {
        // Publish the purge to all datacenters
        $ok = $this->relayDelete( self::TIME_KEY_PREFIX . $key );

        $kClass = $this->determineKeyClassForStats( $key );
        $this->stats->increment( "wanobjectcache.$kClass.ck_reset." . ( $ok ? 'ok' : 'error' ) );

        return $ok;
    }

    final public function getWithSetCallback( $key, $ttl, $callback, array $opts = [] ) {
        $pcTTL = $opts['pcTTL'] ?? self::TTL_UNCACHEABLE;

        // Try the process cache if enabled and the cache callback is not within a cache callback.
        // Process cache use in nested callbacks is not lag-safe with regard to HOLDOFF_TTL since
        // the in-memory value is further lagged than the shared one since it uses a blind TTL.
        if ( $pcTTL >= 0 && $this->callbackDepth == 0 ) {
            $group = $opts['pcGroup'] ?? self::PC_PRIMARY;
            $procCache = $this->getProcessCache( $group );
            $value = $procCache->has( $key, $pcTTL ) ? $procCache->get( $key ) : false;
        } else {
            $procCache = false;
            $value = false;
        }

        if ( $value === false ) {
            // Fetch the value over the network
            if ( isset( $opts['version'] ) ) {
                $version = $opts['version'];
                $asOf = null;
                $cur = $this->doGetWithSetCallback(
                    $key,
                    $ttl,
                    function ( $oldValue, &$ttl, &$setOpts, $oldAsOf )
                    use ( $callback, $version ) {
                        if ( is_array( $oldValue )
                            && array_key_exists( self::VFLD_DATA, $oldValue )
                            && array_key_exists( self::VFLD_VERSION, $oldValue )
                            && $oldValue[self::VFLD_VERSION] === $version
                        ) {
                            $oldData = $oldValue[self::VFLD_DATA];
                        } else {
                            // VFLD_DATA is not set if an old, unversioned, key is present
                            $oldData = false;
                            $oldAsOf = null;
                        }

                        return [
                            self::VFLD_DATA => $callback( $oldData, $ttl, $setOpts, $oldAsOf ),
                            self::VFLD_VERSION => $version
                        ];
                    },
                    $opts,
                    $asOf
                );
                if ( $cur[self::VFLD_VERSION] === $version ) {
                    // Value created or existed before with version; use it
                    $value = $cur[self::VFLD_DATA];
                } else {
                    // Value existed before with a different version; use variant key.
                    // Reflect purges to $key by requiring that this key value be newer.
                    $value = $this->doGetWithSetCallback(
                        $this->makeGlobalKey( 'WANCache-key-variant', md5( $key ), $version ),
                        $ttl,
                        $callback,
                        // Regenerate value if not newer than $key
                        [ 'version' => null, 'minAsOf' => $asOf ] + $opts
                    );
                }
            } else {
                $value = $this->doGetWithSetCallback( $key, $ttl, $callback, $opts );
            }

            // Update the process cache if enabled
            if ( $procCache && $value !== false ) {
                $procCache->set( $key, $value );
            }
        }

        return $value;
    }

    protected function doGetWithSetCallback( $key, $ttl, $callback, array $opts, &$asOf = null ) {
        $lowTTL = $opts['lowTTL'] ?? min( self::LOW_TTL, $ttl );
        $lockTSE = $opts['lockTSE'] ?? self::TSE_NONE;
        $staleTTL = $opts['staleTTL'] ?? self::STALE_TTL_NONE;
        $graceTTL = $opts['graceTTL'] ?? self::GRACE_TTL_NONE;
        $checkKeys = $opts['checkKeys'] ?? [];
        $busyValue = $opts['busyValue'] ?? null;
        $popWindow = $opts['hotTTR'] ?? self::HOT_TTR;
        $ageNew = $opts['ageNew'] ?? self::AGE_NEW;
        $minTime = $opts['minAsOf'] ?? self::MIN_TIMESTAMP_NONE;
        $needsVersion = isset( $opts['version'] );
        $touchedCb = $opts['touchedCallback'] ?? null;
        $initialTime = $this->getCurrentTime();

        $kClass = $this->determineKeyClassForStats( $key );

        // Get the current key value
        $curTTL = self::PASS_BY_REF;
        $curInfo = self::PASS_BY_REF; 
        $curValue = $this->get( $key, $curTTL, $checkKeys, $curInfo );
        // Apply any $touchedCb invalidation timestamp to get the "last purge timestamp"
        list( $curTTL, $LPT ) = $this->resolveCTL( $curValue, $curTTL, $curInfo, $touchedCb );
        // Keep track of the best candidate value and its timestamp
        $value = $curValue; // return value
        $asOf = $curInfo['asOf']; // return value timestamp

        // Determine if a cached value regeneration is needed or desired
        if (
            $this->isValid( $value, $needsVersion, $asOf, $minTime ) &&
            $this->isAliveOrInGracePeriod( $curTTL, $graceTTL )
        ) {
            $preemptiveRefresh = (
                $this->worthRefreshExpiring( $curTTL, $lowTTL ) ||
                $this->worthRefreshPopular( $asOf, $ageNew, $popWindow, $initialTime )
            );

            if ( !$preemptiveRefresh ) {
                $this->stats->increment( "wanobjectcache.$kClass.hit.good" );

                return $value;
            } elseif ( $this->scheduleAsyncRefresh( $key, $ttl, $callback, $opts ) ) {
                $this->stats->increment( "wanobjectcache.$kClass.hit.refresh" );

                return $value;
            }
        }

        $isKeyTombstoned = ( $curInfo['tombAsOf'] !== null );
        if ( $isKeyTombstoned ) {
            // Get the interim key value since the key is tombstoned (write-holed)
            list( $value, $asOf ) = $this->getInterimValue( $key, $needsVersion, $minTime );
            // Update the "last purge time" since the $touchedCb timestamp depends on $value
            $LPT = $this->resolveTouched( $value, $LPT, $touchedCb );
        }

        // Reduce mutex and cache set spam while keys are in the tombstone/holdoff period by
        // checking if $value was genereated by a recent thread much less than a second ago.
        if (
            $this->isValid( $value, $needsVersion, $asOf, $minTime, $LPT ) &&
            $this->isVolatileValueAgeNegligible( $initialTime - $asOf )
        ) {
            $this->stats->increment( "wanobjectcache.$kClass.hit.volatile" );

            return $value;
        }

        // Decide if only one thread should handle regeneration at a time
        $useMutex =
            // Note that since tombstones no-op set(), $lockTSE and $curTTL cannot be used to
            // deduce the key hotness because |$curTTL| will always keep increasing until the
            // tombstone expires or is overwritten by a new tombstone. Also, even if $lockTSE
            // is not set, constant regeneration of a key for the tombstone lifetime might be
            // very expensive. Assume tombstoned keys are possibly hot in order to reduce
            // the risk of high regeneration load after the delete() method is called.
            $isKeyTombstoned ||
            // Assume a key is hot if requested soon ($lockTSE seconds) after invalidation.
            // This avoids stampedes when timestamps from $checkKeys/$touchedCb bump.
            ( $curTTL !== null && $curTTL <= 0 && abs( $curTTL ) <= $lockTSE ) ||
            // Assume a key is hot if there is no value and a busy fallback is given.
            // This avoids stampedes on eviction or preemptive regeneration taking too long.
            ( $busyValue !== null && $value === false );

        $hasLock = false;
        if ( $useMutex ) {
            // Acquire a datacenter-local non-blocking lock
            if ( $this->cache->add( self::MUTEX_KEY_PREFIX . $key, 1, self::LOCK_TTL ) ) {
                // Lock acquired; this thread will recompute the value and update cache
                $hasLock = true;
            } elseif ( $this->isValid( $value, $needsVersion, $asOf, $minTime ) ) {
                // Lock not acquired and a stale value exists; use the stale value
                $this->stats->increment( "wanobjectcache.$kClass.hit.stale" );

                return $value;
            } else {
                // Lock not acquired and no stale value exists
                if ( $busyValue !== null ) {
                    // Use the busy fallback value if nothing else
                    $miss = is_infinite( $minTime ) ? 'renew' : 'miss';
                    $this->stats->increment( "wanobjectcache.$kClass.$miss.busy" );

                    return is_callable( $busyValue ) ? $busyValue() : $busyValue;
                }
            }
        }

        if ( !is_callable( $callback ) ) {
            throw new InvalidArgumentException( "Invalid cache miss callback provided." );
        }

        $preCallbackTime = $this->getCurrentTime();
        // Generate the new value from the callback...
        $setOpts = [];
        ++$this->callbackDepth;
        try {
            $value = call_user_func_array( $callback, [ $curValue, &$ttl, &$setOpts, $asOf ] );
        } finally {
            --$this->callbackDepth;
        }
        $valueIsCacheable = ( $value !== false && $ttl >= 0 );

        if ( $valueIsCacheable ) {
            $ago = max( $this->getCurrentTime() - $initialTime, 0.0 );
            $this->stats->timing( "wanobjectcache.$kClass.regen_set_delay", 1000 * $ago );

            if ( $isKeyTombstoned ) {
                if ( $this->checkAndSetCooloff( $key, $kClass, $ago, $lockTSE, $hasLock ) ) {
                    // Use the interim key value since the key is tombstoned (write-holed)
                    $tempTTL = max( self::INTERIM_KEY_TTL, (int)$lockTSE );
                    $this->setInterimValue( $key, $value, $tempTTL, $this->getCurrentTime() );
                }
            } elseif ( !$useMutex || $hasLock ) {
                if ( $this->checkAndSetCooloff( $key, $kClass, $ago, $lockTSE, $hasLock ) ) {
                    $setOpts['creating'] = ( $curValue === false );
                    // Save the value unless a lock-winning thread is already expected to do that
                    $setOpts['lockTSE'] = $lockTSE;
                    $setOpts['staleTTL'] = $staleTTL;
                    // Use best known "since" timestamp if not provided
                    $setOpts += [ 'since' => $preCallbackTime ];
                    // Update the cache; this will fail if the key is tombstoned
                    $this->set( $key, $value, $ttl, $setOpts );
                }
            }
        }

        if ( $hasLock ) {
            $this->cache->changeTTL( self::MUTEX_KEY_PREFIX . $key, (int)$initialTime - 60 );
        }

        $miss = is_infinite( $minTime ) ? 'renew' : 'miss';
        $this->stats->increment( "wanobjectcache.$kClass.$miss.compute" );

        return $value;
    }

    private function isVolatileValueAgeNegligible( $age ) {
        return ( $age < mt_rand( self::RECENT_SET_LOW_MS, self::RECENT_SET_HIGH_MS ) / 1e3 );
    }

    private function checkAndSetCooloff( $key, $kClass, $elapsed, $lockTSE, $hasLock ) {
        // If $lockTSE is set, the lock was bypassed because there was no stale/interim value,
        // and $elapsed indicates that regeration is slow, then there is a risk of set()
        // stampedes with large blobs. With a typical scale-out infrastructure, CPU and query
        // load from $callback invocations is distributed among appservers and replica DBs,
        // but cache operations for a given key route to a single cache server (e.g. striped
        // consistent hashing).
        if ( $lockTSE < 0 || $hasLock ) {
            return true; // either not a priori hot or thread has the lock
        } elseif ( $elapsed <= self::SET_DELAY_HIGH_MS * 1e3 ) {
            return true; // not enough time for threads to pile up
        }

        $this->cache->clearLastError();
        if (
            !$this->cache->add( self::COOLOFF_KEY_PREFIX . $key, 1, self::COOLOFF_TTL ) &&
            // Don't treat failures due to I/O errors as the key being in cooloff
            $this->cache->getLastError() === BagOStuff::ERR_NONE
        ) {
            $this->stats->increment( "wanobjectcache.$kClass.cooloff_bounce" );

            return false;
        }

        return true;
    }

    protected function resolveCTL( $value, $curTTL, $curInfo, $touchedCallback ) {
        if ( $touchedCallback === null || $value === false ) {
            return [ $curTTL, max( $curInfo['tombAsOf'], $curInfo['lastCKPurge'] ) ];
        }

        if ( !is_callable( $touchedCallback ) ) {
            throw new InvalidArgumentException( "Invalid expiration callback provided." );
        }

        $touched = $touchedCallback( $value );
        if ( $touched !== null && $touched >= $curInfo['asOf'] ) {
            $curTTL = min( $curTTL, self::TINY_NEGATIVE, $curInfo['asOf'] - $touched );
        }

        return [ $curTTL, max( $curInfo['tombAsOf'], $curInfo['lastCKPurge'], $touched ) ];
    }

    protected function resolveTouched( $value, $lastPurge, $touchedCallback ) {
        if ( $touchedCallback === null || $value === false ) {
            return $lastPurge;
        }

        if ( !is_callable( $touchedCallback ) ) {
            throw new InvalidArgumentException( "Invalid expiration callback provided." );
        }

        return max( $touchedCallback( $value ), $lastPurge );
    }

    protected function getInterimValue( $key, $versioned, $minTime ) {
        if ( !$this->useInterimHoldOffCaching ) {
            return [ false, null ]; // disabled
        }

        $wrapped = $this->cache->get( self::INTERIM_KEY_PREFIX . $key );
        list( $value ) = $this->unwrap( $wrapped, $this->getCurrentTime() );
        $valueAsOf = $wrapped[self::FLD_TIME] ?? null;
        if ( $this->isValid( $value, $versioned, $valueAsOf, $minTime ) ) {
            return [ $value, $valueAsOf ];
        }

        return [ false, null ];
    }

    protected function setInterimValue( $key, $value, $tempTTL, $newAsOf ) {
        $wrapped = $this->wrap( $value, $tempTTL, $newAsOf );

        $this->cache->merge(
            self::INTERIM_KEY_PREFIX . $key,
            function () use ( $wrapped ) {
                return $wrapped;
            },
            $tempTTL,
            1
        );
    }

    final public function getMultiWithSetCallback(
        ArrayIterator $keyedIds, $ttl, callable $callback, array $opts = []
    ) {
        $valueKeys = array_keys( $keyedIds->getArrayCopy() );
        $checkKeys = $opts['checkKeys'] ?? [];
        $pcTTL = $opts['pcTTL'] ?? self::TTL_UNCACHEABLE;

        // Load required keys into process cache in one go
        $this->warmupCache = $this->getRawKeysForWarmup(
            $this->getNonProcessCachedKeys( $valueKeys, $opts, $pcTTL ),
            $checkKeys
        );
        $this->warmupKeyMisses = 0;

        // Wrap $callback to match the getWithSetCallback() format while passing $id to $callback
        $id = null; // current entity ID
        $func = function ( $oldValue, &$ttl, &$setOpts, $oldAsOf ) use ( $callback, &$id ) {
            return $callback( $id, $oldValue, $ttl, $setOpts, $oldAsOf );
        };

        $values = [];
        foreach ( $keyedIds as $key => $id ) { // preserve order
            $values[$key] = $this->getWithSetCallback( $key, $ttl, $func, $opts );
        }

        $this->warmupCache = [];

        return $values;
    }

    final public function getMultiWithUnionSetCallback(
        ArrayIterator $keyedIds, $ttl, callable $callback, array $opts = []
    ) {
        $idsByValueKey = $keyedIds->getArrayCopy();
        $valueKeys = array_keys( $idsByValueKey );
        $checkKeys = $opts['checkKeys'] ?? [];
        $pcTTL = $opts['pcTTL'] ?? self::TTL_UNCACHEABLE;
        unset( $opts['lockTSE'] ); // incompatible
        unset( $opts['busyValue'] ); // incompatible

        // Load required keys into process cache in one go
        $keysGet = $this->getNonProcessCachedKeys( $valueKeys, $opts, $pcTTL );
        $this->warmupCache = $this->getRawKeysForWarmup( $keysGet, $checkKeys );
        $this->warmupKeyMisses = 0;

        // IDs of entities known to be in need of regeneration
        $idsRegen = [];

        // Find out which keys are missing/deleted/stale
        $curTTLs = [];
        $asOfs = [];
        $curByKey = $this->getMulti( $keysGet, $curTTLs, $checkKeys, $asOfs );
        foreach ( $keysGet as $key ) {
            if ( !array_key_exists( $key, $curByKey ) || $curTTLs[$key] < 0 ) {
                $idsRegen[] = $idsByValueKey[$key];
            }
        }

        // Run the callback to populate the regeneration value map for all required IDs
        $newSetOpts = [];
        $newTTLsById = array_fill_keys( $idsRegen, $ttl );
        $newValsById = $idsRegen ? $callback( $idsRegen, $newTTLsById, $newSetOpts ) : [];

        // Wrap $callback to match the getWithSetCallback() format while passing $id to $callback
        $id = null; // current entity ID
        $func = function ( $oldValue, &$ttl, &$setOpts, $oldAsOf )
            use ( $callback, &$id, $newValsById, $newTTLsById, $newSetOpts )
        {
            if ( array_key_exists( $id, $newValsById ) ) {
                // Value was already regerated as expected, so use the value in $newValsById
                $newValue = $newValsById[$id];
                $ttl = $newTTLsById[$id];
                $setOpts = $newSetOpts;
            } else {
                // Pre-emptive/popularity refresh and version mismatch cases are not detected
                // above and thus $newValsById has no entry. Run $callback on this single entity.
                $ttls = [ $id => $ttl ];
                $newValue = $callback( [ $id ], $ttls, $setOpts )[$id];
                $ttl = $ttls[$id];
            }

            return $newValue;
        };

        // Run the cache-aside logic using warmupCache instead of persistent cache queries
        $values = [];
        foreach ( $idsByValueKey as $key => $id ) { // preserve order
            $values[$key] = $this->getWithSetCallback( $key, $ttl, $func, $opts );
        }

        $this->warmupCache = [];

        return $values;
    }

    final public function reap( $key, $purgeTimestamp, &$isStale = false ) {
        $minAsOf = $purgeTimestamp + self::HOLDOFF_TTL;
        $wrapped = $this->cache->get( self::VALUE_KEY_PREFIX . $key );
        if ( is_array( $wrapped ) && $wrapped[self::FLD_TIME] < $minAsOf ) {
            $isStale = true;
            $this->logger->warning( "Reaping stale value key '$key'." );
            $ttlReap = self::HOLDOFF_TTL; // avoids races with tombstone creation
            $ok = $this->cache->changeTTL( self::VALUE_KEY_PREFIX . $key, $ttlReap );
            if ( !$ok ) {
                $this->logger->error( "Could not complete reap of key '$key'." );
            }

            return $ok;
        }

        $isStale = false;

        return true;
    }

    final public function reapCheckKey( $key, $purgeTimestamp, &$isStale = false ) {
        $purge = $this->parsePurgeValue( $this->cache->get( self::TIME_KEY_PREFIX . $key ) );
        if ( $purge && $purge[self::FLD_TIME] < $purgeTimestamp ) {
            $isStale = true;
            $this->logger->warning( "Reaping stale check key '$key'." );
            $ok = $this->cache->changeTTL( self::TIME_KEY_PREFIX . $key, self::TTL_SECOND );
            if ( !$ok ) {
                $this->logger->error( "Could not complete reap of check key '$key'." );
            }

            return $ok;
        }

        $isStale = false;

        return false;
    }

    public function makeKey( $class, $component = null ) {
        return $this->cache->makeKey( ...func_get_args() );
    }

    public function makeGlobalKey( $class, $component = null ) {
        return $this->cache->makeGlobalKey( ...func_get_args() );
    }

    final public function makeMultiKeys( array $entities, callable $keyFunc ) {
        $map = [];
        foreach ( $entities as $entity ) {
            $map[$keyFunc( $entity, $this )] = $entity;
        }

        return new ArrayIterator( $map );
    }

    final public function getLastError() {
        $code = $this->cache->getLastError();
        switch ( $code ) {
            case BagOStuff::ERR_NONE:
                return self::ERR_NONE;
            case BagOStuff::ERR_NO_RESPONSE:
                return self::ERR_NO_RESPONSE;
            case BagOStuff::ERR_UNREACHABLE:
                return self::ERR_UNREACHABLE;
            default:
                return self::ERR_UNEXPECTED;
        }
    }

    final public function clearLastError() {
        $this->cache->clearLastError();
    }

    public function clearProcessCache() {
        $this->processCaches = [];
    }

    final public function useInterimHoldOffCaching( $enabled ) {
        $this->useInterimHoldOffCaching = $enabled;
    }

    public function getQoS( $flag ) {
        return $this->cache->getQoS( $flag );
    }

    public function adaptiveTTL( $mtime, $maxTTL, $minTTL = 30, $factor = 0.2 ) {
        if ( is_float( $mtime ) || ctype_digit( $mtime ) ) {
            $mtime = (int)$mtime; // handle fractional seconds and string integers
        }

        if ( !is_int( $mtime ) || $mtime <= 0 ) {
            return $minTTL; // no last-modified time provided
        }

        $age = $this->getCurrentTime() - $mtime;

        return (int)min( $maxTTL, max( $minTTL, $factor * $age ) );
    }

    final public function getWarmupKeyMisses() {
        return $this->warmupKeyMisses;
    }

    protected function relayPurge( $key, $ttl, $holdoff ) {
        if ( $this->mcrouterAware ) {
            // See https://github.com/facebook/mcrouter/wiki/Multi-cluster-broadcast-setup
            // Wildcards select all matching routes, e.g. the WAN cluster on all DCs
            $ok = $this->cache->set(
                "/*/{$this->cluster}/{$key}",
                $this->makePurgeValue( $this->getCurrentTime(), self::HOLDOFF_NONE ),
                $ttl
            );
        } else {
            // This handles the mcrouter and the single-DC case
            $ok = $this->cache->set(
                $key,
                $this->makePurgeValue( $this->getCurrentTime(), self::HOLDOFF_NONE ),
                $ttl
            );
        }

        return $ok;
    }

    protected function relayDelete( $key ) {
        if ( $this->mcrouterAware ) {
            // See https://github.com/facebook/mcrouter/wiki/Multi-cluster-broadcast-setup
            // Wildcards select all matching routes, e.g. the WAN cluster on all DCs
            $ok = $this->cache->delete( "/*/{$this->cluster}/{$key}" );
        } else {
            // Some other proxy handles broadcasting or there is only one datacenter
            $ok = $this->cache->delete( $key );
        }

        return $ok;
    }

    private function scheduleAsyncRefresh( $key, $ttl, $callback, $opts ) {
        if ( !$this->asyncHandler ) {
            return false;
        }
        // Update the cache value later, such during post-send of an HTTP request
        $func = $this->asyncHandler;
        $func( function () use ( $key, $ttl, $callback, $opts ) {
            $asOf = null; // unused
            $opts['minAsOf'] = INF; // force a refresh
            $this->doGetWithSetCallback( $key, $ttl, $callback, $opts, $asOf );
        } );

        return true;
    }

    protected function isAliveOrInGracePeriod( $curTTL, $graceTTL ) {
        if ( $curTTL > 0 ) {
            return true;
        } elseif ( $graceTTL <= 0 ) {
            return false;
        }

        $ageStale = abs( $curTTL ); // seconds of staleness
        $curGTTL = ( $graceTTL - $ageStale ); // current grace-time-to-live
        if ( $curGTTL <= 0 ) {
            return false; //  already out of grace period
        }

        // Chance of using a stale value is the complement of the chance of refreshing it
        return !$this->worthRefreshExpiring( $curGTTL, $graceTTL );
    }

    protected function worthRefreshExpiring( $curTTL, $lowTTL ) {
        if ( $lowTTL <= 0 ) {
            return false;
        } elseif ( $curTTL >= $lowTTL ) {
            return false;
        } elseif ( $curTTL <= 0 ) {
            return false;
        }

        $chance = ( 1 - $curTTL / $lowTTL );

        return mt_rand( 1, 1e9 ) <= 1e9 * $chance;
    }

    protected function worthRefreshPopular( $asOf, $ageNew, $timeTillRefresh, $now ) {
        if ( $ageNew < 0 || $timeTillRefresh <= 0 ) {
            return false;
        }

        $age = $now - $asOf;
        $timeOld = $age - $ageNew;
        if ( $timeOld <= 0 ) {
            return false;
        }

        // Lifecycle is: new, ramp-up refresh chance, full refresh chance.
        // Note that the "expected # of refreshes" for the ramp-up time range is half of what it
        // would be if P(refresh) was at its full value during that time range.
        $refreshWindowSec = max( $timeTillRefresh - $ageNew - self::RAMPUP_TTL / 2, 1 );
        // P(refresh) * (# hits in $refreshWindowSec) = (expected # of refreshes)
        // P(refresh) * ($refreshWindowSec * $popularHitsPerSec) = 1
        // P(refresh) = 1/($refreshWindowSec * $popularHitsPerSec)
        $chance = 1 / ( self::HIT_RATE_HIGH * $refreshWindowSec );

        // Ramp up $chance from 0 to its nominal value over RAMPUP_TTL seconds to avoid stampedes
        $chance *= ( $timeOld <= self::RAMPUP_TTL ) ? $timeOld / self::RAMPUP_TTL : 1;

        return mt_rand( 1, 1e9 ) <= 1e9 * $chance;
    }

    protected function isValid( $value, $versioned, $asOf, $minTime, $purgeTime = null ) {
        // Avoid reading any key not generated after the latest delete() or touch
        $safeMinTime = max( $minTime, $purgeTime + self::TINY_POSTIVE );

        if ( $value === false ) {
            return false;
        } elseif ( $versioned && !isset( $value[self::VFLD_VERSION] ) ) {
            return false;
        } elseif ( $safeMinTime > 0 && $asOf < $minTime ) {
            return false;
        }

        return true;
    }

    protected function wrap( $value, $ttl, $now ) {
        return [
            self::FLD_VERSION => self::VERSION,
            self::FLD_VALUE => $value,
            self::FLD_TTL => $ttl,
            self::FLD_TIME => $now
        ];
    }

    protected function unwrap( $wrapped, $now ) {
        // Check if the value is a tombstone
        $purge = $this->parsePurgeValue( $wrapped );
        if ( $purge !== false ) {
            // Purged values should always have a negative current $ttl
            $curTTL = min( $purge[self::FLD_TIME] - $now, self::TINY_NEGATIVE );
            return [ false, $curTTL, null, $purge[self::FLD_TIME] ];
        }

        if ( !is_array( $wrapped ) // not found
            || !isset( $wrapped[self::FLD_VERSION] ) // wrong format
            || $wrapped[self::FLD_VERSION] !== self::VERSION // wrong version
        ) {
            return [ false, null, null, null ];
        }

        if ( $wrapped[self::FLD_TTL] > 0 ) {
            // Get the approximate time left on the key
            $age = $now - $wrapped[self::FLD_TIME];
            $curTTL = max( $wrapped[self::FLD_TTL] - $age, 0.0 );
        } else {
            // Key had no TTL, so the time left is unbounded
            $curTTL = INF;
        }

        if ( $wrapped[self::FLD_TIME] < $this->epoch ) {
            // Values this old are ignored
            return [ false, null, null, null ];
        }

        return [ $wrapped[self::FLD_VALUE], $curTTL, $wrapped[self::FLD_TIME], null ];
    }

    protected static function prefixCacheKeys( array $keys, $prefix ) {
        $res = [];
        foreach ( $keys as $key ) {
            $res[] = $prefix . $key;
        }

        return $res;
    }

    protected function determineKeyClassForStats( $key ) {
        $parts = explode( ':', $key );

        return $parts[1] ?? $parts[0]; // sanity
    }

    protected function parsePurgeValue( $value ) {
        if ( !is_string( $value ) ) {
            return false;
        }

        $segments = explode( ':', $value, 3 );
        if ( !isset( $segments[0] ) || !isset( $segments[1] )
            || "{$segments[0]}:" !== self::PURGE_VAL_PREFIX
        ) {
            return false;
        }

        if ( !isset( $segments[2] ) ) {
            // Back-compat with old purge values without holdoff
            $segments[2] = self::HOLDOFF_TTL;
        }

        if ( $segments[1] < $this->epoch ) {
            // Values this old are ignored
            return false;
        }

        return [
            self::FLD_TIME => (float)$segments[1],
            self::FLD_HOLDOFF => (int)$segments[2],
        ];
    }

    protected function makePurgeValue( $timestamp, $holdoff ) {
        return self::PURGE_VAL_PREFIX . (float)$timestamp . ':' . (int)$holdoff;
    }

    protected function getProcessCache( $group ) {
        if ( !isset( $this->processCaches[$group] ) ) {
            list( , $n ) = explode( ':', $group );
            $this->processCaches[$group] = new MapCacheLRU( (int)$n );
        }

        return $this->processCaches[$group];
    }

    private function getNonProcessCachedKeys( array $keys, array $opts, $pcTTL ) {
        $keysFound = [];
        if ( isset( $opts['pcTTL'] ) && $opts['pcTTL'] > 0 && $this->callbackDepth == 0 ) {
            $pcGroup = $opts['pcGroup'] ?? self::PC_PRIMARY;
            $procCache = $this->getProcessCache( $pcGroup );
            foreach ( $keys as $key ) {
                if ( $procCache->has( $key, $pcTTL ) ) {
                    $keysFound[] = $key;
                }
            }
        }

        return array_diff( $keys, $keysFound );
    }

    private function getRawKeysForWarmup( array $keys, array $checkKeys ) {
        if ( !$keys ) {
            return [];
        }

        $keysWarmUp = [];
        // Get all the value keys to fetch...
        foreach ( $keys as $key ) {
            $keysWarmUp[] = self::VALUE_KEY_PREFIX . $key;
        }
        // Get all the check keys to fetch...
        foreach ( $checkKeys as $i => $checkKeyOrKeys ) {
            if ( is_int( $i ) ) {
                // Single check key that applies to all value keys
                $keysWarmUp[] = self::TIME_KEY_PREFIX . $checkKeyOrKeys;
            } else {
                // List of check keys that apply to value key $i
                $keysWarmUp = array_merge(
                    $keysWarmUp,
                    self::prefixCacheKeys( $checkKeyOrKeys, self::TIME_KEY_PREFIX )
                );
            }
        }

        $warmupCache = $this->cache->getMulti( $keysWarmUp );
        $warmupCache += array_fill_keys( $keysWarmUp, false );

        return $warmupCache;
    }

    protected function getCurrentTime() {
        if ( $this->wallClockOverride ) {
            return $this->wallClockOverride;
        }

        $clockTime = (float)time(); // call this first
        // microtime() uses an initial gettimeofday() call added to usage clocks.
        // This can severely drift from time() and the microtime() value of other threads
        // due to undercounting of the amount of time elapsed. Instead of seeing the current
        // time as being in the past, use the value of time(). This avoids setting cache values
        // that will immediately be seen as expired and possibly cause stampedes.
        return max( microtime( true ), $clockTime );
    }

    public function setMockTime( &$time ) {
        $this->wallClockOverride =& $time;
        $this->cache->setMockTime( $time );
    }
}