forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 10:22:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

core,eth,trie: use common/prque instead of karalabe's

Browse source
This commit is contained in:
Delweng Zheng 2018-08-25 23:26:55 +08:00
parent 70398d300d
commit 7f3cf38627
5 changed files with 30 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
core/blockchain.go
View file

@ -29,6 +29,7 @@ import (
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/common/mclock"
"github.com/ethereum/go-ethereum/common/prque"
"github.com/ethereum/go-ethereum/consensus" "github.com/ethereum/go-ethereum/consensus"
"github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/rawdb"
"github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/state"
@ -43,7 +44,6 @@ import (
"github.com/ethereum/go-ethereum/rlp" "github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/trie" "github.com/ethereum/go-ethereum/trie"
"github.com/hashicorp/golang-lru" "github.com/hashicorp/golang-lru"
"gopkg.in/karalabe/cookiejar.v2/collections/prque"
) )
var ( var (
@ -151,7 +151,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
chainConfig: chainConfig, chainConfig: chainConfig,
cacheConfig: cacheConfig, cacheConfig: cacheConfig,
db: db, db: db,
triegc: prque.New(), triegc: prque.New(nil),
stateCache: state.NewDatabase(db), stateCache: state.NewDatabase(db),
quit: make(chan struct{}), quit: make(chan struct{}),
bodyCache: bodyCache, bodyCache: bodyCache,
@ -915,7 +915,7 @@ func (bc *BlockChain) WriteBlockWithState(block *types.Block, receipts []*types.
} else { } else {
// Full but not archive node, do proper garbage collection // Full but not archive node, do proper garbage collection
triedb.Reference(root, common.Hash{}) // metadata reference to keep trie alive triedb.Reference(root, common.Hash{}) // metadata reference to keep trie alive
bc.triegc.Push(root, -float32(block.NumberU64())) bc.triegc.Push(root, -int64(block.NumberU64()))
if current := block.NumberU64(); current > triesInMemory { if current := block.NumberU64(); current > triesInMemory {
// If we exceeded our memory allowance, flush matured singleton nodes to disk // If we exceeded our memory allowance, flush matured singleton nodes to disk

6
core/tx_pool.go
View file

@ -26,13 +26,13 @@ import (
"time" "time"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/prque"
"github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/state"
"github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics" "github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/params" "github.com/ethereum/go-ethereum/params"
"gopkg.in/karalabe/cookiejar.v2/collections/prque"
) )
const ( const (
@ -987,11 +987,11 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
if pending > pool.config.GlobalSlots { if pending > pool.config.GlobalSlots {
pendingBeforeCap := pending pendingBeforeCap := pending
// Assemble a spam order to penalize large transactors first // Assemble a spam order to penalize large transactors first
spammers := prque.New() spammers := prque.New(nil)
for addr, list := range pool.pending { for addr, list := range pool.pending {
// Only evict transactions from high rollers // Only evict transactions from high rollers
if !pool.locals.contains(addr) && uint64(list.Len()) > pool.config.AccountSlots { if !pool.locals.contains(addr) && uint64(list.Len()) > pool.config.AccountSlots {
spammers.Push(addr, float32(list.Len())) spammers.Push(addr, int64(list.Len()))
} }
} }
// Gradually drop transactions from offenders // Gradually drop transactions from offenders

34
eth/downloader/queue.go
View file

@ -26,10 +26,10 @@ import (
"time" "time"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/prque"
"github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics" "github.com/ethereum/go-ethereum/metrics"
"gopkg.in/karalabe/cookiejar.v2/collections/prque"
) )
var ( var (
@ -105,11 +105,11 @@ func newQueue() *queue {
headerPendPool: make(map[string]*fetchRequest), headerPendPool: make(map[string]*fetchRequest),
headerContCh: make(chan bool), headerContCh: make(chan bool),
blockTaskPool: make(map[common.Hash]*types.Header), blockTaskPool: make(map[common.Hash]*types.Header),
blockTaskQueue: prque.New(), blockTaskQueue: prque.New(nil),
blockPendPool: make(map[string]*fetchRequest), blockPendPool: make(map[string]*fetchRequest),
blockDonePool: make(map[common.Hash]struct{}), blockDonePool: make(map[common.Hash]struct{}),
receiptTaskPool: make(map[common.Hash]*types.Header), receiptTaskPool: make(map[common.Hash]*types.Header),
receiptTaskQueue: prque.New(), receiptTaskQueue: prque.New(nil),
receiptPendPool: make(map[string]*fetchRequest), receiptPendPool: make(map[string]*fetchRequest),
receiptDonePool: make(map[common.Hash]struct{}), receiptDonePool: make(map[common.Hash]struct{}),
resultCache: make([]*fetchResult, blockCacheItems), resultCache: make([]*fetchResult, blockCacheItems),
@ -277,7 +277,7 @@ func (q *queue) ScheduleSkeleton(from uint64, skeleton []*types.Header) {
} }
// Schedule all the header retrieval tasks for the skeleton assembly // Schedule all the header retrieval tasks for the skeleton assembly
q.headerTaskPool = make(map[uint64]*types.Header) q.headerTaskPool = make(map[uint64]*types.Header)
q.headerTaskQueue = prque.New() q.headerTaskQueue = prque.New(nil)
q.headerPeerMiss = make(map[string]map[uint64]struct{}) // Reset availability to correct invalid chains q.headerPeerMiss = make(map[string]map[uint64]struct{}) // Reset availability to correct invalid chains
q.headerResults = make([]*types.Header, len(skeleton)*MaxHeaderFetch) q.headerResults = make([]*types.Header, len(skeleton)*MaxHeaderFetch)
q.headerProced = 0 q.headerProced = 0
@ -288,7 +288,7 @@ func (q *queue) ScheduleSkeleton(from uint64, skeleton []*types.Header) {
index := from + uint64(i*MaxHeaderFetch) index := from + uint64(i*MaxHeaderFetch)
q.headerTaskPool[index] = header q.headerTaskPool[index] = header
q.headerTaskQueue.Push(index, -float32(index)) q.headerTaskQueue.Push(index, -int64(index))
} }
} }
@ -334,11 +334,11 @@ func (q *queue) Schedule(headers []*types.Header, from uint64) []*types.Header {
} }
// Queue the header for content retrieval // Queue the header for content retrieval
q.blockTaskPool[hash] = header q.blockTaskPool[hash] = header
q.blockTaskQueue.Push(header, -float32(header.Number.Uint64())) q.blockTaskQueue.Push(header, -int64(header.Number.Uint64()))
if q.mode == FastSync { if q.mode == FastSync {
q.receiptTaskPool[hash] = header q.receiptTaskPool[hash] = header
q.receiptTaskQueue.Push(header, -float32(header.Number.Uint64())) q.receiptTaskQueue.Push(header, -int64(header.Number.Uint64()))
} }
inserts = append(inserts, header) inserts = append(inserts, header)
q.headerHead = hash q.headerHead = hash
@ -436,7 +436,7 @@ func (q *queue) ReserveHeaders(p *peerConnection, count int) *fetchRequest {
} }
// Merge all the skipped batches back // Merge all the skipped batches back
for _, from := range skip { for _, from := range skip {
q.headerTaskQueue.Push(from, -float32(from)) q.headerTaskQueue.Push(from, -int64(from))
} }
// Assemble and return the block download request // Assemble and return the block download request
if send == 0 { if send == 0 {
@ -542,7 +542,7 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common
} }
// Merge all the skipped headers back // Merge all the skipped headers back
for _, header := range skip { for _, header := range skip {
taskQueue.Push(header, -float32(header.Number.Uint64())) taskQueue.Push(header, -int64(header.Number.Uint64()))
} }
if progress { if progress {
// Wake WaitResults, resultCache was modified // Wake WaitResults, resultCache was modified
@ -585,10 +585,10 @@ func (q *queue) cancel(request *fetchRequest, taskQueue *prque.Prque, pendPool m
defer q.lock.Unlock() defer q.lock.Unlock()
if request.From > 0 { if request.From > 0 {
taskQueue.Push(request.From, -float32(request.From)) taskQueue.Push(request.From, -int64(request.From))
} }
for _, header := range request.Headers { for _, header := range request.Headers {
taskQueue.Push(header, -float32(header.Number.Uint64())) taskQueue.Push(header, -int64(header.Number.Uint64()))
} }
delete(pendPool, request.Peer.id) delete(pendPool, request.Peer.id)
} }
@ -602,13 +602,13 @@ func (q *queue) Revoke(peerID string) {
if request, ok := q.blockPendPool[peerID]; ok { if request, ok := q.blockPendPool[peerID]; ok {
for _, header := range request.Headers { for _, header := range request.Headers {
q.blockTaskQueue.Push(header, -float32(header.Number.Uint64())) q.blockTaskQueue.Push(header, -int64(header.Number.Uint64()))
} }
delete(q.blockPendPool, peerID) delete(q.blockPendPool, peerID)
} }
if request, ok := q.receiptPendPool[peerID]; ok { if request, ok := q.receiptPendPool[peerID]; ok {
for _, header := range request.Headers { for _, header := range request.Headers {
q.receiptTaskQueue.Push(header, -float32(header.Number.Uint64())) q.receiptTaskQueue.Push(header, -int64(header.Number.Uint64()))
} }
delete(q.receiptPendPool, peerID) delete(q.receiptPendPool, peerID)
} }
@ -657,10 +657,10 @@ func (q *queue) expire(timeout time.Duration, pendPool map[string]*fetchRequest,
// Return any non satisfied requests to the pool // Return any non satisfied requests to the pool
if request.From > 0 { if request.From > 0 {
taskQueue.Push(request.From, -float32(request.From)) taskQueue.Push(request.From, -int64(request.From))
} }
for _, header := range request.Headers { for _, header := range request.Headers {
taskQueue.Push(header, -float32(header.Number.Uint64())) taskQueue.Push(header, -int64(header.Number.Uint64()))
} }
// Add the peer to the expiry report along the the number of failed requests // Add the peer to the expiry report along the the number of failed requests
expiries[id] = len(request.Headers) expiries[id] = len(request.Headers)
@ -731,7 +731,7 @@ func (q *queue) DeliverHeaders(id string, headers []*types.Header, headerProcCh
} }
miss[request.From] = struct{}{} miss[request.From] = struct{}{}
q.headerTaskQueue.Push(request.From, -float32(request.From)) q.headerTaskQueue.Push(request.From, -int64(request.From))
return 0, errors.New("delivery not accepted") return 0, errors.New("delivery not accepted")
} }
// Clean up a successful fetch and try to deliver any sub-results // Clean up a successful fetch and try to deliver any sub-results
@ -854,7 +854,7 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, taskQ
// Return all failed or missing fetches to the queue // Return all failed or missing fetches to the queue
for _, header := range request.Headers { for _, header := range request.Headers {
if header != nil { if header != nil {
taskQueue.Push(header, -float32(header.Number.Uint64())) taskQueue.Push(header, -int64(header.Number.Uint64()))
} }
} }
// Wake up WaitResults // Wake up WaitResults

8
eth/fetcher/fetcher.go
View file

@ -23,10 +23,10 @@ import (
"time" "time"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/prque"
"github.com/ethereum/go-ethereum/consensus" "github.com/ethereum/go-ethereum/consensus"
"github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/log"
"gopkg.in/karalabe/cookiejar.v2/collections/prque"
) )
const ( const (
@ -160,7 +160,7 @@ func New(getBlock blockRetrievalFn, verifyHeader headerVerifierFn, broadcastBloc
fetching: make(map[common.Hash]*announce), fetching: make(map[common.Hash]*announce),
fetched: make(map[common.Hash][]*announce), fetched: make(map[common.Hash][]*announce),
completing: make(map[common.Hash]*announce), completing: make(map[common.Hash]*announce),
queue: prque.New(), queue: prque.New(nil),
queues: make(map[string]int), queues: make(map[string]int),
queued: make(map[common.Hash]*inject), queued: make(map[common.Hash]*inject),
getBlock: getBlock, getBlock: getBlock,
@ -299,7 +299,7 @@ func (f *Fetcher) loop() {
// If too high up the chain or phase, continue later // If too high up the chain or phase, continue later
number := op.block.NumberU64() number := op.block.NumberU64()
if number > height+1 { if number > height+1 {
f.queue.Push(op, -float32(number)) f.queue.Push(op, -int64(number))
if f.queueChangeHook != nil { if f.queueChangeHook != nil {
f.queueChangeHook(hash, true) f.queueChangeHook(hash, true)
} }
@ -624,7 +624,7 @@ func (f *Fetcher) enqueue(peer string, block *types.Block) {
} }
f.queues[peer] = count f.queues[peer] = count
f.queued[hash] = op f.queued[hash] = op
f.queue.Push(op, -float32(block.NumberU64())) f.queue.Push(op, -int64(block.NumberU64()))
if f.queueChangeHook != nil { if f.queueChangeHook != nil {
f.queueChangeHook(op.block.Hash(), true) f.queueChangeHook(op.block.Hash(), true)
} }

6
trie/sync.go
View file

@ -21,8 +21,8 @@ import (
"fmt" "fmt"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/prque"
"github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/ethdb"
"gopkg.in/karalabe/cookiejar.v2/collections/prque"
) )
// ErrNotRequested is returned by the trie sync when it's requested to process a // ErrNotRequested is returned by the trie sync when it's requested to process a
@ -84,7 +84,7 @@ func NewSync(root common.Hash, database DatabaseReader, callback LeafCallback) *
database: database, database: database,
membatch: newSyncMemBatch(), membatch: newSyncMemBatch(),
requests: make(map[common.Hash]*request), requests: make(map[common.Hash]*request),
queue: prque.New(), queue: prque.New(nil),
} }
ts.AddSubTrie(root, 0, common.Hash{}, callback) ts.AddSubTrie(root, 0, common.Hash{}, callback)
return ts return ts
@ -242,7 +242,7 @@ func (s *Sync) schedule(req *request) {
return return
} }
// Schedule the request for future retrieval // Schedule the request for future retrieval
s.queue.Push(req.hash, float32(req.depth)) s.queue.Push(req.hash, int64(req.depth))
s.requests[req.hash] = req s.requests[req.hash] = req
} }