memcached-client.php

来自「php 开发的内容管理系统」· PHP 代码 · 共 1,061 行 · 第 1/2 页

    *       has shown that, on FreeBSD at least, each line is terminated only
    *       with a \n.  This is with the PHP flag auto_detect_line_endings set
    *       to falase (the default).
    *
    * @param   resource $sock    Socket to send command on
    * @param   string   $cmd     Command to run
    *
    * @return  array    Output array
    * @access  public
    */
   function run_command ($sock, $cmd)
   {
      if (!is_resource($sock))
         return array();

      if (!$this->_safe_fwrite($sock, $cmd, strlen($cmd)))
         return array();

      while (true)
      {
         $res = fgets($sock);
         $ret[] = $res;
         if (preg_match('/^END/', $res))
            break;
         if (strlen($res) == 0)
            break;
      }
      return $ret;
   }

   // }}}
   // {{{ set()

   /**
    * Unconditionally sets a key to a given value in the memcache.  Returns true
    * if set successfully.
    *
    * @param   string   $key     Key to set value as
    * @param   mixed    $value   Value to set
    * @param   interger $exp     (optional) Experiation time
    *
    * @return  boolean  TRUE on success
    * @access  public
    */
   function set ($key, $value, $exp=0)
   {
      return $this->_set('set', $key, $value, $exp);
   }

   // }}}
   // {{{ set_compress_threshold()

   /**
    * Sets the compression threshold
    *
    * @param   interger $thresh  Threshold to compress if larger than
    *
    * @access  public
    */
   function set_compress_threshold ($thresh)
   {
      $this->_compress_threshold = $thresh;
   }

   // }}}
   // {{{ set_debug()

   /**
    * Sets the debug flag
    *
    * @param   boolean  $dbg     TRUE for debugging, FALSE otherwise
    *
    * @access  public
    *
    * @see     memcahced::memcached
    */
   function set_debug ($dbg)
   {
      $this->_debug = $dbg;
   }

   // }}}
   // {{{ set_servers()

   /**
    * Sets the server list to distribute key gets and puts between
    *
    * @param   array    $list    Array of servers to connect to
    *
    * @access  public
    *
    * @see     memcached::memcached()
    */
   function set_servers ($list)
   {
      $this->_servers = $list;
      $this->_active = count($list);
      $this->_buckets = null;
      $this->_bucketcount = 0;

      $this->_single_sock = null;
      if ($this->_active == 1)
         $this->_single_sock = $this->_servers[0];
   }

   /**
    * Sets the timeout for new connections
    *
    * @param   integer  $seconds Number of seconds
    * @param   integer  $microseconds  Number of microseconds
    *
    * @access  public
    */
   function set_timeout ($seconds, $microseconds)
   {
      $this->_timeout_seconds = $seconds;
      $this->_timeout_microseconds = $microseconds;
   }

   // }}}
   // }}}
   // {{{ private methods
   // {{{ _close_sock()

   /**
    * Close the specified socket
    *
    * @param   string   $sock    Socket to close
    *
    * @access  private
    */
   function _close_sock ($sock)
   {
      $host = array_search($sock, $this->_cache_sock);
      fclose($this->_cache_sock[$host]);
      unset($this->_cache_sock[$host]);
   }

   // }}}
   // {{{ _connect_sock()

   /**
    * Connects $sock to $host, timing out after $timeout
    *
    * @param   interger $sock    Socket to connect
    * @param   string   $host    Host:IP to connect to
    * @param   float    $timeout (optional) Timeout value, defaults to 0.25s
    *
    * @return  boolean
    * @access  private
    */
   function _connect_sock (&$sock, $host, $timeout = 0.25)
   {
      list ($ip, $port) = explode(":", $host);
      if ($this->_persistant == 1)
      {
         $sock = @pfsockopen($ip, $port, $errno, $errstr, $timeout);
      } else
      {
         $sock = @fsockopen($ip, $port, $errno, $errstr, $timeout);
      }

      if (!$sock) {
         if ($this->_debug)
            $this->_debugprint( "Error connecting to $host: $errstr\n" );
         return false;
      }

      // Initialise timeout
      stream_set_timeout($sock, $this->_timeout_seconds, $this->_timeout_microseconds);

      return true;
   }

   // }}}
   // {{{ _dead_sock()

   /**
    * Marks a host as dead until 30-40 seconds in the future
    *
    * @param   string   $sock    Socket to mark as dead
    *
    * @access  private
    */
   function _dead_sock ($sock)
   {
      $host = array_search($sock, $this->_cache_sock);
      @list ($ip, $port) = explode(":", $host);
      $this->_host_dead[$ip] = time() + 30 + intval(rand(0, 10));
      $this->_host_dead[$host] = $this->_host_dead[$ip];
      unset($this->_cache_sock[$host]);
   }

   // }}}
   // {{{ get_sock()

   /**
    * get_sock
    *
    * @param   string   $key     Key to retrieve value for;
    *
    * @return  mixed    resource on success, false on failure
    * @access  private
    */
   function get_sock ($key)
   {
      if (!$this->_active)
         return false;

      if ($this->_single_sock !== null) {
         $this->_flush_read_buffer($this->_single_sock);
         return $this->sock_to_host($this->_single_sock);
      }

      $hv = is_array($key) ? intval($key[0]) : $this->_hashfunc($key);

      if ($this->_buckets === null)
      {
         foreach ($this->_servers as $v)
         {
            if (is_array($v))
            {
               for ($i=0; $i<$v[1]; $i++)
                  $bu[] = $v[0];
            } else
            {
               $bu[] = $v;
            }
         }
         $this->_buckets = $bu;
         $this->_bucketcount = count($bu);
      }

      $realkey = is_array($key) ? $key[1] : $key;
      for ($tries = 0; $tries<20; $tries++)
      {
         $host = $this->_buckets[$hv % $this->_bucketcount];
         $sock = $this->sock_to_host($host);
         if (is_resource($sock)) {
            $this->_flush_read_buffer($sock);
            return $sock;
         }
         $hv += $this->_hashfunc($tries . $realkey);
      }

      return false;
   }

   // }}}
   // {{{ _hashfunc()

   /**
    * Creates a hash interger based on the $key
    *
    * @param   string   $key     Key to hash
    *
    * @return  interger Hash value
    * @access  private
    */
   function _hashfunc ($key)
   {
      # Hash function must on [0,0x7ffffff]
      # We take the first 31 bits of the MD5 hash, which unlike the hash
      # function used in a previous version of this client, works
      return hexdec(substr(md5($key),0,8)) & 0x7fffffff;
   }

   // }}}
   // {{{ _incrdecr()

   /**
    * Perform increment/decriment on $key
    *
    * @param   string   $cmd     Command to perform
    * @param   string   $key     Key to perform it on
    * @param   interger $amt     Amount to adjust
    *
    * @return  interger    New value of $key
    * @access  private
    */
   function _incrdecr ($cmd, $key, $amt=1)
   {
      if (!$this->_active)
         return null;

      $sock = $this->get_sock($key);
      if (!is_resource($sock))
         return null;

      $key = is_array($key) ? $key[1] : $key;
      @$this->stats[$cmd]++;
      if (!$this->_safe_fwrite($sock, "$cmd $key $amt\r\n"))
         return $this->_dead_sock($sock);

      stream_set_timeout($sock, 1, 0);
      $line = fgets($sock);
      if (!preg_match('/^(\d+)/', $line, $match))
         return null;
      return $match[1];
   }

   // }}}
   // {{{ _load_items()

   /**
    * Load items into $ret from $sock
    *
    * @param   resource $sock    Socket to read from
    * @param   array    $ret     Returned values
    *
    * @access  private
    */
   function _load_items ($sock, &$ret)
   {
      while (1)
      {
         $decl = fgets($sock);
         if ($decl == "END\r\n")
         {
            return true;
         } elseif (preg_match('/^VALUE (\S+) (\d+) (\d+)\r\n$/', $decl, $match))
         {
            list($rkey, $flags, $len) = array($match[1], $match[2], $match[3]);
            $bneed = $len+2;
            $offset = 0;

            while ($bneed > 0)
            {
               $data = fread($sock, $bneed);
               $n = strlen($data);
               if ($n == 0)
                  break;
               $offset += $n;
               $bneed -= $n;
               @$ret[$rkey] .= $data;
            }

            if ($offset != $len+2)
            {
               // Something is borked!
               if ($this->_debug)
                  $this->_debugprint(sprintf("Something is borked!  key %s expecting %d got %d length\n", $rkey, $len+2, $offset));

               unset($ret[$rkey]);
               $this->_close_sock($sock);
               return false;
            }

            if ($this->_have_zlib && $flags & MEMCACHE_COMPRESSED)
               $ret[$rkey] = gzuncompress($ret[$rkey]);

            $ret[$rkey] = rtrim($ret[$rkey]);

            if ($flags & MEMCACHE_SERIALIZED)
               $ret[$rkey] = unserialize($ret[$rkey]);

         } else
         {
            $this->_debugprint("Error parsing memcached response\n");
            return 0;
         }
      }
   }

   // }}}
   // {{{ _set()

   /**
    * Performs the requested storage operation to the memcache server
    *
    * @param   string   $cmd     Command to perform
    * @param   string   $key     Key to act on
    * @param   mixed    $val     What we need to store
    * @param   interger $exp     When it should expire
    *
    * @return  boolean
    * @access  private
    */
   function _set ($cmd, $key, $val, $exp)
   {
      if (!$this->_active)
         return false;

      $sock = $this->get_sock($key);
      if (!is_resource($sock))
         return false;

      @$this->stats[$cmd]++;

      $flags = 0;

      if (!is_scalar($val))
      {
         $val = serialize($val);
         $flags |= MEMCACHE_SERIALIZED;
         if ($this->_debug)
            $this->_debugprint(sprintf("client: serializing data as it is not scalar\n"));
      }

      $len = strlen($val);

      if ($this->_have_zlib && $this->_compress_enable &&
          $this->_compress_threshold && $len >= $this->_compress_threshold)
      {
         $c_val = gzcompress($val, 9);
         $c_len = strlen($c_val);

         if ($c_len < $len*(1 - COMPRESSION_SAVINGS))
         {
            if ($this->_debug)
               $this->_debugprint(sprintf("client: compressing data; was %d bytes is now %d bytes\n", $len, $c_len));
            $val = $c_val;
            $len = $c_len;
            $flags |= MEMCACHE_COMPRESSED;
         }
      }
      if (!$this->_safe_fwrite($sock, "$cmd $key $flags $exp $len\r\n$val\r\n"))
         return $this->_dead_sock($sock);

      $line = trim(fgets($sock));

      if ($this->_debug)
      {
         if ($flags & MEMCACHE_COMPRESSED)
            $val = 'compressed data';
         $this->_debugprint(sprintf("MemCache: %s %s => %s (%s)\n", $cmd, $key, $val, $line));
      }
      if ($line == "STORED")
         return true;
      return false;
   }

   // }}}
   // {{{ sock_to_host()

   /**
    * Returns the socket for the host
    *
    * @param   string   $host    Host:IP to get socket for
    *
    * @return  mixed    IO Stream or false
    * @access  private
    */
   function sock_to_host ($host)
   {
      if (isset($this->_cache_sock[$host]))
         return $this->_cache_sock[$host];

      $now = time();
      list ($ip, $port) = explode (":", $host);
      if (isset($this->_host_dead[$host]) && $this->_host_dead[$host] > $now ||
          isset($this->_host_dead[$ip]) && $this->_host_dead[$ip] > $now)
         return null;

      if (!$this->_connect_sock($sock, $host))
         return $this->_dead_sock($host);

      // Do not buffer writes
      stream_set_write_buffer($sock, 0);

      $this->_cache_sock[$host] = $sock;

      return $this->_cache_sock[$host];
   }

   function _debugprint($str){
      print($str);
   }

   /**
    * Write to a stream, timing out after the correct amount of time
    *
    * @return bool false on failure, true on success
    */
    /*
   function _safe_fwrite($f, $buf, $len = false) {
      stream_set_blocking($f, 0);

      if ($len === false) {
         wfDebug("Writing " . strlen( $buf ) . " bytes\n");
         $bytesWritten = fwrite($f, $buf);
      } else {
         wfDebug("Writing $len bytes\n");
         $bytesWritten = fwrite($f, $buf, $len);
      }
      $n = stream_select($r=NULL, $w = array($f), $e = NULL, 10, 0);
      #   $this->_timeout_seconds, $this->_timeout_microseconds);

      wfDebug("stream_select returned $n\n");
      stream_set_blocking($f, 1);
      return $n == 1;
      return $bytesWritten;
   }*/

   /**
    * Original behaviour
    */
   function _safe_fwrite($f, $buf, $len = false) {
      if ($len === false) {
         $bytesWritten = fwrite($f, $buf);
      } else {
         $bytesWritten = fwrite($f, $buf, $len);
      }
      return $bytesWritten;
   }

   /**
    * Flush the read buffer of a stream
    */
   function _flush_read_buffer($f) {
      if (!is_resource($f)) {
         return;
      }
      $n = stream_select($r=array($f), $w = NULL, $e = NULL, 0, 0);
      while ($n == 1 && !feof($f)) {
         fread($f, 1024);
         $n = stream_select($r=array($f), $w = NULL, $e = NULL, 0, 0);
      }
   }

   // }}}
   // }}}
   // }}}
}

// vim: sts=3 sw=3 et

// }}}
?>