lotus/chain/store/coalescer.go

179 lines
4.0 KiB
Go
Raw Normal View History

2020-11-02 11:12:32 +00:00
package store
import (
"context"
"time"
"github.com/filecoin-project/lotus/chain/types"
)
// WrapHeadChangeCoalescer wraps a ReorgNotifee with a head change coalescer.
2020-11-02 11:12:32 +00:00
func WrapHeadChangeCoalescer(fn ReorgNotifee, delay time.Duration) ReorgNotifee {
c := NewHeadChangeCoalescer(fn, delay)
return c.HeadChange
}
// HeadChangeCoalescer is a stateful reorg notifee which coalesces incoming head changes
// with pending head changes to reduce state computations from head change notifications.
2020-11-02 11:12:32 +00:00
type HeadChangeCoalescer struct {
notify ReorgNotifee
ctx context.Context
cancel func()
eventq chan headChange
revert []*types.TipSet
apply []*types.TipSet
}
type headChange struct {
revert, apply []*types.TipSet
}
// NewHeadChangeCoalescer creates a HeadChangeCoalescer.
2020-11-02 11:12:32 +00:00
func NewHeadChangeCoalescer(fn ReorgNotifee, delay time.Duration) *HeadChangeCoalescer {
ctx, cancel := context.WithCancel(context.Background())
c := &HeadChangeCoalescer{
notify: fn,
ctx: ctx,
cancel: cancel,
eventq: make(chan headChange),
}
go c.background(delay)
return c
}
// HeadChange is the ReorgNotifee callback for the stateful coalescer; it receives an incoming
// head change and schedules dispatch of a coalesced head change in the background.
2020-11-02 11:12:32 +00:00
func (c *HeadChangeCoalescer) HeadChange(revert, apply []*types.TipSet) error {
select {
case c.eventq <- headChange{revert: revert, apply: apply}:
return nil
case <-c.ctx.Done():
return c.ctx.Err()
}
}
// Close closes the coalescer and cancels the background dispatch goroutine.
// Any further notification will result in an error.
2020-11-02 11:12:32 +00:00
func (c *HeadChangeCoalescer) Close() {
select {
case <-c.ctx.Done():
default:
c.cancel()
}
}
// Implementation details
2020-11-02 11:12:32 +00:00
func (c *HeadChangeCoalescer) background(delay time.Duration) {
var timerC <-chan time.Time
for {
select {
case evt := <-c.eventq:
c.coalesce(evt.revert, evt.apply)
if timerC == nil {
timerC = time.After(delay)
}
case <-timerC:
c.dispatch()
timerC = nil
case <-c.ctx.Done():
if c.revert != nil || c.apply != nil {
c.dispatch()
}
2020-11-02 11:12:32 +00:00
return
}
}
}
func (c *HeadChangeCoalescer) coalesce(revert, apply []*types.TipSet) {
// newly reverted tipsets cancel out with pending applys.
// similarly, newly applied tipsets cancel out with pending reverts.
2020-11-02 11:12:32 +00:00
// pending tipsets
pendRevert := make(map[types.TipSetKey]struct{}, len(c.revert))
for _, ts := range c.revert {
pendRevert[ts.Key()] = struct{}{}
2020-11-02 11:12:32 +00:00
}
pendApply := make(map[types.TipSetKey]struct{}, len(c.apply))
for _, ts := range c.apply {
pendApply[ts.Key()] = struct{}{}
}
2020-11-02 11:12:32 +00:00
// incoming tipsets
reverting := make(map[types.TipSetKey]struct{}, len(revert))
2020-11-02 11:12:32 +00:00
for _, ts := range revert {
reverting[ts.Key()] = struct{}{}
}
applying := make(map[types.TipSetKey]struct{}, len(apply))
for _, ts := range apply {
applying[ts.Key()] = struct{}{}
}
// coalesced revert set
// - pending reverts are cancelled by incoming applys
// - incoming reverts are cancelled by pending applys
newRevert := make([]*types.TipSet, 0, len(c.revert)+len(revert))
for _, ts := range c.revert {
_, cancel := applying[ts.Key()]
if cancel {
continue
}
2020-11-02 11:12:32 +00:00
newRevert = append(newRevert, ts)
}
for _, ts := range revert {
_, cancel := pendApply[ts.Key()]
if cancel {
2020-11-02 11:12:32 +00:00
continue
}
newRevert = append(newRevert, ts)
2020-11-02 11:12:32 +00:00
}
// coalesced apply set
// - pending applys are cancelled by incoming reverts
// - incoming applys are cancelled by pending reverts
newApply := make([]*types.TipSet, 0, len(c.apply)+len(apply))
2020-11-02 11:12:32 +00:00
for _, ts := range c.apply {
_, cancel := reverting[ts.Key()]
if cancel {
continue
}
newApply = append(newApply, ts)
}
for _, ts := range apply {
_, cancel := pendRevert[ts.Key()]
if cancel {
2020-11-02 11:12:32 +00:00
continue
}
newApply = append(newApply, ts)
}
// commit the coalesced sets
c.revert = newRevert
2020-11-02 11:12:32 +00:00
c.apply = newApply
}
func (c *HeadChangeCoalescer) dispatch() {
err := c.notify(c.revert, c.apply)
if err != nil {
log.Errorf("error dispatching coalesced head change notification: %s", err)
}
c.revert = nil
c.apply = nil
}