performance: dynamic lambda

agreement/actions.go

@@ -19,6 +19,7 @@ package agreement
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/algorand/go-algorand/logging/logspec"
 	"github.com/algorand/go-algorand/logging/telemetryspec"
@@ -209,6 +210,11 @@ type ensureAction struct {
 	Payload proposal
 	// the certificate proving commitment
 	Certificate Certificate
+
+	// The time that the winning proposal-vote was validated, relative to the beginning of the round
+	voteValidatedAt time.Duration
+	// The dynamic filter timeout calculated for this round, even if not enabled, for reporting to telemetry.
+	dynamicFilterTimeout time.Duration
 }
 
 func (a ensureAction) t() actionType {
@@ -231,29 +237,33 @@ func (a ensureAction) do(ctx context.Context, s *Service) {
 		logEvent.Type = logspec.RoundConcluded
 		s.log.with(logEvent).Infof("committed round %d with pre-validated block %v", a.Certificate.Round, a.Certificate.Proposal)
 		s.log.EventWithDetails(telemetryspec.Agreement, telemetryspec.BlockAcceptedEvent, telemetryspec.BlockAcceptedEventDetails{
-			Address:      a.Certificate.Proposal.OriginalProposer.String(),
-			Hash:         a.Certificate.Proposal.BlockDigest.String(),
-			Round:        uint64(a.Certificate.Round),
-			ValidatedAt:  a.Payload.validatedAt,
-			ReceivedAt:   a.Payload.receivedAt,
-			PreValidated: true,
-			PropBufLen:   uint64(len(s.demux.rawProposals)),
-			VoteBufLen:   uint64(len(s.demux.rawVotes)),
+			Address:              a.Certificate.Proposal.OriginalProposer.String(),
+			Hash:                 a.Certificate.Proposal.BlockDigest.String(),
+			Round:                uint64(a.Certificate.Round),
+			ValidatedAt:          a.Payload.validatedAt,
+			ReceivedAt:           a.Payload.receivedAt,
+			VoteValidatedAt:      a.voteValidatedAt,
+			DynamicFilterTimeout: a.dynamicFilterTimeout,
+			PreValidated:         true,
+			PropBufLen:           uint64(len(s.demux.rawProposals)),
+			VoteBufLen:           uint64(len(s.demux.rawVotes)),
 		})
 		s.Ledger.EnsureValidatedBlock(a.Payload.ve, a.Certificate)
 	} else {
 		block := a.Payload.Block
 		logEvent.Type = logspec.RoundConcluded
 		s.log.with(logEvent).Infof("committed round %d with block %v", a.Certificate.Round, a.Certificate.Proposal)
 		s.log.EventWithDetails(telemetryspec.Agreement, telemetryspec.BlockAcceptedEvent, telemetryspec.BlockAcceptedEventDetails{
-			Address:      a.Certificate.Proposal.OriginalProposer.String(),
-			Hash:         a.Certificate.Proposal.BlockDigest.String(),
-			Round:        uint64(a.Certificate.Round),
-			ValidatedAt:  a.Payload.validatedAt,
-			ReceivedAt:   a.Payload.receivedAt,
-			PreValidated: false,
-			PropBufLen:   uint64(len(s.demux.rawProposals)),
-			VoteBufLen:   uint64(len(s.demux.rawVotes)),
+			Address:              a.Certificate.Proposal.OriginalProposer.String(),
+			Hash:                 a.Certificate.Proposal.BlockDigest.String(),
+			Round:                uint64(a.Certificate.Round),
+			ValidatedAt:          a.Payload.validatedAt,
+			ReceivedAt:           a.Payload.receivedAt,
+			VoteValidatedAt:      a.voteValidatedAt,
+			DynamicFilterTimeout: a.dynamicFilterTimeout,
+			PreValidated:         false,
+			PropBufLen:           uint64(len(s.demux.rawProposals)),
+			VoteBufLen:           uint64(len(s.demux.rawVotes)),
 		})
 		s.Ledger.EnsureBlock(block, a.Certificate)
 	}

agreement/credentialArrivalHistory.go

@@ -0,0 +1,83 @@
+// Copyright (C) 2019-2023 Algorand, Inc.
+// This file is part of go-algorand
+//
+// go-algorand is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as
+// published by the Free Software Foundation, either version 3 of the
+// License, or (at your option) any later version.
+//
+// go-algorand is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with go-algorand.  If not, see <https://www.gnu.org/licenses/>.
+
+package agreement
+
+import (
+	"sort"
+	"time"
+)
+
+// credentialArrivalHistory maintains a circular buffer of time.Duration samples.
+type credentialArrivalHistory struct {
+	history  []time.Duration
+	writePtr int
+	full     bool
+}
+
+func makeCredentialArrivalHistory(size int) credentialArrivalHistory {
+	if size < 0 {
+		panic("can't create CredentialArrivalHistory with negative size")
+	}
+	history := credentialArrivalHistory{history: make([]time.Duration, size)}
+	history.reset()
+	return history
+}
+
+// store saves a new sample into the circular buffer.
+// If the buffer is full, it overwrites the oldest sample.
+func (history *credentialArrivalHistory) store(sample time.Duration) {
+	if len(history.history) == 0 {
+		return
+	}
+
+	history.history[history.writePtr] = sample
+	history.writePtr++
+	if history.writePtr == len(history.history) {
+		history.full = true
+		history.writePtr = 0
+	}
+}
+
+// reset marks the history buffer as empty
+func (history *credentialArrivalHistory) reset() {
+	history.writePtr = 0
+	history.full = false
+}
+
+// isFull checks if the circular buffer has been fully populated at least once.
+func (history *credentialArrivalHistory) isFull() bool {
+	return history.full
+}
+
+// orderStatistics returns the idx'th time duration in the sorted history array.
+// It assumes that history is full and the idx is within the array bounds, and
+// panics if either of these assumptions doesn't hold.
+func (history *credentialArrivalHistory) orderStatistics(idx int) time.Duration {
+	if !history.isFull() {
+		panic("history not full")
+	}
+	if idx < 0 || idx >= len(history.history) {
+		panic("index out of bounds")
+	}
+
+	// if history.history is long, then we could optimize this function to use
+	// the linear time order statistics algorithm.
+	sortedArrivals := make([]time.Duration, len(history.history))
+	copy(sortedArrivals[:], history.history[:])
+	sort.Slice(sortedArrivals, func(i, j int) bool { return sortedArrivals[i] < sortedArrivals[j] })
+	return sortedArrivals[idx]
+}

agreement/credentialArrivalHistory_test.go

@@ -0,0 +1,129 @@
+// Copyright (C) 2019-2023 Algorand, Inc.
+// This file is part of go-algorand
+//
+// go-algorand is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as
+// published by the Free Software Foundation, either version 3 of the
+// License, or (at your option) any later version.
+//
+// go-algorand is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with go-algorand.  If not, see <https://www.gnu.org/licenses/>.
+
+package agreement
+
+import (
+	"testing"
+	"time"
+
+	"github.com/algorand/go-algorand/test/partitiontest"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCredentialHistoryStore(t *testing.T) {
+	partitiontest.PartitionTest(t)
+
+	size := 5
+	buffer := makeCredentialArrivalHistory(size)
+	// last store call overwrites the first one
+	for i := 0; i < size+1; i++ {
+		buffer.store(time.Duration(i))
+	}
+
+	require.True(t, buffer.isFull())
+	require.Equal(t, time.Duration(size), buffer.history[0])
+	for i := 1; i < size; i++ {
+		require.Equal(t, time.Duration(i), buffer.history[i])
+	}
+}
+
+func TestCredentialHistoryReset(t *testing.T) {
+	partitiontest.PartitionTest(t)
+
+	size := 5
+	buffer := makeCredentialArrivalHistory(size)
+	// last store call overwrites the first one
+	for i := 0; i < size+1; i++ {
+		buffer.store(time.Duration(i))
+	}
+
+	require.Equal(t, time.Duration(size), buffer.history[0])
+	for i := 1; i < size; i++ {
+		require.Equal(t, time.Duration(i), buffer.history[i])
+	}
+	require.True(t, buffer.isFull())
+	buffer.reset()
+	require.False(t, buffer.isFull())
+	buffer.store(time.Duration(100))
+	require.Equal(t, time.Duration(100), buffer.history[0])
+}
+
+func TestCredentialHistoryIsFull(t *testing.T) {
+	partitiontest.PartitionTest(t)
+	var buffer credentialArrivalHistory
+	require.False(t, buffer.isFull())
+
+	size := 5
+	buffer = makeCredentialArrivalHistory(size)
+	require.False(t, buffer.isFull())
+
+	for i := 1; i < size+10; i++ {
+		buffer.store(time.Duration(i))
+		if i < size {
+			require.False(t, buffer.isFull())
+		} else {
+			require.True(t, buffer.isFull())
+		}
+	}
+
+	// reset the buffer and then fill it again
+	buffer.reset()
+	require.False(t, buffer.isFull())
+
+	for i := 1; i < size+10; i++ {
+		buffer.store(time.Duration(i))
+		if i < size {
+			require.False(t, buffer.isFull())
+		} else {
+			require.True(t, buffer.isFull())
+		}
+	}
+}
+
+func TestCredentialHisotyZeroSize(t *testing.T) {
+	partitiontest.PartitionTest(t)
+
+	var buffer credentialArrivalHistory
+	require.False(t, buffer.isFull())
+
+	size := 0
+	buffer = makeCredentialArrivalHistory(size)
+	require.False(t, buffer.isFull())
+
+	// trying to store new samples won't panic but the history is never full
+	for i := 0; i < size+10; i++ {
+		buffer.store(time.Duration(i))
+		require.False(t, buffer.isFull())
+	}
+}
+
+func TestOrderStatistics(t *testing.T) {
+	partitiontest.PartitionTest(t)
+
+	size := 5
+	buffer := makeCredentialArrivalHistory(size)
+	require.False(t, buffer.isFull())
+
+	for i := 0; i < size; i++ {
+		buffer.store(time.Duration(size - i))
+	}
+	require.True(t, buffer.isFull())
+
+	for i := 0; i < size; i++ {
+		require.Equal(t, time.Duration(i+1), buffer.orderStatistics(i))
+	}
+}

agreement/demux.go

@@ -199,9 +199,14 @@ 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.Vote.R.Step == 0 {
+				e = e.(messageEvent).AttachValidatedAt(s.Clock.Since(), currentRound)
+			}
 		}
 	}()
 

