multi: replace manual CAS with sync.Once in component start/stop
This guarantees callers that the method will not return until it has executed completely at least once.
This commit is contained in:
parent
f802ebddba
commit
aea52f4bef
@ -6,7 +6,6 @@ import (
|
||||
"math/rand"
|
||||
"net"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/btcsuite/btcd/btcec"
|
||||
@ -105,9 +104,8 @@ func (c channelState) ConnectedNodes() map[NodeID]struct{} {
|
||||
//
|
||||
// TODO(roasbeef): prob re-word
|
||||
type Agent struct {
|
||||
// Only to be used atomically.
|
||||
started uint32
|
||||
stopped uint32
|
||||
started sync.Once
|
||||
stopped sync.Once
|
||||
|
||||
// cfg houses the configuration state of the Ant.
|
||||
cfg Config
|
||||
@ -197,10 +195,14 @@ func New(cfg Config, initialState []Channel) (*Agent, error) {
|
||||
// Start starts the agent along with any goroutines it needs to perform its
|
||||
// normal duties.
|
||||
func (a *Agent) Start() error {
|
||||
if !atomic.CompareAndSwapUint32(&a.started, 0, 1) {
|
||||
return nil
|
||||
}
|
||||
var err error
|
||||
a.started.Do(func() {
|
||||
err = a.start()
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
func (a *Agent) start() error {
|
||||
rand.Seed(time.Now().Unix())
|
||||
log.Infof("Autopilot Agent starting")
|
||||
|
||||
@ -213,10 +215,14 @@ func (a *Agent) Start() error {
|
||||
// Stop signals the Agent to gracefully shutdown. This function will block
|
||||
// until all goroutines have exited.
|
||||
func (a *Agent) Stop() error {
|
||||
if !atomic.CompareAndSwapUint32(&a.stopped, 0, 1) {
|
||||
return nil
|
||||
}
|
||||
var err error
|
||||
a.stopped.Do(func() {
|
||||
err = a.stop()
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
func (a *Agent) stop() error {
|
||||
log.Infof("Autopilot Agent stopping")
|
||||
|
||||
close(a.quit)
|
||||
|
@ -3,7 +3,6 @@ package autopilot
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/btcsuite/btcd/btcec"
|
||||
"github.com/lightningnetwork/lnd/lnwallet"
|
||||
@ -40,8 +39,8 @@ type ManagerCfg struct {
|
||||
// It implements the autopilot grpc service, which is used to get data about
|
||||
// the running autopilot, and give it relevant information.
|
||||
type Manager struct {
|
||||
started uint32 // To be used atomically.
|
||||
stopped uint32 // To be used atomically.
|
||||
started sync.Once
|
||||
stopped sync.Once
|
||||
|
||||
cfg *ManagerCfg
|
||||
|
||||
@ -64,27 +63,21 @@ func NewManager(cfg *ManagerCfg) (*Manager, error) {
|
||||
|
||||
// Start starts the Manager.
|
||||
func (m *Manager) Start() error {
|
||||
if !atomic.CompareAndSwapUint32(&m.started, 0, 1) {
|
||||
return nil
|
||||
}
|
||||
|
||||
m.started.Do(func() {})
|
||||
return nil
|
||||
}
|
||||
|
||||
// Stop stops the Manager. If an autopilot agent is active, it will also be
|
||||
// stopped.
|
||||
func (m *Manager) Stop() error {
|
||||
if !atomic.CompareAndSwapUint32(&m.stopped, 0, 1) {
|
||||
return nil
|
||||
}
|
||||
|
||||
if err := m.StopAgent(); err != nil {
|
||||
log.Errorf("Unable to stop pilot: %v", err)
|
||||
}
|
||||
|
||||
close(m.quit)
|
||||
m.wg.Wait()
|
||||
m.stopped.Do(func() {
|
||||
if err := m.StopAgent(); err != nil {
|
||||
log.Errorf("Unable to stop pilot: %v", err)
|
||||
}
|
||||
|
||||
close(m.quit)
|
||||
m.wg.Wait()
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
|
@ -7,7 +7,6 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/btcsuite/btcd/blockchain"
|
||||
"github.com/btcsuite/btcd/chaincfg/chainhash"
|
||||
@ -118,8 +117,8 @@ type BreachConfig struct {
|
||||
// counterparties.
|
||||
// TODO(roasbeef): closures in config for subsystem pointers to decouple?
|
||||
type breachArbiter struct {
|
||||
started uint32 // To be used atomically.
|
||||
stopped uint32 // To be used atomically.
|
||||
started sync.Once
|
||||
stopped sync.Once
|
||||
|
||||
cfg *BreachConfig
|
||||
|
||||
@ -140,10 +139,14 @@ func newBreachArbiter(cfg *BreachConfig) *breachArbiter {
|
||||
// Start is an idempotent method that officially starts the breachArbiter along
|
||||
// with all other goroutines it needs to perform its functions.
|
||||
func (b *breachArbiter) Start() error {
|
||||
if !atomic.CompareAndSwapUint32(&b.started, 0, 1) {
|
||||
return nil
|
||||
}
|
||||
var err error
|
||||
b.started.Do(func() {
|
||||
err = b.start()
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
func (b *breachArbiter) start() error {
|
||||
brarLog.Tracef("Starting breach arbiter")
|
||||
|
||||
// Load all retributions currently persisted in the retribution store.
|
||||
@ -226,15 +229,12 @@ func (b *breachArbiter) Start() error {
|
||||
// graceful shutdown. This function will block until all goroutines spawned by
|
||||
// the breachArbiter have gracefully exited.
|
||||
func (b *breachArbiter) Stop() error {
|
||||
if !atomic.CompareAndSwapUint32(&b.stopped, 0, 1) {
|
||||
return nil
|
||||
}
|
||||
|
||||
brarLog.Infof("Breach arbiter shutting down")
|
||||
|
||||
close(b.quit)
|
||||
b.wg.Wait()
|
||||
b.stopped.Do(func() {
|
||||
brarLog.Infof("Breach arbiter shutting down")
|
||||
|
||||
close(b.quit)
|
||||
b.wg.Wait()
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
|
@ -5,7 +5,6 @@ import (
|
||||
"fmt"
|
||||
"net"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/btcsuite/btcd/wire"
|
||||
"github.com/lightningnetwork/lnd/channeldb"
|
||||
@ -79,8 +78,8 @@ type ChannelNotifier interface {
|
||||
//
|
||||
// TODO(roasbeef): better name lol
|
||||
type SubSwapper struct {
|
||||
started uint32
|
||||
stopped uint32
|
||||
started sync.Once
|
||||
stopped sync.Once
|
||||
|
||||
// backupState are the set of SCBs for all open channels we know of.
|
||||
backupState map[wire.OutPoint]Single
|
||||
@ -135,29 +134,23 @@ func NewSubSwapper(startingChans []Single, chanNotifier ChannelNotifier,
|
||||
|
||||
// Start starts the chanbackup.SubSwapper.
|
||||
func (s *SubSwapper) Start() error {
|
||||
if !atomic.CompareAndSwapUint32(&s.started, 0, 1) {
|
||||
return nil
|
||||
}
|
||||
|
||||
log.Infof("Starting chanbackup.SubSwapper")
|
||||
|
||||
s.wg.Add(1)
|
||||
go s.backupUpdater()
|
||||
s.started.Do(func() {
|
||||
log.Infof("Starting chanbackup.SubSwapper")
|
||||
|
||||
s.wg.Add(1)
|
||||
go s.backupUpdater()
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
// Stop signals the SubSwapper to being a graceful shutdown.
|
||||
func (s *SubSwapper) Stop() error {
|
||||
if !atomic.CompareAndSwapUint32(&s.stopped, 0, 1) {
|
||||
return nil
|
||||
}
|
||||
|
||||
log.Infof("Stopping chanbackup.SubSwapper")
|
||||
|
||||
close(s.quit)
|
||||
s.wg.Wait()
|
||||
s.stopped.Do(func() {
|
||||
log.Infof("Stopping chanbackup.SubSwapper")
|
||||
|
||||
close(s.quit)
|
||||
s.wg.Wait()
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
|
@ -5,7 +5,6 @@ import (
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/btcsuite/btcd/btcec"
|
||||
@ -353,9 +352,8 @@ type fundingConfig struct {
|
||||
// controls between the wallet and remote peers are enforced via the funding
|
||||
// manager.
|
||||
type fundingManager struct {
|
||||
// MUST be used atomically.
|
||||
started int32
|
||||
stopped int32
|
||||
started sync.Once
|
||||
stopped sync.Once
|
||||
|
||||
// cfg is a copy of the configuration struct that the FundingManager
|
||||
// was initialized with.
|
||||
@ -471,10 +469,14 @@ func newFundingManager(cfg fundingConfig) (*fundingManager, error) {
|
||||
// Start launches all helper goroutines required for handling requests sent
|
||||
// to the funding manager.
|
||||
func (f *fundingManager) Start() error {
|
||||
if atomic.AddInt32(&f.started, 1) != 1 { // TODO(roasbeef): CAS instead
|
||||
return nil
|
||||
}
|
||||
var err error
|
||||
f.started.Do(func() {
|
||||
err = f.start()
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
func (f *fundingManager) start() error {
|
||||
fndgLog.Tracef("Funding manager running")
|
||||
|
||||
// Upon restart, the Funding Manager will check the database to load any
|
||||
@ -710,10 +712,14 @@ func (f *fundingManager) Start() error {
|
||||
// Stop signals all helper goroutines to execute a graceful shutdown. This
|
||||
// method will block until all goroutines have exited.
|
||||
func (f *fundingManager) Stop() error {
|
||||
if atomic.AddInt32(&f.stopped, 1) != 1 {
|
||||
return nil
|
||||
}
|
||||
var err error
|
||||
f.stopped.Do(func() {
|
||||
err = f.stop()
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
func (f *fundingManager) stop() error {
|
||||
fndgLog.Infof("Funding manager shutting down")
|
||||
|
||||
close(f.quit)
|
||||
|
Loading…
Reference in New Issue
Block a user