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->_active) {
 406          wfProfileOut( $fname );
 407          return false;
 408       }
 409
 410       $sock = $this->get_sock($key);
 411
 412       if (!is_resource($sock)) {
 413          wfProfileOut( $fname );
 414          return false;
 415       }
 416
 417       @$this->stats['get']++;
 418
 419       $cmd = "get $key\r\n";
 420       if (!$this->_safe_fwrite($sock, $cmd, strlen($cmd)))
 421       {
 422          $this->_dead_sock($sock);
 423          wfProfileOut( $fname );
 424          return false;
 425       }
 426
 427       $val = array();
 428       $this->_load_items($sock, $val);
 429
 430       if ($this->_debug)
 431          foreach ($val as $k => $v)
 432             $this->_debugprint(@sprintf("MemCache: sock %s got %s => %s\r\n", serialize($sock), $k, $v));
 433
 434       wfProfileOut( $fname );
 435       return @$val[$key];
 436    }
 437
 438    // }}}
 439    // {{{ get_multi()
 440
 441    /**
 442     * Get multiple keys from the server(s)
 443     *
 444     * @param   array    $keys    Keys to retrieve
 445     *
 446     * @return  array
 447     * @access  public
 448     */
 449    function get_multi ($keys)
 450    {
 451       if (!$this->_active)
 452          return false;
 453
 454       $this->stats['get_multi']++;
 455       $sock_keys = array();
 456
 457       foreach ($keys as $key)
 458       {
 459          $sock = $this->get_sock($key);
 460          if (!is_resource($sock)) continue;
 461          $key = is_array($key) ? $key[1] : $key;
 462          if (!isset($sock_keys[$sock]))
 463          {
 464             $sock_keys[$sock] = array();
 465             $socks[] = $sock;
 466          }
 467          $sock_keys[$sock][] = $key;
 468       }
 469
 470       // Send out the requests
 471       foreach ($socks as $sock)
 472       {
 473          $cmd = "get";
 474          foreach ($sock_keys[$sock] as $key)
 475          {
 476             $cmd .= " ". $key;
 477          }
 478          $cmd .= "\r\n";
 479
 480          if ($this->_safe_fwrite($sock, $cmd, strlen($cmd)))
 481          {
 482             $gather[] = $sock;
 483          } else
 484          {
 485             $this->_dead_sock($sock);
 486          }
 487       }
 488
 489       // Parse responses
 490       $val = array();
 491       foreach ($gather as $sock)
 492       {
 493          $this->_load_items($sock, $val);
 494       }
 495
 496       if ($this->_debug)
 497          foreach ($val as $k => $v)
 498             $this->_debugprint(sprintf("MemCache: got %s => %s\r\n", $k, $v));
 499
 500       return $val;
 501    }
 502
 503    // }}}
 504    // {{{ incr()
 505
 506    /**
 507     * Increments $key (optionally) by $amt
 508     *
 509     * @param   string   $key     Key to increment
 510     * @param   integer  $amt     (optional) amount to increment
 511     *
 512     * @return  integer  New key value?
 513     * @access  public
 514     */
 515    function incr ($key, $amt=1)
 516    {
 517       return $this->_incrdecr('incr', $key, $amt);
 518    }
 519
 520    // }}}
 521    // {{{ replace()
 522
 523    /**
 524     * Overwrites an existing value for key; only works if key is already set
 525     *
 526     * @param   string   $key     Key to set value as
 527     * @param   mixed    $value   Value to store
 528     * @param   integer  $exp     (optional) Experiation time
 529     *
 530     * @return  boolean
 531     * @access  public
 532     */
 533    function replace ($key, $value, $exp=0)
 534    {
 535       return $this->_set('replace', $key, $value, $exp);
 536    }
 537
 538    // }}}
 539    // {{{ run_command()
 540
 541    /**
 542     * Passes through $cmd to the memcache server connected by $sock; returns
 543     * output as an array (null array if no output)
 544     *
 545     * NOTE: due to a possible bug in how PHP reads while using fgets(), each
 546     *       line may not be terminated by a \r\n.  More specifically, my testing
 547     *       has shown that, on FreeBSD at least, each line is terminated only
 548     *       with a \n.  This is with the PHP flag auto_detect_line_endings set
 549     *       to falase (the default).
 550     *
 551     * @param   resource $sock    Socket to send command on
 552     * @param   string   $cmd     Command to run
 553     *
 554     * @return  array    Output array
 555     * @access  public
 556     */
 557    function run_command ($sock, $cmd)
 558    {
 559       if (!is_resource($sock))
 560          return array();
 561
 562       if (!$this->_safe_fwrite($sock, $cmd, strlen($cmd)))
 563          return array();
 564
 565       while (true)
 566       {
 567          $res = fgets($sock);
 568          $ret[] = $res;
 569          if (preg_match('/^END/', $res))
 570             break;
 571          if (strlen($res) == 0)
 572             break;
 573       }
 574       return $ret;
 575    }
 576
 577    // }}}
 578    // {{{ set()
 579
 580    /**
 581     * Unconditionally sets a key to a given value in the memcache.  Returns true
 582     * if set successfully.
 583     *
 584     * @param   string   $key     Key to set value as
 585     * @param   mixed    $value   Value to set
 586     * @param   integer  $exp     (optional) Experiation time
 587     *
 588     * @return  boolean  TRUE on success
 589     * @access  public
 590     */
 591    function set ($key, $value, $exp=0)
 592    {
 593       return $this->_set('set', $key, $value, $exp);
 594    }
 595
 596    // }}}
 597    // {{{ set_compress_threshold()
 598
 599    /**
 600     * Sets the compression threshold
 601     *
 602     * @param   integer  $thresh  Threshold to compress if larger than
 603     *
 604     * @access  public
 605     */
 606    function set_compress_threshold ($thresh)
 607    {
 608       $this->_compress_threshold = $thresh;
 609    }
 610
 611    // }}}
 612    // {{{ set_debug()
 613
 614    /**
 615     * Sets the debug flag
 616     *
 617     * @param   boolean  $dbg     TRUE for debugging, FALSE otherwise
 618     *
 619     * @access  public
 620     *
 621     * @see     memcahced::memcached
 622     */
 623    function set_debug ($dbg)
 624    {
 625       $this->_debug = $dbg;
 626    }
 627
 628    // }}}
 629    // {{{ set_servers()
 630
 631    /**
 632     * Sets the server list to distribute key gets and puts between
 633     *
 634     * @param   array    $list    Array of servers to connect to
 635     *
 636     * @access  public
 637     *
 638     * @see     memcached::memcached()
 639     */
 640    function set_servers ($list)
 641    {
 642       $this->_servers = $list;
 643       $this->_active = count($list);
 644       $this->_buckets = null;
 645       $this->_bucketcount = 0;
 646
 647       $this->_single_sock = null;
 648       if ($this->_active == 1)
 649          $this->_single_sock = $this->_servers[0];
 650    }
 651
 652    /**
 653     * Sets the timeout for new connections
 654     *
 655     * @param   integer  $seconds Number of seconds
 656     * @param   integer  $microseconds  Number of microseconds
 657     *
 658     * @access  public
 659     */
 660    function set_timeout ($seconds, $microseconds)
 661    {
 662       $this->_timeout_seconds = $seconds;
 663       $this->_timeout_microseconds = $microseconds;
 664    }
 665
 666    // }}}
 667    // }}}
 668    // {{{ private methods
 669    // {{{ _close_sock()
 670
 671    /**
 672     * Close the specified socket
 673     *
 674     * @param   string   $sock    Socket to close
 675     *
 676     * @access  private
 677     */
 678    function _close_sock ($sock)
 679    {
 680       $host = array_search($sock, $this->_cache_sock);
 681       fclose($this->_cache_sock[$host]);
 682       unset($this->_cache_sock[$host]);
 683    }
 684
 685    // }}}
 686    // {{{ _connect_sock()
 687
 688    /**
 689     * Connects $sock to $host, timing out after $timeout
 690     *
 691     * @param   integer  $sock    Socket to connect
 692     * @param   string   $host    Host:IP to connect to
 693     *
 694     * @return  boolean
 695     * @access  private
 696     */
 697    function _connect_sock (&$sock, $host)
 698    {
 699       list ($ip, $port) = explode(":", $host);
 700       $sock = false;
 701       $timeout = $this->_connect_timeout;
 702       $errno = $errstr = null;
 703       for ($i = 0; !$sock && $i < $this->_connect_attempts; $i++) {
 704          if ($i > 0) {
 705             # Sleep until the timeout, in case it failed fast
 706             $elapsed = microtime(true) - $t;
 707             if ( $elapsed < $timeout ) {
 708                usleep(($timeout - $elapsed) * 1e6);
 709             }
 710             $timeout *= 2;
 711          }
 712          $t = microtime(true);
 713          if ($this->_persistant == 1)
 714          {
 715             $sock = @pfsockopen($ip, $port, $errno, $errstr, $timeout);
 716          } else
 717          {
 718             $sock = @fsockopen($ip, $port, $errno, $errstr, $timeout);
 719          }
 720       }
 721       if (!$sock) {
 722          if ($this->_debug)
 723             $this->_debugprint( "Error connecting to $host: $errstr\n" );
 724          return false;
 725       }
 726
 727       // Initialise timeout
 728       stream_set_timeout($sock, $this->_timeout_seconds, $this->_timeout_microseconds);
 729
 730       return true;
 731    }
 732
 733    // }}}
 734    // {{{ _dead_sock()
 735
 736    /**
 737     * Marks a host as dead until 30-40 seconds in the future
 738     *
 739     * @param   string   $sock    Socket to mark as dead
 740     *
 741     * @access  private
 742     */
 743    function _dead_sock ($sock)
 744    {
 745       $host = array_search($sock, $this->_cache_sock);
 746       @list ($ip, /* $port */) = explode(":", $host);
 747       $this->_host_dead[$ip] = time() + 30 + intval(rand(0, 10));
 748       $this->_host_dead[$host] = $this->_host_dead[$ip];
 749       unset($this->_cache_sock[$host]);
 750    }
 751
 752    // }}}
 753    // {{{ get_sock()
 754
 755    /**
 756     * get_sock
 757     *
 758     * @param   string   $key     Key to retrieve value for;
 759     *
 760     * @return  mixed    resource on success, false on failure
 761     * @access  private
 762     */
 763    function get_sock ($key)
 764    {
 765       if (!$this->_active)
 766          return false;
 767
 768       if ($this->_single_sock !== null) {
 769          $this->_flush_read_buffer($this->_single_sock);
 770          return $this->sock_to_host($this->_single_sock);
 771       }
 772
 773       $hv = is_array($key) ? intval($key[0]) : $this->_hashfunc($key);
 774
 775       if ($this->_buckets === null)
 776       {
 777          foreach ($this->_servers as $v)
 778          {
 779             if (is_array($v))
 780             {
 781                for ($i=0; $i<$v[1]; $i++)
 782                   $bu[] = $v[0];
 783             } else
 784             {
 785                $bu[] = $v;
 786             }
 787          }
 788          $this->_buckets = $bu;
 789          $this->_bucketcount = count($bu);
 790       }
 791
 792       $realkey = is_array($key) ? $key[1] : $key;
 793       for ($tries = 0; $tries<20; $tries++)
 794       {
 795          $host = $this->_buckets[$hv % $this->_bucketcount];
 796          $sock = $this->sock_to_host($host);
 797          if (is_resource($sock)) {
 798             $this->_flush_read_buffer($sock);
 799             return $sock;
 800          }
 801          $hv += $this->_hashfunc($tries . $realkey);
 802       }
 803
 804       return false;
 805    }
 806
 807    // }}}
 808    // {{{ _hashfunc()
 809
 810    /**
 811     * Creates a hash integer  based on the $key
 812     *
 813     * @param   string   $key     Key to hash
 814     *
 815     * @return  integer  Hash value
 816     * @access  private
 817     */
 818    function _hashfunc ($key)
 819    {
 820       # Hash function must on [0,0x7ffffff]
 821       # We take the first 31 bits of the MD5 hash, which unlike the hash
 822       # function used in a previous version of this client, works
 823       return hexdec(substr(md5($key),0,8)) & 0x7fffffff;
 824    }
 825
 826    // }}}
 827    // {{{ _incrdecr()
 828
 829    /**
 830     * Perform increment/decriment on $key
 831     *
 832     * @param   string   $cmd     Command to perform
 833     * @param   string   $key     Key to perform it on
 834     * @param   integer  $amt     Amount to adjust
 835     *
 836     * @return  integer     New value of $key
 837     * @access  private
 838     */
 839    function _incrdecr ($cmd, $key, $amt=1)
 840    {
 841       if (!$this->_active)
 842          return null;
 843
 844       $sock = $this->get_sock($key);
 845       if (!is_resource($sock))
 846          return null;
 847
 848       $key = is_array($key) ? $key[1] : $key;
 849       @$this->stats[$cmd]++;
 850       if (!$this->_safe_fwrite($sock, "$cmd $key $amt\r\n"))
 851          return $this->_dead_sock($sock);
 852
 853       stream_set_timeout($sock, 1, 0);
 854       $line = fgets($sock);
 855       $match = array();
 856       if (!preg_match('/^(\d+)/', $line, $match))
 857          return null;
 858       return $match[1];
 859    }
 860
 861    // }}}
 862    // {{{ _load_items()
 863
 864    /**
 865     * Load items into $ret from $sock
 866     *
 867     * @param   resource $sock    Socket to read from
 868     * @param   array    $ret     Returned values
 869     *
 870     * @access  private
 871     */
 872    function _load_items ($sock, &$ret)
 873    {
 874       while (1)
 875       {
 876          $decl = fgets($sock);
 877          if ($decl == "END\r\n")
 878          {
 879             return true;
 880          } elseif (preg_match('/^VALUE (\S+) (\d+) (\d+)\r\n$/', $decl, $match))
 881          {
 882             list($rkey, $flags, $len) = array($match[1], $match[2], $match[3]);
 883             $bneed = $len+2;
 884             $offset = 0;
 885
 886             while ($bneed > 0)
 887             {
 888                $data = fread($sock, $bneed);
 889                $n = strlen($data);
 890                if ($n == 0)
 891                   break;
 892                $offset += $n;
 893                $bneed -= $n;
 894                @$ret[$rkey] .= $data;
 895             }
 896
 897             if ($offset != $len+2)
 898             {
 899                // Something is borked!
 900                if ($this->_debug)
 901                   $this->_debugprint(sprintf("Something is borked!  key %s expecting %d got %d length\n", $rkey, $len+2, $offset));
 902
 903                unset($ret[$rkey]);
 904                $this->_close_sock($sock);
 905                return false;
 906             }
 907
 908             if ($this->_have_zlib && $flags & memcached::COMPRESSED)
 909                $ret[$rkey] = gzuncompress($ret[$rkey]);
 910
 911             $ret[$rkey] = rtrim($ret[$rkey]);
 912
 913             if ($flags & memcached::SERIALIZED)
 914                $ret[$rkey] = unserialize($ret[$rkey]);
 915
 916          } else
 917          {
 918             $this->_debugprint("Error parsing memcached response\n");
 919             return 0;
 920          }
 921       }
 922    }
 923
 924    // }}}
 925    // {{{ _set()
 926
 927    /**
 928     * Performs the requested storage operation to the memcache server
 929     *
 930     * @param   string   $cmd     Command to perform
 931     * @param   string   $key     Key to act on
 932     * @param   mixed    $val     What we need to store
 933     * @param   integer  $exp     When it should expire
 934     *
 935     * @return  boolean
 936     * @access  private
 937     */
 938    function _set ($cmd, $key, $val, $exp)
 939    {
 940       if (!$this->_active)
 941          return false;
 942
 943       $sock = $this->get_sock($key);
 944       if (!is_resource($sock))
 945          return false;
 946
 947       @$this->stats[$cmd]++;
 948
 949       $flags = 0;
 950
 951       if (!is_scalar($val))
 952       {
 953          $val = serialize($val);
 954          $flags |= memcached::SERIALIZED;
 955          if ($this->_debug)
 956             $this->_debugprint(sprintf("client: serializing data as it is not scalar\n"));
 957       }
 958
 959       $len = strlen($val);
 960
 961       if ($this->_have_zlib && $this->_compress_enable &&
 962           $this->_compress_threshold && $len >= $this->_compress_threshold)
 963       {
 964          $c_val = gzcompress($val, 9);
 965          $c_len = strlen($c_val);
 966
 967          if ($c_len < $len*(1 - memcached::COMPRESSION_SAVINGS))
 968          {
 969             if ($this->_debug)
 970                $this->_debugprint(sprintf("client: compressing data; was %d bytes is now %d bytes\n", $len, $c_len));
 971             $val = $c_val;
 972             $len = $c_len;
 973             $flags |= memcached::COMPRESSED;
 974          }
 975       }
 976       if (!$this->_safe_fwrite($sock, "$cmd $key $flags $exp $len\r\n$val\r\n"))
 977          return $this->_dead_sock($sock);
 978
 979       $line = trim(fgets($sock));
 980
 981       if ($this->_debug)
 982       {
 983          if ($flags & memcached::COMPRESSED)
 984             $val = 'compressed data';
 985          $this->_debugprint(sprintf("MemCache: %s %s => %s (%s)\n", $cmd, $key, $val, $line));
 986       }
 987       if ($line == "STORED")
 988          return true;
 989       return false;
 990    }
 991
 992    // }}}
 993    // {{{ sock_to_host()
 994
 995    /**
 996     * Returns the socket for the host
 997     *
 998     * @param   string   $host    Host:IP to get socket for
 999     *
1000     * @return  mixed    IO Stream or false
1001     * @access  private
1002     */
1003    function sock_to_host ($host)
1004    {
1005       if (isset($this->_cache_sock[$host]))
1006          return $this->_cache_sock[$host];
1007
1008       $sock = null;
1009       $now = time();
1010       list ($ip, /* $port */) = explode (":", $host);
1011       if (isset($this->_host_dead[$host]) && $this->_host_dead[$host] > $now ||
1012           isset($this->_host_dead[$ip]) && $this->_host_dead[$ip] > $now)
1013          return null;
1014
1015       if (!$this->_connect_sock($sock, $host))
1016          return $this->_dead_sock($host);
1017
1018       // Do not buffer writes
1019       stream_set_write_buffer($sock, 0);
1020
1021       $this->_cache_sock[$host] = $sock;
1022
1023       return $this->_cache_sock[$host];
1024    }
1025
1026    function _debugprint($str){
1027       print($str);
1028    }
1029
1030    /**
1031     * Write to a stream, timing out after the correct amount of time
1032     *
1033     * @return bool false on failure, true on success
1034     */
1035     /*
1036    function _safe_fwrite($f, $buf, $len = false) {
1037       stream_set_blocking($f, 0);
1038
1039       if ($len === false) {
1040          wfDebug("Writing " . strlen( $buf ) . " bytes\n");
1041          $bytesWritten = fwrite($f, $buf);
1042       } else {
1043          wfDebug("Writing $len bytes\n");
1044          $bytesWritten = fwrite($f, $buf, $len);
1045       }
1046       $n = stream_select($r=NULL, $w = array($f), $e = NULL, 10, 0);
1047       #   $this->_timeout_seconds, $this->_timeout_microseconds);
1048
1049       wfDebug("stream_select returned $n\n");
1050       stream_set_blocking($f, 1);
1051       return $n == 1;
1052       return $bytesWritten;
1053    }*/
1054
1055    /**
1056     * Original behaviour
1057     */
1058    function _safe_fwrite($f, $buf, $len = false) {
1059       if ($len === false) {
1060          $bytesWritten = fwrite($f, $buf);
1061       } else {
1062          $bytesWritten = fwrite($f, $buf, $len);
1063       }
1064       return $bytesWritten;
1065    }
1066
1067    /**
1068     * Flush the read buffer of a stream
1069     */
1070    function _flush_read_buffer($f) {
1071       if (!is_resource($f)) {
1072          return;
1073       }
1074       $n = stream_select($r=array($f), $w = NULL, $e = NULL, 0, 0);
1075       while ($n == 1 && !feof($f)) {
1076          fread($f, 1024);
1077          $n = stream_select($r=array($f), $w = NULL, $e = NULL, 0, 0);
1078       }
1079    }
1080
1081    // }}}
1082    // }}}
1083    // }}}
1084 }
1085
1086 // vim: sts=3 sw=3 et
1087
1088 // }}}
1089