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