MediaWiki  master
MemcLockManager.php
Go to the documentation of this file.
1 <?php
24 
41  protected $lockTypeMap = [
42  self::LOCK_SH => self::LOCK_SH,
43  self::LOCK_UW => self::LOCK_SH,
44  self::LOCK_EX => self::LOCK_EX
45  ];
46 
48  protected $cacheServers = [];
50  protected $statusCache;
51 
64  public function __construct( array $config ) {
65  parent::__construct( $config );
66 
67  // Sanitize srvsByBucket config to prevent PHP errors
68  $this->srvsByBucket = array_filter( $config['srvsByBucket'], 'is_array' );
69  $this->srvsByBucket = array_values( $this->srvsByBucket ); // consecutive
70 
71  $memcConfig = $config['memcConfig'] ?? [];
72  $memcConfig += [ 'class' => MemcachedPhpBagOStuff::class ]; // default
73 
74  $class = $memcConfig['class'];
75  if ( !is_subclass_of( $class, MemcachedBagOStuff::class ) ) {
76  throw new InvalidArgumentException( "$class is not of type MemcachedBagOStuff." );
77  }
78 
79  foreach ( $config['lockServers'] as $name => $address ) {
80  $params = [ 'servers' => [ $address ] ] + $memcConfig;
81  $this->cacheServers[$name] = new $class( $params );
82  }
83 
84  $this->statusCache = new MapCacheLRU( 100 );
85  }
86 
87  protected function getLocksOnServer( $lockSrv, array $pathsByType ) {
88  $status = StatusValue::newGood();
89 
90  $memc = $this->getCache( $lockSrv );
91  // List of affected paths
92  $paths = array_merge( ...array_values( $pathsByType ) );
93  $paths = array_unique( $paths );
94  // List of affected lock record keys
95  $keys = array_map( [ $this, 'recordKeyForPath' ], $paths );
96 
97  // Lock all of the active lock record keys...
98  if ( !$this->acquireMutexes( $memc, $keys ) ) {
99  foreach ( $paths as $path ) {
100  $status->fatal( 'lockmanager-fail-acquirelock', $path );
101  }
102 
103  return $status;
104  }
105 
106  // Fetch all the existing lock records...
107  $lockRecords = $memc->getMulti( $keys );
108 
109  $now = time();
110  // Check if the requested locks conflict with existing ones...
111  foreach ( $pathsByType as $type => $paths ) {
112  foreach ( $paths as $path ) {
113  $locksKey = $this->recordKeyForPath( $path );
114  $locksHeld = isset( $lockRecords[$locksKey] )
115  ? self::sanitizeLockArray( $lockRecords[$locksKey] )
116  : self::newLockArray(); // init
117  foreach ( $locksHeld[self::LOCK_EX] as $session => $expiry ) {
118  if ( $expiry < $now ) { // stale?
119  unset( $locksHeld[self::LOCK_EX][$session] );
120  } elseif ( $session !== $this->session ) {
121  $status->fatal( 'lockmanager-fail-acquirelock', $path );
122  }
123  }
124  if ( $type === self::LOCK_EX ) {
125  foreach ( $locksHeld[self::LOCK_SH] as $session => $expiry ) {
126  if ( $expiry < $now ) { // stale?
127  unset( $locksHeld[self::LOCK_SH][$session] );
128  } elseif ( $session !== $this->session ) {
129  $status->fatal( 'lockmanager-fail-acquirelock', $path );
130  }
131  }
132  }
133  if ( $status->isOK() ) {
134  // Register the session in the lock record array
136  // We will update this record if none of the other locks conflict
137  $lockRecords[$locksKey] = $locksHeld;
138  }
139  }
140  }
141 
142  // If there were no lock conflicts, update all the lock records...
143  if ( $status->isOK() ) {
144  foreach ( $paths as $path ) {
145  $locksKey = $this->recordKeyForPath( $path );
146  $locksHeld = $lockRecords[$locksKey];
147  $ok = $memc->set( $locksKey, $locksHeld, self::MAX_LOCK_TTL );
148  if ( !$ok ) {
149  $status->fatal( 'lockmanager-fail-acquirelock', $path );
150  } else {
151  $this->logger->debug( __METHOD__ . ": acquired lock on key $locksKey.\n" );
152  }
153  }
154  }
155 
156  // Unlock all of the active lock record keys...
157  $this->releaseMutexes( $memc, $keys );
158 
159  return $status;
160  }
161 
162  protected function freeLocksOnServer( $lockSrv, array $pathsByType ) {
163  $status = StatusValue::newGood();
164 
165  $memc = $this->getCache( $lockSrv );
166  // List of affected paths
167  $paths = array_merge( ...array_values( $pathsByType ) );
168  $paths = array_unique( $paths );
169  // List of affected lock record keys
170  $keys = array_map( [ $this, 'recordKeyForPath' ], $paths );
171 
172  // Lock all of the active lock record keys...
173  if ( !$this->acquireMutexes( $memc, $keys ) ) {
174  foreach ( $paths as $path ) {
175  $status->fatal( 'lockmanager-fail-releaselock', $path );
176  }
177 
178  return $status;
179  }
180 
181  // Fetch all the existing lock records...
182  $lockRecords = $memc->getMulti( $keys );
183 
184  // Remove the requested locks from all records...
185  foreach ( $pathsByType as $type => $paths ) {
186  foreach ( $paths as $path ) {
187  $locksKey = $this->recordKeyForPath( $path ); // lock record
188  if ( !isset( $lockRecords[$locksKey] ) ) {
189  $status->warning( 'lockmanager-fail-releaselock', $path );
190  continue; // nothing to do
191  }
192  $locksHeld = $this->sanitizeLockArray( $lockRecords[$locksKey] );
193  if ( isset( $locksHeld[$type][$this->session] ) ) {
194  unset( $locksHeld[$type][$this->session] ); // unregister this session
195  $lockRecords[$locksKey] = $locksHeld;
196  } else {
197  $status->warning( 'lockmanager-fail-releaselock', $path );
198  }
199  }
200  }
201 
202  // Persist the new lock record values...
203  foreach ( $paths as $path ) {
204  $locksKey = $this->recordKeyForPath( $path );
205  if ( !isset( $lockRecords[$locksKey] ) ) {
206  continue; // nothing to do
207  }
208  $locksHeld = $lockRecords[$locksKey];
209  if ( $locksHeld === $this->newLockArray() ) {
210  $ok = $memc->delete( $locksKey );
211  } else {
212  $ok = $memc->set( $locksKey, $locksHeld, self::MAX_LOCK_TTL );
213  }
214  if ( $ok ) {
215  $this->logger->debug( __METHOD__ . ": released lock on key $locksKey.\n" );
216  } else {
217  $status->fatal( 'lockmanager-fail-releaselock', $path );
218  }
219  }
220 
221  // Unlock all of the active lock record keys...
222  $this->releaseMutexes( $memc, $keys );
223 
224  return $status;
225  }
226 
231  protected function releaseAllLocks() {
232  return StatusValue::newGood(); // not supported
233  }
234 
240  protected function isServerUp( $lockSrv ) {
241  return (bool)$this->getCache( $lockSrv );
242  }
243 
250  protected function getCache( $lockSrv ) {
251  if ( !isset( $this->cacheServers[$lockSrv] ) ) {
252  throw new InvalidArgumentException( "Invalid cache server '$lockSrv'." );
253  }
254 
255  $online = $this->statusCache->get( "online:$lockSrv", 30 );
256  if ( $online === null ) {
257  $online = $this->cacheServers[$lockSrv]->set( __CLASS__ . ':ping', 1, 1 );
258  if ( !$online ) { // server down?
259  $this->logger->warning( __METHOD__ . ": Could not contact $lockSrv." );
260  }
261  $this->statusCache->set( "online:$lockSrv", (int)$online );
262  }
263 
264  return $online ? $this->cacheServers[$lockSrv] : null;
265  }
266 
271  protected function recordKeyForPath( $path ) {
272  return implode( ':', [ __CLASS__, 'locks', $this->sha1Base36Absolute( $path ) ] );
273  }
274 
278  protected function newLockArray() {
279  return [ self::LOCK_SH => [], self::LOCK_EX => [] ];
280  }
281 
286  protected function sanitizeLockArray( $a ) {
287  if ( is_array( $a ) && isset( $a[self::LOCK_EX] ) && isset( $a[self::LOCK_SH] ) ) {
288  return $a;
289  }
290 
291  $this->logger->error( __METHOD__ . ": reset invalid lock array." );
292 
293  return $this->newLockArray();
294  }
295 
301  protected function acquireMutexes( MemcachedBagOStuff $memc, array $keys ) {
302  $lockedKeys = [];
303 
304  // Acquire the keys in lexicographical order, to avoid deadlock problems.
305  // If P1 is waiting to acquire a key P2 has, P2 can't also be waiting for a key P1 has.
306  sort( $keys );
307 
308  // Try to quickly loop to acquire the keys, but back off after a few rounds.
309  // This reduces memcached spam, especially in the rare case where a server acquires
310  // some lock keys and dies without releasing them. Lock keys expire after a few minutes.
311  $loop = new WaitConditionLoop(
312  function () use ( $memc, $keys, &$lockedKeys ) {
313  foreach ( array_diff( $keys, $lockedKeys ) as $key ) {
314  if ( $memc->add( "$key:mutex", 1, 180 ) ) { // lock record
315  $lockedKeys[] = $key;
316  }
317  }
318 
319  return array_diff( $keys, $lockedKeys )
320  ? WaitConditionLoop::CONDITION_CONTINUE
321  : true;
322  },
323  3.0 // timeout
324  );
325  $loop->invoke();
326 
327  if ( count( $lockedKeys ) != count( $keys ) ) {
328  $this->releaseMutexes( $memc, $lockedKeys ); // failed; release what was locked
329  return false;
330  }
331 
332  return true;
333  }
334 
339  protected function releaseMutexes( MemcachedBagOStuff $memc, array $keys ) {
340  foreach ( $keys as $key ) {
341  $memc->delete( "$key:mutex" );
342  }
343  }
344 
348  function __destruct() {
349  while ( count( $this->locksHeld ) ) {
350  foreach ( $this->locksHeld as $path => $locks ) {
351  $this->doUnlock( [ $path ], self::LOCK_EX );
352  $this->doUnlock( [ $path ], self::LOCK_SH );
353  }
354  }
355  }
356 }
delete( $key, $flags=0)
Delete an item.
add( $key, $value, $exptime=0, $flags=0)
__construct(array $config)
Construct a new instance from configuration.
array $lockTypeMap
Mapping of lock types to the type actually used.
MapCacheLRU $statusCache
Server status cache.
string $session
Random 32-char hex number.
Definition: LockManager.php:65
static newGood( $value=null)
Factory function for good results.
Definition: StatusValue.php:81
array $locksHeld
Map of (resource path => lock type => count)
Definition: LockManager.php:59
doUnlock(array $paths, $type)
acquireMutexes(MemcachedBagOStuff $memc, array $keys)
getLocksOnServer( $lockSrv, array $pathsByType)
Manage locks using memcached servers.
isServerUp( $lockSrv)
releaseMutexes(MemcachedBagOStuff $memc, array $keys)
Base class for memcached clients.
__destruct()
Make sure remaining locks get cleared for sanity.
MemcachedBagOStuff [] $cacheServers
Map of (server name => MemcachedBagOStuff)
getCache( $lockSrv)
Get the MemcachedBagOStuff object for a $lockSrv.
Version of LockManager that uses a quorum from peer servers for locks.
freeLocksOnServer( $lockSrv, array $pathsByType)
sha1Base36Absolute( $path)
Get the base 36 SHA-1 of a string, padded to 31 digits.