[Execution] Ingestion Block Queue #5248
Conversation
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #5248 +/- ##
==========================================
- Coverage 55.97% 55.64% -0.33%
==========================================
Files 1022 1038 +16
Lines 99705 101598 +1893
==========================================
+ Hits 55807 56539 +732
- Misses 39598 40715 +1117
- Partials 4300 4344 +44
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Sentry. |
| "github.com/onflow/flow-go/module/mempool/entity" | ||
| ) | ||
|
|
||
| // BlockQueue keeps track of state of blocks and determines which blocks are executable |
There was a problem hiding this comment.
Not sure if there is a better name than this. Suggestions are welcome.
0b6a5ba to
0ed04d1
Compare
There was a problem hiding this comment.
- I'm concerned how this will be used singe a lot of different methods (that are presumably called from different places) are returning a
[]*entity.ExecutableBlock.
This might be messy if that information needs to then be forwarded to another component.
Maybe having a channel where *entity.ExecutableBlock would be pushed instead of always returning them would be better.
-
All of the
On*methods are blocking. Is this ok? -
A lot of locking is going on here. Will this be a problem? Would a syncmap be better in this case?
| // we have already received this block, and its parent still has not been executed yet | ||
| if executable.StartState == nil && parentFinalState == nil { | ||
| return nil, nil, nil | ||
| } | ||
|
|
||
| // this is an edge case where parentFinalState is provided, and its parent block exists | ||
| // in the queue but has not been marked as executed yet (OnBlockExecuted(parent) is not called), | ||
| // in this case, we will internally call OnBlockExecuted(parentBlockID, parentFinalState). | ||
| // there is no need to create the executable block again, since it's already created. | ||
| if executable.StartState == nil && parentFinalState != nil { | ||
| executables, err := q.onBlockExecuted(block.Header.ParentID, *parentFinalState) | ||
| if err != nil { | ||
| return nil, nil, fmt.Errorf("receiving block %v with parent commitment %v, but parent block %v already exists with no commitment, fail to call mark parent as executed: %w", | ||
| blockID, *parentFinalState, block.Header.ParentID, err) | ||
| } | ||
|
|
||
| // we already have this block, its collection must have been fetched, so we only return the | ||
| // executables from marking its parent as executed. | ||
| return nil, executables, nil | ||
| } | ||
|
|
||
| // this is an edge case could be ignored | ||
| if executable.StartState != nil && parentFinalState == nil { | ||
| q.log.Warn(). | ||
| Str("blockID", blockID.String()). | ||
| Hex("parentID", block.Header.ParentID[:]). | ||
| Msg("edge case: receiving block with no parent commitment, but its parent block actually has been executed") | ||
| return nil, nil, nil | ||
| } | ||
|
|
||
| // this is an exception that should not happen | ||
| if *executable.StartState != *parentFinalState { | ||
| return nil, nil, | ||
| fmt.Errorf("block %s has already been executed with a different parent final state, %v != %v", | ||
| blockID, *executable.StartState, parentFinalState) | ||
| } | ||
|
|
||
| return nil, nil, nil |
There was a problem hiding this comment.
This could be packaged into a new function to reduce the complexity of this one
| func (q *BlockQueue) GetMissingCollections(blockID flow.Identifier) ( | ||
| []*MissingCollection, *flow.StateCommitment, error) { |
There was a problem hiding this comment.
formatting here is a bit weird.
There was a problem hiding this comment.
What do you mean? Can you be more specific?
| return nil, fmt.Errorf("parent block %s of block %s is in the queue", | ||
| block.Block.Header.ParentID, blockID) |
There was a problem hiding this comment.
Sanity checks like this make me think that this should use the irrecoverable.SignalerContext to throw this error and shutdown the component entirely.
There was a problem hiding this comment.
We could leave the decision to the caller. The caller can decide how to handle this exception, either throw as a irrecoverable error or handle it differently.
| return nil, executables, nil | ||
| } | ||
|
|
||
| // this is an edge case could be ignored |
There was a problem hiding this comment.
can you elaborate in this comment about how this case would happen and why it's safe to ignore
| // update collection | ||
| colInfo.Collection.Transactions = collection.Transactions | ||
|
|
||
| // check if any block, which includes this collection, become executable |
There was a problem hiding this comment.
| // check if any block, which includes this collection, become executable | |
| // check if any block, which includes this collection, became executable |
| // 2. if a block's parent is not executed, then the parent block must be passed in first | ||
| // 3. if a block's parent is executed, then the parent's finalState must be passed in | ||
| // It returns (nil, nil, nil) if this block is a duplication | ||
| func (q *BlockQueue) OnBlock(block *flow.Block, parentFinalState *flow.StateCommitment) ( |
There was a problem hiding this comment.
The On* naming is a little confusing since we typically use that for non-blocking signal functions. Here they are blocking and return some data. How about renaming them to something like HandleBlock?
2e1f681 to
2ddeaf7
Compare
Yes, they are called in multiple places, but all in the same file (the ingestion core module), and the logic to handle them are reused, such as here the Regarding the ingestion core, it is a stateless module, the point of the separating the ingestion core and block queue is so that so a stateful machine, the block queue handles all state transition, and uses lock to protect its consistency, and ingestion core wraps the block queue, and connect all external dependencies and run side effects, such as reading and writing to database, listening to stop control. And similarly, there is this
blocking should be OK. Because the BlockQueue's data are all hold in memory, and its methods are only doing updating internal state in memory, such be very fast.
The lock is useful to guarantee
|
Working towards #5297
This PR simplifies the ingestion engine's mempool module with a block queue module.