includes/libs/objectcache/utils/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
 282         /**
 283          * @param mixed $value
 284          * @return string|integer
 285          */
 286         public function serialize( $value ) {
 287                 return serialize( $value );
 288         }
 289
 290         /**
 291          * @param string $value
 292          * @return mixed
 293          */
 294         public function unserialize( $value ) {
 295                 return unserialize( $value );
 296         }
 297
 298         // {{{ add()
 299
 300         /**
 301          * Adds a key/value to the memcache server if one isn't already set with
 302          * that key
 303          *
 304          * @param string $key Key to set with data
 305          * @param mixed $val Value to store
 306          * @param int $exp (optional) Expiration time. This can be a number of seconds
 307          * to cache for (up to 30 days inclusive).  Any timespans of 30 days + 1 second or
 308          * longer must be the timestamp of the time at which the mapping should expire. It
 309          * is safe to use timestamps in all cases, regardless of expiration
 310          * eg: strtotime("+3 hour")
 311          *
 312          * @return bool
 313          */
 314         public function add( $key, $val, $exp = 0 ) {
 315                 return $this->_set( 'add', $key, $val, $exp );
 316         }
 317
 318         // }}}
 319         // {{{ decr()
 320
 321         /**
 322          * Decrease a value stored on the memcache server
 323          *
 324          * @param string $key Key to decrease
 325          * @param int $amt (optional) amount to decrease
 326          *
 327          * @return mixed False on failure, value on success
 328          */
 329         public function decr( $key, $amt = 1 ) {
 330                 return $this->_incrdecr( 'decr', $key, $amt );
 331         }
 332
 333         // }}}
 334         // {{{ delete()
 335
 336         /**
 337          * Deletes a key from the server, optionally after $time
 338          *
 339          * @param string $key Key to delete
 340          * @param int $time (optional) how long to wait before deleting
 341          *
 342          * @return bool True on success, false on failure
 343          */
 344         public function delete( $key, $time = 0 ) {
 345                 if ( !$this->_active ) {
 346                         return false;
 347                 }
 348
 349                 $sock = $this->get_sock( $key );
 350                 if ( !is_resource( $sock ) ) {
 351                         return false;
 352                 }
 353
 354                 $key = is_array( $key ) ? $key[1] : $key;
 355
 356                 if ( isset( $this->stats['delete'] ) ) {
 357                         $this->stats['delete']++;
 358                 } else {
 359                         $this->stats['delete'] = 1;
 360                 }
 361                 $cmd = "delete $key $time\r\n";
 362                 if ( !$this->_fwrite( $sock, $cmd ) ) {
 363                         return false;
 364                 }
 365                 $res = $this->_fgets( $sock );
 366
 367                 if ( $this->_debug ) {
 368                         $this->_debugprint( sprintf( "MemCache: delete %s (%s)", $key, $res ) );
 369                 }
 370
 371                 if ( $res == "DELETED" || $res == "NOT_FOUND" ) {
 372                         return true;
 373                 }
 374
 375                 return false;
 376         }
 377
 378         /**
 379          * Changes the TTL on a key from the server to $time
 380          *
 381          * @param string $key
 382          * @param int $time TTL in seconds
 383          *
 384          * @return bool True on success, false on failure
 385          */
 386         public function touch( $key, $time = 0 ) {
 387                 if ( !$this->_active ) {
 388                         return false;
 389                 }
 390
 391                 $sock = $this->get_sock( $key );
 392                 if ( !is_resource( $sock ) ) {
 393                         return false;
 394                 }
 395
 396                 $key = is_array( $key ) ? $key[1] : $key;
 397
 398                 if ( isset( $this->stats['touch'] ) ) {
 399                         $this->stats['touch']++;
 400                 } else {
 401                         $this->stats['touch'] = 1;
 402                 }
 403                 $cmd = "touch $key $time\r\n";
 404                 if ( !$this->_fwrite( $sock, $cmd ) ) {
 405                         return false;
 406                 }
 407                 $res = $this->_fgets( $sock );
 408
 409                 if ( $this->_debug ) {
 410                         $this->_debugprint( sprintf( "MemCache: touch %s (%s)", $key, $res ) );
 411                 }
 412
 413                 if ( $res == "TOUCHED" ) {
 414                         return true;
 415                 }
 416
 417                 return false;
 418         }
 419
 420         // }}}
 421         // {{{ disconnect_all()
 422
 423         /**
 424          * Disconnects all connected sockets
 425          */
 426         public function disconnect_all() {
 427                 foreach ( $this->_cache_sock as $sock ) {
 428                         fclose( $sock );
 429                 }
 430
 431                 $this->_cache_sock = array();
 432         }
 433
 434         // }}}
 435         // {{{ enable_compress()
 436
 437         /**
 438          * Enable / Disable compression
 439          *
 440          * @param bool $enable True to enable, false to disable
 441          */
 442         public function enable_compress( $enable ) {
 443                 $this->_compress_enable = $enable;
 444         }
 445
 446         // }}}
 447         // {{{ forget_dead_hosts()
 448
 449         /**
 450          * Forget about all of the dead hosts
 451          */
 452         public function forget_dead_hosts() {
 453                 $this->_host_dead = array();
 454         }
 455
 456         // }}}
 457         // {{{ get()
 458
 459         /**
 460          * Retrieves the value associated with the key from the memcache server
 461          *
 462          * @param array|string $key key to retrieve
 463          * @param float $casToken [optional]
 464          *
 465          * @return mixed
 466          */
 467         public function get( $key, &$casToken = null ) {
 468                 if ( $this->_debug ) {
 469                         $this->_debugprint( "get($key)" );
 470                 }
 471
 472                 if ( !is_array( $key ) && strval( $key ) === '' ) {
 473                         $this->_debugprint( "Skipping key which equals to an empty string" );
 474                         return false;
 475                 }
 476
 477                 if ( !$this->_active ) {
 478                         return false;
 479                 }
 480
 481                 $sock = $this->get_sock( $key );
 482
 483                 if ( !is_resource( $sock ) ) {
 484                         return false;
 485                 }
 486
 487                 $key = is_array( $key ) ? $key[1] : $key;
 488                 if ( isset( $this->stats['get'] ) ) {
 489                         $this->stats['get']++;
 490                 } else {
 491                         $this->stats['get'] = 1;
 492                 }
 493
 494                 $cmd = "gets $key\r\n";
 495                 if ( !$this->_fwrite( $sock, $cmd ) ) {
 496                         return false;
 497                 }
 498
 499                 $val = array();
 500                 $this->_load_items( $sock, $val, $casToken );
 501
 502                 if ( $this->_debug ) {
 503                         foreach ( $val as $k => $v ) {
 504                                 $this->_debugprint(
 505                                         sprintf( "MemCache: sock %s got %s", $this->serialize( $sock ), $k ) );
 506                         }
 507                 }
 508
 509                 $value = false;
 510                 if ( isset( $val[$key] ) ) {
 511                         $value = $val[$key];
 512                 }
 513                 return $value;
 514         }
 515
 516         // }}}
 517         // {{{ get_multi()
 518
 519         /**
 520          * Get multiple keys from the server(s)
 521          *
 522          * @param array $keys Keys to retrieve
 523          *
 524          * @return array
 525          */
 526         public function get_multi( $keys ) {
 527                 if ( !$this->_active ) {
 528                         return array();
 529                 }
 530
 531                 if ( isset( $this->stats['get_multi'] ) ) {
 532                         $this->stats['get_multi']++;
 533                 } else {
 534                         $this->stats['get_multi'] = 1;
 535                 }
 536                 $sock_keys = array();
 537                 $socks = array();
 538                 foreach ( $keys as $key ) {
 539                         $sock = $this->get_sock( $key );
 540                         if ( !is_resource( $sock ) ) {
 541                                 continue;
 542                         }
 543                         $key = is_array( $key ) ? $key[1] : $key;
 544                         if ( !isset( $sock_keys[$sock] ) ) {
 545                                 $sock_keys[intval( $sock )] = array();
 546                                 $socks[] = $sock;
 547                         }
 548                         $sock_keys[intval( $sock )][] = $key;
 549                 }
 550
 551                 $gather = array();
 552                 // Send out the requests
 553                 foreach ( $socks as $sock ) {
 554                         $cmd = 'gets';
 555                         foreach ( $sock_keys[intval( $sock )] as $key ) {
 556                                 $cmd .= ' ' . $key;
 557                         }
 558                         $cmd .= "\r\n";
 559
 560                         if ( $this->_fwrite( $sock, $cmd ) ) {
 561                                 $gather[] = $sock;
 562                         }
 563                 }
 564
 565                 // Parse responses
 566                 $val = array();
 567                 foreach ( $gather as $sock ) {
 568                         $this->_load_items( $sock, $val, $casToken );
 569                 }
 570
 571                 if ( $this->_debug ) {
 572                         foreach ( $val as $k => $v ) {
 573                                 $this->_debugprint( sprintf( "MemCache: got %s", $k ) );
 574                         }
 575                 }
 576
 577                 return $val;
 578         }
 579
 580         // }}}
 581         // {{{ incr()
 582
 583         /**
 584          * Increments $key (optionally) by $amt
 585          *
 586          * @param string $key Key to increment
 587          * @param int $amt (optional) amount to increment
 588          *
 589          * @return int|null Null if the key does not exist yet (this does NOT
 590          * create new mappings if the key does not exist). If the key does
 591          * exist, this returns the new value for that key.
 592          */
 593         public function incr( $key, $amt = 1 ) {
 594                 return $this->_incrdecr( 'incr', $key, $amt );
 595         }
 596
 597         // }}}
 598         // {{{ replace()
 599
 600         /**
 601          * Overwrites an existing value for key; only works if key is already set
 602          *
 603          * @param string $key Key to set value as
 604          * @param mixed $value Value to store
 605          * @param int $exp (optional) Expiration time. This can be a number of seconds
 606          * to cache for (up to 30 days inclusive).  Any timespans of 30 days + 1 second or
 607          * longer must be the timestamp of the time at which the mapping should expire. It
 608          * is safe to use timestamps in all cases, regardless of exipration
 609          * eg: strtotime("+3 hour")
 610          *
 611          * @return bool
 612          */
 613         public function replace( $key, $value, $exp = 0 ) {
 614                 return $this->_set( 'replace', $key, $value, $exp );
 615         }
 616
 617         // }}}
 618         // {{{ run_command()
 619
 620         /**
 621          * Passes through $cmd to the memcache server connected by $sock; returns
 622          * output as an array (null array if no output)
 623          *
 624          * @param Resource $sock Socket to send command on
 625          * @param string $cmd Command to run
 626          *
 627          * @return array Output array
 628          */
 629         public function run_command( $sock, $cmd ) {
 630                 if ( !is_resource( $sock ) ) {
 631                         return array();
 632                 }
 633
 634                 if ( !$this->_fwrite( $sock, $cmd ) ) {
 635                         return array();
 636                 }
 637
 638                 $ret = array();
 639                 while ( true ) {
 640                         $res = $this->_fgets( $sock );
 641                         $ret[] = $res;
 642                         if ( preg_match( '/^END/', $res ) ) {
 643                                 break;
 644                         }
 645                         if ( strlen( $res ) == 0 ) {
 646                                 break;
 647                         }
 648                 }
 649                 return $ret;
 650         }
 651
 652         // }}}
 653         // {{{ set()
 654
 655         /**
 656          * Unconditionally sets a key to a given value in the memcache.  Returns true
 657          * if set successfully.
 658          *
 659          * @param string $key Key to set value as
 660          * @param mixed $value Value to set
 661          * @param int $exp (optional) Expiration time. This can be a number of seconds
 662          * to cache for (up to 30 days inclusive).  Any timespans of 30 days + 1 second or
 663          * longer must be the timestamp of the time at which the mapping should expire. It
 664          * is safe to use timestamps in all cases, regardless of exipration
 665          * eg: strtotime("+3 hour")
 666          *
 667          * @return bool True on success
 668          */
 669         public function set( $key, $value, $exp = 0 ) {
 670                 return $this->_set( 'set', $key, $value, $exp );
 671         }
 672
 673         // }}}
 674         // {{{ cas()
 675
 676         /**
 677          * Sets a key to a given value in the memcache if the current value still corresponds
 678          * to a known, given value.  Returns true if set successfully.
 679          *
 680          * @param float $casToken Current known value
 681          * @param string $key Key to set value as
 682          * @param mixed $value Value to set
 683          * @param int $exp (optional) Expiration time. This can be a number of seconds
 684          * to cache for (up to 30 days inclusive).  Any timespans of 30 days + 1 second or
 685          * longer must be the timestamp of the time at which the mapping should expire. It
 686          * is safe to use timestamps in all cases, regardless of exipration
 687          * eg: strtotime("+3 hour")
 688          *
 689          * @return bool True on success
 690          */
 691         public function cas( $casToken, $key, $value, $exp = 0 ) {
 692                 return $this->_set( 'cas', $key, $value, $exp, $casToken );
 693         }
 694
 695         // }}}
 696         // {{{ set_compress_threshold()
 697
 698         /**
 699          * Set the compression threshold
 700          *
 701          * @param int $thresh Threshold to compress if larger than
 702          */
 703         public function set_compress_threshold( $thresh ) {
 704                 $this->_compress_threshold = $thresh;
 705         }
 706
 707         // }}}
 708         // {{{ set_debug()
 709
 710         /**
 711          * Set the debug flag
 712          *
 713          * @see __construct()
 714          * @param bool $dbg True for debugging, false otherwise
 715          */
 716         public function set_debug( $dbg ) {
 717                 $this->_debug = $dbg;
 718         }
 719
 720         // }}}
 721         // {{{ set_servers()
 722
 723         /**
 724          * Set the server list to distribute key gets and puts between
 725          *
 726          * @see __construct()
 727          * @param array $list Array of servers to connect to
 728          */
 729         public function set_servers( $list ) {
 730                 $this->_servers = $list;
 731                 $this->_active = count( $list );
 732                 $this->_buckets = null;
 733                 $this->_bucketcount = 0;
 734
 735                 $this->_single_sock = null;
 736                 if ( $this->_active == 1 ) {
 737                         $this->_single_sock = $this->_servers[0];
 738                 }
 739         }
 740
 741         /**
 742          * Sets the timeout for new connections
 743          *
 744          * @param int $seconds Number of seconds
 745          * @param int $microseconds Number of microseconds
 746          */
 747         public function set_timeout( $seconds, $microseconds ) {
 748                 $this->_timeout_seconds = $seconds;
 749                 $this->_timeout_microseconds = $microseconds;
 750         }
 751
 752         // }}}
 753         // }}}
 754         // {{{ private methods
 755         // {{{ _close_sock()
 756
 757         /**
 758          * Close the specified socket
 759          *
 760          * @param string $sock Socket to close
 761          *
 762          * @access private
 763          */
 764         function _close_sock( $sock ) {
 765                 $host = array_search( $sock, $this->_cache_sock );
 766                 fclose( $this->_cache_sock[$host] );
 767                 unset( $this->_cache_sock[$host] );
 768         }
 769
 770         // }}}
 771         // {{{ _connect_sock()
 772
 773         /**
 774          * Connects $sock to $host, timing out after $timeout
 775          *
 776          * @param int $sock Socket to connect
 777          * @param string $host Host:IP to connect to
 778          *
 779          * @return bool
 780          * @access private
 781          */
 782         function _connect_sock( &$sock, $host ) {
 783                 list( $ip, $port ) = preg_split( '/:(?=\d)/', $host );
 784                 $sock = false;
 785                 $timeout = $this->_connect_timeout;
 786                 $errno = $errstr = null;
 787                 for ( $i = 0; !$sock && $i < $this->_connect_attempts; $i++ ) {
 788                         Wikimedia\suppressWarnings();
 789                         if ( $this->_persistent == 1 ) {
 790                                 $sock = pfsockopen( $ip, $port, $errno, $errstr, $timeout );
 791                         } else {
 792                                 $sock = fsockopen( $ip, $port, $errno, $errstr, $timeout );
 793                         }
 794                         Wikimedia\restoreWarnings();
 795                 }
 796                 if ( !$sock ) {
 797                         $this->_error_log( "Error connecting to $host: $errstr" );
 798                         $this->_dead_host( $host );
 799                         return false;
 800                 }
 801
 802                 // Initialise timeout
 803                 stream_set_timeout( $sock, $this->_timeout_seconds, $this->_timeout_microseconds );
 804
 805                 // If the connection was persistent, flush the read buffer in case there
 806                 // was a previous incomplete request on this connection
 807                 if ( $this->_persistent ) {
 808                         $this->_flush_read_buffer( $sock );
 809                 }
 810                 return true;
 811         }
 812
 813         // }}}
 814         // {{{ _dead_sock()
 815
 816         /**
 817          * Marks a host as dead until 30-40 seconds in the future
 818          *
 819          * @param string $sock Socket to mark as dead
 820          *
 821          * @access private
 822          */
 823         function _dead_sock( $sock ) {
 824                 $host = array_search( $sock, $this->_cache_sock );
 825                 $this->_dead_host( $host );
 826         }
 827
 828         /**
 829          * @param string $host
 830          */
 831         function _dead_host( $host ) {
 832                 $ip = explode( ':', $host )[0];
 833                 $this->_host_dead[$ip] = time() + 30 + intval( rand( 0, 10 ) );
 834                 $this->_host_dead[$host] = $this->_host_dead[$ip];
 835                 unset( $this->_cache_sock[$host] );
 836         }
 837
 838         // }}}
 839         // {{{ get_sock()
 840
 841         /**
 842          * get_sock
 843          *
 844          * @param string $key Key to retrieve value for;
 845          *
 846          * @return Resource|bool Resource on success, false on failure
 847          * @access private
 848          */
 849         function get_sock( $key ) {
 850                 if ( !$this->_active ) {
 851                         return false;
 852                 }
 853
 854                 if ( $this->_single_sock !== null ) {
 855                         return $this->sock_to_host( $this->_single_sock );
 856                 }
 857
 858                 $hv = is_array( $key ) ? intval( $key[0] ) : $this->_hashfunc( $key );
 859                 if ( $this->_buckets === null ) {
 860                         $bu = array();
 861                         foreach ( $this->_servers as $v ) {
 862                                 if ( is_array( $v ) ) {
 863                                         for ( $i = 0; $i < $v[1]; $i++ ) {
 864                                                 $bu[] = $v[0];
 865                                         }
 866                                 } else {
 867                                         $bu[] = $v;
 868                                 }
 869                         }
 870                         $this->_buckets = $bu;
 871                         $this->_bucketcount = count( $bu );
 872                 }
 873
 874                 $realkey = is_array( $key ) ? $key[1] : $key;
 875                 for ( $tries = 0; $tries < 20; $tries++ ) {
 876                         $host = $this->_buckets[$hv % $this->_bucketcount];
 877                         $sock = $this->sock_to_host( $host );
 878                         if ( is_resource( $sock ) ) {
 879                                 return $sock;
 880                         }
 881                         $hv = $this->_hashfunc( $hv . $realkey );
 882                 }
 883
 884                 return false;
 885         }
 886
 887         // }}}
 888         // {{{ _hashfunc()
 889
 890         /**
 891          * Creates a hash integer based on the $key
 892          *
 893          * @param string $key Key to hash
 894          *
 895          * @return int Hash value
 896          * @access private
 897          */
 898         function _hashfunc( $key ) {
 899                 # Hash function must be in [0,0x7ffffff]
 900                 # We take the first 31 bits of the MD5 hash, which unlike the hash
 901                 # function used in a previous version of this client, works
 902                 return hexdec( substr( md5( $key ), 0, 8 ) ) & 0x7fffffff;
 903         }
 904
 905         // }}}
 906         // {{{ _incrdecr()
 907
 908         /**
 909          * Perform increment/decriment on $key
 910          *
 911          * @param string $cmd Command to perform
 912          * @param string|array $key Key to perform it on
 913          * @param int $amt Amount to adjust
 914          *
 915          * @return int New value of $key
 916          * @access private
 917          */
 918         function _incrdecr( $cmd, $key, $amt = 1 ) {
 919                 if ( !$this->_active ) {
 920                         return null;
 921                 }
 922
 923                 $sock = $this->get_sock( $key );
 924                 if ( !is_resource( $sock ) ) {
 925                         return null;
 926                 }
 927
 928                 $key = is_array( $key ) ? $key[1] : $key;
 929                 if ( isset( $this->stats[$cmd] ) ) {
 930                         $this->stats[$cmd]++;
 931                 } else {
 932                         $this->stats[$cmd] = 1;
 933                 }
 934                 if ( !$this->_fwrite( $sock, "$cmd $key $amt\r\n" ) ) {
 935                         return null;
 936                 }
 937
 938                 $line = $this->_fgets( $sock );
 939                 $match = array();
 940                 if ( !preg_match( '/^(\d+)/', $line, $match ) ) {
 941                         return null;
 942                 }
 943                 return $match[1];
 944         }
 945
 946         // }}}
 947         // {{{ _load_items()
 948
 949         /**
 950          * Load items into $ret from $sock
 951          *
 952          * @param Resource $sock Socket to read from
 953          * @param array $ret returned values
 954          * @param float $casToken [optional]
 955          * @return bool True for success, false for failure
 956          *
 957          * @access private
 958          */
 959         function _load_items( $sock, &$ret, &$casToken = null ) {
 960                 $results = array();
 961
 962                 while ( 1 ) {
 963                         $decl = $this->_fgets( $sock );
 964
 965                         if ( $decl === false ) {
 966                                 /*
 967                                  * If nothing can be read, something is wrong because we know exactly when
 968                                  * to stop reading (right after "END") and we return right after that.
 969                                  */
 970                                 return false;
 971                         } elseif ( preg_match( '/^VALUE (\S+) (\d+) (\d+) (\d+)$/', $decl, $match ) ) {
 972                                 /*
 973                                  * Read all data returned. This can be either one or multiple values.
 974                                  * Save all that data (in an array) to be processed later: we'll first
 975                                  * want to continue reading until "END" before doing anything else,
 976                                  * to make sure that we don't leave our client in a state where it's
 977                                  * output is not yet fully read.
 978                                  */
 979                                 $results[] = array(
 980                                         $match[1], // rkey
 981                                         $match[2], // flags
 982                                         $match[3], // len
 983                                         $match[4], // casToken
 984                                         $this->_fread( $sock, $match[3] + 2 ), // data
 985                                 );
 986                         } elseif ( $decl == "END" ) {
 987                                 if ( count( $results ) == 0 ) {
 988                                         return false;
 989                                 }
 990
 991                                 /**
 992                                  * All data has been read, time to process the data and build
 993                                  * meaningful return values.
 994                                  */
 995                                 foreach ( $results as $vars ) {
 996                                         list( $rkey, $flags, $len, $casToken, $data ) = $vars;
 997
 998                                         if ( $data === false || substr( $data, -2 ) !== "\r\n" ) {
 999                                                 $this->_handle_error( $sock,
1000                                                         'line ending missing from data block from $1' );
1001                                                 return false;
1002                                         }
1003                                         $data = substr( $data, 0, -2 );
1004                                         $ret[$rkey] = $data;
1005
1006                                         if ( $this->_have_zlib && $flags & self::COMPRESSED ) {
1007                                                 $ret[$rkey] = gzuncompress( $ret[$rkey] );
1008                                         }
1009
1010                                         /*
1011                                          * This unserialize is the exact reason that we only want to
1012                                          * process data after having read until "END" (instead of doing
1013                                          * this right away): "unserialize" can trigger outside code:
1014                                          * in the event that $ret[$rkey] is a serialized object,
1015                                          * unserializing it will trigger __wakeup() if present. If that
1016                                          * function attempted to read from memcached (while we did not
1017                                          * yet read "END"), these 2 calls would collide.
1018                                          */
1019                                         if ( $flags & self::SERIALIZED ) {
1020                                                 $ret[$rkey] = $this->unserialize( $ret[$rkey] );
1021                                         } elseif ( $flags & self::INTVAL ) {
1022                                                 $ret[$rkey] = intval( $ret[$rkey] );
1023                                         }
1024                                 }
1025
1026                                 return true;
1027                         } else {
1028                                 $this->_handle_error( $sock, 'Error parsing response from $1' );
1029                                 return false;
1030                         }
1031                 }
1032         }
1033
1034         // }}}
1035         // {{{ _set()
1036
1037         /**
1038          * Performs the requested storage operation to the memcache server
1039          *
1040          * @param string $cmd Command to perform
1041          * @param string $key Key to act on
1042          * @param mixed $val What we need to store
1043          * @param int $exp (optional) Expiration time. This can be a number of seconds
1044          * to cache for (up to 30 days inclusive).  Any timespans of 30 days + 1 second or
1045          * longer must be the timestamp of the time at which the mapping should expire. It
1046          * is safe to use timestamps in all cases, regardless of exipration
1047          * eg: strtotime("+3 hour")
1048          * @param float $casToken [optional]
1049          *
1050          * @return bool
1051          * @access private
1052          */
1053         function _set( $cmd, $key, $val, $exp, $casToken = null ) {
1054                 if ( !$this->_active ) {
1055                         return false;
1056                 }
1057
1058                 $sock = $this->get_sock( $key );
1059                 if ( !is_resource( $sock ) ) {
1060                         return false;
1061                 }
1062
1063                 if ( isset( $this->stats[$cmd] ) ) {
1064                         $this->stats[$cmd]++;
1065                 } else {
1066                         $this->stats[$cmd] = 1;
1067                 }
1068
1069                 $flags = 0;
1070
1071                 if ( is_int( $val ) ) {
1072                         $flags |= self::INTVAL;
1073                 } elseif ( !is_scalar( $val ) ) {
1074                         $val = $this->serialize( $val );
1075                         $flags |= self::SERIALIZED;
1076                         if ( $this->_debug ) {
1077                                 $this->_debugprint( sprintf( "client: serializing data as it is not scalar" ) );
1078                         }
1079                 }
1080
1081                 $len = strlen( $val );
1082
1083                 if ( $this->_have_zlib && $this->_compress_enable
1084                         && $this->_compress_threshold && $len >= $this->_compress_threshold
1085                 ) {
1086                         $c_val = gzcompress( $val, 9 );
1087                         $c_len = strlen( $c_val );
1088
1089                         if ( $c_len < $len * ( 1 - self::COMPRESSION_SAVINGS ) ) {
1090                                 if ( $this->_debug ) {
1091                                         $this->_debugprint( sprintf( "client: compressing data; was %d bytes is now %d bytes", $len, $c_len ) );
1092                                 }
1093                                 $val = $c_val;
1094                                 $len = $c_len;
1095                                 $flags |= self::COMPRESSED;
1096                         }
1097                 }
1098
1099                 $command = "$cmd $key $flags $exp $len";
1100                 if ( $casToken ) {
1101                         $command .= " $casToken";
1102                 }
1103
1104                 if ( !$this->_fwrite( $sock, "$command\r\n$val\r\n" ) ) {
1105                         return false;
1106                 }
1107
1108                 $line = $this->_fgets( $sock );
1109
1110                 if ( $this->_debug ) {
1111                         $this->_debugprint( sprintf( "%s %s (%s)", $cmd, $key, $line ) );
1112                 }
1113                 if ( $line === "STORED" ) {
1114                         return true;
1115                 } elseif ( $line === "NOT_STORED" && $cmd === "set" ) {
1116                         // "Not stored" is always used as the mcrouter response with AllAsyncRoute
1117                         return true;
1118                 }
1119
1120                 return false;
1121         }
1122
1123         // }}}
1124         // {{{ sock_to_host()
1125
1126         /**
1127          * Returns the socket for the host
1128          *
1129          * @param string $host Host:IP to get socket for
1130          *
1131          * @return Resource|bool IO Stream or false
1132          * @access private
1133          */
1134         function sock_to_host( $host ) {
1135                 if ( isset( $this->_cache_sock[$host] ) ) {
1136                         return $this->_cache_sock[$host];
1137                 }
1138
1139                 $sock = null;
1140                 $now = time();
1141                 list( $ip, /* $port */) = explode( ':', $host );
1142                 if ( isset( $this->_host_dead[$host] ) && $this->_host_dead[$host] > $now ||
1143                         isset( $this->_host_dead[$ip] ) && $this->_host_dead[$ip] > $now
1144                 ) {
1145                         return null;
1146                 }
1147
1148                 if ( !$this->_connect_sock( $sock, $host ) ) {
1149                         return null;
1150                 }
1151
1152                 // Do not buffer writes
1153                 stream_set_write_buffer( $sock, 0 );
1154
1155                 $this->_cache_sock[$host] = $sock;
1156
1157                 return $this->_cache_sock[$host];
1158         }
1159
1160         /**
1161          * @param string $text
1162          */
1163         function _debugprint( $text ) {
1164                 $this->_logger->debug( $text );
1165         }
1166
1167         /**
1168          * @param string $text
1169          */
1170         function _error_log( $text ) {
1171                 $this->_logger->error( "Memcached error: $text" );
1172         }
1173
1174         /**
1175          * Write to a stream. If there is an error, mark the socket dead.
1176          *
1177          * @param Resource $sock The socket
1178          * @param string $buf The string to write
1179          * @return bool True on success, false on failure
1180          */
1181         function _fwrite( $sock, $buf ) {
1182                 $bytesWritten = 0;
1183                 $bufSize = strlen( $buf );
1184                 while ( $bytesWritten < $bufSize ) {
1185                         $result = fwrite( $sock, $buf );
1186                         $data = stream_get_meta_data( $sock );
1187                         if ( $data['timed_out'] ) {
1188                                 $this->_handle_error( $sock, 'timeout writing to $1' );
1189                                 return false;
1190                         }
1191                         // Contrary to the documentation, fwrite() returns zero on error in PHP 5.3.
1192                         if ( $result === false || $result === 0 ) {
1193                                 $this->_handle_error( $sock, 'error writing to $1' );
1194                                 return false;
1195                         }
1196                         $bytesWritten += $result;
1197                 }
1198
1199                 return true;
1200         }
1201
1202         /**
1203          * Handle an I/O error. Mark the socket dead and log an error.
1204          *
1205          * @param Resource $sock
1206          * @param string $msg
1207          */
1208         function _handle_error( $sock, $msg ) {
1209                 $peer = stream_socket_get_name( $sock, true /** remote **/ );
1210                 if ( strval( $peer ) === '' ) {
1211                         $peer = array_search( $sock, $this->_cache_sock );
1212                         if ( $peer === false ) {
1213                                 $peer = '[unknown host]';
1214                         }
1215                 }
1216                 $msg = str_replace( '$1', $peer, $msg );
1217                 $this->_error_log( "$msg" );
1218                 $this->_dead_sock( $sock );
1219         }
1220
1221         /**
1222          * Read the specified number of bytes from a stream. If there is an error,
1223          * mark the socket dead.
1224          *
1225          * @param Resource $sock The socket
1226          * @param int $len The number of bytes to read
1227          * @return string|bool The string on success, false on failure.
1228          */
1229         function _fread( $sock, $len ) {
1230                 $buf = '';
1231                 while ( $len > 0 ) {
1232                         $result = fread( $sock, $len );
1233                         $data = stream_get_meta_data( $sock );
1234                         if ( $data['timed_out'] ) {
1235                                 $this->_handle_error( $sock, 'timeout reading from $1' );
1236                                 return false;
1237                         }
1238                         if ( $result === false ) {
1239                                 $this->_handle_error( $sock, 'error reading buffer from $1' );
1240                                 return false;
1241                         }
1242                         if ( $result === '' ) {
1243                                 // This will happen if the remote end of the socket is shut down
1244                                 $this->_handle_error( $sock, 'unexpected end of file reading from $1' );
1245                                 return false;
1246                         }
1247                         $len -= strlen( $result );
1248                         $buf .= $result;
1249                 }
1250                 return $buf;
1251         }
1252
1253         /**
1254          * Read a line from a stream. If there is an error, mark the socket dead.
1255          * The \r\n line ending is stripped from the response.
1256          *
1257          * @param Resource $sock The socket
1258          * @return string|bool The string on success, false on failure
1259          */
1260         function _fgets( $sock ) {
1261                 $result = fgets( $sock );
1262                 // fgets() may return a partial line if there is a select timeout after
1263                 // a successful recv(), so we have to check for a timeout even if we
1264                 // got a string response.
1265                 $data = stream_get_meta_data( $sock );
1266                 if ( $data['timed_out'] ) {
1267                         $this->_handle_error( $sock, 'timeout reading line from $1' );
1268                         return false;
1269                 }
1270                 if ( $result === false ) {
1271                         $this->_handle_error( $sock, 'error reading line from $1' );
1272                         return false;
1273                 }
1274                 if ( substr( $result, -2 ) === "\r\n" ) {
1275                         $result = substr( $result, 0, -2 );
1276                 } elseif ( substr( $result, -1 ) === "\n" ) {
1277                         $result = substr( $result, 0, -1 );
1278                 } else {
1279                         $this->_handle_error( $sock, 'line ending missing in response from $1' );
1280                         return false;
1281                 }
1282                 return $result;
1283         }
1284
1285         /**
1286          * Flush the read buffer of a stream
1287          * @param Resource $f
1288          */
1289         function _flush_read_buffer( $f ) {
1290                 if ( !is_resource( $f ) ) {
1291                         return;
1292                 }
1293                 $r = array( $f );
1294                 $w = null;
1295                 $e = null;
1296                 $n = stream_select( $r, $w, $e, 0, 0 );
1297                 while ( $n == 1 && !feof( $f ) ) {
1298                         fread( $f, 1024 );
1299                         $r = array( $f );
1300                         $w = null;
1301                         $e = null;
1302                         $n = stream_select( $r, $w, $e, 0, 0 );
1303                 }
1304         }
1305
1306         // }}}
1307         // }}}
1308         // }}}
1309 }
1310
1311 // }}}