includes/memcached-client.php

   1 <?php
   2 //
   3 // +---------------------------------------------------------------------------+
   4 // | memcached client, PHP                                                     |
   5 // +---------------------------------------------------------------------------+
   6 // | Copyright (c) 2003 Ryan T. Dean <rtdean@cytherianage.net>                 |
   7 // | All rights reserved.                                                      |
   8 // |                                                                           |
   9 // | Redistribution and use in source and binary forms, with or without        |
  10 // | modification, are permitted provided that the following conditions        |
  11 // | are met:                                                                  |
  12 // |                                                                           |
  13 // | 1. Redistributions of source code must retain the above copyright         |
  14 // |    notice, this list of conditions and the following disclaimer.          |
  15 // | 2. Redistributions in binary form must reproduce the above copyright      |
  16 // |    notice, this list of conditions and the following disclaimer in the    |
  17 // |    documentation and/or other materials provided with the distribution.   |
  18 // |                                                                           |
  19 // | THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR      |
  20 // | IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES |
  21 // | OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.   |
  22 // | IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,          |
  23 // | INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT  |
  24 // | NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
  25 // | DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY     |
  26 // | THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT       |
  27 // | (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF  |
  28 // | THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.         |
  29 // +---------------------------------------------------------------------------+
  30 // | Author: Ryan T. Dean <rtdean@cytherianage.net>                            |
  31 // | Heavily influenced by the Perl memcached client by Brad Fitzpatrick.      |
  32 // |   Permission granted by Brad Fitzpatrick for relicense of ported Perl     |
  33 // |   client logic under 2-clause BSD license.                                |
  34 // +---------------------------------------------------------------------------+
  35 //
  36 // $TCAnet$
  37 //
  38
  39 /**
  40  * This is the PHP client for memcached - a distributed memory cache daemon.
  41  * More information is available at http://www.danga.com/memcached/
  42  *
  43  * Usage example:
  44  *
  45  * require_once 'memcached.php';
  46  *
  47  * $mc = new MWMemcached(array(
  48  *              'servers' => array('127.0.0.1:10000',
  49  *                                 array('192.0.0.1:10010', 2),
  50  *                                 '127.0.0.1:10020'),
  51  *              'debug'   => false,
  52  *              'compress_threshold' => 10240,
  53  *              'persistant' => true));
  54  *
  55  * $mc->add('key', array('some', 'array'));
  56  * $mc->replace('key', 'some random string');
  57  * $val = $mc->get('key');
  58  *
  59  * @author  Ryan T. Dean <rtdean@cytherianage.net>
  60  * @version 0.1.2
  61  */
  62
  63 // {{{ requirements
  64 // }}}
  65
  66 // {{{ class MWMemcached
  67 /**
  68  * memcached client class implemented using (p)fsockopen()
  69  *
  70  * @author  Ryan T. Dean <rtdean@cytherianage.net>
  71  * @ingroup Cache
  72  */
  73 class MWMemcached
  74 {
  75    // {{{ properties
  76    // {{{ public
  77
  78         // {{{ constants
  79         // {{{ flags
  80
  81         /**
  82          * Flag: indicates data is serialized
  83          */
  84         const SERIALIZED = 1;
  85
  86         /**
  87          * Flag: indicates data is compressed
  88          */
  89         const COMPRESSED = 2;
  90
  91         // }}}
  92
  93         /**
  94          * Minimum savings to store data compressed
  95          */
  96         const COMPRESSION_SAVINGS = 0.20;
  97
  98         // }}}
  99
 100
 101         /**
 102          * Command statistics
 103          *
 104          * @var     array
 105          * @access  public
 106          */
 107         var $stats;
 108
 109         // }}}
 110         // {{{ private
 111
 112         /**
 113          * Cached Sockets that are connected
 114          *
 115          * @var     array
 116          * @access  private
 117          */
 118         var $_cache_sock;
 119
 120         /**
 121          * Current debug status; 0 - none to 9 - profiling
 122          *
 123          * @var     boolean
 124          * @access  private
 125          */
 126         var $_debug;
 127
 128         /**
 129          * Dead hosts, assoc array, 'host'=>'unixtime when ok to check again'
 130          *
 131          * @var     array
 132          * @access  private
 133          */
 134         var $_host_dead;
 135
 136         /**
 137          * Is compression available?
 138          *
 139          * @var     boolean
 140          * @access  private
 141          */
 142         var $_have_zlib;
 143
 144         /**
 145          * Do we want to use compression?
 146          *
 147          * @var     boolean
 148          * @access  private
 149          */
 150         var $_compress_enable;
 151
 152         /**
 153          * At how many bytes should we compress?
 154          *
 155          * @var     integer
 156          * @access  private
 157          */
 158         var $_compress_threshold;
 159
 160         /**
 161          * Are we using persistant links?
 162          *
 163          * @var     boolean
 164          * @access  private
 165          */
 166         var $_persistant;
 167
 168         /**
 169          * If only using one server; contains ip:port to connect to
 170          *
 171          * @var     string
 172          * @access  private
 173          */
 174         var $_single_sock;
 175
 176         /**
 177          * Array containing ip:port or array(ip:port, weight)
 178          *
 179          * @var     array
 180          * @access  private
 181          */
 182         var $_servers;
 183
 184         /**
 185          * Our bit buckets
 186          *
 187          * @var     array
 188          * @access  private
 189          */
 190         var $_buckets;
 191
 192         /**
 193          * Total # of bit buckets we have
 194          *
 195          * @var     integer
 196          * @access  private
 197          */
 198         var $_bucketcount;
 199
 200         /**
 201          * # of total servers we have
 202          *
 203          * @var     integer
 204          * @access  private
 205          */
 206         var $_active;
 207
 208         /**
 209          * Stream timeout in seconds. Applies for example to fread()
 210          *
 211          * @var     integer
 212          * @access  private
 213          */
 214         var $_timeout_seconds;
 215
 216         /**
 217          * Stream timeout in microseconds
 218          *
 219          * @var     integer
 220          * @access  private
 221          */
 222         var $_timeout_microseconds;
 223
 224         /**
 225          * Connect timeout in seconds
 226          */
 227         var $_connect_timeout;
 228
 229         /**
 230          * Number of connection attempts for each server
 231          */
 232         var $_connect_attempts;
 233
 234         // }}}
 235         // }}}
 236         // {{{ methods
 237         // {{{ public functions
 238         // {{{ memcached()
 239
 240         /**
 241          * Memcache initializer
 242          *
 243          * @param   array    $args    Associative array of settings
 244          *
 245          * @return  mixed
 246          * @access  public
 247          */
 248         function __construct ($args)
 249         {
 250                 $this->set_servers(@$args['servers']);
 251                 $this->_debug = @$args['debug'];
 252                 $this->stats = array();
 253                 $this->_compress_threshold = @$args['compress_threshold'];
 254                 $this->_persistant = array_key_exists('persistant', $args) ? (@$args['persistant']) : false;
 255                 $this->_compress_enable = true;
 256                 $this->_have_zlib = function_exists("gzcompress");
 257
 258                 $this->_cache_sock = array();
 259                 $this->_host_dead = array();
 260
 261                 $this->_timeout_seconds = 0;
 262                 $this->_timeout_microseconds = 50000;
 263
 264                 $this->_connect_timeout = 0.01;
 265                 $this->_connect_attempts = 2;
 266         }
 267
 268         // }}}
 269         // {{{ add()
 270
 271         /**
 272          * Adds a key/value to the memcache server if one isn't already set with
 273          * that key
 274          *
 275          * @param   string  $key     Key to set with data
 276          * @param   mixed   $val     Value to store
 277          * @param   integer $exp     (optional) Time to expire data at
 278          *
 279          * @return  boolean
 280          * @access  public
 281          */
 282         function add ($key, $val, $exp = 0)
 283         {
 284                 return $this->_set('add', $key, $val, $exp);
 285         }
 286
 287         // }}}
 288         // {{{ decr()
 289
 290         /**
 291          * Decriment a value stored on the memcache server
 292          *
 293          * @param   string   $key     Key to decriment
 294          * @param   integer  $amt     (optional) Amount to decriment
 295          *
 296          * @return  mixed    FALSE on failure, value on success
 297          * @access  public
 298          */
 299         function decr ($key, $amt=1)
 300         {
 301                 return $this->_incrdecr('decr', $key, $amt);
 302         }
 303
 304         // }}}
 305         // {{{ delete()
 306
 307         /**
 308          * Deletes a key from the server, optionally after $time
 309          *
 310          * @param   string   $key     Key to delete
 311          * @param   integer  $time    (optional) How long to wait before deleting
 312          *
 313          * @return  boolean  TRUE on success, FALSE on failure
 314          * @access  public
 315          */
 316         function delete ($key, $time = 0)
 317         {
 318                 if (!$this->_active)
 319                         return false;
 320
 321                 $sock = $this->get_sock($key);
 322                 if (!is_resource($sock))
 323                         return false;
 324
 325                 $key = is_array($key) ? $key[1] : $key;
 326
 327                 @$this->stats['delete']++;
 328                 $cmd = "delete $key $time\r\n";
 329                 if(!$this->_safe_fwrite($sock, $cmd, strlen($cmd)))
 330                 {
 331                         $this->_dead_sock($sock);
 332                         return false;
 333                 }
 334                 $res = trim(fgets($sock));
 335
 336                 if ($this->_debug)
 337                         $this->_debugprint(sprintf("MemCache: delete %s (%s)\n", $key, $res));
 338
 339                 if ($res == "DELETED")
 340                         return true;
 341                 return false;
 342         }
 343
 344         // }}}
 345         // {{{ disconnect_all()
 346
 347         /**
 348          * Disconnects all connected sockets
 349          *
 350          * @access  public
 351          */
 352         function disconnect_all ()
 353         {
 354                 foreach ($this->_cache_sock as $sock)
 355                         fclose($sock);
 356
 357                 $this->_cache_sock = array();
 358         }
 359
 360         // }}}
 361         // {{{ enable_compress()
 362
 363         /**
 364          * Enable / Disable compression
 365          *
 366          * @param   boolean  $enable  TRUE to enable, FALSE to disable
 367          *
 368          * @access  public
 369          */
 370         function enable_compress ($enable)
 371         {
 372                 $this->_compress_enable = $enable;
 373         }
 374
 375         // }}}
 376         // {{{ forget_dead_hosts()
 377
 378         /**
 379          * Forget about all of the dead hosts
 380          *
 381          * @access  public
 382          */
 383         function forget_dead_hosts ()
 384         {
 385                 $this->_host_dead = array();
 386         }
 387
 388         // }}}
 389         // {{{ get()
 390
 391         /**
 392          * Retrieves the value associated with the key from the memcache server
 393          *
 394          * @param  string   $key     Key to retrieve
 395          *
 396          * @return  mixed
 397          * @access  public
 398          */
 399         function get ($key)
 400         {
 401                 $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 ($this->_persistant == 1)
 708                         {
 709                                 $sock = @pfsockopen($ip, $port, $errno, $errstr, $timeout);
 710                         } else
 711                         {
 712                                 $sock = @fsockopen($ip, $port, $errno, $errstr, $timeout);
 713                         }
 714                 }
 715                 if (!$sock) {
 716                         if ($this->_debug)
 717                                 $this->_debugprint( "Error connecting to $host: $errstr\n" );
 718                         return false;
 719                 }
 720
 721                 // Initialise timeout
 722                 stream_set_timeout($sock, $this->_timeout_seconds, $this->_timeout_microseconds);
 723
 724                 return true;
 725         }
 726
 727         // }}}
 728         // {{{ _dead_sock()
 729
 730         /**
 731          * Marks a host as dead until 30-40 seconds in the future
 732          *
 733          * @param   string   $sock    Socket to mark as dead
 734          *
 735          * @access  private
 736          */
 737         function _dead_sock ($sock)
 738         {
 739                 $host = array_search($sock, $this->_cache_sock);
 740                 $this->_dead_host($host);
 741         }
 742
 743         function _dead_host ($host)
 744         {
 745                 @list ($ip, /* $port */) = explode(":", $host);
 746                 $this->_host_dead[$ip] = time() + 30 + intval(rand(0, 10));
 747                 $this->_host_dead[$host] = $this->_host_dead[$ip];
 748                 unset($this->_cache_sock[$host]);
 749         }
 750
 751         // }}}
 752         // {{{ get_sock()
 753
 754         /**
 755          * get_sock
 756          *
 757          * @param   string   $key     Key to retrieve value for;
 758          *
 759          * @return  mixed    resource on success, false on failure
 760          * @access  private
 761          */
 762         function get_sock ($key)
 763         {
 764                 if (!$this->_active)
 765                         return false;
 766
 767                 if ($this->_single_sock !== null) {
 768                         $this->_flush_read_buffer($this->_single_sock);
 769                         return $this->sock_to_host($this->_single_sock);
 770                 }
 771
 772                 $hv = is_array($key) ? intval($key[0]) : $this->_hashfunc($key);
 773
 774                 if ($this->_buckets === null)
 775                 {
 776                         foreach ($this->_servers as $v)
 777                         {
 778                                 if (is_array($v))
 779                                 {
 780                                         for ($i=0; $i<$v[1]; $i++)
 781                                                 $bu[] = $v[0];
 782                                 } else
 783                                 {
 784                                         $bu[] = $v;
 785                                 }
 786                         }
 787                         $this->_buckets = $bu;
 788                         $this->_bucketcount = count($bu);
 789                 }
 790
 791                 $realkey = is_array($key) ? $key[1] : $key;
 792                 for ($tries = 0; $tries<20; $tries++)
 793                 {
 794                         $host = $this->_buckets[$hv % $this->_bucketcount];
 795                         $sock = $this->sock_to_host($host);
 796                         if (is_resource($sock)) {
 797                                 $this->_flush_read_buffer($sock);
 798                                 return $sock;
 799                  }
 800                         $hv = $this->_hashfunc( $hv . $realkey );
 801                 }
 802
 803                 return false;
 804         }
 805
 806         // }}}
 807         // {{{ _hashfunc()
 808
 809         /**
 810          * Creates a hash integer       based on the $key
 811          *
 812          * @param   string   $key     Key to hash
 813          *
 814          * @return  integer  Hash value
 815          * @access  private
 816          */
 817         function _hashfunc ($key)
 818         {
 819                 # Hash function must on [0,0x7ffffff]
 820                 # We take the first 31 bits of the MD5 hash, which unlike the hash
 821                 # function used in a previous version of this client, works
 822                 return hexdec(substr(md5($key),0,8)) & 0x7fffffff;
 823         }
 824
 825         // }}}
 826         // {{{ _incrdecr()
 827
 828         /**
 829          * Perform increment/decriment on $key
 830          *
 831          * @param   string   $cmd     Command to perform
 832          * @param   string   $key     Key to perform it on
 833          * @param   integer  $amt     Amount to adjust
 834          *
 835          * @return  integer     New value of $key
 836          * @access  private
 837          */
 838         function _incrdecr ($cmd, $key, $amt=1)
 839         {
 840                 if (!$this->_active)
 841                         return null;
 842
 843                 $sock = $this->get_sock($key);
 844                 if (!is_resource($sock))
 845                         return null;
 846
 847                 $key = is_array($key) ? $key[1] : $key;
 848                 @$this->stats[$cmd]++;
 849                 if (!$this->_safe_fwrite($sock, "$cmd $key $amt\r\n"))
 850                         return $this->_dead_sock($sock);
 851
 852                 $line = fgets($sock);
 853                 $match = array();
 854                 if (!preg_match('/^(\d+)/', $line, $match))
 855                         return null;
 856                 return $match[1];
 857         }
 858
 859         // }}}
 860         // {{{ _load_items()
 861
 862         /**
 863          * Load items into $ret from $sock
 864          *
 865          * @param   resource $sock    Socket to read from
 866          * @param   array    $ret     Returned values
 867          *
 868          * @access  private
 869          */
 870         function _load_items ($sock, &$ret)
 871         {
 872                 while (1)
 873                 {
 874                         $decl = fgets($sock);
 875                         if ($decl == "END\r\n")
 876                         {
 877                                 return true;
 878                         } elseif (preg_match('/^VALUE (\S+) (\d+) (\d+)\r\n$/', $decl, $match))
 879                         {
 880                                 list($rkey, $flags, $len) = array($match[1], $match[2], $match[3]);
 881                                 $bneed = $len+2;
 882                                 $offset = 0;
 883
 884                                 while ($bneed > 0)
 885                                 {
 886                                         $data = fread($sock, $bneed);
 887                                         $n = strlen($data);
 888                                         if ($n == 0)
 889                                                 break;
 890                                         $offset += $n;
 891                                         $bneed -= $n;
 892                                         @$ret[$rkey] .= $data;
 893                                 }
 894
 895                                 if ($offset != $len+2)
 896                                 {
 897                                         // Something is borked!
 898                                         if ($this->_debug)
 899                                                 $this->_debugprint(sprintf("Something is borked!  key %s expecting %d got %d length\n", $rkey, $len+2, $offset));
 900
 901                                         unset($ret[$rkey]);
 902                                         $this->_close_sock($sock);
 903                                         return false;
 904                                 }
 905
 906                                 if ($this->_have_zlib && $flags & memcached::COMPRESSED)
 907                                         $ret[$rkey] = gzuncompress($ret[$rkey]);
 908
 909                                 $ret[$rkey] = rtrim($ret[$rkey]);
 910
 911                                 if ($flags & memcached::SERIALIZED)
 912                                         $ret[$rkey] = unserialize($ret[$rkey]);
 913
 914                         } else
 915                         {
 916                                 $this->_debugprint("Error parsing memcached response\n");
 917                                 return 0;
 918                         }
 919                 }
 920         }
 921
 922         // }}}
 923         // {{{ _set()
 924
 925         /**
 926          * Performs the requested storage operation to the memcache server
 927          *
 928          * @param   string   $cmd     Command to perform
 929          * @param   string   $key     Key to act on
 930          * @param   mixed    $val     What we need to store
 931          * @param   integer  $exp     When it should expire
 932          *
 933          * @return  boolean
 934          * @access  private
 935          */
 936         function _set ($cmd, $key, $val, $exp)
 937         {
 938                 if (!$this->_active)
 939                         return false;
 940
 941                 $sock = $this->get_sock($key);
 942                 if (!is_resource($sock))
 943                         return false;
 944
 945                 @$this->stats[$cmd]++;
 946
 947                 $flags = 0;
 948
 949                 if (!is_scalar($val))
 950                 {
 951                         $val = serialize($val);
 952                         $flags |= memcached::SERIALIZED;
 953                         if ($this->_debug)
 954                                 $this->_debugprint(sprintf("client: serializing data as it is not scalar\n"));
 955                 }
 956
 957                 $len = strlen($val);
 958
 959                 if ($this->_have_zlib && $this->_compress_enable &&
 960                          $this->_compress_threshold && $len >= $this->_compress_threshold)
 961                 {
 962                         $c_val = gzcompress($val, 9);
 963                         $c_len = strlen($c_val);
 964
 965                         if ($c_len < $len*(1 - memcached::COMPRESSION_SAVINGS))
 966                         {
 967                                 if ($this->_debug)
 968                                         $this->_debugprint(sprintf("client: compressing data; was %d bytes is now %d bytes\n", $len, $c_len));
 969                                 $val = $c_val;
 970                                 $len = $c_len;
 971                                 $flags |= memcached::COMPRESSED;
 972                         }
 973                 }
 974                 if (!$this->_safe_fwrite($sock, "$cmd $key $flags $exp $len\r\n$val\r\n"))
 975                         return $this->_dead_sock($sock);
 976
 977                 $line = trim(fgets($sock));
 978
 979                 if ($this->_debug)
 980                 {
 981                         $this->_debugprint(sprintf("%s %s (%s)\n", $cmd, $key, $line));
 982                 }
 983                 if ($line == "STORED")
 984                         return true;
 985                 return false;
 986         }
 987
 988         // }}}
 989         // {{{ sock_to_host()
 990
 991         /**
 992          * Returns the socket for the host
 993          *
 994          * @param   string   $host    Host:IP to get socket for
 995          *
 996          * @return  mixed    IO Stream or false
 997          * @access  private
 998          */
 999         function sock_to_host ($host)
