RedisBagOStuff.php

Go to the documentation of this file.

<?php
class RedisBagOStuff extends MediumSpecificBagOStuff {
    protected $redisPool;
    protected $servers;
    protected $serverTagMap;
    protected $automaticFailover;
 
    public function __construct( $params ) {
        parent::__construct( $params );
        $redisConf = [ 'serializer' => 'none' ]; // manage that in this class
        foreach ( [ 'connectTimeout', 'persistent', 'password' ] as $opt ) {
            if ( isset( $params[$opt] ) ) {
                $redisConf[$opt] = $params[$opt];
            }
        }
        $this->redisPool = RedisConnectionPool::singleton( $redisConf );
 
        $this->servers = $params['servers'];
        foreach ( $this->servers as $key => $server ) {
            $this->serverTagMap[is_int( $key ) ? $server : $key] = $server;
        }
 
        $this->automaticFailover = $params['automaticFailover'] ?? true;
 
        // ...and uses rdb snapshots (redis.conf default)
        $this->attrMap[self::ATTR_DURABILITY] = self::QOS_DURABILITY_DISK;
    }
 
    protected function doGet( $key, $flags = 0, &$casToken = null ) {
        $getToken = ( $casToken === self::PASS_BY_REF );
        $casToken = null;
 
        $conn = $this->getConnection( $key );
        if ( !$conn ) {
            return false;
        }
 
        $e = null;
        try {
            $blob = $conn->get( $key );
            if ( $blob !== false ) {
                $value = $this->unserialize( $blob );
                $valueSize = strlen( $blob );
            } else {
                $value = false;
                $valueSize = false;
            }
            if ( $getToken && $value !== false ) {
                $casToken = $blob;
            }
        } catch ( RedisException $e ) {
            $value = false;
            $valueSize = false;
            $this->handleException( $conn, $e );
        }
 
        $this->logRequest( 'get', $key, $conn->getServer(), $e );
 
        $this->updateOpStats( self::METRIC_OP_GET, [ $key => [ 0, $valueSize ] ] );
 
        return $value;
    }
 
    protected function doSet( $key, $value, $exptime = 0, $flags = 0 ) {
        $conn = $this->getConnection( $key );
        if ( !$conn ) {
            return false;
        }
 
        $ttl = $this->getExpirationAsTTL( $exptime );
        $serialized = $this->getSerialized( $value, $key );
        $valueSize = strlen( $serialized );
 
        $e = null;
        try {
            if ( $ttl ) {
                $result = $conn->setex( $key, $ttl, $serialized );
            } else {
                $result = $conn->set( $key, $serialized );
            }
        } catch ( RedisException $e ) {
            $result = false;
            $this->handleException( $conn, $e );
        }
 
        $this->logRequest( 'set', $key, $conn->getServer(), $e );
 
        $this->updateOpStats( self::METRIC_OP_SET, [ $key => [ $valueSize, 0 ] ] );
 
        return $result;
    }
 
    protected function doDelete( $key, $flags = 0 ) {
        $conn = $this->getConnection( $key );
        if ( !$conn ) {
            return false;
        }
 
        $e = null;
        try {
            // Note that redis does not return false if the key was not there
            $result = ( $conn->del( $key ) !== false );
        } catch ( RedisException $e ) {
            $result = false;
            $this->handleException( $conn, $e );
        }
 
        $this->logRequest( 'delete', $key, $conn->getServer(), $e );
 
        $this->updateOpStats( self::METRIC_OP_DELETE, [ $key ] );
 
        return $result;
    }
 
