SquidPurgeClient.php

Go to the documentation of this file.

<?php
class SquidPurgeClient {
        var $host, $port, $ip;

        var $readState = 'idle';
        var $writeBuffer = '';
        var $requests = array();
        var $currentRequestIndex;

        const EINTR = 4;
        const EAGAIN = 11;
        const EINPROGRESS = 115;
        const BUFFER_SIZE = 8192;

        var $socket;
        
        public function __construct( $server, $options = array() ) {
                $parts = explode( ':', $server, 2 );
                $this->host = $parts[0];
                $this->port = isset( $parts[1] ) ? $parts[1] : 80;
        }

        protected function getSocket() {
                if ( $this->socket !== null ) {
                        return $this->socket;
                }

                $ip = $this->getIP();
                if ( !$ip ) {
                        $this->log( "DNS error" );
                        $this->markDown();
                        return false;
                }
                $this->socket = socket_create( AF_INET, SOCK_STREAM, SOL_TCP );
                socket_set_nonblock( $this->socket );
                wfSuppressWarnings();
                $ok = socket_connect( $this->socket, $ip, $this->port );
                wfRestoreWarnings();
                if ( !$ok ) {
                        $error = socket_last_error( $this->socket );
                        if ( $error !== self::EINPROGRESS ) {
                                $this->log( "connection error: " . socket_strerror( $error ) );
                                $this->markDown();
                                return false;
                        }
                }

                return $this->socket;
        }

        public function getReadSocketsForSelect() {
                if ( $this->readState == 'idle' ) {
                        return array();
                }
                $socket = $this->getSocket();
                if ( $socket === false ) {
                        return array();
                }
                return array( $socket );
        }

        public function getWriteSocketsForSelect() {
                if ( !strlen( $this->writeBuffer ) ) {
                        return array();
                }
                $socket = $this->getSocket();
                if ( $socket === false ) {
                        return array();
                }
                return array( $socket );
        }

        protected function getIP() {
                if ( $this->ip === null ) {
                        if ( IP::isIPv4( $this->host ) ) {
                                $this->ip = $this->host;
                        } elseif ( IP::isIPv6( $this->host ) ) {
                                throw new MWException( '$wgSquidServers does not support IPv6' );
                        } else {
                                wfSuppressWarnings();
                                $this->ip = gethostbyname( $this->host );
                                if ( $this->ip === $this->host ) {
                                        $this->ip = false;
                                }
                                wfRestoreWarnings();
                        }
                }
                return $this->ip;
        }

        protected function markDown() {
                $this->close();
                $this->socket = false;
        }

        public function close() {
                if ( $this->socket ) {
                        wfSuppressWarnings();
                        socket_set_block( $this->socket );
                        socket_shutdown( $this->socket );
                        socket_close( $this->socket );
                        wfRestoreWarnings();
                }
                $this->socket = null;
                $this->readBuffer = '';
                // Write buffer is kept since it may contain a request for the next socket
        }

        public function queuePurge( $url ) {
                $url = SquidUpdate::expand( str_replace( "\n", '', $url ) );
                $this->requests[] = "PURGE $url HTTP/1.0\r\n" .
                        "Connection: Keep-Alive\r\n" .
                        "Proxy-Connection: Keep-Alive\r\n" .
                        "User-Agent: " . Http::userAgent() . ' ' . __CLASS__ . "\r\n\r\n";
                if ( $this->currentRequestIndex === null ) {
                        $this->nextRequest();
                }
        }

        public function isIdle() {
                return strlen( $this->writeBuffer ) == 0 && $this->readState == 'idle';
        }

        public function doWrites() {
                if ( !strlen( $this->writeBuffer ) ) {
                        return;
                }
                $socket = $this->getSocket();
                if ( !$socket ) {
                        return;
                }

                if ( strlen( $this->writeBuffer ) <= self::BUFFER_SIZE ) {
                        $buf = $this->writeBuffer;
                        $flags = MSG_EOR;
                } else {
                        $buf = substr( $this->writeBuffer, 0, self::BUFFER_SIZE );
                        $flags = 0;
                }
                wfSuppressWarnings();
                $bytesSent = socket_send( $socket, $buf, strlen( $buf ), $flags );
                wfRestoreWarnings();

                if ( $bytesSent === false ) {
                        $error = socket_last_error( $socket );
                        if ( $error != self::EAGAIN && $error != self::EINTR ) {
                                $this->log( 'write error: ' . socket_strerror( $error ) );
                                $this->markDown();
                        }
                        return;
                }

                $this->writeBuffer = substr( $this->writeBuffer, $bytesSent );
        }

        public function doReads() {
                $socket = $this->getSocket();
                if ( !$socket ) {
                        return;
                }

                $buf = '';
                wfSuppressWarnings();
                $bytesRead = socket_recv( $socket, $buf, self::BUFFER_SIZE, 0 );
                wfRestoreWarnings();
                if ( $bytesRead === false ) {
                        $error = socket_last_error( $socket );
                        if ( $error != self::EAGAIN && $error != self::EINTR ) {
                                $this->log( 'read error: ' . socket_strerror( $error ) );
                                $this->markDown();
                                return;
                        }
                } elseif ( $bytesRead === 0 ) {
                        // Assume EOF
                        $this->close();
                        return;
                }

                $this->readBuffer .= $buf;
                while ( $this->socket && $this->processReadBuffer() === 'continue' );
        }

