From e9024f4b88f472699a7379b69dc81285316ef2f4 Mon Sep 17 00:00:00 2001 From: Giulio Date: Wed, 24 Jul 2024 01:07:20 +0200 Subject: [PATCH 1/3] parrallel flushing --- turbo/execution/eth1/forkchoice.go | 31 ++++++++++++++++++++---------- 1 file changed, 21 insertions(+), 10 deletions(-) diff --git a/turbo/execution/eth1/forkchoice.go b/turbo/execution/eth1/forkchoice.go index fbb0d720807..eb9eb774369 100644 --- a/turbo/execution/eth1/forkchoice.go +++ b/turbo/execution/eth1/forkchoice.go @@ -46,7 +46,10 @@ 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: @@ -142,7 +145,7 @@ 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) @@ -228,13 +231,13 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original 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 } @@ -243,7 +246,7 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original sendForkchoiceReceiptWithoutWaiting(outcomeCh, &execution.ForkChoiceReceipt{ LatestValidHash: gointerfaces.ConvertHashToH256(common.Hash{}), Status: execution.ExecutionStatus_MissingSegment, - }) + }, false) return } @@ -274,7 +277,7 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original sendForkchoiceReceiptWithoutWaiting(outcomeCh, &execution.ForkChoiceReceipt{ LatestValidHash: gointerfaces.ConvertHashToH256(common.Hash{}), Status: execution.ExecutionStatus_MissingSegment, - }) + }, false) return } currentParentHash = currentHeader.ParentHash @@ -357,7 +360,7 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original LatestValidHash: gointerfaces.ConvertHashToH256(common.Hash{}), Status: execution.ExecutionStatus_TooFarAway, ValidationError: "domain ahead of blocks", - }) + }, false) return } @@ -378,12 +381,19 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original sendForkchoiceErrorWithoutWaiting(outcomeCh, err) return } - if blockHash == e.forkValidator.ExtendingForkHeadHash() { + + flushExtendingFork := blockHash == e.forkValidator.ExtendingForkHeadHash() + if flushExtendingFork { e.logger.Info("[updateForkchoice] Fork choice update: flushing in-memory state (built by previous newPayload)") if err := e.forkValidator.FlushExtendingFork(tx, e.accumulator); err != nil { sendForkchoiceErrorWithoutWaiting(outcomeCh, err) return } + sendForkchoiceReceiptWithoutWaiting(outcomeCh, &execution.ForkChoiceReceipt{ + LatestValidHash: gointerfaces.ConvertHashToH256(blockHash), + Status: execution.ExecutionStatus_Success, + ValidationError: validationError, + }, false) } // Run the forkchoice initialCycle := limitedBigJump @@ -421,7 +431,7 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original 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 { @@ -470,11 +480,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) { From 212424709e356302584f0c2b374818cd6f0a2332 Mon Sep 17 00:00:00 2001 From: Giulio Date: Wed, 24 Jul 2024 01:11:18 +0200 Subject: [PATCH 2/3] parrallel flushing --- turbo/execution/eth1/forkchoice.go | 77 ++++++++++++++++-------------- 1 file changed, 41 insertions(+), 36 deletions(-) diff --git a/turbo/execution/eth1/forkchoice.go b/turbo/execution/eth1/forkchoice.go index eb9eb774369..13f3b89a44f 100644 --- a/turbo/execution/eth1/forkchoice.go +++ b/turbo/execution/eth1/forkchoice.go @@ -56,7 +56,12 @@ func sendForkchoiceReceiptWithoutWaiting(ch chan forkchoiceOutcome, receipt *exe } } -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: @@ -151,7 +156,7 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original 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 //} @@ -166,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 } @@ -177,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() @@ -186,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 } @@ -215,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 { @@ -224,7 +229,7 @@ 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 { @@ -254,7 +259,7 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original 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 @@ -270,7 +275,7 @@ 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 { @@ -287,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 @@ -321,39 +326,39 @@ 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{ @@ -366,19 +371,19 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original // 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 } @@ -386,7 +391,7 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original if flushExtendingFork { e.logger.Info("[updateForkchoice] Fork choice update: flushing in-memory state (built by previous newPayload)") if err := e.forkValidator.FlushExtendingFork(tx, e.accumulator); err != nil { - sendForkchoiceErrorWithoutWaiting(outcomeCh, err) + sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false) return } sendForkchoiceReceiptWithoutWaiting(outcomeCh, &execution.ForkChoiceReceipt{ @@ -401,7 +406,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, false) return } @@ -424,7 +429,7 @@ 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 { @@ -435,17 +440,17 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original 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) @@ -454,7 +459,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 } } @@ -463,7 +468,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 } From 3a951a0bd4e8bfda9aeb604aff9a6e6799d1569e Mon Sep 17 00:00:00 2001 From: Giulio Date: Wed, 24 Jul 2024 13:46:13 +0200 Subject: [PATCH 3/3] save --- turbo/execution/eth1/forkchoice.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/turbo/execution/eth1/forkchoice.go b/turbo/execution/eth1/forkchoice.go index ae3b90d3b9b..08627ef9613 100644 --- a/turbo/execution/eth1/forkchoice.go +++ b/turbo/execution/eth1/forkchoice.go @@ -397,7 +397,7 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original ValidationError: validationError, }, false) if err := e.forkValidator.FlushExtendingFork(tx, e.accumulator); err != nil { - sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, false) + sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, flushExtendingFork) return } } @@ -407,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(e.logger, outcomeCh, err, false) + sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, flushExtendingFork) return }