agreement/dynamicFilterTimeoutParams.go

@@ -0,0 +1,45 @@
+// Copyright (C) 2019-2023 Algorand, Inc.
+// This file is part of go-algorand
+//
+// go-algorand is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as
+// published by the Free Software Foundation, either version 3 of the
+// License, or (at your option) any later version.
+//
+// go-algorand is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with go-algorand.  If not, see <https://www.gnu.org/licenses/>.
+
+package agreement
+
+import "time"
+
+// This file contains parameters for the dynamic filter timeout mechanism. When
+// this feature is enabled (dynamicFilterTimeout is true), these parameters
+// should migrate to be consensus params.
+
+// DynamicFilterCredentialArrivalHistory specifies the number of past
+// credential arrivals that are measured to determine the next filter
+// timeout. If DynamicFilterCredentialArrivalHistory <= 0, then the dynamic
+// timeout feature is off and the filter step timeout is calculated using
+// the static configuration.
+const dynamicFilterCredentialArrivalHistory int = 40
+
+// DynamicFilterTimeoutLowerBound specifies a minimal duration that the
+// filter timeout must meet.
+const dynamicFilterTimeoutLowerBound time.Duration = 500 * time.Millisecond
+
+// DynamicFilterTimeoutCredentialArrivalHistoryIdx specified which sample to use
+// out of a sorted DynamicFilterCredentialArrivalHistory-sized array of time
+// samples. The 95th percentile of dynamicFilterCredentialArrivalHistory = 40
+// sorted samples, is at index 37.
+const dynamicFilterTimeoutCredentialArrivalHistoryIdx int = 37
+
+// DynamicFilterTimeoutGraceInterval is additional extension to the dynamic
+// filter time atop the one calculated based on the history of credential
+// arrivals.
+const dynamicFilterTimeoutGraceInterval time.Duration = 50 * time.Millisecond

