diff --git a/.github/matic-cli-config.yml b/.github/matic-cli-config.yml index c7dab39b0e..2b83b684c6 100644 --- a/.github/matic-cli-config.yml +++ b/.github/matic-cli-config.yml @@ -8,4 +8,4 @@ numOfNonValidators: 0 ethURL: http://ganache:9545 devnetType: docker borDockerBuildContext: "../../bor" -heimdallDockerBuildContext: "https://github.com/maticnetwork/heimdall.git#develop" \ No newline at end of file +heimdallDockerBuildContext: "https://github.com/maticnetwork/heimdall.git#develop" diff --git a/cmd/evm/internal/t8ntool/block.go b/cmd/evm/internal/t8ntool/block.go index d4edd33bde..4d75a30d4b 100644 --- a/cmd/evm/internal/t8ntool/block.go +++ b/cmd/evm/internal/t8ntool/block.go @@ -17,6 +17,7 @@ package t8ntool import ( + "context" "crypto/ecdsa" "encoding/json" "errors" @@ -188,7 +189,7 @@ func (i *bbInput) sealEthash(block *types.Block) (*types.Block, error) { // If the testmode is used, the sealer will return quickly, and complain // "Sealing result is not read by miner" if it cannot write the result. results := make(chan *types.Block, 1) - if err := engine.Seal(nil, block, results, nil); err != nil { + if err := engine.Seal(context.Background(), nil, block, results, nil); err != nil { panic(fmt.Sprintf("failed to seal block: %v", err)) } found := <-results diff --git a/cmd/evm/internal/t8ntool/execution.go b/cmd/evm/internal/t8ntool/execution.go index 874685f15e..c848b953f8 100644 --- a/cmd/evm/internal/t8ntool/execution.go +++ b/cmd/evm/internal/t8ntool/execution.go @@ -223,6 +223,7 @@ func (pre *Prestate) Apply(vmConfig vm.Config, chainConfig *params.ChainConfig, txIndex++ } + statedb.IntermediateRoot(chainConfig.IsEIP158(vmContext.BlockNumber)) // Add mining reward? if miningReward > 0 { diff --git a/common/tracing/context.go b/common/tracing/context.go new file mode 100644 index 0000000000..510e45d775 --- /dev/null +++ b/common/tracing/context.go @@ -0,0 +1,96 @@ +package tracing + +import ( + "context" + "time" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" +) + +type tracerKey struct{} + +type Option func(context.Context, trace.Span) + +func WithTracer(ctx context.Context, tr trace.Tracer) context.Context { + return context.WithValue(ctx, tracerKey{}, tr) +} + +func FromContext(ctx context.Context) trace.Tracer { + tr, _ := ctx.Value(tracerKey{}).(trace.Tracer) + + return tr +} + +func StartSpan(ctx context.Context, snapName string) (context.Context, trace.Span) { + tr := FromContext(ctx) + + if tr == nil { + return ctx, nil + } + + ctx, span := tr.Start(ctx, snapName) + ctx = WithTracer(ctx, tr) + + return ctx, span +} + +func EndSpan(span trace.Span) { + if span != nil { + span.End() + } +} + +func Trace(ctx context.Context, spanName string) (context.Context, trace.Span) { + tr := FromContext(ctx) + + if tr == nil { + return ctx, nil + } + + return tr.Start(ctx, spanName) +} + +func Exec(ctx context.Context, spanName string, opts ...Option) { + var span trace.Span + + tr := FromContext(ctx) + + if tr != nil { + ctx, span = tr.Start(ctx, spanName) + } + + for _, optFn := range opts { + optFn(ctx, span) + } + + if tr != nil { + span.End() + } +} + +func WithTime(fn func(context.Context, trace.Span)) Option { + return func(ctx context.Context, span trace.Span) { + ElapsedTime(ctx, span, "elapsed", fn) + } +} + +func ElapsedTime(ctx context.Context, span trace.Span, msg string, fn func(context.Context, trace.Span)) { + var now time.Time + + if span != nil { + now = time.Now() + } + + fn(ctx, span) + + if span != nil { + span.SetAttributes(attribute.Int(msg, int(time.Since(now).Milliseconds()))) + } +} + +func SetAttributes(span trace.Span, kvs ...attribute.KeyValue) { + if span != nil { + span.SetAttributes(kvs...) + } +} diff --git a/consensus/beacon/consensus.go b/consensus/beacon/consensus.go index 1fd7deb872..b7102c1e67 100644 --- a/consensus/beacon/consensus.go +++ b/consensus/beacon/consensus.go @@ -17,6 +17,7 @@ package beacon import ( + "context" "errors" "fmt" "math/big" @@ -170,10 +171,11 @@ func (beacon *Beacon) VerifyUncles(chain consensus.ChainReader, block *types.Blo // verifyHeader checks whether a header conforms to the consensus rules of the // stock Ethereum consensus engine. The difference between the beacon and classic is // (a) The following fields are expected to be constants: -// - difficulty is expected to be 0 -// - nonce is expected to be 0 -// - unclehash is expected to be Hash(emptyHeader) +// - difficulty is expected to be 0 +// - nonce is expected to be 0 +// - unclehash is expected to be Hash(emptyHeader) // to be the desired constants +// // (b) the timestamp is not verified anymore // (c) the extradata is limited to 32 bytes func (beacon *Beacon) verifyHeader(chain consensus.ChainHeaderReader, header, parent *types.Header) error { @@ -278,11 +280,11 @@ func (beacon *Beacon) Finalize(chain consensus.ChainHeaderReader, header *types. // FinalizeAndAssemble implements consensus.Engine, setting the final state and // assembling the block. -func (beacon *Beacon) FinalizeAndAssemble(chain consensus.ChainHeaderReader, header *types.Header, state *state.StateDB, txs []*types.Transaction, uncles []*types.Header, receipts []*types.Receipt) (*types.Block, error) { +func (beacon *Beacon) FinalizeAndAssemble(ctx context.Context, chain consensus.ChainHeaderReader, header *types.Header, state *state.StateDB, txs []*types.Transaction, uncles []*types.Header, receipts []*types.Receipt) (*types.Block, error) { // FinalizeAndAssemble is different with Prepare, it can be used in both block // generation and verification. So determine the consensus rules by header type. if !beacon.IsPoSHeader(header) { - return beacon.ethone.FinalizeAndAssemble(chain, header, state, txs, uncles, receipts) + return beacon.ethone.FinalizeAndAssemble(ctx, chain, header, state, txs, uncles, receipts) } // Finalize and assemble the block beacon.Finalize(chain, header, state, txs, uncles) @@ -294,9 +296,9 @@ func (beacon *Beacon) FinalizeAndAssemble(chain consensus.ChainHeaderReader, hea // // Note, the method returns immediately and will send the result async. More // than one result may also be returned depending on the consensus algorithm. -func (beacon *Beacon) Seal(chain consensus.ChainHeaderReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error { +func (beacon *Beacon) Seal(ctx context.Context, chain consensus.ChainHeaderReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error { if !beacon.IsPoSHeader(block.Header()) { - return beacon.ethone.Seal(chain, block, results, stop) + return beacon.ethone.Seal(ctx, chain, block, results, stop) } // The seal verification is done by the external consensus engine, // return directly without pushing any block back. In another word diff --git a/consensus/bor/bor.go b/consensus/bor/bor.go index cc9d750258..dee3998703 100644 --- a/consensus/bor/bor.go +++ b/consensus/bor/bor.go @@ -16,10 +16,13 @@ import ( "time" lru "github.com/hashicorp/golang-lru" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" "golang.org/x/crypto/sha3" "github.com/ethereum/go-ethereum/accounts" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/tracing" "github.com/ethereum/go-ethereum/consensus" "github.com/ethereum/go-ethereum/consensus/bor/api" "github.com/ethereum/go-ethereum/consensus/bor/clerk" @@ -735,9 +738,9 @@ func (c *Bor) Finalize(chain consensus.ChainHeaderReader, header *types.Header, headerNumber := header.Number.Uint64() - if IsSprintStart(headerNumber, c.config.Sprint) { - ctx := context.Background() + ctx := context.Background() + if IsSprintStart(headerNumber, c.config.Sprint) { cx := statefull.ChainContext{Chain: chain, Bor: c} // check and commit span if err := c.checkAndCommitSpan(ctx, state, header, cx); err != nil { @@ -804,26 +807,35 @@ func (c *Bor) changeContractCodeIfNeeded(headerNumber uint64, state *state.State // FinalizeAndAssemble implements consensus.Engine, ensuring no uncles are set, // nor block rewards given, and returns the final block. -func (c *Bor) FinalizeAndAssemble(chain consensus.ChainHeaderReader, header *types.Header, state *state.StateDB, txs []*types.Transaction, _ []*types.Header, receipts []*types.Receipt) (*types.Block, error) { - var stateSyncData []*types.StateSyncData +func (c *Bor) FinalizeAndAssemble(ctx context.Context, chain consensus.ChainHeaderReader, header *types.Header, state *state.StateDB, txs []*types.Transaction, uncles []*types.Header, receipts []*types.Receipt) (*types.Block, error) { + finalizeCtx, finalizeSpan := tracing.StartSpan(ctx, "bor.FinalizeAndAssemble") + defer tracing.EndSpan(finalizeSpan) + + stateSyncData := []*types.StateSyncData{} headerNumber := header.Number.Uint64() - if IsSprintStart(headerNumber, c.config.Sprint) { - ctx := context.Background() + var err error + if IsSprintStart(headerNumber, c.config.Sprint) { cx := statefull.ChainContext{Chain: chain, Bor: c} - // check and commit span - err := c.checkAndCommitSpan(ctx, state, header, cx) + tracing.Exec(finalizeCtx, "bor.checkAndCommitSpan", func(ctx context.Context, span trace.Span) { + // check and commit span + err = c.checkAndCommitSpan(finalizeCtx, state, header, cx) + }) + if err != nil { log.Error("Error while committing span", "error", err) return nil, err } if c.HeimdallClient != nil { - // commit states - stateSyncData, err = c.CommitStates(ctx, state, header, cx) + tracing.Exec(finalizeCtx, "bor.checkAndCommitSpan", func(ctx context.Context, span trace.Span) { + // commit states + stateSyncData, err = c.CommitStates(finalizeCtx, state, header, cx) + }) + if err != nil { log.Error("Error while committing states", "error", err) return nil, err @@ -831,13 +843,21 @@ func (c *Bor) FinalizeAndAssemble(chain consensus.ChainHeaderReader, header *typ } } - if err := c.changeContractCodeIfNeeded(headerNumber, state); err != nil { + tracing.Exec(finalizeCtx, "bor.changeContractCodeIfNeeded", func(ctx context.Context, span trace.Span) { + err = c.changeContractCodeIfNeeded(headerNumber, state) + }) + + if err != nil { log.Error("Error changing contract code", "error", err) return nil, err } - // No block rewards in PoA, so the state remains as is and uncles are dropped - header.Root = state.IntermediateRoot(chain.Config().IsEIP158(header.Number)) + // No block rewards in PoA, so the state remains as it is + tracing.Exec(finalizeCtx, "bor.IntermediateRoot", func(ctx context.Context, span trace.Span) { + header.Root = state.IntermediateRoot(chain.Config().IsEIP158(header.Number)) + }) + + // Uncles are dropped header.UncleHash = types.CalcUncleHash(nil) // Assemble block @@ -847,6 +867,14 @@ func (c *Bor) FinalizeAndAssemble(chain consensus.ChainHeaderReader, header *typ bc := chain.(core.BorStateSyncer) bc.SetStateSync(stateSyncData) + tracing.SetAttributes( + finalizeSpan, + attribute.Int("number", int(header.Number.Int64())), + attribute.String("hash", header.Hash().String()), + attribute.Int("number of txs", len(txs)), + attribute.Int("gas used", int(block.GasUsed())), + ) + // return the final block for sealing return block, nil } @@ -862,7 +890,18 @@ func (c *Bor) Authorize(currentSigner common.Address, signFn SignerFn) { // Seal implements consensus.Engine, attempting to create a sealed block using // the local signing credentials. -func (c *Bor) Seal(chain consensus.ChainHeaderReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error { +func (c *Bor) Seal(ctx context.Context, chain consensus.ChainHeaderReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error { + _, sealSpan := tracing.StartSpan(ctx, "bor.Seal") + + var endSpan bool = true + + defer func() { + // Only end span in case of early-returns/errors + if endSpan { + tracing.EndSpan(sealSpan) + } + }() + header := block.Header() // Sealing the genesis block is not supported number := header.Number.Uint64() @@ -908,7 +947,7 @@ func (c *Bor) Seal(chain consensus.ChainHeaderReader, block *types.Block, result // Wait until sealing is terminated or delay timeout. log.Info("Waiting for slot to sign and propagate", "number", number, "hash", header.Hash, "delay-in-sec", uint(delay), "delay", common.PrettyDuration(delay)) - go func() { + go func(sealSpan trace.Span) { select { case <-stop: log.Debug("Discarding sealing operation for block", "number", number) @@ -931,13 +970,27 @@ func (c *Bor) Seal(chain consensus.ChainHeaderReader, block *types.Block, result "delay", delay, "headerDifficulty", header.Difficulty, ) + + tracing.SetAttributes( + sealSpan, + attribute.Int("number", int(number)), + attribute.String("hash", header.Hash().String()), + attribute.Int("delay", int(delay.Milliseconds())), + attribute.Int("wiggle", int(wiggle.Milliseconds())), + attribute.Bool("out-of-turn", wiggle > 0), + ) + + tracing.EndSpan(sealSpan) } select { case results <- block.WithSeal(header): default: log.Warn("Sealing result was not read by miner", "number", number, "sealhash", SealHash(header, c.config)) } - }() + }(sealSpan) + + // Set the endSpan flag to false, as the go routine will handle it + endSpan = false return nil } @@ -1000,13 +1053,13 @@ func (c *Bor) checkAndCommitSpan( ) error { headerNumber := header.Number.Uint64() - currentSpan, err := c.spanner.GetCurrentSpan(ctx, header.ParentHash) + span, err := c.spanner.GetCurrentSpan(ctx, header.ParentHash) if err != nil { return err } - if c.needToCommitSpan(currentSpan, headerNumber) { - return c.FetchAndCommitSpan(ctx, currentSpan.ID+1, state, header, chain) + if c.needToCommitSpan(span, headerNumber) { + return c.FetchAndCommitSpan(ctx, span.ID+1, state, header, chain) } return nil @@ -1076,6 +1129,7 @@ func (c *Bor) CommitStates( header *types.Header, chain statefull.ChainContext, ) ([]*types.StateSyncData, error) { + fetchStart := time.Now() number := header.Number.Uint64() _lastStateID, err := c.GenesisContractsClient.LastStateId(number - 1) @@ -1102,6 +1156,8 @@ func (c *Bor) CommitStates( } } + fetchTime := time.Since(fetchStart) + processStart := time.Now() totalGas := 0 /// limit on gas for state sync per block chainID := c.chainConfig.ChainID.String() stateSyncs := make([]*types.StateSyncData, len(eventRecords)) @@ -1140,7 +1196,9 @@ func (c *Bor) CommitStates( lastStateID++ } - log.Info("StateSyncData", "Gas", totalGas, "Block-number", number, "LastStateID", lastStateID, "TotalRecords", len(eventRecords)) + processTime := time.Since(processStart) + + log.Info("StateSyncData", "gas", totalGas, "number", number, "lastStateID", lastStateID, "total records", len(eventRecords), "fetch time", int(fetchTime.Milliseconds()), "process time", int(processTime.Milliseconds())) return stateSyncs, nil } diff --git a/consensus/bor/validators_getter.go b/consensus/bor/validators_getter.go deleted file mode 100644 index 90d1fccf6e..0000000000 --- a/consensus/bor/validators_getter.go +++ /dev/null @@ -1,11 +0,0 @@ -package bor - -import ( - "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/consensus/bor/valset" -) - -//go:generate mockgen -destination=./validators_getter_mock.go -package=bor . ValidatorsGetter -type ValidatorsGetter interface { - GetCurrentValidators(headerHash common.Hash, blockNumber uint64) ([]*valset.Validator, error) -} diff --git a/consensus/bor/validators_getter_mock.go b/consensus/bor/validators_getter_mock.go deleted file mode 100644 index ad99489d8e..0000000000 --- a/consensus/bor/validators_getter_mock.go +++ /dev/null @@ -1,51 +0,0 @@ -// Code generated by MockGen. DO NOT EDIT. -// Source: github.com/ethereum/go-ethereum/consensus/bor (interfaces: ValidatorsGetter) - -// Package bor is a generated GoMock package. -package bor - -import ( - reflect "reflect" - - common "github.com/ethereum/go-ethereum/common" - valset "github.com/ethereum/go-ethereum/consensus/bor/valset" - gomock "github.com/golang/mock/gomock" -) - -// MockValidatorsGetter is a mock of ValidatorsGetter interface. -type MockValidatorsGetter struct { - ctrl *gomock.Controller - recorder *MockValidatorsGetterMockRecorder -} - -// MockValidatorsGetterMockRecorder is the mock recorder for MockValidatorsGetter. -type MockValidatorsGetterMockRecorder struct { - mock *MockValidatorsGetter -} - -// NewMockValidatorsGetter creates a new mock instance. -func NewMockValidatorsGetter(ctrl *gomock.Controller) *MockValidatorsGetter { - mock := &MockValidatorsGetter{ctrl: ctrl} - mock.recorder = &MockValidatorsGetterMockRecorder{mock} - return mock -} - -// EXPECT returns an object that allows the caller to indicate expected use. -func (m *MockValidatorsGetter) EXPECT() *MockValidatorsGetterMockRecorder { - return m.recorder -} - -// GetCurrentValidators mocks base method. -func (m *MockValidatorsGetter) GetCurrentValidators(arg0 common.Hash, arg1 uint64) ([]*valset.Validator, error) { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "GetCurrentValidators", arg0, arg1) - ret0, _ := ret[0].([]*valset.Validator) - ret1, _ := ret[1].(error) - return ret0, ret1 -} - -// GetCurrentValidators indicates an expected call of GetCurrentValidators. -func (mr *MockValidatorsGetterMockRecorder) GetCurrentValidators(arg0, arg1 interface{}) *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetCurrentValidators", reflect.TypeOf((*MockValidatorsGetter)(nil).GetCurrentValidators), arg0, arg1) -} diff --git a/consensus/clique/clique.go b/consensus/clique/clique.go index 685186817d..ad09552469 100644 --- a/consensus/clique/clique.go +++ b/consensus/clique/clique.go @@ -19,6 +19,7 @@ package clique import ( "bytes" + "context" "errors" "fmt" "io" @@ -569,7 +570,7 @@ func (c *Clique) Finalize(chain consensus.ChainHeaderReader, header *types.Heade // FinalizeAndAssemble implements consensus.Engine, ensuring no uncles are set, // nor block rewards given, and returns the final block. -func (c *Clique) FinalizeAndAssemble(chain consensus.ChainHeaderReader, header *types.Header, state *state.StateDB, txs []*types.Transaction, uncles []*types.Header, receipts []*types.Receipt) (*types.Block, error) { +func (c *Clique) FinalizeAndAssemble(ctx context.Context, chain consensus.ChainHeaderReader, header *types.Header, state *state.StateDB, txs []*types.Transaction, uncles []*types.Header, receipts []*types.Receipt) (*types.Block, error) { // Finalize block c.Finalize(chain, header, state, txs, uncles) @@ -589,7 +590,7 @@ func (c *Clique) Authorize(signer common.Address, signFn SignerFn) { // Seal implements consensus.Engine, attempting to create a sealed block using // the local signing credentials. -func (c *Clique) Seal(chain consensus.ChainHeaderReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error { +func (c *Clique) Seal(ctx context.Context, chain consensus.ChainHeaderReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error { header := block.Header() // Sealing the genesis block is not supported diff --git a/consensus/consensus.go b/consensus/consensus.go index af8ce98ff3..c287972beb 100644 --- a/consensus/consensus.go +++ b/consensus/consensus.go @@ -18,6 +18,7 @@ package consensus import ( + "context" "math/big" "github.com/ethereum/go-ethereum/common" @@ -97,7 +98,7 @@ type Engine interface { // // Note: The block header and state database might be updated to reflect any // consensus rules that happen at finalization (e.g. block rewards). - FinalizeAndAssemble(chain ChainHeaderReader, header *types.Header, state *state.StateDB, txs []*types.Transaction, + FinalizeAndAssemble(ctx context.Context, chain ChainHeaderReader, header *types.Header, state *state.StateDB, txs []*types.Transaction, uncles []*types.Header, receipts []*types.Receipt) (*types.Block, error) // Seal generates a new sealing request for the given input block and pushes @@ -105,7 +106,7 @@ type Engine interface { // // Note, the method returns immediately and will send the result async. More // than one result may also be returned depending on the consensus algorithm. - Seal(chain ChainHeaderReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error + Seal(ctx context.Context, chain ChainHeaderReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error // SealHash returns the hash of a block prior to it being sealed. SealHash(header *types.Header) common.Hash diff --git a/consensus/ethash/consensus.go b/consensus/ethash/consensus.go index 7dec436a26..761442f44e 100644 --- a/consensus/ethash/consensus.go +++ b/consensus/ethash/consensus.go @@ -18,6 +18,7 @@ package ethash import ( "bytes" + "context" "errors" "fmt" "math/big" @@ -598,7 +599,7 @@ func (ethash *Ethash) Finalize(chain consensus.ChainHeaderReader, header *types. // FinalizeAndAssemble implements consensus.Engine, accumulating the block and // uncle rewards, setting the final state and assembling the block. -func (ethash *Ethash) FinalizeAndAssemble(chain consensus.ChainHeaderReader, header *types.Header, state *state.StateDB, txs []*types.Transaction, uncles []*types.Header, receipts []*types.Receipt) (*types.Block, error) { +func (ethash *Ethash) FinalizeAndAssemble(ctx context.Context, chain consensus.ChainHeaderReader, header *types.Header, state *state.StateDB, txs []*types.Transaction, uncles []*types.Header, receipts []*types.Receipt) (*types.Block, error) { // Finalize block ethash.Finalize(chain, header, state, txs, uncles) diff --git a/consensus/ethash/ethash_test.go b/consensus/ethash/ethash_test.go index 382eefeecf..ef8cd3e8fa 100644 --- a/consensus/ethash/ethash_test.go +++ b/consensus/ethash/ethash_test.go @@ -17,6 +17,7 @@ package ethash import ( + "context" "io/ioutil" "math/big" "math/rand" @@ -38,7 +39,7 @@ func TestTestMode(t *testing.T) { defer ethash.Close() results := make(chan *types.Block) - err := ethash.Seal(nil, types.NewBlockWithHeader(header), results, nil) + err := ethash.Seal(context.Background(), nil, types.NewBlockWithHeader(header), results, nil) if err != nil { t.Fatalf("failed to seal block: %v", err) } @@ -111,12 +112,13 @@ func TestRemoteSealer(t *testing.T) { // Push new work. results := make(chan *types.Block) - ethash.Seal(nil, block, results, nil) + err := ethash.Seal(context.Background(), nil, block, results, nil) - var ( - work [4]string - err error - ) + if err != nil { + t.Error("error in sealing block") + } + + var work [4]string if work, err = api.GetWork(); err != nil || work[0] != sealhash.Hex() { t.Error("expect to return a mining work has same hash") } @@ -128,7 +130,11 @@ func TestRemoteSealer(t *testing.T) { header = &types.Header{Number: big.NewInt(1), Difficulty: big.NewInt(1000)} block = types.NewBlockWithHeader(header) sealhash = ethash.SealHash(header) - ethash.Seal(nil, block, results, nil) + err = ethash.Seal(context.Background(), nil, block, results, nil) + + if err != nil { + t.Error("error in sealing block") + } if work, err = api.GetWork(); err != nil || work[0] != sealhash.Hex() { t.Error("expect to return the latest pushed work") diff --git a/consensus/ethash/sealer.go b/consensus/ethash/sealer.go index 6fa60ef6a8..d851a065f2 100644 --- a/consensus/ethash/sealer.go +++ b/consensus/ethash/sealer.go @@ -48,7 +48,7 @@ var ( // Seal implements consensus.Engine, attempting to find a nonce that satisfies // the block's difficulty requirements. -func (ethash *Ethash) Seal(chain consensus.ChainHeaderReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error { +func (ethash *Ethash) Seal(ctx context.Context, chain consensus.ChainHeaderReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error { // If we're running a fake PoW, simply return a 0 nonce immediately if ethash.config.PowMode == ModeFake || ethash.config.PowMode == ModeFullFake { header := block.Header() @@ -62,7 +62,7 @@ func (ethash *Ethash) Seal(chain consensus.ChainHeaderReader, block *types.Block } // If we're running a shared PoW, delegate sealing to it if ethash.shared != nil { - return ethash.shared.Seal(chain, block, results, stop) + return ethash.shared.Seal(ctx, chain, block, results, stop) } // Create a runner and the multiple search threads it directs abort := make(chan struct{}) @@ -117,7 +117,8 @@ func (ethash *Ethash) Seal(chain consensus.ChainHeaderReader, block *types.Block case <-ethash.update: // Thread count was changed on user request, restart close(abort) - if err := ethash.Seal(chain, block, results, stop); err != nil { + + if err := ethash.Seal(ctx, chain, block, results, stop); err != nil { ethash.config.Log.Error("Failed to restart sealing after update", "err", err) } } diff --git a/consensus/ethash/sealer_test.go b/consensus/ethash/sealer_test.go index 9ddfcd840a..ad03fdff75 100644 --- a/consensus/ethash/sealer_test.go +++ b/consensus/ethash/sealer_test.go @@ -17,6 +17,7 @@ package ethash import ( + "context" "encoding/json" "io/ioutil" "math/big" @@ -57,7 +58,11 @@ func TestRemoteNotify(t *testing.T) { header := &types.Header{Number: big.NewInt(1), Difficulty: big.NewInt(100)} block := types.NewBlockWithHeader(header) - ethash.Seal(nil, block, nil, nil) + err := ethash.Seal(context.Background(), nil, block, nil, nil) + + if err != nil { + t.Error("error in sealing block") + } select { case work := <-sink: if want := ethash.SealHash(header).Hex(); work[0] != want { @@ -105,7 +110,11 @@ func TestRemoteNotifyFull(t *testing.T) { header := &types.Header{Number: big.NewInt(1), Difficulty: big.NewInt(100)} block := types.NewBlockWithHeader(header) - ethash.Seal(nil, block, nil, nil) + err := ethash.Seal(context.Background(), nil, block, nil, nil) + + if err != nil { + t.Error("error in sealing block") + } select { case work := <-sink: if want := "0x" + strconv.FormatUint(header.Number.Uint64(), 16); work["number"] != want { @@ -151,7 +160,11 @@ func TestRemoteMultiNotify(t *testing.T) { for i := 0; i < cap(sink); i++ { header := &types.Header{Number: big.NewInt(int64(i)), Difficulty: big.NewInt(100)} block := types.NewBlockWithHeader(header) - ethash.Seal(nil, block, results, nil) + err := ethash.Seal(context.Background(), nil, block, results, nil) + + if err != nil { + t.Error("error in sealing block") + } } for i := 0; i < cap(sink); i++ { @@ -204,7 +217,11 @@ func TestRemoteMultiNotifyFull(t *testing.T) { for i := 0; i < cap(sink); i++ { header := &types.Header{Number: big.NewInt(int64(i)), Difficulty: big.NewInt(100)} block := types.NewBlockWithHeader(header) - ethash.Seal(nil, block, results, nil) + err := ethash.Seal(context.Background(), nil, block, results, nil) + + if err != nil { + t.Error("error in sealing block") + } } for i := 0; i < cap(sink); i++ { @@ -270,7 +287,11 @@ func TestStaleSubmission(t *testing.T) { for id, c := range testcases { for _, h := range c.headers { - ethash.Seal(nil, types.NewBlockWithHeader(h), results, nil) + err := ethash.Seal(context.Background(), nil, types.NewBlockWithHeader(h), results, nil) + + if err != nil { + t.Error("error in sealing block") + } } if res := api.SubmitWork(fakeNonce, ethash.SealHash(c.headers[c.submitIndex]), fakeDigest); res != c.submitRes { t.Errorf("case %d submit result mismatch, want %t, get %t", id+1, c.submitRes, res) diff --git a/core/blockchain.go b/core/blockchain.go index fe8172e41e..48d5e966db 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -223,6 +223,7 @@ type BlockChain struct { // NewBlockChain returns a fully initialised block chain using information // available in the database. It initialises the default Ethereum Validator // and Processor. +// //nolint:gocognit func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *params.ChainConfig, engine consensus.Engine, vmConfig vm.Config, shouldPreserve func(header *types.Header) bool, txLookupLimit *uint64, checker ethereum.ChainValidator) (*BlockChain, error) { if cacheConfig == nil { diff --git a/core/chain_makers.go b/core/chain_makers.go index 4b2d2082df..e9944e4744 100644 --- a/core/chain_makers.go +++ b/core/chain_makers.go @@ -17,6 +17,7 @@ package core import ( + "context" "fmt" "math/big" @@ -258,7 +259,7 @@ func GenerateChain(config *params.ChainConfig, parent *types.Block, engine conse } if b.engine != nil { // Finalize and seal the block - block, _ := b.engine.FinalizeAndAssemble(chainreader, b.header, statedb, b.txs, b.uncles, b.receipts) + block, _ := b.engine.FinalizeAndAssemble(context.Background(), chainreader, b.header, statedb, b.txs, b.uncles, b.receipts) // Write state changes to db root, err := statedb.Commit(config.IsEIP158(b.header.Number)) diff --git a/core/types/transaction.go b/core/types/transaction.go index 29820a0d78..e0e52f25bc 100644 --- a/core/types/transaction.go +++ b/core/types/transaction.go @@ -508,6 +508,10 @@ func NewTransactionsByPriceAndNonce(signer Signer, txs map[common.Address]Transa // Initialize a price and received time based heap with the head transactions heads := make(TxByPriceAndTime, 0, len(txs)) for from, accTxs := range txs { + if len(accTxs) == 0 { + continue + } + acc, _ := Sender(signer, accTxs[0]) wrapped, err := NewTxWithMinerFee(accTxs[0], baseFee) // Remove transaction if sender doesn't match from, or if wrapping fails. @@ -550,6 +554,10 @@ func (t *TransactionsByPriceAndNonce) Shift() { heap.Pop(&t.heads) } +func (t *TransactionsByPriceAndNonce) GetTxs() int { + return len(t.txs) +} + // Pop removes the best transaction, *not* replacing it with the next one from // the same account. This should be used when a transaction cannot be executed // and hence all subsequent ones should be discarded from the same account. diff --git a/go.mod b/go.mod index 7a643a251c..18a2ca0ae7 100644 --- a/go.mod +++ b/go.mod @@ -122,7 +122,7 @@ require ( github.com/mitchellh/pointerstructure v1.2.0 // indirect github.com/mitchellh/reflectwalk v1.0.0 // indirect github.com/opentracing/opentracing-go v1.1.0 // indirect - github.com/pelletier/go-toml v1.9.5 // indirect + github.com/pelletier/go-toml v1.9.5 github.com/pkg/errors v0.9.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect github.com/posener/complete v1.1.1 // indirect @@ -130,7 +130,7 @@ require ( github.com/tklauser/numcpus v0.2.2 // indirect github.com/zclconf/go-cty v1.8.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.2.0 // indirect - go.opentelemetry.io/otel/trace v1.2.0 // indirect + go.opentelemetry.io/otel/trace v1.2.0 go.opentelemetry.io/proto/otlp v0.10.0 // indirect golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e // indirect golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 // indirect diff --git a/miner/worker.go b/miner/worker.go index 922eba7e9f..797e7ea980 100644 --- a/miner/worker.go +++ b/miner/worker.go @@ -17,6 +17,7 @@ package miner import ( + "context" "errors" "fmt" "math/big" @@ -25,8 +26,12 @@ import ( "time" mapset "github.com/deckarep/golang-set" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/tracing" "github.com/ethereum/go-ethereum/consensus" "github.com/ethereum/go-ethereum/consensus/misc" "github.com/ethereum/go-ethereum/core" @@ -144,6 +149,8 @@ func (env *environment) discard() { // task contains all information for consensus engine sealing and result submitting. type task struct { + //nolint:containedctx + ctx context.Context receipts []*types.Receipt state *state.StateDB block *types.Block @@ -158,6 +165,8 @@ const ( // newWorkReq represents a request for new sealing work submitting with relative interrupt notifier. type newWorkReq struct { + //nolint:containedctx + ctx context.Context interrupt *int32 noempty bool timestamp int64 @@ -165,6 +174,8 @@ type newWorkReq struct { // getWorkReq represents a request for getting a new sealing work with provided parameters. type getWorkReq struct { + //nolint:containedctx + ctx context.Context params *generateParams err error result chan *types.Block @@ -287,9 +298,12 @@ func newWorker(config *Config, chainConfig *params.ChainConfig, engine consensus recommit = minRecommitInterval } + ctx := tracing.WithTracer(context.Background(), otel.GetTracerProvider().Tracer("MinerWorker")) + worker.wg.Add(4) - go worker.mainLoop() - go worker.newWorkLoop(recommit) + + go worker.mainLoop(ctx) + go worker.newWorkLoop(ctx, recommit) go worker.resultLoop() go worker.taskLoop() @@ -419,7 +433,9 @@ func recalcRecommit(minRecommit, prev time.Duration, target float64, inc bool) t } // newWorkLoop is a standalone goroutine to submit new sealing work upon received events. -func (w *worker) newWorkLoop(recommit time.Duration) { +// +//nolint:gocognit +func (w *worker) newWorkLoop(ctx context.Context, recommit time.Duration) { defer w.wg.Done() var ( interrupt *int32 @@ -433,12 +449,16 @@ func (w *worker) newWorkLoop(recommit time.Duration) { // commit aborts in-flight transaction execution with given signal and resubmits a new one. commit := func(noempty bool, s int32) { + // we close spans only by the place we created them + ctx, span := tracing.Trace(ctx, "worker.newWorkLoop.commit") + tracing.EndSpan(span) + if interrupt != nil { atomic.StoreInt32(interrupt, s) } interrupt = new(int32) select { - case w.newWorkCh <- &newWorkReq{interrupt: interrupt, noempty: noempty, timestamp: timestamp}: + case w.newWorkCh <- &newWorkReq{interrupt: interrupt, noempty: noempty, timestamp: timestamp, ctx: ctx}: case <-w.exitCh: return } @@ -447,6 +467,9 @@ func (w *worker) newWorkLoop(recommit time.Duration) { } // clearPending cleans the stale pending tasks. clearPending := func(number uint64) { + _, span := tracing.Trace(ctx, "worker.newWorkLoop.clearPending") + tracing.EndSpan(span) + w.pendingMu.Lock() for h, t := range w.pendingTasks { if t.block.NumberU64()+staleThreshold <= number { @@ -519,7 +542,8 @@ func (w *worker) newWorkLoop(recommit time.Duration) { // mainLoop is responsible for generating and submitting sealing work based on // the received event. It can support two modes: automatically generate task and // submit it or return task according to given parameters for various proposes. -func (w *worker) mainLoop() { +// nolint: gocognit +func (w *worker) mainLoop(ctx context.Context) { defer w.wg.Done() defer w.txsSub.Unsubscribe() defer w.chainHeadSub.Unsubscribe() @@ -536,10 +560,10 @@ func (w *worker) mainLoop() { for { select { case req := <-w.newWorkCh: - w.commitWork(req.interrupt, req.noempty, req.timestamp) + w.commitWork(req.ctx, req.interrupt, req.noempty, req.timestamp) case req := <-w.getWorkCh: - block, err := w.generateWork(req.params) + block, err := w.generateWork(req.ctx, req.params) if err != nil { req.err = err req.result <- nil @@ -567,7 +591,10 @@ func (w *worker) mainLoop() { if w.isRunning() && w.current != nil && len(w.current.uncles) < 2 { start := time.Now() if err := w.commitUncle(w.current, ev.Block.Header()); err == nil { - w.commit(w.current.copy(), nil, true, start) + commitErr := w.commit(ctx, w.current.copy(), nil, true, start) + if commitErr != nil { + log.Error("error while committing work for mining", "err", commitErr) + } } } @@ -614,7 +641,7 @@ func (w *worker) mainLoop() { // submit sealing work here since all empty submission will be rejected // by clique. Of course the advance sealing(empty submission) is disabled. if w.chainConfig.Clique != nil && w.chainConfig.Clique.Period == 0 { - w.commitWork(nil, true, time.Now().Unix()) + w.commitWork(ctx, nil, true, time.Now().Unix()) } } atomic.AddInt32(&w.newTxs, int32(len(ev.Txs))) @@ -670,7 +697,7 @@ func (w *worker) taskLoop() { w.pendingTasks[sealHash] = task w.pendingMu.Unlock() - if err := w.engine.Seal(w.chain, task.block, w.resultCh, stopCh); err != nil { + if err := w.engine.Seal(task.ctx, w.chain, task.block, w.resultCh, stopCh); err != nil { log.Warn("Block sealing failed", "err", err) w.pendingMu.Lock() delete(w.pendingTasks, sealHash) @@ -694,10 +721,12 @@ func (w *worker) resultLoop() { if block == nil { continue } + // Short circuit when receiving duplicate result caused by resubmitting. if w.chain.HasBlock(block.Hash(), block.NumberU64()) { continue } + oldBlock := w.chain.GetBlockByNumber(block.NumberU64()) if oldBlock != nil { oldBlockAuthor, _ := w.chain.Engine().Author(oldBlock.Header()) @@ -707,49 +736,72 @@ func (w *worker) resultLoop() { continue } } + var ( sealhash = w.engine.SealHash(block.Header()) hash = block.Hash() ) + w.pendingMu.RLock() task, exist := w.pendingTasks[sealhash] w.pendingMu.RUnlock() + if !exist { log.Error("Block found but no relative pending task", "number", block.Number(), "sealhash", sealhash, "hash", hash) continue } + // Different block could share same sealhash, deep copy here to prevent write-write conflict. var ( receipts = make([]*types.Receipt, len(task.receipts)) logs []*types.Log + err error ) - for i, taskReceipt := range task.receipts { - receipt := new(types.Receipt) - receipts[i] = receipt - *receipt = *taskReceipt - - // add block location fields - receipt.BlockHash = hash - receipt.BlockNumber = block.Number() - receipt.TransactionIndex = uint(i) - - // Update the block hash in all logs since it is now available and not when the - // receipt/log of individual transactions were created. - receipt.Logs = make([]*types.Log, len(taskReceipt.Logs)) - for i, taskLog := range taskReceipt.Logs { - log := new(types.Log) - receipt.Logs[i] = log - *log = *taskLog - log.BlockHash = hash + + tracing.Exec(task.ctx, "resultLoop", func(ctx context.Context, span trace.Span) { + for i, taskReceipt := range task.receipts { + receipt := new(types.Receipt) + receipts[i] = receipt + *receipt = *taskReceipt + + // add block location fields + receipt.BlockHash = hash + receipt.BlockNumber = block.Number() + receipt.TransactionIndex = uint(i) + + // Update the block hash in all logs since it is now available and not when the + // receipt/log of individual transactions were created. + receipt.Logs = make([]*types.Log, len(taskReceipt.Logs)) + for i, taskLog := range taskReceipt.Logs { + log := new(types.Log) + receipt.Logs[i] = log + *log = *taskLog + log.BlockHash = hash + } + logs = append(logs, receipt.Logs...) } - logs = append(logs, receipt.Logs...) - } - // Commit block and state to database. - _, err := w.chain.WriteBlockAndSetHead(block, receipts, logs, task.state, true) + + // Commit block and state to database. + tracing.ElapsedTime(ctx, span, "WriteBlockAndSetHead time taken", func(_ context.Context, _ trace.Span) { + _, err = w.chain.WriteBlockAndSetHead(block, receipts, logs, task.state, true) + }) + + tracing.SetAttributes( + span, + attribute.String("hash", hash.String()), + attribute.Int("number", int(block.Number().Uint64())), + attribute.Int("txns", block.Transactions().Len()), + attribute.Int("gas used", int(block.GasUsed())), + attribute.Int("elapsed", int(time.Since(task.createdAt).Milliseconds())), + attribute.Bool("error", err != nil), + ) + }) + if err != nil { log.Error("Failed writing block to chain", "err", err) continue } + log.Info("Successfully sealed new block", "number", block.Number(), "sealhash", sealhash, "hash", hash, "elapsed", common.PrettyDuration(time.Since(task.createdAt))) @@ -758,7 +810,6 @@ func (w *worker) resultLoop() { // Insert the block into the set of pending ones to resultLoop for confirmations w.unconfirmed.Insert(block.NumberU64(), block.Hash()) - case <-w.exitCh: return } @@ -1069,78 +1120,176 @@ func (w *worker) prepareWork(genParams *generateParams) (*environment, error) { // fillTransactions retrieves the pending transactions from the txpool and fills them // into the given sealing block. The transaction selection and ordering strategy can // be customized with the plugin in the future. -func (w *worker) fillTransactions(interrupt *int32, env *environment) { +func (w *worker) fillTransactions(ctx context.Context, interrupt *int32, env *environment) { + ctx, span := tracing.StartSpan(ctx, "fillTransactions") + defer tracing.EndSpan(span) + // Split the pending transactions into locals and remotes // Fill the block with all available pending transactions. - pending := w.eth.TxPool().Pending(true) - localTxs, remoteTxs := make(map[common.Address]types.Transactions), pending - for _, account := range w.eth.TxPool().Locals() { - if txs := remoteTxs[account]; len(txs) > 0 { - delete(remoteTxs, account) - localTxs[account] = txs + + var ( + localTxsCount int + remoteTxsCount int + localTxs = make(map[common.Address]types.Transactions) + remoteTxs map[common.Address]types.Transactions + ) + + tracing.Exec(ctx, "worker.SplittingTransactions", func(ctx context.Context, span trace.Span) { + pending := w.eth.TxPool().Pending(true) + remoteTxs = pending + + for _, account := range w.eth.TxPool().Locals() { + if txs := remoteTxs[account]; len(txs) > 0 { + delete(remoteTxs, account) + localTxs[account] = txs + } } - } - if len(localTxs) > 0 { - txs := types.NewTransactionsByPriceAndNonce(env.signer, localTxs, env.header.BaseFee) - if w.commitTransactions(env, txs, interrupt) { + + localTxsCount = len(localTxs) + remoteTxsCount = len(remoteTxs) + + tracing.SetAttributes( + span, + attribute.Int("len of local txs", localTxsCount), + attribute.Int("len of remote txs", remoteTxsCount), + ) + }) + + var ( + localEnvTCount int + remoteEnvTCount int + committed bool + ) + + if localTxsCount > 0 { + var txs *types.TransactionsByPriceAndNonce + + tracing.Exec(ctx, "worker.LocalTransactionsByPriceAndNonce", func(ctx context.Context, span trace.Span) { + txs = types.NewTransactionsByPriceAndNonce(env.signer, localTxs, env.header.BaseFee) + + tracing.SetAttributes( + span, + attribute.Int("len of tx local Heads", txs.GetTxs()), + ) + }) + + tracing.Exec(ctx, "worker.LocalCommitTransactions", func(ctx context.Context, span trace.Span) { + committed = w.commitTransactions(env, txs, interrupt) + }) + + if committed { return } + + localEnvTCount = env.tcount } - if len(remoteTxs) > 0 { - txs := types.NewTransactionsByPriceAndNonce(env.signer, remoteTxs, env.header.BaseFee) - if w.commitTransactions(env, txs, interrupt) { + + if remoteTxsCount > 0 { + var txs *types.TransactionsByPriceAndNonce + + tracing.Exec(ctx, "worker.RemoteTransactionsByPriceAndNonce", func(ctx context.Context, span trace.Span) { + txs = types.NewTransactionsByPriceAndNonce(env.signer, remoteTxs, env.header.BaseFee) + + tracing.SetAttributes( + span, + attribute.Int("len of tx remote Heads", txs.GetTxs()), + ) + }) + + tracing.Exec(ctx, "worker.RemoteCommitTransactions", func(ctx context.Context, span trace.Span) { + committed = w.commitTransactions(env, txs, interrupt) + }) + + if committed { return } + + remoteEnvTCount = env.tcount } + + tracing.SetAttributes( + span, + attribute.Int("len of final local txs ", localEnvTCount), + attribute.Int("len of final remote txs", remoteEnvTCount), + ) } // generateWork generates a sealing block based on the given parameters. -func (w *worker) generateWork(params *generateParams) (*types.Block, error) { +func (w *worker) generateWork(ctx context.Context, params *generateParams) (*types.Block, error) { work, err := w.prepareWork(params) if err != nil { return nil, err } defer work.discard() - w.fillTransactions(nil, work) - return w.engine.FinalizeAndAssemble(w.chain, work.header, work.state, work.txs, work.unclelist(), work.receipts) + w.fillTransactions(ctx, nil, work) + + return w.engine.FinalizeAndAssemble(ctx, w.chain, work.header, work.state, work.txs, work.unclelist(), work.receipts) } // commitWork generates several new sealing tasks based on the parent block // and submit them to the sealer. -func (w *worker) commitWork(interrupt *int32, noempty bool, timestamp int64) { +func (w *worker) commitWork(ctx context.Context, interrupt *int32, noempty bool, timestamp int64) { start := time.Now() - // Set the coinbase if the worker is running or it's required - var coinbase common.Address - if w.isRunning() { - if w.coinbase == (common.Address{}) { - log.Error("Refusing to mine without etherbase") - return + var ( + work *environment + err error + ) + + tracing.Exec(ctx, "worker.prepareWork", func(ctx context.Context, span trace.Span) { + // Set the coinbase if the worker is running or it's required + var coinbase common.Address + if w.isRunning() { + if w.coinbase == (common.Address{}) { + log.Error("Refusing to mine without etherbase") + return + } + + coinbase = w.coinbase // Use the preset address as the fee recipient } - coinbase = w.coinbase // Use the preset address as the fee recipient - } - work, err := w.prepareWork(&generateParams{ - timestamp: uint64(timestamp), - coinbase: coinbase, + + work, err = w.prepareWork(&generateParams{ + timestamp: uint64(timestamp), + coinbase: coinbase, + }) }) + if err != nil { return } + + ctx, span := tracing.StartSpan(ctx, "commitWork") + defer tracing.EndSpan(span) + + tracing.SetAttributes( + span, + attribute.Int("number", int(work.header.Number.Uint64())), + ) + // Create an empty block based on temporary copied state for // sealing in advance without waiting block execution finished. if !noempty && atomic.LoadUint32(&w.noempty) == 0 { - w.commit(work.copy(), nil, false, start) + err = w.commit(ctx, work.copy(), nil, false, start) + if err != nil { + return + } } + // Fill pending transactions from the txpool - w.fillTransactions(interrupt, work) - w.commit(work.copy(), w.fullTaskHook, true, start) + w.fillTransactions(ctx, interrupt, work) + + err = w.commit(ctx, work.copy(), w.fullTaskHook, true, start) + if err != nil { + return + } // Swap out the old work with the new one, terminating any leftover // prefetcher processes in the mean time and starting a new one. if w.current != nil { w.current.discard() } + w.current = work } @@ -1148,22 +1297,38 @@ func (w *worker) commitWork(interrupt *int32, noempty bool, timestamp int64) { // and commits new work if consensus engine is running. // Note the assumption is held that the mutation is allowed to the passed env, do // the deep copy first. -func (w *worker) commit(env *environment, interval func(), update bool, start time.Time) error { +func (w *worker) commit(ctx context.Context, env *environment, interval func(), update bool, start time.Time) error { if w.isRunning() { + ctx, span := tracing.StartSpan(ctx, "commit") + defer tracing.EndSpan(span) + if interval != nil { interval() } + // Create a local environment copy, avoid the data race with snapshot state. // https://github.com/ethereum/go-ethereum/issues/24299 env := env.copy() - block, err := w.engine.FinalizeAndAssemble(w.chain, env.header, env.state, env.txs, env.unclelist(), env.receipts) + + block, err := w.engine.FinalizeAndAssemble(ctx, w.chain, env.header, env.state, env.txs, env.unclelist(), env.receipts) + + tracing.SetAttributes( + span, + attribute.Int("number", int(block.Number().Uint64())), + attribute.String("hash", block.Hash().String()), + attribute.String("sealhash", w.engine.SealHash(block.Header()).String()), + attribute.Int("len of env.txs", len(env.txs)), + attribute.Bool("error", err != nil), + ) + if err != nil { return err } + // If we're post merge, just ignore if !w.isTTDReached(block.Header()) { select { - case w.taskCh <- &task{receipts: env.receipts, state: env.state, block: block, createdAt: time.Now()}: + case w.taskCh <- &task{ctx: ctx, receipts: env.receipts, state: env.state, block: block, createdAt: time.Now()}: w.unconfirmed.Shift(block.NumberU64() - 1) log.Info("Commit new sealing work", "number", block.Number(), "sealhash", w.engine.SealHash(block.Header()), "uncles", len(env.uncles), "txs", env.tcount, @@ -1178,11 +1343,14 @@ func (w *worker) commit(env *environment, interval func(), update bool, start ti if update { w.updateSnapshot(env) } + return nil } // getSealingBlock generates the sealing block based on the given parameters. func (w *worker) getSealingBlock(parent common.Hash, timestamp uint64, coinbase common.Address, random common.Hash) (*types.Block, error) { + ctx := tracing.WithTracer(context.Background(), otel.GetTracerProvider().Tracer("getSealingBlock")) + req := &getWorkReq{ params: &generateParams{ timestamp: timestamp, @@ -1194,13 +1362,16 @@ func (w *worker) getSealingBlock(parent common.Hash, timestamp uint64, coinbase noExtra: true, }, result: make(chan *types.Block, 1), + ctx: ctx, } + select { case w.getWorkCh <- req: block := <-req.result if block == nil { return nil, req.err } + return block, nil case <-w.exitCh: return nil, errors.New("miner closed") diff --git a/tests/bor/helper.go b/tests/bor/helper.go index f5d80bbebf..ddddc97572 100644 --- a/tests/bor/helper.go +++ b/tests/bor/helper.go @@ -3,6 +3,7 @@ package bor import ( + "context" "encoding/hex" "encoding/json" "fmt" @@ -172,8 +173,10 @@ func buildNextBlock(t *testing.T, _bor consensus.Engine, chain *core.BlockChain, b.addTxWithChain(chain, state, tx, addr) } + ctx := context.Background() + // Finalize and seal the block - block, _ := _bor.FinalizeAndAssemble(chain, b.header, state, b.txs, nil, b.receipts) + block, _ := _bor.FinalizeAndAssemble(ctx, chain, b.header, state, b.txs, nil, b.receipts) // Write state changes to db root, err := state.Commit(chain.Config().IsEIP158(b.header.Number)) @@ -187,7 +190,7 @@ func buildNextBlock(t *testing.T, _bor consensus.Engine, chain *core.BlockChain, res := make(chan *types.Block, 1) - err = _bor.Seal(chain, block, res, nil) + err = _bor.Seal(ctx, chain, block, res, nil) if err != nil { // an error case - sign manually sign(t, header, signer, borConfig) diff --git a/tests/state_test_util.go b/tests/state_test_util.go index 4fd3cf76b2..65f93bfbe3 100644 --- a/tests/state_test_util.go +++ b/tests/state_test_util.go @@ -24,6 +24,8 @@ import ( "strconv" "strings" + "golang.org/x/crypto/sha3" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" "github.com/ethereum/go-ethereum/common/math" @@ -37,7 +39,6 @@ import ( "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/params" "github.com/ethereum/go-ethereum/rlp" - "golang.org/x/crypto/sha3" ) // StateTest checks transaction processing without block context. @@ -217,15 +218,16 @@ func (t *StateTest) RunNoVerify(subtest StateSubtest, vmconfig vm.Config, snapsh // Prepare the EVM. txContext := core.NewEVMTxContext(msg) - context := core.NewEVMBlockContext(block.Header(), nil, &t.json.Env.Coinbase) - context.GetHash = vmTestBlockHash - context.BaseFee = baseFee + evmContext := core.NewEVMBlockContext(block.Header(), nil, &t.json.Env.Coinbase) + evmContext.GetHash = vmTestBlockHash + evmContext.BaseFee = baseFee if t.json.Env.Random != nil { rnd := common.BigToHash(t.json.Env.Random) - context.Random = &rnd - context.Difficulty = big.NewInt(0) + evmContext.Random = &rnd + evmContext.Difficulty = big.NewInt(0) } - evm := vm.NewEVM(context, txContext, statedb, config, vmconfig) + + evm := vm.NewEVM(evmContext, txContext, statedb, config, vmconfig) // Execute the message. snapshot := statedb.Snapshot() gaspool := new(core.GasPool)