DBLockManager.php

Go to the documentation of this file.

<?php
abstract class DBLockManager extends QuorumLockManager {
    protected $dbServers; // (DB name => server config array)
    protected $statusCache;

    protected $lockExpiry; // integer number of seconds
    protected $safeDelay; // integer number of seconds

    protected $session = 0; // random integer
    protected $conns = [];

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

        $this->dbServers = isset( $config['dbServers'] )
            ? $config['dbServers']
            : []; // likely just using 'localDBMaster'
        // Sanitize srvsByBucket config to prevent PHP errors
        $this->srvsByBucket = array_filter( $config['dbsByBucket'], 'is_array' );
        $this->srvsByBucket = array_values( $this->srvsByBucket ); // consecutive

        if ( isset( $config['lockExpiry'] ) ) {
            $this->lockExpiry = $config['lockExpiry'];
        } else {
            $met = ini_get( 'max_execution_time' );
            $this->lockExpiry = $met ? $met : 60; // use some sane amount if 0
        }
        $this->safeDelay = ( $this->lockExpiry <= 0 )
            ? 60 // pick a safe-ish number to match DB timeout default
            : $this->lockExpiry; // cover worst case

        foreach ( $this->srvsByBucket as $bucket ) {
            if ( count( $bucket ) > 1 ) { // multiple peers
                // Tracks peers that couldn't be queried recently to avoid lengthy
                // connection timeouts. This is useless if each bucket has one peer.
                $this->statusCache = ObjectCache::getLocalServerInstance();
                break;
            }
        }

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

    // @todo change this code to work in one batch
    protected function getLocksOnServer( $lockSrv, array $pathsByType ) {
        $status = Status::newGood();
        foreach ( $pathsByType as $type => $paths ) {
            $status->merge( $this->doGetLocksOnServer( $lockSrv, $paths, $type ) );
        }

        return $status;
    }

    abstract protected function doGetLocksOnServer( $lockSrv, array $paths, $type );

    protected function freeLocksOnServer( $lockSrv, array $pathsByType ) {
        return Status::newGood();
    }

    protected function isServerUp( $lockSrv ) {
        if ( !$this->cacheCheckFailures( $lockSrv ) ) {
            return false; // recent failure to connect
        }
        try {
            $this->getConnection( $lockSrv );
        } catch ( DBError $e ) {
            $this->cacheRecordFailure( $lockSrv );

            return false; // failed to connect
        }

        return true;
    }

    protected function getConnection( $lockDb ) {
        if ( !isset( $this->conns[$lockDb] ) ) {
            $db = null;
            if ( $lockDb === 'localDBMaster' ) {
                $lb = wfGetLBFactory()->getMainLB( $this->domain );
                $db = $lb->getConnection( DB_MASTER, [], $this->domain );
            } elseif ( isset( $this->dbServers[$lockDb] ) ) {
                $config = $this->dbServers[$lockDb];
                $db = DatabaseBase::factory( $config['type'], $config );
            }
            if ( !$db ) {
                return null; // config error?
            }
            $this->conns[$lockDb] = $db;
            $this->conns[$lockDb]->clearFlag( DBO_TRX );
            # If the connection drops, try to avoid letting the DB rollback
            # and release the locks before the file operations are finished.
            # This won't handle the case of DB server restarts however.
            $options = [];
            if ( $this->lockExpiry > 0 ) {
                $options['connTimeout'] = $this->lockExpiry;
            }
            $this->conns[$lockDb]->setSessionOptions( $options );
            $this->initConnection( $lockDb, $this->conns[$lockDb] );
        }
        if ( !$this->conns[$lockDb]->trxLevel() ) {
            $this->conns[$lockDb]->begin( __METHOD__ ); // start transaction
        }

        return $this->conns[$lockDb];
    }

    protected function initConnection( $lockDb, IDatabase $db ) {
    }

    protected function cacheCheckFailures( $lockDb ) {
        return ( $this->statusCache && $this->safeDelay > 0 )
            ? !$this->statusCache->get( $this->getMissKey( $lockDb ) )
            : true;
    }

    protected function cacheRecordFailure( $lockDb ) {
        return ( $this->statusCache && $this->safeDelay > 0 )
            ? $this->statusCache->set( $this->getMissKey( $lockDb ), 1, $this->safeDelay )
            : true;
    }

    protected function getMissKey( $lockDb ) {
        $lockDb = ( $lockDb === 'localDBMaster' ) ? wfWikiID() : $lockDb; // non-relative
        return 'dblockmanager:downservers:' . str_replace( ' ', '_', $lockDb );
    }

    function __destruct() {
        $this->releaseAllLocks();
        foreach ( $this->conns as $db ) {
            $db->close();
        }
    }
}

class MySqlLockManager extends DBLockManager {
    protected $lockTypeMap = [
        self::LOCK_SH => self::LOCK_SH,
        self::LOCK_UW => self::LOCK_SH,
        self::LOCK_EX => self::LOCK_EX
    ];

    protected function initConnection( $lockDb, IDatabase $db ) {
        # Let this transaction see lock rows from other transactions
        $db->query( "SET SESSION TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;" );
    }

