includes/libs/objectcache/MemcachedClient.php

   1 <?php
   2 // phpcs:ignoreFile -- It's an external lib and it isn't. Let's not bother.
   3 /**
   4  * Memcached client for PHP.
   5  *
   6  * +---------------------------------------------------------------------------+
   7  * | memcached client, PHP                                                     |
   8  * +---------------------------------------------------------------------------+
   9  * | Copyright (c) 2003 Ryan T. Dean <rtdean@cytherianage.net>                 |
  10  * | All rights reserved.                                                      |
  11  * |                                                                           |
  12  * | Redistribution and use in source and binary forms, with or without        |
  13  * | modification, are permitted provided that the following conditions        |
  14  * | are met:                                                                  |
  15  * |                                                                           |
  16  * | 1. Redistributions of source code must retain the above copyright         |
  17  * |    notice, this list of conditions and the following disclaimer.          |
  18  * | 2. Redistributions in binary form must reproduce the above copyright      |
  19  * |    notice, this list of conditions and the following disclaimer in the    |
  20  * |    documentation and/or other materials provided with the distribution.   |
  21  * |                                                                           |
  22  * | THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR      |
  23  * | IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES |
  24  * | OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.   |
  25  * | IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,          |
  26  * | INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT  |
  27  * | NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
  28  * | DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY     |
  29  * | THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT       |
  30  * | (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF  |
  31  * | THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.         |
  32  * +---------------------------------------------------------------------------+
  33  * | Author: Ryan T. Dean <rtdean@cytherianage.net>                            |
  34  * | Heavily influenced by the Perl memcached client by Brad Fitzpatrick.      |
  35  * |   Permission granted by Brad Fitzpatrick for relicense of ported Perl     |
  36  * |   client logic under 2-clause BSD license.                                |
  37  * +---------------------------------------------------------------------------+
  38  *
  39  * @file
  40  * $TCAnet$
  41  */
  42
  43 /**
  44  * This is a PHP client for memcached - a distributed memory cache daemon.
  45  *
  46  * More information is available at http://www.danga.com/memcached/
  47  *
  48  * Usage example:
  49  *
  50  *     $mc = new MemcachedClient(array(
  51  *         'servers' => array(
  52  *             '127.0.0.1:10000',
  53  *             array( '192.0.0.1:10010', 2 ),
  54  *             '127.0.0.1:10020'
  55  *         ),
  56  *         'debug'   => false,
  57  *         'compress_threshold' => 10240,
  58  *         'persistent' => true
  59  *     ));
  60  *
  61  *     $mc->add( 'key', array( 'some', 'array' ) );
  62  *     $mc->replace( 'key', 'some random string' );
  63  *     $val = $mc->get( 'key' );
  64  *
  65  * @author Ryan T. Dean <rtdean@cytherianage.net>
  66  * @version 0.1.2
  67  */
  68
  69 use Psr\Log\LoggerInterface;
  70 use Psr\Log\NullLogger;
  71
  72 // {{{ class MemcachedClient
  73 /**
  74  * memcached client class implemented using (p)fsockopen()
  75  *
  76  * @author  Ryan T. Dean <rtdean@cytherianage.net>
  77  * @ingroup Cache
  78  */
  79 class MemcachedClient {
  80         // {{{ properties
  81         // {{{ public
  82
  83         // {{{ constants
  84         // {{{ flags
  85
  86         /**
  87          * Flag: indicates data is serialized
  88          */
  89         const SERIALIZED = 1;
  90
  91         /**
  92          * Flag: indicates data is compressed
  93          */
  94         const COMPRESSED = 2;
  95
  96         /**
  97          * Flag: indicates data is an integer
  98          */
  99         const INTVAL = 4;
 100
 101         // }}}
 102
 103         /**
 104          * Minimum savings to store data compressed
 105          */
 106         const COMPRESSION_SAVINGS = 0.20;
 107
 108         // }}}
 109
 110         /**
 111          * Command statistics
 112          *
 113          * @var array
 114          * @access public
 115          */
 116         public $stats;
 117
 118         // }}}
 119         // {{{ private
 120
 121         /**
 122          * Cached Sockets that are connected
 123          *
 124          * @var array
 125          * @access private
 126          */
 127         public $_cache_sock;
 128
 129         /**
 130          * Current debug status; 0 - none to 9 - profiling
 131          *
 132          * @var bool
 133          * @access private
 134          */
 135         public $_debug;
 136
 137         /**
 138          * Dead hosts, assoc array, 'host'=>'unixtime when ok to check again'
 139          *
 140          * @var array
 141          * @access private
 142          */
 143         public $_host_dead;
 144
 145         /**
 146          * Is compression available?
 147          *
 148          * @var bool
 149          * @access private
 150          */
 151         public $_have_zlib;
 152
 153         /**
 154          * Do we want to use compression?
 155          *
 156          * @var bool
 157          * @access private
 158          */
 159         public $_compress_enable;
 160
 161         /**
 162          * At how many bytes should we compress?
 163          *
 164          * @var int
 165          * @access private
 166          */
 167         public $_compress_threshold;
 168
 169         /**
 170          * Are we using persistent links?
 171          *
 172          * @var bool
 173          * @access private
 174          */
 175         public $_persistent;
 176
 177         /**
 178          * If only using one server; contains ip:port to connect to
 179          *
 180          * @var string
 181          * @access private
 182          */
 183         public $_single_sock;
 184
 185         /**
 186          * Array containing ip:port or array(ip:port, weight)
 187          *
 188          * @var array
 189          * @access private
 190          */
 191         public $_servers;
 192
 193         /**
 194          * Our bit buckets
 195          *
 196          * @var array
 197          * @access private
 198          */
 199         public $_buckets;
 200
 201         /**
 202          * Total # of bit buckets we have
 203          *
 204          * @var int
 205          * @access private
 206          */
 207         public $_bucketcount;
 208
 209         /**
 210          * # of total servers we have
 211          *
 212          * @var int
 213          * @access private
 214          */
 215         public $_active;
 216
 217         /**
 218          * Stream timeout in seconds. Applies for example to fread()
 219          *
 220          * @var int
 221          * @access private
 222          */
 223         public $_timeout_seconds;
 224
 225         /**
 226          * Stream timeout in microseconds
 227          *
 228          * @var int
 229          * @access private
 230          */
 231         public $_timeout_microseconds;
 232
 233         /**
 234          * Connect timeout in seconds
 235          */
 236         public $_connect_timeout;
 237
 238         /**
 239          * Number of connection attempts for each server
 240          */
 241         public $_connect_attempts;
 242
 243         /**
 244          * @var LoggerInterface
 245          */
 246         private $_logger;
 247
 248         // }}}
 249         // }}}
 250         // {{{ methods
 251         // {{{ public functions
 252         // {{{ memcached()
 253
 254         /**
 255          * Memcache initializer
 256          *
 257          * @param array $args Associative array of settings
 258          */
 259         public function __construct( $args ) {
 260                 $this->set_servers( $args['servers'] ?? array() );
 261                 $this->_debug = $args['debug'] ?? false;
 262                 $this->stats = array();
 263                 $this->_compress_threshold = $args['compress_threshold'] ?? 0;
 264                 $this->_persistent = $args['persistent'] ?? false;
 265                 $this->_compress_enable = true;
 266                 $this->_have_zlib = function_exists( 'gzcompress' );
 267
 268                 $this->_cache_sock = array();
 269                 $this->_host_dead = array();
 270
 271                 $this->_timeout_seconds = 0;
 272                 $this->_timeout_microseconds = $args['timeout'] ?? 500000;
 273
 274                 $this->_connect_timeout = $args['connect_timeout'] ?? 0.1;
 275                 $this->_connect_attempts = 2;
 276
 277                 $this->_logger = $args['logger'] ?? new NullLogger();
 278         }
 279
 280         // }}}
 281         // {{{ add()
 282
 283         /**
 284          * Adds a key/value to the memcache server if one isn't already set with
 285          * that key
 286          *
 287          * @param string $key Key to set with data
 288          * @param mixed $val Value to store
 289          * @param int $exp (optional) Expiration time. This can be a number of seconds
 290          * to cache for (up to 30 days inclusive).  Any timespans of 30 days + 1 second or
 291          * longer must be the timestamp of the time at which the mapping should expire. It
 292          * is safe to use timestamps in all cases, regardless of expiration
 293          * eg: strtotime("+3 hour")
 294          *
 295          * @return bool
 296          */
 297         public function add( $key, $val, $exp = 0 ) {
 298                 return $this->_set( 'add', $key, $val, $exp );
 299         }
 300
 301         // }}}
 302         // {{{ decr()
 303
 304         /**
 305          * Decrease a value stored on the memcache server
 306          *
 307          * @param string $key Key to decrease
 308          * @param int $amt (optional) amount to decrease
 309          *
 310          * @return mixed False on failure, value on success
 311          */
 312         public function decr( $key, $amt = 1 ) {
 313                 return $this->_incrdecr( 'decr', $key, $amt );
 314         }
 315
 316         // }}}
 317         // {{{ delete()
 318
 319         /**
 320          * Deletes a key from the server, optionally after $time
 321          *
 322          * @param string $key Key to delete
 323          * @param int $time (optional) how long to wait before deleting
 324          *
 325          * @return bool True on success, false on failure
 326          */
 327         public function delete( $key, $time = 0 ) {
 328                 if ( !$this->_active ) {
 329                         return false;
 330                 }
 331
 332                 $sock = $this->get_sock( $key );
 333                 if ( !is_resource( $sock ) ) {
 334                         return false;
 335                 }
 336
 337                 $key = is_array( $key ) ? $key[1] : $key;
 338
 339                 if ( isset( $this->stats['delete'] ) ) {
 340                         $this->stats['delete']++;
 341                 } else {
 342                         $this->stats['delete'] = 1;
 343                 }
 344                 $cmd = "delete $key $time\r\n";
 345                 if ( !$this->_fwrite( $sock, $cmd ) ) {
 346                         return false;
 347                 }
 348                 $res = $this->_fgets( $sock );
 349
 350                 if ( $this->_debug ) {
 351                         $this->_debugprint( sprintf( "MemCache: delete %s (%s)", $key, $res ) );
 352                 }
 353
 354                 if ( $res == "DELETED" || $res == "NOT_FOUND" ) {
 355                         return true;
 356                 }
 357
 358                 return false;
 359         }
 360
 361         /**
 362          * Changes the TTL on a key from the server to $time
 363          *
 364          * @param string $key
 365          * @param int $time TTL in seconds
 366          *
 367          * @return bool True on success, false on failure
 368          */
 369         public function touch( $key, $time = 0 ) {
 370                 if ( !$this->_active ) {
 371                         return false;
 372                 }
 373
 374                 $sock = $this->get_sock( $key );
 375                 if ( !is_resource( $sock ) ) {
 376                         return false;
 377                 }
 378
 379                 $key = is_array( $key ) ? $key[1] : $key;
 380
 381                 if ( isset( $this->stats['touch'] ) ) {
 382                         $this->stats['touch']++;
 383                 } else {
 384                         $this->stats['touch'] = 1;
 385                 }
 386                 $cmd = "touch $key $time\r\n";
 387                 if ( !$this->_fwrite( $sock, $cmd ) ) {
 388                         return false;
 389                 }
 390                 $res = $this->_fgets( $sock );
 391
 392                 if ( $this->_debug ) {
 393                         $this->_debugprint( sprintf( "MemCache: touch %s (%s)", $key, $res ) );
 394                 }
 395
 396                 if ( $res == "TOUCHED" ) {
 397                         return true;
 398                 }
 399
 400                 return false;
 401         }
 402
 403         /**
 404          * @param string $key
 405          * @param int $timeout
 406          * @return bool
 407          */
 408         public function lock( $key, $timeout = 0 ) {
 409                 /* stub */
 410                 return true;
 411         }
 412
 413         /**
 414          * @param string $key
 415          * @return bool
 416          */
 417         public function unlock( $key ) {
 418                 /* stub */
 419                 return true;
 420         }
 421
 422         // }}}
 423         // {{{ disconnect_all()
 424
 425         /**
 426          * Disconnects all connected sockets
 427          */
 428         public function disconnect_all() {
 429                 foreach ( $this->_cache_sock as $sock ) {
 430                         fclose( $sock );
 431                 }
 432
 433                 $this->_cache_sock = array();
 434         }
 435
 436         // }}}
 437         // {{{ enable_compress()
 438
 439         /**
 440          * Enable / Disable compression
 441          *
 442          * @param bool $enable True to enable, false to disable
 443          */
 444         public function enable_compress( $enable ) {
 445                 $this->_compress_enable = $enable;
 446         }
 447
 448         // }}}
 449         // {{{ forget_dead_hosts()
 450
 451         /**
 452          * Forget about all of the dead hosts
 453          */
 454         public function forget_dead_hosts() {
 455                 $this->_host_dead = array();
 456         }
 457
 458         // }}}
 459         // {{{ get()
 460
 461         /**
 462          * Retrieves the value associated with the key from the memcache server
 463          *
 464          * @param array|string $key key to retrieve
 465          * @param float $casToken [optional]
 466          *
 467          * @return mixed
 468          */
 469         public function get( $key, &$casToken = null ) {
 470                 if ( $this->_debug ) {
 471                         $this->_debugprint( "get($key)" );
 472                 }
 473
 474                 if ( !is_array( $key ) && strval( $key ) === '' ) {
 475                         $this->_debugprint( "Skipping key which equals to an empty string" );
 476                         return false;
 477                 }
 478
 479                 if ( !$this->_active ) {
 480                         return false;
 481                 }
 482
 483                 $sock = $this->get_sock( $key );
 484
 485                 if ( !is_resource( $sock ) ) {
 486                         return false;
 487                 }
 488
 489                 $key = is_array( $key ) ? $key[1] : $key;
 490                 if ( isset( $this->stats['get'] ) ) {
 491                         $this->stats['get']++;
 492                 } else {
 493                         $this->stats['get'] = 1;
 494                 }
 495
 496                 $cmd = "gets $key\r\n";
 497                 if ( !$this->_fwrite( $sock, $cmd ) ) {
 498                         return false;
 499                 }
 500
 501                 $val = array();
 502                 $this->_load_items( $sock, $val, $casToken );
 503
 504                 if ( $this->_debug ) {
 505                         foreach ( $val as $k => $v ) {
 506                                 $this->_debugprint( sprintf( "MemCache: sock %s got %s", serialize( $sock ), $k ) );
 507                         }
 508                 }
 509
 510                 $value = false;
 511                 if ( isset( $val[$key] ) ) {
 512                         $value = $val[$key];
 513                 }
 514                 return $value;
 515         }
 516
 517         // }}}
 518         // {{{ get_multi()
 519
 520         /**
 521          * Get multiple keys from the server(s)
 522          *
 523          * @param array $keys Keys to retrieve
 524          *
 525          * @return array
 526          */
 527         public function get_multi( $keys ) {
 528                 if ( !$this->_active ) {
 529                         return array();
 530                 }
 531
 532                 if ( isset( $this->stats['get_multi'] ) ) {
 533                         $this->stats['get_multi']++;
 534                 } else {
 535                         $this->stats['get_multi'] = 1;
 536                 }
 537                 $sock_keys = array();
 538                 $socks = array();
 539                 foreach ( $keys as $key ) {
 540                         $sock = $this->get_sock( $key );
 541                         if ( !is_resource( $sock ) ) {
 542                                 continue;
 543                         }
 544                         $key = is_array( $key ) ? $key[1] : $key;
 545                         if ( !isset( $sock_keys[$sock] ) ) {
 546                                 $sock_keys[intval( $sock )] = array();
 547                                 $socks[] = $sock;
 548                         }
 549                         $sock_keys[intval( $sock )][] = $key;
 550                 }
 551
 552                 $gather = array();
 553                 // Send out the requests
 554                 foreach ( $socks as $sock ) {
 555                         $cmd = 'gets';
 556                         foreach ( $sock_keys[intval( $sock )] as $key ) {
 557                                 $cmd .= ' ' . $key;
 558                         }
 559                         $cmd .= "\r\n";
 560
 561                         if ( $this->_fwrite( $sock, $cmd ) ) {
 562                                 $gather[] = $sock;
 563                         }
 564                 }
 565
 566                 // Parse responses
 567                 $val = array();
 568                 foreach ( $gather as $sock ) {
 569                         $this->_load_items( $sock, $val, $casToken );
 570                 }
 571
 572                 if ( $this->_debug ) {
 573                         foreach ( $val as $k => $v ) {
 574                                 $this->_debugprint( sprintf( "MemCache: got %s", $k ) );
 575                         }
 576                 }
 577
 578                 return $val;
 579         }
 580
 581         // }}}
 582         // {{{ incr()
 583
 584         /**
 585          * Increments $key (optionally) by $amt
 586          *
 587          * @param string $key Key to increment
 588          * @param int $amt (optional) amount to increment
 589          *
 590          * @return int|null Null if the key does not exist yet (this does NOT
 591          * create new mappings if the key does not exist). If the key does
 592          * exist, this returns the new value for that key.
 593          */
 594         public function incr( $key, $amt = 1 ) {
 595                 return $this->_incrdecr( 'incr', $key, $amt );
 596         }
 597
 598         // }}}
 599         // {{{ replace()
 600
 601         /**
 602          * Overwrites an existing value for key; only works if key is already set
 603          *
 604          * @param string $key Key to set value as
 605          * @param mixed $value Value to store
 606          * @param int $exp (optional) Expiration time. This can be a number of seconds
 607          * to cache for (up to 30 days inclusive).  Any timespans of 30 days + 1 second or
 608          * longer must be the timestamp of the time at which the mapping should expire. It
 609          * is safe to use timestamps in all cases, regardless of exipration
 610          * eg: strtotime("+3 hour")
 611          *
 612          * @return bool
 613          */
 614         public function replace( $key, $value, $exp = 0 ) {
 615                 return $this->_set( 'replace', $key, $value, $exp );
 616         }
 617
 618         // }}}
 619         // {{{ run_command()
 620
 621         /**
 622          * Passes through $cmd to the memcache server connected by $sock; returns
 623          * output as an array (null array if no output)
 624          *
 625          * @param Resource $sock Socket to send command on
 626          * @param string $cmd Command to run
 627          *
 628          * @return array Output array
 629          */
 630         public function run_command( $sock, $cmd ) {
 631                 if ( !is_resource( $sock ) ) {
 632                         return array();
 633                 }
 634
 635                 if ( !$this->_fwrite( $sock, $cmd ) ) {
 636                         return array();
 637                 }
 638
 639                 $ret = array();
 640                 while ( true ) {
 641                         $res = $this->_fgets( $sock );
 642                         $ret[] = $res;
 643                         if ( preg_match( '/^END/', $res ) ) {
 644                                 break;
 645                         }
 646                         if ( strlen( $res ) == 0 ) {
 647                                 break;
 648                         }
 649                 }
 650                 return $ret;
 651         }
 652
 653         // }}}
 654         // {{{ set()
 655
 656         /**
 657          * Unconditionally sets a key to a given value in the memcache.  Returns true
 658          * if set successfully.
 659          *
 660          * @param string $key Key to set value as
 661          * @param mixed $value Value to set
 662          * @param int $exp (optional) Expiration time. This can be a number of seconds
 663          * to cache for (up to 30 days inclusive).  Any timespans of 30 days + 1 second or
 664          * longer must be the timestamp of the time at which the mapping should expire. It
 665          * is safe to use timestamps in all cases, regardless of exipration
 666          * eg: strtotime("+3 hour")
 667          *
 668          * @return bool True on success
 669          */
 670         public function set( $key, $value, $exp = 0 ) {
 671                 return $this->_set( 'set', $key, $value, $exp );
 672         }
 673
 674         // }}}
 675         // {{{ cas()
 676
 677         /**
 678          * Sets a key to a given value in the memcache if the current value still corresponds
 679          * to a known, given value.  Returns true if set successfully.
 680          *
 681          * @param float $casToken Current known value
 682          * @param string $key Key to set value as
 683          * @param mixed $value Value to set
 684          * @param int $exp (optional) Expiration time. This can be a number of seconds
 685          * to cache for (up to 30 days inclusive).  Any timespans of 30 days + 1 second or
 686          * longer must be the timestamp of the time at which the mapping should expire. It
 687          * is safe to use timestamps in all cases, regardless of exipration
 688          * eg: strtotime("+3 hour")
 689          *
 690          * @return bool True on success
 691          */
 692         public function cas( $casToken, $key, $value, $exp = 0 ) {
 693                 return $this->_set( 'cas', $key, $value, $exp, $casToken );
 694         }
 695
 696         // }}}
 697         // {{{ set_compress_threshold()
 698
 699         /**
 700          * Set the compression threshold
 701          *
 702          * @param int $thresh Threshold to compress if larger than
 703          */
 704         public function set_compress_threshold( $thresh ) {
 705                 $this->_compress_threshold = $thresh;
 706         }
 707
 708         // }}}
 709         // {{{ set_debug()
 710
 711         /**
 712          * Set the debug flag
 713          *
 714          * @see __construct()
 715          * @param bool $dbg True for debugging, false otherwise
 716          */
 717         public function set_debug( $dbg ) {
 718                 $this->_debug = $dbg;
 719         }
 720
 721         // }}}
 722         // {{{ set_servers()
 723
 724         /**
 725          * Set the server list to distribute key gets and puts between
 726          *
 727          * @see __construct()
 728          * @param array $list Array of servers to connect to
 729          */
 730         public function set_servers( $list ) {
 731                 $this->_servers = $list;
 732                 $this->_active = count( $list );
 733                 $this->_buckets = null;
 734                 $this->_bucketcount = 0;
 735
 736                 $this->_single_sock = null;
 737                 if ( $this->_active == 1 ) {
 738                         $this->_single_sock = $this->_servers[0];
 739                 }
 740         }
 741
 742         /**
 743          * Sets the timeout for new connections
 744          *
 745          * @param int $seconds Number of seconds
 746          * @param int $microseconds Number of microseconds
 747          */
 748         public function set_timeout( $seconds, $microseconds ) {
 749                 $this->_timeout_seconds = $seconds;
 750                 $this->_timeout_microseconds = $microseconds;
 751         }
 752
 753         // }}}
 754         // }}}
 755         // {{{ private methods
 756         // {{{ _close_sock()
 757
 758         /**
 759          * Close the specified socket
 760          *
 761          * @param string $sock Socket to close
 762          *
 763          * @access private
 764          */
 765         function _close_sock( $sock ) {
 766                 $host = array_search( $sock, $this->_cache_sock );
 767                 fclose( $this->_cache_sock[$host] );
 768                 unset( $this->_cache_sock[$host] );
 769         }
 770
 771         // }}}
 772         // {{{ _connect_sock()
 773
 774         /**
 775          * Connects $sock to $host, timing out after $timeout
 776          *
 777          * @param int $sock Socket to connect
 778          * @param string $host Host:IP to connect to
 779          *
 780          * @return bool
 781          * @access private
 782          */
 783         function _connect_sock( &$sock, $host ) {
 784                 list( $ip, $port ) = preg_split( '/:(?=\d)/', $host );
 785                 $sock = false;
 786                 $timeout = $this->_connect_timeout;
 787                 $errno = $errstr = null;
 788                 for ( $i = 0; !$sock && $i < $this->_connect_attempts; $i++ ) {
 789                         Wikimedia\suppressWarnings();
 790                         if ( $this->_persistent == 1 ) {
 791                                 $sock = pfsockopen( $ip, $port, $errno, $errstr, $timeout );
 792                         } else {
 793                                 $sock = fsockopen( $ip, $port, $errno, $errstr, $timeout );
 794                         }
 795                         Wikimedia\restoreWarnings();
 796                 }
 797                 if ( !$sock ) {
 798                         $this->_error_log( "Error connecting to $host: $errstr" );
 799                         $this->_dead_host( $host );
 800                         return false;
 801                 }
 802
 803                 // Initialise timeout
 804                 stream_set_timeout( $sock, $this->_timeout_seconds, $this->_timeout_microseconds );
 805
 806                 // If the connection was persistent, flush the read buffer in case there
 807                 // was a previous incomplete request on this connection
 808                 if ( $this->_persistent ) {
 809                         $this->_flush_read_buffer( $sock );
 810                 }
 811                 return true;
 812         }
 813
 814         // }}}
 815         // {{{ _dead_sock()
 816
 817         /**
 818          * Marks a host as dead until 30-40 seconds in the future
 819          *
 820          * @param string $sock Socket to mark as dead
 821          *
 822          * @access private
 823          */
 824         function _dead_sock( $sock ) {
 825                 $host = array_search( $sock, $this->_cache_sock );
 826                 $this->_dead_host( $host );
 827         }
 828
 829         /**
 830          * @param string $host
 831          */
 832         function _dead_host( $host ) {
 833                 $ip = explode( ':', $host )[0];
 834                 $this->_host_dead[$ip] = time() + 30 + intval( rand( 0, 10 ) );
 835                 $this->_host_dead[$host] = $this->_host_dead[$ip];
 836                 unset( $this->_cache_sock[$host] );
 837         }
 838
 839         // }}}
 840         // {{{ get_sock()
 841
 842         /**
 843          * get_sock
 844          *
 845          * @param string $key Key to retrieve value for;
 846          *
 847          * @return Resource|bool Resource on success, false on failure
 848          * @access private
 849          */
 850         function get_sock( $key ) {
 851                 if ( !$this->_active ) {
 852                         return false;
 853                 }
 854
 855                 if ( $this->_single_sock !== null ) {
 856                         return $this->sock_to_host( $this->_single_sock );
 857                 }
 858
 859                 $hv = is_array( $key ) ? intval( $key[0] ) : $this->_hashfunc( $key );
 860                 if ( $this->_buckets === null ) {
 861                         $bu = array();
 862                         foreach ( $this->_servers as $v ) {
 863                                 if ( is_array( $v ) ) {
 864                                         for ( $i = 0; $i < $v[1]; $i++ ) {
 865                                                 $bu[] = $v[0];
 866                                         }
 867                                 } else {
 868                                         $bu[] = $v;
 869                                 }
 870                         }
 871                         $this->_buckets = $bu;
 872                         $this->_bucketcount = count( $bu );
 873                 }
 874
 875                 $realkey = is_array( $key ) ? $key[1] : $key;
 876                 for ( $tries = 0; $tries < 20; $tries++ ) {
 877                         $host = $this->_buckets[$hv % $this->_bucketcount];
 878                         $sock = $this->sock_to_host( $host );
 879                         if ( is_resource( $sock ) ) {
 880                                 return $sock;
 881                         }
 882                         $hv = $this->_hashfunc( $hv . $realkey );
 883                 }
 884
 885                 return false;
 886         }
 887
 888         // }}}
 889         // {{{ _hashfunc()
 890
 891         /**
 892          * Creates a hash integer based on the $key
 893          *
 894          * @param string $key Key to hash
 895          *
 896          * @return int Hash value
 897          * @access private
 898          */
 899         function _hashfunc( $key ) {
 900                 # Hash function must be in [0,0x7ffffff]
 901                 # We take the first 31 bits of the MD5 hash, which unlike the hash
 902                 # function used in a previous version of this client, works
 903                 return hexdec( substr( md5( $key ), 0, 8 ) ) & 0x7fffffff;
 904         }
 905
 906         // }}}
 907         // {{{ _incrdecr()
 908
 909         /**
 910          * Perform increment/decriment on $key
 911          *
 912          * @param string $cmd Command to perform
 913          * @param string|array $key Key to perform it on
 914          * @param int $amt Amount to adjust
 915          *
 916          * @return int New value of $key
 917          * @access private
 918          */
 919         function _incrdecr( $cmd, $key, $amt = 1 ) {
 920                 if ( !$this->_active ) {
 921                         return null;
 922                 }
 923
 924                 $sock = $this->get_sock( $key );
 925                 if ( !is_resource( $sock ) ) {
 926                         return null;
 927                 }
 928
 929                 $key = is_array( $key ) ? $key[1] : $key;
 930                 if ( isset( $this->stats[$cmd] ) ) {
 931                         $this->stats[$cmd]++;
 932                 } else {
 933                         $this->stats[$cmd] = 1;
 934                 }
 935                 if ( !$this->_fwrite( $sock, "$cmd $key $amt\r\n" ) ) {
 936                         return null;
 937                 }
 938
 939                 $line = $this->_fgets( $sock );
 940                 $match = array();
 941                 if ( !preg_match( '/^(\d+)/', $line, $match ) ) {
 942                         return null;
 943                 }
 944                 return $match[1];
 945         }
 946
 947         // }}}
 948         // {{{ _load_items()
 949
 950         /**
 951          * Load items into $ret from $sock
 952          *
 953          * @param Resource $sock Socket to read from
 954          * @param array $ret returned values
 955          * @param float $casToken [optional]
 956          * @return bool True for success, false for failure
 957          *
 958          * @access private
 959          */
 960         function _load_items( $sock, &$ret, &$casToken = null ) {
 961                 $results = array();
 962
 963                 while ( 1 ) {
 964                         $decl = $this->_fgets( $sock );
 965
 966                         if ( $decl === false ) {
 967                                 /*
 968                                  * If nothing can be read, something is wrong because we know exactly when
 969                                  * to stop reading (right after "END") and we return right after that.
 970                                  */
 971                                 return false;
 972                         } elseif ( preg_match( '/^VALUE (\S+) (\d+) (\d+) (\d+)$/', $decl, $match ) ) {
 973                                 /*
 974                                  * Read all data returned. This can be either one or multiple values.
 975                                  * Save all that data (in an array) to be processed later: we'll first
 976                                  * want to continue reading until "END" before doing anything else,
 977                                  * to make sure that we don't leave our client in a state where it's
 978                                  * output is not yet fully read.
 979                                  */
 980                                 $results[] = array(
 981                                         $match[1], // rkey
 982                                         $match[2], // flags
 983                                         $match[3], // len
 984                                         $match[4], // casToken
 985                                         $this->_fread( $sock, $match[3] + 2 ), // data
 986                                 );
 987                         } elseif ( $decl == "END" ) {
 988                                 if ( count( $results ) == 0 ) {
 989                                         return false;
 990                                 }
 991
 992                                 /**
 993                                  * All data has been read, time to process the data and build
 994                                  * meaningful return values.
 995                                  */
 996                                 foreach ( $results as $vars ) {
 997                                         list( $rkey, $flags, $len, $casToken, $data ) = $vars;
 998
 999                                         if ( $data === false || substr( $data, -2 ) !== "\r\n" ) {
1000                                                 $this->_handle_error( $sock,
1001                                                         'line ending missing from data block from $1' );
1002                                                 return false;
1003                                         }
1004                                         $data = substr( $data, 0, -2 );
1005                                         $ret[$rkey] = $data;
1006
1007                                         if ( $this->_have_zlib && $flags & self::COMPRESSED ) {
1008                                                 $ret[$rkey] = gzuncompress( $ret[$rkey] );
1009                                         }
1010
1011                                         /*
1012                                          * This unserialize is the exact reason that we only want to
1013                                          * process data after having read until "END" (instead of doing
1014                                          * this right away): "unserialize" can trigger outside code:
1015                                          * in the event that $ret[$rkey] is a serialized object,
1016                                          * unserializing it will trigger __wakeup() if present. If that
1017                                          * function attempted to read from memcached (while we did not
1018                                          * yet read "END"), these 2 calls would collide.
1019                                          */
1020                                         if ( $flags & self::SERIALIZED ) {
1021                                                 $ret[$rkey] = unserialize( $ret[$rkey] );
1022                                         } elseif ( $flags & self::INTVAL ) {
1023                                                 $ret[$rkey] = intval( $ret[$rkey] );
1024                                         }
1025                                 }
1026
1027                                 return true;
1028                         } else {
1029                                 $this->_handle_error( $sock, 'Error parsing response from $1' );
1030                                 return false;
1031                         }
1032                 }
1033         }
1034
1035         // }}}
1036         // {{{ _set()
1037
1038         /**
1039          * Performs the requested storage operation to the memcache server
1040          *
1041          * @param string $cmd Command to perform
1042          * @param string $key Key to act on
1043          * @param mixed $val What we need to store
1044          * @param int $exp (optional) Expiration time. This can be a number of seconds
1045          * to cache for (up to 30 days inclusive).  Any timespans of 30 days + 1 second or
1046          * longer must be the timestamp of the time at which the mapping should expire. It
1047          * is safe to use timestamps in all cases, regardless of exipration
1048          * eg: strtotime("+3 hour")
1049          * @param float $casToken [optional]
1050          *
1051          * @return bool
1052          * @access private
1053          */
1054         function _set( $cmd, $key, $val, $exp, $casToken = null ) {
1055                 if ( !$this->_active ) {
1056                         return false;
1057                 }
1058
1059                 $sock = $this->get_sock( $key );
1060                 if ( !is_resource( $sock ) ) {
1061                         return false;
1062                 }
1063
1064                 if ( isset( $this->stats[$cmd] ) ) {
1065                         $this->stats[$cmd]++;
1066                 } else {
1067                         $this->stats[$cmd] = 1;
1068                 }
1069
1070                 $flags = 0;
1071
1072                 if ( is_int( $val ) ) {
1073                         $flags |= self::INTVAL;
1074                 } elseif ( !is_scalar( $val ) ) {
1075                         $val = serialize( $val );
1076                         $flags |= self::SERIALIZED;
1077                         if ( $this->_debug ) {
1078                                 $this->_debugprint( sprintf( "client: serializing data as it is not scalar" ) );
1079                         }
1080                 }
1081
1082                 $len = strlen( $val );
1083
1084                 if ( $this->_have_zlib && $this->_compress_enable
1085                         && $this->_compress_threshold && $len >= $this->_compress_threshold
1086                 ) {
1087                         $c_val = gzcompress( $val, 9 );
1088                         $c_len = strlen( $c_val );
1089
1090                         if ( $c_len < $len * ( 1 - self::COMPRESSION_SAVINGS ) ) {
1091                                 if ( $this->_debug ) {
1092                                         $this->_debugprint( sprintf( "client: compressing data; was %d bytes is now %d bytes", $len, $c_len ) );
1093                                 }
1094                                 $val = $c_val;
1095                                 $len = $c_len;
1096                                 $flags |= self::COMPRESSED;
1097                         }
1098                 }
1099
1100                 $command = "$cmd $key $flags $exp $len";
1101                 if ( $casToken ) {
1102                         $command .= " $casToken";
1103                 }
1104
1105                 if ( !$this->_fwrite( $sock, "$command\r\n$val\r\n" ) ) {
1106                         return false;
1107                 }
1108
1109                 $line = $this->_fgets( $sock );
1110
1111                 if ( $this->_debug ) {
1112                         $this->_debugprint( sprintf( "%s %s (%s)", $cmd, $key, $line ) );
1113                 }
1114                 if ( $line === "STORED" ) {
1115                         return true;
1116                 } elseif ( $line === "NOT_STORED" && $cmd === "set" ) {
1117                         // "Not stored" is always used as the mcrouter response with AllAsyncRoute
1118                         return true;
1119                 }
1120
1121                 return false;
1122         }
1123
1124         // }}}
1125         // {{{ sock_to_host()
1126
1127         /**
1128          * Returns the socket for the host
1129          *
1130          * @param string $host Host:IP to get socket for
1131          *
1132          * @return Resource|bool IO Stream or false
1133          * @access private
1134          */
1135         function sock_to_host( $host ) {
1136                 if ( isset( $this->_cache_sock[$host] ) ) {
1137                         return $this->_cache_sock[$host];
1138                 }
1139
1140                 $sock = null;
1141                 $now = time();
1142                 list( $ip, /* $port */) = explode( ':', $host );
1143                 if ( isset( $this->_host_dead[$host] ) && $this->_host_dead[$host] > $now ||
1144                         isset( $this->_host_dead[$ip] ) && $this->_host_dead[$ip] > $now
1145                 ) {
1146                         return null;
1147                 }
1148
1149                 if ( !$this->_connect_sock( $sock, $host ) ) {
1150                         return null;
1151                 }
1152
1153                 // Do not buffer writes
1154                 stream_set_write_buffer( $sock, 0 );
1155
1156                 $this->_cache_sock[$host] = $sock;
1157
1158                 return $this->_cache_sock[$host];
1159         }
1160
1161         /**
1162          * @param string $text
1163          */
1164         function _debugprint( $text ) {
1165                 $this->_logger->debug( $text );
1166         }
1167
1168         /**
1169          * @param string $text
1170          */
1171         function _error_log( $text ) {
1172                 $this->_logger->error( "Memcached error: $text" );
1173         }
1174
1175         /**
1176          * Write to a stream. If there is an error, mark the socket dead.
1177          *
1178          * @param Resource $sock The socket
1179          * @param string $buf The string to write
1180          * @return bool True on success, false on failure
1181          */
1182         function _fwrite( $sock, $buf ) {
1183                 $bytesWritten = 0;
1184                 $bufSize = strlen( $buf );
1185                 while ( $bytesWritten < $bufSize ) {
1186                         $result = fwrite( $sock, $buf );
1187                         $data = stream_get_meta_data( $sock );
1188                         if ( $data['timed_out'] ) {
1189                                 $this->_handle_error( $sock, 'timeout writing to $1' );
1190                                 return false;
1191                         }
1192                         // Contrary to the documentation, fwrite() returns zero on error in PHP 5.3.
1193                         if ( $result === false || $result === 0 ) {
1194                                 $this->_handle_error( $sock, 'error writing to $1' );
1195                                 return false;
1196                         }
1197                         $bytesWritten += $result;
1198                 }
1199
1200                 return true;
1201         }
1202
1203         /**
1204          * Handle an I/O error. Mark the socket dead and log an error.
1205          *
1206          * @param Resource $sock
1207          * @param string $msg
1208          */
1209         function _handle_error( $sock, $msg ) {
1210                 $peer = stream_socket_get_name( $sock, true /** remote **/ );
1211                 if ( strval( $peer ) === '' ) {
1212                         $peer = array_search( $sock, $this->_cache_sock );
1213                         if ( $peer === false ) {
1214                                 $peer = '[unknown host]';
1215                         }
1216                 }
1217                 $msg = str_replace( '$1', $peer, $msg );
1218                 $this->_error_log( "$msg" );
1219                 $this->_dead_sock( $sock );
1220         }
1221
1222         /**
1223          * Read the specified number of bytes from a stream. If there is an error,
1224          * mark the socket dead.
1225          *
1226          * @param Resource $sock The socket
1227          * @param int $len The number of bytes to read
1228          * @return string|bool The string on success, false on failure.
1229          */
1230         function _fread( $sock, $len ) {
1231                 $buf = '';
1232                 while ( $len > 0 ) {
1233                         $result = fread( $sock, $len );
1234                         $data = stream_get_meta_data( $sock );
1235                         if ( $data['timed_out'] ) {
1236                                 $this->_handle_error( $sock, 'timeout reading from $1' );
1237                                 return false;
1238                         }
1239                         if ( $result === false ) {
1240                                 $this->_handle_error( $sock, 'error reading buffer from $1' );
1241                                 return false;
1242                         }
1243                         if ( $result === '' ) {
1244                                 // This will happen if the remote end of the socket is shut down
1245                                 $this->_handle_error( $sock, 'unexpected end of file reading from $1' );
1246                                 return false;
1247                         }
1248                         $len -= strlen( $result );
1249                         $buf .= $result;
1250                 }
1251                 return $buf;
1252         }
1253
1254         /**
1255          * Read a line from a stream. If there is an error, mark the socket dead.
1256          * The \r\n line ending is stripped from the response.
1257          *
1258          * @param Resource $sock The socket
1259          * @return string|bool The string on success, false on failure
1260          */
1261         function _fgets( $sock ) {
1262                 $result = fgets( $sock );
1263                 // fgets() may return a partial line if there is a select timeout after
1264                 // a successful recv(), so we have to check for a timeout even if we
1265                 // got a string response.
1266                 $data = stream_get_meta_data( $sock );
1267                 if ( $data['timed_out'] ) {
1268                         $this->_handle_error( $sock, 'timeout reading line from $1' );
1269                         return false;
1270                 }
1271                 if ( $result === false ) {
1272                         $this->_handle_error( $sock, 'error reading line from $1' );
1273                         return false;
1274                 }
1275                 if ( substr( $result, -2 ) === "\r\n" ) {
1276                         $result = substr( $result, 0, -2 );
1277                 } elseif ( substr( $result, -1 ) === "\n" ) {
1278                         $result = substr( $result, 0, -1 );
1279                 } else {
1280                         $this->_handle_error( $sock, 'line ending missing in response from $1' );
1281                         return false;
1282                 }
1283                 return $result;
1284         }
1285
1286         /**
1287          * Flush the read buffer of a stream
1288          * @param Resource $f
1289          */
1290         function _flush_read_buffer( $f ) {
1291                 if ( !is_resource( $f ) ) {
1292                         return;
1293                 }
1294                 $r = array( $f );
1295                 $w = null;
1296                 $e = null;
1297                 $n = stream_select( $r, $w, $e, 0, 0 );
1298                 while ( $n == 1 && !feof( $f ) ) {
1299                         fread( $f, 1024 );
1300                         $r = array( $f );
1301                         $w = null;
1302                         $e = null;
1303                         $n = stream_select( $r, $w, $e, 0, 0 );
1304                 }
1305         }
1306
1307         // }}}
1308         // }}}
1309         // }}}
1310 }
1311
1312 // }}}