multi: Add atomic start/stop functions.
Make sure that each subsystem only starts and stop once. This makes sure we don't close e.g. quit channels twice.
This commit is contained in:
4 files changed
+92
-18
No files matched your search
+32
-11
@@ -101,6 +101,9 @@ func (r *htlcReleaseEvent) Less(other queue.PriorityQueueItem) bool {
|
||||
// created by the daemon. The registry is a thin wrapper around a map in order
|
||||
// to ensure that all updates/reads are thread safe.
|
||||
type InvoiceRegistry struct {
|
||||
started atomic.Bool
|
||||
stopped atomic.Bool
|
||||
|
||||
sync.RWMutex
|
||||
|
||||
nextClientID uint32 // must be used atomically
|
||||
@@ -213,33 +216,48 @@ func (i *InvoiceRegistry) scanInvoicesOnStart(ctx context.Context) error {
|
||||
|
||||
// Start starts the registry and all goroutines it needs to carry out its task.
|
||||
func (i *InvoiceRegistry) Start() error {
|
||||
// Start InvoiceExpiryWatcher and prepopulate it with existing active
|
||||
// invoices.
|
||||
err := i.expiryWatcher.Start(func(hash lntypes.Hash, force bool) error {
|
||||
return i.cancelInvoiceImpl(context.Background(), hash, force)
|
||||
})
|
||||
var err error
|
||||
|
||||
log.Info("InvoiceRegistry starting...")
|
||||
|
||||
if i.started.Swap(true) {
|
||||
return fmt.Errorf("InvoiceRegistry started more than once")
|
||||
}
|
||||
// Start InvoiceExpiryWatcher and prepopulate it with existing
|
||||
// active invoices.
|
||||
err = i.expiryWatcher.Start(
|
||||
func(hash lntypes.Hash, force bool) error {
|
||||
return i.cancelInvoiceImpl(
|
||||
context.Background(), hash, force,
|
||||
)
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
log.Info("InvoiceRegistry starting")
|
||||
|
||||
i.wg.Add(1)
|
||||
go i.invoiceEventLoop()
|
||||
|
||||
// Now scan all pending and removable invoices to the expiry watcher or
|
||||
// delete them.
|
||||
// Now scan all pending and removable invoices to the expiry
|
||||
// watcher or delete them.
|
||||
err = i.scanInvoicesOnStart(context.Background())
|
||||
if err != nil {
|
||||
_ = i.Stop()
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
log.Debug("InvoiceRegistry started")
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
// Stop signals the registry for a graceful shutdown.
|
||||
func (i *InvoiceRegistry) Stop() error {
|
||||
log.Info("InvoiceRegistry shutting down...")
|
||||
|
||||
if i.stopped.Swap(true) {
|
||||
return fmt.Errorf("InvoiceRegistry stopped more than once")
|
||||
}
|
||||
|
||||
log.Info("InvoiceRegistry shutting down...")
|
||||
defer log.Debug("InvoiceRegistry shutdown complete")
|
||||
|
||||
@@ -248,6 +266,9 @@ func (i *InvoiceRegistry) Stop() error {
|
||||
close(i.quit)
|
||||
|
||||
i.wg.Wait()
|
||||
|
||||
log.Debug("InvoiceRegistry shutdown complete")
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user