Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
les: implement client connection logic (#16899)
This PR implements les.freeClientPool. It also adds a simulated clock in common/mclock, which enables time-sensitive tests to run quickly and still produce accurate results, and package common/prque which is a generalised variant of prque that enables removing elements other than the top one from the queue. les.freeClientPool implements a client database that limits the connection time of each client and manages accepting/rejecting incoming connections and even kicking out some connected clients. The pool calculates recent usage time for each known client (a value that increases linearly when the client is connected and decreases exponentially when not connected). Clients with lower recent usage are preferred, unknown nodes have the highest priority. Already connected nodes receive a small bias in their favor in order to avoid accepting and instantly kicking out clients. Note: the pool can use any string for client identification. Using signature keys for that purpose would not make sense when being known has a negative value for the client. Currently the LES protocol manager uses IP addresses (without port address) to identify clients.
- Loading branch information
1 parent
a1783d1
commit b2ddb1f
Showing
7 changed files
with
761 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,129 @@ | ||
// Copyright 2018 The go-ethereum Authors | ||
// This file is part of the go-ethereum library. | ||
// | ||
// The go-ethereum library is free software: you can redistribute it and/or modify | ||
// it under the terms of the GNU Lesser General Public License as published by | ||
// the Free Software Foundation, either version 3 of the License, or | ||
// (at your option) any later version. | ||
// | ||
// The go-ethereum library is distributed in the hope that it will be useful, | ||
// but WITHOUT ANY WARRANTY; without even the implied warranty of | ||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the | ||
// GNU Lesser General Public License for more details. | ||
// | ||
// You should have received a copy of the GNU Lesser General Public License | ||
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>. | ||
|
||
package mclock | ||
|
||
import ( | ||
"sync" | ||
"time" | ||
) | ||
|
||
// Simulated implements a virtual Clock for reproducible time-sensitive tests. It | ||
// simulates a scheduler on a virtual timescale where actual processing takes zero time. | ||
// | ||
// The virtual clock doesn't advance on its own, call Run to advance it and execute timers. | ||
// Since there is no way to influence the Go scheduler, testing timeout behaviour involving | ||
// goroutines needs special care. A good way to test such timeouts is as follows: First | ||
// perform the action that is supposed to time out. Ensure that the timer you want to test | ||
// is created. Then run the clock until after the timeout. Finally observe the effect of | ||
// the timeout using a channel or semaphore. | ||
type Simulated struct { | ||
now AbsTime | ||
scheduled []event | ||
mu sync.RWMutex | ||
cond *sync.Cond | ||
} | ||
|
||
type event struct { | ||
do func() | ||
at AbsTime | ||
} | ||
|
||
// Run moves the clock by the given duration, executing all timers before that duration. | ||
func (s *Simulated) Run(d time.Duration) { | ||
s.mu.Lock() | ||
defer s.mu.Unlock() | ||
s.init() | ||
|
||
end := s.now + AbsTime(d) | ||
for len(s.scheduled) > 0 { | ||
ev := s.scheduled[0] | ||
if ev.at > end { | ||
break | ||
} | ||
s.now = ev.at | ||
ev.do() | ||
s.scheduled = s.scheduled[1:] | ||
} | ||
s.now = end | ||
} | ||
|
||
func (s *Simulated) ActiveTimers() int { | ||
s.mu.RLock() | ||
defer s.mu.RUnlock() | ||
|
||
return len(s.scheduled) | ||
} | ||
|
||
func (s *Simulated) WaitForTimers(n int) { | ||
s.mu.Lock() | ||
defer s.mu.Unlock() | ||
s.init() | ||
|
||
for len(s.scheduled) < n { | ||
s.cond.Wait() | ||
} | ||
} | ||
|
||
// Now implements Clock. | ||
func (s *Simulated) Now() AbsTime { | ||
s.mu.RLock() | ||
defer s.mu.RUnlock() | ||
|
||
return s.now | ||
} | ||
|
||
// Sleep implements Clock. | ||
func (s *Simulated) Sleep(d time.Duration) { | ||
<-s.After(d) | ||
} | ||
|
||
// After implements Clock. | ||
func (s *Simulated) After(d time.Duration) <-chan time.Time { | ||
after := make(chan time.Time, 1) | ||
s.insert(d, func() { | ||
after <- (time.Time{}).Add(time.Duration(s.now)) | ||
}) | ||
return after | ||
} | ||
|
||
func (s *Simulated) insert(d time.Duration, do func()) { | ||
s.mu.Lock() | ||
defer s.mu.Unlock() | ||
s.init() | ||
|
||
at := s.now + AbsTime(d) | ||
l, h := 0, len(s.scheduled) | ||
ll := h | ||
for l != h { | ||
m := (l + h) / 2 | ||
if at < s.scheduled[m].at { | ||
h = m | ||
} else { | ||
l = m + 1 | ||
} | ||
} | ||
s.scheduled = append(s.scheduled, event{}) | ||
copy(s.scheduled[l+1:], s.scheduled[l:ll]) | ||
s.scheduled[l] = event{do: do, at: at} | ||
s.cond.Broadcast() | ||
} | ||
|
||
func (s *Simulated) init() { | ||
if s.cond == nil { | ||
s.cond = sync.NewCond(&s.mu) | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,57 @@ | ||
// This is a duplicated and slightly modified version of "gopkg.in/karalabe/cookiejar.v2/collections/prque". | ||
|
||
package prque | ||
|
||
import ( | ||
"container/heap" | ||
) | ||
|
||
// Priority queue data structure. | ||
type Prque struct { | ||
cont *sstack | ||
} | ||
|
||
// Creates a new priority queue. | ||
func New(setIndex setIndexCallback) *Prque { | ||
return &Prque{newSstack(setIndex)} | ||
} | ||
|
||
// 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}) | ||
} | ||
|
||
// Pops the value with the greates priority off the stack and returns it. | ||
// Currently no shrinking is done. | ||
func (p *Prque) Pop() (interface{}, int64) { | ||
item := heap.Pop(p.cont).(*item) | ||
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 | ||
} | ||
|
||
// 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) | ||
} | ||
|
||
// Checks whether the priority queue is empty. | ||
func (p *Prque) Empty() bool { | ||
return p.cont.Len() == 0 | ||
} | ||
|
||
// Returns the number of element in the priority queue. | ||
func (p *Prque) Size() int { | ||
return p.cont.Len() | ||
} | ||
|
||
// Clears the contents of the priority queue. | ||
func (p *Prque) Reset() { | ||
*p = *New(p.cont.setIndex) | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,106 @@ | ||
// This is a duplicated and slightly modified version of "gopkg.in/karalabe/cookiejar.v2/collections/prque". | ||
|
||
package prque | ||
|
||
// The size of a block of data | ||
const blockSize = 4096 | ||
|
||
// A prioritized item in the sorted stack. | ||
// | ||
// 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{} | ||
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(a interface{}, i 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 | ||
size int | ||
capacity int | ||
offset int | ||
|
||
blocks [][]*item | ||
active []*item | ||
} | ||
|
||
// Creates a new, empty stack. | ||
func newSstack(setIndex setIndexCallback) *sstack { | ||
result := new(sstack) | ||
result.setIndex = setIndex | ||
result.active = make([]*item, blockSize) | ||
result.blocks = [][]*item{result.active} | ||
result.capacity = blockSize | ||
return result | ||
} | ||
|
||
// Pushes a value onto the stack, expanding it if necessary. Required by | ||
// heap.Interface. | ||
func (s *sstack) Push(data interface{}) { | ||
if s.size == s.capacity { | ||
s.active = make([]*item, blockSize) | ||
s.blocks = append(s.blocks, s.active) | ||
s.capacity += blockSize | ||
s.offset = 0 | ||
} else if s.offset == blockSize { | ||
s.active = s.blocks[s.size/blockSize] | ||
s.offset = 0 | ||
} | ||
if s.setIndex != nil { | ||
s.setIndex(data.(*item).value, s.size) | ||
} | ||
s.active[s.offset] = data.(*item) | ||
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{}) { | ||
s.size-- | ||
s.offset-- | ||
if s.offset < 0 { | ||
s.offset = blockSize - 1 | ||
s.active = s.blocks[s.size/blockSize] | ||
} | ||
res, s.active[s.offset] = s.active[s.offset], nil | ||
if s.setIndex != nil { | ||
s.setIndex(res.(*item).value, -1) | ||
} | ||
return | ||
} | ||
|
||
// Returns the length of the stack. Required by sort.Interface. | ||
func (s *sstack) 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 { | ||
return (s.blocks[i/blockSize][i%blockSize].priority - s.blocks[j/blockSize][j%blockSize].priority) > 0 | ||
} | ||
|
||
// Swaps two elements in the stack. Required by sort.Interface. | ||
func (s *sstack) 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 { | ||
s.setIndex(a.value, i) | ||
s.setIndex(b.value, j) | ||
} | ||
s.blocks[ib][io], s.blocks[jb][jo] = a, b | ||
} | ||
|
||
// Resets the stack, effectively clearing its contents. | ||
func (s *sstack) Reset() { | ||
*s = *newSstack(s.setIndex) | ||
} |
Oops, something went wrong.