agreement/dynamicFilterTimeoutParams_test.go

@@ -0,0 +1,41 @@
+// Copyright (C) 2019-2023 Algorand, Inc.
+// This file is part of go-algorand
+//
+// go-algorand is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as
+// published by the Free Software Foundation, either version 3 of the
+// License, or (at your option) any later version.
+//
+// go-algorand is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with go-algorand.  If not, see <https://www.gnu.org/licenses/>.
+
+package agreement
+
+import (
+	"testing"
+	"time"
+
+	"github.com/algorand/go-algorand/test/partitiontest"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSampleIndexIsValid(t *testing.T) {
+	partitiontest.PartitionTest(t)
+
+	require.GreaterOrEqual(t, dynamicFilterCredentialArrivalHistory, 0)
+	require.GreaterOrEqual(t, dynamicFilterTimeoutCredentialArrivalHistoryIdx, 0)
+	if dynamicFilterCredentialArrivalHistory > 0 {
+		require.Less(t, dynamicFilterTimeoutCredentialArrivalHistoryIdx, dynamicFilterCredentialArrivalHistory)
+	}
+}
+
+func TestLowerBound(t *testing.T) {
+	partitiontest.PartitionTest(t)
+
+	require.Less(t, 20*time.Millisecond, dynamicFilterTimeoutLowerBound)
+}