CdnCacheUpdate.php

Go to the documentation of this file.

<?php
use Wikimedia\Assert\Assert;
use MediaWiki\MediaWikiServices;
 
class CdnCacheUpdate implements DeferrableUpdate, MergeableUpdate {
    protected $urls = [];
 
    public function __construct( array $urlArr ) {
        $this->urls = $urlArr;
    }
 
    public function merge( MergeableUpdate $update ) {
        Assert::parameterType( __CLASS__, $update, '$update' );
 
        $this->urls = array_merge( $this->urls, $update->urls );
    }
 
    public static function newFromTitles( $titles, $urlArr = [] ) {
        ( new LinkBatch( $titles ) )->execute();
        foreach ( $titles as $title ) {
            $urlArr = array_merge( $urlArr, $title->getCdnUrls() );
        }
 
        return new CdnCacheUpdate( $urlArr );
    }
 
    public function doUpdate() {
        global $wgCdnReboundPurgeDelay;
 
        self::purge( $this->urls );
 
        if ( $wgCdnReboundPurgeDelay > 0 ) {
            JobQueueGroup::singleton()->lazyPush( new CdnPurgeJob(
                Title::makeTitle( NS_SPECIAL, 'Badtitle/' . __CLASS__ ),
                [
                    'urls' => $this->urls,
                    'jobReleaseTimestamp' => time() + $wgCdnReboundPurgeDelay
                ]
            ) );
        }
    }
 
    public static function purge( array $urlArr ) {
        global $wgSquidServers, $wgHTCPRouting;
 
        if ( !$urlArr ) {
            return;
        }
 
        // Remove duplicate URLs from list
        $urlArr = array_unique( $urlArr );
 
        wfDebugLog( 'squid', __METHOD__ . ': ' . implode( ' ', $urlArr ) );
 
        // Reliably broadcast the purge to all edge nodes
        $relayer = MediaWikiServices::getInstance()->getEventRelayerGroup()
                    ->getRelayer( 'cdn-url-purges' );
        $ts = microtime( true );
        $relayer->notifyMulti(
            'cdn-url-purges',
            array_map(
                function ( $url ) use ( $ts ) {
                    return [
                        'url' => $url,
                        'timestamp' => $ts,
                    ];
                },
                $urlArr
            )
        );
 
        // Send lossy UDP broadcasting if enabled
        if ( $wgHTCPRouting ) {
            self::HTCPPurge( $urlArr );
        }
 
        // Do direct server purges if enabled (this does not scale very well)
        if ( $wgSquidServers ) {
            // Maximum number of parallel connections per squid
            $maxSocketsPerSquid = 8;
            // Number of requests to send per socket
            // 400 seems to be a good tradeoff, opening a socket takes a while
            $urlsPerSocket = 400;
            $socketsPerSquid = ceil( count( $urlArr ) / $urlsPerSocket );
            if ( $socketsPerSquid > $maxSocketsPerSquid ) {
                $socketsPerSquid = $maxSocketsPerSquid;
            }
 
            $pool = new SquidPurgeClientPool;
            $chunks = array_chunk( $urlArr, ceil( count( $urlArr ) / $socketsPerSquid ) );
            foreach ( $wgSquidServers as $server ) {
                foreach ( $chunks as $chunk ) {
                    $client = new SquidPurgeClient( $server );
                    foreach ( $chunk as $url ) {
                        $client->queuePurge( $url );
                    }
                    $pool->addClient( $client );
                }
            }
 
            $pool->run();
        }
    }
 
    private static function HTCPPurge( array $urlArr ) {
        global $wgHTCPRouting, $wgHTCPMulticastTTL;
 
        // HTCP CLR operation
        $htcpOpCLR = 4;
 
        // @todo FIXME: PHP doesn't support these socket constants (include/linux/in.h)
        if ( !defined( "IPPROTO_IP" ) ) {
            define( "IPPROTO_IP", 0 );
            define( "IP_MULTICAST_LOOP", 34 );
            define( "IP_MULTICAST_TTL", 33 );
        }
 
        // pfsockopen doesn't work because we need set_sock_opt
        $conn = socket_create( AF_INET, SOCK_DGRAM, SOL_UDP );
        if ( !$conn ) {
            $errstr = socket_strerror( socket_last_error() );
            wfDebugLog( 'squid', __METHOD__ .
                ": Error opening UDP socket: $errstr" );
 
            return;
        }
 
        // Set socket options
        socket_set_option( $conn, IPPROTO_IP, IP_MULTICAST_LOOP, 0 );
        if ( $wgHTCPMulticastTTL != 1 ) {
            // Set multicast time to live (hop count) option on socket
            socket_set_option( $conn, IPPROTO_IP, IP_MULTICAST_TTL,
                $wgHTCPMulticastTTL );
        }
 
        // Get sequential trx IDs for packet loss counting
        $ids = UIDGenerator::newSequentialPerNodeIDs(
            'squidhtcppurge', 32, count( $urlArr ), UIDGenerator::QUICK_VOLATILE
        );
 
        foreach ( $urlArr as $url ) {
            if ( !is_string( $url ) ) {
                throw new MWException( 'Bad purge URL' );
            }
            $url = self::expand( $url );
            $conf = self::getRuleForURL( $url, $wgHTCPRouting );
            if ( !$conf ) {
                wfDebugLog( 'squid', __METHOD__ .
                    "No HTCP rule configured for URL {$url} , skipping" );
                continue;
            }
 
            if ( isset( $conf['host'] ) && isset( $conf['port'] ) ) {
                // Normalize single entries
                $conf = [ $conf ];
            }
            foreach ( $conf as $subconf ) {
                if ( !isset( $subconf['host'] ) || !isset( $subconf['port'] ) ) {
                    throw new MWException( "Invalid HTCP rule for URL $url\n" );
                }
            }
 
            // Construct a minimal HTCP request diagram
            // as per RFC 2756
            // Opcode 'CLR', no response desired, no auth
            $htcpTransID = current( $ids );
            next( $ids );
 
            $htcpSpecifier = pack( 'na4na*na8n',
                4, 'HEAD', strlen( $url ), $url,
                8, 'HTTP/1.0', 0 );
 
            $htcpDataLen = 8 + 2 + strlen( $htcpSpecifier );
            $htcpLen = 4 + $htcpDataLen + 2;
 
            // Note! Squid gets the bit order of the first
            // word wrong, wrt the RFC. Apparently no other
            // implementation exists, so adapt to Squid
            $htcpPacket = pack( 'nxxnCxNxxa*n',
                $htcpLen, $htcpDataLen, $htcpOpCLR,
                $htcpTransID, $htcpSpecifier, 2 );
 
            wfDebugLog( 'squid', __METHOD__ .
                "Purging URL $url via HTCP" );
            foreach ( $conf as $subconf ) {
                socket_sendto( $conn, $htcpPacket, $htcpLen, 0,
                    $subconf['host'], $subconf['port'] );
            }
        }
    }
 
    public static function expand( $url ) {
        return wfExpandUrl( $url, PROTO_INTERNAL );
    }
 
    private static function getRuleForURL( $url, $rules ) {
        foreach ( $rules as $regex => $routing ) {
            if ( $regex === '' || preg_match( $regex, $url ) ) {
                return $routing;
            }
        }
 
        return false;
    }
}