    protected function doGetLocksOnServer( $lockSrv, array $paths, $type ) {
        $status = Status::newGood();

        $db = $this->getConnection( $lockSrv ); // checked in isServerUp()

        $keys = []; // list of hash keys for the paths
        $data = []; // list of rows to insert
        $checkEXKeys = []; // list of hash keys that this has no EX lock on
        # Build up values for INSERT clause
        foreach ( $paths as $path ) {
            $key = $this->sha1Base36Absolute( $path );
            $keys[] = $key;
            $data[] = [ 'fls_key' => $key, 'fls_session' => $this->session ];
            if ( !isset( $this->locksHeld[$path][self::LOCK_EX] ) ) {
                $checkEXKeys[] = $key;
            }
        }

        # Block new writers (both EX and SH locks leave entries here)...
        $db->insert( 'filelocks_shared', $data, __METHOD__, [ 'IGNORE' ] );
        # Actually do the locking queries...
        if ( $type == self::LOCK_SH ) { // reader locks
            $blocked = false;
            # Bail if there are any existing writers...
            if ( count( $checkEXKeys ) ) {
                $blocked = $db->selectField( 'filelocks_exclusive', '1',
                    [ 'fle_key' => $checkEXKeys ],
                    __METHOD__
                );
            }
            # Other prospective writers that haven't yet updated filelocks_exclusive
            # will recheck filelocks_shared after doing so and bail due to this entry.
        } else { // writer locks
            $encSession = $db->addQuotes( $this->session );
            # Bail if there are any existing writers...
            # This may detect readers, but the safe check for them is below.
            # Note: if two writers come at the same time, both bail :)
            $blocked = $db->selectField( 'filelocks_shared', '1',
                [ 'fls_key' => $keys, "fls_session != $encSession" ],
                __METHOD__
            );
            if ( !$blocked ) {
                # Build up values for INSERT clause
                $data = [];
                foreach ( $keys as $key ) {
                    $data[] = [ 'fle_key' => $key ];
                }
                # Block new readers/writers...
                $db->insert( 'filelocks_exclusive', $data, __METHOD__ );
                # Bail if there are any existing readers...
                $blocked = $db->selectField( 'filelocks_shared', '1',
                    [ 'fls_key' => $keys, "fls_session != $encSession" ],
                    __METHOD__
                );
            }
        }

        if ( $blocked ) {
            foreach ( $paths as $path ) {
                $status->fatal( 'lockmanager-fail-acquirelock', $path );
            }
        }

        return $status;
    }

    protected function releaseAllLocks() {
        $status = Status::newGood();

        foreach ( $this->conns as $lockDb => $db ) {
            if ( $db->trxLevel() ) { // in transaction
                try {
                    $db->rollback( __METHOD__ ); // finish transaction and kill any rows
                } catch ( DBError $e ) {
                    $status->fatal( 'lockmanager-fail-db-release', $lockDb );
                }
            }
        }

        return $status;
    }
}

class PostgreSqlLockManager extends DBLockManager {
    protected $lockTypeMap = [
        self::LOCK_SH => self::LOCK_SH,
        self::LOCK_UW => self::LOCK_SH,
        self::LOCK_EX => self::LOCK_EX
    ];

    protected function doGetLocksOnServer( $lockSrv, array $paths, $type ) {
        $status = Status::newGood();
        if ( !count( $paths ) ) {
            return $status; // nothing to lock
        }

        $db = $this->getConnection( $lockSrv ); // checked in isServerUp()
        $bigints = array_unique( array_map(
            function ( $key ) {
                return Wikimedia\base_convert( substr( $key, 0, 15 ), 16, 10 );
            },
            array_map( [ $this, 'sha1Base16Absolute' ], $paths )
        ) );

        // Try to acquire all the locks...
        $fields = [];
        foreach ( $bigints as $bigint ) {
            $fields[] = ( $type == self::LOCK_SH )
                ? "pg_try_advisory_lock_shared({$db->addQuotes( $bigint )}) AS K$bigint"
                : "pg_try_advisory_lock({$db->addQuotes( $bigint )}) AS K$bigint";
        }
        $res = $db->query( 'SELECT ' . implode( ', ', $fields ), __METHOD__ );
        $row = $res->fetchRow();

        if ( in_array( 'f', $row ) ) {
            // Release any acquired locks if some could not be acquired...
            $fields = [];
            foreach ( $row as $kbigint => $ok ) {
                if ( $ok === 't' ) { // locked
                    $bigint = substr( $kbigint, 1 ); // strip off the "K"
                    $fields[] = ( $type == self::LOCK_SH )
                        ? "pg_advisory_unlock_shared({$db->addQuotes( $bigint )})"
                        : "pg_advisory_unlock({$db->addQuotes( $bigint )})";
                }
            }
            if ( count( $fields ) ) {
                $db->query( 'SELECT ' . implode( ', ', $fields ), __METHOD__ );
            }
            foreach ( $paths as $path ) {
                $status->fatal( 'lockmanager-fail-acquirelock', $path );
            }
        }

        return $status;
    }

    protected function releaseAllLocks() {
        $status = Status::newGood();

        foreach ( $this->conns as $lockDb => $db ) {
            try {
                $db->query( "SELECT pg_advisory_unlock_all()", __METHOD__ );
            } catch ( DBError $e ) {
                $status->fatal( 'lockmanager-fail-db-release', $lockDb );
            }
        }

        return $status;
    }
}