Skip to content
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

miner/worker: broadcast block immediately once sealed #2576

Merged
merged 1 commit into from
Jul 16, 2024
Merged

miner/worker: broadcast block immediately once sealed #2576

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
18 changes: 11 additions & 7 deletions core/blockchain.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1908,26 +1908,30 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.

// WriteBlockAndSetHead writes the given block and all associated state to the database,
// and applies the block as the new chain head.
func (bc *BlockChain) WriteBlockAndSetHead(block *types.Block, receipts []*types.Receipt, logs []*types.Log, state *state.StateDB, emitHeadEvent bool) (status WriteStatus, err error) {
func (bc *BlockChain) WriteBlockAndSetHead(block *types.Block, receipts []*types.Receipt, logs []*types.Log, state *state.StateDB, emitHeadEvent bool, mux *event.TypeMux) (status WriteStatus, err error) {
if !bc.chainmu.TryLock() {
return NonStatTy, errChainStopped
}
defer bc.chainmu.Unlock()

return bc.writeBlockAndSetHead(block, receipts, logs, state, emitHeadEvent)
return bc.writeBlockAndSetHead(block, receipts, logs, state, emitHeadEvent, mux)
}

// writeBlockAndSetHead is the internal implementation of WriteBlockAndSetHead.
// This function expects the chain mutex to be held.
func (bc *BlockChain) writeBlockAndSetHead(block *types.Block, receipts []*types.Receipt, logs []*types.Log, state *state.StateDB, emitHeadEvent bool) (status WriteStatus, err error) {
if err := bc.writeBlockWithState(block, receipts, state); err != nil {
return NonStatTy, err
}
func (bc *BlockChain) writeBlockAndSetHead(block *types.Block, receipts []*types.Receipt, logs []*types.Log, state *state.StateDB, emitHeadEvent bool, mux *event.TypeMux) (status WriteStatus, err error) {
currentBlock := bc.CurrentBlock()
reorg, err := bc.forker.ReorgNeededWithFastFinality(currentBlock, block.Header())
if err != nil {
return NonStatTy, err
}
if reorg && mux != nil {
mux.Post(NewSealedBlockEvent{Block: block})
}

if err := bc.writeBlockWithState(block, receipts, state); err != nil {
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

writeBlockWithState cost much, may be hundreds of milliseconds

we can post a NewSealedBlockEvent before it, so broadcast it early @emailtovamos

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, understood.
But if writeBlockWithState() fails for some reason then it is still okay to post a NewSealedBlockEvent ? This is my only doubt.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, understood. But if writeBlockWithState() fails for some reason then it is still okay to post a NewSealedBlockEvent ? This is my only doubt.

every verification has been passed when do writeBlockWithState.
if a miner has a bad db, cause failed to writeBlockWithState. It's ok, others will write the mined block into db.
if the block is bad, cause failed to writeBlockWithState. It's ok, others will reject the mined block into db.

return NonStatTy, err
}
if reorg {
// Reorganise the chain if the parent is not the head block
if block.ParentHash() != currentBlock.Hash() {
Expand Down Expand Up @@ -2300,7 +2304,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error)
// Don't set the head, only insert the block
err = bc.writeBlockWithState(block, receipts, statedb)
} else {
status, err = bc.writeBlockAndSetHead(block, receipts, logs, statedb, false)
status, err = bc.writeBlockAndSetHead(block, receipts, logs, statedb, false, nil)
}
if err != nil {
return it.index, err
Expand Down
5 changes: 4 additions & 1 deletion core/events.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,10 @@ type NewTxsEvent struct{ Txs []*types.Transaction }
// ReannoTxsEvent is posted when a batch of local pending transactions exceed a specified duration.
type ReannoTxsEvent struct{ Txs []*types.Transaction }

// NewMinedBlockEvent is posted when a block has been imported.
// NewSealedBlockEvent is posted when a block has been sealed.
zzzckck marked this conversation as resolved.
Show resolved Hide resolved
type NewSealedBlockEvent struct{ Block *types.Block }

// NewMinedBlockEvent is posted when a block has been mined.
type NewMinedBlockEvent struct{ Block *types.Block }

// RemovedLogsEvent is posted when a reorg happens
Expand Down
7 changes: 4 additions & 3 deletions eth/handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -729,7 +729,7 @@ func (h *handler) Start(maxPeers int, maxPeersPerIP int) {

// broadcast mined blocks
h.wg.Add(1)
h.minedBlockSub = h.eventMux.Subscribe(core.NewMinedBlockEvent{})
h.minedBlockSub = h.eventMux.Subscribe(core.NewMinedBlockEvent{}, core.NewSealedBlockEvent{})
go h.minedBroadcastLoop()

// start sync handlers
Expand Down Expand Up @@ -946,8 +946,9 @@ func (h *handler) minedBroadcastLoop() {
if obj == nil {
continue
}
if ev, ok := obj.Data.(core.NewMinedBlockEvent); ok {
h.BroadcastBlock(ev.Block, true) // First propagate block to peers
if ev, ok := obj.Data.(core.NewSealedBlockEvent); ok {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am not sure how exactly this improves things? Could you clarify bit more?
Now we are waiting for a sealing to happen before broadcasting rather than mining to happen before broadcasting it.
But in (w *worker) resultLoop() function, there are several checks made before a mining event is emitted which could fail e.g. double sign. Wouldn't that be a problem?

h.BroadcastBlock(ev.Block, true) // Propagate block to peers
} else if ev, ok := obj.Data.(core.NewMinedBlockEvent); ok {
h.BroadcastBlock(ev.Block, false) // Only then announce to the rest
}
case <-h.stopCh:
Expand Down
2 changes: 1 addition & 1 deletion miner/worker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -665,7 +665,7 @@ func (w *worker) resultLoop() {
// Commit block and state to database.
task.state.SetExpectedStateRoot(block.Root())
start := time.Now()
status, err := w.chain.WriteBlockAndSetHead(block, receipts, logs, task.state, true)
status, err := w.chain.WriteBlockAndSetHead(block, receipts, logs, task.state, true, w.mux)
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@emailtovamos this happens after double sign
maybe NewSealedBlockEvent is not exactly accurate, but I have not got a better word

if status != core.CanonStatTy {
if err != nil {
log.Error("Failed writing block to chain", "err", err, "status", status)
Expand Down
Loading