RedisConnectionPool.php

Go to the documentation of this file.

<?php
use MediaWiki\Logger\LoggerFactory;
use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerInterface;

class RedisConnectionPool implements LoggerAwareInterface {
    protected $connectTimeout;
    protected $readTimeout;
    protected $password;
    protected $persistent;
    protected $serializer;
    protected $idlePoolSize = 0;

    protected $connections = [];
    protected $downServers = [];

    protected static $instances = [];

    const SERVER_DOWN_TTL = 30;

    protected $logger;

    protected function __construct( array $options ) {
        if ( !class_exists( 'Redis' ) ) {
            throw new Exception( __CLASS__ . ' requires a Redis client library. ' .
                'See https://www.mediawiki.org/wiki/Redis#Setup' );
        }
        if ( isset( $options['logger'] ) ) {
            $this->setLogger( $options['logger'] );
        } else {
            $this->setLogger( LoggerFactory::getInstance( 'redis' ) );
        }
        $this->connectTimeout = $options['connectTimeout'];
        $this->readTimeout = $options['readTimeout'];
        $this->persistent = $options['persistent'];
        $this->password = $options['password'];
        if ( !isset( $options['serializer'] ) || $options['serializer'] === 'php' ) {
            $this->serializer = Redis::SERIALIZER_PHP;
        } elseif ( $options['serializer'] === 'igbinary' ) {
            $this->serializer = Redis::SERIALIZER_IGBINARY;
        } elseif ( $options['serializer'] === 'none' ) {
            $this->serializer = Redis::SERIALIZER_NONE;
        } else {
            throw new InvalidArgumentException( "Invalid serializer specified." );
        }
    }

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

    protected static function applyDefaultConfig( array $options ) {
        if ( !isset( $options['connectTimeout'] ) ) {
            $options['connectTimeout'] = 1;
        }
        if ( !isset( $options['readTimeout'] ) ) {
            $options['readTimeout'] = 1;
        }
        if ( !isset( $options['persistent'] ) ) {
            $options['persistent'] = false;
        }
        if ( !isset( $options['password'] ) ) {
            $options['password'] = null;
        }

        return $options;
    }

    public static function singleton( array $options ) {
        $options = self::applyDefaultConfig( $options );
        // Map the options to a unique hash...
        ksort( $options ); // normalize to avoid pool fragmentation
        $id = sha1( serialize( $options ) );
        // Initialize the object at the hash as needed...
        if ( !isset( self::$instances[$id] ) ) {
            self::$instances[$id] = new self( $options );
            LoggerFactory::getInstance( 'redis' )->debug(
                "Creating a new " . __CLASS__ . " instance with id $id."
            );
        }

        return self::$instances[$id];
    }

    public static function destroySingletons() {
        self::$instances = [];
    }

