htlcswitch+peer: remove fwd pkgs once before tick, bump timer to 1hr

This commit changes the logic when garbage collecting forwarding
packages such that they are removed once when the function is called,
and then again upon subsequent ticks. This allows us to bump the
peer timer to 1 hour to limit the number of db transactions happening
in lnd. The forwarding packages need to be removed initially as
otherwise a flappy node will never have them garbage collected.
This commit is contained in:
eugene 2020-08-19 11:09:48 -04:00
parent 62e19185f1
commit ea467783e9
2 changed files with 34 additions and 18 deletions

@ -843,35 +843,51 @@ func (l *channelLink) fwdPkgGarbager() {
l.cfg.FwdPkgGCTicker.Resume() l.cfg.FwdPkgGCTicker.Resume()
defer l.cfg.FwdPkgGCTicker.Stop() defer l.cfg.FwdPkgGCTicker.Stop()
if err := l.loadAndRemove(); err != nil {
l.log.Warnf("unable to run initial fwd pkgs gc: %v", err)
}
for { for {
select { select {
case <-l.cfg.FwdPkgGCTicker.Ticks(): case <-l.cfg.FwdPkgGCTicker.Ticks():
fwdPkgs, err := l.channel.LoadFwdPkgs() if err := l.loadAndRemove(); err != nil {
if err != nil { l.log.Warnf("unable to remove fwd pkgs: %v",
l.log.Warnf("unable to load fwdpkgs for gc: %v",
err) err)
continue continue
} }
// TODO(conner): batch removal of forward packages.
for _, fwdPkg := range fwdPkgs {
if fwdPkg.State != channeldb.FwdStateCompleted {
continue
}
err = l.channel.RemoveFwdPkgs(fwdPkg.Height)
if err != nil {
l.log.Warnf("unable to remove fwd pkg "+
"for height=%d: %v",
fwdPkg.Height, err)
}
}
case <-l.quit: case <-l.quit:
return return
} }
} }
} }
// loadAndRemove loads all the channels forwarding packages and determines if
// they can be removed. It is called once before the FwdPkgGCTicker ticks so that
// a longer tick interval can be used.
func (l *channelLink) loadAndRemove() error {
fwdPkgs, err := l.channel.LoadFwdPkgs()
if err != nil {
return err
}
var removeHeights []uint64
for _, fwdPkg := range fwdPkgs {
if fwdPkg.State != channeldb.FwdStateCompleted {
continue
}
removeHeights = append(removeHeights, fwdPkg.Height)
}
// If removeHeights is empty, return early so we don't use a db
// transaction.
if len(removeHeights) == 0 {
return nil
}
return l.channel.RemoveFwdPkgs(removeHeights...)
}
// htlcManager is the primary goroutine which drives a channel's commitment // htlcManager is the primary goroutine which drives a channel's commitment
// update state-machine in response to messages received via several channels. // update state-machine in response to messages received via several channels.
// This goroutine reads messages from the upstream (remote) peer, and also from // This goroutine reads messages from the upstream (remote) peer, and also from

@ -588,7 +588,7 @@ func (p *Brontide) addLink(chanPoint *wire.OutPoint,
OnChannelFailure: onChannelFailure, OnChannelFailure: onChannelFailure,
SyncStates: syncStates, SyncStates: syncStates,
BatchTicker: ticker.New(50 * time.Millisecond), BatchTicker: ticker.New(50 * time.Millisecond),
FwdPkgGCTicker: ticker.New(time.Minute), FwdPkgGCTicker: ticker.New(time.Hour),
PendingCommitTicker: ticker.New(time.Minute), PendingCommitTicker: ticker.New(time.Minute),
BatchSize: 10, BatchSize: 10,
UnsafeReplay: p.cfg.UnsafeReplay, UnsafeReplay: p.cfg.UnsafeReplay,