chainntnfs+queue: move ConcurrentQueue to own package 'queue'

chainntnfs/bitcoindnotify/bitcoind.go

@@ -14,6 +14,7 @@ import (
 	"github.com/btcsuite/btcwallet/chain"
 	"github.com/btcsuite/btcwallet/wtxmgr"
 	"github.com/lightningnetwork/lnd/chainntnfs"
+	"github.com/lightningnetwork/lnd/queue"
 )
 
 const (
@@ -996,7 +997,7 @@ type blockEpochRegistration struct {
 
 	epochChan chan *chainntnfs.BlockEpoch
 
-	epochQueue *chainntnfs.ConcurrentQueue
+	epochQueue *queue.ConcurrentQueue
 
 	bestBlock *chainntnfs.BlockEpoch
 
@@ -1021,7 +1022,7 @@ func (b *BitcoindNotifier) RegisterBlockEpochNtfn(
 	bestBlock *chainntnfs.BlockEpoch) (*chainntnfs.BlockEpochEvent, error) {
 
 	reg := &blockEpochRegistration{
-		epochQueue: chainntnfs.NewConcurrentQueue(20),
+		epochQueue: queue.NewConcurrentQueue(20),
 		epochChan:  make(chan *chainntnfs.BlockEpoch, 20),
 		cancelChan: make(chan struct{}),
 		epochID:    atomic.AddUint64(&b.epochClientCounter, 1),

chainntnfs/btcdnotify/btcd.go

@@ -14,6 +14,7 @@ import (
 	"github.com/btcsuite/btcd/wire"
 	"github.com/btcsuite/btcutil"
 	"github.com/lightningnetwork/lnd/chainntnfs"
+	"github.com/lightningnetwork/lnd/queue"
 )
 
 const (
@@ -81,8 +82,8 @@ type BtcdNotifier struct {
 
 	bestBlock chainntnfs.BlockEpoch
 
-	chainUpdates *chainntnfs.ConcurrentQueue
-	txUpdates    *chainntnfs.ConcurrentQueue
+	chainUpdates *queue.ConcurrentQueue
+	txUpdates    *queue.ConcurrentQueue
 
 	// spendHintCache is a cache used to query and update the latest height
 	// hints for an outpoint. Each height hint represents the earliest
@@ -115,8 +116,8 @@ func New(config *rpcclient.ConnConfig, spendHintCache chainntnfs.SpendHintCache,
 
 		spendNotifications: make(map[wire.OutPoint]map[uint64]*spendNotification),
 
-		chainUpdates: chainntnfs.NewConcurrentQueue(10),
-		txUpdates:    chainntnfs.NewConcurrentQueue(10),
+		chainUpdates: queue.NewConcurrentQueue(10),
+		txUpdates:    queue.NewConcurrentQueue(10),
 
 		spendHintCache:   spendHintCache,
 		confirmHintCache: confirmHintCache,
@@ -1056,7 +1057,7 @@ type blockEpochRegistration struct {
 
 	epochChan chan *chainntnfs.BlockEpoch
 
-	epochQueue *chainntnfs.ConcurrentQueue
+	epochQueue *queue.ConcurrentQueue
 
 	bestBlock *chainntnfs.BlockEpoch
 
@@ -1081,7 +1082,7 @@ func (b *BtcdNotifier) RegisterBlockEpochNtfn(
 	bestBlock *chainntnfs.BlockEpoch) (*chainntnfs.BlockEpochEvent, error) {
 
 	reg := &blockEpochRegistration{
-		epochQueue: chainntnfs.NewConcurrentQueue(20),
+		epochQueue: queue.NewConcurrentQueue(20),
 		epochChan:  make(chan *chainntnfs.BlockEpoch, 20),
 		cancelChan: make(chan struct{}),
 		epochID:    atomic.AddUint64(&b.epochClientCounter, 1),

chainntnfs/neutrinonotify/neutrino.go

@@ -18,6 +18,7 @@ import (
 	"github.com/btcsuite/btcwallet/waddrmgr"
 	"github.com/lightninglabs/neutrino"
 	"github.com/lightningnetwork/lnd/chainntnfs"
+	"github.com/lightningnetwork/lnd/queue"
 )
 
 const (
@@ -75,7 +76,7 @@ type NeutrinoNotifier struct {
 
 	rescanErr <-chan error
 
-	chainUpdates *chainntnfs.ConcurrentQueue
+	chainUpdates *queue.ConcurrentQueue
 
 	// spendHintCache is a cache used to query and update the latest height
 	// hints for an outpoint. Each height hint represents the earliest
@@ -114,7 +115,7 @@ func New(node *neutrino.ChainService, spendHintCache chainntnfs.SpendHintCache,
 
 		rescanErr: make(chan error),
 
-		chainUpdates: chainntnfs.NewConcurrentQueue(10),
+		chainUpdates: queue.NewConcurrentQueue(10),
 
 		spendHintCache:   spendHintCache,
 		confirmHintCache: confirmHintCache,
@@ -968,7 +969,7 @@ type blockEpochRegistration struct {
 
 	epochChan chan *chainntnfs.BlockEpoch
 
-	epochQueue *chainntnfs.ConcurrentQueue
+	epochQueue *queue.ConcurrentQueue
 
 	cancelChan chan struct{}
 
@@ -993,7 +994,7 @@ func (n *NeutrinoNotifier) RegisterBlockEpochNtfn(
 	bestBlock *chainntnfs.BlockEpoch) (*chainntnfs.BlockEpochEvent, error) {
 
 	reg := &blockEpochRegistration{
-		epochQueue: chainntnfs.NewConcurrentQueue(20),
+		epochQueue: queue.NewConcurrentQueue(20),
 		epochChan:  make(chan *chainntnfs.BlockEpoch, 20),
 		cancelChan: make(chan struct{}),
 		epochID:    atomic.AddUint64(&n.epochClientCounter, 1),

chainntnfs/queue.go

@@ -1,86 +0,0 @@
-package chainntnfs
-
-import (
-	"container/list"
-)
-
-// ConcurrentQueue is a concurrent-safe FIFO queue with unbounded capacity.
-// Clients interact with the queue by pushing items into the in channel and
-// popping items from the out channel. There is a goroutine that manages moving
-// items from the in channel to the out channel in the correct order that must
-// be started by calling Start().
-type ConcurrentQueue struct {
-	chanIn   chan interface{}
-	chanOut  chan interface{}
-	quit     chan struct{}
-	overflow *list.List
-}
-
-// NewConcurrentQueue constructs a ConcurrentQueue. The bufferSize parameter is
-// the capacity of the output channel. When the size of the queue is below this
-// threshold, pushes do not incur the overhead of the less efficient overflow
-// structure.
-func NewConcurrentQueue(bufferSize int) *ConcurrentQueue {
-	return &ConcurrentQueue{
-		chanIn:   make(chan interface{}),
-		chanOut:  make(chan interface{}, bufferSize),
-		quit:     make(chan struct{}),
-		overflow: list.New(),
-	}
-}
-
-// ChanIn returns a channel that can be used to push new items into the queue.
-func (cq *ConcurrentQueue) ChanIn() chan<- interface{} {
-	return cq.chanIn
-}
-
-// ChanOut returns a channel that can be used to pop items from the queue.
-func (cq *ConcurrentQueue) ChanOut() <-chan interface{} {
-	return cq.chanOut
-}
-
-// Start begins a goroutine that manages moving items from the in channel to the
-// out channel. The queue tries to move items directly to the out channel
-// minimize overhead, but if the out channel is full it pushes items to an
-// overflow queue. This must be called before using the queue.
-func (cq *ConcurrentQueue) Start() {
-	go func() {
-		for {
-			nextElement := cq.overflow.Front()
-			if nextElement == nil {
-				// Overflow queue is empty so incoming items can be pushed
-				// directly to the output channel. If output channel is full
-				// though, push to overflow.
-				select {
-				case item := <-cq.chanIn:
-					select {
-					case cq.chanOut <- item:
-						// Optimistically push directly to chanOut
-					default:
-						cq.overflow.PushBack(item)
-					}
-				case <-cq.quit:
-					return
-				}
-			} else {
-				// Overflow queue is not empty, so any new items get pushed to
-				// the back to preserve order.
-				select {
-				case item := <-cq.chanIn:
-					cq.overflow.PushBack(item)
-				case cq.chanOut <- nextElement.Value:
-					cq.overflow.Remove(nextElement)
-				case <-cq.quit:
-					return
-				}
-			}
-		}
-	}()
-}
-
-// Stop ends the goroutine that moves items from the in channel to the out
-// channel. This does not clear the queue state, so the queue can be restarted
-// without dropping items.
-func (cq *ConcurrentQueue) Stop() {
-	cq.quit <- struct{}{}
-}

chainntnfs/queue_test.go

@@ -1,26 +0,0 @@
-package chainntnfs_test
-
-import (
-	"testing"
-
-	"github.com/lightningnetwork/lnd/chainntnfs"
-)
-
-func TestConcurrentQueue(t *testing.T) {
-	queue := chainntnfs.NewConcurrentQueue(100)
-	queue.Start()
-	defer queue.Stop()
-
-	// Pushes should never block for long.
-	for i := 0; i < 1000; i++ {
-		queue.ChanIn() <- i
-	}
-
-	// Pops also should not block for long. Expect elements in FIFO order.
-	for i := 0; i < 1000; i++ {
-		item := <-queue.ChanOut()
-		if i != item.(int) {
-			t.Fatalf("Dequeued wrong value: expected %d, got %d", i, item.(int))
-		}
-	}
-}