From ee6c890d6c2abd46050dc98a04211d42486acbee Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Thu, 1 Dec 2022 09:00:21 +0200 Subject: [PATCH 1/2] common, core, eth, les, trie: make prque generic --- common/prque/lazyqueue.go | 75 +++++++++++++++----------------- common/prque/prque.go | 39 ++++++++--------- common/prque/prque_test.go | 8 ++-- common/prque/sstack.go | 42 +++++++++--------- common/prque/sstack_test.go | 30 ++++++------- core/blockchain.go | 18 ++++---- core/rawdb/chain_iterator.go | 8 ++-- core/txpool/txpool.go | 6 +-- eth/downloader/queue.go | 36 +++++++-------- eth/fetcher/block_fetcher.go | 6 +-- les/downloader/queue.go | 38 ++++++++-------- les/fetcher/block_fetcher.go | 6 +-- les/flowcontrol/manager.go | 6 +-- les/servingqueue.go | 16 +++---- les/vflux/server/prioritypool.go | 23 +++++----- trie/sync.go | 4 +- 16 files changed, 177 insertions(+), 184 deletions(-) diff --git a/common/prque/lazyqueue.go b/common/prque/lazyqueue.go index 13ef3ed2cdb..df85b5f272a 100644 --- a/common/prque/lazyqueue.go +++ b/common/prque/lazyqueue.go @@ -32,31 +32,31 @@ import ( // // If the upper estimate is exceeded then Update should be called for that item. // A global Refresh function should also be called periodically. -type LazyQueue struct { +type LazyQueue[V any] struct { clock mclock.Clock // Items are stored in one of two internal queues ordered by estimated max // priority until the next and the next-after-next refresh. Update and Refresh // always places items in queue[1]. - queue [2]*sstack - popQueue *sstack + queue [2]*sstack[V] + popQueue *sstack[V] period time.Duration maxUntil mclock.AbsTime indexOffset int - setIndex SetIndexCallback - priority PriorityCallback - maxPriority MaxPriorityCallback + setIndex SetIndexCallback[V] + priority PriorityCallback[V] + maxPriority MaxPriorityCallback[V] lastRefresh1, lastRefresh2 mclock.AbsTime } type ( - PriorityCallback func(data interface{}) int64 // actual priority callback - MaxPriorityCallback func(data interface{}, until mclock.AbsTime) int64 // estimated maximum priority callback + PriorityCallback[V any] func(data V) int64 // actual priority callback + MaxPriorityCallback[V any] func(data V, until mclock.AbsTime) int64 // estimated maximum priority callback ) // NewLazyQueue creates a new lazy queue -func NewLazyQueue(setIndex SetIndexCallback, priority PriorityCallback, maxPriority MaxPriorityCallback, clock mclock.Clock, refreshPeriod time.Duration) *LazyQueue { - q := &LazyQueue{ - popQueue: newSstack(nil, false), +func NewLazyQueue[V any](setIndex SetIndexCallback[V], priority PriorityCallback[V], maxPriority MaxPriorityCallback[V], clock mclock.Clock, refreshPeriod time.Duration) *LazyQueue[V] { + q := &LazyQueue[V]{ + popQueue: newSstack[V](nil, false), setIndex: setIndex, priority: priority, maxPriority: maxPriority, @@ -71,13 +71,13 @@ func NewLazyQueue(setIndex SetIndexCallback, priority PriorityCallback, maxPrior } // Reset clears the contents of the queue -func (q *LazyQueue) Reset() { - q.queue[0] = newSstack(q.setIndex0, false) - q.queue[1] = newSstack(q.setIndex1, false) +func (q *LazyQueue[V]) Reset() { + q.queue[0] = newSstack[V](q.setIndex0, false) + q.queue[1] = newSstack[V](q.setIndex1, false) } // Refresh performs queue re-evaluation if necessary -func (q *LazyQueue) Refresh() { +func (q *LazyQueue[V]) Refresh() { now := q.clock.Now() for time.Duration(now-q.lastRefresh2) >= q.period*2 { q.refresh(now) @@ -87,10 +87,10 @@ func (q *LazyQueue) Refresh() { } // refresh re-evaluates items in the older queue and swaps the two queues -func (q *LazyQueue) refresh(now mclock.AbsTime) { +func (q *LazyQueue[V]) refresh(now mclock.AbsTime) { q.maxUntil = now.Add(q.period) for q.queue[0].Len() != 0 { - q.Push(heap.Pop(q.queue[0]).(*item).value) + q.Push(heap.Pop(q.queue[0]).(*item[V]).value) } q.queue[0], q.queue[1] = q.queue[1], q.queue[0] q.indexOffset = 1 - q.indexOffset @@ -98,22 +98,22 @@ func (q *LazyQueue) refresh(now mclock.AbsTime) { } // Push adds an item to the queue -func (q *LazyQueue) Push(data interface{}) { - heap.Push(q.queue[1], &item{data, q.maxPriority(data, q.maxUntil)}) +func (q *LazyQueue[V]) Push(data V) { + heap.Push(q.queue[1], &item[V]{data, q.maxPriority(data, q.maxUntil)}) } // Update updates the upper priority estimate for the item with the given queue index -func (q *LazyQueue) Update(index int) { +func (q *LazyQueue[V]) Update(index int) { q.Push(q.Remove(index)) } // Pop removes and returns the item with the greatest actual priority -func (q *LazyQueue) Pop() (interface{}, int64) { +func (q *LazyQueue[V]) Pop() (V, int64) { var ( - resData interface{} + resData V resPri int64 ) - q.MultiPop(func(data interface{}, priority int64) bool { + q.MultiPop(func(data V, priority int64) bool { resData = data resPri = priority return false @@ -123,7 +123,7 @@ func (q *LazyQueue) Pop() (interface{}, int64) { // peekIndex returns the index of the internal queue where the item with the // highest estimated priority is or -1 if both are empty -func (q *LazyQueue) peekIndex() int { +func (q *LazyQueue[V]) peekIndex() int { if q.queue[0].Len() != 0 { if q.queue[1].Len() != 0 && q.queue[1].blocks[0][0].priority > q.queue[0].blocks[0][0].priority { return 1 @@ -139,17 +139,17 @@ func (q *LazyQueue) peekIndex() int { // MultiPop pops multiple items from the queue and is more efficient than calling // Pop multiple times. Popped items are passed to the callback. MultiPop returns // when the callback returns false or there are no more items to pop. -func (q *LazyQueue) MultiPop(callback func(data interface{}, priority int64) bool) { +func (q *LazyQueue[V]) MultiPop(callback func(data V, priority int64) bool) { nextIndex := q.peekIndex() for nextIndex != -1 { - data := heap.Pop(q.queue[nextIndex]).(*item).value - heap.Push(q.popQueue, &item{data, q.priority(data)}) + data := heap.Pop(q.queue[nextIndex]).(*item[V]).value + heap.Push(q.popQueue, &item[V]{data, q.priority(data)}) nextIndex = q.peekIndex() for q.popQueue.Len() != 0 && (nextIndex == -1 || q.queue[nextIndex].blocks[0][0].priority < q.popQueue.blocks[0][0].priority) { - i := heap.Pop(q.popQueue).(*item) + i := heap.Pop(q.popQueue).(*item[V]) if !callback(i.value, i.priority) { for q.popQueue.Len() != 0 { - q.Push(heap.Pop(q.popQueue).(*item).value) + q.Push(heap.Pop(q.popQueue).(*item[V]).value) } return } @@ -159,31 +159,28 @@ func (q *LazyQueue) MultiPop(callback func(data interface{}, priority int64) boo } // PopItem pops the item from the queue only, dropping the associated priority value. -func (q *LazyQueue) PopItem() interface{} { +func (q *LazyQueue[V]) PopItem() V { i, _ := q.Pop() return i } // Remove removes the item with the given index. -func (q *LazyQueue) Remove(index int) interface{} { - if index < 0 { - return nil - } - return heap.Remove(q.queue[index&1^q.indexOffset], index>>1).(*item).value +func (q *LazyQueue[V]) Remove(index int) V { + return heap.Remove(q.queue[index&1^q.indexOffset], index>>1).(*item[V]).value } // Empty checks whether the priority queue is empty. -func (q *LazyQueue) Empty() bool { +func (q *LazyQueue[V]) Empty() bool { return q.queue[0].Len() == 0 && q.queue[1].Len() == 0 } // Size returns the number of items in the priority queue. -func (q *LazyQueue) Size() int { +func (q *LazyQueue[V]) Size() int { return q.queue[0].Len() + q.queue[1].Len() } // setIndex0 translates internal queue item index to the virtual index space of LazyQueue -func (q *LazyQueue) setIndex0(data interface{}, index int) { +func (q *LazyQueue[V]) setIndex0(data V, index int) { if index == -1 { q.setIndex(data, -1) } else { @@ -192,6 +189,6 @@ func (q *LazyQueue) setIndex0(data interface{}, index int) { } // setIndex1 translates internal queue item index to the virtual index space of LazyQueue -func (q *LazyQueue) setIndex1(data interface{}, index int) { +func (q *LazyQueue[V]) setIndex1(data V, index int) { q.setIndex(data, index+index+1) } diff --git a/common/prque/prque.go b/common/prque/prque.go index fb02e3418c2..8ea06c2f7a0 100755 --- a/common/prque/prque.go +++ b/common/prque/prque.go @@ -22,62 +22,59 @@ import ( ) // Priority queue data structure. -type Prque struct { - cont *sstack +type Prque[V any] struct { + cont *sstack[V] } // New creates a new priority queue. -func New(setIndex SetIndexCallback) *Prque { - return &Prque{newSstack(setIndex, false)} +func New[V any](setIndex SetIndexCallback[V]) *Prque[V] { + return &Prque[V]{newSstack(setIndex, false)} } // NewWrapAround creates a new priority queue with wrap-around priority handling. -func NewWrapAround(setIndex SetIndexCallback) *Prque { - return &Prque{newSstack(setIndex, true)} +func NewWrapAround[V any](setIndex SetIndexCallback[V]) *Prque[V] { + return &Prque[V]{newSstack(setIndex, true)} } // Pushes a value with a given priority into the queue, expanding if necessary. -func (p *Prque) Push(data interface{}, priority int64) { - heap.Push(p.cont, &item{data, priority}) +func (p *Prque[V]) Push(data V, priority int64) { + heap.Push(p.cont, &item[V]{data, priority}) } // Peek returns the value with the greatest priority but does not pop it off. -func (p *Prque) Peek() (interface{}, int64) { +func (p *Prque[V]) Peek() (V, int64) { item := p.cont.blocks[0][0] return item.value, item.priority } // Pops the value with the greatest priority off the stack and returns it. // Currently no shrinking is done. -func (p *Prque) Pop() (interface{}, int64) { - item := heap.Pop(p.cont).(*item) +func (p *Prque[V]) Pop() (V, int64) { + item := heap.Pop(p.cont).(*item[V]) return item.value, item.priority } // Pops only the item from the queue, dropping the associated priority value. -func (p *Prque) PopItem() interface{} { - return heap.Pop(p.cont).(*item).value +func (p *Prque[V]) PopItem() V { + return heap.Pop(p.cont).(*item[V]).value } // Remove removes the element with the given index. -func (p *Prque) Remove(i int) interface{} { - if i < 0 { - return nil - } - return heap.Remove(p.cont, i) +func (p *Prque[V]) Remove(i int) V { + return heap.Remove(p.cont, i).(*item[V]).value } // Checks whether the priority queue is empty. -func (p *Prque) Empty() bool { +func (p *Prque[V]) Empty() bool { return p.cont.Len() == 0 } // Returns the number of element in the priority queue. -func (p *Prque) Size() int { +func (p *Prque[V]) Size() int { return p.cont.Len() } // Clears the contents of the priority queue. -func (p *Prque) Reset() { +func (p *Prque[V]) Reset() { *p = *New(p.cont.setIndex) } diff --git a/common/prque/prque_test.go b/common/prque/prque_test.go index 1cffcebad43..94ed684767b 100644 --- a/common/prque/prque_test.go +++ b/common/prque/prque_test.go @@ -21,7 +21,7 @@ func TestPrque(t *testing.T) { for i := 0; i < size; i++ { data[i] = rand.Int() } - queue := New(nil) + queue := New[int](nil) for rep := 0; rep < 2; rep++ { // Fill a priority queue with the above data for i := 0; i < size; i++ { @@ -59,7 +59,7 @@ func TestReset(t *testing.T) { for i := 0; i < size; i++ { data[i] = rand.Int() } - queue := New(nil) + queue := New[int](nil) for rep := 0; rep < 2; rep++ { // Fill a priority queue with the above data for i := 0; i < size; i++ { @@ -104,7 +104,7 @@ func BenchmarkPush(b *testing.B) { } // Execute the benchmark b.ResetTimer() - queue := New(nil) + queue := New[int](nil) for i := 0; i < len(data); i++ { queue.Push(data[i], prio[i]) } @@ -118,7 +118,7 @@ func BenchmarkPop(b *testing.B) { data[i] = rand.Int() prio[i] = rand.Int63() } - queue := New(nil) + queue := New[int](nil) for i := 0; i < len(data); i++ { queue.Push(data[i], prio[i]) } diff --git a/common/prque/sstack.go b/common/prque/sstack.go index b06a95413df..005f9797827 100755 --- a/common/prque/sstack.go +++ b/common/prque/sstack.go @@ -17,36 +17,36 @@ const blockSize = 4096 // // Note: priorities can "wrap around" the int64 range, a comes before b if (a.priority - b.priority) > 0. // The difference between the lowest and highest priorities in the queue at any point should be less than 2^63. -type item struct { - value interface{} +type item[V any] struct { + value V priority int64 } // SetIndexCallback is called when the element is moved to a new index. // Providing SetIndexCallback is optional, it is needed only if the application needs // to delete elements other than the top one. -type SetIndexCallback func(data interface{}, index int) +type SetIndexCallback[V any] func(data V, index int) // Internal sortable stack data structure. Implements the Push and Pop ops for // the stack (heap) functionality and the Len, Less and Swap methods for the // sortability requirements of the heaps. -type sstack struct { - setIndex SetIndexCallback +type sstack[V any] struct { + setIndex SetIndexCallback[V] size int capacity int offset int wrapAround bool - blocks [][]*item - active []*item + blocks [][]*item[V] + active []*item[V] } // Creates a new, empty stack. -func newSstack(setIndex SetIndexCallback, wrapAround bool) *sstack { - result := new(sstack) +func newSstack[V any](setIndex SetIndexCallback[V], wrapAround bool) *sstack[V] { + result := new(sstack[V]) result.setIndex = setIndex - result.active = make([]*item, blockSize) - result.blocks = [][]*item{result.active} + result.active = make([]*item[V], blockSize) + result.blocks = [][]*item[V]{result.active} result.capacity = blockSize result.wrapAround = wrapAround return result @@ -54,9 +54,9 @@ func newSstack(setIndex SetIndexCallback, wrapAround bool) *sstack { // Pushes a value onto the stack, expanding it if necessary. Required by // heap.Interface. -func (s *sstack) Push(data interface{}) { +func (s *sstack[V]) Push(data any) { if s.size == s.capacity { - s.active = make([]*item, blockSize) + s.active = make([]*item[V], blockSize) s.blocks = append(s.blocks, s.active) s.capacity += blockSize s.offset = 0 @@ -65,16 +65,16 @@ func (s *sstack) Push(data interface{}) { s.offset = 0 } if s.setIndex != nil { - s.setIndex(data.(*item).value, s.size) + s.setIndex(data.(*item[V]).value, s.size) } - s.active[s.offset] = data.(*item) + s.active[s.offset] = data.(*item[V]) s.offset++ s.size++ } // Pops a value off the stack and returns it. Currently no shrinking is done. // Required by heap.Interface. -func (s *sstack) Pop() (res interface{}) { +func (s *sstack[V]) Pop() (res any) { s.size-- s.offset-- if s.offset < 0 { @@ -83,19 +83,19 @@ func (s *sstack) Pop() (res interface{}) { } res, s.active[s.offset] = s.active[s.offset], nil if s.setIndex != nil { - s.setIndex(res.(*item).value, -1) + s.setIndex(res.(*item[V]).value, -1) } return } // Returns the length of the stack. Required by sort.Interface. -func (s *sstack) Len() int { +func (s *sstack[V]) Len() int { return s.size } // Compares the priority of two elements of the stack (higher is first). // Required by sort.Interface. -func (s *sstack) Less(i, j int) bool { +func (s *sstack[V]) Less(i, j int) bool { a, b := s.blocks[i/blockSize][i%blockSize].priority, s.blocks[j/blockSize][j%blockSize].priority if s.wrapAround { return a-b > 0 @@ -104,7 +104,7 @@ func (s *sstack) Less(i, j int) bool { } // Swaps two elements in the stack. Required by sort.Interface. -func (s *sstack) Swap(i, j int) { +func (s *sstack[V]) Swap(i, j int) { ib, io, jb, jo := i/blockSize, i%blockSize, j/blockSize, j%blockSize a, b := s.blocks[jb][jo], s.blocks[ib][io] if s.setIndex != nil { @@ -115,6 +115,6 @@ func (s *sstack) Swap(i, j int) { } // Resets the stack, effectively clearing its contents. -func (s *sstack) Reset() { +func (s *sstack[V]) Reset() { *s = *newSstack(s.setIndex, false) } diff --git a/common/prque/sstack_test.go b/common/prque/sstack_test.go index bc6298979cb..4818f7d5414 100644 --- a/common/prque/sstack_test.go +++ b/common/prque/sstack_test.go @@ -17,23 +17,23 @@ import ( func TestSstack(t *testing.T) { // Create some initial data size := 16 * blockSize - data := make([]*item, size) + data := make([]*item[int], size) for i := 0; i < size; i++ { - data[i] = &item{rand.Int(), rand.Int63()} + data[i] = &item[int]{rand.Int(), rand.Int63()} } - stack := newSstack(nil, false) + stack := newSstack[int](nil, false) for rep := 0; rep < 2; rep++ { // Push all the data into the stack, pop out every second - secs := []*item{} + secs := []*item[int]{} for i := 0; i < size; i++ { stack.Push(data[i]) if i%2 == 0 { - secs = append(secs, stack.Pop().(*item)) + secs = append(secs, stack.Pop().(*item[int])) } } - rest := []*item{} + rest := []*item[int]{} for stack.Len() > 0 { - rest = append(rest, stack.Pop().(*item)) + rest = append(rest, stack.Pop().(*item[int])) } // Make sure the contents of the resulting slices are ok for i := 0; i < size; i++ { @@ -50,12 +50,12 @@ func TestSstack(t *testing.T) { func TestSstackSort(t *testing.T) { // Create some initial data size := 16 * blockSize - data := make([]*item, size) + data := make([]*item[int], size) for i := 0; i < size; i++ { - data[i] = &item{rand.Int(), int64(i)} + data[i] = &item[int]{rand.Int(), int64(i)} } // Push all the data into the stack - stack := newSstack(nil, false) + stack := newSstack[int](nil, false) for _, val := range data { stack.Push(val) } @@ -72,18 +72,18 @@ func TestSstackSort(t *testing.T) { func TestSstackReset(t *testing.T) { // Create some initial data size := 16 * blockSize - data := make([]*item, size) + data := make([]*item[int], size) for i := 0; i < size; i++ { - data[i] = &item{rand.Int(), rand.Int63()} + data[i] = &item[int]{rand.Int(), rand.Int63()} } - stack := newSstack(nil, false) + stack := newSstack[int](nil, false) for rep := 0; rep < 2; rep++ { // Push all the data into the stack, pop out every second - secs := []*item{} + secs := []*item[int]{} for i := 0; i < size; i++ { stack.Push(data[i]) if i%2 == 0 { - secs = append(secs, stack.Pop().(*item)) + secs = append(secs, stack.Pop().(*item[int])) } } // Reset and verify both pulled and stack contents diff --git a/core/blockchain.go b/core/blockchain.go index 992e5a0f6b4..c3995a3df43 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -169,12 +169,12 @@ type BlockChain struct { chainConfig *params.ChainConfig // Chain & network configuration cacheConfig *CacheConfig // Cache configuration for pruning - db ethdb.Database // Low level persistent database to store final content in - snaps *snapshot.Tree // Snapshot tree for fast trie leaf access - triegc *prque.Prque // Priority queue mapping block numbers to tries to gc - gcproc time.Duration // Accumulates canonical block processing for trie dumping - triedb *trie.Database // The database handler for maintaining trie nodes. - stateCache state.Database // State database to reuse between imports (contains state cache) + db ethdb.Database // Low level persistent database to store final content in + snaps *snapshot.Tree // Snapshot tree for fast trie leaf access + triegc *prque.Prque[common.Hash] // Priority queue mapping block numbers to tries to gc + gcproc time.Duration // Accumulates canonical block processing for trie dumping + triedb *trie.Database // The database handler for maintaining trie nodes. + stateCache state.Database // State database to reuse between imports (contains state cache) // txLookupLimit is the maximum number of blocks from head whose tx indices // are reserved: @@ -258,7 +258,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, genesis *Genesis cacheConfig: cacheConfig, db: db, triedb: triedb, - triegc: prque.New(nil), + triegc: prque.New[common.Hash](nil), quit: make(chan struct{}), chainmu: syncx.NewClosableMutex(), bodyCache: lru.NewCache[common.Hash, *types.Body](bodyCacheLimit), @@ -922,7 +922,7 @@ func (bc *BlockChain) Stop() { } } for !bc.triegc.Empty() { - triedb.Dereference(bc.triegc.PopItem().(common.Hash)) + triedb.Dereference(bc.triegc.PopItem()) } if size, _ := triedb.Size(); size != 0 { log.Error("Dangling trie nodes after full cleanup") @@ -1354,7 +1354,7 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types. bc.triegc.Push(root, number) break } - bc.triedb.Dereference(root.(common.Hash)) + bc.triedb.Dereference(root) } } } diff --git a/core/rawdb/chain_iterator.go b/core/rawdb/chain_iterator.go index 121f6d39dda..214f2d2020d 100644 --- a/core/rawdb/chain_iterator.go +++ b/core/rawdb/chain_iterator.go @@ -191,7 +191,7 @@ func indexTransactions(db ethdb.Database, from uint64, to uint64, interrupt chan // in to be [to-1]. Therefore, setting lastNum to means that the // prqueue gap-evaluation will work correctly lastNum = to - queue = prque.New(nil) + queue = prque.New[*blockTxHashes](nil) // for stats reporting blocks, txs = 0, 0 ) @@ -210,7 +210,7 @@ func indexTransactions(db ethdb.Database, from uint64, to uint64, interrupt chan break } // Next block available, pop it off and index it - delivery := queue.PopItem().(*blockTxHashes) + delivery := queue.PopItem() lastNum = delivery.number WriteTxLookupEntries(batch, delivery.number, delivery.hashes) blocks++ @@ -282,7 +282,7 @@ func unindexTransactions(db ethdb.Database, from uint64, to uint64, interrupt ch // we expect the first number to come in to be [from]. Therefore, setting // nextNum to from means that the prqueue gap-evaluation will work correctly nextNum = from - queue = prque.New(nil) + queue = prque.New[*blockTxHashes](nil) // for stats reporting blocks, txs = 0, 0 ) @@ -299,7 +299,7 @@ func unindexTransactions(db ethdb.Database, from uint64, to uint64, interrupt ch if hook != nil && !hook(nextNum) { break } - delivery := queue.PopItem().(*blockTxHashes) + delivery := queue.PopItem() nextNum = delivery.number + 1 DeleteTxLookupEntries(batch, delivery.hashes) txs += len(delivery.hashes) diff --git a/core/txpool/txpool.go b/core/txpool/txpool.go index 8905140fdb6..22c9d123592 100644 --- a/core/txpool/txpool.go +++ b/core/txpool/txpool.go @@ -1388,7 +1388,7 @@ func (pool *TxPool) truncatePending() { pendingBeforeCap := pending // Assemble a spam order to penalize large transactors first - spammers := prque.New(nil) + spammers := prque.New[common.Address](nil) for addr, list := range pool.pending { // Only evict transactions from high rollers if !pool.locals.contains(addr) && uint64(list.Len()) > pool.config.AccountSlots { @@ -1400,12 +1400,12 @@ func (pool *TxPool) truncatePending() { for pending > pool.config.GlobalSlots && !spammers.Empty() { // Retrieve the next offender if not local address offender, _ := spammers.Pop() - offenders = append(offenders, offender.(common.Address)) + offenders = append(offenders, offender) // Equalize balances until all the same or below threshold if len(offenders) > 1 { // Calculate the equalization threshold for all current offenders - threshold := pool.pending[offender.(common.Address)].Len() + threshold := pool.pending[offender].Len() // Iteratively reduce all offenders until below limit or threshold reached for pending > pool.config.GlobalSlots && pool.pending[offenders[len(offenders)-2]].Len() > threshold { diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go index 60a83a7fb0d..0d44831351e 100644 --- a/eth/downloader/queue.go +++ b/eth/downloader/queue.go @@ -114,7 +114,7 @@ type queue struct { // Headers are "special", they download in batches, supported by a skeleton chain headerHead common.Hash // Hash of the last queued header to verify order headerTaskPool map[uint64]*types.Header // Pending header retrieval tasks, mapping starting indexes to skeleton headers - headerTaskQueue *prque.Prque // Priority queue of the skeleton indexes to fetch the filling headers for + headerTaskQueue *prque.Prque[uint64] // Priority queue of the skeleton indexes to fetch the filling headers for headerPeerMiss map[string]map[uint64]struct{} // Set of per-peer header batches known to be unavailable headerPendPool map[string]*fetchRequest // Currently pending header retrieval operations headerResults []*types.Header // Result cache accumulating the completed headers @@ -125,12 +125,12 @@ type queue struct { // All data retrievals below are based on an already assembles header chain blockTaskPool map[common.Hash]*types.Header // Pending block (body) retrieval tasks, mapping hashes to headers - blockTaskQueue *prque.Prque // Priority queue of the headers to fetch the blocks (bodies) for + blockTaskQueue *prque.Prque[*types.Header] // Priority queue of the headers to fetch the blocks (bodies) for blockPendPool map[string]*fetchRequest // Currently pending block (body) retrieval operations blockWakeCh chan bool // Channel to notify the block fetcher of new tasks receiptTaskPool map[common.Hash]*types.Header // Pending receipt retrieval tasks, mapping hashes to headers - receiptTaskQueue *prque.Prque // Priority queue of the headers to fetch the receipts for + receiptTaskQueue *prque.Prque[*types.Header] // Priority queue of the headers to fetch the receipts for receiptPendPool map[string]*fetchRequest // Currently pending receipt retrieval operations receiptWakeCh chan bool // Channel to notify when receipt fetcher of new tasks @@ -149,9 +149,9 @@ func newQueue(blockCacheLimit int, thresholdInitialSize int) *queue { lock := new(sync.RWMutex) q := &queue{ headerContCh: make(chan bool, 1), - blockTaskQueue: prque.New(nil), + blockTaskQueue: prque.New[*types.Header](nil), blockWakeCh: make(chan bool, 1), - receiptTaskQueue: prque.New(nil), + receiptTaskQueue: prque.New[*types.Header](nil), receiptWakeCh: make(chan bool, 1), active: sync.NewCond(lock), lock: lock, @@ -257,7 +257,7 @@ func (q *queue) ScheduleSkeleton(from uint64, skeleton []*types.Header) { } // Schedule all the header retrieval tasks for the skeleton assembly q.headerTaskPool = make(map[uint64]*types.Header) - q.headerTaskQueue = prque.New(nil) + q.headerTaskQueue = prque.New[uint64](nil) q.headerPeerMiss = make(map[string]map[uint64]struct{}) // Reset availability to correct invalid chains q.headerResults = make([]*types.Header, len(skeleton)*MaxHeaderFetch) q.headerHashes = make([]common.Hash, len(skeleton)*MaxHeaderFetch) @@ -427,12 +427,12 @@ func (q *queue) ReserveHeaders(p *peerConnection, count int) *fetchRequest { for send == 0 && !q.headerTaskQueue.Empty() { from, _ := q.headerTaskQueue.Pop() if q.headerPeerMiss[p.id] != nil { - if _, ok := q.headerPeerMiss[p.id][from.(uint64)]; ok { - skip = append(skip, from.(uint64)) + if _, ok := q.headerPeerMiss[p.id][from]; ok { + skip = append(skip, from) continue } } - send = from.(uint64) + send = from } // Merge all the skipped batches back for _, from := range skip { @@ -484,7 +484,7 @@ func (q *queue) ReserveReceipts(p *peerConnection, count int) (*fetchRequest, bo // item - the fetchRequest // progress - whether any progress was made // throttle - if the caller should throttle for a while -func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common.Hash]*types.Header, taskQueue *prque.Prque, +func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common.Hash]*types.Header, taskQueue *prque.Prque[*types.Header], pendPool map[string]*fetchRequest, kind uint) (*fetchRequest, bool, bool) { // Short circuit if the pool has been depleted, or if the peer's already // downloading something (sanity check not to corrupt state) @@ -502,8 +502,8 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common for proc := 0; len(send) < count && !taskQueue.Empty(); proc++ { // the task queue will pop items in order, so the highest prio block // is also the lowest block number. - h, _ := taskQueue.Peek() - header := h.(*types.Header) + header, _ := taskQueue.Peek() + // we can ask the resultcache if this header is within the // "prioritized" segment of blocks. If it is not, we need to throttle @@ -626,12 +626,14 @@ func (q *queue) ExpireReceipts(peer string) int { } // expire is the generic check that moves a specific expired task from a pending -// pool back into a task pool. +// pool back into a task pool. The syntax on the passed taskQueue is a bit weird +// as we would need a generic expire method to handle both types, but that is not +// supported at the moment at least (Go 1.19). // // Note, this method expects the queue lock to be already held. The reason the // lock is not obtained in here is that the parameters already need to access // the queue, so they already need a lock anyway. -func (q *queue) expire(peer string, pendPool map[string]*fetchRequest, taskQueue *prque.Prque) int { +func (q *queue) expire(peer string, pendPool map[string]*fetchRequest, taskQueue interface{}) int { // Retrieve the request being expired and log an error if it's non-existent, // as there's no order of events that should lead to such expirations. req := pendPool[peer] @@ -643,10 +645,10 @@ func (q *queue) expire(peer string, pendPool map[string]*fetchRequest, taskQueue // Return any non-satisfied requests to the pool if req.From > 0 { - taskQueue.Push(req.From, -int64(req.From)) + taskQueue.(*prque.Prque[uint64]).Push(req.From, -int64(req.From)) } for _, header := range req.Headers { - taskQueue.Push(header, -int64(header.Number.Uint64())) + taskQueue.(*prque.Prque[*types.Header]).Push(header, -int64(header.Number.Uint64())) } return len(req.Headers) } @@ -814,7 +816,7 @@ func (q *queue) DeliverReceipts(id string, receiptList [][]*types.Receipt, recei // reason this lock is not obtained in here is because the parameters already need // to access the queue, so they already need a lock anyway. func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, - taskQueue *prque.Prque, pendPool map[string]*fetchRequest, + taskQueue *prque.Prque[*types.Header], pendPool map[string]*fetchRequest, reqTimer metrics.Timer, resInMeter metrics.Meter, resDropMeter metrics.Meter, results int, validate func(index int, header *types.Header) error, reconstruct func(index int, result *fetchResult)) (int, error) { diff --git a/eth/fetcher/block_fetcher.go b/eth/fetcher/block_fetcher.go index bd1a34c83c0..41477232519 100644 --- a/eth/fetcher/block_fetcher.go +++ b/eth/fetcher/block_fetcher.go @@ -175,7 +175,7 @@ type BlockFetcher struct { completing map[common.Hash]*blockAnnounce // Blocks with headers, currently body-completing // Block cache - queue *prque.Prque // Queue containing the import operations (block number sorted) + queue *prque.Prque[*blockOrHeaderInject] // Queue containing the import operations (block number sorted) queues map[string]int // Per peer block counts to prevent memory exhaustion queued map[common.Hash]*blockOrHeaderInject // Set of already queued blocks (to dedup imports) @@ -212,7 +212,7 @@ func NewBlockFetcher(light bool, getHeader HeaderRetrievalFn, getBlock blockRetr fetching: make(map[common.Hash]*blockAnnounce), fetched: make(map[common.Hash][]*blockAnnounce), completing: make(map[common.Hash]*blockAnnounce), - queue: prque.New(nil), + queue: prque.New[*blockOrHeaderInject](nil), queues: make(map[string]int), queued: make(map[common.Hash]*blockOrHeaderInject), getHeader: getHeader, @@ -351,7 +351,7 @@ func (f *BlockFetcher) loop() { // Import any queued blocks that could potentially fit height := f.chainHeight() for !f.queue.Empty() { - op := f.queue.PopItem().(*blockOrHeaderInject) + op := f.queue.PopItem() hash := op.hash() if f.queueChangeHook != nil { f.queueChangeHook(hash, false) diff --git a/les/downloader/queue.go b/les/downloader/queue.go index 5b7054cf35c..4859261f7e4 100644 --- a/les/downloader/queue.go +++ b/les/downloader/queue.go @@ -115,7 +115,7 @@ type queue struct { // Headers are "special", they download in batches, supported by a skeleton chain headerHead common.Hash // Hash of the last queued header to verify order headerTaskPool map[uint64]*types.Header // Pending header retrieval tasks, mapping starting indexes to skeleton headers - headerTaskQueue *prque.Prque // Priority queue of the skeleton indexes to fetch the filling headers for + headerTaskQueue *prque.Prque[uint64] // Priority queue of the skeleton indexes to fetch the filling headers for headerPeerMiss map[string]map[uint64]struct{} // Set of per-peer header batches known to be unavailable headerPendPool map[string]*fetchRequest // Currently pending header retrieval operations headerResults []*types.Header // Result cache accumulating the completed headers @@ -125,11 +125,11 @@ type queue struct { // All data retrievals below are based on an already assembles header chain blockTaskPool map[common.Hash]*types.Header // Pending block (body) retrieval tasks, mapping hashes to headers - blockTaskQueue *prque.Prque // Priority queue of the headers to fetch the blocks (bodies) for + blockTaskQueue *prque.Prque[*types.Header] // Priority queue of the headers to fetch the blocks (bodies) for blockPendPool map[string]*fetchRequest // Currently pending block (body) retrieval operations receiptTaskPool map[common.Hash]*types.Header // Pending receipt retrieval tasks, mapping hashes to headers - receiptTaskQueue *prque.Prque // Priority queue of the headers to fetch the receipts for + receiptTaskQueue *prque.Prque[*types.Header] // Priority queue of the headers to fetch the receipts for receiptPendPool map[string]*fetchRequest // Currently pending receipt retrieval operations resultCache *resultStore // Downloaded but not yet delivered fetch results @@ -147,8 +147,8 @@ func newQueue(blockCacheLimit int, thresholdInitialSize int) *queue { lock := new(sync.RWMutex) q := &queue{ headerContCh: make(chan bool), - blockTaskQueue: prque.New(nil), - receiptTaskQueue: prque.New(nil), + blockTaskQueue: prque.New[*types.Header](nil), + receiptTaskQueue: prque.New[*types.Header](nil), active: sync.NewCond(lock), lock: lock, } @@ -262,7 +262,7 @@ func (q *queue) ScheduleSkeleton(from uint64, skeleton []*types.Header) { } // Schedule all the header retrieval tasks for the skeleton assembly q.headerTaskPool = make(map[uint64]*types.Header) - q.headerTaskQueue = prque.New(nil) + q.headerTaskQueue = prque.New[uint64](nil) q.headerPeerMiss = make(map[string]map[uint64]struct{}) // Reset availability to correct invalid chains q.headerResults = make([]*types.Header, len(skeleton)*MaxHeaderFetch) q.headerProced = 0 @@ -424,12 +424,12 @@ func (q *queue) ReserveHeaders(p *peerConnection, count int) *fetchRequest { for send == 0 && !q.headerTaskQueue.Empty() { from, _ := q.headerTaskQueue.Pop() if q.headerPeerMiss[p.id] != nil { - if _, ok := q.headerPeerMiss[p.id][from.(uint64)]; ok { - skip = append(skip, from.(uint64)) + if _, ok := q.headerPeerMiss[p.id][from]; ok { + skip = append(skip, from) continue } } - send = from.(uint64) + send = from } // Merge all the skipped batches back for _, from := range skip { @@ -481,7 +481,7 @@ func (q *queue) ReserveReceipts(p *peerConnection, count int) (*fetchRequest, bo // item - the fetchRequest // progress - whether any progress was made // throttle - if the caller should throttle for a while -func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common.Hash]*types.Header, taskQueue *prque.Prque, +func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common.Hash]*types.Header, taskQueue *prque.Prque[*types.Header], pendPool map[string]*fetchRequest, kind uint) (*fetchRequest, bool, bool) { // Short circuit if the pool has been depleted, or if the peer's already // downloading something (sanity check not to corrupt state) @@ -499,8 +499,8 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common for proc := 0; len(send) < count && !taskQueue.Empty(); proc++ { // the task queue will pop items in order, so the highest prio block // is also the lowest block number. - h, _ := taskQueue.Peek() - header := h.(*types.Header) + header, _ := taskQueue.Peek() + // we can ask the resultcache if this header is within the // "prioritized" segment of blocks. If it is not, we need to throttle @@ -591,12 +591,12 @@ func (q *queue) CancelReceipts(request *fetchRequest) { } // Cancel aborts a fetch request, returning all pending hashes to the task queue. -func (q *queue) cancel(request *fetchRequest, taskQueue *prque.Prque, pendPool map[string]*fetchRequest) { +func (q *queue) cancel(request *fetchRequest, taskQueue interface{}, pendPool map[string]*fetchRequest) { if request.From > 0 { - taskQueue.Push(request.From, -int64(request.From)) + taskQueue.(*prque.Prque[uint64]).Push(request.From, -int64(request.From)) } for _, header := range request.Headers { - taskQueue.Push(header, -int64(header.Number.Uint64())) + taskQueue.(*prque.Prque[*types.Header]).Push(header, -int64(header.Number.Uint64())) } delete(pendPool, request.Peer.id) } @@ -655,7 +655,7 @@ func (q *queue) ExpireReceipts(timeout time.Duration) map[string]int { // Note, this method expects the queue lock to be already held. The // reason the lock is not obtained in here is because the parameters already need // to access the queue, so they already need a lock anyway. -func (q *queue) expire(timeout time.Duration, pendPool map[string]*fetchRequest, taskQueue *prque.Prque, timeoutMeter metrics.Meter) map[string]int { +func (q *queue) expire(timeout time.Duration, pendPool map[string]*fetchRequest, taskQueue interface{}, timeoutMeter metrics.Meter) map[string]int { // Iterate over the expired requests and return each to the queue expiries := make(map[string]int) for id, request := range pendPool { @@ -665,10 +665,10 @@ func (q *queue) expire(timeout time.Duration, pendPool map[string]*fetchRequest, // Return any non satisfied requests to the pool if request.From > 0 { - taskQueue.Push(request.From, -int64(request.From)) + taskQueue.(*prque.Prque[uint64]).Push(request.From, -int64(request.From)) } for _, header := range request.Headers { - taskQueue.Push(header, -int64(header.Number.Uint64())) + taskQueue.(*prque.Prque[*types.Header]).Push(header, -int64(header.Number.Uint64())) } // Add the peer to the expiry report along the number of failed requests expiries[id] = len(request.Headers) @@ -831,7 +831,7 @@ func (q *queue) DeliverReceipts(id string, receiptList [][]*types.Receipt) (int, // reason this lock is not obtained in here is because the parameters already need // to access the queue, so they already need a lock anyway. func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, - taskQueue *prque.Prque, pendPool map[string]*fetchRequest, reqTimer metrics.Timer, + taskQueue *prque.Prque[*types.Header], pendPool map[string]*fetchRequest, reqTimer metrics.Timer, results int, validate func(index int, header *types.Header) error, reconstruct func(index int, result *fetchResult)) (int, error) { // Short circuit if the data was never requested diff --git a/les/fetcher/block_fetcher.go b/les/fetcher/block_fetcher.go index 42cf9500a2a..395b94d388b 100644 --- a/les/fetcher/block_fetcher.go +++ b/les/fetcher/block_fetcher.go @@ -177,7 +177,7 @@ type BlockFetcher struct { completing map[common.Hash]*blockAnnounce // Blocks with headers, currently body-completing // Block cache - queue *prque.Prque // Queue containing the import operations (block number sorted) + queue *prque.Prque[*blockOrHeaderInject] // Queue containing the import operations (block number sorted) queues map[string]int // Per peer block counts to prevent memory exhaustion queued map[common.Hash]*blockOrHeaderInject // Set of already queued blocks (to dedup imports) @@ -214,7 +214,7 @@ func NewBlockFetcher(light bool, getHeader HeaderRetrievalFn, getBlock blockRetr fetching: make(map[common.Hash]*blockAnnounce), fetched: make(map[common.Hash][]*blockAnnounce), completing: make(map[common.Hash]*blockAnnounce), - queue: prque.New(nil), + queue: prque.New[*blockOrHeaderInject](nil), queues: make(map[string]int), queued: make(map[common.Hash]*blockOrHeaderInject), getHeader: getHeader, @@ -353,7 +353,7 @@ func (f *BlockFetcher) loop() { // Import any queued blocks that could potentially fit height := f.chainHeight() for !f.queue.Empty() { - op := f.queue.PopItem().(*blockOrHeaderInject) + op := f.queue.PopItem() hash := op.hash() if f.queueChangeHook != nil { f.queueChangeHook(hash, false) diff --git a/les/flowcontrol/manager.go b/les/flowcontrol/manager.go index 497f91eeda7..3402ac71510 100644 --- a/les/flowcontrol/manager.go +++ b/les/flowcontrol/manager.go @@ -78,7 +78,7 @@ type ClientManager struct { // recharge queue is a priority queue with currently recharging client nodes // as elements. The priority value is rcFullIntValue which allows to quickly // determine which client will first finish recharge. - rcQueue *prque.Prque + rcQueue *prque.Prque[*ClientNode] } // NewClientManager returns a new client manager. @@ -107,7 +107,7 @@ type ClientManager struct { func NewClientManager(curve PieceWiseLinear, clock mclock.Clock) *ClientManager { cm := &ClientManager{ clock: clock, - rcQueue: prque.NewWrapAround(func(a interface{}, i int) { a.(*ClientNode).queueIndex = i }), + rcQueue: prque.NewWrapAround(func(a *ClientNode, i int) { a.queueIndex = i }), capLastUpdate: clock.Now(), stop: make(chan chan struct{}), } @@ -288,7 +288,7 @@ func (cm *ClientManager) updateRecharge(now mclock.AbsTime) { } dt := now - lastUpdate // fetch the client that finishes first - rcqNode := cm.rcQueue.PopItem().(*ClientNode) // if sumRecharge > 0 then the queue cannot be empty + rcqNode := cm.rcQueue.PopItem() // if sumRecharge > 0 then the queue cannot be empty // check whether it has already finished dtNext := mclock.AbsTime(float64(rcqNode.rcFullIntValue-cm.rcLastIntValue) / bonusRatio) if dt < dtNext { diff --git a/les/servingqueue.go b/les/servingqueue.go index 10c7e6f48cb..55a03452818 100644 --- a/les/servingqueue.go +++ b/les/servingqueue.go @@ -38,10 +38,10 @@ type servingQueue struct { setThreadsCh chan int wg sync.WaitGroup - threadCount int // number of currently running threads - queue *prque.Prque // priority queue for waiting or suspended tasks - best *servingTask // the highest priority task (not included in the queue) - suspendBias int64 // priority bias against suspending an already running task + threadCount int // number of currently running threads + queue *prque.Prque[*servingTask] // priority queue for waiting or suspended tasks + best *servingTask // the highest priority task (not included in the queue) + suspendBias int64 // priority bias against suspending an already running task } // servingTask represents a request serving task. Tasks can be implemented to @@ -123,7 +123,7 @@ func (t *servingTask) waitOrStop() bool { // newServingQueue returns a new servingQueue func newServingQueue(suspendBias int64, utilTarget float64) *servingQueue { sq := &servingQueue{ - queue: prque.NewWrapAround(nil), + queue: prque.NewWrapAround[*servingTask](nil), suspendBias: suspendBias, queueAddCh: make(chan *servingTask, 100), queueBestCh: make(chan *servingTask), @@ -214,7 +214,7 @@ func (sq *servingQueue) freezePeers() { } sq.best = nil for sq.queue.Size() > 0 { - task := sq.queue.PopItem().(*servingTask) + task := sq.queue.PopItem() tasks := peerMap[task.peer] if tasks == nil { bufValue, bufLimit := task.peer.fcClient.BufferStatus() @@ -251,7 +251,7 @@ func (sq *servingQueue) freezePeers() { } } if sq.queue.Size() > 0 { - sq.best = sq.queue.PopItem().(*servingTask) + sq.best = sq.queue.PopItem() } } @@ -310,7 +310,7 @@ func (sq *servingQueue) queueLoop() { if sq.queue.Size() == 0 { sq.best = nil } else { - sq.best, _ = sq.queue.PopItem().(*servingTask) + sq.best = sq.queue.PopItem() } case <-sq.quit: return diff --git a/les/vflux/server/prioritypool.go b/les/vflux/server/prioritypool.go index 059dac0d46d..f2bf39e0b24 100644 --- a/les/vflux/server/prioritypool.go +++ b/les/vflux/server/prioritypool.go @@ -77,8 +77,8 @@ type priorityPool struct { // temporary state if tempState is not empty tempState []*ppNodeInfo activeCount, activeCap uint64 - activeQueue *prque.LazyQueue - inactiveQueue *prque.Prque + activeQueue *prque.LazyQueue[*ppNodeInfo] + inactiveQueue *prque.Prque[*ppNodeInfo] } // ppNodeInfo is the internal node descriptor of priorityPool @@ -250,13 +250,13 @@ func (pp *priorityPool) Limits() (uint64, uint64) { } // inactiveSetIndex callback updates ppNodeInfo item index in inactiveQueue -func inactiveSetIndex(a interface{}, index int) { - a.(*ppNodeInfo).inactiveIndex = index +func inactiveSetIndex(a *ppNodeInfo, index int) { + a.inactiveIndex = index } // activeSetIndex callback updates ppNodeInfo item index in activeQueue -func activeSetIndex(a interface{}, index int) { - a.(*ppNodeInfo).activeIndex = index +func activeSetIndex(a *ppNodeInfo, index int) { + a.activeIndex = index } // invertPriority inverts a priority value. The active queue uses inverted priorities @@ -269,8 +269,7 @@ func invertPriority(p int64) int64 { } // activePriority callback returns actual priority of ppNodeInfo item in activeQueue -func activePriority(a interface{}) int64 { - c := a.(*ppNodeInfo) +func activePriority(c *ppNodeInfo) int64 { if c.bias == 0 { return invertPriority(c.nodePriority.priority(c.tempCapacity)) } else { @@ -279,8 +278,7 @@ func activePriority(a interface{}) int64 { } // activeMaxPriority callback returns estimated maximum priority of ppNodeInfo item in activeQueue -func (pp *priorityPool) activeMaxPriority(a interface{}, until mclock.AbsTime) int64 { - c := a.(*ppNodeInfo) +func (pp *priorityPool) activeMaxPriority(c *ppNodeInfo, until mclock.AbsTime) int64 { future := time.Duration(until - pp.clock.Now()) if future < 0 { future = 0 @@ -414,8 +412,7 @@ func (pp *priorityPool) enforceLimits() (*ppNodeInfo, int64) { c *ppNodeInfo maxActivePriority int64 ) - pp.activeQueue.MultiPop(func(data interface{}, priority int64) bool { - c = data.(*ppNodeInfo) + pp.activeQueue.MultiPop(func(c *ppNodeInfo, priority int64) bool { pp.setTempState(c) maxActivePriority = priority if c.tempCapacity == c.minTarget || pp.activeCount > pp.maxCount { @@ -496,7 +493,7 @@ func (pp *priorityPool) updateFlags(updates []capUpdate) { // tryActivate tries to activate inactive nodes if possible func (pp *priorityPool) tryActivate(commit bool) []capUpdate { for pp.inactiveQueue.Size() > 0 { - c := pp.inactiveQueue.PopItem().(*ppNodeInfo) + c := pp.inactiveQueue.PopItem() pp.setTempState(c) pp.setTempBias(c, pp.activeBias) pp.setTempCapacity(c, pp.minCap) diff --git a/trie/sync.go b/trie/sync.go index 19976698357..b9c6593815b 100644 --- a/trie/sync.go +++ b/trie/sync.go @@ -160,7 +160,7 @@ type Sync struct { membatch *syncMemBatch // Memory buffer to avoid frequent database writes nodeReqs map[string]*nodeRequest // Pending requests pertaining to a trie node path codeReqs map[common.Hash]*codeRequest // Pending requests pertaining to a code hash - queue *prque.Prque // Priority queue with the pending requests + queue *prque.Prque[any] // Priority queue with the pending requests fetches map[int]int // Number of active fetches per trie node depth } @@ -172,7 +172,7 @@ func NewSync(root common.Hash, database ethdb.KeyValueReader, callback LeafCallb membatch: newSyncMemBatch(), nodeReqs: make(map[string]*nodeRequest), codeReqs: make(map[common.Hash]*codeRequest), - queue: prque.New(nil), + queue: prque.New[any](nil), // Ugh, can contain both string and hash, whyyy fetches: make(map[int]int), } ts.AddSubTrie(root, nil, common.Hash{}, nil, callback) From 7028706895d57971b0d7a729bb65111c2a1e31e6 Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Thu, 1 Dec 2022 23:40:46 +0100 Subject: [PATCH 2/2] les/vflux/server: fixed issues in priorityPool --- les/vflux/server/prioritypool.go | 27 +++++++++++++++++---------- 1 file changed, 17 insertions(+), 10 deletions(-) diff --git a/les/vflux/server/prioritypool.go b/les/vflux/server/prioritypool.go index f2bf39e0b24..a09f0f3b2fd 100644 --- a/les/vflux/server/prioritypool.go +++ b/les/vflux/server/prioritypool.go @@ -183,8 +183,7 @@ func (pp *priorityPool) requestCapacity(node *enode.Node, minTarget, maxTarget u } pp.setTempCapacity(c, maxTarget) c.minTarget = minTarget - pp.activeQueue.Remove(c.activeIndex) - pp.inactiveQueue.Remove(c.inactiveIndex) + pp.removeFromQueues(c) pp.activeQueue.Push(c) pp.enforceLimits() updates := pp.finalizeChanges(c.tempCapacity >= minTarget && c.tempCapacity <= maxTarget && c.tempCapacity != c.capacity) @@ -291,6 +290,16 @@ func (pp *priorityPool) inactivePriority(p *ppNodeInfo) int64 { return p.nodePriority.priority(pp.minCap) } +// removeFromQueues removes the node from the active/inactive queues +func (pp *priorityPool) removeFromQueues(c *ppNodeInfo) { + if c.activeIndex >= 0 { + pp.activeQueue.Remove(c.activeIndex) + } + if c.inactiveIndex >= 0 { + pp.inactiveQueue.Remove(c.inactiveIndex) + } +} + // connectNode is called when a new node has been added to the pool (inactiveFlag set) // Note: this function should run inside a NodeStateMachine operation func (pp *priorityPool) connectNode(c *ppNodeInfo) { @@ -318,8 +327,7 @@ func (pp *priorityPool) disconnectNode(c *ppNodeInfo) { return } c.connected = false - pp.activeQueue.Remove(c.activeIndex) - pp.inactiveQueue.Remove(c.inactiveIndex) + pp.removeFromQueues(c) var updates []capUpdate if c.capacity != 0 { @@ -409,10 +417,11 @@ func (pp *priorityPool) enforceLimits() (*ppNodeInfo, int64) { return nil, math.MinInt64 } var ( - c *ppNodeInfo + lastNode *ppNodeInfo maxActivePriority int64 ) pp.activeQueue.MultiPop(func(c *ppNodeInfo, priority int64) bool { + lastNode = c pp.setTempState(c) maxActivePriority = priority if c.tempCapacity == c.minTarget || pp.activeCount > pp.maxCount { @@ -430,7 +439,7 @@ func (pp *priorityPool) enforceLimits() (*ppNodeInfo, int64) { } return pp.activeCap > pp.maxCap || pp.activeCount > pp.maxCount }) - return c, invertPriority(maxActivePriority) + return lastNode, invertPriority(maxActivePriority) } // finalizeChanges either commits or reverts temporary changes. The necessary capacity @@ -439,8 +448,7 @@ func (pp *priorityPool) enforceLimits() (*ppNodeInfo, int64) { func (pp *priorityPool) finalizeChanges(commit bool) (updates []capUpdate) { for _, c := range pp.tempState { // always remove and push back in order to update biased priority - pp.activeQueue.Remove(c.activeIndex) - pp.inactiveQueue.Remove(c.inactiveIndex) + pp.removeFromQueues(c) oldCapacity := c.capacity if commit { c.capacity = c.tempCapacity @@ -521,8 +529,7 @@ func (pp *priorityPool) updatePriority(node *enode.Node) { pp.lock.Unlock() return } - pp.activeQueue.Remove(c.activeIndex) - pp.inactiveQueue.Remove(c.inactiveIndex) + pp.removeFromQueues(c) if c.capacity != 0 { pp.activeQueue.Push(c) } else {