From 58463cc5f4452eb0e4b92eb2bc0b47d13699d375 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/peer.go | 15 ++++---- les/{execqueue.go => utilities/exec_queue.go} | 37 ++++++++++--------- .../exec_queue_test.go} | 20 +++++----- 3 files changed, 36 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/peer.go b/les/peer.go index d308fd249e..579787313c 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" @@ -135,7 +136,7 @@ type peerCommons struct { headInfo blockInfo // Latest block information. // Background task queue for caching peer tasks and executing in order. - sendQueue *execQueue + sendQueue *utilities.ExecQueue // Flow control agreement. fcParams flowcontrol.ServerParams // The config for token bucket. @@ -153,13 +154,13 @@ func (p *peerCommons) isFrozen() bool { // canQueue returns an indicator whether the peer can queue a operation. func (p *peerCommons) canQueue() bool { - return p.sendQueue.canQueue() && !p.isFrozen() + return p.sendQueue.CanQueue() && !p.isFrozen() } // queueSend caches a peer operation in the background task queue. // Please ensure to check `canQueue` before call this function func (p *peerCommons) queueSend(f func()) bool { - return p.sendQueue.queue(f) + return p.sendQueue.Queue(f) } // mustQueueSend starts a for loop and retry the caching if failed. @@ -337,7 +338,7 @@ func (p *peerCommons) handshake(td *big.Int, head common.Hash, headNum uint64, g // close closes the channel and notifies all background routines to exit. func (p *peerCommons) close() { close(p.closeCh) - p.sendQueue.quit() + p.sendQueue.Quit() } // serverPeer represents each node to which the client is connected. @@ -375,7 +376,7 @@ func newServerPeer(version int, network uint64, trusted bool, p *p2p.Peer, rw p2 id: peerIdToString(p.ID()), version: version, network: network, - sendQueue: newExecQueue(100), + sendQueue: utilities.NewExecQueue(100), closeCh: make(chan struct{}), }, trusted: trusted, @@ -407,7 +408,7 @@ func (p *serverPeer) rejectUpdate(size uint64) bool { // frozen. func (p *serverPeer) freeze() { if atomic.CompareAndSwapUint32(&p.frozen, 0, 1) { - p.sendQueue.clear() + p.sendQueue.Clear() } } @@ -652,7 +653,7 @@ func newClientPeer(version int, network uint64, p *p2p.Peer, rw p2p.MsgReadWrite id: peerIdToString(p.ID()), version: version, network: network, - sendQueue: newExecQueue(100), + sendQueue: utilities.NewExecQueue(100), closeCh: make(chan struct{}), }, errCh: make(chan error, 1), 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) }