RedisBagOStuff.php

Go to the documentation of this file.

<?php
class RedisBagOStuff extends BagOStuff {
    protected $redisPool;
    protected $servers;
    protected $automaticFailover;

    function __construct( $params ) {
        $redisConf = array( 'serializer' => 'none' ); // manage that in this class
        foreach ( array( 'connectTimeout', 'persistent', 'password' ) as $opt ) {
            if ( isset( $params[$opt] ) ) {
                $redisConf[$opt] = $params[$opt];
            }
        }
        $this->redisPool = RedisConnectionPool::singleton( $redisConf );

        $this->servers = $params['servers'];
        if ( isset( $params['automaticFailover'] ) ) {
            $this->automaticFailover = $params['automaticFailover'];
        } else {
            $this->automaticFailover = true;
        }
    }

    public function get( $key, &$casToken = null ) {
        $section = new ProfileSection( __METHOD__ );

        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 ) {
        $section = new ProfileSection( __METHOD__ );

        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 cas( $casToken, $key, $value, $expiry = 0 ) {
        $section = new ProfileSection( __METHOD__ );

        list( $server, $conn ) = $this->getConnection( $key );
        if ( !$conn ) {
            return false;
        }
        $expiry = $this->convertToRelative( $expiry );
        try {
            $conn->watch( $key );

            if ( $this->serialize( $this->get( $key ) ) !== $casToken ) {
                $conn->unwatch();
                return false;
            }

            // multi()/exec() will fail atomically if the key changed since watch()
            $conn->multi();
            if ( $expiry ) {
                $conn->setex( $key, $expiry, $this->serialize( $value ) );
            } else {
                // No expiry, that is very different from zero expiry in Redis
                $conn->set( $key, $this->serialize( $value ) );
            }
            $result = ( $conn->exec() == array( true ) );
        } catch ( RedisException $e ) {
            $result = false;
            $this->handleException( $conn, $e );
        }

        $this->logRequest( 'cas', $key, $server, $result );
        return $result;
    }

    public function delete( $key, $time = 0 ) {
        $section = new ProfileSection( __METHOD__ );

        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 ) {
        $section = new ProfileSection( __METHOD__ );

        $batches = array();
        $conns = array();
        foreach ( $keys as $key ) {
            list( $server, $conn ) = $this->getConnection( $key );
            if ( !$conn ) {
                continue;
            }
            $conns[$server] = $conn;
            $batches[$server][] = $key;
        }
        $result = array();
        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 ) {
        $section = new ProfileSection( __METHOD__ );

        $batches = array();
        $conns = array();
        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 ) {
        $section = new ProfileSection( __METHOD__ );

        list( $server, $conn ) = $this->getConnection( $key );
        if ( !$conn ) {
            return false;
        }
        $expiry = $this->convertToRelative( $expiry );
        try {
            if ( $expiry ) {
                $conn->multi();
                $conn->setnx( $key, $this->serialize( $value ) );
                $conn->expire( $key, $expiry );
                $result = ( $conn->exec() == array( true, true ) );
            } 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 incr( $key, $value = 1 ) {
        $section = new ProfileSection( __METHOD__ );

        list( $server, $conn ) = $this->getConnection( $key );
        if ( !$conn ) {
            return false;
        }
        if ( !$conn->exists( $key ) ) {
            return null;
        }
        try {
            $result = $conn->incrBy( $key, $value );
        } catch ( RedisException $e ) {
            $result = false;
            $this->handleException( $conn, $e );
        }

        $this->logRequest( 'incr', $key, $server, $result );
        return $result;
    }
    protected function serialize( $data ) {
        // Serialize anything but integers so INCR/DECR work
        // Do not store integer-like strings as integers to avoid type confusion (bug 60563)
        return is_int( $data ) ? $data : serialize( $data );
    }

    protected function unserialize( $data ) {
        return ctype_digit( $data ) ? intval( $data ) : unserialize( $data );
    }

    protected function getConnection( $key ) {
        if ( count( $this->servers ) === 1 ) {
            $candidates = $this->servers;
        } else {
            $candidates = $this->servers;
            ArrayUtils::consistentHashSort( $candidates, $key, '/' );
            if ( !$this->automaticFailover ) {
                $candidates = array_slice( $candidates, 0, 1 );
            }
        }

        foreach ( $candidates as $server ) {
            $conn = $this->redisPool->getConnection( $server );
            if ( $conn ) {
                return array( $server, $conn );
            }
        }
        $this->setLastError( BagOStuff::ERR_UNREACHABLE );
        return array( false, false );
    }

    protected function logError( $msg ) {
        wfDebugLog( 'redis', "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" ) );
    }
}