Skip to content

dynamic lambda: current round awareness for AttachReceivedAt, AttachValidatedAt #5715

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

Closed
cce wants to merge 2 commits into from
Closed

dynamic lambda: current round awareness for AttachReceivedAt, AttachValidatedAt #5715

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
93a437a
try adding pipeline awareness to AttachValidatedAt
cce Aug 29, 2023
326561c
add currentRound handling to AttachReceivedAt and AttachValidatedAt
cce Aug 30, 2023
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
8 changes: 4 additions & 4 deletions agreement/demux.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -199,13 +199,13 @@ func (d *demux) next(s *Service, deadline Deadline, fastDeadline Deadline, curre

switch e.t() {
case payloadVerified:
e = e.(messageEvent).AttachValidatedAt(s.Clock.Since())
e = e.(messageEvent).AttachValidatedAt(s.Clock.Since(), currentRound)
case payloadPresent, votePresent:
e = e.(messageEvent).AttachReceivedAt(s.Clock.Since())
e = e.(messageEvent).AttachReceivedAt(s.Clock.Since(), currentRound)
case voteVerified:
// if this is a proposal vote (step 0), record the validatedAt time on the vote
if e.(messageEvent).Input.UnauthenticatedVote.R.Step == 0 {
e = e.(messageEvent).AttachValidatedAt(s.Clock.Since())
if e.(messageEvent).Input.Vote.R.Step == 0 {
e = e.(messageEvent).AttachValidatedAt(s.Clock.Since(), currentRound)
}
}
}()
Expand Down
41 changes: 34 additions & 7 deletions agreement/events.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -975,22 +975,44 @@ func (e checkpointEvent) AttachConsensusVersion(v ConsensusVersionView) external
return e
}

