9 * Helper class for representing batch file operations.
10 * Do not use this class from places outside FileBackend.
12 * Methods should avoid throwing exceptions at all costs.
14 * @ingroup FileBackend
18 /* Timeout related parameters */
19 const MAX_BATCH_SIZE
= 1000; // integer
22 * Attempt to perform a series of file operations.
23 * Callers are responsible for handling file locking.
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.
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
38 * @param $performOps Array List of FileOp operations
39 * @param $opts Array Batch operation options
40 * @param $journal FileJournal Journal to log operations to
43 public static function attempt( array $performOps, array $opts, FileJournal
$journal ) {
44 wfProfileIn( __METHOD__
);
45 $status = Status
::newGood();
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__
);
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;
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 ) )
77 $pPerformOps[] = $curBatch; // push this batch
78 $curBatch = array(); // start a new sub-batch
79 $curBatchDeps = FileOp
::newDependencies();
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 ) );
94 } else { // operation failed?
95 $status->success
[$index] = false;
97 if ( !$ignoreErrors ) {
98 wfProfileOut( __METHOD__
);
99 return $status; // abort
103 // Push the last sub-batch
104 if ( count( $curBatch ) ) {
105 $pPerformOps[] = $curBatch;
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
117 if ( $ignoreErrors ) { // treat precheck() fatals as mere warnings
118 $status->setResult( true, $status->value
);
121 // Attempt each operation (in parallel if allowed and possible)...
122 if ( count( $pPerformOps ) < count( $performOps ) ) {
123 self
::runBatchParallel( $pPerformOps, $status );
125 self
::runBatchSeries( $performOps, $status );
128 wfProfileOut( __METHOD__
);
133 * Attempt a list of file operations in series.
134 * This will abort remaining ops on failure.
136 * @param $performOps Array
137 * @param $status Status
138 * @return bool Success
140 protected static function runBatchSeries( array $performOps, Status
$status ) {
141 foreach ( $performOps as $index => $fileOp ) {
142 if ( $fileOp->failed() ) {
143 continue; // nothing to do
145 $subStatus = $fileOp->attempt();
146 $status->merge( $subStatus );
147 if ( $subStatus->isOK() ) {
148 $status->success
[$index] = true;
149 ++
$status->successCount
;
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' );
158 return false; // bail out
165 * Attempt a list of file operations sub-batches in series.
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.
172 * @param $performOps Array
173 * @param $status Status
174 * @return bool Success
176 protected static function runBatchParallel( array $pPerformOps, Status
$status ) {
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' );
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
199 $statuses[$i] = $subStatus; // done already
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
;
214 $status->success
[$i] = false;
215 ++
$status->failCount
;
216 $aborted = true; // set abort flag; we can't continue