    protected function doGetMulti( array $keys, $flags = 0 ) {
        $blobsFound = [];
 
        [ $keysByServer, $connByServer ] = $this->getConnectionsForKeys( $keys );
        foreach ( $keysByServer as $server => $batchKeys ) {
            $conn = $connByServer[$server];
 
            $e = null;
            try {
                // Avoid mget() to reduce CPU hogging from a single request
                $conn->multi( Redis::PIPELINE );
                foreach ( $batchKeys as $key ) {
                    $conn->get( $key );
                }
                $batchResult = $conn->exec();
                if ( $batchResult === false ) {
                    $this->logRequest( 'get', implode( ',', $batchKeys ), $server, true );
                    continue;
                }
 
                foreach ( $batchResult as $i => $blob ) {
                    if ( $blob !== false ) {
                        $blobsFound[$batchKeys[$i]] = $blob;
                    }
                }
            } catch ( RedisException $e ) {
                $this->handleException( $conn, $e );
            }
 
            $this->logRequest( 'get', implode( ',', $batchKeys ), $server, $e );
        }
 
        // Preserve the order of $keys
        $result = [];
        $valueSizesByKey = [];
        foreach ( $keys as $key ) {
            if ( array_key_exists( $key, $blobsFound ) ) {
                $blob = $blobsFound[$key];
                $value = $this->unserialize( $blob );
                if ( $value !== false ) {
                    $result[$key] = $value;
                }
                $valueSize = strlen( $blob );
            } else {
                $valueSize = false;
            }
            $valueSizesByKey[$key] = [ 0, $valueSize ];
        }
 
        $this->updateOpStats( self::METRIC_OP_GET, $valueSizesByKey );
 
        return $result;
    }
 
    protected function doSetMulti( array $data, $exptime = 0, $flags = 0 ) {
        $ttl = $this->getExpirationAsTTL( $exptime );
        $op = $ttl ? 'setex' : 'set';
 
        $keys = array_keys( $data );
        $valueSizesByKey = [];
 
        [ $keysByServer, $connByServer, $result ] = $this->getConnectionsForKeys( $keys );
        foreach ( $keysByServer as $server => $batchKeys ) {
            $conn = $connByServer[$server];
 
            $e = null;
            try {
                // Avoid mset() to reduce CPU hogging from a single request
                $conn->multi( Redis::PIPELINE );
                foreach ( $batchKeys as $key ) {
                    $serialized = $this->getSerialized( $data[$key], $key );
                    if ( $ttl ) {
                        $conn->setex( $key, $ttl, $serialized );
                    } else {
                        $conn->set( $key, $serialized );
                    }
                    $valueSizesByKey[$key] = [ strlen( $serialized ), 0 ];
                }
                $batchResult = $conn->exec();
                if ( $batchResult === false ) {
                    $result = false;
                    $this->logRequest( $op, implode( ',', $batchKeys ), $server, true );
                    continue;
                }
 
                $result = $result && !in_array( false, $batchResult, true );
            } catch ( RedisException $e ) {
                $this->handleException( $conn, $e );
                $result = false;
            }
 
            $this->logRequest( $op, implode( ',', $batchKeys ), $server, $e );
        }
 
        $this->updateOpStats( self::METRIC_OP_SET, $valueSizesByKey );
 
        return $result;
    }
 
    protected function doDeleteMulti( array $keys, $flags = 0 ) {
        [ $keysByServer, $connByServer, $result ] = $this->getConnectionsForKeys( $keys );
        foreach ( $keysByServer as $server => $batchKeys ) {
            $conn = $connByServer[$server];
 
            $e = null;
            try {
                // Avoid delete() with array to reduce CPU hogging from a single request
                $conn->multi( Redis::PIPELINE );
                foreach ( $batchKeys as $key ) {
                    $conn->del( $key );
                }
                $batchResult = $conn->exec();
                if ( $batchResult === false ) {
                    $result = false;
                    $this->logRequest( 'delete', implode( ',', $batchKeys ), $server, true );
                    continue;
                }
                // Note that redis does not return false if the key was not there
                $result = $result && !in_array( false, $batchResult, true );
            } catch ( RedisException $e ) {
                $this->handleException( $conn, $e );
                $result = false;
            }
 
            $this->logRequest( 'delete', implode( ',', $batchKeys ), $server, $e );
        }
 
        $this->updateOpStats( self::METRIC_OP_DELETE, array_values( $keys ) );
 
        return $result;
    }
 
