WANObjectCache.php

Go to the documentation of this file.

<?php
use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;

class WANObjectCache implements IExpiringStore, LoggerAwareInterface {
    protected $cache;
    protected $procCache;
    protected $purgeChannel;
    protected $purgeRelayer;
    protected $logger;

    protected $lastRelayError = self::ERR_NONE;

    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 LOCK_TTL = 10;
    const LOW_TTL = 30;
    const LOCK_TSE = 1;

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

    const TINY_NEGATIVE = -0.000001;

    const VERSION = 1;

    const FLD_VERSION = 0;
    const FLD_VALUE = 1;
    const FLD_TTL = 2;
    const FLD_TIME = 3;
    const FLD_FLAGS = 4;
    const FLD_HOLDOFF = 5;

    const FLG_STALE = 1;

    const ERR_NONE = 0; // no error
    const ERR_NO_RESPONSE = 1; // no response
    const ERR_UNREACHABLE = 2; // can't connect
    const ERR_UNEXPECTED = 3; // response gave some error
    const ERR_RELAY = 4; // relay broadcast failed

    const VALUE_KEY_PREFIX = 'WANCache:v:';
    const STASH_KEY_PREFIX = 'WANCache:s:';
    const TIME_KEY_PREFIX = 'WANCache:t:';

    const PURGE_VAL_PREFIX = 'PURGED:';

    const MAX_PC_KEYS = 1000; // max keys to keep in process cache

    const DEFAULT_PURGE_CHANNEL = 'wancache-purge';

    public function __construct( array $params ) {
        $this->cache = $params['cache'];
        $this->procCache = new HashBagOStuff( [ 'maxKeys' => self::MAX_PC_KEYS ] );
        $this->purgeChannel = isset( $params['channels']['purge'] )
            ? $params['channels']['purge']
            : self::DEFAULT_PURGE_CHANNEL;
        $this->purgeRelayer = isset( $params['relayers']['purge'] )
            ? $params['relayers']['purge']
            : new EventRelayerNull( [] );
        $this->setLogger( isset( $params['logger'] ) ? $params['logger'] : new NullLogger() );
    }

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

    public static function newEmpty() {
        return new self( [
            'cache'   => new EmptyBagOStuff(),
            'pool'    => 'empty',
            'relayer' => new EventRelayerNull( [] )
        ] );
    }

    final public function get( $key, &$curTTL = null, array $checkKeys = [] ) {
        $curTTLs = [];
        $values = $this->getMulti( [ $key ], $curTTLs, $checkKeys );
        $curTTL = isset( $curTTLs[$key] ) ? $curTTLs[$key] : null;

        return isset( $values[$key] ) ? $values[$key] : false;
    }

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

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

