Skip to content

Erigon 3: Parallel state flushing #11300

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

Merged
Giulio2002 merged 4 commits into from
Jul 25, 2024
Merged

Erigon 3: Parallel state flushing #11300

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
e9024f4
parrallel flushing
Giulio2002 Jul 23, 2024
2124247
parrallel flushing
Giulio2002 Jul 23, 2024
fd52c04
save
Giulio2002 Jul 24, 2024
3a951a0
save
Giulio2002 Jul 24, 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
109 changes: 62 additions & 47 deletions turbo/execution/eth1/forkchoice.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,14 +46,22 @@ type forkchoiceOutcome struct {
err error
}

func sendForkchoiceReceiptWithoutWaiting(ch chan forkchoiceOutcome, receipt *execution.ForkChoiceReceipt) {
func sendForkchoiceReceiptWithoutWaiting(ch chan forkchoiceOutcome, receipt *execution.ForkChoiceReceipt, alreadySent bool) {
if alreadySent {
return
}
select {
case ch <- forkchoiceOutcome{receipt: receipt}:
default:
}
}

func sendForkchoiceErrorWithoutWaiting(ch chan forkchoiceOutcome, err error) {
func sendForkchoiceErrorWithoutWaiting(logger log.Logger, ch chan forkchoiceOutcome, err error, alreadySent bool) {
if alreadySent {
logger.Warn("forkchoice: error received after result was sent", "error", err)
return
}

select {
case ch <- forkchoiceOutcome{err: err}:
default:
Expand Down Expand Up @@ -142,13 +150,13 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original
sendForkchoiceReceiptWithoutWaiting(outcomeCh, &execution.ForkChoiceReceipt{
LatestValidHash: gointerfaces.ConvertHashToH256(common.Hash{}),
Status: execution.ExecutionStatus_Busy,
})
}, false)
return
}
defer e.semaphore.Release(1)

//if err := stages2.ProcessFrozenBlocks(ctx, e.db, e.blockReader, e.executionPipeline); err != nil {
// sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
// sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
// e.logger.Warn("ProcessFrozenBlocks", "error", err)
// return
//}
Expand All @@ -163,7 +171,7 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original
}
return rawdb.WriteLastNewBlockSeen(tx, *num)
}); err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}

Expand All @@ -174,7 +182,7 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original
}
tx, err := e.db.BeginRwNosync(ctx)
if err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}
defer tx.Rollback()
Expand All @@ -183,23 +191,23 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original

finishProgressBefore, err := stages.GetStageProgress(tx, stages.Finish)
if err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}
headersProgressBefore, err := stages.GetStageProgress(tx, stages.Headers)
if err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}

// Step one, find reconnection point, and mark all of those headers as canonical.
fcuHeader, err := e.blockReader.HeaderByHash(ctx, tx, originalBlockHash)
if err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}
if fcuHeader == nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, fmt.Errorf("forkchoice: block %x not found or was marked invalid", blockHash))
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, fmt.Errorf("forkchoice: block %x not found or was marked invalid", blockHash), false)
return
}

Expand All @@ -212,7 +220,7 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original

canonicalHash, err := rawdb.ReadCanonicalHash(tx, fcuHeader.Number.Uint64())
if err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}
if fcuHeader.Number.Uint64() > 0 {
Expand All @@ -221,20 +229,20 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original
writeForkChoiceHashes(tx, blockHash, safeHash, finalizedHash)
valid, err := e.verifyForkchoiceHashes(ctx, tx, blockHash, finalizedHash, safeHash)
if err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}
if !valid {
sendForkchoiceReceiptWithoutWaiting(outcomeCh, &execution.ForkChoiceReceipt{
LatestValidHash: gointerfaces.ConvertHashToH256(common.Hash{}),
Status: execution.ExecutionStatus_InvalidForkchoice,
})
}, false)
return
}
sendForkchoiceReceiptWithoutWaiting(outcomeCh, &execution.ForkChoiceReceipt{
LatestValidHash: gointerfaces.ConvertHashToH256(blockHash),
Status: execution.ExecutionStatus_Success,
})
}, false)
return
}

