ChronologyProtector.php

Go to the documentation of this file.

<?php
use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerInterface;
use Wikimedia\WaitConditionLoop;

class ChronologyProtector implements LoggerAwareInterface {
    protected $store;
    protected $logger;

    protected $key;
    protected $clientId;
    protected $waitForPosTime;
    protected $waitForPosTimeout = self::POS_WAIT_TIMEOUT;
    protected $enabled = true;
    protected $wait = true;

    protected $initialized = false;
    protected $startupPositions = [];
    protected $shutdownPositions = [];
    protected $shutdownTouchDBs = [];

    const POSITION_TTL = 60;
    const POS_WAIT_TIMEOUT = 5;

    public function __construct( BagOStuff $store, array $client, $posTime = null ) {
        $this->store = $store;
        $this->clientId = md5( $client['ip'] . "\n" . $client['agent'] );
        $this->key = $store->makeGlobalKey( __CLASS__, $this->clientId );
        $this->waitForPosTime = $posTime;
        $this->logger = new \Psr\Log\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 ( !empty( $this->startupPositions[$masterName] ) ) {
            $pos = $this->startupPositions[$masterName];
            $this->logger->info( __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();
            $this->logger->info( __METHOD__ . ": LB for '$masterName' has pos $pos\n" );
            $this->shutdownPositions[$masterName] = $pos;
        } else {
            $this->logger->info( __METHOD__ . ": DB '$masterName' touched\n" );
        }
        $this->shutdownTouchDBs[$masterName] = 1;
    }

    public function shutdown( callable $workCallback = null, $mode = 'sync' ) {
        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->info( __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,
                self::mergePositions( $store->get( $this->key ), $this->shutdownPositions ),
                self::POSITION_TTL,
                ( $mode === 'sync' ) ? $store::WRITE_SYNC : 0
            );
            $store->unlock( $this->key );
        } else {
            $ok = false;
        }

        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 with a certain min
            // timestamp, then block until they appear, or until a timeout is reached.
            if ( $this->waitForPosTime > 0.0 ) {
                $data = null;
                $loop = new WaitConditionLoop(
                    function () use ( &$data ) {
                        $data = $this->store->get( $this->key );

                        return ( self::minPosTime( $data ) >= $this->waitForPosTime )
                            ? WaitConditionLoop::CONDITION_REACHED
                            : WaitConditionLoop::CONDITION_CONTINUE;
                    },
                    $this->waitForPosTimeout
                );
                $result = $loop->invoke();
                $waitedMs = $loop->getLastWaitTime() * 1e3;

                if ( $result == $loop::CONDITION_REACHED ) {
                    $msg = "expected and found pos time {$this->waitForPosTime} ({$waitedMs}ms)";
                    $this->logger->debug( $msg );
                } else {
                    $msg = "expected but missed pos time {$this->waitForPosTime} ({$waitedMs}ms)";
                    $this->logger->info( $msg );
                }
            } else {
                $data = $this->store->get( $this->key );
            }

            $this->startupPositions = $data ? $data['positions'] : [];
            $this->logger->info( __METHOD__ . ": key is {$this->key} (read)\n" );
        } else {
            $this->startupPositions = [];
            $this->logger->info( __METHOD__ . ": key is {$this->key} (unread)\n" );
        }
    }

    private static function minPosTime( $data ) {
        if ( !isset( $data['positions'] ) ) {
            return null;
        }

        $min = null;
        foreach ( $data['positions'] as $pos ) {
            $min = $min ? min( $pos->asOfTime(), $min ) : $pos->asOfTime();
        }

        return $min;
    }

    private static function mergePositions( $curValue, array $shutdownPositions ) {
        if ( $curValue === false ) {
            $curPositions = $shutdownPositions;
        } else {
            $curPositions = $curValue['positions'];
            // Use the newest positions for each DB master
            foreach ( $shutdownPositions as $db => $pos ) {
                if ( !isset( $curPositions[$db] )
                    || $pos->asOfTime() > $curPositions[$db]->asOfTime()
                ) {
                    $curPositions[$db] = $pos;
                }
            }
        }

        return [ 'positions' => $curPositions ];
    }
}