diff --git a/core/blockchain.go b/core/blockchain.go index ae92386dc2e5..2705ed1349df 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -309,6 +309,7 @@ type BlockChain struct { chainHeadFeed event.Feed logsFeed event.Feed blockProcFeed event.Feed + newPayloadFeed event.Feed // Feed for engine API newPayload events blockProcCounter int32 scope event.SubscriptionScope genesisBlock *types.Block diff --git a/core/blockchain_reader.go b/core/blockchain_reader.go index 4894523b0e58..ee15c152c498 100644 --- a/core/blockchain_reader.go +++ b/core/blockchain_reader.go @@ -522,3 +522,13 @@ func (bc *BlockChain) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscript func (bc *BlockChain) SubscribeBlockProcessingEvent(ch chan<- bool) event.Subscription { return bc.scope.Track(bc.blockProcFeed.Subscribe(ch)) } + +// SubscribeNewPayloadEvent registers a subscription for NewPayloadEvent. +func (bc *BlockChain) SubscribeNewPayloadEvent(ch chan<- NewPayloadEvent) event.Subscription { + return bc.scope.Track(bc.newPayloadFeed.Subscribe(ch)) +} + +// SendNewPayloadEvent sends a NewPayloadEvent to subscribers. +func (bc *BlockChain) SendNewPayloadEvent(ev NewPayloadEvent) { + bc.newPayloadFeed.Send(ev) +} diff --git a/core/events.go b/core/events.go index ef0de3242621..ed853f1790ac 100644 --- a/core/events.go +++ b/core/events.go @@ -17,6 +17,9 @@ package core import ( + "time" + + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" ) @@ -35,3 +38,10 @@ type ChainEvent struct { type ChainHeadEvent struct { Header *types.Header } + +// NewPayloadEvent is posted when engine_newPayloadVX processes a block. +type NewPayloadEvent struct { + Hash common.Hash + Number uint64 + ProcessingTime time.Duration +} diff --git a/eth/api_backend.go b/eth/api_backend.go index 766a99fc1ef6..3f826b786184 100644 --- a/eth/api_backend.go +++ b/eth/api_backend.go @@ -315,6 +315,11 @@ func (b *EthAPIBackend) SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) e return b.eth.BlockChain().SubscribeChainHeadEvent(ch) } +// SubscribeNewPayloadEvent registers a subscription for NewPayloadEvent. +func (b *EthAPIBackend) SubscribeNewPayloadEvent(ch chan<- core.NewPayloadEvent) event.Subscription { + return b.eth.BlockChain().SubscribeNewPayloadEvent(ch) +} + func (b *EthAPIBackend) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription { return b.eth.BlockChain().SubscribeLogsEvent(ch) } diff --git a/eth/catalyst/api.go b/eth/catalyst/api.go index d6d3f579364d..4fabe95ef3cb 100644 --- a/eth/catalyst/api.go +++ b/eth/catalyst/api.go @@ -30,6 +30,7 @@ import ( "github.com/ethereum/go-ethereum/beacon/engine" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" + "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/eth" @@ -773,7 +774,9 @@ func (api *ConsensusAPI) newPayload(params engine.ExecutableData, versionedHashe return engine.PayloadStatusV1{Status: engine.ACCEPTED}, nil } log.Trace("Inserting block without sethead", "hash", block.Hash(), "number", block.Number()) + start := time.Now() proofs, err := api.eth.BlockChain().InsertBlockWithoutSetHead(block, witness) + processingTime := time.Since(start) if err != nil { log.Warn("NewPayload: inserting block failed", "error", err) @@ -786,6 +789,13 @@ func (api *ConsensusAPI) newPayload(params engine.ExecutableData, versionedHashe } hash := block.Hash() + // Emit NewPayloadEvent for ethstats reporting + api.eth.BlockChain().SendNewPayloadEvent(core.NewPayloadEvent{ + Hash: hash, + Number: block.NumberU64(), + ProcessingTime: processingTime, + }) + // If witness collection was requested, inject that into the result too var ow *hexutil.Bytes if proofs != nil { diff --git a/ethstats/ethstats.go b/ethstats/ethstats.go index b6191baa12b9..c17e22516504 100644 --- a/ethstats/ethstats.go +++ b/ethstats/ethstats.go @@ -63,6 +63,7 @@ const ( type backend interface { SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription + SubscribeNewPayloadEvent(ch chan<- core.NewPayloadEvent) event.Subscription CurrentHeader() *types.Header HeaderByNumber(ctx context.Context, number rpc.BlockNumber) (*types.Header, error) Stats() (pending int, queued int) @@ -92,8 +93,9 @@ type Service struct { pongCh chan struct{} // Pong notifications are fed into this channel histCh chan []uint64 // History request block numbers are fed into this channel - headSub event.Subscription - txSub event.Subscription + headSub event.Subscription + txSub event.Subscription + newPayloadSub event.Subscription } // connWrapper is a wrapper to prevent concurrent-write or concurrent-read on the @@ -198,7 +200,9 @@ func (s *Service) Start() error { s.headSub = s.backend.SubscribeChainHeadEvent(chainHeadCh) txEventCh := make(chan core.NewTxsEvent, txChanSize) s.txSub = s.backend.SubscribeNewTxsEvent(txEventCh) - go s.loop(chainHeadCh, txEventCh) + newPayloadCh := make(chan core.NewPayloadEvent, chainHeadChanSize) + s.newPayloadSub = s.backend.SubscribeNewPayloadEvent(newPayloadCh) + go s.loop(chainHeadCh, txEventCh, newPayloadCh) log.Info("Stats daemon started") return nil @@ -208,18 +212,20 @@ func (s *Service) Start() error { func (s *Service) Stop() error { s.headSub.Unsubscribe() s.txSub.Unsubscribe() + s.newPayloadSub.Unsubscribe() log.Info("Stats daemon stopped") return nil } // loop keeps trying to connect to the netstats server, reporting chain events // until termination. -func (s *Service) loop(chainHeadCh chan core.ChainHeadEvent, txEventCh chan core.NewTxsEvent) { +func (s *Service) loop(chainHeadCh chan core.ChainHeadEvent, txEventCh chan core.NewTxsEvent, newPayloadCh chan core.NewPayloadEvent) { // Start a goroutine that exhausts the subscriptions to avoid events piling up var ( - quitCh = make(chan struct{}) - headCh = make(chan *types.Header, 1) - txCh = make(chan struct{}, 1) + quitCh = make(chan struct{}) + headCh = make(chan *types.Header, 1) + txCh = make(chan struct{}, 1) + newPayloadEvCh = make(chan core.NewPayloadEvent, 1) ) go func() { var lastTx mclock.AbsTime @@ -246,11 +252,20 @@ func (s *Service) loop(chainHeadCh chan core.ChainHeadEvent, txEventCh chan core default: } + // Notify of new payload events, but drop if too frequent + case ev := <-newPayloadCh: + select { + case newPayloadEvCh <- ev: + default: + } + // node stopped case <-s.txSub.Err(): break HandleLoop case <-s.headSub.Err(): break HandleLoop + case <-s.newPayloadSub.Err(): + break HandleLoop } } close(quitCh) @@ -336,6 +351,10 @@ func (s *Service) loop(chainHeadCh chan core.ChainHeadEvent, txEventCh chan core if err = s.reportPending(conn); err != nil { log.Warn("Post-block transaction stats report failed", "err", err) } + case ev := <-newPayloadEvCh: + if err = s.reportNewPayload(conn, ev); err != nil { + log.Warn("New payload stats report failed", "err", err) + } case <-txCh: if err = s.reportPending(conn); err != nil { log.Warn("Transaction stats report failed", "err", err) @@ -600,6 +619,33 @@ func (s uncleStats) MarshalJSON() ([]byte, error) { return []byte("[]"), nil } +// newPayloadStats is the information to report about new payload events. +type newPayloadStats struct { + Number uint64 `json:"number"` + Hash common.Hash `json:"hash"` + ProcessingTime uint64 `json:"processingTime"` // nanoseconds +} + +// reportNewPayload reports a new payload event to the stats server. +func (s *Service) reportNewPayload(conn *connWrapper, ev core.NewPayloadEvent) error { + details := &newPayloadStats{ + Number: ev.Number, + Hash: ev.Hash, + ProcessingTime: uint64(ev.ProcessingTime.Nanoseconds()), + } + + log.Trace("Sending new payload to ethstats", "number", details.Number, "hash", details.Hash) + + stats := map[string]interface{}{ + "id": s.node, + "block": details, + } + report := map[string][]interface{}{ + "emit": {"block_new_payload", stats}, + } + return conn.WriteJSON(report) +} + // reportBlock retrieves the current chain head and reports it to the stats server. func (s *Service) reportBlock(conn *connWrapper, header *types.Header) error { // Gather the block details from the header or block chain