Clean up gossip batching/frequency (#1253)

Author: aaronbuchwald

plugin/evm/block_builder.go

@@ -17,16 +17,6 @@ import (
 )
 
 const (
-	// waitBlockTime is the amount of time to wait for BuildBlock to be
-	// called by the engine before deciding whether or not to gossip the
-	// transaction that triggered the PendingTxs message to the engine.
-	//
-	// This is done to reduce contention in the network when there is no
-	// preferred producer. If we did not wait here, we may gossip a new
-	// transaction to a peer while building a block that will conflict with
-	// whatever the peer makes.
-	waitBlockTime = 100 * time.Millisecond
-
 	// Minimum amount of time to wait after building a block before attempting to build a block
 	// a second time without changing the contents of the mempool.
 	minBlockBuildingRetryDelay = 500 * time.Millisecond
@@ -168,9 +158,6 @@ func (b *blockBuilder) awaitSubmittedTxs() {
 				b.signalTxsReady()
 
 				if b.gossiper != nil && len(ethTxsEvent.Txs) > 0 {
-					// Give time for this node to build a block before attempting to
-					// gossip
-					time.Sleep(waitBlockTime)
 					// [GossipEthTxs] will block unless [gossiper.ethTxsToGossipChan] (an
 					// unbuffered channel) is listened on
 					if err := b.gossiper.GossipEthTxs(ethTxsEvent.Txs); err != nil {
@@ -186,9 +173,6 @@ func (b *blockBuilder) awaitSubmittedTxs() {
 
 				newTxs := b.mempool.GetNewTxs()
 				if b.gossiper != nil && len(newTxs) > 0 {
-					// Give time for this node to build a block before attempting to
-					// gossip
-					time.Sleep(waitBlockTime)
 					if err := b.gossiper.GossipAtomicTxs(newTxs); err != nil {
 						log.Warn(
 							"failed to gossip new atomic transactions",

plugin/evm/gossiper.go

@@ -36,6 +36,10 @@ const (
 	// [ethTxsGossipInterval] is how often we attempt to gossip newly seen
 	// transactions to other nodes.
 	ethTxsGossipInterval = 500 * time.Millisecond
+
+	// [minGossipBatchInterval] is the minimum amount of time that must pass
+	// before our last gossip to peers.
+	minGossipBatchInterval = 50 * time.Millisecond
 )
 
 // Gossiper handles outgoing gossip of transactions
@@ -326,7 +330,7 @@ func (n *pushGossiper) sendEthTxs(txs []*types.Transaction) error {
 }
 
 func (n *pushGossiper) gossipEthTxs(force bool) (int, error) {
-	if (!force && time.Since(n.lastGossiped) < ethTxsGossipInterval) || len(n.ethTxsToGossip) == 0 {
+	if (!force && time.Since(n.lastGossiped) < minGossipBatchInterval) || len(n.ethTxsToGossip) == 0 {
 		return 0, nil
 	}
 	n.lastGossiped = time.Now()

plugin/evm/gossiper_atomic_gossiping_test.go

@@ -58,7 +58,7 @@ func TestMempoolAtmTxsIssueTxAndGossiping(t *testing.T) {
 
 	// Optimistically gossip raw tx
 	assert.NoError(vm.issueTx(tx, true /*=local*/))
-	time.Sleep(waitBlockTime * 3)
+	time.Sleep(500 * time.Millisecond)
 	gossipedLock.Lock()
 	assert.Equal(1, gossiped)
 	gossipedLock.Unlock()
@@ -119,7 +119,7 @@ func TestMempoolAtmTxsAppGossipHandling(t *testing.T) {
 
 	// show that no txID is requested
 	assert.NoError(vm.AppGossip(context.Background(), nodeID, msgBytes))
-	time.Sleep(waitBlockTime * 3)
+	time.Sleep(500 * time.Millisecond)
 
 	assert.False(txRequested, "tx should not have been requested")
 	txGossipedLock.Lock()
@@ -218,7 +218,7 @@ func TestMempoolAtmTxsAppGossipHandlingDiscardedTx(t *testing.T) {
 	assert.NoError(err)
 
 	assert.NoError(vm.AppGossip(context.Background(), nodeID, msgBytes))
-	time.Sleep(waitBlockTime * 3)
+	time.Sleep(500 * time.Millisecond)
 	assert.False(txRequested, "tx shouldn't be requested")
 	txGossipedLock.Lock()
 	assert.Equal(1, txGossiped, "conflicting tx should have been gossiped")