        protected function processReadBuffer() {
                switch ( $this->readState ) {
                case 'idle':
                        return 'done';
                case 'status':
                case 'header':
                        $lines = explode( "\r\n", $this->readBuffer, 2 );
                        if ( count( $lines ) < 2 ) {
                                return 'done';
                        }
                        if ( $this->readState == 'status' )  {
                                $this->processStatusLine( $lines[0] );
                        } else { // header
                                $this->processHeaderLine( $lines[0] );
                        }
                        $this->readBuffer = $lines[1];
                        return 'continue';
                case 'body':
                        if ( $this->bodyRemaining !== null ) {
                                if ( $this->bodyRemaining > strlen( $this->readBuffer ) ) {
                                        $this->bodyRemaining -= strlen( $this->readBuffer );
                                        $this->readBuffer = '';
                                        return 'done';
                                } else {
                                        $this->readBuffer = substr( $this->readBuffer, $this->bodyRemaining );
                                        $this->bodyRemaining = 0;
                                        $this->nextRequest();
                                        return 'continue';
                                }
                        } else {
                                // No content length, read all data to EOF
                                $this->readBuffer = '';
                                return 'done';
                        }
                default:
                        throw new MWException( __METHOD__.': unexpected state' );
                }
        }

        protected function processStatusLine( $line ) {
                if ( !preg_match( '!^HTTP/(\d+)\.(\d+) (\d{3}) (.*)$!', $line, $m ) ) {
                        $this->log( 'invalid status line' );
                        $this->markDown();
                        return;
                }
                list( , , , $status, $reason ) = $m;
                $status = intval( $status );
                if ( $status !== 200 && $status !== 404 ) {
                        $this->log( "unexpected status code: $status $reason" );
                        $this->markDown();
                        return;
                }
                $this->readState = 'header';
        }

        protected function processHeaderLine( $line ) {
                if ( preg_match( '/^Content-Length: (\d+)$/i', $line, $m ) ) {
                        $this->bodyRemaining = intval( $m[1] );
                } elseif ( $line === '' ) {
                        $this->readState = 'body';
                }
        }

        protected function nextRequest() {
                if ( $this->currentRequestIndex !== null ) {
                        unset( $this->requests[$this->currentRequestIndex] );
                }
                if ( count( $this->requests ) ) {
                        $this->readState = 'status';
                        $this->currentRequestIndex = key( $this->requests );
                        $this->writeBuffer = $this->requests[$this->currentRequestIndex];
                } else {
                        $this->readState = 'idle';
                        $this->currentRequestIndex = null;
                        $this->writeBuffer = '';
                }
                $this->bodyRemaining = null;
        }

        protected function log( $msg ) {
                wfDebugLog( 'squid', __CLASS__." ($this->host): $msg\n" );
        }
}

class SquidPurgeClientPool {

        var $clients = array();
        var $timeout = 5;

        function __construct( $options = array() ) {
                if ( isset( $options['timeout'] ) ) {
                        $this->timeout = $options['timeout'];
                }
        }

        public function addClient( $client ) {
                $this->clients[] = $client;
        }

        public function run() {
                $done = false;
                $startTime = microtime( true );
                while ( !$done ) {
                        $readSockets = $writeSockets = array();
                        foreach ( $this->clients as $clientIndex => $client ) {
                                $sockets = $client->getReadSocketsForSelect();
                                foreach ( $sockets as $i => $socket ) {
                                        $readSockets["$clientIndex/$i"] = $socket;
                                }
                                $sockets = $client->getWriteSocketsForSelect();
                                foreach ( $sockets as $i => $socket ) {
                                        $writeSockets["$clientIndex/$i"] = $socket;
                                }
                        }
                        if ( !count( $readSockets ) && !count( $writeSockets ) ) {
                                break;
                        }
                        $exceptSockets = null;
                        $timeout = min( $startTime + $this->timeout - microtime( true ), 1 );
                        wfSuppressWarnings();
                        $numReady = socket_select( $readSockets, $writeSockets, $exceptSockets, $timeout );
                        wfRestoreWarnings();
                        if ( $numReady === false ) {
                                wfDebugLog( 'squid', __METHOD__.': Error in stream_select: ' . 
                                        socket_strerror( socket_last_error() ) . "\n" );
                                break;
                        }
                        // Check for timeout, use 1% tolerance since we aimed at having socket_select()
                        // exit at precisely the overall timeout
                        if ( microtime( true ) - $startTime > $this->timeout * 0.99 ) {
                                wfDebugLog( 'squid', __CLASS__.": timeout ({$this->timeout}s)\n" );
                                break;
                        } elseif ( !$numReady ) {
                                continue;
                        }

                        foreach ( $readSockets as $key => $socket ) {
                                list( $clientIndex, ) = explode( '/', $key );
                                $client = $this->clients[$clientIndex];
                                $client->doReads();
                        }
                        foreach ( $writeSockets as $key => $socket ) {
                                list( $clientIndex, ) = explode( '/', $key );
                                $client = $this->clients[$clientIndex];
                                $client->doWrites();
                        }

                        $done = true;
                        foreach ( $this->clients as $client ) {
                                if ( !$client->isIdle() ) {
                                        $done = false;
                                }
                        }
                }
                foreach ( $this->clients as $client ) {
                        $client->close();
                }
        }
}