X-Git-Url: https://git.cyclocoop.org/?a=blobdiff_plain;f=includes%2Flibs%2Frdbms%2Fdatabase%2FDatabase.php;h=2b5aaf59da111a147375467ec80a1fff8a948c55;hb=7065200b036d4bbf0c46f4b236d761a79b57215e;hp=5b259bd31dc8187d5bdc71da5f1f478ce8c63c34;hpb=0618227f7f2303a883f5c2b351647f550c345ba0;p=lhc%2Fweb%2Fwiklou.git diff --git a/includes/libs/rdbms/database/Database.php b/includes/libs/rdbms/database/Database.php index 5b259bd31d..2b5aaf59da 100644 --- a/includes/libs/rdbms/database/Database.php +++ b/includes/libs/rdbms/database/Database.php @@ -109,11 +109,11 @@ abstract class Database implements IDatabase, IMaintainableDatabase, LoggerAware /** @var bool */ protected $opened = false; - /** @var array[] List of (callable, method name) */ + /** @var array[] List of (callable, method name, atomic section id) */ protected $trxIdleCallbacks = []; - /** @var array[] List of (callable, method name) */ + /** @var array[] List of (callable, method name, atomic section id) */ protected $trxPreCommitCallbacks = []; - /** @var array[] List of (callable, method name) */ + /** @var array[] List of (callable, method name, atomic section id) */ protected $trxEndCallbacks = []; /** @var callable[] Map of (name => callable) */ protected $trxRecurringCallbacks = []; @@ -212,7 +212,7 @@ abstract class Database implements IDatabase, IMaintainableDatabase, LoggerAware /** * Array of levels of atomicity within transactions * - * @var array + * @var array List of (name, unique ID, savepoint ID) */ private $trxAtomicLevels = []; /** @@ -274,6 +274,11 @@ abstract class Database implements IDatabase, IMaintainableDatabase, LoggerAware /** @var int */ protected $nonNativeInsertSelectBatchSize = 10000; + /** @var string Idiom used when a cancelable atomic section started the transaction */ + private static $NOT_APPLICABLE = 'n/a'; + /** @var string Prefix to the atomic section counter used to make savepoint IDs */ + private static $SAVEPOINT_PREFIX = 'wikimedia_rdbms_atomic'; + /** @var int Transaction is in a error state requiring a full or savepoint rollback */ const STATUS_TRX_ERROR = 1; /** @var int Transaction is active and in a normal state */ @@ -898,34 +903,48 @@ abstract class Database implements IDatabase, IMaintainableDatabase, LoggerAware if ( $this->conn ) { // Resolve any dangling transaction first if ( $this->trxLevel ) { - // Meaningful transactions should ideally have been resolved by now - if ( $this->writesOrCallbacksPending() ) { - $this->queryLogger->warning( - __METHOD__ . ": writes or callbacks still pending.", - [ 'trace' => ( new RuntimeException() )->getTraceAsString() ] - ); + if ( $this->trxAtomicLevels ) { // Cannot let incomplete atomic sections be committed - if ( $this->trxAtomicLevels ) { - $levels = $this->flatAtomicSectionList(); - $exception = new DBUnexpectedError( - $this, - __METHOD__ . ": atomic sections $levels are still open." - ); - // Check if it is possible to properly commit and trigger callbacks - } elseif ( $this->trxEndCallbacksSuppressed ) { + $levels = $this->flatAtomicSectionList(); + $exception = new DBUnexpectedError( + $this, + __METHOD__ . ": atomic sections $levels are still open." + ); + } elseif ( $this->trxAutomatic ) { + // Only the connection manager can commit non-empty DBO_TRX transactions + if ( $this->writesOrCallbacksPending() ) { $exception = new DBUnexpectedError( $this, - __METHOD__ . ': callbacks are suppressed; cannot properly commit.' + __METHOD__ . + ": mass commit/rollback of peer transaction required (DBO_TRX set)." ); } + } elseif ( $this->trxLevel ) { + // Commit explicit transactions as if this was commit() + $this->queryLogger->warning( + __METHOD__ . ": writes or callbacks still pending.", + [ 'trace' => ( new RuntimeException() )->getTraceAsString() ] + ); } + + if ( $this->trxEndCallbacksSuppressed ) { + $exception = $exception ?: new DBUnexpectedError( + $this, + __METHOD__ . ': callbacks are suppressed; cannot properly commit.' + ); + } + // Commit or rollback the changes and run any callbacks as needed if ( $this->trxStatus === self::STATUS_TRX_OK && !$exception ) { - $this->commit( __METHOD__, self::TRANSACTION_INTERNAL ); + $this->commit( + __METHOD__, + $this->trxAutomatic ? self::FLUSHING_INTERNAL : self::FLUSHING_ONE + ); } else { - $this->rollback( __METHOD__, self::TRANSACTION_INTERNAL ); + $this->rollback( __METHOD__, self::FLUSHING_INTERNAL ); } } + // Close the actual connection in the binding handle $closed = $this->closeConnection(); $this->conn = false; @@ -1167,8 +1186,7 @@ abstract class Database implements IDatabase, IMaintainableDatabase, LoggerAware } else { # Nothing prior was there to lose from the transaction, # so just roll it back. - $this->doRollback( __METHOD__ . " ($fname)" ); - $this->trxStatus = self::STATUS_TRX_OK; + $this->rollback( __METHOD__ . " ($fname)", self::FLUSHING_INTERNAL ); } $this->trxStatusIgnoredCause = null; } else { @@ -1293,7 +1311,7 @@ abstract class Database implements IDatabase, IMaintainableDatabase, LoggerAware if ( $this->trxStatus < self::STATUS_TRX_OK ) { throw new DBTransactionStateError( $this, - "Cannot execute query from $fname while transaction status is ERROR. ", + "Cannot execute query from $fname while transaction status is ERROR.", [], $this->trxStatusCause ); @@ -1485,14 +1503,14 @@ abstract class Database implements IDatabase, IMaintainableDatabase, LoggerAware $options = [ $options ]; } - $res = $this->select( $table, $var, $cond, $fname, $options, $join_conds ); + $res = $this->select( $table, [ 'value' => $var ], $cond, $fname, $options, $join_conds ); if ( $res === false ) { return false; } $values = []; foreach ( $res as $row ) { - $values[] = $row->$var; + $values[] = $row->value; } return $values; @@ -3217,7 +3235,7 @@ abstract class Database implements IDatabase, IMaintainableDatabase, LoggerAware if ( !$this->trxLevel ) { throw new DBUnexpectedError( $this, "No transaction is active." ); } - $this->trxEndCallbacks[] = [ $callback, $fname ]; + $this->trxEndCallbacks[] = [ $callback, $fname, $this->currentAtomicSectionId() ]; } final public function onTransactionIdle( callable $callback, $fname = __METHOD__ ) { @@ -3227,7 +3245,7 @@ abstract class Database implements IDatabase, IMaintainableDatabase, LoggerAware $this->trxAutomatic = true; } - $this->trxIdleCallbacks[] = [ $callback, $fname ]; + $this->trxIdleCallbacks[] = [ $callback, $fname, $this->currentAtomicSectionId() ]; if ( !$this->trxLevel ) { $this->runOnTransactionIdleCallbacks( self::TRIGGER_IDLE ); } @@ -3241,7 +3259,7 @@ abstract class Database implements IDatabase, IMaintainableDatabase, LoggerAware } if ( $this->trxLevel ) { - $this->trxPreCommitCallbacks[] = [ $callback, $fname ]; + $this->trxPreCommitCallbacks[] = [ $callback, $fname, $this->currentAtomicSectionId() ]; } else { // No transaction is active nor will start implicitly, so make one for this callback $this->startAtomic( __METHOD__, self::ATOMIC_CANCELABLE ); @@ -3255,6 +3273,72 @@ abstract class Database implements IDatabase, IMaintainableDatabase, LoggerAware } } + /** + * @return AtomicSectionIdentifier|null ID of the topmost atomic section level + */ + private function currentAtomicSectionId() { + if ( $this->trxLevel && $this->trxAtomicLevels ) { + $levelInfo = end( $this->trxAtomicLevels ); + + return $levelInfo[1]; + } + + return null; + } + + /** + * @param AtomicSectionIdentifier $old + * @param AtomicSectionIdentifier $new + */ + private function reassignCallbacksForSection( + AtomicSectionIdentifier $old, AtomicSectionIdentifier $new + ) { + foreach ( $this->trxPreCommitCallbacks as $key => $info ) { + if ( $info[2] === $old ) { + $this->trxPreCommitCallbacks[$key][2] = $new; + } + } + foreach ( $this->trxIdleCallbacks as $key => $info ) { + if ( $info[2] === $old ) { + $this->trxIdleCallbacks[$key][2] = $new; + } + } + foreach ( $this->trxEndCallbacks as $key => $info ) { + if ( $info[2] === $old ) { + $this->trxEndCallbacks[$key][2] = $new; + } + } + } + + /** + * @param AtomicSectionIdentifier[] $sectionIds ID of an actual savepoint + * @throws UnexpectedValueException + */ + private function modifyCallbacksForCancel( array $sectionIds ) { + // Cancel the "on commit" callbacks owned by this savepoint + $this->trxIdleCallbacks = array_filter( + $this->trxIdleCallbacks, + function ( $entry ) use ( $sectionIds ) { + return !in_array( $entry[2], $sectionIds, true ); + } + ); + $this->trxPreCommitCallbacks = array_filter( + $this->trxPreCommitCallbacks, + function ( $entry ) use ( $sectionIds ) { + return !in_array( $entry[2], $sectionIds, true ); + } + ); + // Make "on resolution" callbacks owned by this savepoint to perceive a rollback + foreach ( $this->trxEndCallbacks as $key => $entry ) { + if ( in_array( $entry[2], $sectionIds, true ) ) { + $callback = $entry[0]; + $this->trxEndCallbacks[$key][0] = function () use ( $callback ) { + return $callback( self::TRIGGER_ROLLBACK ); + }; + } + } + } + final public function setTransactionListener( $name, callable $callback = null ) { if ( $callback ) { $this->trxRecurringCallbacks[$name] = $callback; @@ -3428,83 +3512,159 @@ abstract class Database implements IDatabase, IMaintainableDatabase, LoggerAware $this->query( 'ROLLBACK TO SAVEPOINT ' . $this->addIdentifierQuotes( $identifier ), $fname ); } + /** + * @param string $fname + * @return string + */ + private function nextSavepointId( $fname ) { + $savepointId = self::$SAVEPOINT_PREFIX . ++$this->trxAtomicCounter; + if ( strlen( $savepointId ) > 30 ) { + // 30 == Oracle's identifier length limit (pre 12c) + // With a 22 character prefix, that puts the highest number at 99999999. + throw new DBUnexpectedError( + $this, + 'There have been an excessively large number of atomic sections in a transaction' + . " started by $this->trxFname (at $fname)" + ); + } + + return $savepointId; + } + final public function startAtomic( $fname = __METHOD__, $cancelable = self::ATOMIC_NOT_CANCELABLE ) { - $savepointId = $cancelable === self::ATOMIC_CANCELABLE ? 'n/a' : null; + $savepointId = $cancelable === self::ATOMIC_CANCELABLE ? self::$NOT_APPLICABLE : null; + if ( !$this->trxLevel ) { $this->begin( $fname, self::TRANSACTION_INTERNAL ); // If DBO_TRX is set, a series of startAtomic/endAtomic pairs will result // in all changes being in one transaction to keep requests transactional. - if ( !$this->getFlag( self::DBO_TRX ) ) { + if ( $this->getFlag( self::DBO_TRX ) ) { + // Since writes could happen in between the topmost atomic sections as part + // of the transaction, those sections will need savepoints. + $savepointId = $this->nextSavepointId( $fname ); + $this->doSavepoint( $savepointId, $fname ); + } else { $this->trxAutomaticAtomic = true; } } elseif ( $cancelable === self::ATOMIC_CANCELABLE ) { - $savepointId = 'wikimedia_rdbms_atomic' . ++$this->trxAtomicCounter; - if ( strlen( $savepointId ) > 30 ) { // 30 == Oracle's identifier length limit (pre 12c) - $this->queryLogger->warning( - 'There have been an excessively large number of atomic sections in a transaction' - . " started by $this->trxFname, reusing IDs (at $fname)", - [ 'trace' => ( new RuntimeException() )->getTraceAsString() ] - ); - $this->trxAtomicCounter = 0; - $savepointId = 'wikimedia_rdbms_atomic' . ++$this->trxAtomicCounter; - } + $savepointId = $this->nextSavepointId( $fname ); $this->doSavepoint( $savepointId, $fname ); } - $this->trxAtomicLevels[] = [ $fname, $savepointId ]; + $sectionId = new AtomicSectionIdentifier; + $this->trxAtomicLevels[] = [ $fname, $sectionId, $savepointId ]; + + return $sectionId; } final public function endAtomic( $fname = __METHOD__ ) { - if ( !$this->trxLevel ) { - throw new DBUnexpectedError( $this, "No atomic transaction is open (got $fname)." ); + if ( !$this->trxLevel || !$this->trxAtomicLevels ) { + throw new DBUnexpectedError( $this, "No atomic section is open (got $fname)." ); } - list( $savedFname, $savepointId ) = $this->trxAtomicLevels - ? array_pop( $this->trxAtomicLevels ) : [ null, null ]; + // Check if the current section matches $fname + $pos = count( $this->trxAtomicLevels ) - 1; + list( $savedFname, $sectionId, $savepointId ) = $this->trxAtomicLevels[$pos]; + if ( $savedFname !== $fname ) { - throw new DBUnexpectedError( $this, "Invalid atomic section ended (got $fname)." ); + throw new DBUnexpectedError( + $this, + "Invalid atomic section ended (got $fname but expected $savedFname)." + ); } + // Remove the last section (no need to re-index the array) + array_pop( $this->trxAtomicLevels ); + if ( !$this->trxAtomicLevels && $this->trxAutomaticAtomic ) { $this->commit( $fname, self::FLUSHING_INTERNAL ); - } elseif ( $savepointId && $savepointId !== 'n/a' ) { + } elseif ( $savepointId !== null && $savepointId !== self::$NOT_APPLICABLE ) { $this->doReleaseSavepoint( $savepointId, $fname ); } + + // Hoist callback ownership for callbacks in the section that just ended; + // all callbacks should have an owner that is present in trxAtomicLevels. + $currentSectionId = $this->currentAtomicSectionId(); + if ( $currentSectionId ) { + $this->reassignCallbacksForSection( $sectionId, $currentSectionId ); + } } - final public function cancelAtomic( $fname = __METHOD__ ) { - if ( !$this->trxLevel ) { - throw new DBUnexpectedError( $this, "No atomic transaction is open (got $fname)." ); + final public function cancelAtomic( + $fname = __METHOD__, AtomicSectionIdentifier $sectionId = null + ) { + if ( !$this->trxLevel || !$this->trxAtomicLevels ) { + throw new DBUnexpectedError( $this, "No atomic section is open (got $fname)." ); } - list( $savedFname, $savepointId ) = $this->trxAtomicLevels - ? array_pop( $this->trxAtomicLevels ) : [ null, null ]; - if ( $savedFname !== $fname ) { - throw new DBUnexpectedError( $this, "Invalid atomic section ended (got $fname)." ); + if ( $sectionId !== null ) { + // Find the (last) section with the given $sectionId + $pos = -1; + foreach ( $this->trxAtomicLevels as $i => list( $asFname, $asId, $spId ) ) { + if ( $asId === $sectionId ) { + $pos = $i; + } + } + if ( $pos < 0 ) { + throw new DBUnexpectedError( "Atomic section not found (for $fname)" ); + } + // Remove all descendant sections and re-index the array + $excisedIds = []; + $len = count( $this->trxAtomicLevels ); + for ( $i = $pos + 1; $i < $len; ++$i ) { + $excisedIds[] = $this->trxAtomicLevels[$i][1]; + } + $this->trxAtomicLevels = array_slice( $this->trxAtomicLevels, 0, $pos + 1 ); + $this->modifyCallbacksForCancel( $excisedIds ); } - if ( !$savepointId ) { - throw new DBUnexpectedError( $this, "Uncancelable atomic section canceled (got $fname)." ); + + // Check if the current section matches $fname + $pos = count( $this->trxAtomicLevels ) - 1; + list( $savedFname, $savedSectionId, $savepointId ) = $this->trxAtomicLevels[$pos]; + + if ( $savedFname !== $fname ) { + throw new DBUnexpectedError( + $this, + "Invalid atomic section ended (got $fname but expected $savedFname)." + ); } - if ( !$this->trxAtomicLevels && $this->trxAutomaticAtomic ) { - $this->rollback( $fname, self::FLUSHING_INTERNAL ); - } elseif ( $savepointId !== 'n/a' ) { - $this->doRollbackToSavepoint( $savepointId, $fname ); - $this->trxStatus = self::STATUS_TRX_OK; // no exception; recovered - $this->trxStatusIgnoredCause = null; + // Remove the last section (no need to re-index the array) + array_pop( $this->trxAtomicLevels ); + $this->modifyCallbacksForCancel( [ $savedSectionId ] ); + + if ( $savepointId !== null ) { + // Rollback the transaction to the state just before this atomic section + if ( $savepointId === self::$NOT_APPLICABLE ) { + $this->rollback( $fname, self::FLUSHING_INTERNAL ); + } else { + $this->doRollbackToSavepoint( $savepointId, $fname ); + $this->trxStatus = self::STATUS_TRX_OK; // no exception; recovered + $this->trxStatusIgnoredCause = null; + } + } elseif ( $this->trxStatus > self::STATUS_TRX_ERROR ) { + // Put the transaction into an error state if it's not already in one + $this->trxStatus = self::STATUS_TRX_ERROR; + $this->trxStatusCause = new DBUnexpectedError( + $this, + "Uncancelable atomic section canceled (got $fname)." + ); } $this->affectedRowCount = 0; // for the sake of consistency } - final public function doAtomicSection( $fname, callable $callback ) { - $this->startAtomic( $fname, self::ATOMIC_CANCELABLE ); + final public function doAtomicSection( + $fname, callable $callback, $cancelable = self::ATOMIC_NOT_CANCELABLE + ) { + $sectionId = $this->startAtomic( $fname, $cancelable ); try { $res = call_user_func_array( $callback, [ $this, $fname ] ); } catch ( Exception $e ) { - $this->cancelAtomic( $fname ); + $this->cancelAtomic( $fname, $sectionId ); + throw $e; } $this->endAtomic( $fname ); @@ -3513,6 +3673,11 @@ abstract class Database implements IDatabase, IMaintainableDatabase, LoggerAware } final public function begin( $fname = __METHOD__, $mode = self::TRANSACTION_EXPLICIT ) { + static $modes = [ self::TRANSACTION_EXPLICIT, self::TRANSACTION_INTERNAL ]; + if ( !in_array( $mode, $modes, true ) ) { + throw new DBUnexpectedError( $this, "$fname: invalid mode parameter '$mode'." ); + } + // Protect against mismatched atomic section, transaction nesting, and snapshot loss if ( $this->trxLevel ) { if ( $this->trxAtomicLevels ) { @@ -3571,9 +3736,14 @@ abstract class Database implements IDatabase, IMaintainableDatabase, LoggerAware $this->trxLevel = 1; } - final public function commit( $fname = __METHOD__, $flush = '' ) { + final public function commit( $fname = __METHOD__, $flush = self::FLUSHING_ONE ) { + static $modes = [ self::FLUSHING_ONE, self::FLUSHING_ALL_PEERS, self::FLUSHING_INTERNAL ]; + if ( !in_array( $flush, $modes, true ) ) { + throw new DBUnexpectedError( $this, "$fname: invalid flush parameter '$flush'." ); + } + if ( $this->trxLevel && $this->trxAtomicLevels ) { - // There are still atomic sections open. This cannot be ignored + // There are still atomic sections open; this cannot be ignored $levels = $this->flatAtomicSectionList(); throw new DBUnexpectedError( $this,