[Consensus] Queue for synchronization.Engine
#910
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…ed implementing queue consuming in engine
…tributor. Updated all nodes
durkmurder
requested review from
AlexHentschel,
jordanschalm,
m4ksio,
ramtinms,
yhassanzadeh13 and
zhangchiqing
as code owners
durkmurder
removed request for
m4ksio,
ramtinms,
jordanschalm and
yhassanzadeh13
zhangchiqing
approved these changes
Jul 2, 2021
|
|
||
| // setupRequestMessageHandler initializes the inbound queues and the MessageHandler for UNTRUSTED requests. | ||
| func (e *Engine) setupRequestMessageHandler() { | ||
| e.pendingSyncRequests = NewRequestQueue(defaultSyncRequestQueueCapacity) |
There was a problem hiding this comment.
Suggested change
| e.pendingSyncRequests = NewRequestQueue(defaultSyncRequestQueueCapacity) | |
| // RequestQueue deduplicates requests by keeping only one sync request for each requester. | |
| e.pendingSyncRequests = NewRequestQueue(defaultSyncRequestQueueCapacity) |
added 2 commits
July 4, 2021 21:36
… implemented by `badger.Headers`
• reduced locking in `synchronization.Core`
AlexHentschel
approved these changes
Jul 5, 2021
There was a problem hiding this comment.
Comments & Suggestions:
- It would be great to fix this performance bottleneck:
- In contrast to
badger.Headers, which has a cache for heights,badger.blocksdoes not have such a cache and instead always hits the data base.
- In contrast to
- I think we could inline method as it is only used once.
flow-go/engine/common/synchronization/engine.go
Lines 581 to 596 in 5fd2ff2
- We might be able to reduce lock congestion in
Core. HandleHeight:- we always lock right away, but on the happy path when the node is up to date, this code would return right away: .
flow-go/module/synchronization/core.go
Lines 102 to 105 in 5fd2ff2
WithinToleranceis fully concurrency safe without any locks (we also call it externally, here). Hence, we could move the lock further down into theifstatement, where we actually updateCore's state.
- we always lock right away, but on the happy path when the node is up to date, this code would return right away:
Implemented these suggestions☝️ in my PR #926.
The suggestions below (code comments) are not implemented in PR 926.
Comment on lines
65
to
77
|
|
||
| // we keep reducing the cache size until we are at limit again | ||
| for len(q.requests) >= int(q.limit) { | ||
|
|
||
| // eject first element using go map properties | ||
| var key flow.Identifier | ||
| for originID := range q.requests { | ||
| key = originID | ||
| break | ||
| } | ||
|
|
||
| delete(q.requests, key) | ||
| } |
There was a problem hiding this comment.
I feel we could write this much more concisely:
Suggested change
| // we keep reducing the cache size until we are at limit again | |
| for len(q.requests) >= int(q.limit) { | |
| // eject first element using go map properties | |
| var key flow.Identifier | |
| for originID := range q.requests { | |
| key = originID | |
| break | |
| } | |
| delete(q.requests, key) | |
| } | |
| // we keep reducing the cache size until we are at limit again | |
| for overCapacity := len(q.requests) - int(q.limit); overCapacity >= 0; overCapacity -- { | |
| for originID := range q.requests { | |
| delete(q.requests, originID) | |
| } | |
| } |
Did I miss something?
Comment on lines
32
to
34
| // first try to eject if we are at max capacity, we need to do this way | ||
| // to prevent a situation where just inserted item gets ejected | ||
| q.reduce() |
There was a problem hiding this comment.
Consider the following scenario:
- the queue is at max capacity
- it contains already an element from Origin
A - we are attempting to put a new element from
Ainto the queue
The result is that:
- the old message from A is overwritten by the new one (desired).
- but we also ejected some other message (which seems not very intuitive)
Suggestion:
Suggested change
| // first try to eject if we are at max capacity, we need to do this way | |
| // to prevent a situation where just inserted item gets ejected | |
| q.reduce() | |
| if _, found := q.requests[message.OriginID]; !found { | |
| // if no message from the origin is stored, make sure we have room to store the new message: | |
| q.reduce() | |
| } |
| filter.Not(filter.HasNodeID(e.me.NodeID())), | ||
| )) | ||
| if err != nil { | ||
| return fmt.Errorf("could not send get consensus participants: %w", err) |
There was a problem hiding this comment.
Suggested change
| return fmt.Errorf("could not send get consensus participants: %w", err) | |
| return fmt.Errorf("could get consensus participants at latest finalized block: %w", err) |
Merged
…queue_-_suggestions suggestions for PR 910
Codecov Report
@@ Coverage Diff @@
## master #910 +/- ##
==========================================
+ Coverage 56.56% 56.69% +0.13%
==========================================
Files 424 425 +1
Lines 25008 25146 +138
==========================================
+ Hits 14145 14257 +112
- Misses 8941 8955 +14
- Partials 1922 1934 +12
Flags with carried forward coverage won't be shown. Click here to find out more.
Continue to review full report at Codecov.
|
…hub.com/onflow/flow-go into yurii/5615-synchronization-engine-queue
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
dapperlabs/flow-go/issues/5615
Context
This PR implements support of inbound queue for messages that are submitted by network layer and/or other engines.
As described in issue this PR mainly implements messages queening logic as it is implemented in other consensus engines(sealing, matching, compliance). Also currently there is quite inefficient access to final protocol state, this PR contains changes for this as well.
In proposed implementation we have two separate goroutines which handle requests and responses in parallel, maybe that's not needed but I though that it's a good idea to more evenly distribute load.
Contributions of this PR