lnrpc+wtclient: refactor ClientStats
This commit removes the mutex from ClientStats and instead puts that in clientStats which wraps ClientStats with a mutex. This is so that the tower client interface can return a ClientStats struct without worrying about copying a mutex.
This commit is contained in:
3 files changed
+34
-31
No files matched your search
@@ -413,8 +413,8 @@ func constructFunctionalOptions(includeSessions,
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Stats returns the in-memory statistics of the client since startup.
|
// Stats returns the in-memory statistics of the client since startup.
|
||||||
func (c *WatchtowerClient) Stats(ctx context.Context,
|
func (c *WatchtowerClient) Stats(_ context.Context,
|
||||||
req *StatsRequest) (*StatsResponse, error) {
|
_ *StatsRequest) (*StatsResponse, error) {
|
||||||
|
|
||||||
if err := c.isActive(); err != nil {
|
if err := c.isActive(); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
@@ -426,11 +426,7 @@ func (c *WatchtowerClient) Stats(ctx context.Context,
|
|||||||
}
|
}
|
||||||
|
|
||||||
var stats wtclient.ClientStats
|
var stats wtclient.ClientStats
|
||||||
for i := range clientStats {
|
for _, stat := range clientStats {
|
||||||
// Grab a reference to the slice index rather than copying bc
|
|
||||||
// ClientStats contains a lock which cannot be copied by value.
|
|
||||||
stat := &clientStats[i]
|
|
||||||
|
|
||||||
stats.NumTasksAccepted += stat.NumTasksAccepted
|
stats.NumTasksAccepted += stat.NumTasksAccepted
|
||||||
stats.NumTasksIneligible += stat.NumTasksIneligible
|
stats.NumTasksIneligible += stat.NumTasksIneligible
|
||||||
stats.NumTasksPending += stat.NumTasksPending
|
stats.NumTasksPending += stat.NumTasksPending
|
||||||
|
|||||||
@@ -200,7 +200,7 @@ type TowerClient struct {
|
|||||||
chanInfos wtdb.ChannelInfos
|
chanInfos wtdb.ChannelInfos
|
||||||
|
|
||||||
statTicker *time.Ticker
|
statTicker *time.Ticker
|
||||||
stats *ClientStats
|
stats *clientStats
|
||||||
|
|
||||||
newTowers chan *newTowerMsg
|
newTowers chan *newTowerMsg
|
||||||
staleTowers chan *staleTowerMsg
|
staleTowers chan *staleTowerMsg
|
||||||
@@ -245,7 +245,7 @@ func newTowerClient(cfg *towerClientCfg) (*TowerClient, error) {
|
|||||||
chanInfos: chanInfos,
|
chanInfos: chanInfos,
|
||||||
closableSessionQueue: newSessionCloseMinHeap(),
|
closableSessionQueue: newSessionCloseMinHeap(),
|
||||||
statTicker: time.NewTicker(DefaultStatInterval),
|
statTicker: time.NewTicker(DefaultStatInterval),
|
||||||
stats: new(ClientStats),
|
stats: new(clientStats),
|
||||||
newTowers: make(chan *newTowerMsg),
|
newTowers: make(chan *newTowerMsg),
|
||||||
staleTowers: make(chan *staleTowerMsg),
|
staleTowers: make(chan *staleTowerMsg),
|
||||||
quit: make(chan struct{}),
|
quit: make(chan struct{}),
|
||||||
@@ -1637,7 +1637,7 @@ func (c *TowerClient) LookupTower(pubKey *btcec.PublicKey,
|
|||||||
|
|
||||||
// Stats returns the in-memory statistics of the client since startup.
|
// Stats returns the in-memory statistics of the client since startup.
|
||||||
func (c *TowerClient) Stats() ClientStats {
|
func (c *TowerClient) Stats() ClientStats {
|
||||||
return c.stats.Copy()
|
return c.stats.getStatsCopy()
|
||||||
}
|
}
|
||||||
|
|
||||||
// Policy returns the active client policy configuration.
|
// Policy returns the active client policy configuration.
|
||||||
|
|||||||
@@ -8,8 +8,6 @@ import (
|
|||||||
// ClientStats is a collection of in-memory statistics of the actions the client
|
// ClientStats is a collection of in-memory statistics of the actions the client
|
||||||
// has performed since its creation.
|
// has performed since its creation.
|
||||||
type ClientStats struct {
|
type ClientStats struct {
|
||||||
mu sync.Mutex
|
|
||||||
|
|
||||||
// NumTasksPending is the total number of backups that are pending to
|
// NumTasksPending is the total number of backups that are pending to
|
||||||
// be acknowledged by all active and exhausted watchtower sessions.
|
// be acknowledged by all active and exhausted watchtower sessions.
|
||||||
NumTasksPending int
|
NumTasksPending int
|
||||||
@@ -31,68 +29,77 @@ type ClientStats struct {
|
|||||||
NumSessionsExhausted int
|
NumSessionsExhausted int
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// clientStats wraps ClientStats with a mutex so that it's members can be
|
||||||
|
// accessed in a thread safe manner.
|
||||||
|
type clientStats struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
|
||||||
|
ClientStats
|
||||||
|
}
|
||||||
|
|
||||||
// taskReceived increments the number of backup requests the client has received
|
// taskReceived increments the number of backup requests the client has received
|
||||||
// from active channels.
|
// from active channels.
|
||||||
func (s *ClientStats) taskReceived() {
|
func (s *clientStats) taskReceived() {
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
defer s.mu.Unlock()
|
defer s.mu.Unlock()
|
||||||
|
|
||||||
s.NumTasksPending++
|
s.NumTasksPending++
|
||||||
}
|
}
|
||||||
|
|
||||||
// taskAccepted increments the number of tasks that have been assigned to active
|
// taskAccepted increments the number of tasks that have been assigned to active
|
||||||
// session queues, and are awaiting upload to a tower.
|
// session queues, and are awaiting upload to a tower.
|
||||||
func (s *ClientStats) taskAccepted() {
|
func (s *clientStats) taskAccepted() {
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
defer s.mu.Unlock()
|
defer s.mu.Unlock()
|
||||||
|
|
||||||
s.NumTasksAccepted++
|
s.NumTasksAccepted++
|
||||||
s.NumTasksPending--
|
s.NumTasksPending--
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// getStatsCopy returns a copy of the ClientStats.
|
||||||
|
func (s *clientStats) getStatsCopy() ClientStats {
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
|
||||||
|
return s.ClientStats
|
||||||
|
}
|
||||||
|
|
||||||
// taskIneligible increments the number of tasks that were unable to satisfy the
|
// taskIneligible increments the number of tasks that were unable to satisfy the
|
||||||
// active session queue's policy. These can potentially be retried later, but
|
// active session queue's policy. These can potentially be retried later, but
|
||||||
// typically this means that the balance created dust outputs, so it may not be
|
// typically this means that the balance created dust outputs, so it may not be
|
||||||
// worth backing up at all.
|
// worth backing up at all.
|
||||||
func (s *ClientStats) taskIneligible() {
|
func (s *clientStats) taskIneligible() {
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
defer s.mu.Unlock()
|
defer s.mu.Unlock()
|
||||||
|
|
||||||
s.NumTasksIneligible++
|
s.NumTasksIneligible++
|
||||||
}
|
}
|
||||||
|
|
||||||
// sessionAcquired increments the number of sessions that have been successfully
|
// sessionAcquired increments the number of sessions that have been successfully
|
||||||
// negotiated by the client during this execution.
|
// negotiated by the client during this execution.
|
||||||
func (s *ClientStats) sessionAcquired() {
|
func (s *clientStats) sessionAcquired() {
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
defer s.mu.Unlock()
|
defer s.mu.Unlock()
|
||||||
|
|
||||||
s.NumSessionsAcquired++
|
s.NumSessionsAcquired++
|
||||||
}
|
}
|
||||||
|
|
||||||
// sessionExhausted increments the number of session that have become full as a
|
// sessionExhausted increments the number of session that have become full as a
|
||||||
// result of accepting backup tasks.
|
// result of accepting backup tasks.
|
||||||
func (s *ClientStats) sessionExhausted() {
|
func (s *clientStats) sessionExhausted() {
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
defer s.mu.Unlock()
|
defer s.mu.Unlock()
|
||||||
|
|
||||||
s.NumSessionsExhausted++
|
s.NumSessionsExhausted++
|
||||||
}
|
}
|
||||||
|
|
||||||
// String returns a human-readable summary of the client's metrics.
|
// String returns a human-readable summary of the client's metrics.
|
||||||
func (s *ClientStats) String() string {
|
func (s *clientStats) String() string {
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
defer s.mu.Unlock()
|
defer s.mu.Unlock()
|
||||||
|
|
||||||
return fmt.Sprintf("tasks(received=%d accepted=%d ineligible=%d) "+
|
return fmt.Sprintf("tasks(received=%d accepted=%d ineligible=%d) "+
|
||||||
"sessions(acquired=%d exhausted=%d)", s.NumTasksPending,
|
"sessions(acquired=%d exhausted=%d)", s.NumTasksPending,
|
||||||
s.NumTasksAccepted, s.NumTasksIneligible, s.NumSessionsAcquired,
|
s.NumTasksAccepted, s.NumTasksIneligible, s.NumSessionsAcquired,
|
||||||
s.NumSessionsExhausted)
|
s.NumSessionsExhausted)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Copy returns a copy of the current stats.
|
|
||||||
func (s *ClientStats) Copy() ClientStats {
|
|
||||||
s.mu.Lock()
|
|
||||||
defer s.mu.Unlock()
|
|
||||||
return ClientStats{
|
|
||||||
NumTasksPending: s.NumTasksPending,
|
|
||||||
NumTasksAccepted: s.NumTasksAccepted,
|
|
||||||
NumTasksIneligible: s.NumTasksIneligible,
|
|
||||||
NumSessionsAcquired: s.NumSessionsAcquired,
|
|
||||||
NumSessionsExhausted: s.NumSessionsExhausted,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
Reference in new issue
Block a user