includes/SquidPurgeClient.php

   1 <?php
   2 /**
   3  * Squid and Varnish cache purging.
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License as published by
   7  * the Free Software Foundation; either version 2 of the License, or
   8  * (at your option) any later version.
   9  *
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  13  * GNU General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU General Public License along
  16  * with this program; if not, write to the Free Software Foundation, Inc.,
  17  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
  18  * http://www.gnu.org/copyleft/gpl.html
  19  *
  20  * @file
  21  */
  22
  23 /**
  24  * An HTTP 1.0 client built for the purposes of purging Squid and Varnish.
  25  * Uses asynchronous I/O, allowing purges to be done in a highly parallel
  26  * manner.
  27  *
  28  * Could be replaced by curl_multi_exec() or some such.
  29  */
  30 class SquidPurgeClient {
  31         /** @var string */
  32         protected $host;
  33
  34         /** @var int */
  35         protected $port;
  36
  37         /** @var string|bool */
  38         protected $ip;
  39
  40         /** @var string */
  41         protected $readState = 'idle';
  42
  43         /** @var string */
  44         protected $writeBuffer = '';
  45
  46         /** @var array */
  47         protected $requests = array();
  48
  49         /** @var mixed */
  50         protected $currentRequestIndex;
  51
  52         const EINTR = 4;
  53         const EAGAIN = 11;
  54         const EINPROGRESS = 115;
  55         const BUFFER_SIZE = 8192;
  56
  57         /**
  58          * @var resource|null The socket resource, or null for unconnected, or false
  59          *   for disabled due to error.
  60          */
  61         protected $socket;
  62
  63         /** @var string */
  64         protected $readBuffer;
  65
  66         /** @var int */
  67         protected $bodyRemaining;
  68
  69         /**
  70          * @param string $server
  71          * @param array $options
  72          */
  73         public function __construct( $server, $options = array() ) {
  74                 $parts = explode( ':', $server, 2 );
  75                 $this->host = $parts[0];
  76                 $this->port = isset( $parts[1] ) ? $parts[1] : 80;
  77         }
  78
  79         /**
  80          * Open a socket if there isn't one open already, return it.
  81          * Returns false on error.
  82          *
  83          * @return bool|resource
  84          */
  85         protected function getSocket() {
  86                 if ( $this->socket !== null ) {
  87                         return $this->socket;
  88                 }
  89
  90                 $ip = $this->getIP();
  91                 if ( !$ip ) {
  92                         $this->log( "DNS error" );
  93                         $this->markDown();
  94                         return false;
  95                 }
  96                 $this->socket = socket_create( AF_INET, SOCK_STREAM, SOL_TCP );
  97                 socket_set_nonblock( $this->socket );
  98                 wfSuppressWarnings();
  99                 $ok = socket_connect( $this->socket, $ip, $this->port );
 100                 wfRestoreWarnings();
 101                 if ( !$ok ) {
 102                         $error = socket_last_error( $this->socket );
 103                         if ( $error !== self::EINPROGRESS ) {
 104                                 $this->log( "connection error: " . socket_strerror( $error ) );
 105                                 $this->markDown();
 106                                 return false;
 107                         }
 108                 }
 109
 110                 return $this->socket;
 111         }
 112
 113         /**
 114          * Get read socket array for select()
 115          * @return array
 116          */
 117         public function getReadSocketsForSelect() {
 118                 if ( $this->readState == 'idle' ) {
 119                         return array();
 120                 }
 121                 $socket = $this->getSocket();
 122                 if ( $socket === false ) {
 123                         return array();
 124                 }
 125                 return array( $socket );
 126         }
 127
 128         /**
 129          * Get write socket array for select()
 130          * @return array
 131          */
 132         public function getWriteSocketsForSelect() {
 133                 if ( !strlen( $this->writeBuffer ) ) {
 134                         return array();
 135                 }
 136                 $socket = $this->getSocket();
 137                 if ( $socket === false ) {
 138                         return array();
 139                 }
 140                 return array( $socket );
 141         }
 142
 143         /**
 144          * Get the host's IP address.
 145          * Does not support IPv6 at present due to the lack of a convenient interface in PHP.
 146          * @throws MWException
 147          * @return string
 148          */
 149         protected function getIP() {
 150                 if ( $this->ip === null ) {
 151                         if ( IP::isIPv4( $this->host ) ) {
 152                                 $this->ip = $this->host;
 153                         } elseif ( IP::isIPv6( $this->host ) ) {
 154                                 throw new MWException( '$wgSquidServers does not support IPv6' );
 155                         } else {
 156                                 wfSuppressWarnings();
 157                                 $this->ip = gethostbyname( $this->host );
 158                                 if ( $this->ip === $this->host ) {
 159                                         $this->ip = false;
 160                                 }
 161                                 wfRestoreWarnings();
 162                         }
 163                 }
 164                 return $this->ip;
 165         }
 166
 167         /**
 168          * Close the socket and ignore any future purge requests.
 169          * This is called if there is a protocol error.
 170          */
 171         protected function markDown() {
 172                 $this->close();
 173                 $this->socket = false;
 174         }
 175
 176         /**
 177          * Close the socket but allow it to be reopened for future purge requests
 178          */
 179         public function close() {
 180                 if ( $this->socket ) {
 181                         wfSuppressWarnings();
 182                         socket_set_block( $this->socket );
 183                         socket_shutdown( $this->socket );
 184                         socket_close( $this->socket );
 185                         wfRestoreWarnings();
 186                 }
 187                 $this->socket = null;
 188                 $this->readBuffer = '';
 189                 // Write buffer is kept since it may contain a request for the next socket
 190         }
 191
 192         /**
 193          * Queue a purge operation
 194          *
 195          * @param string $url
 196          */
 197         public function queuePurge( $url ) {
 198                 global $wgSquidPurgeUseHostHeader;
 199                 $url = SquidUpdate::expand( str_replace( "\n", '', $url ) );
 200                 $request = array();
 201                 if ( $wgSquidPurgeUseHostHeader ) {
 202                         $url = wfParseUrl( $url );
 203                         $host = $url['host'];
 204                         if ( isset( $url['port'] ) && strlen( $url['port'] ) > 0 ) {
 205                                 $host .= ":" . $url['port'];
 206                         }
 207                         $path = $url['path'];
 208                         if ( isset( $url['query'] ) && is_string( $url['query'] ) ) {
 209                                 $path = wfAppendQuery( $path, $url['query'] );
 210                         }
 211                         $request[] = "PURGE $path HTTP/1.1";
 212                         $request[] = "Host: $host";
 213                 } else {
 214                         $request[] = "PURGE $url HTTP/1.0";
 215                 }
 216                 $request[] = "Connection: Keep-Alive";
 217                 $request[] = "Proxy-Connection: Keep-Alive";
 218                 $request[] = "User-Agent: " . Http::userAgent() . ' ' . __CLASS__;
 219                 // Two ''s to create \r\n\r\n
 220                 $request[] = '';
 221                 $request[] = '';
 222
 223                 $this->requests[] = implode( "\r\n", $request );
 224                 if ( $this->currentRequestIndex === null ) {
 225                         $this->nextRequest();
 226                 }
 227         }
 228
 229         /**
 230          * @return bool
 231          */
 232         public function isIdle() {
 233                 return strlen( $this->writeBuffer ) == 0 && $this->readState == 'idle';
 234         }
 235
 236         /**
 237          * Perform pending writes. Call this when socket_select() indicates that writing will not block.
 238          */
 239         public function doWrites() {
 240                 if ( !strlen( $this->writeBuffer ) ) {
 241                         return;
 242                 }
 243                 $socket = $this->getSocket();
 244                 if ( !$socket ) {
 245                         return;
 246                 }
 247
 248                 if ( strlen( $this->writeBuffer ) <= self::BUFFER_SIZE ) {
 249                         $buf = $this->writeBuffer;
 250                         $flags = MSG_EOR;
 251                 } else {
 252                         $buf = substr( $this->writeBuffer, 0, self::BUFFER_SIZE );
 253                         $flags = 0;
 254                 }
 255                 wfSuppressWarnings();
 256                 $bytesSent = socket_send( $socket, $buf, strlen( $buf ), $flags );
 257                 wfRestoreWarnings();
 258
 259                 if ( $bytesSent === false ) {
 260                         $error = socket_last_error( $socket );
 261                         if ( $error != self::EAGAIN && $error != self::EINTR ) {
 262                                 $this->log( 'write error: ' . socket_strerror( $error ) );
 263                                 $this->markDown();
 264                         }
 265                         return;
 266                 }
 267
 268                 $this->writeBuffer = substr( $this->writeBuffer, $bytesSent );
 269         }
 270
 271         /**
 272          * Read some data. Call this when socket_select() indicates that the read buffer is non-empty.
 273          */
 274         public function doReads() {
 275                 $socket = $this->getSocket();
 276                 if ( !$socket ) {
 277                         return;
 278                 }
 279
 280                 $buf = '';
 281                 wfSuppressWarnings();
 282                 $bytesRead = socket_recv( $socket, $buf, self::BUFFER_SIZE, 0 );
 283                 wfRestoreWarnings();
 284                 if ( $bytesRead === false ) {
 285                         $error = socket_last_error( $socket );
 286                         if ( $error != self::EAGAIN && $error != self::EINTR ) {
 287                                 $this->log( 'read error: ' . socket_strerror( $error ) );
 288                                 $this->markDown();
 289                                 return;
 290                         }
 291                 } elseif ( $bytesRead === 0 ) {
 292                         // Assume EOF
 293                         $this->close();
 294                         return;
 295                 }
 296
 297                 $this->readBuffer .= $buf;
 298                 while ( $this->socket && $this->processReadBuffer() === 'continue' );
 299         }
 300
 301         /**
 302          * @throws MWException
 303          * @return string
 304          */
 305         protected function processReadBuffer() {
 306                 switch ( $this->readState ) {
 307                 case 'idle':
 308                         return 'done';
 309                 case 'status':
 310                 case 'header':
 311                         $lines = explode( "\r\n", $this->readBuffer, 2 );
 312                         if ( count( $lines ) < 2 ) {
 313                                 return 'done';
 314                         }
 315                         if ( $this->readState == 'status' ) {
 316                                 $this->processStatusLine( $lines[0] );
 317                         } else { // header
 318                                 $this->processHeaderLine( $lines[0] );
 319                         }
 320                         $this->readBuffer = $lines[1];
 321                         return 'continue';
 322                 case 'body':
 323                         if ( $this->bodyRemaining !== null ) {
 324                                 if ( $this->bodyRemaining > strlen( $this->readBuffer ) ) {
 325                                         $this->bodyRemaining -= strlen( $this->readBuffer );
 326                                         $this->readBuffer = '';
 327                                         return 'done';
 328                                 } else {
 329                                         $this->readBuffer = substr( $this->readBuffer, $this->bodyRemaining );
 330                                         $this->bodyRemaining = 0;
 331                                         $this->nextRequest();
 332                                         return 'continue';
 333                                 }
 334                         } else {
 335                                 // No content length, read all data to EOF
 336                                 $this->readBuffer = '';
 337                                 return 'done';
 338                         }
 339                 default:
 340                         throw new MWException( __METHOD__ . ': unexpected state' );
 341                 }
 342         }
 343
 344         /**
 345          * @param string $line
 346          * @return
 347          */
 348         protected function processStatusLine( $line ) {
 349                 if ( !preg_match( '!^HTTP/(\d+)\.(\d+) (\d{3}) (.*)$!', $line, $m ) ) {
 350                         $this->log( 'invalid status line' );
 351                         $this->markDown();
 352                         return;
 353                 }
 354                 list( , , , $status, $reason ) = $m;
 355                 $status = intval( $status );
 356                 if ( $status !== 200 && $status !== 404 ) {
 357                         $this->log( "unexpected status code: $status $reason" );
 358                         $this->markDown();
 359                         return;
 360                 }
 361                 $this->readState = 'header';
 362         }
 363
 364         /**
 365          * @param string $line
 366          */
 367         protected function processHeaderLine( $line ) {
 368                 if ( preg_match( '/^Content-Length: (\d+)$/i', $line, $m ) ) {
 369                         $this->bodyRemaining = intval( $m[1] );
 370                 } elseif ( $line === '' ) {
 371                         $this->readState = 'body';
 372                 }
 373         }
 374
 375         protected function nextRequest() {
 376                 if ( $this->currentRequestIndex !== null ) {
 377                         unset( $this->requests[$this->currentRequestIndex] );
 378                 }
 379                 if ( count( $this->requests ) ) {
 380                         $this->readState = 'status';
 381                         $this->currentRequestIndex = key( $this->requests );
 382                         $this->writeBuffer = $this->requests[$this->currentRequestIndex];
 383                 } else {
 384                         $this->readState = 'idle';
 385                         $this->currentRequestIndex = null;
 386                         $this->writeBuffer = '';
 387                 }
 388                 $this->bodyRemaining = null;
 389         }
 390
 391         /**
 392          * @param string $msg
 393          */
 394         protected function log( $msg ) {
 395                 wfDebugLog( 'squid', __CLASS__ . " ($this->host): $msg" );
 396         }
 397 }
 398
 399 class SquidPurgeClientPool {
 400         /** @var array of SquidPurgeClient */
 401         protected $clients = array();
 402
 403         /** @var int */
 404         protected $timeout = 5;
 405
 406         /**
 407          * @param array $options
 408          */
 409         function __construct( $options = array() ) {
 410                 if ( isset( $options['timeout'] ) ) {
 411                         $this->timeout = $options['timeout'];
 412                 }
 413         }
 414
 415         /**
 416          * @param SquidPurgeClient $client
 417          * @return void
 418          */
 419         public function addClient( $client ) {
 420                 $this->clients[] = $client;
 421         }
 422
 423         public function run() {
 424                 $done = false;
 425                 $startTime = microtime( true );
 426                 while ( !$done ) {
 427                         $readSockets = $writeSockets = array();
 428                         /**
 429                          * @var $client SquidPurgeClient
 430                          */
 431                         foreach ( $this->clients as $clientIndex => $client ) {
 432                                 $sockets = $client->getReadSocketsForSelect();
 433                                 foreach ( $sockets as $i => $socket ) {
 434                                         $readSockets["$clientIndex/$i"] = $socket;
 435                                 }
 436                                 $sockets = $client->getWriteSocketsForSelect();
 437                                 foreach ( $sockets as $i => $socket ) {
 438                                         $writeSockets["$clientIndex/$i"] = $socket;
 439                                 }
 440                         }
 441                         if ( !count( $readSockets ) && !count( $writeSockets ) ) {
 442                                 break;
 443                         }
 444                         $exceptSockets = null;
 445                         $timeout = min( $startTime + $this->timeout - microtime( true ), 1 );
 446                         wfSuppressWarnings();
 447                         $numReady = socket_select( $readSockets, $writeSockets, $exceptSockets, $timeout );
 448                         wfRestoreWarnings();
 449                         if ( $numReady === false ) {
 450                                 wfDebugLog( 'squid', __METHOD__ . ': Error in stream_select: ' .
 451                                         socket_strerror( socket_last_error() ) . "\n" );
 452                                 break;
 453                         }
 454                         // Check for timeout, use 1% tolerance since we aimed at having socket_select()
 455                         // exit at precisely the overall timeout
 456                         if ( microtime( true ) - $startTime > $this->timeout * 0.99 ) {
 457                                 wfDebugLog( 'squid', __CLASS__ . ": timeout ({$this->timeout}s)\n" );
 458                                 break;
 459                         } elseif ( !$numReady ) {
 460                                 continue;
 461                         }
 462
 463                         foreach ( $readSockets as $key => $socket ) {
 464                                 list( $clientIndex, ) = explode( '/', $key );
 465                                 $client = $this->clients[$clientIndex];
 466                                 $client->doReads();
 467                         }
 468                         foreach ( $writeSockets as $key => $socket ) {
 469                                 list( $clientIndex, ) = explode( '/', $key );
 470                                 $client = $this->clients[$clientIndex];
 471                                 $client->doWrites();
 472                         }
 473
 474                         $done = true;
 475                         foreach ( $this->clients as $client ) {
 476                                 if ( !$client->isIdle() ) {
 477                                         $done = false;
 478                                 }
 479                         }
 480                 }
 481                 foreach ( $this->clients as $client ) {
 482                         $client->close();
 483                 }
 484         }
 485 }