core, eth: implement eth/65 transaction fetcher

Author: rjl493456442

core/tx_pool.go

@@ -864,6 +864,12 @@ func (pool *TxPool) Get(hash common.Hash) *types.Transaction {
 	return pool.all.Get(hash)
 }
 
+// Has returns an indicator whether txpool has a transaction cached with the
+// given hash.
+func (pool *TxPool) Has(hash common.Hash) bool {
+	return pool.all.Get(hash) != nil
+}
+
 // removeTx removes a single transaction from the queue, moving all subsequent
 // transactions back to the future queue.
 func (pool *TxPool) removeTx(hash common.Hash, outofbound bool) {

eth/downloader/peer.go

@@ -470,7 +470,7 @@ func (ps *peerSet) HeaderIdlePeers() ([]*peerConnection, int) {
 		defer p.lock.RUnlock()
 		return p.headerThroughput
 	}
-	return ps.idlePeers(62, 64, idle, throughput)
+	return ps.idlePeers(62, 65, idle, throughput)
 }
 
 // BodyIdlePeers retrieves a flat list of all the currently body-idle peers within
@@ -484,7 +484,7 @@ func (ps *peerSet) BodyIdlePeers() ([]*peerConnection, int) {
 		defer p.lock.RUnlock()
 		return p.blockThroughput
 	}
-	return ps.idlePeers(62, 64, idle, throughput)
+	return ps.idlePeers(62, 65, idle, throughput)
 }
 
 // ReceiptIdlePeers retrieves a flat list of all the currently receipt-idle peers
@@ -498,7 +498,7 @@ func (ps *peerSet) ReceiptIdlePeers() ([]*peerConnection, int) {
 		defer p.lock.RUnlock()
 		return p.receiptThroughput
 	}
-	return ps.idlePeers(63, 64, idle, throughput)
+	return ps.idlePeers(63, 65, idle, throughput)
 }
 
 // NodeDataIdlePeers retrieves a flat list of all the currently node-data-idle
@@ -512,7 +512,7 @@ func (ps *peerSet) NodeDataIdlePeers() ([]*peerConnection, int) {
 		defer p.lock.RUnlock()
 		return p.stateThroughput
 	}
-	return ps.idlePeers(63, 64, idle, throughput)
+	return ps.idlePeers(63, 65, idle, throughput)
 }
 
 // idlePeers retrieves a flat list of all currently idle peers satisfying the

eth/fetcher/fetcher.go renamed to eth/fetcher/block_fetcher.go

