Implement early termination case for avalanche polling

snow/engine/avalanche/polls.go

@@ -32,9 +32,19 @@ import (
 type polls struct {
 	log      logging.Logger
 	numPolls prometheus.Gauge
+	alpha    int
 	m        map[uint32]poll
 }
 
+func newPolls(alpha int, log logging.Logger, numPolls prometheus.Gauge) polls {
+	return polls{
+		log:      log,
+		numPolls: numPolls,
+		alpha:    alpha,
+		m:        make(map[uint32]poll),
+	}
+}
+
 // Add to the current set of polls
 // Returns true if the poll was registered correctly and the network sample
 //         should be made.
@@ -42,6 +52,7 @@ func (p *polls) Add(requestID uint32, vdrs ids.ShortSet) bool {
 	poll, exists := p.m[requestID]
 	if !exists {
 		poll.polled = vdrs
+		poll.alpha = p.alpha
 		p.m[requestID] = poll
 
 		p.numPolls.Set(float64(len(p.m))) // Tracks performance statistics
@@ -85,6 +96,7 @@ func (p *polls) String() string {
 type poll struct {
 	votes  ids.UniqueBag
 	polled ids.ShortSet
+	alpha  int
 }
 
 // Vote registers a vote for this poll
@@ -97,5 +109,32 @@ func (p *poll) Vote(votes []ids.ID, vdr ids.ShortID) {
 
 // Finished returns true if the poll has completed, with no more required
 // responses
-func (p poll) Finished() bool { return p.polled.Len() == 0 }
+func (p poll) Finished() bool {
+	// If I have no outstanding polls, I'm finished
+	numPending := p.polled.Len()
+	if numPending == 0 {
+		return true
+	}
+	// If there are still enough pending responses to include another vertex,
+	// then I can't stop polling
+	if numPending > p.alpha {
+		return false
+	}
+
+	// I ignore any vertex that has already received alpha votes.
+	// To safely skip DAG traversal, assume that all votes for
+	// vertices with less than alpha votes will be applied to a
+	// single shared ancestor.
+	// In this case, I can terminate early, iff there are not enough
+	// pending votes for this ancestor to receive alpha votes.
+	partialVotes := ids.BitSet(0)
+	for _, vote := range p.votes.List() {
+		voters := p.votes.GetSet(vote)
+		if voters.Len() >= p.alpha {
+			continue
+		}
+		partialVotes.Union(voters)
+	}
+	return partialVotes.Len()+numPending < p.alpha
+}
 func (p poll) String() string { return fmt.Sprintf("Waiting on %d chits", p.polled.Len()) }

snow/engine/avalanche/polls_test.go

@@ -0,0 +1,99 @@
+package avalanche
+
+import (
+	"testing"
+
+	"github.com/ava-labs/gecko/ids"
+)
+
+func TestPollTerminatesEarlyVirtuousCase(t *testing.T) {
+	alpha := 3
+
+	vtxID := GenerateID()
+	votes := []ids.ID{vtxID}
+
+	vdr1 := ids.NewShortID([20]byte{1})
+	vdr2 := ids.NewShortID([20]byte{2})
+	vdr3 := ids.NewShortID([20]byte{3})
+	vdr4 := ids.NewShortID([20]byte{4})
+	vdr5 := ids.NewShortID([20]byte{5}) // k = 5
+
+	vdrs := ids.ShortSet{}
+	vdrs.Add(vdr1)
+	vdrs.Add(vdr2)
+	vdrs.Add(vdr3)
+	vdrs.Add(vdr4)
+	vdrs.Add(vdr5)
+
+	poll := poll{
+		votes:  make(ids.UniqueBag),
+		polled: vdrs,
+		alpha:  alpha,
+	}
+
+	poll.Vote(votes, vdr1)
+	if poll.Finished() {
+		t.Fatalf("Poll finished after less than alpha votes")
+	}
+	poll.Vote(votes, vdr2)
+	if poll.Finished() {
+		t.Fatalf("Poll finished after less than alpha votes")
+	}
+	poll.Vote(votes, vdr3)
+	if !poll.Finished() {
+		t.Fatalf("Poll did not terminate early after receiving alpha votes for one vertex and none for other vertices")
+	}
+}
+
+func TestPollAccountsForSharedAncestor(t *testing.T) {
+	alpha := 4
+
+	vtxA := GenerateID()
+	vtxB := GenerateID()
+	vtxC := GenerateID()
+	vtxD := GenerateID()
+
+	// If validators 1-3 vote for frontier vertices
+	// B, C, and D respectively, which all share the common ancestor
+	// A, then we cannot terminate early with alpha = k = 4
+	// If the final vote is cast for any of A, B, C, or D, then
+	// vertex A will have transitively received alpha = 4 votes
+	vdr1 := ids.NewShortID([20]byte{1})
+	vdr2 := ids.NewShortID([20]byte{2})
+	vdr3 := ids.NewShortID([20]byte{3})
+	vdr4 := ids.NewShortID([20]byte{4})
+
+	vdrs := ids.ShortSet{}
+	vdrs.Add(vdr1)
+	vdrs.Add(vdr2)
+	vdrs.Add(vdr3)
+	vdrs.Add(vdr4)
+
+	poll := poll{
+		votes:  make(ids.UniqueBag),
+		polled: vdrs,
+		alpha:  alpha,
+	}
+
+	votes1 := []ids.ID{vtxB}
+	poll.Vote(votes1, vdr1)
+	if poll.Finished() {
+		t.Fatalf("Poll finished early after receiving one vote")
+	}
+	votes2 := []ids.ID{vtxC}
+	poll.Vote(votes2, vdr2)
+	if poll.Finished() {
+		t.Fatalf("Poll finished early after receiving two votes")
+	}
+	votes3 := []ids.ID{vtxD}
+	poll.Vote(votes3, vdr3)
+	if poll.Finished() {
+		t.Fatalf("Poll terminated early, when a shared ancestor could have received alpha votes")
+	}
+
+	votes4 := []ids.ID{vtxA}
+	poll.Vote(votes4, vdr4)
+	if !poll.Finished() {
+		t.Fatalf("Poll did not terminate after receiving all outstanding votes")
+	}
+}

snow/engine/avalanche/transitive.go

@@ -57,9 +57,7 @@ func (t *Transitive) Initialize(config Config) error {
 
 	t.onFinished = t.finishBootstrapping
 
-	t.polls.log = config.Context.Log
-	t.polls.numPolls = t.numPolls
-	t.polls.m = make(map[uint32]poll)
+	t.polls = newPolls(int(config.Alpha), config.Context.Log, t.numPolls)
 
 	return t.bootstrapper.Initialize(config.BootstrapConfig)
 }