performance: agreement state serialization using msgp (#4644)

Author: iansuvak

agreement/actions.go

@@ -26,8 +26,7 @@ import (
 )
 
 //go:generate stringer -type=actionType
-//msgp:ignore actionType
-type actionType int
+type actionType uint8
 
 const (
 	noop actionType = iota
@@ -103,7 +102,7 @@ type networkAction struct {
 
 	UnauthenticatedVotes []unauthenticatedVote
 
-	Err serializableError
+	Err *serializableError
 }
 
 func (a networkAction) t() actionType {
@@ -181,7 +180,7 @@ type cryptoAction struct {
 	Period    period
 	Step      step
 	Pinned    bool
-	TaskIndex int
+	TaskIndex uint64
 }
 
 func (a cryptoAction) t() actionType {
@@ -388,7 +387,7 @@ func (a pseudonodeAction) do(ctx context.Context, s *Service) {
 		case nil:
 			// no error.
 			persistCompleteEvents := s.persistState(persistStateDone)
-			// we want to place there two one after the other. That way, the second would not get executed up until the first one is complete.
+			// we want to place these two one after the other. That way, the second would not get executed up until the first one is complete.
 			s.demux.prioritize(persistCompleteEvents)
 			s.demux.prioritize(voteEvents)
 		default:
@@ -403,12 +402,12 @@ func (a pseudonodeAction) do(ctx context.Context, s *Service) {
 	}
 }
 
-func ignoreAction(e messageEvent, err serializableError) action {
-	return networkAction{T: ignore, Err: err, h: e.Input.MessageHandle}
+func ignoreAction(e messageEvent, err *serializableError) action {
+	return networkAction{T: ignore, Err: err, h: e.Input.messageHandle}
 }
 
-func disconnectAction(e messageEvent, err serializableError) action {
-	return networkAction{T: disconnect, Err: err, h: e.Input.MessageHandle}
+func disconnectAction(e messageEvent, err *serializableError) action {
+	return networkAction{T: disconnect, Err: err, h: e.Input.messageHandle}
 }
 
 func broadcastAction(tag protocol.Tag, o interface{}) action {
@@ -427,7 +426,7 @@ func broadcastAction(tag protocol.Tag, o interface{}) action {
 }
 
 func relayAction(e messageEvent, tag protocol.Tag, o interface{}) action {
-	a := networkAction{T: relay, h: e.Input.MessageHandle, Tag: tag}
+	a := networkAction{T: relay, h: e.Input.messageHandle, Tag: tag}
 	// TODO would be good to have compiler check this (and related) type switch
 	// by specializing one method per type
 	switch tag {
@@ -441,7 +440,7 @@ func relayAction(e messageEvent, tag protocol.Tag, o interface{}) action {
 	return a
 }
 
-func verifyVoteAction(e messageEvent, r round, p period, taskIndex int) action {
+func verifyVoteAction(e messageEvent, r round, p period, taskIndex uint64) action {
 	return cryptoAction{T: verifyVote, M: e.Input, Round: r, Period: p, TaskIndex: taskIndex}
 }
 
@@ -479,7 +478,7 @@ type checkpointAction struct {
 	Round  round
 	Period period
 	Step   step
-	Err    serializableError
+	Err    *serializableError
 	done   chan error // an output channel to let the pseudonode that we're done processing. We don't want to serialize that, since it's not needed in recovery/autopsy
 }
 

agreement/actiontype_string.go

@@ -29,7 +29,7 @@ type asyncVerifyVoteRequest struct {
 	l       LedgerReader
 	uv      *unauthenticatedVote
 	uev     *unauthenticatedEquivocationVote
-	index   int
+	index   uint64
 	message message
 
 	// a channel that holds the response
@@ -39,7 +39,7 @@ type asyncVerifyVoteRequest struct {
 type asyncVerifyVoteResponse struct {
 	v         vote
 	ev        equivocationVote
-	index     int
+	index     uint64
 	message   message
 	err       error
 	cancelled bool
@@ -131,7 +131,7 @@ func (avv *AsyncVoteVerifier) executeEqVoteVerification(task interface{}) interf
 	}
 }
 
-func (avv *AsyncVoteVerifier) verifyVote(verctx context.Context, l LedgerReader, uv unauthenticatedVote, index int, message message, out chan<- asyncVerifyVoteResponse) error {
+func (avv *AsyncVoteVerifier) verifyVote(verctx context.Context, l LedgerReader, uv unauthenticatedVote, index uint64, message message, out chan<- asyncVerifyVoteResponse) error {
 	select {
 	case <-avv.ctx.Done(): // if we're quitting, don't enqueue the request
 	// case <-verctx.Done(): DO NOT DO THIS! otherwise we will lose the vote (and forget to clean up)!
@@ -151,7 +151,7 @@ func (avv *AsyncVoteVerifier) verifyVote(verctx context.Context, l LedgerReader,
 	return nil
 }
 
-func (avv *AsyncVoteVerifier) verifyEqVote(verctx context.Context, l LedgerReader, uev unauthenticatedEquivocationVote, index int, message message, out chan<- asyncVerifyVoteResponse) error {
+func (avv *AsyncVoteVerifier) verifyEqVote(verctx context.Context, l LedgerReader, uev unauthenticatedEquivocationVote, index uint64, message message, out chan<- asyncVerifyVoteResponse) error {
 	select {
 	case <-avv.ctx.Done(): // if we're quitting, don't enqueue the request
 	// case <-verctx.Done(): DO NOT DO THIS! otherwise we will lose the vote (and forget to clean up)!

agreement/asyncVoteVerifier.go

@@ -202,7 +202,8 @@ func (b unauthenticatedBundle) verifyAsync(ctx context.Context, l LedgerReader,
 
 		rv := rawVote{Sender: auth.Sender, Round: b.Round, Period: b.Period, Step: b.Step, Proposal: b.Proposal}
 		uv := unauthenticatedVote{R: rv, Cred: auth.Cred, Sig: auth.Sig}
-		avv.verifyVote(ctx, l, uv, i, message{}, results)
+
+		avv.verifyVote(ctx, l, uv, uint64(i), message{}, results) //nolint:errcheck // verifyVote will call EnqueueBacklog, which blocks until the verify task is queued, or returns an error when ctx.Done(), which we are already checking
 	}
 
 	// create verification requests for equivocation votes
@@ -222,7 +223,8 @@ func (b unauthenticatedBundle) verifyAsync(ctx context.Context, l LedgerReader,
 			Proposals: auth.Proposals,
 			Sigs:      auth.Sigs,
 		}
-		avv.verifyEqVote(ctx, l, uev, i, message{}, results)
+		avv.verifyEqVote(ctx, l, uev, uint64(i), message{}, results) //nolint:errcheck // verifyVote will call EnqueueBacklog, which blocks until the verify task is queued, or returns an error when ctx.Done(), which we are already checking
+
 	}
 
 	return func() (bundle, error) {

agreement/bundle.go

@@ -82,37 +82,41 @@ type (
 		Quit()
 	}
 
+	//msgp:ignore cryptoVoteRequest
 	cryptoVoteRequest struct {
 		message                   // the message we would like to verify.
-		TaskIndex int             // Caller specific number that would be passed back in the asyncVerifyVoteResponse.TaskIndex field
+		TaskIndex uint64          // Caller specific number that would be passed back in the asyncVerifyVoteResponse.TaskIndex field
 		Round     round           // The round that we're going to test against.
 		Period    period          // The period associated with the message we're going to test.
 		ctx       context.Context // A context for this request, if the context is cancelled then the request is stale.
 	}
 
+	//msgp:ignore cryptoProposalRequest
 	cryptoProposalRequest struct {
 		message                   // the message we would like to verify.
-		TaskIndex int             // Caller specific number that would be passed back in the cryptoResult.TaskIndex field
+		TaskIndex uint64          // Caller specific number that would be passed back in the cryptoResult.TaskIndex field
 		Round     round           // The round that we're going to test against.
 		Period    period          // The period associated with the message we're going to test.
 		Pinned    bool            // A flag that is set if this is a pinned value for the given round.
 		ctx       context.Context // A context for this request, if the context is cancelled then the request is stale.
 	}
 
+	//msgp:ignore cryptoBundleRequest
 	cryptoBundleRequest struct {
 		message                   // the message we would like to verify.
-		TaskIndex int             // Caller specific number that would be passed back in the asyncVerifyVoteResponse.TaskIndex field
+		TaskIndex uint64          // Caller specific number that would be passed back in the asyncVerifyVoteResponse.TaskIndex field
 		Round     round           // The round that we're going to test against.
 		Period    period          // The period associated with the message we're going to test.
 		Certify   bool            // A flag that set if this is a cert bundle.
 		ctx       context.Context // A context for this request, if the context is cancelled then the request is stale.
 	}
 
+	//msgp:ignore cryptoResult
 	cryptoResult struct {
 		message
-		Err       serializableError
-		TaskIndex int  // the TaskIndex that was passed to the cryptoVerifier during the Verify call on the cryptoRequest.TaskIndex
-		Cancelled bool // whether the corresponding request was cancelled before verification completed
+		Err       *serializableError
+		TaskIndex uint64 // the TaskIndex that was passed to the cryptoVerifier during the Verify call on the cryptoRequest.TaskIndex
+		Cancelled bool   // whether the corresponding request was cancelled before verification completed
 	}
 
 	// A poolCryptoVerifier uses asynchronous goroutines to implement cryptoVerifier.
@@ -146,9 +150,10 @@ type (
 		out chan cryptoResult
 	}
 
+	//msgp:ignore bundleFuture
 	bundleFuture struct {
 		message
-		index int
+		index uint64
 		wait  func() (bundle, error)
 		ctx   context.Context
 	}

agreement/cryptoVerifier.go

@@ -93,7 +93,7 @@ func makeMessage(msgHandle int, tag protocol.Tag, sender basics.Address, l Ledge
 		}
 
 		return message{
-			MessageHandle:       MessageHandle(msgHandle),
+			messageHandle:       MessageHandle(msgHandle),
 			Tag:                 tag,
 			UnauthenticatedVote: makeUnauthenticatedVote(l, sender, selection, voting, Round, Period, Step, proposal),
 		}
@@ -103,13 +103,13 @@ func makeMessage(msgHandle int, tag protocol.Tag, sender basics.Address, l Ledge
 			Block: e,
 		}
 		return message{
-			MessageHandle:           MessageHandle(msgHandle),
+			messageHandle:           MessageHandle(msgHandle),
 			Tag:                     tag,
 			UnauthenticatedProposal: payload,
 		}
 	default: // protocol.VoteBundleTag
 		return message{
-			MessageHandle: MessageHandle(msgHandle),
+			messageHandle: MessageHandle(msgHandle),
 			Tag:           tag,
 			UnauthenticatedBundle: unauthenticatedBundle{
 				Round:    Round,
@@ -180,9 +180,9 @@ func TestCryptoVerifierBuffers(t *testing.T) {
 	for _, msgType := range msgTypes {
 		for i := getSelectorCapacity(msgType) * 5; i > 0; i-- {
 			msg := <-verifier.Verified(msgType)
-			_, has := usedMsgIDs[msg.MessageHandle]
+			_, has := usedMsgIDs[msg.messageHandle]
 			assert.True(t, has)
-			delete(usedMsgIDs, msg.MessageHandle)
+			delete(usedMsgIDs, msg.messageHandle)
 		}
 		assert.False(t, verifier.ChannelFull(msgType))
 		assert.Zero(t, len(verifier.Verified(msgType)))
@@ -230,8 +230,8 @@ func TestCryptoVerifierBuffers(t *testing.T) {
 		}
 		msgIDMutex.Lock()
 		defer msgIDMutex.Unlock()
-		_, has := usedMsgIDs[msg.MessageHandle]
-		delete(usedMsgIDs, msg.MessageHandle)
+		_, has := usedMsgIDs[msg.messageHandle]
+		delete(usedMsgIDs, msg.messageHandle)
 		return assert.True(t, has)
 	}
 
@@ -333,7 +333,7 @@ func BenchmarkCryptoVerifierProposalVertification(b *testing.B) {
 	c := verifier.Verified(protocol.ProposalPayloadTag)
 	request := cryptoProposalRequest{
 		message: message{
-			MessageHandle:           MessageHandle(0),
+			messageHandle:           MessageHandle(0),
 			Tag:                     protocol.ProposalPayloadTag,
 			UnauthenticatedProposal: proposals[0].unauthenticatedProposal,
 		},
@@ -402,11 +402,11 @@ func TestCryptoVerifierVerificationFailures(t *testing.T) {
 	cryptoVerifier := makeCryptoVerifier(nil, nil, voteVerifier, logging.TestingLog(t))
 	defer cryptoVerifier.Quit()
 
-	cryptoVerifier.VerifyVote(context.Background(), cryptoVoteRequest{message: message{Tag: protocol.AgreementVoteTag}, Round: basics.Round(8), TaskIndex: 14})
+	cryptoVerifier.VerifyVote(context.Background(), cryptoVoteRequest{message: message{Tag: protocol.AgreementVoteTag}, Round: basics.Round(8), TaskIndex: uint64(14)})
 	// read the failed response from VerifiedVotes:
 	votesout := cryptoVerifier.VerifiedVotes()
 	voteResponse := <-votesout
 	require.Equal(t, context.Canceled, voteResponse.err)
 	require.True(t, voteResponse.cancelled)
-	require.Equal(t, 14, voteResponse.index)
+	require.Equal(t, uint64(14), voteResponse.index)
 }

agreement/cryptoVerifier_test.go

@@ -140,11 +140,11 @@ func (d *demux) tokenizeMessages(ctx context.Context, net Network, tag protocol.
 				var msg message
 				switch tag {
 				case protocol.AgreementVoteTag:
-					msg = message{MessageHandle: raw.MessageHandle, Tag: tag, UnauthenticatedVote: o.(unauthenticatedVote)}
+					msg = message{messageHandle: raw.MessageHandle, Tag: tag, UnauthenticatedVote: o.(unauthenticatedVote)}
 				case protocol.VoteBundleTag:
-					msg = message{MessageHandle: raw.MessageHandle, Tag: tag, UnauthenticatedBundle: o.(unauthenticatedBundle)}
+					msg = message{messageHandle: raw.MessageHandle, Tag: tag, UnauthenticatedBundle: o.(unauthenticatedBundle)}
 				case protocol.ProposalPayloadTag:
-					msg = message{MessageHandle: raw.MessageHandle, Tag: tag, CompoundMessage: o.(compoundMessage)}
+					msg = message{messageHandle: raw.MessageHandle, Tag: tag, CompoundMessage: o.(compoundMessage)}
 				default:
 					err := fmt.Errorf("bad message tag: %v", tag)
 					d.UpdateEventsQueue(fmt.Sprintf("Tokenizing-%s", tag), 0)
@@ -167,7 +167,7 @@ func (d *demux) tokenizeMessages(ctx context.Context, net Network, tag protocol.
 }
 
 // verifyVote enqueues a vote message to be verified.
-func (d *demux) verifyVote(ctx context.Context, m message, taskIndex int, r round, p period) {
+func (d *demux) verifyVote(ctx context.Context, m message, taskIndex uint64, r round, p period) {
 	d.UpdateEventsQueue(eventQueueCryptoVerifierVote, 1)
 	d.monitor.inc(cryptoVerifierCoserviceType)
 	d.crypto.VerifyVote(ctx, cryptoVoteRequest{message: m, TaskIndex: taskIndex, Round: r, Period: p})
@@ -367,7 +367,7 @@ func setupCompoundMessage(l LedgerReader, m message) (res externalEvent) {
 		return
 	}
 
-	tailmsg := message{MessageHandle: m.MessageHandle, Tag: protocol.ProposalPayloadTag, UnauthenticatedProposal: compound.Proposal}
+	tailmsg := message{messageHandle: m.messageHandle, Tag: protocol.ProposalPayloadTag, UnauthenticatedProposal: compound.Proposal}
 	synthetic := messageEvent{T: payloadPresent, Input: tailmsg}
 	proto, err := l.ConsensusVersion(ParamsRound(synthetic.ConsensusRound()))
 	synthetic = synthetic.AttachConsensusVersion(ConsensusVersionView{Err: makeSerErr(err), Version: proto}).(messageEvent)

agreement/demux.go

@@ -16,38 +16,39 @@
 
 package agreement
 
-import "fmt"
+import (
+	"fmt"
+)
 
 // serializableError, or state machine error, is a serializable error that
 // is correctly written to cadaver files.
-type serializableErrorUnderlying string
-type serializableError = *serializableErrorUnderlying
+type serializableError string
 
 // implement error interface
-func (e serializableErrorUnderlying) Error() string {
+func (e serializableError) Error() string {
 	return string(e)
 }
 
-func (e serializableErrorUnderlying) String() string {
+func (e serializableError) String() string {
 	return e.Error()
 }
 
 // makeSerErrStr returns an serializableError that formats as the given text.
-func makeSerErrStr(text string) serializableError {
-	s := serializableErrorUnderlying(text)
+func makeSerErrStr(text string) *serializableError {
+	s := serializableError(text)
 	return &s
 }
 
-func makeSerErrf(format string, a ...interface{}) serializableError {
-	s := serializableErrorUnderlying(fmt.Sprintf(format, a...))
+func makeSerErrf(format string, a ...interface{}) *serializableError {
+	s := serializableError(fmt.Sprintf(format, a...))
 	return &s
 }
 
 // makeSerErr returns an serializableError that formats as the given error.
-func makeSerErr(err error) serializableError {
+func makeSerErr(err error) *serializableError {
 	if err == nil {
 		return nil
 	}
-	s := serializableErrorUnderlying(err.Error())
+	s := serializableError(err.Error())
 	return &s
 }