func (e messageEvent) AttachValidatedAt(d time.Duration) messageEvent {
// AttachValidatedAt looks for an unauthenticatedProposal or unauthenticatedVote
// inside a payloadVerified or voteVerified messageEvent, and attaches the given
// time to the proposal's receivedAt field, if the message's round matches the
// current round. If the message will be pipelined (it is for currentRound+1) then
// a duration of 1ns will be used.
func (e messageEvent) AttachValidatedAt(d time.Duration, currentRound round) messageEvent {
switch e.T {
case payloadVerified:
e.Input.Proposal.validatedAt = d
switch e.Input.Proposal.Round() {
case currentRound:
e.Input.Proposal.validatedAt = d
case currentRound + 1:
e.Input.Proposal.validatedAt = 1
}
case voteVerified:
e.Input.Vote.validatedAt = d
switch e.Input.Vote.R.Round {
case currentRound:
e.Input.Vote.validatedAt = d
case currentRound + 1:
e.Input.Vote.validatedAt = 1
}
}
return e
}

// AttachReceivedAt looks for an unauthenticatedProposal inside a
// payloadPresent or votePresent messageEvent, and attaches the given
// time to the proposal's receivedAt field.
func (e messageEvent) AttachReceivedAt(d time.Duration) messageEvent {
// time to the proposal's receivedAt field, if the message's round matches the
// current round. If the message will be pipelined (it is for currentRound+1) then
// a duration of 1ns will be used.
func (e messageEvent) AttachReceivedAt(d time.Duration, currentRound round) messageEvent {
if e.T == payloadPresent {
e.Input.UnauthenticatedProposal.receivedAt = d
switch e.Input.UnauthenticatedProposal.Round() {
case currentRound:
e.Input.UnauthenticatedProposal.receivedAt = d
case currentRound + 1:
e.Input.UnauthenticatedProposal.receivedAt = 1
}
} else if e.T == votePresent {
// Check for non-nil Tail, indicating this votePresent event
// contains a synthetic payloadPresent event that was attached
Expand All @@ -999,7 +1021,12 @@ func (e messageEvent) AttachReceivedAt(d time.Duration) messageEvent {
// The tail event is payloadPresent, serialized together
// with the proposal vote as a single CompoundMessage
// using a protocol.ProposalPayloadTag network message.
e.Tail.Input.UnauthenticatedProposal.receivedAt = d
switch e.Tail.Input.UnauthenticatedProposal.Round() {
case currentRound:
e.Tail.Input.UnauthenticatedProposal.receivedAt = d
case currentRound + 1:
e.Tail.Input.UnauthenticatedProposal.receivedAt = 1
}
}
}
return e
Expand Down
26 changes: 13 additions & 13 deletions agreement/player_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3249,15 +3249,15 @@ func TestPlayerRetainsReceivedValidatedAtOneSample(t *testing.T) {
// send voteVerified message
vVote := helper.MakeVerifiedVote(t, 0, r-1, p, propose, *pV)
inMsg := messageEvent{T: voteVerified, Input: message{Vote: vVote, UnauthenticatedVote: vVote.u()}}
inMsg = inMsg.AttachValidatedAt(501 * time.Millisecond)
inMsg = inMsg.AttachValidatedAt(501*time.Millisecond, r-1)
err, panicErr := pM.transition(inMsg)
require.NoError(t, err)
require.NoError(t, panicErr)

// send payloadPresent message
m := message{UnauthenticatedProposal: pP.u()}
inMsg = messageEvent{T: payloadPresent, Input: m}
inMsg = inMsg.AttachReceivedAt(time.Second)
inMsg = inMsg.AttachReceivedAt(time.Second, r-1)
err, panicErr = pM.transition(inMsg)
require.NoError(t, err)
require.NoError(t, panicErr)
Expand Down Expand Up @@ -3289,15 +3289,15 @@ func TestPlayerRetainsReceivedValidatedAtForHistoryWindow(t *testing.T) {
vVote := helper.MakeVerifiedVote(t, 0, r+round(i)-1, p, propose, *pV)
inMsg := messageEvent{T: voteVerified, Input: message{Vote: vVote, UnauthenticatedVote: vVote.u()}}
timestamp := 500 + i
inMsg = inMsg.AttachValidatedAt(time.Duration(timestamp) * time.Millisecond)
inMsg = inMsg.AttachValidatedAt(time.Duration(timestamp)*time.Millisecond, r+round(i)-1)
err, panicErr := pM.transition(inMsg)
require.NoError(t, err)
require.NoError(t, panicErr)

// send payloadPresent message
m := message{UnauthenticatedProposal: pP.u()}
inMsg = messageEvent{T: payloadPresent, Input: m}
inMsg = inMsg.AttachReceivedAt(time.Second)
inMsg = inMsg.AttachReceivedAt(time.Second, r+round(i)-1)
err, panicErr = pM.transition(inMsg)
require.NoError(t, err)
require.NoError(t, panicErr)
Expand Down Expand Up @@ -3331,7 +3331,7 @@ func TestPlayerRetainsReceivedValidatedAtPPOneSample(t *testing.T) {
proposalMsg := message{UnauthenticatedProposal: pP.u()}
compoundMsg := messageEvent{T: votePresent, Input: unverifiedVoteMsg,
Tail: &messageEvent{T: payloadPresent, Input: proposalMsg}}
inMsg := compoundMsg.AttachReceivedAt(time.Second) // call AttachReceivedAt like demux would
inMsg := compoundMsg.AttachReceivedAt(time.Second, r-1) // call AttachReceivedAt like demux would
err, panicErr := pM.transition(inMsg)
require.NoError(t, err)
require.NoError(t, panicErr)
Expand All @@ -3343,7 +3343,7 @@ func TestPlayerRetainsReceivedValidatedAtPPOneSample(t *testing.T) {
// send voteVerified
verifiedVoteMsg := message{Vote: vVote, UnauthenticatedVote: vVote.u()}
inMsg = messageEvent{T: voteVerified, Input: verifiedVoteMsg, TaskIndex: 1}
inMsg = inMsg.AttachValidatedAt(502 * time.Millisecond)
inMsg = inMsg.AttachValidatedAt(502*time.Millisecond, r-1)
err, panicErr = pM.transition(inMsg)
require.NoError(t, err)
require.NoError(t, panicErr)
Expand Down Expand Up @@ -3380,7 +3380,7 @@ func TestPlayerRetainsReceivedValidatedAtPPForHistoryWindow(t *testing.T) {
compoundMsg := messageEvent{T: votePresent, Input: unverifiedVoteMsg,
Tail: &messageEvent{T: payloadPresent, Input: proposalMsg}}

inMsg := compoundMsg.AttachReceivedAt(time.Second) // call AttachReceivedAt like demux would
inMsg := compoundMsg.AttachReceivedAt(time.Second, r+round(i)-1) // call AttachReceivedAt like demux would
err, panicErr := pM.transition(inMsg)
require.NoError(t, err)
require.NoError(t, panicErr)
Expand All @@ -3393,7 +3393,7 @@ func TestPlayerRetainsReceivedValidatedAtPPForHistoryWindow(t *testing.T) {
verifiedVoteMsg := message{Vote: vVote, UnauthenticatedVote: vVote.u()}
inMsg = messageEvent{T: voteVerified, Input: verifiedVoteMsg, TaskIndex: 1}
timestamp := 500 + i
inMsg = inMsg.AttachValidatedAt(time.Duration(timestamp) * time.Millisecond)
inMsg = inMsg.AttachValidatedAt(time.Duration(timestamp)*time.Millisecond, r+round(i)-1)
err, panicErr = pM.transition(inMsg)
require.NoError(t, err)
require.NoError(t, panicErr)
Expand Down Expand Up @@ -3437,7 +3437,7 @@ func TestPlayerRetainsReceivedValidatedAtAVPPOneSample(t *testing.T) {
// send voteVerified
verifiedVoteMsg := message{Vote: vVote, UnauthenticatedVote: vVote.u()}
inMsg = messageEvent{T: voteVerified, Input: verifiedVoteMsg, TaskIndex: 1}
inMsg = inMsg.AttachValidatedAt(502 * time.Millisecond)
inMsg = inMsg.AttachValidatedAt(502*time.Millisecond, r-1)
err, panicErr = pM.transition(inMsg)
require.NoError(t, err)
require.NoError(t, panicErr)
Expand All @@ -3446,7 +3446,7 @@ func TestPlayerRetainsReceivedValidatedAtAVPPOneSample(t *testing.T) {
proposalMsg := message{UnauthenticatedProposal: pP.u()}
compoundMsg := messageEvent{T: votePresent, Input: unverifiedVoteMsg,
Tail: &messageEvent{T: payloadPresent, Input: proposalMsg}}
inMsg = compoundMsg.AttachReceivedAt(time.Second) // call AttachReceivedAt like demux would
inMsg = compoundMsg.AttachReceivedAt(time.Second, r-1) // call AttachReceivedAt like demux would
err, panicErr = pM.transition(inMsg)
require.NoError(t, err)
require.NoError(t, panicErr)
Expand Down Expand Up @@ -3491,7 +3491,7 @@ func TestPlayerRetainsReceivedValidatedAtAVPPHistoryWindow(t *testing.T) {
verifiedVoteMsg := message{Vote: vVote, UnauthenticatedVote: vVote.u()}
inMsg = messageEvent{T: voteVerified, Input: verifiedVoteMsg, TaskIndex: 1}
timestamp := 500 + i
inMsg = inMsg.AttachValidatedAt(time.Duration(timestamp) * time.Millisecond)
inMsg = inMsg.AttachValidatedAt(time.Duration(timestamp)*time.Millisecond, r+round(i)-1)
err, panicErr = pM.transition(inMsg)
require.NoError(t, err)
require.NoError(t, panicErr)
Expand All @@ -3500,7 +3500,7 @@ func TestPlayerRetainsReceivedValidatedAtAVPPHistoryWindow(t *testing.T) {
proposalMsg := message{UnauthenticatedProposal: pP.u()}
compoundMsg := messageEvent{T: votePresent, Input: unverifiedVoteMsg,
Tail: &messageEvent{T: payloadPresent, Input: proposalMsg}}
inMsg = compoundMsg.AttachReceivedAt(time.Second) // call AttachReceivedAt like demux would
inMsg = compoundMsg.AttachReceivedAt(time.Second, r+round(i)-1) // call AttachReceivedAt like demux would
err, panicErr = pM.transition(inMsg)
require.NoError(t, err)
require.NoError(t, panicErr)
Expand All @@ -3526,7 +3526,7 @@ func moveToRound(t *testing.T, pWhite *player, pM ioAutomata, helper *voteMakerH

// payloadVerified
inMsg := messageEvent{T: payloadVerified, Input: message{Proposal: *pP}, Proto: ConsensusVersionView{Version: ver}}
inMsg = inMsg.AttachValidatedAt(2 * time.Second) // call AttachValidatedAt like demux would
inMsg = inMsg.AttachValidatedAt(2*time.Second, r-1) // call AttachValidatedAt like demux would
err, panicErr := pM.transition(inMsg)
require.NoError(t, err)
require.NoError(t, panicErr)
Expand Down