    public function getConnection( $server ) {
        // Check the listing "dead" servers which have had a connection errors.
        // Servers are marked dead for a limited period of time, to
        // avoid excessive overhead from repeated connection timeouts.
        if ( isset( $this->downServers[$server] ) ) {
            $now = time();
            if ( $now > $this->downServers[$server] ) {
                // Dead time expired
                unset( $this->downServers[$server] );
            } else {
                // Server is dead
                $this->logger->debug(
                    'Server "{redis_server}" is marked down for another ' .
                    ( $this->downServers[$server] - $now ) . 'seconds',
                    [ 'redis_server' => $server ]
                );

                return false;
            }
        }

        // Check if a connection is already free for use
        if ( isset( $this->connections[$server] ) ) {
            foreach ( $this->connections[$server] as &$connection ) {
                if ( $connection['free'] ) {
                    $connection['free'] = false;
                    --$this->idlePoolSize;

                    return new RedisConnRef(
                        $this, $server, $connection['conn'], $this->logger
                    );
                }
            }
        }

        if ( substr( $server, 0, 1 ) === '/' ) {
            // UNIX domain socket
            // These are required by the redis extension to start with a slash, but
            // we still need to set the port to a special value to make it work.
            $host = $server;
            $port = 0;
        } else {
            // TCP connection
            $hostPort = IP::splitHostAndPort( $server );
            if ( !$server || !$hostPort ) {
                throw new InvalidArgumentException(
                    __CLASS__ . ": invalid configured server \"$server\""
                );
            }
            list( $host, $port ) = $hostPort;
            if ( $port === false ) {
                $port = 6379;
            }
        }

        $conn = new Redis();
        try {
            if ( $this->persistent ) {
                $result = $conn->pconnect( $host, $port, $this->connectTimeout );
            } else {
                $result = $conn->connect( $host, $port, $this->connectTimeout );
            }
            if ( !$result ) {
                $this->logger->error(
                    'Could not connect to server "{redis_server}"',
                    [ 'redis_server' => $server ]
                );
                // Mark server down for some time to avoid further timeouts
                $this->downServers[$server] = time() + self::SERVER_DOWN_TTL;

                return false;
            }
            if ( $this->password !== null ) {
                if ( !$conn->auth( $this->password ) ) {
                    $this->logger->error(
                        'Authentication error connecting to "{redis_server}"',
                        [ 'redis_server' => $server ]
                    );
                }
            }
        } catch ( RedisException $e ) {
            $this->downServers[$server] = time() + self::SERVER_DOWN_TTL;
            $this->logger->error(
                'Redis exception connecting to "{redis_server}"',
                [
                    'redis_server' => $server,
                    'exception' => $e,
                ]
            );

            return false;
        }

        if ( $conn ) {
            $conn->setOption( Redis::OPT_READ_TIMEOUT, $this->readTimeout );
            $conn->setOption( Redis::OPT_SERIALIZER, $this->serializer );
            $this->connections[$server][] = [ 'conn' => $conn, 'free' => false ];

            return new RedisConnRef( $this, $server, $conn, $this->logger );
        } else {
            return false;
        }
    }

    public function freeConnection( $server, Redis $conn ) {
        $found = false;

        foreach ( $this->connections[$server] as &$connection ) {
            if ( $connection['conn'] === $conn && !$connection['free'] ) {
                $connection['free'] = true;
                ++$this->idlePoolSize;
                break;
            }
        }

        $this->closeExcessIdleConections();

        return $found;
    }

    protected function closeExcessIdleConections() {
        if ( $this->idlePoolSize <= count( $this->connections ) ) {
            return; // nothing to do (no more connections than servers)
        }

        foreach ( $this->connections as &$serverConnections ) {
            foreach ( $serverConnections as $key => &$connection ) {
                if ( $connection['free'] ) {
                    unset( $serverConnections[$key] );
                    if ( --$this->idlePoolSize <= count( $this->connections ) ) {
                        return; // done (no more connections than servers)
                    }
                }
            }
        }
    }

    public function handleException( $server, RedisConnRef $cref, RedisException $e ) {
        $this->handleError( $cref, $e );
    }

    public function handleError( RedisConnRef $cref, RedisException $e ) {
        $server = $cref->getServer();
        $this->logger->error(
            'Redis exception on server "{redis_server}"',
            [
                'redis_server' => $server,
                'exception' => $e,
            ]
        );
        foreach ( $this->connections[$server] as $key => $connection ) {
            if ( $cref->isConnIdentical( $connection['conn'] ) ) {
                $this->idlePoolSize -= $connection['free'] ? 1 : 0;
                unset( $this->connections[$server][$key] );
                break;
            }
        }
    }

    public function reauthenticateConnection( $server, Redis $conn ) {
        if ( $this->password !== null ) {
            if ( !$conn->auth( $this->password ) ) {
                $this->logger->error(
                    'Authentication error connecting to "{redis_server}"',
                    [ 'redis_server' => $server ]
                );

                return false;
            }
        }

        return true;
    }

    public function resetTimeout( Redis $conn, $timeout = null ) {
        $conn->setOption( Redis::OPT_READ_TIMEOUT, $timeout ?: $this->readTimeout );
    }

