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