        $checkKeysForAll = [];
        $checkKeysByKey = [];
        $checkKeysFlat = [];
        foreach ( $checkKeys as $i => $keys ) {
            $prefixed = self::prefixCacheKeys( (array)$keys, self::TIME_KEY_PREFIX );
            $checkKeysFlat = array_merge( $checkKeysFlat, $prefixed );
            // Is this 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] = isset( $checkKeysByKey[$i] )
                    ? array_merge( $checkKeysByKey[$i], $prefixed )
                    : $prefixed;
            }
        }

        // Fetch all of the raw values
        $wrappedValues = $this->cache->getMulti( array_merge( $valueKeys, $checkKeysFlat ) );
        // Time used to compare/init "check" keys (derived after getMulti() to be pessimistic)
        $now = microtime( true );

        // 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 ) {
            if ( !isset( $wrappedValues[$vKey] ) ) {
                continue; // not found
            }

            $key = substr( $vKey, $vPrefixLen ); // unprefix

            list( $value, $curTTL ) = $this->unwrap( $wrappedValues[$vKey], $now );
            if ( $value !== false ) {
                $result[$key] = $value;

                // Force dependant keys to be invalid 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] );
                }
                foreach ( $purgeValues as $purge ) {
                    $safeTimestamp = $purge[self::FLD_TIME] + $purge[self::FLD_HOLDOFF];
                    if ( $safeTimestamp >= $wrappedValues[$vKey][self::FLD_TIME] ) {
                        // How long ago this value was expired by *this* check key
                        $ago = min( $purge[self::FLD_TIME] - $now, self::TINY_NEGATIVE );
                        // How long ago this value was expired by *any* known check key
                        $curTTL = min( $curTTL, $ago );
                    }
                }
            }
            $curTTLs[$key] = $curTTL;
        }

        return $result;
    }

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

    final public function set( $key, $value, $ttl = 0, array $opts = [] ) {
        $lockTSE = isset( $opts['lockTSE'] ) ? $opts['lockTSE'] : self::TSE_NONE;
        $age = isset( $opts['since'] ) ? max( 0, microtime( true ) - $opts['since'] ) : 0;
        $lag = isset( $opts['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 $key due to pending writes." );

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

        $wrapExtra = []; // additional wrapped value fields
        // 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: read lag with "lockTSE"; save but record value as stale
            if ( $lockTSE >= 0 ) {
                $ttl = max( 1, (int)$lockTSE ); // set() expects seconds
                $wrapExtra[self::FLD_FLAGS] = self::FLG_STALE; // mark as stale
            // Case B: any long-running transaction; ignore this set()
            } elseif ( $age > self::MAX_READ_LAG ) {
                $this->logger->warning( "Rejected set() for $key due to snapshot lag." );

                return true; // no-op the write for being unsafe
            // Case C: high replication lag; lower TTL instead of ignoring all set()s
            } elseif ( $lag === false || $lag > self::MAX_READ_LAG ) {
                $ttl = $ttl ? min( $ttl, self::TTL_LAGGED ) : self::TTL_LAGGED;
                $this->logger->warning( "Lowered set() TTL for $key due to replication lag." );
            // Case D: medium length request with medium replication lag; ignore this set()
            } else {
                $this->logger->warning( "Rejected set() for $key due to high read lag." );

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

        // Wrap that value with time/TTL/version metadata
        $wrapped = $this->wrap( $value, $ttl ) + $wrapExtra;

        $func = function ( $cache, $key, $cWrapped ) use ( $wrapped ) {
            return ( is_string( $cWrapped ) )
                ? false // key is tombstoned; do nothing
                : $wrapped;
        };

        return $this->cache->merge( self::VALUE_KEY_PREFIX . $key, $func, $ttl, 1 );
    }

    final public function delete( $key, $ttl = self::HOLDOFF_TTL ) {
        $key = self::VALUE_KEY_PREFIX . $key;

        if ( $ttl <= 0 ) {
            // Update the local datacenter immediately
            $ok = $this->cache->delete( $key );
            // Publish the purge to all datacenters
            $ok = $this->relayDelete( $key ) && $ok;
        } else {
            // Update the local datacenter immediately
            $ok = $this->cache->set( $key,
                $this->makePurgeValue( microtime( true ), self::HOLDOFF_NONE ),
                $ttl
            );
            // Publish the purge to all datacenters
            $ok = $this->relayPurge( $key, $ttl, self::HOLDOFF_NONE ) && $ok;
        }

        return $ok;
    }

    final public function getCheckKeyTime( $key ) {
        $key = self::TIME_KEY_PREFIX . $key;

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

        return $time;
    }

    final public function touchCheckKey( $key, $holdoff = self::HOLDOFF_TTL ) {
        $key = self::TIME_KEY_PREFIX . $key;
        // Update the local datacenter immediately
        $ok = $this->cache->set( $key,
            $this->makePurgeValue( microtime( true ), $holdoff ),
            self::CHECK_KEY_TTL
        );
        // Publish the purge to all datacenters
        return $this->relayPurge( $key, self::CHECK_KEY_TTL, $holdoff ) && $ok;
    }

    final public function resetCheckKey( $key ) {
        $key = self::TIME_KEY_PREFIX . $key;
        // Update the local datacenter immediately
        $ok = $this->cache->delete( $key );
        // Publish the purge to all datacenters
        return $this->relayDelete( $key ) && $ok;
    }

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

        // Try the process cache if enabled
        $value = ( $pcTTL >= 0 ) ? $this->procCache->get( $key ) : false;

        if ( $value === false ) {
            // Fetch the value over the network
            $value = $this->doGetWithSetCallback( $key, $ttl, $callback, $opts );
            // Update the process cache if enabled
            if ( $pcTTL >= 0 && $value !== false ) {
                $this->procCache->set( $key, $value, $pcTTL );
            }
        }

        return $value;
    }

    protected function doGetWithSetCallback( $key, $ttl, $callback, array $opts ) {
        $lowTTL = isset( $opts['lowTTL'] ) ? $opts['lowTTL'] : min( self::LOW_TTL, $ttl );
        $lockTSE = isset( $opts['lockTSE'] ) ? $opts['lockTSE'] : self::TSE_NONE;
        $checkKeys = isset( $opts['checkKeys'] ) ? $opts['checkKeys'] : [];

        // Get the current key value
        $curTTL = null;
        $cValue = $this->get( $key, $curTTL, $checkKeys ); // current value
        $value = $cValue; // return value

        // Determine if a regeneration is desired
        if ( $value !== false && $curTTL > 0 && !$this->worthRefresh( $curTTL, $lowTTL ) ) {
            return $value;
        }

        // A deleted key with a negative TTL left must be tombstoned
        $isTombstone = ( $curTTL !== null && $value === false );
        // Assume a key is hot if requested soon after invalidation
        $isHot = ( $curTTL !== null && $curTTL <= 0 && abs( $curTTL ) <= $lockTSE );
        // Decide whether a single thread should handle regenerations.
        // This avoids stampedes when $checkKeys are bumped and when preemptive
        // renegerations take too long. It also reduces regenerations while $key
        // is tombstoned. This balances cache freshness with avoiding DB load.
        $useMutex = ( $isHot || ( $isTombstone && $lockTSE > 0 ) );

        $lockAcquired = false;
        if ( $useMutex ) {
            // Acquire a datacenter-local non-blocking lock
            if ( $this->cache->lock( $key, 0, self::LOCK_TTL ) ) {
                // Lock acquired; this thread should update the key
                $lockAcquired = true;
            } elseif ( $value !== false ) {
                // If it cannot be acquired; then the stale value can be used
                return $value;
            } else {
                // Use the stash value for tombstoned keys to reduce regeneration load.
                // For hot keys, either another thread has the lock or the lock failed;
                // use the stash value from the last thread that regenerated it.
                $value = $this->cache->get( self::STASH_KEY_PREFIX . $key );
                if ( $value !== false ) {
                    return $value;
                }
            }
        }

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

        // Generate the new value from the callback...
        $setOpts = [];
        $value = call_user_func_array( $callback, [ $cValue, &$ttl, &$setOpts ] );
        // When delete() is called, writes are write-holed by the tombstone,
        // so use a special stash key to pass the new value around threads.
        if ( $useMutex && $value !== false && $ttl >= 0 ) {
            $tempTTL = max( 1, (int)$lockTSE ); // set() expects seconds
            $this->cache->set( self::STASH_KEY_PREFIX . $key, $value, $tempTTL );
        }

        if ( $lockAcquired ) {
            $this->cache->unlock( $key );
        }

        if ( $value !== false && $ttl >= 0 ) {
            // Update the cache; this will fail if the key is tombstoned
            $setOpts['lockTSE'] = $lockTSE;
            $this->set( $key, $value, $ttl, $setOpts );
        }

        return $value;
    }

    public function makeKey() {
        return call_user_func_array( [ $this->cache, __FUNCTION__ ], func_get_args() );
    }

    public function makeGlobalKey() {
        return call_user_func_array( [ $this->cache, __FUNCTION__ ], func_get_args() );
    }

    final public function getLastError() {
        if ( $this->lastRelayError ) {
            // If the cache and the relayer failed, focus on the later.
            // An update not making it to the relayer means it won't show up
            // in other DCs (nor will consistent re-hashing see up-to-date values).
            // On the other hand, if just the cache update failed, then it should
            // eventually be applied by the relayer.
            return $this->lastRelayError;
        }

        $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();
        $this->lastRelayError = self::ERR_NONE;
    }

    public function clearProcessCache() {
        $this->procCache->clear();
    }

    protected function relayPurge( $key, $ttl, $holdoff ) {
        $event = $this->cache->modifySimpleRelayEvent( [
            'cmd' => 'set',
            'key' => $key,
            'val' => 'PURGED:$UNIXTIME$:' . (int)$holdoff,
            'ttl' => max( $ttl, 1 ),
            'sbt' => true, // substitute $UNIXTIME$ with actual microtime
        ] );

        $ok = $this->purgeRelayer->notify( $this->purgeChannel, $event );
        if ( !$ok ) {
            $this->lastRelayError = self::ERR_RELAY;
        }

        return $ok;
    }

    protected function relayDelete( $key ) {
        $event = $this->cache->modifySimpleRelayEvent( [
            'cmd' => 'delete',
            'key' => $key,
        ] );

        $ok = $this->purgeRelayer->notify( $this->purgeChannel, $event );
        if ( !$ok ) {
            $this->lastRelayError = self::ERR_RELAY;
        }

        return $ok;
    }

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

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

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

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

    protected function unwrap( $wrapped, $now ) {
        // Check if the value is a tombstone
        $purge = self::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 ];
        }

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

        $flags = isset( $wrapped[self::FLD_FLAGS] ) ? $wrapped[self::FLD_FLAGS] : 0;
        if ( ( $flags & self::FLG_STALE ) == self::FLG_STALE ) {
            // Treat as expired, with the cache time as the expiration
            $age = $now - $wrapped[self::FLD_TIME];
            $curTTL = min( -$age, self::TINY_NEGATIVE );
        } elseif ( $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;
        }

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

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

        return $res;
    }

    protected static 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;
        }
        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;
    }
}