code cleanup, turn off recheck

Author: philipsu522

config/config.go

@@ -82,7 +82,7 @@ func DefaultConfig() *Config {
 		P2P:             DefaultP2PConfig(),
 		Mempool:         DefaultMempoolConfig(),
 		StateSync:       DefaultStateSyncConfig(),
-		Consensus:       DefaultConsensusConfig(),
+		Consensus:       DefaultConsensusConfig().AB,
 		TxIndex:         DefaultTxIndexConfig(),
 		Instrumentation: DefaultInstrumentationConfig(),
 		PrivValidator:   DefaultPrivValidatorConfig(),

internal/consensus/metrics.gen.go

@@ -69,9 +69,6 @@ type Metrics struct {
 	// Number of block parts transmitted by each peer.
 	BlockParts metrics.Counter `metrics_labels:"peer_id"`
 
-	// Number of tx requests transmitted by each peer.
-	TxsSent metrics.Counter `metrics_labels:"peer_id"`
-
 	// Histogram of durations for each step in the consensus protocol.
 	StepDuration metrics.Histogram `metrics_labels:"step" metrics_buckettype:"exprange" metrics_bucketsizes:"0.1, 100, 8"`
 	stepStart    time.Time

internal/consensus/metrics.go

@@ -32,7 +32,6 @@ func init() {
 	jsontypes.MustRegister(&HasVoteMessage{})
 	jsontypes.MustRegister(&VoteSetMaj23Message{})
 	jsontypes.MustRegister(&VoteSetBitsMessage{})
-	jsontypes.MustRegister(&TxRequestMessage{})
 }
 
 // NewRoundStepMessage is sent for every step taken in the ConsensusState.
@@ -186,25 +185,6 @@ func (m *ProposalPOLMessage) String() string {
 	return fmt.Sprintf("[ProposalPOL H:%v POLR:%v POL:%v]", m.Height, m.ProposalPOLRound, m.ProposalPOL)
 }
 
-// TxRequestMessage is sent when a set of Txs are requested
-type TxRequestMessage struct {
-	Height int64 `json:",string"`
-	Round  int32
-	TxKeys []*types.TxKey
-}
-
-func (*TxRequestMessage) TypeTag() string { return "tendermint/TxRequest" }
-
-func (m *TxRequestMessage) ValidateBasic() error {
-	if m.Height < 0 {
-		return errors.New("negative Height")
-	}
-	if m.Round < 0 {
-		return errors.New("negative Round")
-	}
-	return nil
-}
-
 // BlockPartMessage is sent when gossipping a piece of the proposed block.
 type BlockPartMessage struct {
 	Height int64 `json:",string"`
@@ -476,22 +456,6 @@ func MsgToProto(msg Message) (*tmcons.Message, error) {
 			Sum: vsb,
 		}
 
-	case *TxRequestMessage:
-		var txKeys []*tmproto.TxKey
-		for _, txKey := range msg.TxKeys {
-			key := txKey.ToProto()
-			txKeys = append(txKeys, key)
-		}
-		pb = tmcons.Message{
-			Sum: &tmcons.Message_TxRequest{
-				TxRequest: &tmcons.TxRequest{
-					Height: msg.Height,
-					Round:  msg.Round,
-					TxKeys: txKeys,
-				},
-			},
-		}
-
 	default:
 		return nil, fmt.Errorf("consensus: message not recognized: %T", msg)
 	}
@@ -616,17 +580,6 @@ func MsgFromProto(msg *tmcons.Message) (Message, error) {
 			BlockID: *bi,
 			Votes:   bits,
 		}
-	case *tmcons.Message_TxRequest:
-		var txKeys []*types.TxKey
-		for _, txKey := range msg.TxRequest.TxKeys {
-			key, _ := types.TxKeyFromProto(txKey)
-			txKeys = append(txKeys, &key)
-		}
-		pb = &TxRequestMessage{
-			Height: msg.TxRequest.Height,
-			Round:  msg.TxRequest.Round,
-			TxKeys: txKeys,
-		}
 	default:
 		return nil, fmt.Errorf("consensus: message not recognized: %T", msg)
 	}

internal/consensus/msgs.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	protomem "github.com/tendermint/tendermint/proto/tendermint/mempool"
 	"runtime/debug"
 	"sync"
 	"time"
@@ -1194,28 +1193,6 @@ func (r *Reactor) handleDataMessage(ctx context.Context, envelope *p2p.Envelope,
 		case <-ctx.Done():
 			return ctx.Err()
 		}
-	case *tmcons.TxRequest:
-		trMsg := msgI.(*TxRequestMessage)
-		logger.Info("PSULOG: Received request for Txs", "txKeys", trMsg.TxKeys)
-		var txKeys []types.TxKey
-		for _, txKey := range trMsg.TxKeys {
-			txKeys = append(txKeys, *txKey)
-		}
-		txs := r.state.blockExec.GetTxsForKeys(txKeys)
-		r.Metrics.TxsSent.With("peer_id", string(envelope.From)).Add(1)
-		// TODO(psu): send all txs in 1 msg
-		for _, tx := range txs {
-			//logger.Info("PSULOG: Sending mempool ch for tx", "tx", tx)
-			if err := r.channels.mempoolCh.Send(ctx, p2p.Envelope{
-				To: ps.peerID,
-				Message: &protomem.Txs{
-					Txs: [][]byte{tx},
-				},
-			}); err != nil {
-				logger.Error("Unable to send tx for tx request", "peerId", ps.peerID, "txKeys", txKeys)
-			}
-		}
-
 	default:
 		return fmt.Errorf("received unknown message on DataChannel: %T", msg)
 	}

