Merge pull request #1668 from cfromknecht/interface-tickers

Ticker Package

htlcswitch/link.go

@@ -18,6 +18,7 @@ import (
 	"github.com/lightningnetwork/lnd/lnpeer"
 	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/lightningnetwork/lnd/ticker"
 )
 
 func init() {
@@ -94,41 +95,6 @@ func ExpectedFee(f ForwardingPolicy,
 	return f.BaseFee + (htlcAmt*f.FeeRate)/1000000
 }
 
-// Ticker is an interface used to wrap a time.Ticker in a struct, making
-// mocking it easier.
-type Ticker interface {
-	Start() <-chan time.Time
-
-	Stop()
-}
-
-// BatchTicker implements the Ticker interface, and wraps a time.Ticker.
-type BatchTicker struct {
-	duration time.Duration
-	ticker   *time.Ticker
-}
-
-// NewBatchTicker returns a new BatchTicker that wraps the passed time.Ticker.
-func NewBatchTicker(d time.Duration) *BatchTicker {
-	return &BatchTicker{
-		duration: d,
-	}
-}
-
-// Start returns the tick channel for the underlying time.Ticker.
-func (t *BatchTicker) Start() <-chan time.Time {
-	t.ticker = time.NewTicker(t.duration)
-	return t.ticker.C
-}
-
-// Stop stops the underlying time.Ticker.
-func (t *BatchTicker) Stop() {
-	if t.ticker != nil {
-		t.ticker.Stop()
-		t.ticker = nil
-	}
-}
-
 // ChannelLinkConfig defines the configuration for the channel link. ALL
 // elements within the configuration MUST be non-nil for channel link to carry
 // out its duties.
@@ -236,13 +202,13 @@ type ChannelLinkConfig struct {
 	// flush out. By batching updates into a single commit, we attempt to
 	// increase throughput by maximizing the number of updates coalesced
 	// into a single commit.
-	BatchTicker Ticker
+	BatchTicker ticker.Ticker
 
 	// FwdPkgGCTicker is the ticker determining the frequency at which
 	// garbage collection of forwarding packages occurs. We use a
 	// time-based approach, as opposed to block epochs, as to not hinder
 	// syncing.
-	FwdPkgGCTicker Ticker
+	FwdPkgGCTicker ticker.Ticker
 
 	// BatchSize is the max size of a batch of updates done to the link
 	// before we do a state update.
@@ -739,12 +705,12 @@ func (l *channelLink) resolveFwdPkg(fwdPkg *channeldb.FwdPkg) (bool, error) {
 func (l *channelLink) fwdPkgGarbager() {
 	defer l.wg.Done()
 
-	fwdPkgGcTick := l.cfg.FwdPkgGCTicker.Start()
+	l.cfg.FwdPkgGCTicker.Resume()
 	defer l.cfg.FwdPkgGCTicker.Stop()
 
 	for {
 		select {
-		case <-fwdPkgGcTick:
+		case <-l.cfg.FwdPkgGCTicker.Ticks():
 			fwdPkgs, err := l.channel.LoadFwdPkgs()
 			if err != nil {
 				l.warnf("unable to load fwdpkgs for gc: %v", err)
@@ -890,13 +856,6 @@ func (l *channelLink) htlcManager() {
 	l.wg.Add(1)
 	go l.fwdPkgGarbager()
 
-	// We'll only need the batch ticker if we have outgoing updates that are
-	// not covered by our last signature. This value will be nil unless a
-	// downstream packet forces the batchCounter to be positive. After the
-	// batch is cleared, it will return to nil to prevent wasteful CPU time
-	// caused by the batch ticker waking up the htlcManager needlessly.
-	var maybeBatchTick <-chan time.Time
-
 out:
 	for {
 		// We must always check if we failed at some point processing
@@ -977,13 +936,12 @@ out:
 				break out
 			}
 
-		case <-maybeBatchTick:
+		case <-l.cfg.BatchTicker.Ticks():
 			// If the current batch is empty, then we have no work
 			// here. We also disable the batch ticker from waking up
 			// the htlcManager while the batch is empty.
 			if l.batchCounter == 0 {
-				l.cfg.BatchTicker.Stop()
-				maybeBatchTick = nil
+				l.cfg.BatchTicker.Pause()
 				continue
 			}
 
@@ -1012,8 +970,8 @@ out:
 			// If the downstream packet resulted in a non-empty
 			// batch, reinstate the batch ticker so that it can be
 			// cleared.
-			if l.batchCounter > 0 && maybeBatchTick == nil {
-				maybeBatchTick = l.cfg.BatchTicker.Start()
+			if l.batchCounter > 0 {
+				l.cfg.BatchTicker.Resume()
 			}
 
 		// A message from the switch was just received. This indicates
@@ -1041,8 +999,8 @@ out:
 			// If the downstream packet resulted in a non-empty
 			// batch, reinstate the batch ticker so that it can be
 			// cleared.
-			if l.batchCounter > 0 && maybeBatchTick == nil {
-				maybeBatchTick = l.cfg.BatchTicker.Start()
+			if l.batchCounter > 0 {
+				l.cfg.BatchTicker.Resume()
 			}
 
 		// A message from the connected peer was just received. This

htlcswitch/link_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/lightningnetwork/lnd/lnpeer"
 	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/lightningnetwork/lnd/ticker"
 )
 
 const (
@@ -1474,8 +1475,9 @@ func newSingleLinkTestHarness(chanAmt, chanReserve btcutil.Amount) (
 		return nil, nil, nil, nil, nil, nil, err
 	}
 
-	t := make(chan time.Time)
-	ticker := &mockTicker{t}
+	// Instantiate with a long interval, so that we can precisely control
+	// the firing via force feeding.
+	bticker := ticker.MockNew(time.Hour)
 	aliceCfg := ChannelLinkConfig{
 		FwrdingPolicy:      globalPolicy,
 		Peer:               alicePeer,
@@ -1497,8 +1499,8 @@ func newSingleLinkTestHarness(chanAmt, chanReserve btcutil.Amount) (
 		},
 		Registry:       invoiceRegistry,
 		ChainEvents:    &contractcourt.ChainEventSubscription{},
-		BatchTicker:    ticker,
-		FwdPkgGCTicker: NewBatchTicker(5 * time.Second),
+		BatchTicker:    bticker,
+		FwdPkgGCTicker: ticker.MockNew(5 * time.Second),
 		// Make the BatchSize and Min/MaxFeeUpdateTimeout large enough
 		// to not trigger commit updates automatically during tests.
 		BatchSize:           10000,
@@ -1528,7 +1530,7 @@ func newSingleLinkTestHarness(chanAmt, chanReserve btcutil.Amount) (
 		defer bobChannel.Stop()
 	}
 
-	return aliceLink, bobChannel, t, start, cleanUp, restore, nil
+	return aliceLink, bobChannel, bticker.Force, start, cleanUp, restore, nil
 }
 
 func assertLinkBandwidth(t *testing.T, link ChannelLink,
@@ -3846,8 +3848,9 @@ func restartLink(aliceChannel *lnwallet.LightningChannel, aliceSwitch *Switch,
 		}
 	}
 
-	t := make(chan time.Time)
-	ticker := &mockTicker{t}
+	// Instantiate with a long interval, so that we can precisely control
+	// the firing via force feeding.
+	bticker := ticker.MockNew(time.Hour)
 	aliceCfg := ChannelLinkConfig{
 		FwrdingPolicy:      globalPolicy,
 		Peer:               alicePeer,
@@ -3869,8 +3872,8 @@ func restartLink(aliceChannel *lnwallet.LightningChannel, aliceSwitch *Switch,
 		},
 		Registry:       invoiceRegistry,
 		ChainEvents:    &contractcourt.ChainEventSubscription{},
-		BatchTicker:    ticker,
-		FwdPkgGCTicker: NewBatchTicker(5 * time.Second),
+		BatchTicker:    bticker,
+		FwdPkgGCTicker: ticker.New(5 * time.Second),
 		// Make the BatchSize and Min/MaxFeeUpdateTimeout large enough
 		// to not trigger commit updates automatically during tests.
 		BatchSize:           10000,
@@ -3901,7 +3904,7 @@ func restartLink(aliceChannel *lnwallet.LightningChannel, aliceSwitch *Switch,
 		defer aliceLink.Stop()
 	}
 
-	return aliceLink, t, cleanUp, nil
+	return aliceLink, bticker.Force, cleanUp, nil
 }
 
 // gnerateHtlc generates a simple payment from Bob to Alice.

htlcswitch/mock.go

@@ -26,6 +26,7 @@ import (
 	"github.com/lightningnetwork/lnd/lnpeer"
 	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/lightningnetwork/lnd/ticker"
 )
 
 type mockPreimageCache struct {
@@ -145,7 +146,9 @@ func initSwitchWithDB(startingHeight uint32, db *channeldb.DB) (*Switch, error)
 		FetchLastChannelUpdate: func(lnwire.ShortChannelID) (*lnwire.ChannelUpdate, error) {
 			return nil, nil
 		},
-		Notifier: &mockNotifier{},
+		Notifier:       &mockNotifier{},
+		FwdEventTicker: ticker.MockNew(DefaultFwdEventInterval),
+		LogEventTicker: ticker.MockNew(DefaultLogInterval),
 	}
 
 	return New(cfg, startingHeight)
@@ -809,14 +812,3 @@ func (m *mockNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, _ []byte,
 		Spend: make(chan *chainntnfs.SpendDetail),
 	}, nil
 }
-
-type mockTicker struct {
-	ticker <-chan time.Time
-}
-
-func (m *mockTicker) Start() <-chan time.Time {
-	return m.ticker
-}
-
-func (m *mockTicker) Stop() {
-}

htlcswitch/switch.go

@@ -21,6 +21,17 @@ import (
 	"github.com/lightningnetwork/lnd/lnrpc"
 	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/lightningnetwork/lnd/ticker"
+)
+
+const (
+	// DefaultFwdEventInterval is the duration between attempts to flush
+	// pending forwarding events to disk.
+	DefaultFwdEventInterval = 15 * time.Second
+
+	// DefaultLogInterval is the duration between attempts to log statistics
+	// about forwarding events.
+	DefaultLogInterval = 10 * time.Second
 )
 
 var (
@@ -146,6 +157,14 @@ type Config struct {
 	// Notifier is an instance of a chain notifier that we'll use to signal
 	// the switch when a new block has arrived.
 	Notifier chainntnfs.ChainNotifier
+
+	// FwdEventTicker is a signal that instructs the htlcswitch to flush any
+	// pending forwarding events.
+	FwdEventTicker ticker.Ticker
+
+	// LogEventTicker is a signal instructing the htlcswitch to log
+	// aggregate stats about it's forwarding during the last interval.
+	LogEventTicker ticker.Ticker
 }
 
 // Switch is the central messaging bus for all incoming/outgoing HTLCs.
@@ -1390,13 +1409,13 @@ func (s *Switch) htlcForwarder() {
 		totalSatSent    btcutil.Amount
 		totalSatRecv    btcutil.Amount
 	)
-	logTicker := time.NewTicker(10 * time.Second)
-	defer logTicker.Stop()
+	s.cfg.LogEventTicker.Resume()
+	defer s.cfg.LogEventTicker.Stop()
 
 	// Every 15 seconds, we'll flush out the forwarding events that
 	// occurred during that period.
-	fwdEventTicker := time.NewTicker(15 * time.Second)
-	defer fwdEventTicker.Stop()
+	s.cfg.FwdEventTicker.Resume()
+	defer s.cfg.FwdEventTicker.Stop()
 
 out:
 	for {
@@ -1474,7 +1493,7 @@ out:
 		// When this time ticks, then it indicates that we should
 		// collect all the forwarding events since the last internal,
 		// and write them out to our log.
-		case <-fwdEventTicker.C:
+		case <-s.cfg.FwdEventTicker.Ticks():
 			s.wg.Add(1)
 			go func() {
 				defer s.wg.Done()
@@ -1488,7 +1507,7 @@ out:
 		// The log ticker has fired, so we'll calculate some forwarding
 		// stats for the last 10 seconds to display within the logs to
 		// users.
-		case <-logTicker.C:
+		case <-s.cfg.LogEventTicker.Ticks():
 			// First, we'll collate the current running tally of
 			// our forwarding stats.
 			prevSatSent := totalSatSent

htlcswitch/test_utils.go

@@ -28,6 +28,7 @@ import (
 	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/shachain"
+	"github.com/lightningnetwork/lnd/ticker"
 )
 
 var (
@@ -915,8 +916,8 @@ func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 			ChainEvents:         &contractcourt.ChainEventSubscription{},
 			SyncStates:          true,
 			BatchSize:           10,
-			BatchTicker:         &mockTicker{time.NewTicker(batchTimeout).C},
-			FwdPkgGCTicker:      &mockTicker{time.NewTicker(fwdPkgTimeout).C},
+			BatchTicker:         ticker.MockNew(batchTimeout),
+			FwdPkgGCTicker:      ticker.MockNew(fwdPkgTimeout),
 			MinFeeUpdateTimeout: minFeeUpdateTimeout,
 			MaxFeeUpdateTimeout: maxFeeUpdateTimeout,
 			OnChannelFailure:    func(lnwire.ChannelID, lnwire.ShortChannelID, LinkFailureError) {},
@@ -958,8 +959,8 @@ func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 			ChainEvents:         &contractcourt.ChainEventSubscription{},
 			SyncStates:          true,
 			BatchSize:           10,
-			BatchTicker:         &mockTicker{time.NewTicker(batchTimeout).C},
-			FwdPkgGCTicker:      &mockTicker{time.NewTicker(fwdPkgTimeout).C},
+			BatchTicker:         ticker.MockNew(batchTimeout),
+			FwdPkgGCTicker:      ticker.MockNew(fwdPkgTimeout),
 			MinFeeUpdateTimeout: minFeeUpdateTimeout,
 			MaxFeeUpdateTimeout: maxFeeUpdateTimeout,
 			OnChannelFailure:    func(lnwire.ChannelID, lnwire.ShortChannelID, LinkFailureError) {},
@@ -1001,8 +1002,8 @@ func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 			ChainEvents:         &contractcourt.ChainEventSubscription{},
 			SyncStates:          true,
 			BatchSize:           10,
-			BatchTicker:         &mockTicker{time.NewTicker(batchTimeout).C},
-			FwdPkgGCTicker:      &mockTicker{time.NewTicker(fwdPkgTimeout).C},
+			BatchTicker:         ticker.MockNew(batchTimeout),
+			FwdPkgGCTicker:      ticker.MockNew(fwdPkgTimeout),
 			MinFeeUpdateTimeout: minFeeUpdateTimeout,
 			MaxFeeUpdateTimeout: maxFeeUpdateTimeout,
 			OnChannelFailure:    func(lnwire.ChannelID, lnwire.ShortChannelID, LinkFailureError) {},
@@ -1044,8 +1045,8 @@ func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 			ChainEvents:         &contractcourt.ChainEventSubscription{},
 			SyncStates:          true,
 			BatchSize:           10,
-			BatchTicker:         &mockTicker{time.NewTicker(batchTimeout).C},
-			FwdPkgGCTicker:      &mockTicker{time.NewTicker(fwdPkgTimeout).C},
+			BatchTicker:         ticker.MockNew(batchTimeout),
+			FwdPkgGCTicker:      ticker.MockNew(fwdPkgTimeout),
 			MinFeeUpdateTimeout: minFeeUpdateTimeout,
 			MaxFeeUpdateTimeout: maxFeeUpdateTimeout,
 			OnChannelFailure:    func(lnwire.ChannelID, lnwire.ShortChannelID, LinkFailureError) {},