    public function doChangeTTLMulti( array $keys, $exptime, $flags = 0 ) {
        $relative = $this->isRelativeExpiration( $exptime );
        $op = ( $exptime == self::TTL_INDEFINITE )
            ? 'persist'
            : ( $relative ? 'expire' : 'expireAt' );
 
        [ $keysByServer, $connByServer, $result ] = $this->getConnectionsForKeys( $keys );
        foreach ( $keysByServer as $server => $batchKeys ) {
            $conn = $connByServer[$server];
 
            $e = null;
            try {
                $conn->multi( Redis::PIPELINE );
                foreach ( $batchKeys as $key ) {
                    if ( $exptime == self::TTL_INDEFINITE ) {
                        $conn->persist( $key );
                    } elseif ( $relative ) {
                        $conn->expire( $key, $this->getExpirationAsTTL( $exptime ) );
                    } else {
                        $conn->expireAt( $key, $this->getExpirationAsTimestamp( $exptime ) );
                    }
                }
                $batchResult = $conn->exec();
                if ( $batchResult === false ) {
                    $result = false;
                    $this->logRequest( $op, implode( ',', $batchKeys ), $server, true );
                    continue;
                }
                $result = in_array( false, $batchResult, true ) ? false : $result;
            } catch ( RedisException $e ) {
                $this->handleException( $conn, $e );
                $result = false;
            }
 
            $this->logRequest( $op, implode( ',', $batchKeys ), $server, $e );
        }
 
        $this->updateOpStats( self::METRIC_OP_CHANGE_TTL, array_values( $keys ) );
 
        return $result;
    }
 
    protected function doAdd( $key, $value, $exptime = 0, $flags = 0 ) {
        $conn = $this->getConnection( $key );
        if ( !$conn ) {
            return false;
        }
 
        $ttl = $this->getExpirationAsTTL( $exptime );
        $serialized = $this->getSerialized( $value, $key );
        $valueSize = strlen( $serialized );
 
        try {
            $result = $conn->set(
                $key,
                $serialized,
                $ttl ? [ 'nx', 'ex' => $ttl ] : [ 'nx' ]
            );
        } catch ( RedisException $e ) {
            $result = false;
            $this->handleException( $conn, $e );
        }
 
        $this->logRequest( 'add', $key, $conn->getServer(), $result );
 
        $this->updateOpStats( self::METRIC_OP_ADD, [ $key => [ $valueSize, 0 ] ] );
 
        return $result;
    }
 
    protected function doIncrWithInit( $key, $exptime, $step, $init, $flags ) {
        $conn = $this->getConnection( $key );
        if ( !$conn ) {
            return false;
        }
 
        $ttl = $this->getExpirationAsTTL( $exptime );
        try {
            static $script =
<<<LUA
            local key = KEYS[1]
            local ttl, step, init = unpack( ARGV )
            if redis.call( 'exists', key ) == 1 then
                return redis.call( 'incrBy', key, step )
            end
            if 1 * ttl ~= 0 then
                redis.call( 'setex', key, ttl, init )
            else
                redis.call( 'set', key, init )
            end
            return 1 * init
LUA;
            $result = $conn->luaEval( $script, [ $key, $ttl, $step, $init ], 1 );
        } catch ( RedisException $e ) {
            $result = false;
            $this->handleException( $conn, $e );
        }
        $this->logRequest( 'incrWithInit', $key, $conn->getServer(), $result );
 
        return $result;
    }
 
