69use Psr\Log\LoggerInterface;
70use Psr\Log\NullLogger;
71use Wikimedia\AtEase\AtEase;
268 $this->_debug = $args[
'debug'] ??
false;
269 $this->stats = array();
270 $this->_compress_threshold = $args[
'compress_threshold'] ?? 0;
271 $this->_persistent = $args[
'persistent'] ??
false;
272 $this->_compress_enable =
true;
273 $this->_have_zlib = function_exists(
'gzcompress' );
275 $this->_cache_sock = array();
276 $this->_host_dead = array();
278 $this->_timeout_seconds = 0;
279 $this->_timeout_microseconds = $args[
'timeout'] ?? 500000;
281 $this->_connect_timeout = $args[
'connect_timeout'] ?? 0.1;
282 $this->_connect_attempts = 2;
284 $this->_logger = $args[
'logger'] ??
new NullLogger();
321 public function add( $key, $val, $exp = 0 ) {
324 return $this->
_set(
'add', $key, $val, $exp );
338 public function decr( $key, $amt = 1 ) {
341 return $this->
_incrdecr(
'decr', $key, $amt );
355 public function delete( $key, $time = 0 ) {
358 if ( !$this->_active ) {
367 $key = is_array( $key ) ? $key[1] : $key;
369 if ( isset( $this->stats[
'delete'] ) ) {
370 $this->stats[
'delete']++;
372 $this->stats[
'delete'] = 1;
374 $cmd =
"delete $key $time\r\n";
375 if ( !$this->
_fwrite( $sock, $cmd ) ) {
378 $res = $this->
_fgets( $sock );
380 if ( $this->_debug ) {
381 $this->
_debugprint( sprintf(
"MemCache: delete %s (%s)", $key, $res ) );
384 if ( $res ==
"DELETED" || $res ==
"NOT_FOUND" ) {
401 public function touch( $key, $time = 0 ) {
404 if ( !$this->_active ) {
413 $key = is_array( $key ) ? $key[1] : $key;
415 if ( isset( $this->stats[
'touch'] ) ) {
416 $this->stats[
'touch']++;
418 $this->stats[
'touch'] = 1;
420 $cmd =
"touch $key $time\r\n";
421 if ( !$this->
_fwrite( $sock, $cmd ) ) {
424 $res = $this->
_fgets( $sock );
426 if ( $this->_debug ) {
427 $this->
_debugprint( sprintf(
"MemCache: touch %s (%s)", $key, $res ) );
430 if ( $res ==
"TOUCHED" ) {
444 foreach ( $this->_cache_sock as $sock ) {
448 $this->_cache_sock = array();
460 $this->_compress_enable = $enable;
470 $this->_host_dead = array();
484 public function get( $key, &$casToken = null ) {
485 $getToken = ( func_num_args() >= 2 );
489 if ( $this->_debug ) {
493 if ( !is_array( $key ) && strval( $key ) ===
'' ) {
495 $this->
_debugprint(
"Skipping key which equals to an empty string" );
499 if ( !$this->_active ) {
513 $key = is_array( $key ) ? $key[1] : $key;
514 if ( isset( $this->stats[
'get'] ) ) {
515 $this->stats[
'get']++;
517 $this->stats[
'get'] = 1;
520 $cmd = $getToken ?
"gets" :
"get";
522 if ( !$this->
_fwrite( $sock, $cmd ) ) {
529 if ( !$this->
_load_items( $sock, $val, $casToken ) ) {
533 if ( $this->_debug ) {
534 foreach ( $val as $k => $v ) {
536 sprintf(
"MemCache: sock %s got %s", $this->
serialize( $sock ), $k ) );
541 if ( isset( $val[$key] ) ) {
560 if ( !$this->_active ) {
566 if ( isset( $this->stats[
'get_multi'] ) ) {
567 $this->stats[
'get_multi']++;
569 $this->stats[
'get_multi'] = 1;
571 $sock_keys = array();
573 foreach ( $keys as $key ) {
579 $key = is_array( $key ) ? $key[1] : $key;
580 $sockValue = intval( $sock );
582 if ( !isset( $sock_keys[$sockValue] ) ) {
583 $sock_keys[$sockValue] = array();
586 $sock_keys[$sockValue][] = $key;
591 foreach ( $socks as $sock ) {
593 foreach ( $sock_keys[intval( $sock )] as $key ) {
598 if ( $this->
_fwrite( $sock, $cmd ) ) {
607 foreach ( $gather as $sock ) {
613 if ( $this->_debug ) {
614 foreach ( $val as $k => $v ) {
615 $this->
_debugprint( sprintf(
"MemCache: got %s", $k ) );
635 public function incr( $key, $amt = 1 ) {
636 return $this->
_incrdecr(
'incr', $key, $amt );
655 public function replace( $key, $value, $exp = 0 ) {
656 return $this->
_set(
'replace', $key, $value, $exp );
676 if ( !$this->
_fwrite( $sock, $cmd ) ) {
682 $res = $this->
_fgets( $sock );
684 if ( preg_match(
'/^END/', $res ) ) {
687 if ( strlen( $res ) == 0 ) {
711 public function set( $key, $value, $exp = 0 ) {
712 return $this->
_set(
'set', $key, $value, $exp );
733 public function cas( $casToken, $key, $value, $exp = 0 ) {
734 return $this->
_set(
'cas', $key, $value, $exp, $casToken );
746 $this->_compress_threshold = $thresh;
759 $this->_debug = $dbg;
772 $this->_servers = $list;
773 $this->_active = count( $list );
774 $this->_buckets =
null;
775 $this->_bucketcount = 0;
777 $this->_single_sock =
null;
778 if ( $this->_active == 1 ) {
779 $this->_single_sock = $this->_servers[0];
790 $this->_timeout_seconds = $seconds;
791 $this->_timeout_microseconds = $microseconds;
807 $host = array_search( $sock, $this->_cache_sock );
808 fclose( $this->_cache_sock[$host] );
809 unset( $this->_cache_sock[$host] );
826 $hostAndPort = IPUtils::splitHostAndPort( $host );
827 if ( $hostAndPort ) {
828 $ip = $hostAndPort[0];
829 if ( $hostAndPort[1] ) {
830 $port = $hostAndPort[1];
837 $errno = $errstr =
null;
839 AtEase::suppressWarnings();
840 if ( $this->_persistent == 1 ) {
841 $sock = pfsockopen( $ip, $port, $errno, $errstr, $timeout );
843 $sock = fsockopen( $ip, $port, $errno, $errstr, $timeout );
845 AtEase::restoreWarnings();
848 $this->
_error_log(
"Error connecting to $host: $errstr" );
854 stream_set_timeout( $sock, $this->_timeout_seconds, $this->_timeout_microseconds );
858 if ( $this->_persistent ) {
875 $host = array_search( $sock, $this->_cache_sock );
883 $hostAndPort = IPUtils::splitHostAndPort( $host );
884 if ( $hostAndPort ) {
885 $ip = $hostAndPort[0];
889 $this->_host_dead[$ip] = time() + 30 + intval( rand( 0, 10 ) );
890 $this->_host_dead[$host] = $this->_host_dead[$ip];
891 unset( $this->_cache_sock[$host] );
906 if ( !$this->_active ) {
910 if ( $this->_single_sock !==
null ) {
914 $hv = is_array( $key ) ? intval( $key[0] ) : $this->
_hashfunc( $key );
915 if ( $this->_buckets ===
null ) {
917 foreach ( $this->_servers as $v ) {
918 if ( is_array( $v ) ) {
919 for ( $i = 0; $i < $v[1]; $i++ ) {
926 $this->_buckets = $bu;
927 $this->_bucketcount = count( $bu );
930 $realkey = is_array( $key ) ? $key[1] : $key;
931 for ( $tries = 0; $tries < 20; $tries++ ) {
937 $hv = $this->
_hashfunc( $hv . $realkey );
955 # Hash function must be in [0,0x7ffffff]
956 # We take the first 31 bits of the MD5 hash, which unlike the hash
957 # function used in a previous version of this client, works
958 return hexdec( substr( md5( $key ), 0, 8 ) ) & 0x7fffffff;
977 if ( !$this->_active ) {
990 $key = is_array( $key ) ? $key[1] : $key;
991 if ( isset( $this->stats[$cmd] ) ) {
992 $this->stats[$cmd]++;
994 $this->stats[$cmd] = 1;
996 if ( !$this->
_fwrite( $sock,
"$cmd $key $amt\r\n" ) ) {
1002 $line = $this->
_fgets( $sock );
1003 if ( $this->_debug ) {
1008 if ( !preg_match(
'/^(\d+)/', $line, $match ) ) {
1014 return (
int)$match[1];
1034 $decl = $this->
_fgets( $sock );
1036 if ( $decl ===
false ) {
1042 } elseif ( preg_match(
'/^VALUE (\S+) (\d+) (\d+)(?: (\d+))?$/', $decl, $match ) ) {
1055 $this->
_fread( $sock, $match[3] + 2 ),
1057 } elseif ( $decl ==
"END" ) {
1062 foreach ( $results as [ $rkey, $flags, , $casToken, $data ] ) {
1063 if ( $data ===
false || substr( $data, -2 ) !==
"\r\n" ) {
1065 'line ending missing from data block from $1' );
1068 $data = substr( $data, 0, -2 );
1069 $ret[$rkey] = $data;
1071 if ( $this->_have_zlib && $flags & self::COMPRESSED ) {
1072 $ret[$rkey] = gzuncompress( $ret[$rkey] );
1084 if ( $flags & self::SERIALIZED ) {
1086 } elseif ( $flags & self::INTVAL ) {
1087 $ret[$rkey] = intval( $ret[$rkey] );
1093 $this->
_handle_error( $sock,
'Error parsing response from $1' );
1118 function _set( $cmd, $key, $val, $exp, $casToken =
null ) {
1121 if ( !$this->_active ) {
1134 if ( isset( $this->stats[$cmd] ) ) {
1135 $this->stats[$cmd]++;
1137 $this->stats[$cmd] = 1;
1142 if ( is_int( $val ) ) {
1144 } elseif ( !is_scalar( $val ) ) {
1147 if ( $this->_debug ) {
1148 $this->
_debugprint(
"client: serializing data as it is not scalar" );
1152 $len = strlen( $val );
1154 if ( $this->_have_zlib && $this->_compress_enable
1155 && $this->_compress_threshold && $len >= $this->_compress_threshold
1157 $c_val = gzcompress( $val, 9 );
1158 $c_len = strlen( $c_val );
1160 if ( $c_len < $len * ( 1 - self::COMPRESSION_SAVINGS ) ) {
1161 if ( $this->_debug ) {
1162 $this->
_debugprint( sprintf(
"client: compressing data; was %d bytes is now %d bytes", $len, $c_len ) );
1170 $command =
"$cmd $key $flags $exp $len";
1172 $command .=
" $casToken";
1175 if ( !$this->
_fwrite( $sock,
"$command\r\n$val\r\n" ) ) {
1181 $line = $this->
_fgets( $sock );
1182 if ( $this->_debug ) {
1183 $this->
_debugprint( sprintf(
"%s %s (%s)", $cmd, $key, $line ) );
1186 if ( $line ===
"STORED" ) {
1188 } elseif ( $line ===
"NOT_STORED" && $cmd ===
"set" ) {
1193 if ( $line ===
false ) {
1212 if ( isset( $this->_cache_sock[$host] ) ) {
1213 return $this->_cache_sock[$host];
1218 $hostAndPort = IPUtils::splitHostAndPort( $host );
1219 if ( $hostAndPort ) {
1220 $ip = $hostAndPort[0];
1224 if ( isset( $this->_host_dead[$host] ) && $this->_host_dead[$host] > $now ||
1225 isset( $this->_host_dead[$ip] ) && $this->_host_dead[$ip] > $now
1235 stream_set_write_buffer( $sock, 0 );
1237 $this->_cache_sock[$host] = $sock;
1239 return $this->_cache_sock[$host];
1246 $this->_logger->debug( $text );
1253 $this->_logger->error(
"Memcached error: $text" );
1265 $bufSize = strlen( $buf );
1266 while ( $bytesWritten < $bufSize ) {
1267 $result = fwrite( $sock, $buf );
1268 $data = stream_get_meta_data( $sock );
1269 if ( $data[
'timed_out'] ) {
1274 if ( $result ===
false || $result === 0 ) {
1278 $bytesWritten += $result;
1291 $peer = stream_socket_get_name( $sock,
true );
1292 if ( strval( $peer ) ===
'' ) {
1293 $peer = array_search( $sock, $this->_cache_sock );
1294 if ( $peer ===
false ) {
1295 $peer =
'[unknown host]';
1298 $msg = str_replace(
'$1', $peer, $msg );
1313 while ( $len > 0 ) {
1314 $result = fread( $sock, $len );
1315 $data = stream_get_meta_data( $sock );
1316 if ( $data[
'timed_out'] ) {
1320 if ( $result ===
false ) {
1321 $this->
_handle_error( $sock,
'error reading buffer from $1' );
1324 if ( $result ===
'' ) {
1326 $this->
_handle_error( $sock,
'unexpected end of file reading from $1' );
1329 $len -= strlen( $result );
1343 $result = fgets( $sock );
1347 $data = stream_get_meta_data( $sock );
1348 if ( $data[
'timed_out'] ) {
1349 $this->
_handle_error( $sock,
'timeout reading line from $1' );
1352 if ( $result ===
false ) {
1353 $this->
_handle_error( $sock,
'error reading line from $1' );
1356 if ( substr( $result, -2 ) ===
"\r\n" ) {
1357 $result = substr( $result, 0, -2 );
1358 } elseif ( substr( $result, -1 ) ===
"\n" ) {
1359 $result = substr( $result, 0, -1 );
1361 $this->
_handle_error( $sock,
'line ending missing in response from $1' );
1378 $n = stream_select( $r, $w, $e, 0, 0 );
1379 while ( $n == 1 && !feof( $f ) ) {
1384 $n = stream_select( $r, $w, $e, 0, 0 );
memcached client class implemented using (p)fsockopen()
set_debug( $dbg)
Set the debug flag.
array $_cache_sock
Cached Sockets that are connected.
const SERIALIZED
Flag: indicates data is serialized.
const COMPRESSION_SAVINGS
Minimum savings to store data compressed.
_hashfunc( $key)
Creates a hash integer based on the $key.
set_compress_threshold( $thresh)
Set the compression threshold.
_fread( $sock, $len)
Read the specified number of bytes from a stream.
array $_servers
Array containing ip:port or array(ip:port, weight)
bool $_compress_enable
Do we want to use compression?
decr( $key, $amt=1)
Decrease a value stored on the memcache server.
const INTVAL
Flag: indicates data is an integer.
int $_compress_threshold
At how many bytes should we compress?
int $_timeout_seconds
Stream timeout in seconds.
disconnect_all()
Disconnects all connected sockets.
cas( $casToken, $key, $value, $exp=0)
Sets a key to a given value in the memcache if the current value still corresponds to a known,...
array $_buckets
Our bit buckets.
set_timeout( $seconds, $microseconds)
Sets the timeout for new connections.
_fgets( $sock)
Read a line from a stream.
incr( $key, $amt=1)
Increments $key (optionally) by $amt.
forget_dead_hosts()
Forget about all of the dead hosts.
_incrdecr( $cmd, $key, $amt=1)
Perform increment/decrement on $key.
$_connect_attempts
Number of connection attempts for each server.
add( $key, $val, $exp=0)
Adds a key/value to the memcache server if one isn't already set with that key.
_handle_error( $sock, $msg)
Handle an I/O error.
sock_to_host( $host)
Returns the socket for the host.
int $_last_cmd_status
StorageAwareness:ERR_* constant of the last cache command.
$_connect_timeout
Connect timeout in seconds.
touch( $key, $time=0)
Changes the TTL on a key from the server to $time.
bool $_have_zlib
Is compression available?
int $_bucketcount
Total # of bit buckets we have.
_fwrite( $sock, $buf)
Write to a stream.
bool $_debug
Current debug status; 0 - none to 9 - profiling.
replace( $key, $value, $exp=0)
Overwrites an existing value for key; only works if key is already set.
set_servers( $list)
Set the server list to distribute key gets and puts between.
_load_items( $sock, &$ret, &$casToken=null)
Load items into $ret from $sock.
string $_single_sock
If only using one server; contains ip:port to connect to.
get_multi( $keys)
Get multiple keys from the server(s)
_close_sock( $sock)
Close the specified socket.
run_command( $sock, $cmd)
Passes through $cmd to the memcache server connected by $sock; returns output as an array (null array...
_set( $cmd, $key, $val, $exp, $casToken=null)
Performs the requested storage operation to the memcache server.
int $_timeout_microseconds
Stream timeout in microseconds.
__construct( $args)
Memcache initializer.
_flush_read_buffer( $f)
Flush the read buffer of a stream.
enable_compress( $enable)
Enable / Disable compression.
array $_host_dead
Dead hosts, assoc array, 'host'=>'unixtime when ok to check again'.
const COMPRESSED
Flag: indicates data is compressed.
_dead_sock( $sock)
Marks a host as dead until 30-40 seconds in the future.
_connect_sock(&$sock, $host)
Connects $sock to $host, timing out after $timeout.
array $stats
Command statistics.
bool $_persistent
Are we using persistent links?