Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Reduce locking in BlockchainUpdater #3979

Draft
wants to merge 2 commits into
base: version-1.5.x
Choose a base branch
from
Draft

Reduce locking in BlockchainUpdater #3979

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3ae3758
wip
phearnot Dec 5, 2024
52f09ea
Merge branch 'version-1.5.x' into reduce-locking-in-updater
phearnot Dec 5, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion node/src/main/scala/com/wavesplatform/Importer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -351,7 +351,7 @@ object Importer extends ScorexLogging {
val utxPool = new UtxPoolImpl(time, blockchainUpdater, settings.utxSettings, settings.maxTxErrorLogSize, settings.minerSettings.enable)
val pos = PoSSelector(blockchainUpdater, settings.synchronizationSettings.maxBaseTarget)
val extAppender: (Block, Option[BlockSnapshotResponse]) => Task[Either[ValidationError, BlockApplyResult]] =
BlockAppender(blockchainUpdater, time, utxPool, pos, scheduler, importOptions.verify, txSignParCheck = false)
BlockAppender(blockchainUpdater, time, utxPool, pos, scheduler, importOptions.verify)

val extensions = initExtensions(settings, blockchainUpdater, scheduler, time, utxPool, rdb)
checkGenesis(settings, blockchainUpdater, Miner.Disabled)
Expand Down
10 changes: 10 additions & 0 deletions node/src/main/scala/com/wavesplatform/database/Caches.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -398,6 +398,16 @@ object Caches {
lazy val hitSource: Option[ByteStr] = meta.map(toHitSource)
}

case class BlockData(
snapshot: StateSnapshot,
carryFee: Long,
totalFee: Long,
reward: Option[Long],
hitSource: ByteStr,
computedBlockStateHash: ByteStr,
block: Block
)

def toHitSource(m: PBBlockMeta): ByteStr = (if (m.vrf.isEmpty) m.getHeader.generationSignature else m.vrf).toByteStr

def toSignedHeader(m: PBBlockMeta): SignedBlockHeader = SignedBlockHeader(PBBlocks.vanilla(m.getHeader), m.signature.toByteStr)
Expand Down
445 changes: 236 additions & 209 deletions node/src/main/scala/com/wavesplatform/state/BlockchainUpdaterImpl.scala
View file
Open in desktop

Large diffs are not rendered by default.

7 changes: 3 additions & 4 deletions node/src/main/scala/com/wavesplatform/state/appender/BlockAppender.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,18 +31,17 @@ object BlockAppender extends ScorexLogging {
utxStorage: UtxPool,
pos: PoSSelector,
scheduler: Scheduler,
verify: Boolean = true,
txSignParCheck: Boolean = true
verify: Boolean = true
)(newBlock: Block, snapshot: Option[BlockSnapshotResponse]): Task[Either[ValidationError, BlockApplyResult]] =
Task {
if (
blockchainUpdater
.isLastBlockId(newBlock.header.reference) || blockchainUpdater.lastBlockHeader.exists(_.header.reference == newBlock.header.reference)
) {
if (newBlock.header.challengedHeader.isDefined) {
appendChallengeBlock(blockchainUpdater, utxStorage, pos, time, log, verify, txSignParCheck)(newBlock, snapshot)
appendChallengeBlock(blockchainUpdater, utxStorage, pos, time, log, verify)(newBlock, snapshot)
} else {
appendKeyBlock(blockchainUpdater, utxStorage, pos, time, log, verify, txSignParCheck)(newBlock, snapshot)
appendKeyBlock(blockchainUpdater, utxStorage, pos, time, log, verify)(newBlock, snapshot)
}
} else if (blockchainUpdater.contains(newBlock.id()) || blockchainUpdater.isLastBlockId(newBlock.id()))
Right(Ignored)
Expand Down
2 changes: 1 addition & 1 deletion node/src/main/scala/com/wavesplatform/state/appender/ExtensionAppender.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@ object ExtensionAppender extends ScorexLogging {
val forkApplicationResultEi = {
newBlocks.view
.map { b =>
b -> appendExtensionBlock(blockchainUpdater, pos, time, verify = true, txSignParCheck = false)(
b -> appendExtensionBlock(blockchainUpdater, pos, time, verify = true)(
b,
extension.snapshots.get(b.id())
)
Expand Down
24 changes: 9 additions & 15 deletions node/src/main/scala/com/wavesplatform/state/appender/package.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,16 +40,15 @@ package object appender {
pos: PoSSelector,
time: Time,
log: LoggerFacade,
verify: Boolean,
txSignParCheck: Boolean
verify: Boolean
)(block: Block, snapshot: Option[BlockSnapshotResponse]): Either[ValidationError, BlockApplyResult] =
for {
hitSource <- if (verify) validateBlock(blockchainUpdater, pos, time)(block) else pos.validateGenerationSignature(block)
newHeight <-
metrics.appendBlock
.measureSuccessful(
blockchainUpdater
.processBlock(block, hitSource, snapshot.map(responseToSnapshot(block, blockchainUpdater.height + 1)), None, verify, txSignParCheck)
.processBlock(block, hitSource, snapshot.map(responseToSnapshot(block, blockchainUpdater.height + 1)), None, verify)
)
.map {
case res @ Applied(discardedDiffs, _) =>
Expand All @@ -71,11 +70,10 @@ package object appender {
blockchainUpdater: BlockchainUpdater & Blockchain,
pos: PoSSelector,
time: Time,
verify: Boolean,
txSignParCheck: Boolean
verify: Boolean
)(block: Block, snapshot: Option[BlockSnapshotResponse]): Either[ValidationError, (BlockApplyResult, Int)] = {
if (block.header.challengedHeader.nonEmpty) {
processBlockWithChallenge(blockchainUpdater, pos, time, verify, txSignParCheck)(block, snapshot)
processBlockWithChallenge(blockchainUpdater, pos, time, verify)(block, snapshot)
} else {
for {
hitSource <- if (verify) validateBlock(blockchainUpdater, pos, time)(block) else pos.validateGenerationSignature(block)
Expand All @@ -85,8 +83,7 @@ package object appender {
hitSource,
snapshot.map(responseToSnapshot(block, blockchainUpdater.height + 1)),
None,
verify,
txSignParCheck
verify
)
)
} yield applyResult -> blockchainUpdater.height
Expand All @@ -99,10 +96,9 @@ package object appender {
pos: PoSSelector,
time: Time,
log: LoggerFacade,
verify: Boolean,
txSignParCheck: Boolean
verify: Boolean
)(block: Block, snapshot: Option[BlockSnapshotResponse]): Either[ValidationError, BlockApplyResult] =
processBlockWithChallenge(blockchainUpdater, pos, time, verify, txSignParCheck)(block, snapshot).map {
processBlockWithChallenge(blockchainUpdater, pos, time, verify)(block, snapshot).map {
case (res @ Applied(discardedDiffs, _), _) =>
if (block.transactionData.nonEmpty) {
utx.removeAll(block.transactionData)
Expand All @@ -120,8 +116,7 @@ package object appender {
blockchainUpdater: BlockchainUpdater & Blockchain,
pos: PoSSelector,
time: Time,
verify: Boolean,
txSignParCheck: Boolean
verify: Boolean
)(block: Block, snapshot: Option[BlockSnapshotResponse]): Either[ValidationError, (BlockApplyResult, Int)] = {
val challengedBlock = block.toOriginal
for {
Expand All @@ -136,8 +131,7 @@ package object appender {
hitSource,
snapshot.map(responseToSnapshot(block, blockchainUpdater.height + 1)),
Some(challengedHitSource),
verify,
txSignParCheck
verify
)
)
} yield applyResult -> blockchainUpdater.height
Expand Down
10 changes: 1 addition & 9 deletions node/src/main/scala/com/wavesplatform/state/diffs/BlockDiffer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,6 @@ object BlockDiffer {
loadCacheData: (Set[Address], Set[ByteStr]) => Unit = (_, _) => (),
verify: Boolean = true,
enableExecutionLog: Boolean = false,
txSignParCheck: Boolean = true
): Either[ValidationError, Result] = {
challengedHitSource match {
case Some(hs) if snapshot.isEmpty =>
Expand All @@ -67,7 +66,6 @@ object BlockDiffer {
loadCacheData,
verify,
enableExecutionLog,
txSignParCheck
).resultE match {
case Left(_: InvalidStateHash) =>
fromBlockTraced(
Expand All @@ -80,7 +78,6 @@ object BlockDiffer {
loadCacheData,
verify,
enableExecutionLog,
txSignParCheck
).resultE
case Left(err) => Left(GenericError(s"Invalid block challenge: $err"))
case _ => Left(GenericError("Invalid block challenge"))
Expand All @@ -96,7 +93,6 @@ object BlockDiffer {
loadCacheData,
verify,
enableExecutionLog,
txSignParCheck
).resultE
}
}
Expand All @@ -111,7 +107,6 @@ object BlockDiffer {
loadCacheData: (Set[Address], Set[ByteStr]) => Unit,
verify: Boolean,
enableExecutionLog: Boolean,
txSignParCheck: Boolean
): TracedResult[ValidationError, Result] = {
val stateHeight = blockchain.height
val heightWithNewBlock = stateHeight + 1
Expand Down Expand Up @@ -202,7 +197,6 @@ object BlockDiffer {
loadCacheData,
verify = verify,
enableExecutionLog = enableExecutionLog,
txSignParCheck = txSignParCheck
)
}
_ <- checkStateHash(blockchainWithNewBlock, block.header.stateHash, r.computedStateHash)
Expand Down Expand Up @@ -269,7 +263,6 @@ object BlockDiffer {
loadCacheData,
verify = verify,
enableExecutionLog = enableExecutionLog,
txSignParCheck = true
)
}
_ <- checkStateHash(blockchain, micro.stateHash, r.computedStateHash)
Expand Down Expand Up @@ -334,15 +327,14 @@ object BlockDiffer {
loadCacheData: (Set[Address], Set[ByteStr]) => Unit,
verify: Boolean,
enableExecutionLog: Boolean,
txSignParCheck: Boolean
): TracedResult[ValidationError, Result] = {
val timestamp = blockchain.lastBlockTimestamp.get
val blockGenerator = blockchain.lastBlockHeader.get.header.generator.toAddress
val rideV6Activated = blockchain.isFeatureActivated(BlockchainFeatures.RideV6)

val txDiffer = TransactionDiffer(prevBlockTimestamp, timestamp, verify, enableExecutionLog = enableExecutionLog) _

if (verify && txSignParCheck)
if (verify)
ParSignatureChecker.checkTxSignatures(txs, rideV6Activated)

prepareCaches(blockGenerator, txs, loadCacheData)
Expand Down
1 change: 0 additions & 1 deletion node/src/main/scala/com/wavesplatform/transaction/BlockchainUpdater.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@ trait BlockchainUpdater {
snapshot: Option[BlockSnapshot],
challengedHitSource: Option[ByteStr] = None,
verify: Boolean = true,
txSignParCheck: Boolean = true
): Either[ValidationError, BlockApplyResult]
def processMicroBlock(microBlock: MicroBlock, snapshot: Option[MicroBlockSnapshot], verify: Boolean = true): Either[ValidationError, BlockId]
def computeNextReward: Option[Long]
Expand Down
3 changes: 1 addition & 2 deletions node/testkit/src/main/scala/com/wavesplatform/db/WithState.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,8 +180,7 @@ trait WithState extends BeforeAndAfterAll with DBCacheSettings with Matchers wit
b.header.generationSignature,
(_, _) => (),
verify = true,
enableExecutionLog = enableExecutionLog,
txSignParCheck = true
enableExecutionLog = enableExecutionLog
)

preconditions.foreach { precondition =>
Expand Down
6 changes: 3 additions & 3 deletions node/tests/src/test/scala/com/wavesplatform/mining/MiningFailuresSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,10 +99,10 @@ class MiningFailuresSuite extends FlatSpec with PathMockFactory with WithNewDBFo
)

var minedBlock: Block = null
(blockchainUpdater.processBlock _).when(*, *, *, *, *, *).returning(Left(BlockFromFuture(100, 100))).repeated(10)
(blockchainUpdater.processBlock _).when(*, *, *, *, *).returning(Left(BlockFromFuture(100, 100))).repeated(10)
(blockchainUpdater.processBlock _)
.when(*, *, *, *, *, *)
.onCall { (block, _, _, _, _, _) =>
.when(*, *, *, *, *)
.onCall { (block, _, _, _, _) =>
minedBlock = block
Right(Applied(Nil, 0))
}
Expand Down
Loading