Expand All @@ -243,15 +251,15 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original
sendForkchoiceReceiptWithoutWaiting(outcomeCh, &execution.ForkChoiceReceipt{
LatestValidHash: gointerfaces.ConvertHashToH256(common.Hash{}),
Status: execution.ExecutionStatus_MissingSegment,
})
}, false)
return
}

currentParentHash := fcuHeader.ParentHash
currentParentNumber := fcuHeader.Number.Uint64() - 1
isCanonicalHash, err := rawdb.IsCanonicalHash(tx, currentParentHash, currentParentNumber)
if err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}
// Find such point, and collect all hashes
Expand All @@ -267,14 +275,14 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original
})
currentHeader, err := e.blockReader.Header(ctx, tx, currentParentHash, currentParentNumber)
if err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}
if currentHeader == nil {
sendForkchoiceReceiptWithoutWaiting(outcomeCh, &execution.ForkChoiceReceipt{
LatestValidHash: gointerfaces.ConvertHashToH256(common.Hash{}),
Status: execution.ExecutionStatus_MissingSegment,
})
}, false)
return
}
currentParentHash = currentHeader.ParentHash
Expand All @@ -284,30 +292,30 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original
currentParentNumber = currentHeader.Number.Uint64() - 1
isCanonicalHash, err = rawdb.IsCanonicalHash(tx, currentParentHash, currentParentNumber)
if err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}
}

if err := e.executionPipeline.UnwindTo(currentParentNumber, stagedsync.ForkChoice, tx); err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}
if e.hook != nil {
if err = e.hook.BeforeRun(tx, isSynced); err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}
}
// Run the unwind
if err := e.executionPipeline.RunUnwind(e.db, wrap.TxContainer{Tx: tx}); err != nil {
err = fmt.Errorf("updateForkChoice: %w", err)
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}

if err := rawdbv3.TxNums.Truncate(tx, currentParentNumber+1); err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}
// Mark all new canonicals as canonicals
Expand All @@ -318,72 +326,78 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original
h := rawdb.ReadHeader(tx, canonicalSegment.hash, canonicalSegment.number)

if b == nil || h == nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, fmt.Errorf("unexpected chain cap: %d", canonicalSegment.number))
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, fmt.Errorf("unexpected chain cap: %d", canonicalSegment.number), false)
return
}

if err := e.engine.VerifyHeader(chainReader, h, true); err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}

if err := e.engine.VerifyUncles(chainReader, h, b.Uncles); err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}

if err := rawdb.WriteCanonicalHash(tx, canonicalSegment.hash, canonicalSegment.number); err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}
}
if len(newCanonicals) > 0 {
if err := rawdbv3.TxNums.Truncate(tx, newCanonicals[0].number); err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}
if err := rawdb.AppendCanonicalTxNums(tx, newCanonicals[len(newCanonicals)-1].number); err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}
}
}
if isDomainAheadOfBlocks(tx) {
if err := tx.Commit(); err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}
sendForkchoiceReceiptWithoutWaiting(outcomeCh, &execution.ForkChoiceReceipt{
LatestValidHash: gointerfaces.ConvertHashToH256(common.Hash{}),
Status: execution.ExecutionStatus_TooFarAway,
ValidationError: "domain ahead of blocks",
})
}, false)
return
}

// Set Progress for headers and bodies accordingly.
if err := stages.SaveStageProgress(tx, stages.Headers, fcuHeader.Number.Uint64()); err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}
if err := stages.SaveStageProgress(tx, stages.BlockHashes, fcuHeader.Number.Uint64()); err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}
if err := stages.SaveStageProgress(tx, stages.Bodies, fcuHeader.Number.Uint64()); err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}
if err = rawdb.WriteHeadHeaderHash(tx, blockHash); err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false)
return
}

