includes/memcached-client.php

   1 <?php
   2 //
   3 // +---------------------------------------------------------------------------+
   4 // | memcached client, PHP                                                     |
   5 // +---------------------------------------------------------------------------+
   6 // | Copyright (c) 2003 Ryan T. Dean <rtdean@cytherianage.net>                 |
   7 // | All rights reserved.                                                      |
   8 // |                                                                           |
   9 // | Redistribution and use in source and binary forms, with or without        |
  10 // | modification, are permitted provided that the following conditions        |
  11 // | are met:                                                                  |
  12 // |                                                                           |
  13 // | 1. Redistributions of source code must retain the above copyright         |
  14 // |    notice, this list of conditions and the following disclaimer.          |
  15 // | 2. Redistributions in binary form must reproduce the above copyright      |
  16 // |    notice, this list of conditions and the following disclaimer in the    |
  17 // |    documentation and/or other materials provided with the distribution.   |
  18 // |                                                                           |
  19 // | THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR      |
  20 // | IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES |
  21 // | OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.   |
  22 // | IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,          |
  23 // | INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT  |
  24 // | NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
  25 // | DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY     |
  26 // | THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT       |
  27 // | (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF  |
  28 // | THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.         |
  29 // +---------------------------------------------------------------------------+
  30 // | Author: Ryan T. Dean <rtdean@cytherianage.net>                            |
  31 // | Heavily influenced by the Perl memcached client by Brad Fitzpatrick.      |
  32 // |   Permission granted by Brad Fitzpatrick for relicense of ported Perl     |
  33 // |   client logic under 2-clause BSD license.                                |
  34 // +---------------------------------------------------------------------------+
  35 //
  36 // $TCAnet$
  37 //
  38
  39 /**
  40  * This is the PHP client for memcached - a distributed memory cache daemon.
  41  * More information is available at http://www.danga.com/memcached/
  42  *
  43  * Usage example:
  44  *
  45  * require_once 'memcached.php';
  46  *
  47  * $mc = new MWMemcached(array(
  48  *              'servers' => array('127.0.0.1:10000',
  49  *                                 array('192.0.0.1:10010', 2),
  50  *                                 '127.0.0.1:10020'),
  51  *              'debug'   => false,
  52  *              'compress_threshold' => 10240,
  53  *              'persistant' => true));
  54  *
  55  * $mc->add('key', array('some', 'array'));
  56  * $mc->replace('key', 'some random string');
  57  * $val = $mc->get('key');
  58  *
  59  * @author  Ryan T. Dean <rtdean@cytherianage.net>
  60  * @version 0.1.2
  61  */
  62
  63 // {{{ requirements
  64 // }}}
  65
  66 // {{{ class MWMemcached
  67 /**
  68  * memcached client class implemented using (p)fsockopen()
  69  *
  70  * @author  Ryan T. Dean <rtdean@cytherianage.net>
  71  * @ingroup Cache
  72  */
  73 class MWMemcached
  74 {
  75    // {{{ properties
  76    // {{{ public
  77
  78         // {{{ constants
  79         // {{{ flags
  80
  81         /**
  82          * Flag: indicates data is serialized
  83          */
  84         const SERIALIZED = 1;
  85
  86         /**
  87          * Flag: indicates data is compressed
  88          */
  89         const COMPRESSED = 2;
  90
  91         // }}}
  92
  93         /**
  94          * Minimum savings to store data compressed
  95          */
  96         const COMPRESSION_SAVINGS = 0.20;
  97
  98         // }}}
  99
 100
 101         /**
 102          * Command statistics
 103          *
 104          * @var     array
 105          * @access  public
 106          */
 107         var $stats;
 108
 109         // }}}
 110         // {{{ private
 111
 112         /**
 113          * Cached Sockets that are connected
 114          *
 115          * @var     array
 116          * @access  private
 117          */
 118         var $_cache_sock;
 119
 120         /**
 121          * Current debug status; 0 - none to 9 - profiling
 122          *
 123          * @var     boolean
 124          * @access  private
 125          */
 126         var $_debug;
 127
 128         /**
 129          * Dead hosts, assoc array, 'host'=>'unixtime when ok to check again'
 130          *
 131          * @var     array
 132          * @access  private
 133          */
 134         var $_host_dead;
 135
 136         /**
 137          * Is compression available?
 138          *
 139          * @var     boolean
 140          * @access  private
 141          */
 142         var $_have_zlib;
 143
 144         /**
 145          * Do we want to use compression?
 146          *
 147          * @var     boolean
 148          * @access  private
 149          */
 150         var $_compress_enable;
 151
 152         /**
 153          * At how many bytes should we compress?
 154          *
 155          * @var     integer
 156          * @access  private
 157          */
 158         var $_compress_threshold;
 159
 160         /**
 161          * Are we using persistant links?
 162          *
 163          * @var     boolean
 164          * @access  private
 165          */
 166         var $_persistant;
 167
 168         /**
 169          * If only using one server; contains ip:port to connect to
 170          *
 171          * @var     string
 172          * @access  private
 173          */
 174         var $_single_sock;
 175
 176         /**
 177          * Array containing ip:port or array(ip:port, weight)
 178          *
 179          * @var     array
 180          * @access  private
 181          */
 182         var $_servers;
 183
 184         /**
 185          * Our bit buckets
 186          *
 187          * @var     array
 188          * @access  private
 189          */
 190         var $_buckets;
 191
 192         /**
 193          * Total # of bit buckets we have
 194          *
 195          * @var     integer
 196          * @access  private
 197          */
 198         var $_bucketcount;
 199
 200         /**
 201          * # of total servers we have
 202          *
 203          * @var     integer
 204          * @access  private
 205          */
 206         var $_active;
 207
 208         /**
 209          * Stream timeout in seconds. Applies for example to fread()
 210          *
 211          * @var     integer
 212          * @access  private
 213          */
 214         var $_timeout_seconds;
 215
 216         /**
 217          * Stream timeout in microseconds
 218          *
 219          * @var     integer
 220          * @access  private
 221          */
 222         var $_timeout_microseconds;
 223
 224         /**
 225          * Connect timeout in seconds
 226          */
 227         var $_connect_timeout;
 228
 229         /**
 230          * Number of connection attempts for each server
 231          */
 232         var $_connect_attempts;
 233
 234         // }}}
 235         // }}}
 236         // {{{ methods
 237         // {{{ public functions
 238         // {{{ memcached()
 239
 240         /**
 241          * Memcache initializer
 242          *
 243          * @param   array    $args    Associative array of settings
 244          *
 245          * @return  mixed
 246          * @access  public
 247          */
 248         function __construct ($args)
 249         {
 250                 $this->set_servers(@$args['servers']);
 251                 $this->_debug = @$args['debug'];
 252                 $this->stats = array();
 253                 $this->_compress_threshold = @$args['compress_threshold'];
 254                 $this->_persistant = array_key_exists('persistant', $args) ? (@$args['persistant']) : false;
 255                 $this->_compress_enable = true;
 256                 $this->_have_zlib = function_exists("gzcompress");
 257
 258                 $this->_cache_sock = array();
 259                 $this->_host_dead = array();
 260
 261                 $this->_timeout_seconds = 0;
 262                 $this->_timeout_microseconds = 50000;
 263
 264                 $this->_connect_timeout = 0.01;
 265                 $this->_connect_attempts = 2;
 266         }
 267
 268         // }}}
 269         // {{{ add()
 270
 271         /**
 272          * Adds a key/value to the memcache server if one isn't already set with
 273          * that key
 274          *
 275          * @param   string  $key     Key to set with data
 276          * @param   mixed   $val     Value to store
 277          * @param   integer $exp     (optional) Time to expire data at
 278          *
 279          * @return  boolean
 280          * @access  public
 281          */
 282         function add ($key, $val, $exp = 0)
 283         {
 284                 return $this->_set('add', $key, $val, $exp);
 285         }
 286
 287         // }}}
 288         // {{{ decr()
 289
 290         /**
 291          * Decriment a value stored on the memcache server
 292          *
 293          * @param   string   $key     Key to decriment
 294          * @param   integer  $amt     (optional) Amount to decriment
 295          *
 296          * @return  mixed    FALSE on failure, value on success
 297          * @access  public
 298          */
 299         function decr ($key, $amt=1)
 300         {
 301                 return $this->_incrdecr('decr', $key, $amt);
 302         }
 303
 304         // }}}
 305         // {{{ delete()
 306
 307         /**
 308          * Deletes a key from the server, optionally after $time
 309          *
 310          * @param   string   $key     Key to delete
 311          * @param   integer  $time    (optional) How long to wait before deleting
 312          *
 313          * @return  boolean  TRUE on success, FALSE on failure
 314          * @access  public
 315          */
 316         function delete ($key, $time = 0)
 317         {
 318                 if (!$this->_active)
 319                         return false;
 320
 321                 $sock = $this->get_sock($key);
 322                 if (!is_resource($sock))
 323                         return false;
 324
 325                 $key = is_array($key) ? $key[1] : $key;
 326
 327                 @$this->stats['delete']++;
 328                 $cmd = "delete $key $time\r\n";
 329                 if(!$this->_safe_fwrite($sock, $cmd, strlen($cmd)))
 330                 {
 331                         $this->_dead_sock($sock);
 332                         return false;
 333                 }
 334                 $res = trim(fgets($sock));
 335
 336                 if ($this->_debug)
 337                         $this->_debugprint(sprintf("MemCache: delete %s (%s)\n", $key, $res));
 338
 339                 if ($res == "DELETED")
 340                         return true;
 341                 return false;
 342         }
 343
 344         // }}}
 345         // {{{ disconnect_all()
 346
 347         /**
 348          * Disconnects all connected sockets
 349          *
 350          * @access  public
 351          */
 352         function disconnect_all ()
 353         {
 354                 foreach ($this->_cache_sock as $sock)
 355                         fclose($sock);
 356
 357                 $this->_cache_sock = array();
 358         }
 359
 360         // }}}
 361         // {{{ enable_compress()
 362
 363         /**
 364          * Enable / Disable compression
 365          *
 366          * @param   boolean  $enable  TRUE to enable, FALSE to disable
 367          *
 368          * @access  public
 369          */
 370         function enable_compress ($enable)
 371         {
 372                 $this->_compress_enable = $enable;
 373         }
 374
 375         // }}}
 376         // {{{ forget_dead_hosts()
 377
 378         /**
 379          * Forget about all of the dead hosts
 380          *
 381          * @access  public
 382          */
 383         function forget_dead_hosts ()
 384         {
 385                 $this->_host_dead = array();
 386         }
 387
 388         // }}}
 389         // {{{ get()
 390
 391         /**
 392          * Retrieves the value associated with the key from the memcache server
 393          *
 394          * @param  string   $key     Key to retrieve
 395          *
 396          * @return  mixed
 397          * @access  public
 398          */
 399         function get ($key)
 400         {
 401                 wfProfileIn( __METHOD__ );
 402
 403                 if ( $this->_debug ) {
 404                         $this->_debugprint( "get($key)\n" );
 405                 }
 406
 407                 if (!$this->_active) {
 408                         wfProfileOut( __METHOD__ );
 409                         return false;
 410                 }
 411
 412                 $sock = $this->get_sock($key);
 413
 414                 if (!is_resource($sock)) {
 415                         wfProfileOut( __METHOD__ );
 416                         return false;
 417                 }
 418
 419                 @$this->stats['get']++;
 420
 421                 $cmd = "get $key\r\n";
 422                 if (!$this->_safe_fwrite($sock, $cmd, strlen($cmd)))
 423                 {
 424                         $this->_dead_sock($sock);
 425                         wfProfileOut( __METHOD__ );
 426                         return false;
 427                 }
 428
 429                 $val = array();
 430                 $this->_load_items($sock, $val);
 431
 432                 if ($this->_debug)
 433                         foreach ($val as $k => $v)
 434                                 $this->_debugprint(sprintf("MemCache: sock %s got %s\n", serialize($sock), $k));
 435
 436                 wfProfileOut( __METHOD__ );
 437                 return @$val[$key];
 438         }
 439
 440         // }}}
 441         // {{{ get_multi()
 442
 443         /**
 444          * Get multiple keys from the server(s)
 445          *
 446          * @param   array    $keys    Keys to retrieve
 447          *
 448          * @return  array
 449          * @access  public
 450          */
 451         function get_multi ($keys)
 452         {
 453                 if (!$this->_active)
 454                         return false;
 455
 456                 @$this->stats['get_multi']++;
 457                 $sock_keys = array();
 458
 459                 foreach ($keys as $key)
 460                 {
 461                         $sock = $this->get_sock($key);
 462                         if (!is_resource($sock)) continue;
 463                         $key = is_array($key) ? $key[1] : $key;
 464                         if (!isset($sock_keys[$sock]))
 465                         {
 466                                 $sock_keys[$sock] = array();
 467                                 $socks[] = $sock;
 468                         }
 469                         $sock_keys[$sock][] = $key;
 470                 }
 471
 472                 // Send out the requests
 473                 foreach ($socks as $sock)
 474                 {
 475                         $cmd = "get";
 476                         foreach ($sock_keys[$sock] as $key)
 477                         {
 478                                 $cmd .= " ". $key;
 479                         }
 480                         $cmd .= "\r\n";
 481
 482                         if ($this->_safe_fwrite($sock, $cmd, strlen($cmd)))
 483                         {
 484                                 $gather[] = $sock;
 485                         } else
 486                         {
 487                                 $this->_dead_sock($sock);
 488                         }
 489                 }
 490
 491                 // Parse responses
 492                 $val = array();
 493                 foreach ($gather as $sock)
 494                 {
 495                         $this->_load_items($sock, $val);
 496                 }
 497
 498                 if ($this->_debug)
 499                         foreach ($val as $k => $v)
 500                                 $this->_debugprint(sprintf("MemCache: got %s\n", $k));
 501
 502                 return $val;
 503         }
 504
 505         // }}}
 506         // {{{ incr()
 507
 508         /**
 509          * Increments $key (optionally) by $amt
 510          *
 511          * @param   string   $key     Key to increment
 512          * @param   integer  $amt     (optional) amount to increment
 513          *
 514          * @return  integer  New key value?
 515          * @access  public
 516          */
 517         function incr ($key, $amt=1)
 518         {
 519                 return $this->_incrdecr('incr', $key, $amt);
 520         }
 521
 522         // }}}
 523         // {{{ replace()
 524
 525         /**
 526          * Overwrites an existing value for key; only works if key is already set
 527          *
 528          * @param   string   $key     Key to set value as
 529          * @param   mixed    $value   Value to store
 530          * @param   integer  $exp     (optional) Experiation time
 531          *
 532          * @return  boolean
 533          * @access  public
 534          */
 535         function replace ($key, $value, $exp=0)
 536         {
 537                 return $this->_set('replace', $key, $value, $exp);
 538         }
 539
 540         // }}}
 541         // {{{ run_command()
 542
 543         /**
 544          * Passes through $cmd to the memcache server connected by $sock; returns
 545          * output as an array (null array if no output)
 546          *
 547          * NOTE: due to a possible bug in how PHP reads while using fgets(), each
 548          *       line may not be terminated by a \r\n.  More specifically, my testing
 549          *       has shown that, on FreeBSD at least, each line is terminated only
 550          *       with a \n.  This is with the PHP flag auto_detect_line_endings set
 551          *       to falase (the default).
 552          *
 553          * @param   resource $sock    Socket to send command on
 554          * @param   string   $cmd     Command to run
 555          *
 556          * @return  array    Output array
 557          * @access  public
 558          */
 559         function run_command ($sock, $cmd)
 560         {
 561                 if (!is_resource($sock))
 562                         return array();
 563
 564                 if (!$this->_safe_fwrite($sock, $cmd, strlen($cmd)))
 565                         return array();
 566
 567                 while (true)
 568                 {
 569                         $res = fgets($sock);
 570                         $ret[] = $res;
 571                         if (preg_match('/^END/', $res))
 572                                 break;
 573                         if (strlen($res) == 0)
 574                                 break;
 575                 }
 576                 return $ret;
 577         }
 578
 579         // }}}
 580         // {{{ set()
 581
 582         /**
 583          * Unconditionally sets a key to a given value in the memcache.  Returns true
 584          * if set successfully.
 585          *
 586          * @param   string   $key     Key to set value as
 587          * @param   mixed    $value   Value to set
 588          * @param   integer  $exp     (optional) Experiation time
 589          *
 590          * @return  boolean  TRUE on success
 591          * @access  public
 592          */
 593         function set ($key, $value, $exp=0)
 594         {
 595                 return $this->_set('set', $key, $value, $exp);
 596         }
 597
 598         // }}}
 599         // {{{ set_compress_threshold()
 600
 601         /**
 602          * Sets the compression threshold
 603          *
 604          * @param   integer  $thresh  Threshold to compress if larger than
 605          *
 606          * @access  public
 607          */
 608         function set_compress_threshold ($thresh)
 609         {
 610                 $this->_compress_threshold = $thresh;
 611         }
 612
 613         // }}}
 614         // {{{ set_debug()
 615
 616         /**
 617          * Sets the debug flag
 618          *
 619          * @param   boolean  $dbg     TRUE for debugging, FALSE otherwise
 620          *
 621          * @access  public
 622          *
 623          * @see     MWMemcached::__construct
 624          */
 625         function set_debug ($dbg)
 626         {
 627                 $this->_debug = $dbg;
 628         }
 629
 630         // }}}
 631         // {{{ set_servers()
 632
 633         /**
 634          * Sets the server list to distribute key gets and puts between
 635          *
 636          * @param   array    $list    Array of servers to connect to
 637          *
 638          * @access  public
 639          *
 640          * @see     MWMemcached::__construct()
 641          */
 642         function set_servers ($list)
 643         {
 644                 $this->_servers = $list;
 645                 $this->_active = count($list);
 646                 $this->_buckets = null;
 647                 $this->_bucketcount = 0;
 648
 649                 $this->_single_sock = null;
 650                 if ($this->_active == 1)
 651                         $this->_single_sock = $this->_servers[0];
 652         }
 653
 654         /**
 655          * Sets the timeout for new connections
 656          *
 657          * @param   integer  $seconds Number of seconds
 658          * @param   integer  $microseconds  Number of microseconds
 659          *
 660          * @access  public
 661          */
 662         function set_timeout ($seconds, $microseconds)
 663         {
 664                 $this->_timeout_seconds = $seconds;
 665                 $this->_timeout_microseconds = $microseconds;
 666         }
 667
 668         // }}}
 669         // }}}
 670         // {{{ private methods
 671         // {{{ _close_sock()
 672
 673         /**
 674          * Close the specified socket
 675          *
 676          * @param   string   $sock    Socket to close
 677          *
 678          * @access  private
 679          */
 680         function _close_sock ($sock)
 681         {
 682                 $host = array_search($sock, $this->_cache_sock);
 683                 fclose($this->_cache_sock[$host]);
 684                 unset($this->_cache_sock[$host]);
 685         }
 686
 687         // }}}
 688         // {{{ _connect_sock()
 689
 690         /**
 691          * Connects $sock to $host, timing out after $timeout
 692          *
 693          * @param   integer  $sock    Socket to connect
 694          * @param   string   $host    Host:IP to connect to
 695          *
 696          * @return  boolean
 697          * @access  private
 698          */
 699         function _connect_sock (&$sock, $host)
 700         {
 701                 list ($ip, $port) = explode(":", $host);
 702                 $sock = false;
 703                 $timeout = $this->_connect_timeout;
 704                 $errno = $errstr = null;
 705                 for ($i = 0; !$sock && $i < $this->_connect_attempts; $i++) {
 706                         if ($this->_persistant == 1)
 707                         {
 708                                 $sock = @pfsockopen($ip, $port, $errno, $errstr, $timeout);
 709                         } else
 710                         {
 711                                 $sock = @fsockopen($ip, $port, $errno, $errstr, $timeout);
 712                         }
 713                 }
 714                 if (!$sock) {
 715                         if ($this->_debug)
 716                                 $this->_debugprint( "Error connecting to $host: $errstr\n" );
 717                         return false;
 718                 }
 719
 720                 // Initialise timeout
 721                 stream_set_timeout($sock, $this->_timeout_seconds, $this->_timeout_microseconds);
 722
 723                 return true;
 724         }
 725
 726         // }}}
 727         // {{{ _dead_sock()
 728
 729         /**
 730          * Marks a host as dead until 30-40 seconds in the future
 731          *
 732          * @param   string   $sock    Socket to mark as dead
 733          *
 734          * @access  private
 735          */
 736         function _dead_sock ($sock)
 737         {
 738                 $host = array_search($sock, $this->_cache_sock);
 739                 $this->_dead_host($host);
 740         }
 741
 742         function _dead_host ($host)
 743         {
 744                 @list ($ip, /* $port */) = explode(":", $host);
 745                 $this->_host_dead[$ip] = time() + 30 + intval(rand(0, 10));
 746                 $this->_host_dead[$host] = $this->_host_dead[$ip];
 747                 unset($this->_cache_sock[$host]);
 748         }
 749
 750         // }}}
 751         // {{{ get_sock()
 752
 753         /**
 754          * get_sock
 755          *
 756          * @param   string   $key     Key to retrieve value for;
 757          *
 758          * @return  mixed    resource on success, false on failure
 759          * @access  private
 760          */
 761         function get_sock ($key)
 762         {
 763                 if (!$this->_active)
 764                         return false;
 765
 766                 if ($this->_single_sock !== null) {
 767                         $this->_flush_read_buffer($this->_single_sock);
 768                         return $this->sock_to_host($this->_single_sock);
 769                 }
 770
 771                 $hv = is_array($key) ? intval($key[0]) : $this->_hashfunc($key);
 772
 773                 if ($this->_buckets === null)
 774                 {
 775                         foreach ($this->_servers as $v)
 776                         {
 777                                 if (is_array($v))
 778                                 {
 779                                         for ($i=0; $i<$v[1]; $i++)
 780                                                 $bu[] = $v[0];
 781                                 } else
 782                                 {
 783                                         $bu[] = $v;
 784                                 }
 785                         }
 786                         $this->_buckets = $bu;
 787                         $this->_bucketcount = count($bu);
 788                 }
 789
 790                 $realkey = is_array($key) ? $key[1] : $key;
 791                 for ($tries = 0; $tries<20; $tries++)
 792                 {
 793                         $host = $this->_buckets[$hv % $this->_bucketcount];
 794                         $sock = $this->sock_to_host($host);
 795                         if (is_resource($sock)) {
 796                                 $this->_flush_read_buffer($sock);
 797                                 return $sock;
 798                  }
 799                         $hv = $this->_hashfunc( $hv . $realkey );
 800                 }
 801
 802                 return false;
 803         }
 804
 805         // }}}
 806         // {{{ _hashfunc()
 807
 808         /**
 809          * Creates a hash integer       based on the $key
 810          *
 811          * @param   string   $key     Key to hash
 812          *
 813          * @return  integer  Hash value
 814          * @access  private
 815          */
 816         function _hashfunc ($key)
 817         {
 818                 # Hash function must on [0,0x7ffffff]
 819                 # We take the first 31 bits of the MD5 hash, which unlike the hash
 820                 # function used in a previous version of this client, works
 821                 return hexdec(substr(md5($key),0,8)) & 0x7fffffff;
 822         }
 823
 824         // }}}
 825         // {{{ _incrdecr()
 826
 827         /**
 828          * Perform increment/decriment on $key
 829          *
 830          * @param   string   $cmd     Command to perform
 831          * @param   string   $key     Key to perform it on
 832          * @param   integer  $amt     Amount to adjust
 833          *
 834          * @return  integer     New value of $key
 835          * @access  private
 836          */
 837         function _incrdecr ($cmd, $key, $amt=1)
 838         {
 839                 if (!$this->_active)
 840                         return null;
 841
 842                 $sock = $this->get_sock($key);
 843                 if (!is_resource($sock))
 844                         return null;
 845
 846                 $key = is_array($key) ? $key[1] : $key;
 847                 @$this->stats[$cmd]++;
 848                 if (!$this->_safe_fwrite($sock, "$cmd $key $amt\r\n"))
 849                         return $this->_dead_sock($sock);
 850
 851                 $line = fgets($sock);
 852                 $match = array();
 853                 if (!preg_match('/^(\d+)/', $line, $match))
 854                         return null;
 855                 return $match[1];
 856         }
 857
 858         // }}}
 859         // {{{ _load_items()
 860
 861         /**
 862          * Load items into $ret from $sock
 863          *
 864          * @param   resource $sock    Socket to read from
 865          * @param   array    $ret     Returned values
 866          *
 867          * @access  private
 868          */
 869         function _load_items ($sock, &$ret)
 870         {
 871                 while (1)
 872                 {
 873                         $decl = fgets($sock);
 874                         if ($decl == "END\r\n")
 875                         {
 876                                 return true;
 877                         } elseif (preg_match('/^VALUE (\S+) (\d+) (\d+)\r\n$/', $decl, $match))
 878                         {
 879                                 list($rkey, $flags, $len) = array($match[1], $match[2], $match[3]);
 880                                 $bneed = $len+2;
 881                                 $offset = 0;
 882
 883                                 while ($bneed > 0)
 884                                 {
 885                                         $data = fread($sock, $bneed);
 886                                         $n = strlen($data);
 887                                         if ($n == 0)
 888                                                 break;
 889                                         $offset += $n;
 890                                         $bneed -= $n;
 891                                         @$ret[$rkey] .= $data;
 892                                 }
 893
 894                                 if ($offset != $len+2)
 895                                 {
 896                                         // Something is borked!
 897                                         if ($this->_debug)
 898                                                 $this->_debugprint(sprintf("Something is borked!  key %s expecting %d got %d length\n", $rkey, $len+2, $offset));
 899
 900                                         unset($ret[$rkey]);
 901                                         $this->_close_sock($sock);
 902                                         return false;
 903                                 }
 904
 905                                 if ($this->_have_zlib && $flags & self::COMPRESSED)
 906                                         $ret[$rkey] = gzuncompress($ret[$rkey]);
 907
 908                                 $ret[$rkey] = rtrim($ret[$rkey]);
 909
 910                                 if ($flags & self::SERIALIZED)
 911                                         $ret[$rkey] = unserialize($ret[$rkey]);
 912
 913                         } else
 914                         {
 915                                 $this->_debugprint("Error parsing memcached response\n");
 916                                 return 0;
 917                         }
 918                 }
 919         }
 920
 921         // }}}
 922         // {{{ _set()
 923
 924         /**
 925          * Performs the requested storage operation to the memcache server
 926          *
 927          * @param   string   $cmd     Command to perform
 928          * @param   string   $key     Key to act on
 929          * @param   mixed    $val     What we need to store
 930          * @param   integer  $exp     When it should expire
 931          *
 932          * @return  boolean
 933          * @access  private
 934          */
 935         function _set ($cmd, $key, $val, $exp)
 936         {
 937                 if (!$this->_active)
 938                         return false;
 939
 940                 $sock = $this->get_sock($key);
 941                 if (!is_resource($sock))
 942                         return false;
 943
 944                 @$this->stats[$cmd]++;
 945
 946                 $flags = 0;
 947
 948                 if (!is_scalar($val))
 949                 {
 950                         $val = serialize($val);
 951                         $flags |= self::SERIALIZED;
 952                         if ($this->_debug)
 953                                 $this->_debugprint(sprintf("client: serializing data as it is not scalar\n"));
 954                 }
 955
 956                 $len = strlen($val);
 957
 958                 if ($this->_have_zlib && $this->_compress_enable &&
 959                          $this->_compress_threshold && $len >= $this->_compress_threshold)
 960                 {
 961                         $c_val = gzcompress($val, 9);
 962                         $c_len = strlen($c_val);
 963
 964                         if ($c_len < $len*(1 - self::COMPRESSION_SAVINGS))
 965                         {
 966                                 if ($this->_debug)
 967                                         $this->_debugprint(sprintf("client: compressing data; was %d bytes is now %d bytes\n", $len, $c_len));
 968                                 $val = $c_val;
 969                                 $len = $c_len;
 970                                 $flags |= self::COMPRESSED;
 971                         }
 972                 }
 973                 if (!$this->_safe_fwrite($sock, "$cmd $key $flags $exp $len\r\n$val\r\n"))
 974                         return $this->_dead_sock($sock);
 975
 976                 $line = trim(fgets($sock));
 977
 978                 if ($this->_debug)
 979                 {
 980                         $this->_debugprint(sprintf("%s %s (%s)\n", $cmd, $key, $line));
 981                 }
 982                 if ($line == "STORED")
 983                         return true;
 984                 return false;
 985         }
 986
 987         // }}}
 988         // {{{ sock_to_host()
 989
 990         /**
 991          * Returns the socket for the host
 992          *
 993          * @param   string   $host    Host:IP to get socket for
 994          *
 995          * @return  mixed    IO Stream or false
 996          * @access  private
 997          */
 998         function sock_to_host ($host)
 999         {
1000                 if (isset($this->_cache_sock[$host]))
1001                         return $this->_cache_sock[$host];
1002
1003                 $sock = null;
1004                 $now = time();
1005                 list ($ip, /* $port */) = explode (":", $host);
1006                 if (isset($this->_host_dead[$host]) && $this->_host_dead[$host] > $now ||
1007                         isset($this->_host_dead[$ip]) && $this->_host_dead[$ip] > $now)
1008                         return null;
1009
1010                 if (!$this->_connect_sock($sock, $host))
1011                         return $this->_dead_host($host);
1012
1013                 // Do not buffer writes
1014                 stream_set_write_buffer($sock, 0);
1015
1016                 $this->_cache_sock[$host] = $sock;
1017
1018                 return $this->_cache_sock[$host];
1019         }
1020
1021         function _debugprint($str){
1022                 print($str);
1023         }
1024
1025         /**
1026          * Write to a stream, timing out after the correct amount of time
1027          *
1028          * @return bool false on failure, true on success
1029          */
1030         /*
1031         function _safe_fwrite($f, $buf, $len = false) {
1032                 stream_set_blocking($f, 0);
1033
1034                 if ($len === false) {
1035                         wfDebug("Writing " . strlen( $buf ) . " bytes\n");
1036                         $bytesWritten = fwrite($f, $buf);
1037                 } else {
1038                         wfDebug("Writing $len bytes\n");
1039                         $bytesWritten = fwrite($f, $buf, $len);
1040                 }
1041                 $n = stream_select($r=NULL, $w = array($f), $e = NULL, 10, 0);
1042                 #   $this->_timeout_seconds, $this->_timeout_microseconds);
1043
1044                 wfDebug("stream_select returned $n\n");
1045                 stream_set_blocking($f, 1);
1046                 return $n == 1;
1047                 return $bytesWritten;
1048         }*/
1049
1050         /**
1051          * Original behaviour
1052          */
1053         function _safe_fwrite($f, $buf, $len = false) {
1054                 if ($len === false) {
1055                         $bytesWritten = fwrite($f, $buf);
1056                 } else {
1057                         $bytesWritten = fwrite($f, $buf, $len);
1058                 }
1059                 return $bytesWritten;
1060         }
1061
1062         /**
1063          * Flush the read buffer of a stream
1064          */
1065         function _flush_read_buffer($f) {
1066                 if (!is_resource($f)) {
1067                         return;
1068                 }
1069                 $n = stream_select($r=array($f), $w = NULL, $e = NULL, 0, 0);
1070                 while ($n == 1 && !feof($f)) {
1071                         fread($f, 1024);
1072                         $n = stream_select($r=array($f), $w = NULL, $e = NULL, 0, 0);
1073                 }
1074         }
1075
1076         // }}}
1077         // }}}
1078         // }}}
1079 }
1080
1081 // vim: sts=3 sw=3 et
1082
1083 // }}}
1084
1085 class MemCachedClientforWiki extends MWMemcached {
1086         function _debugprint( $text ) {
1087                 wfDebug( "memcached: $text" );
1088         }
1089 }