includes/filerepo/backend/FileOpBatch.php

   1 <?php
   2 /**
   3  * @file
   4  * @ingroup FileBackend
   5  * @author Aaron Schulz
   6  */
   7
   8 /**
   9  * Helper class for representing batch file operations.
  10  * Do not use this class from places outside FileBackend.
  11  *
  12  * Methods should avoid throwing exceptions at all costs.
  13  *
  14  * @ingroup FileBackend
  15  * @since 1.20
  16  */
  17 class FileOpBatch {
  18         /* Timeout related parameters */
  19         const MAX_BATCH_SIZE = 1000; // integer
  20
  21         /**
  22          * Attempt to perform a series of file operations.
  23          * Callers are responsible for handling file locking.
  24          *
  25          * $opts is an array of options, including:
  26          * 'force'        : Errors that would normally cause a rollback do not.
  27          *                  The remaining operations are still attempted if any fail.
  28          * 'allowStale'   : Don't require the latest available data.
  29          *                  This can increase performance for non-critical writes.
  30          *                  This has no effect unless the 'force' flag is set.
  31          * 'nonJournaled' : Don't log this operation batch in the file journal.
  32          * 'concurrency'  : Try to do this many operations in parallel when possible.
  33          *
  34          * The resulting Status will be "OK" unless:
  35          *     a) unexpected operation errors occurred (network partitions, disk full...)
  36          *     b) significant operation errors occured and 'force' was not set
  37          *
  38          * @param $performOps Array List of FileOp operations
  39          * @param $opts Array Batch operation options
  40          * @param $journal FileJournal Journal to log operations to
  41          * @return Status
  42          */
  43         public static function attempt( array $performOps, array $opts, FileJournal $journal ) {
  44                 wfProfileIn( __METHOD__ );
  45                 $status = Status::newGood();
  46
  47                 $n = count( $performOps );
  48                 if ( $n > self::MAX_BATCH_SIZE ) {
  49                         $status->fatal( 'backend-fail-batchsize', $n, self::MAX_BATCH_SIZE );
  50                         wfProfileOut( __METHOD__ );
  51                         return $status;
  52                 }
  53
  54                 $batchId = $journal->getTimestampedUUID();
  55                 $allowStale = !empty( $opts['allowStale'] );
  56                 $ignoreErrors = !empty( $opts['force'] );
  57                 $journaled = empty( $opts['nonJournaled'] );
  58                 $maxConcurrency = isset( $opts['concurrency'] ) ? $opts['concurrency'] : 1;
  59
  60                 $entries = array(); // file journal entry list
  61                 $predicates = FileOp::newPredicates(); // account for previous ops in prechecks
  62                 $curBatch = array(); // concurrent FileOp sub-batch accumulation
  63                 $curBatchDeps = FileOp::newDependencies(); // paths used in FileOp sub-batch
  64                 $pPerformOps = array(); // ordered list of concurrent FileOp sub-batches
  65                 $lastBackend = null; // last op backend name
  66                 // Do pre-checks for each operation; abort on failure...
  67                 foreach ( $performOps as $index => $fileOp ) {
  68                         $backendName = $fileOp->getBackend()->getName();
  69                         $fileOp->setBatchId( $batchId ); // transaction ID
  70                         $fileOp->allowStaleReads( $allowStale ); // consistency level
  71                         // Decide if this op can be done concurrently within this sub-batch
  72                         // or if a new concurrent sub-batch must be started after this one...
  73                         if ( $fileOp->dependsOn( $curBatchDeps )
  74                                 || count( $curBatch ) >= $maxConcurrency
  75                                 || ( $backendName !== $lastBackend && count( $curBatch ) )
  76                         ) {
  77                                 $pPerformOps[] = $curBatch; // push this batch
  78                                 $curBatch = array(); // start a new sub-batch
  79                                 $curBatchDeps = FileOp::newDependencies();
  80                         }
  81                         $lastBackend = $backendName;
  82                         $curBatch[$index] = $fileOp; // keep index
  83                         // Update list of affected paths in this batch
  84                         $curBatchDeps = $fileOp->applyDependencies( $curBatchDeps );
  85                         // Simulate performing the operation...
  86                         $oldPredicates = $predicates;
  87                         $subStatus = $fileOp->precheck( $predicates ); // updates $predicates
  88                         $status->merge( $subStatus );
  89                         if ( $subStatus->isOK() ) {
  90                                 if ( $journaled ) { // journal log entries
  91                                         $entries = array_merge( $entries,
  92                                                 $fileOp->getJournalEntries( $oldPredicates, $predicates ) );
  93                                 }
  94                         } else { // operation failed?
  95                                 $status->success[$index] = false;
  96                                 ++$status->failCount;
  97                                 if ( !$ignoreErrors ) {
  98                                         wfProfileOut( __METHOD__ );
  99                                         return $status; // abort
 100                                 }
 101                         }
 102                 }
 103                 // Push the last sub-batch
 104                 if ( count( $curBatch ) ) {
 105                         $pPerformOps[] = $curBatch;
 106                 }
 107
 108                 // Log the operations in the file journal...
 109                 if ( count( $entries ) ) {
 110                         $subStatus = $journal->logChangeBatch( $entries, $batchId );
 111                         if ( !$subStatus->isOK() ) {
 112                                 wfProfileOut( __METHOD__ );
 113                                 return $subStatus; // abort
 114                         }
 115                 }
 116
 117                 if ( $ignoreErrors ) { // treat precheck() fatals as mere warnings
 118                         $status->setResult( true, $status->value );
 119                 }
 120
 121                 // Attempt each operation (in parallel if allowed and possible)...
 122                 if ( count( $pPerformOps ) < count( $performOps ) ) {
 123                         self::runBatchParallel( $pPerformOps, $status );
 124                 } else {
 125                         self::runBatchSeries( $performOps, $status );
 126                 }
 127
 128                 wfProfileOut( __METHOD__ );
 129                 return $status;
 130         }
 131
 132         /**
 133          * Attempt a list of file operations in series.
 134          * This will abort remaining ops on failure.
 135          *
 136          * @param $performOps Array
 137          * @param $status Status
 138          * @return bool Success
 139          */
 140         protected static function runBatchSeries( array $performOps, Status $status ) {
 141                 foreach ( $performOps as $index => $fileOp ) {
 142                         if ( $fileOp->failed() ) {
 143                                 continue; // nothing to do
 144                         }
 145                         $subStatus = $fileOp->attempt();
 146                         $status->merge( $subStatus );
 147                         if ( $subStatus->isOK() ) {
 148                                 $status->success[$index] = true;
 149                                 ++$status->successCount;
 150                         } else {
 151                                 $status->success[$index] = false;
 152                                 ++$status->failCount;
 153                                 // We can't continue (even with $ignoreErrors) as $predicates is wrong.
 154                                 // Log the remaining ops as failed for recovery...
 155                                 for ( $i = ($index + 1); $i < count( $performOps ); $i++ ) {
 156                                         $performOps[$i]->logFailure( 'attempt_aborted' );
 157                                 }
 158                                 return false; // bail out
 159                         }
 160                 }
 161                 return true;
 162         }
 163
 164         /**
 165          * Attempt a list of file operations sub-batches in series.
 166          *
 167          * The operations *in* each sub-batch will be done in parallel.
 168          * The caller is responsible for making sure the operations
 169          * within any given sub-batch do not depend on each other.
 170          * This will abort remaining ops on failure.
 171          *
 172          * @param $performOps Array
 173          * @param $status Status
 174          * @return bool Success
 175          */
 176         protected static function runBatchParallel( array $pPerformOps, Status $status ) {
 177                 $aborted = false;
 178                 foreach ( $pPerformOps as $performOpsBatch ) {
 179                         if ( $aborted ) { // check batch op abort flag...
 180                                 // We can't continue (even with $ignoreErrors) as $predicates is wrong.
 181                                 // Log the remaining ops as failed for recovery...
 182                                 foreach ( $performOpsBatch as $i => $fileOp ) {
 183                                         $performOpsBatch[$i]->logFailure( 'attempt_aborted' );
 184                                 }
 185                                 continue;
 186                         }
 187                         $statuses = array();
 188                         $opHandles = array();
 189                         // Get the backend; all sub-batch ops belong to a single backend
 190                         $backend = reset( $performOpsBatch )->getBackend();
 191                         // If attemptAsync() returns synchronously, it was either an
 192                         // error Status or the backend just doesn't support async ops.
 193                         foreach ( $performOpsBatch as $i => $fileOp ) {
 194                                 if ( !$fileOp->failed() ) { // failed => already has Status
 195                                         $subStatus = $fileOp->attemptAsync();
 196                                         if ( $subStatus->value instanceof FileBackendStoreOpHandle ) {
 197                                                 $opHandles[$i] = $subStatus->value; // deferred
 198                                         } else {
 199                                                 $statuses[$i] = $subStatus; // done already
 200                                         }
 201                                 }
 202                         }
 203                         // Try to do all the operations concurrently...
 204                         $statuses = $statuses + $backend->executeOpHandlesInternal( $opHandles );
 205                         // Marshall and merge all the responses (blocking)...
 206                         foreach ( $performOpsBatch as $i => $fileOp ) {
 207                                 if ( !$fileOp->failed() ) { // failed => already has Status
 208                                         $subStatus = $statuses[$i];
 209                                         $status->merge( $subStatus );
 210                                         if ( $subStatus->isOK() ) {
 211                                                 $status->success[$i] = true;
 212                                                 ++$status->successCount;
 213                                         } else {
 214                                                 $status->success[$i] = false;
 215                                                 ++$status->failCount;
 216                                                 $aborted = true; // set abort flag; we can't continue
 217                                         }
 218                                 }
 219                         }
 220                 }
 221                 return $status;
 222         }
 223 }