MediaWiki
master
|
00001 <?php 00034 class FileOpBatch { 00035 /* Timeout related parameters */ 00036 const MAX_BATCH_SIZE = 1000; // integer 00037 00060 public static function attempt( array $performOps, array $opts, FileJournal $journal ) { 00061 wfProfileIn( __METHOD__ ); 00062 $status = Status::newGood(); 00063 00064 $n = count( $performOps ); 00065 if ( $n > self::MAX_BATCH_SIZE ) { 00066 $status->fatal( 'backend-fail-batchsize', $n, self::MAX_BATCH_SIZE ); 00067 wfProfileOut( __METHOD__ ); 00068 return $status; 00069 } 00070 00071 $batchId = $journal->getTimestampedUUID(); 00072 $allowStale = !empty( $opts['allowStale'] ); 00073 $ignoreErrors = !empty( $opts['force'] ); 00074 $journaled = empty( $opts['nonJournaled'] ); 00075 $maxConcurrency = isset( $opts['concurrency'] ) ? $opts['concurrency'] : 1; 00076 00077 $entries = array(); // file journal entry list 00078 $predicates = FileOp::newPredicates(); // account for previous ops in prechecks 00079 $curBatch = array(); // concurrent FileOp sub-batch accumulation 00080 $curBatchDeps = FileOp::newDependencies(); // paths used in FileOp sub-batch 00081 $pPerformOps = array(); // ordered list of concurrent FileOp sub-batches 00082 $lastBackend = null; // last op backend name 00083 // Do pre-checks for each operation; abort on failure... 00084 foreach ( $performOps as $index => $fileOp ) { 00085 $backendName = $fileOp->getBackend()->getName(); 00086 $fileOp->setBatchId( $batchId ); // transaction ID 00087 $fileOp->allowStaleReads( $allowStale ); // consistency level 00088 // Decide if this op can be done concurrently within this sub-batch 00089 // or if a new concurrent sub-batch must be started after this one... 00090 if ( $fileOp->dependsOn( $curBatchDeps ) 00091 || count( $curBatch ) >= $maxConcurrency 00092 || ( $backendName !== $lastBackend && count( $curBatch ) ) 00093 ) { 00094 $pPerformOps[] = $curBatch; // push this batch 00095 $curBatch = array(); // start a new sub-batch 00096 $curBatchDeps = FileOp::newDependencies(); 00097 } 00098 $lastBackend = $backendName; 00099 $curBatch[$index] = $fileOp; // keep index 00100 // Update list of affected paths in this batch 00101 $curBatchDeps = $fileOp->applyDependencies( $curBatchDeps ); 00102 // Simulate performing the operation... 00103 $oldPredicates = $predicates; 00104 $subStatus = $fileOp->precheck( $predicates ); // updates $predicates 00105 $status->merge( $subStatus ); 00106 if ( $subStatus->isOK() ) { 00107 if ( $journaled ) { // journal log entries 00108 $entries = array_merge( $entries, 00109 $fileOp->getJournalEntries( $oldPredicates, $predicates ) ); 00110 } 00111 } else { // operation failed? 00112 $status->success[$index] = false; 00113 ++$status->failCount; 00114 if ( !$ignoreErrors ) { 00115 wfProfileOut( __METHOD__ ); 00116 return $status; // abort 00117 } 00118 } 00119 } 00120 // Push the last sub-batch 00121 if ( count( $curBatch ) ) { 00122 $pPerformOps[] = $curBatch; 00123 } 00124 00125 // Log the operations in the file journal... 00126 if ( count( $entries ) ) { 00127 $subStatus = $journal->logChangeBatch( $entries, $batchId ); 00128 if ( !$subStatus->isOK() ) { 00129 wfProfileOut( __METHOD__ ); 00130 return $subStatus; // abort 00131 } 00132 } 00133 00134 if ( $ignoreErrors ) { // treat precheck() fatals as mere warnings 00135 $status->setResult( true, $status->value ); 00136 } 00137 00138 // Attempt each operation (in parallel if allowed and possible)... 00139 self::runParallelBatches( $pPerformOps, $status ); 00140 00141 wfProfileOut( __METHOD__ ); 00142 return $status; 00143 } 00144 00157 protected static function runParallelBatches( array $pPerformOps, Status $status ) { 00158 $aborted = false; // set to true on unexpected errors 00159 foreach ( $pPerformOps as $performOpsBatch ) { 00160 if ( $aborted ) { // check batch op abort flag... 00161 // We can't continue (even with $ignoreErrors) as $predicates is wrong. 00162 // Log the remaining ops as failed for recovery... 00163 foreach ( $performOpsBatch as $i => $fileOp ) { 00164 $performOpsBatch[$i]->logFailure( 'attempt_aborted' ); 00165 } 00166 continue; 00167 } 00168 $statuses = array(); 00169 $opHandles = array(); 00170 // Get the backend; all sub-batch ops belong to a single backend 00171 $backend = reset( $performOpsBatch )->getBackend(); 00172 // Get the operation handles or actually do it if there is just one. 00173 // If attemptAsync() returns a Status, it was either due to an error 00174 // or the backend does not support async ops and did it synchronously. 00175 foreach ( $performOpsBatch as $i => $fileOp ) { 00176 if ( !$fileOp->failed() ) { // failed => already has Status 00177 // If the batch is just one operation, it's faster to avoid 00178 // pipelining as that can involve creating new TCP connections. 00179 $subStatus = ( count( $performOpsBatch ) > 1 ) 00180 ? $fileOp->attemptAsync() 00181 : $fileOp->attempt(); 00182 if ( $subStatus->value instanceof FileBackendStoreOpHandle ) { 00183 $opHandles[$i] = $subStatus->value; // deferred 00184 } else { 00185 $statuses[$i] = $subStatus; // done already 00186 } 00187 } 00188 } 00189 // Try to do all the operations concurrently... 00190 $statuses = $statuses + $backend->executeOpHandlesInternal( $opHandles ); 00191 // Marshall and merge all the responses (blocking)... 00192 foreach ( $performOpsBatch as $i => $fileOp ) { 00193 if ( !$fileOp->failed() ) { // failed => already has Status 00194 $subStatus = $statuses[$i]; 00195 $status->merge( $subStatus ); 00196 if ( $subStatus->isOK() ) { 00197 $status->success[$i] = true; 00198 ++$status->successCount; 00199 } else { 00200 $status->success[$i] = false; 00201 ++$status->failCount; 00202 $aborted = true; // set abort flag; we can't continue 00203 } 00204 } 00205 } 00206 } 00207 return $status; 00208 } 00209 }