RedisConnectionPool.php

Go to the documentation of this file.

<?php
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 RuntimeException(
                __CLASS__ . ' requires a Redis client library. ' .
                'See https://www.mediawiki.org/wiki/Redis#Setup' );
        }
        $this->logger = isset( $options['logger'] )
            ? $options['logger']
            : new \Psr\Log\NullLogger();
        $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 );
        }

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

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

    public function getConnection( $server, LoggerInterface $logger = null ) {
        $logger = $logger ?: $this->logger;
        // 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
                $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'], $logger
                    );
                }
            }
        }

        if ( !$server ) {
            throw new InvalidArgumentException(
                __CLASS__ . ": invalid configured server \"$server\"" );
        } elseif ( 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
            if ( preg_match( '/^\[(.+)\]:(\d+)$/', $server, $m ) ) {
                list( $host, $port ) = [ $m[1], (int)$m[2] ]; // (ip, port)
            } elseif ( preg_match( '/^([^:]+):(\d+)$/', $server, $m ) ) {
                list( $host, $port ) = [ $m[1], (int)$m[2] ]; // (ip or path, port)
            } else {
                list( $host, $port ) = [ $server, 6379 ]; // (ip or path, port)
            }
        }

        $conn = new Redis();
        try {
            if ( $this->persistent ) {
                $result = $conn->pconnect( $host, $port, $this->connectTimeout );
            } else {
                $result = $conn->connect( $host, $port, $this->connectTimeout );
            }
            if ( !$result ) {
                $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 ) ) {
                    $logger->error(
                        'Authentication error connecting to "{redis_server}"',
                        [ 'redis_server' => $server ]
                    );
                }
            }
        } catch ( RedisException $e ) {
            $this->downServers[$server] = time() + self::SERVER_DOWN_TTL;
            $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, $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 ) {
                $conn = $connection['conn'];
                $conn->close();
            }
        }
    }
}