X-Git-Url: http://git.heureux-cyclage.org/?p=lhc%2Fweb%2Fwiklou.git;a=blobdiff_plain;f=includes%2Flibs%2Frdbms%2Fdatabase%2FDatabase.php;h=ab4ab688275cd3f207a550cd79fb14bd92ec0f50;hp=f6817950ed690c164935e8c65ca3045211e0cec7;hb=6fd8b71064f0d14ec3e44e20d69e60f8505ba3ba;hpb=193b6f8d2ebd5dee97946c3fc6358f4f0e7e26fd diff --git a/includes/libs/rdbms/database/Database.php b/includes/libs/rdbms/database/Database.php index f6817950ed..ab4ab68827 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 = []; @@ -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 */ @@ -1181,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 { @@ -1499,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; @@ -2231,7 +2235,7 @@ abstract class Database implements IDatabase, IMaintainableDatabase, LoggerAware if ( preg_match( '/(^|\s)(DISTINCT|JOIN|ON|AS)(\s|$)/i', $name ) !== 0 ) { $this->queryLogger->warning( __METHOD__ . ": use of subqueries is not supported this way.", - [ 'trace' => ( new RuntimeException() )->getTraceAsString() ] + [ 'exception' => new RuntimeException() ] ); return $name; @@ -3231,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__ ) { @@ -3241,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 ); } @@ -3255,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 ); @@ -3269,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; @@ -3442,28 +3512,44 @@ 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 ); } @@ -3480,7 +3566,7 @@ abstract class Database implements IDatabase, IMaintainableDatabase, LoggerAware // Check if the current section matches $fname $pos = count( $this->trxAtomicLevels ) - 1; - list( $savedFname, , $savepointId ) = $this->trxAtomicLevels[$pos]; + list( $savedFname, $sectionId, $savepointId ) = $this->trxAtomicLevels[$pos]; if ( $savedFname !== $fname ) { throw new DBUnexpectedError( @@ -3489,14 +3575,21 @@ abstract class Database implements IDatabase, IMaintainableDatabase, LoggerAware ); } - // Remove the last section and re-index the array - $this->trxAtomicLevels = array_slice( $this->trxAtomicLevels, 0, $pos ); + // 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 !== null && $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( @@ -3518,12 +3611,18 @@ abstract class Database implements IDatabase, IMaintainableDatabase, LoggerAware 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 ); } // Check if the current section matches $fname $pos = count( $this->trxAtomicLevels ) - 1; - list( $savedFname, , $savepointId ) = $this->trxAtomicLevels[$pos]; + list( $savedFname, $savedSectionId, $savepointId ) = $this->trxAtomicLevels[$pos]; if ( $savedFname !== $fname ) { throw new DBUnexpectedError( @@ -3532,12 +3631,13 @@ abstract class Database implements IDatabase, IMaintainableDatabase, LoggerAware ); } - // Remove the last section and re-index the array - $this->trxAtomicLevels = array_slice( $this->trxAtomicLevels, 0, $pos ); + // 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 === 'n/a' ) { + if ( $savepointId === self::$NOT_APPLICABLE ) { $this->rollback( $fname, self::FLUSHING_INTERNAL ); } else { $this->doRollbackToSavepoint( $savepointId, $fname ); @@ -3894,7 +3994,7 @@ abstract class Database implements IDatabase, IMaintainableDatabase, LoggerAware $fname . ': lost connection to {dbserver}; reconnected', [ 'dbserver' => $this->getServer(), - 'trace' => ( new RuntimeException() )->getTraceAsString() + 'exception' => new RuntimeException() ] ); } catch ( DBConnectionError $e ) { @@ -4410,8 +4510,8 @@ abstract class Database implements IDatabase, IMaintainableDatabase, LoggerAware */ public function __clone() { $this->connLogger->warning( - "Cloning " . static::class . " is not recomended; forking connection:\n" . - ( new RuntimeException() )->getTraceAsString() + "Cloning " . static::class . " is not recommended; forking connection", + [ 'exception' => new RuntimeException() ] ); if ( $this->isOpen() ) {