diff --git a/queue/queue.go b/queue/queue.go new file mode 100644 index 00000000..597e02e3 --- /dev/null +++ b/queue/queue.go @@ -0,0 +1,132 @@ +package queue + +import ( + "container/list" + "sync" +) + +const ( + // DefaultQueueSize is the default size to use for concurrent queues. + DefaultQueueSize = 10 +) + +// ConcurrentQueue is a typed 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[T any] struct { + started sync.Once + stopped sync.Once + + chanIn chan T + chanOut chan T + overflow *list.List + + wg sync.WaitGroup + quit chan struct{} +} + +// 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[T any](bufferSize int) *ConcurrentQueue[T] { + return &ConcurrentQueue[T]{ + chanIn: make(chan T), + chanOut: make(chan T, bufferSize), + overflow: list.New(), + quit: make(chan struct{}), + } +} + +// ChanIn returns a channel that can be used to push new items into the queue. +func (cq *ConcurrentQueue[T]) ChanIn() chan<- T { + return cq.chanIn +} + +// ChanOut returns a channel that can be used to pop items from the queue. +func (cq *ConcurrentQueue[T]) ChanOut() <-chan T { + 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[T]) Start() { + cq.started.Do(cq.start) +} + +func (cq *ConcurrentQueue[T]) start() { + cq.wg.Add(1) + go func() { + defer cq.wg.Done() + + readLoop: + 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, ok := <-cq.chanIn: + if !ok { + break readLoop + } + 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, ok := <-cq.chanIn: + if !ok { + break readLoop + } + cq.overflow.PushBack(item) + case cq.chanOut <- nextElement.Value.(T): + cq.overflow.Remove(nextElement) + case <-cq.quit: + return + } + } + } + + // Incoming channel has been closed. Empty overflow queue into + // the outgoing channel. + nextElement := cq.overflow.Front() + for nextElement != nil { + select { + case cq.chanOut <- nextElement.Value.(T): + cq.overflow.Remove(nextElement) + case <-cq.quit: + return + } + nextElement = cq.overflow.Front() + } + + // Close outgoing channel. + close(cq.chanOut) + }() +} + +// 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[T]) Stop() { + cq.stopped.Do(func() { + close(cq.quit) + cq.wg.Wait() + }) +} diff --git a/terminal.go b/terminal.go index 73f892e6..013def75 100644 --- a/terminal.go +++ b/terminal.go @@ -22,6 +22,7 @@ import ( "github.com/lightninglabs/faraday/frdrpc" "github.com/lightninglabs/faraday/frdrpcserver" "github.com/lightninglabs/lightning-terminal/litrpc" + "github.com/lightninglabs/lightning-terminal/queue" mid "github.com/lightninglabs/lightning-terminal/rpcmiddleware" "github.com/lightninglabs/lightning-terminal/session" "github.com/lightninglabs/lndclient" @@ -198,6 +199,13 @@ func (g *LightningTerminal) Run() error { // Show version at startup. log.Infof("LiT version: %s", Version()) + // This concurrent error queue can be used by every component that can + // raise runtime errors. Using a queue will prevent us from blocking on + // sending errors to it, as long as the queue is running. + errQueue := queue.NewConcurrentQueue[error](queue.DefaultQueueSize) + errQueue.Start() + defer errQueue.Stop() + // Construct a new PermissionsManager. g.permsMgr, err = NewPermissionsManager() if err != nil {