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