RedisLockManager.php

Go to the documentation of this file.

<?php
class RedisLockManager extends QuorumLockManager {
    protected $lockTypeMap = array(
        self::LOCK_SH => self::LOCK_SH,
        self::LOCK_UW => self::LOCK_SH,
        self::LOCK_EX => self::LOCK_EX
    );

    protected $redisPool;

    protected $lockServers = array();

    protected $session = '';

    public function __construct( array $config ) {
        parent::__construct( $config );

        $this->lockServers = $config['lockServers'];
        // Sanitize srvsByBucket config to prevent PHP errors
        $this->srvsByBucket = array_filter( $config['srvsByBucket'], 'is_array' );
        $this->srvsByBucket = array_values( $this->srvsByBucket ); // consecutive

        $config['redisConfig']['serializer'] = 'none';
        $this->redisPool = RedisConnectionPool::singleton( $config['redisConfig'] );

        $this->session = wfRandomString( 32 );
    }

    protected function getLocksOnServer( $lockSrv, array $pathsByType ) {
        $status = Status::newGood();

        $server = $this->lockServers[$lockSrv];
        $conn = $this->redisPool->getConnection( $server );
        if ( !$conn ) {
            foreach ( array_merge( array_values( $pathsByType ) ) as $path ) {
                $status->fatal( 'lockmanager-fail-acquirelock', $path );
            }

            return $status;
        }

        $pathsByKey = array(); // (type:hash => path) map
        foreach ( $pathsByType as $type => $paths ) {
            $typeString = ( $type == LockManager::LOCK_SH ) ? 'SH' : 'EX';
            foreach ( $paths as $path ) {
                $pathsByKey[$this->recordKeyForPath( $path, $typeString )] = $path;
            }
        }

        try {
            static $script =
<<<LUA
            local failed = {}
            -- Load input params (e.g. session, ttl, time of request)
            local rSession, rTTL, rTime = unpack(ARGV)
            -- Check that all the locks can be acquired
            for i,requestKey in ipairs(KEYS) do
                local _, _, rType, resourceKey = string.find(requestKey,"(%w+):(%w+)$")
                local keyIsFree = true
                local currentLocks = redis.call('hKeys',resourceKey)
                for i,lockKey in ipairs(currentLocks) do
                    -- Get the type and session of this lock
                    local _, _, type, session = string.find(lockKey,"(%w+):(%w+)")
                    -- Check any locks that are not owned by this session
                    if session ~= rSession then
                        local lockExpiry = redis.call('hGet',resourceKey,lockKey)
                        if 1*lockExpiry < 1*rTime then
                            -- Lock is stale, so just prune it out
                            redis.call('hDel',resourceKey,lockKey)
                        elseif rType == 'EX' or type == 'EX' then
                            keyIsFree = false
                            break
                        end
                    end
                end
                if not keyIsFree then
                    failed[#failed+1] = requestKey
                end
            end
            -- If all locks could be acquired, then do so
            if #failed == 0 then
                for i,requestKey in ipairs(KEYS) do
                    local _, _, rType, resourceKey = string.find(requestKey,"(%w+):(%w+)$")
                    redis.call('hSet',resourceKey,rType .. ':' .. rSession,rTime + rTTL)
                    -- In addition to invalidation logic, be sure to garbage collect
                    redis.call('expire',resourceKey,rTTL)
                end
            end
            return failed
LUA;
            $res = $conn->luaEval( $script,
                array_merge(
                    array_keys( $pathsByKey ), // KEYS[0], KEYS[1],...,KEYS[N]
                    array(
                        $this->session, // ARGV[1]
                        $this->lockTTL, // ARGV[2]
                        time() // ARGV[3]
                    )
                ),
                count( $pathsByKey ) # number of first argument(s) that are keys
            );
        } catch ( RedisException $e ) {
            $res = false;
            $this->redisPool->handleError( $conn, $e );
        }

        if ( $res === false ) {
            foreach ( array_merge( array_values( $pathsByType ) ) as $path ) {
                $status->fatal( 'lockmanager-fail-acquirelock', $path );
            }
        } else {
            foreach ( $res as $key ) {
                $status->fatal( 'lockmanager-fail-acquirelock', $pathsByKey[$key] );
            }
        }

        return $status;
    }

    protected function freeLocksOnServer( $lockSrv, array $pathsByType ) {
        $status = Status::newGood();

        $server = $this->lockServers[$lockSrv];
        $conn = $this->redisPool->getConnection( $server );
        if ( !$conn ) {
            foreach ( array_merge( array_values( $pathsByType ) ) as $path ) {
                $status->fatal( 'lockmanager-fail-releaselock', $path );
            }

            return $status;
        }

        $pathsByKey = array(); // (type:hash => path) map
        foreach ( $pathsByType as $type => $paths ) {
            $typeString = ( $type == LockManager::LOCK_SH ) ? 'SH' : 'EX';
            foreach ( $paths as $path ) {
                $pathsByKey[$this->recordKeyForPath( $path, $typeString )] = $path;
            }
        }

        try {
            static $script =
<<<LUA
            local failed = {}
            -- Load input params (e.g. session)
            local rSession = unpack(ARGV)
            for i,requestKey in ipairs(KEYS) do
                local _, _, rType, resourceKey = string.find(requestKey,"(%w+):(%w+)$")
                local released = redis.call('hDel',resourceKey,rType .. ':' .. rSession)
                if released > 0 then
                    -- Remove the whole structure if it is now empty
                    if redis.call('hLen',resourceKey) == 0 then
                        redis.call('del',resourceKey)
                    end
                else
                    failed[#failed+1] = requestKey
                end
            end
            return failed
LUA;
            $res = $conn->luaEval( $script,
                array_merge(
                    array_keys( $pathsByKey ), // KEYS[0], KEYS[1],...,KEYS[N]
                    array(
                        $this->session, // ARGV[1]
                    )
                ),
                count( $pathsByKey ) # number of first argument(s) that are keys
            );
        } catch ( RedisException $e ) {
            $res = false;
            $this->redisPool->handleError( $conn, $e );
        }

        if ( $res === false ) {
            foreach ( array_merge( array_values( $pathsByType ) ) as $path ) {
                $status->fatal( 'lockmanager-fail-releaselock', $path );
            }
        } else {
            foreach ( $res as $key ) {
                $status->fatal( 'lockmanager-fail-releaselock', $pathsByKey[$key] );
            }
        }

        return $status;
    }

    protected function releaseAllLocks() {
        return Status::newGood(); // not supported
    }

    protected function isServerUp( $lockSrv ) {
        return (bool)$this->redisPool->getConnection( $this->lockServers[$lockSrv] );
    }

    protected function recordKeyForPath( $path, $type ) {
        return implode( ':',
            array( __CLASS__, 'locks', "$type:" . $this->sha1Base36Absolute( $path ) ) );
    }

    function __destruct() {
        while ( count( $this->locksHeld ) ) {
            $pathsByType = array();
            foreach ( $this->locksHeld as $path => $locks ) {
                foreach ( $locks as $type => $count ) {
                    $pathsByType[$type][] = $path;
                }
            }
            $this->unlockByType( $pathsByType );
        }
    }
}