ChronologyProtector.php

Go to the documentation of this file.

<?php
namespace Wikimedia\Rdbms;
 
use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;
use Wikimedia\WaitConditionLoop;
use BagOStuff;
 
class ChronologyProtector implements LoggerAwareInterface {
    protected $store;
    protected $logger;
 
    protected $key;
    protected $clientId;
    protected $waitForPosIndex;
    protected $waitForPosStoreTimeout = self::POS_STORE_WAIT_TIMEOUT;
    protected $enabled = true;
    protected $wait = true;
 
    protected $initialized = false;
    protected $startupPositions = [];
    protected $shutdownPositions = [];
    protected $shutdownTouchDBs = [];
 
    const POSITION_TTL = 60;
    const POS_STORE_WAIT_TIMEOUT = 5;
 
    public function __construct( BagOStuff $store, array $client, $posIndex = null ) {
        $this->store = $store;
        $this->clientId = md5( $client['ip'] . "\n" . $client['agent'] );
        $this->key = $store->makeGlobalKey( __CLASS__, $this->clientId, 'v2' );
        $this->waitForPosIndex = $posIndex;
        $this->logger = new NullLogger();
    }
 
    public function setLogger( LoggerInterface $logger ) {
        $this->logger = $logger;
    }
 
    public function setEnabled( $enabled ) {
        $this->enabled = $enabled;
    }
 
    public function setWaitEnabled( $enabled ) {
        $this->wait = $enabled;
    }
 
    public function initLB( ILoadBalancer $lb ) {
        if ( !$this->enabled || $lb->getServerCount() <= 1 ) {
            return; // non-replicated setup or disabled
        }
 
        $this->initPositions();
 
        $masterName = $lb->getServerName( $lb->getWriterIndex() );
        if (
            isset( $this->startupPositions[$masterName] ) &&
            $this->startupPositions[$masterName] instanceof DBMasterPos
        ) {
            $pos = $this->startupPositions[$masterName];
            $this->logger->debug( __METHOD__ . ": LB for '$masterName' set to pos $pos\n" );
            $lb->waitFor( $pos );
        }
    }
 
    public function shutdownLB( ILoadBalancer $lb ) {
        if ( !$this->enabled ) {
            return; // not enabled
        } elseif ( !$lb->hasOrMadeRecentMasterChanges( INF ) ) {
            // Only save the position if writes have been done on the connection
            return;
        }
 
        $masterName = $lb->getServerName( $lb->getWriterIndex() );
        if ( $lb->getServerCount() > 1 ) {
            $pos = $lb->getMasterPos();
            if ( $pos ) {
                $this->logger->debug( __METHOD__ . ": LB for '$masterName' has pos $pos\n" );
                $this->shutdownPositions[$masterName] = $pos;
            }
        } else {
            $this->logger->debug( __METHOD__ . ": DB '$masterName' touched\n" );
        }
        $this->shutdownTouchDBs[$masterName] = 1;
    }
 
    public function shutdown( callable $workCallback = null, $mode = 'sync', &$cpIndex = null ) {
        if ( !$this->enabled ) {
            return [];
        }
 
        $store = $this->store;
        // Some callers might want to know if a user recently touched a DB.
        // These writes do not need to block on all datacenters receiving them.
        foreach ( $this->shutdownTouchDBs as $dbName => $unused ) {
            $store->set(
                $this->getTouchedKey( $this->store, $dbName ),
                microtime( true ),
                $store::TTL_DAY
            );
        }
 
        if ( !count( $this->shutdownPositions ) ) {
            return []; // nothing to save
        }
 
        $this->logger->debug( __METHOD__ . ": saving master pos for " .
            implode( ', ', array_keys( $this->shutdownPositions ) ) . "\n"
        );
 
        // CP-protected writes should overwhemingly go to the master datacenter, so get DC-local
        // lock to merge the values. Use a DC-local get() and a synchronous all-DC set(). This
        // makes it possible for the BagOStuff class to write in parallel to all DCs with one RTT.
        if ( $store->lock( $this->key, 3 ) ) {
            if ( $workCallback ) {
                // Let the store run the work before blocking on a replication sync barrier. By the
                // time it's done with the work, the barrier should be fast if replication caught up.
                $store->addBusyCallback( $workCallback );
            }
            $ok = $store->set(
                $this->key,
                $this->mergePositions(
                    $store->get( $this->key ),
                    $this->shutdownPositions,
                    $cpIndex
                ),
                self::POSITION_TTL,
                ( $mode === 'sync' ) ? $store::WRITE_SYNC : 0
            );
            $store->unlock( $this->key );
        } else {
            $ok = false;
            $cpIndex = null; // nothing saved
        }
 
        if ( !$ok ) {
            $bouncedPositions = $this->shutdownPositions;
            // Raced out too many times or stash is down
            $this->logger->warning( __METHOD__ . ": failed to save master pos for " .
                implode( ', ', array_keys( $this->shutdownPositions ) ) . "\n"
            );
        } elseif ( $mode === 'sync' &&
            $store->getQoS( $store::ATTR_SYNCWRITES ) < $store::QOS_SYNCWRITES_BE
        ) {
            // Positions may not be in all datacenters, force LBFactory to play it safe
            $this->logger->info( __METHOD__ . ": store may not support synchronous writes." );
            $bouncedPositions = $this->shutdownPositions;
        } else {
            $bouncedPositions = [];
        }
 
        return $bouncedPositions;
    }
 