@@ -76,7 +76,7 @@ func makeChain(n int, seed byte, parent *types.Block) ([]common.Hash, map[common
 
 // fetcherTester is a test simulator for mocking out local block chain.
 type fetcherTester struct {
-	fetcher *Fetcher
+	fetcher *BlockFetcher
 
 	hashes []common.Hash                // Hash chain belonging to the tester
 	blocks map[common.Hash]*types.Block // Blocks belonging to the tester
@@ -92,7 +92,7 @@ func newTester() *fetcherTester {
 		blocks: map[common.Hash]*types.Block{genesis.Hash(): genesis},
 		drops:  make(map[string]bool),
 	}
-	tester.fetcher = New(tester.getBlock, tester.verifyHeader, tester.broadcastBlock, tester.chainHeight, tester.insertChain, tester.dropPeer)
+	tester.fetcher = NewBlockFetcher(tester.getBlock, tester.verifyHeader, tester.broadcastBlock, tester.chainHeight, tester.insertChain, tester.dropPeer)
 	tester.fetcher.Start()
 
 	return tester

eth/fetcher/fetcher_test.go renamed to eth/fetcher/block_fetcher_test.go

@@ -23,15 +23,15 @@ import (
 )
 
 var (
-	propAnnounceInMeter   = metrics.NewRegisteredMeter("eth/fetcher/prop/announces/in", nil)
-	propAnnounceOutTimer  = metrics.NewRegisteredTimer("eth/fetcher/prop/announces/out", nil)
-	propAnnounceDropMeter = metrics.NewRegisteredMeter("eth/fetcher/prop/announces/drop", nil)
-	propAnnounceDOSMeter  = metrics.NewRegisteredMeter("eth/fetcher/prop/announces/dos", nil)
+	blockAnnounceInMeter   = metrics.NewRegisteredMeter("eth/fetcher/prop/block/announces/in", nil)
+	blockAnnounceOutTimer  = metrics.NewRegisteredTimer("eth/fetcher/prop/block/announces/out", nil)
+	blockAnnounceDropMeter = metrics.NewRegisteredMeter("eth/fetcher/prop/block/announces/drop", nil)
+	blockAnnounceDOSMeter  = metrics.NewRegisteredMeter("eth/fetcher/prop/block/announces/dos", nil)
 
-	propBroadcastInMeter   = metrics.NewRegisteredMeter("eth/fetcher/prop/broadcasts/in", nil)
-	propBroadcastOutTimer  = metrics.NewRegisteredTimer("eth/fetcher/prop/broadcasts/out", nil)
-	propBroadcastDropMeter = metrics.NewRegisteredMeter("eth/fetcher/prop/broadcasts/drop", nil)
-	propBroadcastDOSMeter  = metrics.NewRegisteredMeter("eth/fetcher/prop/broadcasts/dos", nil)
+	blockBroadcastInMeter   = metrics.NewRegisteredMeter("eth/fetcher/prop/block/broadcasts/in", nil)
+	blockBroadcastOutTimer  = metrics.NewRegisteredTimer("eth/fetcher/prop/block/broadcasts/out", nil)
+	blockBroadcastDropMeter = metrics.NewRegisteredMeter("eth/fetcher/prop/block/broadcasts/drop", nil)
+	blockBroadcastDOSMeter  = metrics.NewRegisteredMeter("eth/fetcher/prop/block/broadcasts/dos", nil)
 
 	headerFetchMeter = metrics.NewRegisteredMeter("eth/fetcher/fetch/headers", nil)
 	bodyFetchMeter   = metrics.NewRegisteredMeter("eth/fetcher/fetch/bodies", nil)
@@ -40,4 +40,15 @@ var (
 	headerFilterOutMeter = metrics.NewRegisteredMeter("eth/fetcher/filter/headers/out", nil)
 	bodyFilterInMeter    = metrics.NewRegisteredMeter("eth/fetcher/filter/bodies/in", nil)
 	bodyFilterOutMeter   = metrics.NewRegisteredMeter("eth/fetcher/filter/bodies/out", nil)
+
+	txAnnounceInMeter         = metrics.NewRegisteredMeter("eth/fetcher/prop/transaction/announces/in", nil)
+	txAnnounceDOSMeter        = metrics.NewRegisteredMeter("eth/fetcher/prop/transaction/announces/dos", nil)
+	txAnnounceSkipMeter       = metrics.NewRegisteredMeter("eth/fetcher/prop/transaction/announces/skip", nil)
+	txAnnounceUnderpriceMeter = metrics.NewRegisteredMeter("eth/fetcher/prop/transaction/announces/underprice", nil)
+	txBroadcastInMeter        = metrics.NewRegisteredMeter("eth/fetcher/prop/transaction/broadcasts/in", nil)
+	txFetchOutMeter           = metrics.NewRegisteredMeter("eth/fetcher/fetch/transaction/out", nil)
+	txFetchSuccessMeter       = metrics.NewRegisteredMeter("eth/fetcher/fetch/transaction/success", nil)
+	txFetchTimeoutMeter       = metrics.NewRegisteredMeter("eth/fetcher/fetch/transaction/timeout", nil)
+	txFetchInvalidMeter       = metrics.NewRegisteredMeter("eth/fetcher/fetch/transaction/invalid", nil)
+	txFetchDurationTimer      = metrics.NewRegisteredTimer("eth/fetcher/fetch/transaction/duration", nil)
 )