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)
}