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