From 9a9b2de7fae55674cfee16173063d170f8f7534d Mon Sep 17 00:00:00 2001 From: rjl493456442 Date: Mon, 6 Jan 2020 15:18:21 +0800 Subject: [PATCH] les: move execqueue into utilities package execqueue is a util for executing queued functions in a serial order which is used by both les server and les client. Move it to common package. --- les/benchmark.go | 3 +- les/peer.go | 13 ++++--- les/{execqueue.go => utilities/exec_queue.go} | 37 ++++++++++--------- .../exec_queue_test.go} | 20 +++++----- 4 files changed, 37 insertions(+), 36 deletions(-) rename les/{execqueue.go => utilities/exec_queue.go} (71%) rename les/{execqueue_test.go => utilities/exec_queue_test.go} (83%) diff --git a/les/benchmark.go b/les/benchmark.go index 42eeef10f3..993cf7df15 100644 --- a/les/benchmark.go +++ b/les/benchmark.go @@ -30,6 +30,7 @@ import ( "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/les/flowcontrol" + "github.com/ethereum/go-ethereum/les/utilities" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/enode" @@ -285,7 +286,7 @@ func (h *serverHandler) measure(setup *benchmarkSetup, count int) error { clientPeer := newPeer(lpv2, NetworkId, false, p2p.NewPeer(id, "client", nil), clientMeteredPipe) serverPeer := newPeer(lpv2, NetworkId, false, p2p.NewPeer(id, "server", nil), serverMeteredPipe) - serverPeer.sendQueue = newExecQueue(count) + serverPeer.sendQueue = utilities.NewExecQueue(count) serverPeer.announceType = announceTypeNone serverPeer.fcCosts = make(requestCostTable) c := &requestCosts{} diff --git a/les/peer.go b/les/peer.go index feb3910beb..e1bf273524 100644 --- a/les/peer.go +++ b/les/peer.go @@ -32,6 +32,7 @@ import ( "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/eth" "github.com/ethereum/go-ethereum/les/flowcontrol" + "github.com/ethereum/go-ethereum/les/utilities" "github.com/ethereum/go-ethereum/light" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/enode" @@ -91,7 +92,7 @@ type peer struct { headInfo *announceData lock sync.RWMutex - sendQueue *execQueue + sendQueue *utilities.ExecQueue errCh chan error @@ -213,7 +214,7 @@ func (p *peer) freezeServer(frozen bool) { f = 1 } if atomic.SwapUint32(&p.frozen, f) != f && frozen { - p.sendQueue.clear() + p.sendQueue.Clear() } } @@ -224,11 +225,11 @@ func (p *peer) isFrozen() bool { } func (p *peer) canQueue() bool { - return p.sendQueue.canQueue() && !p.isFrozen() + return p.sendQueue.CanQueue() && !p.isFrozen() } func (p *peer) queueSend(f func()) { - p.sendQueue.queue(f) + p.sendQueue.Queue(f) } // Info gathers and returns a collection of metadata known about a peer. @@ -817,7 +818,7 @@ func (ps *peerSet) Register(p *peer) error { return errAlreadyRegistered } ps.peers[p.id] = p - p.sendQueue = newExecQueue(100) + p.sendQueue = utilities.NewExecQueue(100) peers := make([]peerSetNotify, len(ps.notifyList)) copy(peers, ps.notifyList) ps.lock.Unlock() @@ -845,7 +846,7 @@ func (ps *peerSet) Unregister(id string) error { n.unregisterPeer(p) } - p.sendQueue.quit() + p.sendQueue.Quit() p.Peer.Disconnect(p2p.DiscUselessPeer) return nil diff --git a/les/execqueue.go b/les/utilities/exec_queue.go similarity index 71% rename from les/execqueue.go rename to les/utilities/exec_queue.go index e0c88a990f..8a77178051 100644 --- a/les/execqueue.go +++ b/les/utilities/exec_queue.go @@ -14,35 +14,35 @@ // You should have received a copy of the GNU Lesser General Public License // along with the go-ethereum library. If not, see . -package les +package utilities import "sync" -// execQueue implements a queue that executes function calls in a single thread, +// ExecQueue implements a queue that executes function calls in a single thread, // in the same order as they have been queued. -type execQueue struct { +type ExecQueue struct { mu sync.Mutex cond *sync.Cond funcs []func() closeWait chan struct{} } -// newExecQueue creates a new execution queue. -func newExecQueue(capacity int) *execQueue { - q := &execQueue{funcs: make([]func(), 0, capacity)} +// NewExecQueue creates a new execution Queue. +func NewExecQueue(capacity int) *ExecQueue { + q := &ExecQueue{funcs: make([]func(), 0, capacity)} q.cond = sync.NewCond(&q.mu) go q.loop() return q } -func (q *execQueue) loop() { +func (q *ExecQueue) loop() { for f := q.waitNext(false); f != nil; f = q.waitNext(true) { f() } close(q.closeWait) } -func (q *execQueue) waitNext(drop bool) (f func()) { +func (q *ExecQueue) waitNext(drop bool) (f func()) { q.mu.Lock() if drop && len(q.funcs) > 0 { // Remove the function that just executed. We do this here instead of when @@ -60,20 +60,20 @@ func (q *execQueue) waitNext(drop bool) (f func()) { return f } -func (q *execQueue) isClosed() bool { +func (q *ExecQueue) isClosed() bool { return q.closeWait != nil } -// canQueue returns true if more function calls can be added to the execution queue. -func (q *execQueue) canQueue() bool { +// CanQueue returns true if more function calls can be added to the execution Queue. +func (q *ExecQueue) CanQueue() bool { q.mu.Lock() ok := !q.isClosed() && len(q.funcs) < cap(q.funcs) q.mu.Unlock() return ok } -// queue adds a function call to the execution queue. Returns true if successful. -func (q *execQueue) queue(f func()) bool { +// Queue adds a function call to the execution Queue. Returns true if successful. +func (q *ExecQueue) Queue(f func()) bool { q.mu.Lock() ok := !q.isClosed() && len(q.funcs) < cap(q.funcs) if ok { @@ -84,16 +84,17 @@ func (q *execQueue) queue(f func()) bool { return ok } -// clear drops all queued functions -func (q *execQueue) clear() { +// Clear drops all queued functions. +func (q *ExecQueue) Clear() { q.mu.Lock() q.funcs = q.funcs[:0] q.mu.Unlock() } -// quit stops the exec queue. -// quit waits for the current execution to finish before returning. -func (q *execQueue) quit() { +// Quit stops the exec Queue. +// +// Quit waits for the current execution to finish before returning. +func (q *ExecQueue) Quit() { q.mu.Lock() if !q.isClosed() { q.closeWait = make(chan struct{}) diff --git a/les/execqueue_test.go b/les/utilities/exec_queue_test.go similarity index 83% rename from les/execqueue_test.go rename to les/utilities/exec_queue_test.go index cd45b03f22..81a04c381d 100644 --- a/les/execqueue_test.go +++ b/les/utilities/exec_queue_test.go @@ -14,21 +14,19 @@ // You should have received a copy of the GNU Lesser General Public License // along with the go-ethereum library. If not, see . -package les +package utilities -import ( - "testing" -) +import "testing" func TestExecQueue(t *testing.T) { var ( N = 10000 - q = newExecQueue(N) + q = NewExecQueue(N) counter int execd = make(chan int) testexit = make(chan struct{}) ) - defer q.quit() + defer q.Quit() defer close(testexit) check := func(state string, wantOK bool) { @@ -40,11 +38,11 @@ func TestExecQueue(t *testing.T) { case <-testexit: } } - if q.canQueue() != wantOK { - t.Fatalf("canQueue() == %t for %s", !wantOK, state) + if q.CanQueue() != wantOK { + t.Fatalf("CanQueue() == %t for %s", !wantOK, state) } - if q.queue(qf) != wantOK { - t.Fatalf("canQueue() == %t for %s", !wantOK, state) + if q.Queue(qf) != wantOK { + t.Fatalf("Queue() == %t for %s", !wantOK, state) } } @@ -57,6 +55,6 @@ func TestExecQueue(t *testing.T) { t.Fatal("execution out of order") } } - q.quit() + q.Quit() check("closed queue", false) }