includes/ConcurrencyCheck.php

   1 <?php
   2
   3 /**
   4  * Class for cooperative locking of web resources
   5  *
   6  * Each resource is identified by a combination of the "resource type" (the application, the type
   7  * of content, etc), and the resource's primary key or some other unique numeric ID.
   8  *
   9  * Currently, a resource can only be checked out by a single user.  Other attempts to check it out result
  10  * in the checkout failing.  In the future, an option for multiple simulataneous checkouts could be added
  11  * without much trouble.
  12  *
  13  * This could be done with named locks, except then it would be impossible to build a list of all the
  14  * resources currently checked out for a given application.  There's no good way to construct a query
  15  * that answers the question, "What locks do you have starting with [foo]"  This could be done really well
  16  * with a concurrent, reliable, distributed key/value store, but we don't have one of those right now.
  17  *
  18  * @author Ian Baker <ian@wikimedia.org>
  19  */
  20 class ConcurrencyCheck {
  21         /**
  22          * Constructor
  23          *
  24          * @var $resourceType String The calling application or type of resource, conceptually like a namespace
  25          * @var $user User object, the current user
  26          * @var $expirationTime Integer (optional) How long should a checkout last, in seconds
  27          */
  28         public function __construct( $resourceType, $user, $expirationTime = null ) {
  29
  30                 // All database calls are to the master, since the whole point of this class is maintaining
  31                 // concurrency. Most reads should come from cache anyway.
  32                 $this->dbw = wfGetDb( DB_MASTER );
  33
  34                 $this->user = $user;
  35                 // TODO: create a registry of all valid resourceTypes that client app can add to.
  36                 $this->resourceType = $resourceType;
  37                 $this->setExpirationTime( $expirationTime );
  38         }
  39
  40         /**
  41          * Check out a resource.  This establishes an atomically generated, cooperative lock
  42          * on a key.  The lock is tied to the current user.
  43          *
  44          * @var $record Integer containing the record id to check out
  45          * @var $override Boolean (optional) describing whether to override an existing checkout
  46          * @return boolean
  47          */
  48         public function checkout( $record, $override = null ) {
  49                 $memc = wfGetMainCache();
  50                 $this->validateId( $record );
  51                 $dbw = $this->dbw;
  52                 $userId = $this->user->getId();
  53                 $cacheKey = wfMemcKey( $this->resourceType, $record );
  54
  55                 // when operating with a single memcached cluster, it's reasonable to check the cache here.
  56                 global $wgConcurrencyTrustMemc;
  57                 if( $wgConcurrencyTrustMemc ) {
  58                 $cached = $memc->get( $cacheKey );
  59                         if( $cached ) {
  60                                 if( ! $override && $cached['userId'] != $userId && $cached['expiration'] > time() ) {
  61                                         // this is already checked out.
  62                                         return false;
  63                                 }
  64                         }
  65                 }
  66
  67                 // attempt an insert, check success (this is atomic)
  68                 $insertError = null;
  69                 $res = $dbw->insert(
  70                         'concurrencycheck',
  71                         array(
  72                                 'cc_resource_type' => $this->resourceType,
  73                                 'cc_record' => $record,
  74                                 'cc_user' => $userId,
  75                                 'cc_expiration' => time() + $this->expirationTime,
  76                         ),
  77                         __METHOD__,
  78                         array('IGNORE')
  79                 );
  80
  81                 // if the insert succeeded, checkout is done.
  82                 if( $dbw->affectedRows() === 1 ) {
  83                         // delete any existing cache key.  can't create a new key here
  84                         // since the insert didn't happen inside a transaction.
  85                         $memc->delete( $cacheKey );
  86                         return true;
  87                 }
  88
  89                 // if the insert failed, it's necessary to check the expiration.
  90                 $dbw->begin();
  91                 $row = $dbw->selectRow(
  92                         'concurrencycheck',
  93                         array( 'cc_user', 'cc_expiration' ),
  94                         array(
  95                                 'cc_resource_type' => $this->resourceType,
  96                                 'cc_record' => $record,
  97                         ),
  98                         __METHOD__,
  99                         array()
 100                 );
 101
 102                 // not checked out by current user, checkout is unexpired, override is unset
 103                 if( ! ( $override || $row->cc_user == $userId || $row->cc_expiration <= time() ) ) {
 104                         // this was a cache miss.  populate the cache with data from the db.
 105                         // cache is set to expire at the same time as the checkout, since it'll become invalid then anyway.
 106                         // inside this transaction, a row-level lock is established which ensures cache concurrency
 107                         $memc->set( $cacheKey, array( 'userId' => $row->cc_user, 'expiration' => $row->cc_expiration ), $row->cc_expiration - time() );
 108                         $dbw->rollback();
 109                         return false;
 110                 }
 111
 112                 $expiration = time() + $this->expirationTime;
 113
 114                 // execute a replace
 115                 $res = $dbw->replace(
 116                         'concurrencycheck',
 117                         array( array('cc_resource_type', 'cc_record') ),
 118                         array(
 119                                 'cc_resource_type' => $this->resourceType,
 120                                 'cc_record' => $record,
 121                                 'cc_user' => $userId,
 122                                 'cc_expiration' => $expiration,
 123                         ),
 124                         __METHOD__
 125                 );
 126
 127                 // cache the result.
 128                 $memc->set( $cacheKey, array( 'userId' => $userId, 'expiration' => $expiration ), $this->expirationTime );
 129
 130                 $dbw->commit();
 131                 return true;
 132         }
 133
 134         /**
 135          * Check in a resource. Only works if the resource is checked out by the current user.
 136          *
 137          * @var $record Integer containing the record id to checkin
 138          * @return Boolean
 139          */
 140         public function checkin( $record ) {
 141                 $memc = wfGetMainCache();
 142                 $this->validateId( $record );
 143                 $dbw = $this->dbw;
 144                 $userId = $this->user->getId();
 145                 $cacheKey = wfMemcKey( $this->resourceType, $record );
 146
 147                 $dbw->delete(
 148                         'concurrencycheck',
 149                         array(
 150                                 'cc_resource_type' => $this->resourceType,
 151                                 'cc_record' => $record,
 152                                 'cc_user' => $userId,  // only the owner can perform a checkin
 153                         ),
 154                         __METHOD__,
 155                         array()
 156                 );
 157
 158                 // check row count (this is atomic, select would not be)
 159                 if( $dbw->affectedRows() === 1 ) {
 160                         $memc->delete( $cacheKey );
 161                         return true;
 162                 }
 163
 164                 return false;
 165         }
 166
 167         /**
 168          * Remove all expired checkouts.
 169          *
 170          * @return Integer describing the number of records expired.
 171          */
 172         public function expire() {
 173                 $memc = wfGetMainCache();
 174                 $dbw = $this->dbw;
 175                 $now = time();
 176
 177                 // get the rows to remove from cache.
 178                 $res = $dbw->select(
 179                         'concurrencycheck',
 180                         array( '*' ),
 181                         array(
 182                                 'cc_expiration <= ' . $now,
 183                         ),
 184                         __METHOD__,
 185                         array()
 186                 );
 187
 188                 // build a list of rows to delete.
 189                 $toExpire = array();
 190                 while( $res && $record = $res->fetchRow() ) {
 191                         $toExpire[] = $record['cc_record'];
 192                 }
 193
 194                 // remove the rows from the db
 195                 $dbw->delete(
 196                         'concurrencycheck',
 197                         array(
 198                                 'cc_expiration <= ' . $now,
 199                         ),
 200                         __METHOD__,
 201                         array()
 202                 );
 203
 204                 // delete all those rows from cache
 205                 // outside a transaction because deletes don't require atomicity.
 206                 foreach( $toExpire as $expire ) {
 207                         $memc->delete( wfMemcKey( $this->resourceType, $expire ) );
 208                 }
 209
 210                 // return the number of rows removed.
 211                 return $dbw->affectedRows();
 212         }
 213
 214         public function status( $keys ) {
 215                 $memc = wfGetMainCache();
 216                 $dbw = $this->dbw;
 217                 $now = time();
 218
 219                 $checkouts = array();
 220                 $toSelect = array();
 221
 222                 // validate keys, attempt to retrieve from cache.
 223                 foreach( $keys as $key ) {
 224                         $this->validateId( $key );
 225
 226                         $cached = $memc->get( wfMemcKey( $this->resourceType, $key ) );
 227                         if( $cached && $cached['expiration'] > $now ) {
 228                                 $checkouts[$key] = array(
 229                                         'status' => 'valid',
 230                                         'cc_resource_type' => $this->resourceType,
 231                                         'cc_record' => $key,
 232                                         'cc_user' => $cached['userId'],
 233                                         'cc_expiration' => $cached['expiration'],
 234                                         'cache' => 'cached',
 235                                 );
 236                         } else {
 237                                 $toSelect[] = $key;
 238                         }
 239                 }
 240
 241                 // if there were cache misses...
 242                 if( $toSelect ) {
 243                         // If it's time to go to the database, go ahead and expire old rows.
 244                         $this->expire();
 245
 246                         // the transaction seems incongruous, I know, but it's to keep the cache update atomic.
 247                         $dbw->begin();
 248                         $res = $dbw->select(
 249                                 'concurrencycheck',
 250                                 array( '*' ),
 251                                 array(
 252                                         'cc_resource_type' => $this->resourceType,
 253                                         'cc_record IN (' . implode( ',', $toSelect ) . ')',
 254                                         'cc_expiration > unix_timestamp(now())'
 255                                 ),
 256                                 __METHOD__,
 257                                 array()
 258                         );
 259
 260                         while( $res && $record = $res->fetchRow() ) {
 261                                 $record['status'] = 'valid';
 262                                 $checkouts[ $record['cc_record'] ] = $record;
 263
 264                                 // safe to store values since this is inside the transaction
 265                                 $memc->set(
 266                                         wfMemcKey( $this->resourceType, $record['cc_record'] ),
 267                                         array( 'userId' => $record['cc_user'], 'expiration' => $record['cc_expiration'] ),
 268                                         $record['cc_expiration'] - time()
 269                                 );
 270                         }
 271
 272                         // end the transaction.
 273                         $dbw->rollback();
 274                 }
 275
 276                 // if a key was passed in but has no (unexpired) checkout, include it in the
 277                 // result set to make things easier and more consistent on the client-side.
 278                 foreach( $keys as $key ) {
 279                         if( ! array_key_exists( $key, $checkouts ) ) {
 280                                 $checkouts[$key]['status'] = 'invalid';
 281                         }
 282                 }
 283
 284                 return $checkouts;
 285         }
 286
 287         public function listCheckouts() {
 288                 // TODO: fill in the function that lets you get the complete set of checkouts for a given application.
 289         }
 290
 291         public function setUser ( $user ) {
 292                 $this->user = $user;
 293         }
 294
 295         public function setExpirationTime ( $expirationTime = null ) {
 296                 global $wgConcurrencyExpirationDefault, $wgConcurrencyExpirationMax, $wgConcurrencyExpirationMin;
 297
 298                 // check to make sure the time is digits only, so it can be used in queries
 299                 // negative number are allowed, though mostly only used for testing
 300                 if( $expirationTime && preg_match('/^[\d-]+$/', $expirationTime) ) {
 301                         if( $expirationTime > $wgConcurrencyExpirationMax ) {
 302                                 $this->expirationTime = $wgConcurrencyExpirationMax; // if the number is too high, limit it to the max value.
 303                         } elseif ( $expirationTime < $wgConcurrencyExpirationMin ) {
 304                                 $this->expirationTime = $wgConcurrencyExpirationMin; // low limit, default -1 min
 305                         } else {
 306                                 $this->expirationTime = $expirationTime; // the amount of time before a checkout expires.
 307                         }
 308                 } else {
 309                         $this->expirationTime = $wgConcurrencyExpirationDefault; // global default is 15 mins.
 310                 }
 311         }
 312
 313         /**
 314          * Check to make sure a record ID is numeric, throw an exception if not.
 315          *
 316          * @var $record Integer
 317          * @throws ConcurrencyCheckBadRecordIdException
 318          * @return boolean
 319          */
 320         private static function validateId ( $record ) {
 321                 if( ! preg_match('/^\d+$/', $record) ) {
 322                         throw new ConcurrencyCheckBadRecordIdException( 'Record ID ' . $record . ' must be a positive integer' );
 323                 }
 324
 325                 // TODO: add a hook here for client-side validation.
 326                 return true;
 327         }
 328 }
 329
 330 class ConcurrencyCheckBadRecordIdException extends MWException {};