    public function getTouched( $dbName ) {
        return $this->store->get( $this->getTouchedKey( $this->store, $dbName ) );
    }
 
    private function getTouchedKey( BagOStuff $store, $dbName ) {
        return $store->makeGlobalKey( __CLASS__, 'mtime', $this->clientId, $dbName );
    }
 
    protected function initPositions() {
        if ( $this->initialized ) {
            return;
        }
 
        $this->initialized = true;
        if ( $this->wait ) {
            // If there is an expectation to see master positions from a certain write
            // index or higher, then block until it appears, or until a timeout is reached.
            // Since the write index restarts each time the key is created, it is possible that
            // a lagged store has a matching key write index. However, in that case, it should
            // already be expired and thus treated as non-existing, maintaining correctness.
            if ( $this->waitForPosIndex > 0 ) {
                $data = null;
                $loop = new WaitConditionLoop(
                    function () use ( &$data ) {
                        $data = $this->store->get( $this->key );
                        if ( !is_array( $data ) ) {
                            return WaitConditionLoop::CONDITION_CONTINUE; // not found yet
                        } elseif ( !isset( $data['writeIndex'] ) ) {
                            return WaitConditionLoop::CONDITION_REACHED; // b/c
                        }
 
                        return ( $data['writeIndex'] >= $this->waitForPosIndex )
                            ? WaitConditionLoop::CONDITION_REACHED
                            : WaitConditionLoop::CONDITION_CONTINUE;
                    },
                    $this->waitForPosStoreTimeout
                );
                $result = $loop->invoke();
                $waitedMs = $loop->getLastWaitTime() * 1e3;
 
                if ( $result == $loop::CONDITION_REACHED ) {
                    $msg = "expected and found pos index {$this->waitForPosIndex} ({$waitedMs}ms)";
                    $this->logger->debug( $msg );
                } else {
                    $msg = "expected but missed pos index {$this->waitForPosIndex} ({$waitedMs}ms)";
                    $this->logger->info( $msg );
                }
            } else {
                $data = $this->store->get( $this->key );
            }
 
            $this->startupPositions = $data ? $data['positions'] : [];
            $this->logger->debug( __METHOD__ . ": key is {$this->key} (read)\n" );
        } else {
            $this->startupPositions = [];
            $this->logger->debug( __METHOD__ . ": key is {$this->key} (unread)\n" );
        }
    }
 
    protected function mergePositions( $curValue, array $shutdownPositions, &$cpIndex = null ) {
        $curPositions = isset( $curValue['positions'] ) ? $curValue['positions'] : [];
        // Use the newest positions for each DB master
        foreach ( $shutdownPositions as $db => $pos ) {
            if (
                !isset( $curPositions[$db] ) ||
                !( $curPositions[$db] instanceof DBMasterPos ) ||
                $pos->asOfTime() > $curPositions[$db]->asOfTime()
            ) {
                $curPositions[$db] = $pos;
            }
        }
 
        $cpIndex = isset( $curValue['writeIndex'] ) ? $curValue['writeIndex'] : 0;
 
        return [
            'positions' => $curPositions,
            'writeIndex' => ++$cpIndex
        ];
    }
}