+++ /dev/null
-<?php
-/**
- * An HTTP 1.0 client built for the purposes of purging Squid and Varnish.
- * Uses asynchronous I/O, allowing purges to be done in a highly parallel
- * manner.
- *
- * Could be replaced by curl_multi_exec() or some such.
- */
-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;
-
- /**
- * The socket resource, or null for unconnected, or false for disabled due to error
- */
- var $socket;
-
- public function __construct( $server, $options = array() ) {
- $parts = explode( ':', $server, 2 );
- $this->host = $parts[0];
- $this->port = isset( $parts[1] ) ? $parts[1] : 80;
- }
-
- /**
- * Open a socket if there isn't one open already, return it.
- * Returns false on error.
- */
- 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;
- }
-
- /**
- * Get read socket array for select()
- */
- public function getReadSocketsForSelect() {
- if ( $this->readState == 'idle' ) {
- return array();
- }
- $socket = $this->getSocket();
- if ( $socket === false ) {
- return array();
- }
- return array( $socket );
- }
-
- /**
- * Get write socket array for select()
- */
- public function getWriteSocketsForSelect() {
- if ( !strlen( $this->writeBuffer ) ) {
- return array();
- }
- $socket = $this->getSocket();
- if ( $socket === false ) {
- return array();
- }
- return array( $socket );
- }
-
- /**
- * Get the host's IP address.
- * Does not support IPv6 at present due to the lack of a convenient interface in PHP.
- */
- 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;
- }
-
- /**
- * Close the socket and ignore any future purge requests.
- * This is called if there is a protocol error.
- */
- protected function markDown() {
- $this->close();
- $this->socket = false;
- }
-
- /**
- * Close the socket but allow it to be reopened for future purge requests
- */
- 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
- }
-
- /**
- * Queue a purge operation
- */
- public function queuePurge( $url ) {
- $url = 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';
- }
-
- /**
- * Perform pending writes. Call this when socket_select() indicates that writing will not block.
- */
- 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 );
- }
-
- /**
- * Read some data. Call this when socket_select() indicates that the read buffer is non-empty.
- */
- 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();
- }
- }
-}
scripts.mit.edu / autoinstallsdev / mediawiki.git / blobdiff