performance: Speculative block assembly into master

agreement/abstractions.go

@@ -85,6 +85,8 @@ type BlockFactory interface {
 	// nodes on the network can assemble entries, the agreement protocol may
 	// lose liveness.
 	AssembleBlock(basics.Round) (ValidatedBlock, error)
+
+	StartSpeculativeBlockAssembly(context.Context, ValidatedBlock, crypto.Digest, bool)
 }
 
 // A Ledger represents the sequence of Entries agreed upon by the protocol.

agreement/actions.go

@@ -54,6 +54,8 @@ const (
 	attest
 	assemble
 	repropose
+	speculativeAssembly
+	speculativeAssemblyIfStarted
 
 	// disk
 	checkpoint
@@ -309,13 +311,14 @@ func (a rezeroAction) do(ctx context.Context, s *Service) {
 }
 
 type pseudonodeAction struct {
-	// assemble, repropose, attest
+	// assemble, repropose, attest, speculativeAssembly, speculativeAssemblyIfStarted
 	T actionType
 
-	Round    round
-	Period   period
-	Step     step
-	Proposal proposalValue
+	Round          round
+	Period         period
+	Step           step
+	Proposal       proposalValue
+	ValidatedBlock ValidatedBlock
 }
 
 func (a pseudonodeAction) t() actionType {
@@ -346,6 +349,11 @@ func (a pseudonodeAction) do(ctx context.Context, s *Service) {
 		default:
 			s.log.Errorf("pseudonode.MakeProposals call failed %v", err)
 		}
+	case speculativeAssembly:
+		s.loopback.StartSpeculativeBlockAssembly(ctx, a.ValidatedBlock, a.Proposal.BlockDigest, false)
+	case speculativeAssemblyIfStarted:
+		s.loopback.StartSpeculativeBlockAssembly(ctx, a.ValidatedBlock, a.Proposal.BlockDigest, true)
+
 	case repropose:
 		logEvent := logspec.AgreementEvent{
 			Type:   logspec.VoteAttest,
@@ -464,7 +472,7 @@ func zeroAction(t actionType) action {
 		return ensureAction{}
 	case rezero:
 		return rezeroAction{}
-	case attest, assemble, repropose:
+	case attest, assemble, repropose, speculativeAssembly, speculativeAssemblyIfStarted:
 		return pseudonodeAction{}
 	case checkpoint:
 		return checkpointAction{}

agreement/agreementtest/simulate_test.go

@@ -98,6 +98,9 @@ func (f testBlockFactory) AssembleBlock(r basics.Round) (agreement.ValidatedBloc
 	return testValidatedBlock{Inside: bookkeeping.Block{BlockHeader: bookkeeping.BlockHeader{Round: r}}}, nil
 }
 
+func (f testBlockFactory) StartSpeculativeBlockAssembly(context.Context, agreement.ValidatedBlock, crypto.Digest, bool) {
+}
+
 // If we try to read from high rounds, we panic and do not emit an error to find bugs during testing.
 type testLedger struct {
 	mu deadlock.Mutex

agreement/common_test.go

@@ -184,6 +184,10 @@ func (f testBlockFactory) AssembleBlock(r basics.Round) (ValidatedBlock, error)
 	return testValidatedBlock{Inside: bookkeeping.Block{BlockHeader: bookkeeping.BlockHeader{Round: r}}}, nil
 }
 
+func (f testBlockFactory) StartSpeculativeBlockAssembly(context.Context, ValidatedBlock, crypto.Digest, bool) {
+	return
+}
+
 // If we try to read from high rounds, we panic and do not emit an error to find bugs during testing.
 type testLedger struct {
 	mu deadlock.Mutex

agreement/coservice.go

@@ -17,9 +17,10 @@
 package agreement
 
 import (
-	"github.com/algorand/go-deadlock"
+	"runtime"
+	"testing"
 
-	"github.com/algorand/go-algorand/logging"
+	"github.com/algorand/go-deadlock"
 )
 
 //go:generate stringer -type=coserviceType
@@ -35,10 +36,12 @@ const (
 //msgp:ignore coserviceType
 type coserviceType int
 
+// coserviceMonitor is unit test instrumentation
 type coserviceMonitor struct {
 	deadlock.Mutex
 
 	id int
+	t  *testing.T
 	c  map[coserviceType]uint
 
 	coserviceListener
@@ -79,7 +82,10 @@ func (m *coserviceMonitor) dec(t coserviceType) {
 		m.c = make(map[coserviceType]uint)
 	}
 	if m.c[t] == 0 {
-		logging.Base().Panicf("%d: tried to decrement empty coservice queue %v", m.id, t)
+		var stack [1000]byte
+		sl := runtime.Stack(stack[:], false)
+		m.t.Log(string(stack[:sl]))
+		m.t.Fatalf("%d: tried to decrement empty coservice queue %v", m.id, t)
 	}
 	m.c[t]--
 

agreement/demux.go

@@ -60,10 +60,13 @@ type demux struct {
 
 	queue             []<-chan externalEvent
 	processingMonitor EventsProcessingMonitor
-	monitor           *coserviceMonitor
 	cancelTokenizers  context.CancelFunc
 
 	log logging.Logger
+
+	// coserviceMonitor is unit test instrumentation.
+	// should be fast no-op if monitor == nil
+	monitor *coserviceMonitor
 }
 
 // demuxParams contains the parameters required to initliaze a new demux object
@@ -74,7 +77,10 @@ type demuxParams struct {
 	voteVerifier      *AsyncVoteVerifier
 	processingMonitor EventsProcessingMonitor
 	log               logging.Logger
-	monitor           *coserviceMonitor
+
+	// coserviceMonitor is unit test instrumentation.
+	// should be fast no-op if monitor == nil
+	monitor *coserviceMonitor
 }
 
 // makeDemux initializes the goroutines needed to process external events, setting up the appropriate channels.
@@ -190,7 +196,7 @@ func (d *demux) verifyBundle(ctx context.Context, m message, r round, p period,
 // next blocks until it observes an external input event of interest for the state machine.
 //
 // If ok is false, there are no more events so the agreement service should quit.
-func (d *demux) next(s *Service, deadline time.Duration, fastDeadline time.Duration, currentRound round) (e externalEvent, ok bool) {
+func (d *demux) next(s *Service, deadline time.Duration, fastDeadline time.Duration, speculationDeadline time.Duration, currentRound round) (e externalEvent, ok bool) {
 	defer func() {
 		if !ok {
 			return
@@ -253,6 +259,14 @@ func (d *demux) next(s *Service, deadline time.Duration, fastDeadline time.Durat
 	deadlineCh := s.Clock.TimeoutAt(deadline)
 	fastDeadlineCh := s.Clock.TimeoutAt(fastDeadline)
 
+	var speculationDeadlineCh <-chan time.Time
+	// zero timeout means we don't have enough time to speculate on block assembly
+	if speculationDeadline != 0 {
+		speculationDeadlineCh = s.Clock.TimeoutAt(speculationDeadline)
+	}
+
+	//d.log.Infof("demux deadline %d, fastD %d, specD %d, d.monitor %v", deadline, fastDeadline, speculationDeadline, d.monitor) // not threadsafe in some tests
+
 	d.UpdateEventsQueue(eventQueueDemux, 0)
 	d.monitor.dec(demuxCoserviceType)
 
@@ -269,7 +283,7 @@ func (d *demux) next(s *Service, deadline time.Duration, fastDeadline time.Durat
 		// the pseudonode channel got closed. remove it from the queue and try again.
 		d.queue = d.queue[1:]
 		d.UpdateEventsQueue(eventQueuePseudonode, 0)
-		return d.next(s, deadline, fastDeadline, currentRound)
+		return d.next(s, deadline, fastDeadline, speculationDeadline, currentRound)
 
 	// control
 	case <-s.quit:
@@ -303,6 +317,11 @@ func (d *demux) next(s *Service, deadline time.Duration, fastDeadline time.Durat
 		d.UpdateEventsQueue(eventQueueDemux, 1)
 		d.monitor.inc(demuxCoserviceType)
 		d.monitor.dec(clockCoserviceType)
+	case <-speculationDeadlineCh:
+		e = timeoutEvent{T: speculationTimeout, RandomEntropy: s.RandomSource.Uint64(), Round: nextRound}
+		d.UpdateEventsQueue(eventQueueDemux, 1)
+		d.monitor.inc(demuxCoserviceType)
+		d.monitor.dec(clockCoserviceType)
 
 	// raw
 	case m, open := <-rawVotes:
@@ -358,6 +377,7 @@ func (d *demux) next(s *Service, deadline time.Duration, fastDeadline time.Durat
 }
 
 // setupCompoundMessage processes compound messages: distinct messages which are delivered together
+// TODO: does this ever really see something other than empty .Vote?
 func setupCompoundMessage(l LedgerReader, m message) (res externalEvent) {
 	compound := m.CompoundMessage
 	if compound.Vote == (unauthenticatedVote{}) {

agreement/demux_test.go

@@ -37,6 +37,7 @@ import (
 )
 
 const fastTimeoutChTime = 2
+const speculativeBlockAsmTime = time.Duration(5 * time.Microsecond)
 
 type demuxTester struct {
 	*testing.T
@@ -65,8 +66,10 @@ type demuxTestUsecase struct {
 	verifiedProposal    testChanState
 	verifiedBundle      testChanState
 	// expected output
-	e  event
-	ok bool
+	e                  event
+	ok                 bool
+	speculativeAsmTime time.Duration
+	desc               string
 }
 
 var demuxTestUsecases = []demuxTestUsecase{
@@ -195,6 +198,7 @@ var demuxTestUsecases = []demuxTestUsecase{
 		verifiedBundle:      testChanState{eventCount: 0, closed: false},
 		e:                   messageEvent{T: votePresent},
 		ok:                  true,
+		desc:                "one vote one prop",
 	},
 	{
 		queue:               []testChanState{},
@@ -411,6 +415,26 @@ var demuxTestUsecases = []demuxTestUsecase{
 		verifiedBundle:      testChanState{eventCount: 0, closed: false},
 		e:                   messageEvent{T: votePresent},
 		ok:                  true,
+		desc:                "one prop",
+	},
+	{
+		queue:               []testChanState{{eventCount: 0, closed: false}},
+		rawVotes:            testChanState{eventCount: 0, closed: false},
+		rawProposals:        testChanState{eventCount: 0, closed: false},
+		rawBundles:          testChanState{eventCount: 0, closed: false},
+		compoundProposals:   false,
+		quit:                false,
+		voteChannelFull:     false,
+		proposalChannelFull: false,
+		bundleChannelFull:   false,
+		ledgerRoundReached:  false,
+		deadlineReached:     false,
+		verifiedVote:        testChanState{eventCount: 0, closed: false},
+		verifiedProposal:    testChanState{eventCount: 0, closed: false},
+		verifiedBundle:      testChanState{eventCount: 0, closed: false},
+		e:                   timeoutEvent{T: speculationTimeout},
+		ok:                  true,
+		speculativeAsmTime:  speculativeBlockAsmTime,
 	},
 }
 
@@ -432,6 +456,9 @@ func (t *demuxTester) TimeoutAt(delta time.Duration) <-chan time.Time {
 	if delta == fastTimeoutChTime {
 		return nil
 	}
+	if delta == speculativeBlockAsmTime {
+		return time.After(delta)
+	}
 
 	c := make(chan time.Time, 2)
 	if t.currentUsecase.deadlineReached {
@@ -653,6 +680,7 @@ func (t *demuxTester) TestUsecase(testcase demuxTestUsecase) bool {
 
 	dmx := &demux{}
 
+	dmx.log = logging.TestingLog(t)
 	dmx.crypto = t
 	dmx.ledger = t
 	dmx.rawVotes = t.makeRawChannel(protocol.AgreementVoteTag, testcase.rawVotes, false)
@@ -674,14 +702,13 @@ func (t *demuxTester) TestUsecase(testcase demuxTestUsecase) bool {
 	if testcase.quit {
 		close(s.quit)
 	}
-
-	e, ok := dmx.next(s, time.Second, fastTimeoutChTime, 300)
+	e, ok := dmx.next(s, time.Second, fastTimeoutChTime, testcase.speculativeAsmTime, 300)
 
 	if !assert.Equal(t, testcase.ok, ok) {
 		return false
 	}
 
-	if !assert.Equalf(t, strings.Replace(testcase.e.String(), "{test_index}", fmt.Sprintf("%d", t.testIdx), 1), e.String(), "Test case %d failed.", t.testIdx+1) {
+	if !assert.Equalf(t, strings.Replace(testcase.e.String(), "{test_index}", fmt.Sprintf("%d", t.testIdx), 1), e.String(), "Test case %d (%s) failed.", t.testIdx+1, testcase.desc) {
 		return false
 	}
 

agreement/events.go

@@ -122,8 +122,11 @@ const (
 	// period.
 	//
 	// fastTimeout is like timeout but for fast partition recovery.
+	// speculation timeout marks when the player should start speculative
+	// block assembly.
 	timeout
 	fastTimeout
+	speculationTimeout
 
 	// Other events are delivered from one state machine to another to
 	// communicate some message or as a reply to some message.  These events
@@ -197,6 +200,14 @@ const (
 	// readPinned is sent to the proposalStore to read the pinned value, if it exists.
 	readPinned
 
+	// readLowestValue is sent to the proposalPeriodMachine to read the
+	// proposal-vote with the lowest credential.
+	readLowestValue
+
+	// readLowestPayload is sent to the proposalStore to read the payload
+	// corresponding to the lowest-credential proposal-vote, if it exists.
+	readLowestPayload
+
 	/*
 	 * The following are event types that replace queries, and may warrant
 	 * a revision to make them more state-machine-esque.
@@ -363,7 +374,7 @@ func (e roundInterruptionEvent) AttachConsensusVersion(v ConsensusVersionView) e
 }
 
 type timeoutEvent struct {
-	// {timeout,fastTimeout}
+	// {timeout,fastTimeout,speculationTimeout}
 	T eventType
 
 	RandomEntropy uint64
@@ -407,6 +418,38 @@ func (e newRoundEvent) ComparableStr() string {
 	return e.String()
 }
 
+type readLowestEvent struct {
+	// T is either readLowestValue or readLowestPayload
+	T eventType
+
+	// Round and Period are the round and period for which to query
+	// the lowest-credential value and payload.  This type of event
+	// is only sent for pipelining, which only makes sense for period
+	// 0, but the Period is here anyway to route to the appropriate
+	// proposalMachinePeriod.
+	Round  round
+	Period period
+
+	// Proposal holds the lowest-credential value.
+	Proposal proposalValue
+	// Payload holds the payload, if one exists (which is the case if PayloadOK is set).
+	Payload proposal
+	// PayloadOK is set if and only if a payload was received for the lowest-credential value.
+	PayloadOK bool
+}
+
+func (e readLowestEvent) t() eventType {
+	return e.T
+}
+
+func (e readLowestEvent) String() string {
+	return fmt.Sprintf("%v: %.5v", e.t().String(), e.Proposal.BlockDigest.String())
+}
+
+func (e readLowestEvent) ComparableStr() string {
+	return e.String()
+}
+
 type newPeriodEvent struct {
 	// Period holds the latest period relevant to the proposalRoundMachine.
 	Period period
@@ -744,7 +787,7 @@ func zeroEvent(t eventType) event {
 		return messageEvent{}
 	case roundInterruption:
 		return roundInterruptionEvent{}
-	case timeout, fastTimeout:
+	case timeout, fastTimeout, speculationTimeout:
 		return timeoutEvent{}
 	case newRound:
 		return newRoundEvent{}