internal/consensus/reactor.go

@@ -1006,6 +1006,14 @@ func (cs *State) receiveRoutine(ctx context.Context, maxSteps int) {
 		// TODO should we handle context cancels here?
 	}
 }
+func (cs *State) fsyncAndCompleteProposal(ctx context.Context, fsyncUponCompletion bool, height int64, span otrace.Span) {
+	if fsyncUponCompletion {
+		if err := cs.wal.FlushAndSync(); err != nil { // fsync
+			panic("error flushing wal after receiving all block parts")
+		}
+	}
+	cs.handleCompleteProposal(ctx, height, span)
+}
 
 // state transitions on complete-proposal, 2/3-any, 2/3-one
 func (cs *State) handleMsg(ctx context.Context, mi msgInfo, fsyncUponCompletion bool) {
@@ -1032,18 +1040,12 @@ func (cs *State) handleMsg(ctx context.Context, mi msgInfo, fsyncUponCompletion
 			created := cs.tryCreateProposalBlock(msg.Proposal.Height, msg.Proposal.Round, msg.Proposal.Header, msg.Proposal.LastCommit, msg.Proposal.Evidence, msg.Proposal.ProposerAddress)
 			cs.metrics.ProposalBlockCreatedOnPropose.With("success", strconv.FormatBool(created)).Add(1)
 			if created {
-				if fsyncUponCompletion {
-					if err := cs.wal.FlushAndSync(); err != nil { // fsync
-						panic("error flushing wal after receiving all block parts")
-					}
-				}
-				cs.handleCompleteProposal(ctx, msg.Proposal.Height, span)
-
+				cs.fsyncAndCompleteProposal(ctx, fsyncUponCompletion, msg.Proposal.Height, span)
 			}
 		}
 
 	case *BlockPartMessage:
-		spanCtx, span := cs.tracer.Start(cs.getTracingCtx(ctx), "cs.state.handleBlockPartMsg")
+		_, span := cs.tracer.Start(cs.getTracingCtx(ctx), "cs.state.handleBlockPartMsg")
 		span.SetAttributes(attribute.Int("round", int(msg.Round)))
 		defer span.End()
 
@@ -1065,16 +1067,7 @@ func (cs *State) handleMsg(ctx context.Context, mi msgInfo, fsyncUponCompletion
 
 		cs.mtx.Lock()
 		if added && cs.ProposalBlockParts.IsComplete() {
-			if fsyncUponCompletion {
-				_, fsyncSpan := cs.tracer.Start(spanCtx, "cs.state.handleBlockPartMsg.fsync")
-				if err := cs.wal.FlushAndSync(); err != nil { // fsync
-					panic("error flushing wal after receiving all block parts")
-				}
-				fsyncSpan.End()
-			}
-			cs.handleCompleteProposal(ctx, msg.Height, span)
-		} else {
-			//cs.logger.Info("PSULOG - did not complete proposal", "added", added, "proposalblockparts", cs.ProposalBlockParts, "proposal", cs.Proposal)
+			cs.fsyncAndCompleteProposal(ctx, fsyncUponCompletion, msg.Height, span)
 		}
 		if added {
 			select {

internal/consensus/state.go

@@ -28,6 +28,14 @@ func (txKey *TxKey) ToProto() *tmproto.TxKey {
 	return tp
 }
 
+func TxKeysListToProto(txKeys []*TxKey) []*tmproto.TxKey {
+	var tps []*tmproto.TxKey
+	for _, txKey := range txKeys {
+		tps = append(tps, txKey.ToProto())
+	}
+	return tps
+}
+
 // TxKeyFromProto takes a protobuf representation of TxKey &
 // returns the native type.
 func TxKeyFromProto(dp *tmproto.TxKey) (TxKey, error) {
@@ -42,6 +50,18 @@ func TxKeyFromProto(dp *tmproto.TxKey) (TxKey, error) {
 	return txBzs, nil
 }
 
+func TxKeysListFromProto(dps []*tmproto.TxKey) ([]TxKey, error) {
+	var txKeys []TxKey
+	for _, txKey := range dps {
+		txKey, err := TxKeyFromProto(txKey)
+		if err != nil {
+			return nil, err
+		}
+		txKeys = append(txKeys, txKey)
+	}
+	return txKeys, nil
+}
+
 // ErrTxTooLarge defines an error when a transaction is too big to be sent in a
 // message to other peers.
 type ErrTxTooLarge struct {

types/mempool.go

@@ -199,7 +199,7 @@ func DefaultABCIParams() ABCIParams {
 		// When set to 0, vote extensions are not required.
 		VoteExtensionsEnableHeight: 0,
 		// When true, run CheckTx on each transaction in the mempool after each height.
-		RecheckTx: true,
+		RecheckTx: false,
 	}
 }