lotus/chain/events/events.go

175 lines
3.8 KiB
Go
Raw Normal View History

2019-09-05 07:40:50 +00:00
package events
2019-09-03 17:45:55 +00:00
import (
2019-09-18 11:01:52 +00:00
"context"
2019-09-03 17:45:55 +00:00
"sync"
2019-09-18 11:01:52 +00:00
"time"
2019-09-03 17:45:55 +00:00
2019-09-18 11:01:52 +00:00
"github.com/ipfs/go-cid"
2019-09-05 07:40:50 +00:00
logging "github.com/ipfs/go-log"
2019-09-04 16:09:08 +00:00
"golang.org/x/xerrors"
2019-09-18 11:01:52 +00:00
"github.com/filecoin-project/go-lotus/api"
2019-09-03 17:45:55 +00:00
"github.com/filecoin-project/go-lotus/build"
2019-09-18 11:01:52 +00:00
"github.com/filecoin-project/go-lotus/chain/store"
2019-09-03 17:45:55 +00:00
"github.com/filecoin-project/go-lotus/chain/types"
)
2019-09-05 07:40:50 +00:00
var log = logging.Logger("events")
2019-09-03 17:45:55 +00:00
// `curH`-`ts.Height` = `confidence`
2019-09-04 16:09:08 +00:00
type HeightHandler func(ts *types.TipSet, curH uint64) error
type RevertHandler func(ts *types.TipSet) error
2019-09-03 17:45:55 +00:00
2019-09-04 16:09:08 +00:00
type heightHandler struct {
2019-09-03 17:45:55 +00:00
confidence int
2019-09-04 16:09:08 +00:00
handle HeightHandler
revert RevertHandler
2019-09-03 17:45:55 +00:00
}
2019-09-18 11:01:52 +00:00
type eventApi interface {
ChainNotify(context.Context) (<-chan []*store.HeadChange, error)
ChainGetBlockMessages(context.Context, cid.Cid) (*api.BlockMessages, error)
ChainGetTipSetByHeight(context.Context, uint64, *types.TipSet) (*types.TipSet, error)
2019-09-03 17:45:55 +00:00
}
type Events struct {
2019-09-18 11:01:52 +00:00
api eventApi
2019-09-03 17:45:55 +00:00
tsc *tipSetCache
lk sync.Mutex
2019-09-18 11:01:52 +00:00
ready sync.WaitGroup
readyOnce sync.Once
heightEvents
2019-09-04 16:09:08 +00:00
calledEvents
2019-09-03 17:45:55 +00:00
}
func NewEvents(ctx context.Context, api eventApi) *Events {
2019-09-03 17:45:55 +00:00
gcConfidence := 2 * build.ForkLengthThreshold
tsc := newTSCache(gcConfidence, api.ChainGetTipSetByHeight)
2019-09-04 16:09:08 +00:00
2019-09-03 17:45:55 +00:00
e := &Events{
2019-09-18 11:01:52 +00:00
api: api,
2019-09-03 17:45:55 +00:00
2019-09-04 16:09:08 +00:00
tsc: tsc,
2019-09-03 17:45:55 +00:00
heightEvents: heightEvents{
tsc: tsc,
gcConfidence: uint64(gcConfidence),
heightTriggers: map[uint64]*heightHandler{},
htTriggerHeights: map[uint64][]uint64{},
htHeights: map[uint64][]uint64{},
},
2019-09-03 17:45:55 +00:00
2019-09-04 16:09:08 +00:00
calledEvents: calledEvents{
2019-09-18 11:01:52 +00:00
cs: api,
tsc: tsc,
gcConfidence: uint64(gcConfidence),
2019-09-04 16:09:08 +00:00
confQueue: map[triggerH]map[msgH][]*queuedEvent{},
revertQueue: map[msgH][]triggerH{},
triggers: map[triggerId]*callHandler{},
callTuples: map[callTuple][]triggerId{},
timeouts: map[uint64]map[triggerId]int{},
2019-09-04 16:09:08 +00:00
},
2019-09-03 17:45:55 +00:00
}
2019-09-18 11:01:52 +00:00
e.ready.Add(1)
go e.listenHeadChanges(ctx)
2019-09-18 11:01:52 +00:00
e.ready.Wait()
2019-09-03 17:45:55 +00:00
2019-09-03 17:59:32 +00:00
// TODO: cleanup/gc goroutine
2019-09-03 17:45:55 +00:00
return e
}
func (e *Events) listenHeadChanges(ctx context.Context) {
for {
if err := e.listenHeadChangesOnce(ctx); err != nil {
log.Errorf("listen head changes errored: %s", err)
} else {
log.Warn("listenHeadChanges quit")
}
2019-09-18 11:01:52 +00:00
if ctx.Err() != nil {
log.Warnf("not restarting listenHeadChanges: context error: %s", ctx.Err())
return
}
time.Sleep(time.Second)
log.Info("restarting listenHeadChanges")
}
2019-09-18 11:01:52 +00:00
}
func (e *Events) listenHeadChangesOnce(ctx context.Context) error {
2019-09-27 11:37:44 +00:00
ctx, cancel := context.WithCancel(ctx)
defer cancel()
2019-09-18 11:01:52 +00:00
notifs, err := e.api.ChainNotify(ctx)
if err != nil {
// TODO: retry
return xerrors.Errorf("listenHeadChanges ChainNotify call failed: %w", err)
2019-09-18 11:01:52 +00:00
}
cur, ok := <-notifs // TODO: timeout?
if !ok {
return xerrors.Errorf("notification channel closed")
2019-09-18 11:01:52 +00:00
}
if len(cur) != 1 {
return xerrors.Errorf("unexpected initial head notification length: %d", len(cur))
2019-09-18 11:01:52 +00:00
}
if cur[0].Type != store.HCCurrent {
return xerrors.Errorf("expected first head notification type to be 'current', was '%s'", cur[0].Type)
2019-09-18 11:01:52 +00:00
}
if err := e.tsc.add(cur[0].Val); err != nil {
log.Warn("tsc.add: adding current tipset failed: %w", err)
2019-09-18 11:01:52 +00:00
}
e.readyOnce.Do(func() {
e.ready.Done()
})
for notif := range notifs {
var rev, app []*types.TipSet
for _, notif := range notif {
switch notif.Type {
case store.HCRevert:
rev = append(rev, notif.Val)
case store.HCApply:
app = append(app, notif.Val)
default:
log.Warnf("unexpected head change notification type: '%s'", notif.Type)
}
}
if err := e.headChange(rev, app); err != nil {
log.Warnf("headChange failed: %s", err)
}
}
return nil
2019-09-18 11:01:52 +00:00
}
2019-09-03 17:45:55 +00:00
func (e *Events) headChange(rev, app []*types.TipSet) error {
2019-09-03 17:59:32 +00:00
if len(app) == 0 {
return xerrors.New("events.headChange expected at least one applied tipset")
}
2019-09-03 17:45:55 +00:00
e.lk.Lock()
defer e.lk.Unlock()
2019-09-03 17:59:32 +00:00
if err := e.headChangeAt(rev, app); err != nil {
return err
2019-09-03 17:45:55 +00:00
}
2019-09-03 17:59:32 +00:00
return e.headChangeCalled(rev, app)
}