Skip to content

Commit 306dd67

Browse files
phearnotphearnot
committed
wip
1 parent 6fc92bb commit 306dd67

File tree

4 files changed

+48
-55
lines changed

4 files changed

+48
-55
lines changed

‎node/src/main/scala/com/wavesplatform/mining/Miner.scala

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -309,8 +309,12 @@ class MinerImpl(
309309
}.uncancelable
310310

311311
for {
312-
_ <- waitBlockAppendedTask
312+
elapsed <- waitBlockAppendedTask.timed.map(_._1)
313+
newOffset = (offset - elapsed).max(Duration.Zero)
314+
315+
_ <- Task(microBlockAttempt := SerialCancelable()).delayExecution(newOffset)
313316
result <- Task(forgeBlock(account)).executeOn(minerScheduler)
317+
314318
_ <- result match {
315319
case Right((block, totalConstraint)) =>
316320
appendTask(block, totalConstraint)

‎node/src/main/scala/com/wavesplatform/mining/microblocks/MicroBlockMiner.scala

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@ trait MicroBlockMiner {
1717
account: KeyPair,
1818
accumulatedBlock: Block,
1919
restTotalConstraint: MiningConstraint,
20-
lastMicroBlock: Long
20+
prevMicroBlockTs: Long
2121
): Task[Unit]
2222
}
2323

‎node/src/main/scala/com/wavesplatform/mining/microblocks/MicroBlockMinerImpl.scala

Lines changed: 39 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -45,15 +45,15 @@ class MicroBlockMinerImpl(
4545
account: KeyPair,
4646
accumulatedBlock: Block,
4747
restTotalConstraint: MiningConstraint,
48-
lastMicroBlock: Long
48+
prevMicroBlockTs: Long
4949
): Task[Unit] =
50-
generateOneMicroBlockTask(account, accumulatedBlock, restTotalConstraint, lastMicroBlock)
50+
generateOneMicroBlockTask(account, accumulatedBlock, restTotalConstraint, prevMicroBlockTs)
5151
.flatMap {
5252
case res @ Success(newBlock, newConstraint) =>
5353
Task.defer(generateMicroBlockSequence(account, newBlock, newConstraint, res.nanoTime))
5454
case Retry =>
5555
Task
56-
.defer(generateMicroBlockSequence(account, accumulatedBlock, restTotalConstraint, lastMicroBlock))
56+
.defer(generateMicroBlockSequence(account, accumulatedBlock, restTotalConstraint, prevMicroBlockTs))
5757
.delayExecution(1 second)
5858
case Stop =>
5959
setDebugState(MinerDebugInfo.MiningBlocks)
@@ -65,7 +65,7 @@ class MicroBlockMinerImpl(
6565
account: KeyPair,
6666
accumulatedBlock: Block,
6767
restTotalConstraint: MiningConstraint,
68-
lastMicroBlock: Long
68+
prevMicroBlockTs: Long
6969
): Task[MicroBlockMiningResult] = {
7070
val packTask = Task.cancelable[(Option[Seq[Transaction]], MiningConstraint, Option[ByteStr])] { cb =>
7171
@volatile var cancelled = false
@@ -93,8 +93,8 @@ class MicroBlockMinerImpl(
9393
)
9494
)
9595
)
96-
log.trace(s"Finished pack for ${accumulatedBlock.id()}")
9796
val updatedTotalConstraint = updatedMdConstraint.head
97+
log.trace(s"Finished pack for ${accumulatedBlock.id()}, updated total constraint: $updatedTotalConstraint")
9898
cb.onSuccess((unconfirmed, updatedTotalConstraint, stateHash))
9999
}
100100
Task.eval {
@@ -104,24 +104,25 @@ class MicroBlockMinerImpl(
104104

105105
packTask.flatMap {
106106
case (Some(unconfirmed), updatedTotalConstraint, stateHash) if unconfirmed.nonEmpty =>
107-
val delay = {
108-
val delay = System.nanoTime() - lastMicroBlock
109-
val requiredDelay = settings.microBlockInterval.toNanos
110-
if (delay >= requiredDelay) Duration.Zero else (requiredDelay - delay).nanos
111-
}
112-
113107
for {
114-
_ <- Task.now(if (delay > Duration.Zero) log.trace(s"Sleeping ${delay.toMillis} ms before applying microBlock"))
115-
_ <- Task.sleep(delay)
116-
r <-
117-
if (blockchainUpdater.lastBlockId.forall(_ == accumulatedBlock.id())) {
118-
log.trace(s"Generating microBlock for ${account.toAddress}, constraints: $updatedTotalConstraint")
119-
appendAndBroadcastMicroBlock(account, accumulatedBlock, unconfirmed, updatedTotalConstraint, stateHash)
120-
} else {
121-
log.trace(s"Stopping generating microBlock for ${account.toAddress}, new key block was appended")
122-
Task(Stop)
123-
}
124-
} yield r
108+
blocks <- forgeBlocks(account, accumulatedBlock, unconfirmed, stateHash)
109+
.leftWiden[Throwable]
110+
.liftTo[Task]
111+
(signedBlock, microBlock) = blocks
112+
delay = {
113+
val delay = System.nanoTime() - prevMicroBlockTs
114+
val requiredDelay = settings.microBlockInterval.toNanos
115+
if (delay >= requiredDelay) Duration.Zero else (requiredDelay - delay).nanos
116+
}
117+
_ <-
118+
if (delay > Duration.Zero) {
119+
log.trace(s"Sleeping ${delay.toMillis} ms before applying microBlock")
120+
Task.sleep(delay)
121+
} else Task.unit
122+
_ <- appendMicroBlock(microBlock, account)
123+
} yield
124+
if (updatedTotalConstraint.isFull) Stop
125+
else Success(signedBlock, updatedTotalConstraint)
125126

126127
case (_, updatedTotalConstraint, _) =>
127128
if (updatedTotalConstraint.isFull) {
@@ -139,39 +140,27 @@ class MicroBlockMinerImpl(
139140
}
140141
}
141142

142-
private def appendAndBroadcastMicroBlock(
143-
account: KeyPair,
144-
block: Block,
145-
transactions: Seq[Transaction],
146-
constraint: MiningConstraint,
147-
stateHash: Option[BlockId]
148-
): Task[MicroBlockMiningResult] =
149-
for {
150-
(signedBlock, microBlock) <- forgeBlocks(account, block, transactions, stateHash).leftWiden[Throwable].liftTo[Task]
151-
blockId <- appendMicroBlock(microBlock)
152-
_ = BlockStats.mined(microBlock, blockId)
153-
_ <- broadcastMicroBlock(account, microBlock, blockId)
154-
} yield
155-
if (constraint.isFull) Stop
156-
else Success(signedBlock, constraint)
157-
158-
private def broadcastMicroBlock(account: KeyPair, microBlock: MicroBlock, blockId: BlockId): Task[Unit] =
159-
Task(if (allChannels != null) allChannels.broadcast(MicroBlockInv(account, blockId, microBlock.reference)))
160-
161-
private def appendMicroBlock(microBlock: MicroBlock): Task[BlockId] =
162-
MicroblockAppender(blockchainUpdater, utx, appenderScheduler)(microBlock, None)
163-
.flatMap {
164-
case Left(err) => Task.raiseError(MicroBlockAppendError(microBlock, err))
165-
case Right(v) => Task.now(v)
166-
}
143+
private def appendMicroBlock(microBlock: MicroBlock, account: KeyPair): Task[BlockId] =
144+
MicroblockAppender(blockchainUpdater, utx, appenderScheduler)(microBlock, None).flatMap {
145+
case Left(err) => Task.raiseError(MicroBlockAppendError(microBlock, err))
146+
case Right(blockId) =>
147+
Task.evalAsync {
148+
BlockStats.mined(microBlock, blockId)
149+
if (allChannels != null) {
150+
allChannels.broadcast(MicroBlockInv(account, blockId, microBlock.reference))
151+
}
152+
blockId
153+
}
154+
}.uncancelable
167155

168156
private def forgeBlocks(
169157
account: KeyPair,
170158
accumulatedBlock: Block,
171-
unconfirmed: Seq[Transaction],
159+
packedTxs: Seq[Transaction],
172160
stateHash: Option[ByteStr]
173161
): Either[MicroBlockMiningError, (Block, MicroBlock)] =
174162
microBlockBuildTimeStats.measureSuccessful {
163+
log.trace(s"Forging microBlock for ${account.toAddress}")
175164
for {
176165
signedBlock <- Block
177166
.buildAndSign(
@@ -180,7 +169,7 @@ class MicroBlockMinerImpl(
180169
reference = accumulatedBlock.header.reference,
181170
baseTarget = accumulatedBlock.header.baseTarget,
182171
generationSignature = accumulatedBlock.header.generationSignature,
183-
txs = accumulatedBlock.transactionData ++ unconfirmed,
172+
txs = accumulatedBlock.transactionData ++ packedTxs,
184173
signer = account,
185174
featureVotes = accumulatedBlock.header.featureVotes,
186175
rewardVote = accumulatedBlock.header.rewardVote,
@@ -189,7 +178,7 @@ class MicroBlockMinerImpl(
189178
)
190179
.leftMap(BlockBuildError)
191180
microBlock <- MicroBlock
192-
.buildAndSign(signedBlock.header.version, account, unconfirmed, accumulatedBlock.id(), signedBlock.signature, stateHash)
181+
.buildAndSign(signedBlock.header.version, account, packedTxs, accumulatedBlock.id(), signedBlock.signature, stateHash)
193182
.leftMap(MicroBlockBuildError)
194183
} yield (signedBlock, microBlock)
195184
}

‎node/src/main/scala/com/wavesplatform/network/messages.scala

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -80,9 +80,9 @@ case class MicroBlockInv(sender: PublicKey, totalBlockId: ByteStr, reference: By
8080
}
8181

8282
object MicroBlockInv {
83-
def apply(sender: KeyPair, totalBlockRef: ByteStr, prevBlockRef: ByteStr): MicroBlockInv = {
84-
val signature = crypto.sign(sender.privateKey, sender.toAddress.bytes ++ totalBlockRef.arr ++ prevBlockRef.arr)
85-
new MicroBlockInv(sender.publicKey, totalBlockRef, prevBlockRef, signature)
83+
def apply(sender: KeyPair, totalBlockId: ByteStr, prevBlockRef: ByteStr): MicroBlockInv = {
84+
val signature = crypto.sign(sender.privateKey, sender.toAddress.bytes ++ totalBlockId.arr ++ prevBlockRef.arr)
85+
new MicroBlockInv(sender.publicKey, totalBlockId, prevBlockRef, signature)
8686
}
8787
}
8888

0 commit comments

Comments
 (0)