3 * Squid and Varnish cache purging.
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.
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.
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
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
28 * Could be replaced by curl_multi_exec() or some such.
30 class SquidPurgeClient
{
31 var $host, $port, $ip;
33 var $readState = 'idle';
34 var $writeBuffer = '';
35 var $requests = array();
36 var $currentRequestIndex;
40 const EINPROGRESS
= 115;
41 const BUFFER_SIZE
= 8192;
44 * The socket resource, or null for unconnected, or false for disabled due to error
53 * @param $server string
54 * @param $options array
56 public function __construct( $server, $options = array() ) {
57 $parts = explode( ':', $server, 2 );
58 $this->host
= $parts[0];
59 $this->port
= isset( $parts[1] ) ?
$parts[1] : 80;
63 * Open a socket if there isn't one open already, return it.
64 * Returns false on error.
66 * @return bool|resource
68 protected function getSocket() {
69 if ( $this->socket
!== null ) {
75 $this->log( "DNS error" );
79 $this->socket
= socket_create( AF_INET
, SOCK_STREAM
, SOL_TCP
);
80 socket_set_nonblock( $this->socket
);
82 $ok = socket_connect( $this->socket
, $ip, $this->port
);
85 $error = socket_last_error( $this->socket
);
86 if ( $error !== self
::EINPROGRESS
) {
87 $this->log( "connection error: " . socket_strerror( $error ) );
97 * Get read socket array for select()
100 public function getReadSocketsForSelect() {
101 if ( $this->readState
== 'idle' ) {
104 $socket = $this->getSocket();
105 if ( $socket === false ) {
108 return array( $socket );
112 * Get write socket array for select()
115 public function getWriteSocketsForSelect() {
116 if ( !strlen( $this->writeBuffer
) ) {
119 $socket = $this->getSocket();
120 if ( $socket === false ) {
123 return array( $socket );
127 * Get the host's IP address.
128 * Does not support IPv6 at present due to the lack of a convenient interface in PHP.
130 protected function getIP() {
131 if ( $this->ip
=== null ) {
132 if ( IP
::isIPv4( $this->host
) ) {
133 $this->ip
= $this->host
;
134 } elseif ( IP
::isIPv6( $this->host
) ) {
135 throw new MWException( '$wgSquidServers does not support IPv6' );
137 wfSuppressWarnings();
138 $this->ip
= gethostbyname( $this->host
);
139 if ( $this->ip
=== $this->host
) {
149 * Close the socket and ignore any future purge requests.
150 * This is called if there is a protocol error.
152 protected function markDown() {
154 $this->socket
= false;
158 * Close the socket but allow it to be reopened for future purge requests
160 public function close() {
161 if ( $this->socket
) {
162 wfSuppressWarnings();
163 socket_set_block( $this->socket
);
164 socket_shutdown( $this->socket
);
165 socket_close( $this->socket
);
168 $this->socket
= null;
169 $this->readBuffer
= '';
170 // Write buffer is kept since it may contain a request for the next socket
174 * Queue a purge operation
178 public function queuePurge( $url ) {
179 global $wgSquidPurgeUseHostHeader;
180 $url = SquidUpdate
::expand( str_replace( "\n", '', $url ) );
182 if ( $wgSquidPurgeUseHostHeader ) {
183 $url = wfParseUrl( $url );
184 $host = $url['host'];
185 if ( isset( $url['port'] ) && strlen( $url['port'] ) > 0 ) {
186 $host .= ":" . $url['port'];
188 $path = $url['path'];
189 if ( isset( $url['query'] ) && is_string( $url['query'] ) ) {
190 $path = wfAppendQuery( $path, $url['query'] );
192 $request[] = "PURGE $path HTTP/1.1";
193 $request[] = "Host: $host";
195 $request[] = "PURGE $url HTTP/1.0";
197 $request[] = "Connection: Keep-Alive";
198 $request[] = "Proxy-Connection: Keep-Alive";
199 $request[] = "User-Agent: " . Http
::userAgent() . ' ' . __CLASS__
;
200 // Two ''s to create \r\n\r\n
204 $this->requests
[] = implode( "\r\n", $request );
205 if ( $this->currentRequestIndex
=== null ) {
206 $this->nextRequest();
213 public function isIdle() {
214 return strlen( $this->writeBuffer
) == 0 && $this->readState
== 'idle';
218 * Perform pending writes. Call this when socket_select() indicates that writing will not block.
220 public function doWrites() {
221 if ( !strlen( $this->writeBuffer
) ) {
224 $socket = $this->getSocket();
229 if ( strlen( $this->writeBuffer
) <= self
::BUFFER_SIZE
) {
230 $buf = $this->writeBuffer
;
233 $buf = substr( $this->writeBuffer
, 0, self
::BUFFER_SIZE
);
236 wfSuppressWarnings();
237 $bytesSent = socket_send( $socket, $buf, strlen( $buf ), $flags );
240 if ( $bytesSent === false ) {
241 $error = socket_last_error( $socket );
242 if ( $error != self
::EAGAIN
&& $error != self
::EINTR
) {
243 $this->log( 'write error: ' . socket_strerror( $error ) );
249 $this->writeBuffer
= substr( $this->writeBuffer
, $bytesSent );
253 * Read some data. Call this when socket_select() indicates that the read buffer is non-empty.
255 public function doReads() {
256 $socket = $this->getSocket();
262 wfSuppressWarnings();
263 $bytesRead = socket_recv( $socket, $buf, self
::BUFFER_SIZE
, 0 );
265 if ( $bytesRead === false ) {
266 $error = socket_last_error( $socket );
267 if ( $error != self
::EAGAIN
&& $error != self
::EINTR
) {
268 $this->log( 'read error: ' . socket_strerror( $error ) );
272 } elseif ( $bytesRead === 0 ) {
278 $this->readBuffer
.= $buf;
279 while ( $this->socket
&& $this->processReadBuffer() === 'continue' );
283 * @throws MWException
286 protected function processReadBuffer() {
287 switch ( $this->readState
) {
292 $lines = explode( "\r\n", $this->readBuffer
, 2 );
293 if ( count( $lines ) < 2 ) {
296 if ( $this->readState
== 'status' ) {
297 $this->processStatusLine( $lines[0] );
299 $this->processHeaderLine( $lines[0] );
301 $this->readBuffer
= $lines[1];
304 if ( $this->bodyRemaining
!== null ) {
305 if ( $this->bodyRemaining
> strlen( $this->readBuffer
) ) {
306 $this->bodyRemaining
-= strlen( $this->readBuffer
);
307 $this->readBuffer
= '';
310 $this->readBuffer
= substr( $this->readBuffer
, $this->bodyRemaining
);
311 $this->bodyRemaining
= 0;
312 $this->nextRequest();
316 // No content length, read all data to EOF
317 $this->readBuffer
= '';
321 throw new MWException( __METHOD__
. ': unexpected state' );
329 protected function processStatusLine( $line ) {
330 if ( !preg_match( '!^HTTP/(\d+)\.(\d+) (\d{3}) (.*)$!', $line, $m ) ) {
331 $this->log( 'invalid status line' );
335 list( , , , $status, $reason ) = $m;
336 $status = intval( $status );
337 if ( $status !== 200 && $status !== 404 ) {
338 $this->log( "unexpected status code: $status $reason" );
342 $this->readState
= 'header';
346 * @param $line string
348 protected function processHeaderLine( $line ) {
349 if ( preg_match( '/^Content-Length: (\d+)$/i', $line, $m ) ) {
350 $this->bodyRemaining
= intval( $m[1] );
351 } elseif ( $line === '' ) {
352 $this->readState
= 'body';
356 protected function nextRequest() {
357 if ( $this->currentRequestIndex
!== null ) {
358 unset( $this->requests
[$this->currentRequestIndex
] );
360 if ( count( $this->requests
) ) {
361 $this->readState
= 'status';
362 $this->currentRequestIndex
= key( $this->requests
);
363 $this->writeBuffer
= $this->requests
[$this->currentRequestIndex
];
365 $this->readState
= 'idle';
366 $this->currentRequestIndex
= null;
367 $this->writeBuffer
= '';
369 $this->bodyRemaining
= null;
375 protected function log( $msg ) {
376 wfDebugLog( 'squid', __CLASS__
. " ($this->host): $msg" );
380 class SquidPurgeClientPool
{
383 * @var array of SquidPurgeClient
385 var $clients = array();
389 * @param $options array
391 function __construct( $options = array() ) {
392 if ( isset( $options['timeout'] ) ) {
393 $this->timeout
= $options['timeout'];
398 * @param $client SquidPurgeClient
401 public function addClient( $client ) {
402 $this->clients
[] = $client;
405 public function run() {
407 $startTime = microtime( true );
409 $readSockets = $writeSockets = array();
411 * @var $client SquidPurgeClient
413 foreach ( $this->clients
as $clientIndex => $client ) {
414 $sockets = $client->getReadSocketsForSelect();
415 foreach ( $sockets as $i => $socket ) {
416 $readSockets["$clientIndex/$i"] = $socket;
418 $sockets = $client->getWriteSocketsForSelect();
419 foreach ( $sockets as $i => $socket ) {
420 $writeSockets["$clientIndex/$i"] = $socket;
423 if ( !count( $readSockets ) && !count( $writeSockets ) ) {
426 $exceptSockets = null;
427 $timeout = min( $startTime +
$this->timeout
- microtime( true ), 1 );
428 wfSuppressWarnings();
429 $numReady = socket_select( $readSockets, $writeSockets, $exceptSockets, $timeout );
431 if ( $numReady === false ) {
432 wfDebugLog( 'squid', __METHOD__
. ': Error in stream_select: ' .
433 socket_strerror( socket_last_error() ) . "\n" );
436 // Check for timeout, use 1% tolerance since we aimed at having socket_select()
437 // exit at precisely the overall timeout
438 if ( microtime( true ) - $startTime > $this->timeout
* 0.99 ) {
439 wfDebugLog( 'squid', __CLASS__
. ": timeout ({$this->timeout}s)\n" );
441 } elseif ( !$numReady ) {
445 foreach ( $readSockets as $key => $socket ) {
446 list( $clientIndex, ) = explode( '/', $key );
447 $client = $this->clients
[$clientIndex];
450 foreach ( $writeSockets as $key => $socket ) {
451 list( $clientIndex, ) = explode( '/', $key );
452 $client = $this->clients
[$clientIndex];
457 foreach ( $this->clients
as $client ) {
458 if ( !$client->isIdle() ) {
463 foreach ( $this->clients
as $client ) {