1000         {
1001                 if (isset($this->_cache_sock[$host]))
1002                         return $this->_cache_sock[$host];
1003
1004                 $sock = null;
1005                 $now = time();
1006                 list ($ip, /* $port */) = explode (":", $host);
1007                 if (isset($this->_host_dead[$host]) && $this->_host_dead[$host] > $now ||
1008                         isset($this->_host_dead[$ip]) && $this->_host_dead[$ip] > $now)
1009                         return null;
1010
1011                 if (!$this->_connect_sock($sock, $host))
1012                         return $this->_dead_host($host);
1013
1014                 // Do not buffer writes
1015                 stream_set_write_buffer($sock, 0);
1016
1017                 $this->_cache_sock[$host] = $sock;
1018
1019                 return $this->_cache_sock[$host];
1020         }
1021
1022         function _debugprint($str){
1023                 print($str);
1024         }
1025
1026         /**
1027          * Write to a stream, timing out after the correct amount of time
1028          *
1029          * @return bool false on failure, true on success
1030          */
1031         /*
1032         function _safe_fwrite($f, $buf, $len = false) {
1033                 stream_set_blocking($f, 0);
1034
1035                 if ($len === false) {
1036                         wfDebug("Writing " . strlen( $buf ) . " bytes\n");
1037                         $bytesWritten = fwrite($f, $buf);
1038                 } else {
1039                         wfDebug("Writing $len bytes\n");
1040                         $bytesWritten = fwrite($f, $buf, $len);
1041                 }
1042                 $n = stream_select($r=NULL, $w = array($f), $e = NULL, 10, 0);
1043                 #   $this->_timeout_seconds, $this->_timeout_microseconds);
1044
1045                 wfDebug("stream_select returned $n\n");
1046                 stream_set_blocking($f, 1);
1047                 return $n == 1;
1048                 return $bytesWritten;
1049         }*/
1050
1051         /**
1052          * Original behaviour
1053          */
1054         function _safe_fwrite($f, $buf, $len = false) {
1055                 if ($len === false) {
1056                         $bytesWritten = fwrite($f, $buf);
1057                 } else {
1058                         $bytesWritten = fwrite($f, $buf, $len);
1059                 }
1060                 return $bytesWritten;
1061         }
1062
1063         /**
1064          * Flush the read buffer of a stream
1065          */
1066         function _flush_read_buffer($f) {
1067                 if (!is_resource($f)) {
1068                         return;
1069                 }
1070                 $n = stream_select($r=array($f), $w = NULL, $e = NULL, 0, 0);
1071                 while ($n == 1 && !feof($f)) {
1072                         fread($f, 1024);
1073                         $n = stream_select($r=array($f), $w = NULL, $e = NULL, 0, 0);
1074                 }
1075         }
1076
1077         // }}}
1078         // }}}
1079         // }}}
1080 }
1081
1082 // vim: sts=3 sw=3 et
1083
1084 // }}}
1085
1086 class MemCachedClientforWiki extends MWMemcached {
1087         function _debugprint( $text ) {
1088                 wfDebug( "memcached: $text" );
1089         }
1090 }