    protected function doChangeTTL( $key, $exptime, $flags ) {
        $conn = $this->getConnection( $key );
        if ( !$conn ) {
            return false;
        }
 
        $relative = $this->isRelativeExpiration( $exptime );
        try {
            if ( $exptime == self::TTL_INDEFINITE ) {
                $result = $conn->persist( $key );
                $this->logRequest( 'persist', $key, $conn->getServer(), $result );
            } elseif ( $relative ) {
                $result = $conn->expire( $key, $this->getExpirationAsTTL( $exptime ) );
                $this->logRequest( 'expire', $key, $conn->getServer(), $result );
            } else {
                $result = $conn->expireAt( $key, $this->getExpirationAsTimestamp( $exptime ) );
                $this->logRequest( 'expireAt', $key, $conn->getServer(), $result );
            }
        } catch ( RedisException $e ) {
            $result = false;
            $this->handleException( $conn, $e );
        }
 
        $this->updateOpStats( self::METRIC_OP_CHANGE_TTL, [ $key ] );
 
        return $result;
    }
 
    protected function getConnectionsForKeys( array $keys ) {
        $keysByServer = [];
        $connByServer = [];
        $success = true;
        foreach ( $keys as $key ) {
            $candidateTags = $this->getCandidateServerTagsForKey( $key );
 
            $conn = null;
            // Find a suitable server for this key...
            while ( ( $tag = array_shift( $candidateTags ) ) !== null ) {
                $server = $this->serverTagMap[$tag];
                // Reuse connection handles for keys mapping to the same server
                if ( isset( $connByServer[$server] ) ) {
                    $conn = $connByServer[$server];
                } else {
                    $conn = $this->redisPool->getConnection( $server, $this->logger );
                    if ( !$conn ) {
                        continue;
                    }
                    // If automatic failover is enabled, check that the server's link
                    // to its master (if any) is up -- but only if there are other
                    // viable candidates left to consider. Also, getMasterLinkStatus()
                    // does not work with twemproxy, though $candidates will be empty
                    // by now in such cases.
                    if ( $this->automaticFailover && $candidateTags ) {
                        try {
                            $info = $conn->info();
                            if ( ( $info['master_link_status'] ?? null ) === 'down' ) {
                                // If the master cannot be reached, fail-over to the next server.
                                // If masters are in data-center A, and replica DBs in data-center B,
                                // this helps avoid the case were fail-over happens in A but not
                                // to the corresponding server in B (e.g. read/write mismatch).
                                continue;
                            }
                        } catch ( RedisException $e ) {
                            // Server is not accepting commands
                            $this->redisPool->handleError( $conn, $e );
                            continue;
                        }
                    }
                    // Use this connection handle
                    $connByServer[$server] = $conn;
                }
                // Use this server for this key
                $keysByServer[$server][] = $key;
                break;
            }
 
            if ( !$conn ) {
                // No suitable server found for this key
                $success = false;
                $this->setLastError( BagOStuff::ERR_UNREACHABLE );
            }
        }
 
        return [ $keysByServer, $connByServer, $success ];
    }
 
    protected function getConnection( $key ) {
        [ , $connByServer ] = $this->getConnectionsForKeys( [ $key ] );
 
        return reset( $connByServer ) ?: null;
    }
 
    private function getCandidateServerTagsForKey( string $key ): array {
        $candidates = array_keys( $this->serverTagMap );
 
        if ( count( $this->servers ) > 1 ) {
            ArrayUtils::consistentHashSort( $candidates, $key, '/' );
            if ( !$this->automaticFailover ) {
                $candidates = array_slice( $candidates, 0, 1 );
            }
        }
 
        return $candidates;
    }
 
    protected function logError( $msg ) {
        $this->logger->error( "Redis error: $msg" );
    }
 
    protected function handleException( RedisConnRef $conn, RedisException $e ) {
        $this->setLastError( BagOStuff::ERR_UNEXPECTED );
        $this->redisPool->handleError( $conn, $e );
    }
 
    public function logRequest( $op, $keys, $server, $e = null ) {
        $this->debug( "$op($keys) on $server: " . ( $e ? "failure" : "success" ) );
    }
}