flushExtendingFork := blockHash == e.forkValidator.ExtendingForkHeadHash()
if flushExtendingFork {
e.logger.Debug("[updateForkchoice] Fork choice update: flushing in-memory state (built by previous newPayload)")
// Send forkchoice early (We already know the fork is valid)
sendForkchoiceReceiptWithoutWaiting(outcomeCh, &execution.ForkChoiceReceipt{
LatestValidHash: gointerfaces.ConvertHashToH256(blockHash),
Status: execution.ExecutionStatus_Success,
ValidationError: validationError,
}, false)
if err := e.forkValidator.FlushExtendingFork(tx, e.accumulator); err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, flushExtendingFork)
return
}
}
Expand All @@ -393,7 +407,7 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original
if _, err := e.executionPipeline.Run(e.db, wrap.TxContainer{Tx: tx}, initialCycle, firstCycle); err != nil {
err = fmt.Errorf("updateForkChoice: %w", err)
e.logger.Warn("Cannot update chain head", "hash", blockHash, "err", err)
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, flushExtendingFork)
return
}

Expand All @@ -414,28 +428,28 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original
} else {
valid, err := e.verifyForkchoiceHashes(ctx, tx, blockHash, finalizedHash, safeHash)
if err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, flushExtendingFork)
return
}
if !valid {
sendForkchoiceReceiptWithoutWaiting(outcomeCh, &execution.ForkChoiceReceipt{
Status: execution.ExecutionStatus_InvalidForkchoice,
LatestValidHash: gointerfaces.ConvertHashToH256(common.Hash{}),
})
}, flushExtendingFork)
return
}
if err := rawdb.TruncateCanonicalChain(ctx, tx, *headNumber+1); err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, flushExtendingFork)
return
}

if err := rawdbv3.TxNums.Truncate(tx, *headNumber+1); err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, flushExtendingFork)
return
}
commitStart := time.Now()
if err := tx.Commit(); err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, flushExtendingFork)
return
}
commitTime := time.Since(commitStart)
Expand All @@ -444,7 +458,7 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original
if err := e.db.View(ctx, func(tx kv.Tx) error {
return e.hook.AfterRun(tx, finishProgressBefore)
}); err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, flushExtendingFork)
return
}
}
Expand All @@ -453,7 +467,7 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original
if err := e.db.Update(ctx, func(tx kv.RwTx) error {
return kv.IncrementKey(tx, kv.DatabaseInfo, []byte("alex"))
}); err != nil {
sendForkchoiceErrorWithoutWaiting(outcomeCh, err)
sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, flushExtendingFork)
return
}

Expand All @@ -462,10 +476,10 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original
blockTimings := e.forkValidator.GetTimings(blockHash)
logArgs := []interface{}{"head", headHash, "hash", blockHash}
if flushExtendingFork {
totalTime := blockTimings[engine_helpers.BlockTimingsValidationIndex] + blockTimings[engine_helpers.BlockTimingsFlushExtendingFork]
totalTime := blockTimings[engine_helpers.BlockTimingsValidationIndex]
gasUsedMgas := float64(fcuHeader.GasUsed) / 1e6
mgasPerSec := gasUsedMgas / totalTime.Seconds()
logArgs = append(logArgs, "number", fcuHeader.Number.Uint64(), "execution", blockTimings[engine_helpers.BlockTimingsValidationIndex], "flushing", blockTimings[engine_helpers.BlockTimingsFlushExtendingFork], "mgas/s", fmt.Sprintf("%.2f", mgasPerSec))
logArgs = append(logArgs, "number", fcuHeader.Number.Uint64(), "execution", blockTimings[engine_helpers.BlockTimingsValidationIndex], "mgas/s", fmt.Sprintf("%.2f", mgasPerSec))
}
logArgs = append(logArgs, "commit", commitTime, "alloc", common.ByteCount(m.Alloc), "sys", common.ByteCount(m.Sys))
if log {
Expand All @@ -475,11 +489,12 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original
if *headNumber >= startPruneFrom {
e.runPostForkchoiceInBackground(initialCycle)
}

sendForkchoiceReceiptWithoutWaiting(outcomeCh, &execution.ForkChoiceReceipt{
LatestValidHash: gointerfaces.ConvertHashToH256(headHash),
Status: status,
ValidationError: validationError,
})
}, flushExtendingFork)
}

func (e *EthereumExecutionModule) runPostForkchoiceInBackground(initialCycle bool) {
Expand Down