28use UnexpectedValueException;
29use Wikimedia\ScopedCallback;
62 'compress_threshold' => 1500,
63 'connect_timeout' => 0.5,
65 'serializer' =>
'php',
66 'use_binary_protocol' =>
false,
67 'allow_tcp_nagle_delay' => true
75 $client =
new Memcached( $connectionPoolId );
86 ini_set(
'memcached.compression_threshold',
$params[
'compress_threshold'] );
98 private function initializeClient( Memcached
$client, array
$params ) {
99 if (
$client->getServerList() ) {
100 $this->logger->debug( __METHOD__ .
": pre-initialized client instance." );
105 $this->logger->debug( __METHOD__ .
": initializing new client instance." );
108 Memcached::OPT_NO_BLOCK =>
false,
109 Memcached::OPT_BUFFER_WRITES =>
false,
110 Memcached::OPT_NOREPLY =>
false,
112 Memcached::OPT_BINARY_PROTOCOL =>
$params[
'use_binary_protocol'],
114 Memcached::OPT_CONNECT_TIMEOUT =>
$params[
'connect_timeout'] * 1000,
115 Memcached::OPT_SEND_TIMEOUT =>
$params[
'timeout'],
116 Memcached::OPT_RECV_TIMEOUT =>
$params[
'timeout'],
117 Memcached::OPT_POLL_TIMEOUT =>
$params[
'timeout'] / 1000,
119 Memcached::OPT_TCP_NODELAY => !
$params[
'allow_tcp_nagle_delay'],
121 Memcached::OPT_LIBKETAMA_COMPATIBLE => true
123 if ( isset(
$params[
'retry_timeout'] ) ) {
124 $options[Memcached::OPT_RETRY_TIMEOUT] =
$params[
'retry_timeout'];
126 if ( isset(
$params[
'server_failure_limit'] ) ) {
127 $options[Memcached::OPT_SERVER_FAILURE_LIMIT] =
$params[
'server_failure_limit'];
129 if (
$params[
'serializer'] ===
'php' ) {
130 $options[Memcached::OPT_SERIALIZER] = Memcached::SERIALIZER_PHP;
131 } elseif (
$params[
'serializer'] ===
'igbinary' ) {
133 if ( !Memcached::HAVE_IGBINARY ) {
134 throw new RuntimeException(
135 __CLASS__ .
': the igbinary extension is not available ' .
136 'but igbinary serialization was requested.'
139 $options[Memcached::OPT_SERIALIZER] = Memcached::SERIALIZER_IGBINARY;
142 if ( !
$client->setOptions( $options ) ) {
143 throw new RuntimeException(
144 "Invalid options: " . json_encode( $options, JSON_PRETTY_PRINT )
149 foreach (
$params[
'servers'] as $host ) {
150 if ( preg_match(
'/^\[(.+)\]:(\d+)$/', $host, $m ) ) {
151 $servers[] = [ $m[1], (int)$m[2] ];
152 } elseif ( preg_match(
'/^([^:]+):(\d+)$/', $host, $m ) ) {
153 $servers[] = [ $m[1], (int)$m[2] ];
155 $servers[] = [ $host, false ];
159 if ( !
$client->addServers( $servers ) ) {
160 throw new RuntimeException(
"Failed to inject server address list" );
173 private function noReplyScope( $flags ) {
174 if ( $flags !==
true && !( $flags & self::WRITE_BACKGROUND ) ) {
178 $client->setOption( Memcached::OPT_NOREPLY,
true );
180 return new ScopedCallback(
static function () use (
$client ) {
181 $client->setOption( Memcached::OPT_NOREPLY,
false );
185 protected function doGet( $key, $flags = 0, &$casToken =
null ) {
189 $this->
debug(
"get($key)" );
196 $flags = Memcached::GET_EXTENDED;
197 $res = $this->client->get( $routeKey,
null, $flags );
198 if ( is_array( $res ) ) {
199 $result = $res[
'value'];
200 $casToken = $res[
'cas'];
205 $result = $this->client->get( $routeKey );
211 protected function doSet( $key, $value, $exptime = 0, $flags = 0 ) {
212 $this->
debug(
"set($key)" );
216 $noReplyScope = $this->noReplyScope( $flags );
217 $result = $this->client->set( $routeKey, $value, $this->
fixExpiry( $exptime ) );
218 ScopedCallback::consume( $noReplyScope );
220 return ( !$result && $this->client->getResultCode() === Memcached::RES_NOTSTORED )
226 protected function doCas( $casToken, $key, $value, $exptime = 0, $flags = 0 ) {
227 $this->
debug(
"cas($key)" );
230 $result = $this->client->cas(
240 $this->
debug(
"delete($key)" );
243 $noReplyScope = $this->noReplyScope( $flags );
244 $result = $this->client->delete( $routeKey );
245 ScopedCallback::consume( $noReplyScope );
247 return ( !$result && $this->client->getResultCode() === Memcached::RES_NOTFOUND )
253 protected function doAdd( $key, $value, $exptime = 0, $flags = 0 ) {
254 $this->
debug(
"add($key)" );
257 $noReplyScope = $this->noReplyScope( $flags );
258 $result = $this->client->add(
263 ScopedCallback::consume( $noReplyScope );
269 $this->
debug(
"incrWithInit($key)" );
272 $scope = $this->noReplyScope(
true );
273 $this->
checkResult( $key, $this->client->add( $routeKey, $init - $step, $this->fixExpiry( $exptime ) ) );
274 $this->
checkResult( $key, $this->client->increment( $routeKey, $step ) );
275 ScopedCallback::consume( $scope );
282 $this->
debug(
"incrWithInit($key)" );
285 $result = $this->client->increment( $routeKey, $step );
287 if ( $newValue ===
false && !$this->
getLastError( $watchPoint ) ) {
289 $result = $this->client->add( $routeKey, $init, $this->
fixExpiry( $exptime ) );
290 $newValue = $this->
checkResult( $key, $result ) ? $init :
false;
291 if ( $newValue ===
false && !$this->
getLastError( $watchPoint ) ) {
293 $result = $this->client->increment( $routeKey, $step );
314 static $statusByCode = [
328 if ( $result !==
false ) {
333 $code =
$client->getResultCode();
335 case Memcached::RES_SUCCESS:
337 case Memcached::RES_DATA_EXISTS:
338 case Memcached::RES_NOTSTORED:
339 case Memcached::RES_NOTFOUND:
340 $this->
debug(
"result: " . $client->getResultMessage() );
343 $msg =
$client->getResultMessage();
345 if ( $key !==
false ) {
346 $server =
$client->getServerByKey( $key );
347 $logCtx[
'memcached-server'] =
"{$server['host']}:{$server['port']}";
348 $logCtx[
'memcached-key'] = $key;
349 $msg =
"Memcached error for key \"{memcached-key}\" " .
350 "on server \"{memcached-server}\": $msg";
352 $msg =
"Memcached error: $msg";
354 $this->logger->error( $msg, $logCtx );
355 $this->
setLastError( $statusByCode[$code] ?? self::ERR_UNEXPECTED );
362 $this->
debug(
'getMulti(' . implode(
', ', $keys ) .
')' );
365 foreach ( $keys as $key ) {
372 $resByRouteKey = $this->client->getMulti( $routeKeys );
374 if ( is_array( $resByRouteKey ) ) {
376 foreach ( $resByRouteKey as $routeKey => $value ) {
385 return $res !==
false ? $res : [];
388 protected function doSetMulti( array $data, $exptime = 0, $flags = 0 ) {
389 $this->
debug(
'setMulti(' . implode(
', ', array_keys( $data ) ) .
')' );
392 $dataByRouteKey = [];
393 foreach ( $data as $key => $value ) {
397 $noReplyScope = $this->noReplyScope( $flags );
401 $result = @$this->client->setMulti( $dataByRouteKey, $exptime );
402 ScopedCallback::consume( $noReplyScope );
408 $this->
debug(
'deleteMulti(' . implode(
', ', $keys ) .
')' );
411 foreach ( $keys as $key ) {
415 $noReplyScope = $this->noReplyScope( $flags );
416 $resultArray = $this->client->deleteMulti( $routeKeys ) ?: [];
417 ScopedCallback::consume( $noReplyScope );
420 foreach ( $resultArray as $code ) {
421 if ( !in_array( $code, [
true, Memcached::RES_NOTFOUND ],
true ) ) {
431 $this->
debug(
"touch($key)" );
436 $result = $this->client->touch( $routeKey, $this->
fixExpiry( $exptime ) );
442 if ( is_int( $value ) ) {
446 $serializer = $this->client->getOption( Memcached::OPT_SERIALIZER );
447 if ( $serializer === Memcached::SERIALIZER_PHP ) {
449 } elseif ( $serializer === Memcached::SERIALIZER_IGBINARY ) {
450 return igbinary_serialize( $value );
453 throw new UnexpectedValueException( __METHOD__ .
": got serializer '$serializer'." );
461 $serializer = $this->client->getOption( Memcached::OPT_SERIALIZER );
462 if ( $serializer === Memcached::SERIALIZER_PHP ) {
464 } elseif ( $serializer === Memcached::SERIALIZER_IGBINARY ) {
465 return igbinary_unserialize( $value );
468 throw new UnexpectedValueException( __METHOD__ .
": got serializer '$serializer'." );
473class_alias( MemcachedPeclBagOStuff::class,
'MemcachedPeclBagOStuff' );
array $params
The job parameters.