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

@@ -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{}

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

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

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