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

@@ -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),

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 <http://www.gnu.org/licenses/>.
 
-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.
+// NewExecQueue creates a new execution Queue.
-func newExecQueue(capacity int) *execQueue {
+func NewExecQueue(capacity int) *ExecQueue {
-	q := &execQueue{funcs: make([]func(), 0, capacity)}
+	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.
+// CanQueue returns true if more function calls can be added to the execution Queue.
-func (q *execQueue) canQueue() bool {
+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.
+// Queue adds a function call to the execution Queue. Returns true if successful.
-func (q *execQueue) queue(f func()) bool {
+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
+// Clear drops all queued functions.
-func (q *execQueue) clear() {
+func (q *ExecQueue) Clear() {
 	q.mu.Lock()
 	q.funcs = q.funcs[:0]
 	q.mu.Unlock()
 }
 
-// quit stops the exec queue.
+// Quit stops the exec Queue.
-// quit waits for the current execution to finish before returning.
+//
-func (q *execQueue) quit() {
+// 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{})

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 <http://www.gnu.org/licenses/>.
 
-package les
+package utilities
 
-import (
+import "testing"
-	"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 {
+		if q.CanQueue() != wantOK {
-			t.Fatalf("canQueue() == %t for %s", !wantOK, state)
+			t.Fatalf("CanQueue() == %t for %s", !wantOK, state)
 		}
-		if q.queue(qf) != wantOK {
+		if q.Queue(qf) != wantOK {
-			t.Fatalf("canQueue() == %t for %s", !wantOK, state)
+			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)
 }