    function __destruct() {
        foreach ( $this->connections as $server => &$serverConnections ) {
            foreach ( $serverConnections as $key => &$connection ) {
                $connection['conn']->close();
            }
        }
    }
}

class RedisConnRef {
    protected $pool;
    protected $conn;

    protected $server; // string
    protected $lastError; // string

    protected $logger;

    public function __construct(
        RedisConnectionPool $pool, $server, Redis $conn, LoggerInterface $logger
    ) {
        $this->pool = $pool;
        $this->server = $server;
        $this->conn = $conn;
        $this->logger = $logger;
    }

    public function getServer() {
        return $this->server;
    }

    public function getLastError() {
        return $this->lastError;
    }

    public function clearLastError() {
        $this->lastError = null;
    }

    public function __call( $name, $arguments ) {
        $conn = $this->conn; // convenience

        // Work around https://github.com/nicolasff/phpredis/issues/70
        $lname = strtolower( $name );
        if ( ( $lname === 'blpop' || $lname == 'brpop' )
            && is_array( $arguments[0] ) && isset( $arguments[1] )
        ) {
            $this->pool->resetTimeout( $conn, $arguments[1] + 1 );
        } elseif ( $lname === 'brpoplpush' && isset( $arguments[2] ) ) {
            $this->pool->resetTimeout( $conn, $arguments[2] + 1 );
        }

        $conn->clearLastError();
        try {
            $res = call_user_func_array( [ $conn, $name ], $arguments );
            if ( preg_match( '/^ERR operation not permitted\b/', $conn->getLastError() ) ) {
                $this->pool->reauthenticateConnection( $this->server, $conn );
                $conn->clearLastError();
                $res = call_user_func_array( [ $conn, $name ], $arguments );
                $this->logger->info(
                    "Used automatic re-authentication for method '$name'.",
                    [ 'redis_server' => $this->server ]
                );
            }
        } catch ( RedisException $e ) {
            $this->pool->resetTimeout( $conn ); // restore
            throw $e;
        }

        $this->lastError = $conn->getLastError() ?: $this->lastError;

        $this->pool->resetTimeout( $conn ); // restore

        return $res;
    }

    public function luaEval( $script, array $params, $numKeys ) {
        $sha1 = sha1( $script ); // 40 char hex
        $conn = $this->conn; // convenience
        $server = $this->server; // convenience

        // Try to run the server-side cached copy of the script
        $conn->clearLastError();
        $res = $conn->evalSha( $sha1, $params, $numKeys );
        // If we got a permission error reply that means that (a) we are not in
        // multi()/pipeline() and (b) some connection problem likely occurred. If
        // the password the client gave was just wrong, an exception should have
        // been thrown back in getConnection() previously.
        if ( preg_match( '/^ERR operation not permitted\b/', $conn->getLastError() ) ) {
            $this->pool->reauthenticateConnection( $server, $conn );
            $conn->clearLastError();
            $res = $conn->eval( $script, $params, $numKeys );
            $this->logger->info(
                "Used automatic re-authentication for Lua script '$sha1'.",
                [ 'redis_server' => $server ]
            );
        }
        // If the script is not in cache, use eval() to retry and cache it
        if ( preg_match( '/^NOSCRIPT/', $conn->getLastError() ) ) {
            $conn->clearLastError();
            $res = $conn->eval( $script, $params, $numKeys );
            $this->logger->info(
                "Used eval() for Lua script '$sha1'.",
                [ 'redis_server' => $server ]
            );
        }

        if ( $conn->getLastError() ) { // script bug?
            $this->logger->error(
                'Lua script error on server "{redis_server}": {lua_error}',
                [
                    'redis_server' => $server,
                    'lua_error' => $conn->getLastError()
                ]
            );
        }

        $this->lastError = $conn->getLastError() ?: $this->lastError;

        return $res;
    }

    public function isConnIdentical( Redis $conn ) {
        return $this->conn === $conn;
    }

    function __destruct() {
        $this->pool->freeConnection( $this->server, $this->conn );
    }
}