RedisBagOStuff.php

Go to the documentation of this file.

<?php
class RedisBagOStuff extends BagOStuff {
    protected $redisPool;
    protected $servers;
    protected $serverTagMap;
    protected $automaticFailover;
 
    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;
        }
 
        if ( isset( $params['automaticFailover'] ) ) {
            $this->automaticFailover = $params['automaticFailover'];
        } else {
            $this->automaticFailover = true;
        }
 
        $this->attrMap[self::ATTR_SYNCWRITES] = self::QOS_SYNCWRITES_NONE;
    }
 
    protected function doGet( $key, $flags = 0 ) {
        $casToken = null;
 
        return $this->getWithToken( $key, $casToken, $flags );
    }
 
    protected function getWithToken( $key, &$casToken, $flags = 0 ) {
        list( $server, $conn ) = $this->getConnection( $key );
        if ( !$conn ) {
            return false;
        }
        try {
            $value = $conn->get( $key );
            $casToken = $value;
            $result = $this->unserialize( $value );
        } catch ( RedisException $e ) {
            $result = false;
            $this->handleException( $conn, $e );
        }
 
        $this->logRequest( 'get', $key, $server, $result );
        return $result;
    }
 
    public function set( $key, $value, $expiry = 0, $flags = 0 ) {
        list( $server, $conn ) = $this->getConnection( $key );
        if ( !$conn ) {
            return false;
        }
        $expiry = $this->convertToRelative( $expiry );
        try {
            if ( $expiry ) {
                $result = $conn->setex( $key, $expiry, $this->serialize( $value ) );
            } else {
                // No expiry, that is very different from zero expiry in Redis
                $result = $conn->set( $key, $this->serialize( $value ) );
            }
        } catch ( RedisException $e ) {
            $result = false;
            $this->handleException( $conn, $e );
        }
 
        $this->logRequest( 'set', $key, $server, $result );
        return $result;
    }
 
    public function delete( $key ) {
        list( $server, $conn ) = $this->getConnection( $key );
        if ( !$conn ) {
            return false;
        }
        try {
            $conn->delete( $key );
            // Return true even if the key didn't exist
            $result = true;
        } catch ( RedisException $e ) {
            $result = false;
            $this->handleException( $conn, $e );
        }
 
        $this->logRequest( 'delete', $key, $server, $result );
        return $result;
    }
 
    public function getMulti( array $keys, $flags = 0 ) {
        $batches = [];
        $conns = [];
        foreach ( $keys as $key ) {
            list( $server, $conn ) = $this->getConnection( $key );
            if ( !$conn ) {
                continue;
            }
            $conns[$server] = $conn;
            $batches[$server][] = $key;
        }
        $result = [];
        foreach ( $batches as $server => $batchKeys ) {
            $conn = $conns[$server];
            try {
                $conn->multi( Redis::PIPELINE );
                foreach ( $batchKeys as $key ) {
                    $conn->get( $key );
                }
                $batchResult = $conn->exec();
                if ( $batchResult === false ) {
                    $this->debug( "multi request to $server failed" );
                    continue;
                }
                foreach ( $batchResult as $i => $value ) {
                    if ( $value !== false ) {
                        $result[$batchKeys[$i]] = $this->unserialize( $value );
                    }
                }
            } catch ( RedisException $e ) {
                $this->handleException( $conn, $e );
            }
        }
 
        $this->debug( "getMulti for " . count( $keys ) . " keys " .
            "returned " . count( $result ) . " results" );
        return $result;
    }
 
    public function setMulti( array $data, $expiry = 0 ) {
        $batches = [];
        $conns = [];
        foreach ( $data as $key => $value ) {
            list( $server, $conn ) = $this->getConnection( $key );
            if ( !$conn ) {
                continue;
            }
            $conns[$server] = $conn;
            $batches[$server][] = $key;
        }
 
        $expiry = $this->convertToRelative( $expiry );
        $result = true;
        foreach ( $batches as $server => $batchKeys ) {
            $conn = $conns[$server];
            try {
                $conn->multi( Redis::PIPELINE );
                foreach ( $batchKeys as $key ) {
                    if ( $expiry ) {
                        $conn->setex( $key, $expiry, $this->serialize( $data[$key] ) );
                    } else {
                        $conn->set( $key, $this->serialize( $data[$key] ) );
                    }
                }
                $batchResult = $conn->exec();
                if ( $batchResult === false ) {
                    $this->debug( "setMulti request to $server failed" );
                    continue;
                }
                foreach ( $batchResult as $value ) {
                    if ( $value === false ) {
                        $result = false;
                    }
                }
            } catch ( RedisException $e ) {
                $this->handleException( $server, $conn, $e );
                $result = false;
            }
        }
 
        return $result;
    }
 
    public function add( $key, $value, $expiry = 0 ) {
        list( $server, $conn ) = $this->getConnection( $key );
        if ( !$conn ) {
            return false;
        }
        $expiry = $this->convertToRelative( $expiry );
        try {
            if ( $expiry ) {
                $result = $conn->set(
                    $key,
                    $this->serialize( $value ),
                    [ 'nx', 'ex' => $expiry ]
                );
            } else {
                $result = $conn->setnx( $key, $this->serialize( $value ) );
            }
        } catch ( RedisException $e ) {
            $result = false;
            $this->handleException( $conn, $e );
        }
 
        $this->logRequest( 'add', $key, $server, $result );
        return $result;
    }
 
    public function merge( $key, callable $callback, $exptime = 0, $attempts = 10, $flags = 0 ) {
        return $this->mergeViaCas( $key, $callback, $exptime, $attempts );
    }
 
    public function incr( $key, $value = 1 ) {
        list( $server, $conn ) = $this->getConnection( $key );
        if ( !$conn ) {
            return false;
        }
        try {
            if ( !$conn->exists( $key ) ) {
                return false;
            }
            // @FIXME: on races, the key may have a 0 TTL
            $result = $conn->incrBy( $key, $value );
        } catch ( RedisException $e ) {
            $result = false;
            $this->handleException( $conn, $e );
        }
 
        $this->logRequest( 'incr', $key, $server, $result );
        return $result;
    }
 
    public function changeTTL( $key, $expiry = 0 ) {
        list( $server, $conn ) = $this->getConnection( $key );
        if ( !$conn ) {
            return false;
        }
 
        $expiry = $this->convertToRelative( $expiry );
        try {
            $result = $conn->expire( $key, $expiry );
        } catch ( RedisException $e ) {
            $result = false;
            $this->handleException( $conn, $e );
        }
 
        $this->logRequest( 'expire', $key, $server, $result );
        return $result;
    }
 
    public function modifySimpleRelayEvent( array $event ) {
        if ( array_key_exists( 'val', $event ) ) {
            $event['val'] = serialize( $event['val'] ); // this class uses PHP serialization
        }
 
        return $event;
    }
 
    protected function serialize( $data ) {
        // Serialize anything but integers so INCR/DECR work
        // Do not store integer-like strings as integers to avoid type confusion (T62563)
        return is_int( $data ) ? $data : serialize( $data );
    }
 
    protected function unserialize( $data ) {
        $int = intval( $data );
        return $data === (string)$int ? $int : unserialize( $data );
    }
 
    protected function getConnection( $key ) {
        $candidates = array_keys( $this->serverTagMap );
 
        if ( count( $this->servers ) > 1 ) {
            ArrayUtils::consistentHashSort( $candidates, $key, '/' );
            if ( !$this->automaticFailover ) {
                $candidates = array_slice( $candidates, 0, 1 );
            }
        }
 
        while ( ( $tag = array_shift( $candidates ) ) !== null ) {
            $server = $this->serverTagMap[$tag];
            $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 && $candidates ) {
                try {
                    if ( $this->getMasterLinkStatus( $conn ) === '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->handleException( $conn, $e );
                    continue;
                }
            }
 
            return [ $server, $conn ];
        }
 
        $this->setLastError( BagOStuff::ERR_UNREACHABLE );
 
        return [ false, false ];
    }
 
    protected function getMasterLinkStatus( RedisConnRef $conn ) {
        $info = $conn->info();
        return $info['master_link_status'] ?? null;
    }
 
    protected function logError( $msg ) {
        $this->logger->error( "Redis error: $msg" );
    }
 
    protected function handleException( RedisConnRef $conn, $e ) {
        $this->setLastError( BagOStuff::ERR_UNEXPECTED );
        $this->redisPool->handleError( $conn, $e );
    }
 
    public function logRequest( $method, $key, $server, $result ) {
        $this->debug( "$method $key on $server: " .
            ( $result === false ? "failure" : "success" ) );
    }
}