switch ( $type ) {
case self::ESTIMATE_DB_APPLY:
- $this->ping( $rtt );
- $rttAdjTotal = $this->trxWriteAdjQueryCount * $rtt;
- $applyTime = max( $this->trxWriteAdjDuration - $rttAdjTotal, 0 );
- // For omitted queries, make them count as something at least
- $omitted = $this->trxWriteQueryCount - $this->trxWriteAdjQueryCount;
- $applyTime += self::TINY_WRITE_SEC * $omitted;
-
- return $applyTime;
+ return $this->pingAndCalculateLastTrxApplyTime();
default: // everything
return $this->trxWriteDuration;
}
}
+ /**
+ * @return float Time to apply writes to replicas based on trxWrite* fields
+ */
+ private function pingAndCalculateLastTrxApplyTime() {
+ $this->ping( $rtt );
+
+ $rttAdjTotal = $this->trxWriteAdjQueryCount * $rtt;
+ $applyTime = max( $this->trxWriteAdjDuration - $rttAdjTotal, 0 );
+ // For omitted queries, make them count as something at least
+ $omitted = $this->trxWriteQueryCount - $this->trxWriteAdjQueryCount;
+ $applyTime += self::TINY_WRITE_SEC * $omitted;
+
+ return $applyTime;
+ }
+
public function pendingWriteCallers() {
return $this->trxLevel ? $this->trxWriteCallers : [];
}
// that transactions by themselves don't make changes, only actual writes
// within the transaction matter, which we still detect.
return !preg_match(
- '/^(?:SELECT|BEGIN|ROLLBACK|COMMIT|SET|SHOW|EXPLAIN|\(SELECT)\b/i', $sql );
+ '/^(?:SELECT|BEGIN|ROLLBACK|COMMIT|SAVEPOINT|RELEASE|SET|SHOW|EXPLAIN|\(SELECT)\b/i',
+ $sql
+ );
}
/**
$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;
$options = [ $options ];
}
- $fh = null;
- if ( isset( $options['fileHandle'] ) ) {
- $fh = $options['fileHandle'];
- }
$options = $this->makeInsertOptions( $options );
if ( isset( $a[0] ) && is_array( $a[0] ) ) {
$sql .= '(' . $this->makeList( $a ) . ')';
}
- if ( $fh !== null && false === fwrite( $fh, $sql ) ) {
- return false;
- } elseif ( $fh !== null ) {
- return true;
- }
+ $this->query( $sql, $fname );
- return (bool)$this->query( $sql, $fname );
+ return true;
}
/**
$sql .= " WHERE " . $this->makeList( $conds, self::LIST_AND );
}
- return (bool)$this->query( $sql, $fname );
+ $this->query( $sql, $fname );
+
+ return true;
}
public function makeList( $a, $mode = self::LIST_COMMA ) {
* @param string $table Table name
* @param array|string $rows Row(s) to insert
* @param string $fname Caller function name
- *
- * @return ResultWrapper
*/
protected function nativeReplace( $table, $rows, $fname ) {
$table = $this->tableName( $table );
$sql .= '(' . $this->makeList( $row ) . ')';
}
- return $this->query( $sql, $fname );
+ $this->query( $sql, $fname );
}
public function upsert( $table, array $rows, array $uniqueIndexes, array $set,
$this->startAtomic( $fname, self::ATOMIC_CANCELABLE );
# Update any existing conflicting row(s)
if ( $where !== false ) {
- $ok = $this->update( $table, $set, $where, $fname );
+ $this->update( $table, $set, $where, $fname );
$affectedRowCount += $this->affectedRows();
- } else {
- $ok = true;
}
# Now insert any non-conflicting row(s)
- $ok = $this->insert( $table, $rows, $fname, [ 'IGNORE' ] ) && $ok;
+ $this->insert( $table, $rows, $fname, [ 'IGNORE' ] );
$affectedRowCount += $this->affectedRows();
$this->endAtomic( $fname );
$this->affectedRowCount = $affectedRowCount;
throw $e;
}
- return $ok;
+ return true;
}
public function deleteJoin( $delTable, $joinTable, $delVar, $joinVar, $conds,
$sql .= ' WHERE ' . $conds;
}
- return $this->query( $sql, $fname );
+ $this->query( $sql, $fname );
+
+ return true;
}
final public function insertSelect(
if ( $this->cliMode && $this->isInsertSelectSafe( $insertOptions, $selectOptions ) ) {
// For massive migrations with downtime, we don't want to select everything
// into memory and OOM, so do all this native on the server side if possible.
- return $this->nativeInsertSelect(
+ $this->nativeInsertSelect(
+ $destTable,
+ $srcTable,
+ $varMap,
+ $conds,
+ $fname,
+ array_diff( $insertOptions, $hints ),
+ $selectOptions,
+ $selectJoinConds
+ );
+ } else {
+ $this->nonNativeInsertSelect(
$destTable,
$srcTable,
$varMap,
);
}
- return $this->nonNativeInsertSelect(
- $destTable,
- $srcTable,
- $varMap,
- $conds,
- $fname,
- array_diff( $insertOptions, $hints ),
- $selectOptions,
- $selectJoinConds
- );
+ return true;
}
/**
* @param array $insertOptions
* @param array $selectOptions
* @param array $selectJoinConds
- * @return bool
*/
protected function nonNativeInsertSelect( $destTable, $srcTable, $varMap, $conds,
$fname = __METHOD__,
$srcTable, implode( ',', $fields ), $conds, $fname, $selectOptions, $selectJoinConds
);
if ( !$res ) {
- return false;
+ return;
}
try {
} else {
$this->cancelAtomic( $fname );
}
- return $ok;
} catch ( Exception $e ) {
$this->cancelAtomic( $fname );
throw $e;
* @param array $insertOptions
* @param array $selectOptions
* @param array $selectJoinConds
- * @return bool
*/
protected function nativeInsertSelect( $destTable, $srcTable, $varMap, $conds,
$fname = __METHOD__,
" INTO $destTable (" . implode( ',', array_keys( $varMap ) ) . ') ' .
$selectSql;
- return $this->query( $sql, $fname );
+ $this->query( $sql, $fname );
}
/**
$sectionId = new AtomicSectionIdentifier;
$this->trxAtomicLevels[] = [ $fname, $sectionId, $savepointId ];
+ $this->queryLogger->debug( 'startAtomic: entering level ' .
+ ( count( $this->trxAtomicLevels ) - 1 ) . " ($fname)" );
return $sectionId;
}
// Check if the current section matches $fname
$pos = count( $this->trxAtomicLevels ) - 1;
list( $savedFname, $sectionId, $savepointId ) = $this->trxAtomicLevels[$pos];
+ $this->queryLogger->debug( "endAtomic: leaving level $pos ($fname)" );
if ( $savedFname !== $fname ) {
throw new DBUnexpectedError(
throw new DBUnexpectedError( $this, "No atomic section is open (got $fname)." );
}
+ $excisedFnames = [];
if ( $sectionId !== null ) {
// Find the (last) section with the given $sectionId
$pos = -1;
}
}
if ( $pos < 0 ) {
- throw new DBUnexpectedError( "Atomic section not found (for $fname)" );
+ throw new DBUnexpectedError( $this, "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 ) {
+ $excisedFnames[] = $this->trxAtomicLevels[$i][0];
$excisedIds[] = $this->trxAtomicLevels[$i][1];
}
$this->trxAtomicLevels = array_slice( $this->trxAtomicLevels, 0, $pos + 1 );
$pos = count( $this->trxAtomicLevels ) - 1;
list( $savedFname, $savedSectionId, $savepointId ) = $this->trxAtomicLevels[$pos];
+ if ( $excisedFnames ) {
+ $this->queryLogger->debug( "cancelAtomic: canceling level $pos ($savedFname) " .
+ "and descendants " . implode( ', ', $excisedFnames ) );
+ } else {
+ $this->queryLogger->debug( "cancelAtomic: canceling level $pos ($savedFname)" );
+ }
+
if ( $savedFname !== $fname ) {
throw new DBUnexpectedError(
$this,
// Avoid fatals if close() was called
$this->assertOpen();
- $writeTime = $this->pendingWriteQueryDuration( self::ESTIMATE_DB_APPLY );
$this->doRollback( $fname );
$this->trxStatus = self::STATUS_TRX_NONE;
$this->trxAtomicLevels = [];
+ // Estimate the RTT via a query now that trxStatus is OK
+ $writeTime = $this->pingAndCalculateLastTrxApplyTime();
if ( $this->trxDoneWrites ) {
$this->trxProfiler->transactionWritingOut(
dépôts / lhc / web / wiklou.git / blobdiff