Revert "eth/fetcher: throttle tx fetches to 128KB responses (ethereum#28304)"

This reverts commit 32f701e.

Author: devopsbo3

eth/fetcher/tx_fetcher.go

@@ -39,22 +39,16 @@ const (
 	// can announce in a short time.
 	maxTxAnnounces = 4096
 
-	// maxTxRetrievals is the maximum number of transactions that can be fetched
-	// in one request. The rationale for picking 256 is to have a reasonabe lower
-	// bound for the transferred data (don't waste RTTs, transfer more meaningful
-	// batch sizes), but also have an upper bound on the sequentiality to allow
-	// using our entire peerset for deliveries.
-	//
-	// This number also acts as a failsafe against malicious announces which might
-	// cause us to request more data than we'd expect.
+	// maxTxRetrievals is the maximum transaction number can be fetched in one
+	// request. The rationale to pick 256 is:
+	//   - In eth protocol, the softResponseLimit is 2MB. Nowadays according to
+	//     Etherscan the average transaction size is around 200B, so in theory
+	//     we can include lots of transaction in a single protocol packet.
+	//   - However the maximum size of a single transaction is raised to 128KB,
+	//     so pick a middle value here to ensure we can maximize the efficiency
+	//     of the retrieval and response size overflow won't happen in most cases.
 	maxTxRetrievals = 256
 
-	// maxTxRetrievalSize is the max number of bytes that delivered transactions
-	// should weigh according to the announcements. The 128KB was chosen to limit
-	// retrieving a maximum of one blob transaction at a time to minimize hogging
-	// a connection between two peers.
-	maxTxRetrievalSize = 128 * 1024
-
 	// maxTxUnderpricedSetSize is the size of the underpriced transaction set that
 	// is used to track recent transactions that have been dropped so we don't
 	// re-request them.
@@ -865,36 +859,25 @@ func (f *TxFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{},
 		if len(f.announces[peer]) == 0 {
 			return // continue in the for-each
 		}
-		var (
-			hashes = make([]common.Hash, 0, maxTxRetrievals)
-			bytes  uint64
-		)
-		f.forEachAnnounce(f.announces[peer], func(hash common.Hash, meta *txMetadata) bool {
-			// If the transaction is alcear fetching, skip to the next one
-			if _, ok := f.fetching[hash]; ok {
-				return true
-			}
-			// Mark the hash as fetching and stash away possible alternates
-			f.fetching[hash] = peer
-
-			if _, ok := f.alternates[hash]; ok {
-				panic(fmt.Sprintf("alternate tracker already contains fetching item: %v", f.alternates[hash]))
-			}
-			f.alternates[hash] = f.announced[hash]
-			delete(f.announced, hash)
+		hashes := make([]common.Hash, 0, maxTxRetrievals)
+		f.forEachHash(f.announces[peer], func(hash common.Hash) bool {
+			if _, ok := f.fetching[hash]; !ok {
+				// Mark the hash as fetching and stash away possible alternates
+				f.fetching[hash] = peer
+
+				if _, ok := f.alternates[hash]; ok {
+					panic(fmt.Sprintf("alternate tracker already contains fetching item: %v", f.alternates[hash]))
+				}
+				f.alternates[hash] = f.announced[hash]
+				delete(f.announced, hash)
 
-			// Accumulate the hash and stop if the limit was reached
-			hashes = append(hashes, hash)
-			if len(hashes) >= maxTxRetrievals {
-				return false // break in the for-each
-			}
-			if meta != nil { // Only set eth/68 and upwards
-				bytes += uint64(meta.size)
-				if bytes >= maxTxRetrievalSize {
-					return false
+				// Accumulate the hash and stop if the limit was reached
+				hashes = append(hashes, hash)
+				if len(hashes) >= maxTxRetrievals {
+					return false // break in the for-each
 				}
 			}
-			return true // scheduled, try to add more
+			return true // continue in the for-each
 		})
 		// If any hashes were allocated, request them from the peer
 		if len(hashes) > 0 {
@@ -939,28 +922,27 @@ func (f *TxFetcher) forEachPeer(peers map[string]struct{}, do func(peer string))
 	}
 }
 
-// forEachAnnounce does a range loop over a map of announcements in production,
-// but during testing it does a deterministic sorted random to allow reproducing
-// issues.
-func (f *TxFetcher) forEachAnnounce(announces map[common.Hash]*txMetadata, do func(hash common.Hash, meta *txMetadata) bool) {
+// forEachHash does a range loop over a map of hashes in production, but during
+// testing it does a deterministic sorted random to allow reproducing issues.
+func (f *TxFetcher) forEachHash(hashes map[common.Hash]*txMetadata, do func(hash common.Hash) bool) {
 	// If we're running production, use whatever Go's map gives us
 	if f.rand == nil {
-		for hash, meta := range announces {
-			if !do(hash, meta) {
+		for hash := range hashes {
+			if !do(hash) {
 				return
 			}
 		}
 		return
 	}
 	// We're running the test suite, make iteration deterministic
-	list := make([]common.Hash, 0, len(announces))
-	for hash := range announces {
+	list := make([]common.Hash, 0, len(hashes))
+	for hash := range hashes {
 		list = append(list, hash)
 	}
 	sortHashes(list)
 	rotateHashes(list, f.rand.Intn(len(list)))
 	for _, hash := range list {
-		if !do(hash, announces[hash]) {
+		if !do(hash) {
 			return
 		}
 	}

eth/fetcher/tx_fetcher_test.go

@@ -27,7 +27,6 @@ import (
 	"github.com/ethereum/go-ethereum/common/mclock"
 	"github.com/ethereum/go-ethereum/core/txpool"
 	"github.com/ethereum/go-ethereum/core/types"
-	"github.com/ethereum/go-ethereum/params"
 )
 
 var (
@@ -994,14 +993,15 @@ func TestTransactionFetcherTimeoutTimerResets(t *testing.T) {
 	})
 }
 
-// Tests that if thousands of transactions are announced, only a small
+// Tests that if thousands of transactions are announces, only a small
 // number of them will be requested at a time.
 func TestTransactionFetcherRateLimiting(t *testing.T) {
-	// Create a slew of transactions and announce them
+	// Create a slew of transactions and to announce them
 	var hashes []common.Hash
 	for i := 0; i < maxTxAnnounces; i++ {
 		hashes = append(hashes, common.Hash{byte(i / 256), byte(i % 256)})
 	}
+
 	testTransactionFetcherParallel(t, txFetcherTest{
 		init: func() *TxFetcher {
 			return NewTxFetcher(
@@ -1029,68 +1029,6 @@ func TestTransactionFetcherRateLimiting(t *testing.T) {
 	})
 }
 
-// Tests that if huge transactions are announced, only a small number of them will
-// be requested at a time, to keep the responses below a resonable level.
-func TestTransactionFetcherBandwidthLimiting(t *testing.T) {
-	testTransactionFetcherParallel(t, txFetcherTest{
-		init: func() *TxFetcher {
-			return NewTxFetcher(
-				func(common.Hash) bool { return false },
-				nil,
-				func(string, []common.Hash) error { return nil },
-				nil,
-			)
-		},
-		steps: []interface{}{
-			// Announce mid size transactions from A to verify that multiple
-			// ones can be piled into a single request.
-			doTxNotify{peer: "A",
-				hashes: []common.Hash{{0x01}, {0x02}, {0x03}, {0x04}},
-				types:  []byte{types.LegacyTxType, types.LegacyTxType, types.LegacyTxType, types.LegacyTxType},
-				sizes:  []uint32{48 * 1024, 48 * 1024, 48 * 1024, 48 * 1024},
-			},
-			// Announce exactly on the limit transactions to see that only one
-			// gets requested
-			doTxNotify{peer: "B",
-				hashes: []common.Hash{{0x05}, {0x06}},
-				types:  []byte{types.LegacyTxType, types.LegacyTxType},
-				sizes:  []uint32{maxTxRetrievalSize, maxTxRetrievalSize},
-			},
-			// Announce oversized blob transactions to see that overflows are ok
-			doTxNotify{peer: "C",
-				hashes: []common.Hash{{0x07}, {0x08}},
-				types:  []byte{types.BlobTxType, types.BlobTxType},
-				sizes:  []uint32{params.MaxBlobGasPerBlock, params.MaxBlobGasPerBlock},
-			},
-			doWait{time: txArriveTimeout, step: true},
-			isWaiting(nil),
-			isScheduledWithMeta{
-				tracking: map[string][]announce{
-					"A": {
-						{common.Hash{0x01}, typeptr(types.LegacyTxType), sizeptr(48 * 1024)},
-						{common.Hash{0x02}, typeptr(types.LegacyTxType), sizeptr(48 * 1024)},
-						{common.Hash{0x03}, typeptr(types.LegacyTxType), sizeptr(48 * 1024)},
-						{common.Hash{0x04}, typeptr(types.LegacyTxType), sizeptr(48 * 1024)},
-					},
-					"B": {
-						{common.Hash{0x05}, typeptr(types.LegacyTxType), sizeptr(maxTxRetrievalSize)},
-						{common.Hash{0x06}, typeptr(types.LegacyTxType), sizeptr(maxTxRetrievalSize)},
-					},
-					"C": {
-						{common.Hash{0x07}, typeptr(types.BlobTxType), sizeptr(params.MaxBlobGasPerBlock)},
-						{common.Hash{0x08}, typeptr(types.BlobTxType), sizeptr(params.MaxBlobGasPerBlock)},
-					},
-				},
-				fetching: map[string][]common.Hash{
-					"A": {{0x02}, {0x03}, {0x04}},
-					"B": {{0x06}},
-					"C": {{0x08}},
-				},
-			},
-		},
-	})
-}
-
 // Tests that then number of transactions a peer is allowed to announce and/or
 // request at the same time is hard capped.
 func TestTransactionFetcherDoSProtection(t *testing.T) {
@@ -1726,7 +1664,7 @@ func testTransactionFetcher(t *testing.T, tt txFetcherTest) {
 						if (meta == nil && (ann.kind != nil || ann.size != nil)) ||
 							(meta != nil && (ann.kind == nil || ann.size == nil)) ||
 							(meta != nil && (meta.kind != *ann.kind || meta.size != *ann.size)) {
-							t.Errorf("step %d, peer %s, hash %x: waitslot metadata mismatch: want %v, have %v/%v", i, peer, ann.hash, meta, *ann.kind, *ann.size)
+							t.Errorf("step %d, peer %s, hash %x: waitslot metadata mismatch: want %v, have %v/%v", i, peer, ann.hash, meta, ann.kind, ann.size)
 						}
 					}
 				}
@@ -1795,7 +1733,7 @@ func testTransactionFetcher(t *testing.T, tt txFetcherTest) {
 						if (meta == nil && (ann.kind != nil || ann.size != nil)) ||
 							(meta != nil && (ann.kind == nil || ann.size == nil)) ||
 							(meta != nil && (meta.kind != *ann.kind || meta.size != *ann.size)) {
-							t.Errorf("step %d, peer %s, hash %x: announce metadata mismatch: want %v, have %v/%v", i, peer, ann.hash, meta, *ann.kind, *ann.size)
+							t.Errorf("step %d, peer %s, hash %x: announce metadata mismatch: want %v, have %v/%v", i, peer, ann.hash, meta, ann.kind, ann.size)
 						}
 					}
 				}