cerc-io/lotus
10
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

Merge pull request #5159 from iand/fix/event-hang

Browse Source 
fix(events): avoid potential hang when starting event listener
This commit is contained in:
Łukasz Magiera 2020-12-09 19:40:02 +01:00 committed by GitHub
commit e1be89b442
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 25 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
chain/events/events.go
View File

@ -20,8 +20,10 @@ import (
var log = logging.Logger("events") var log = logging.Logger("events")
// HeightHandler `curH`-`ts.Height` = `confidence` // HeightHandler `curH`-`ts.Height` = `confidence`
type HeightHandler func(ctx context.Context, ts *types.TipSet, curH abi.ChainEpoch) error type (
type RevertHandler func(ctx context.Context, ts *types.TipSet) error HeightHandler func(ctx context.Context, ts *types.TipSet, curH abi.ChainEpoch) error
RevertHandler func(ctx context.Context, ts *types.TipSet) error
)
type heightHandler struct { type heightHandler struct {
confidence int confidence int
@ -48,7 +50,7 @@ type Events struct {
tsc *tipSetCache tsc *tipSetCache
lk sync.Mutex lk sync.Mutex
ready sync.WaitGroup ready chan struct{}
readyOnce sync.Once readyOnce sync.Once
heightEvents heightEvents
@ -76,15 +78,16 @@ func NewEvents(ctx context.Context, api eventAPI) *Events {
}, },
hcEvents: newHCEvents(ctx, api, tsc, uint64(gcConfidence)), hcEvents: newHCEvents(ctx, api, tsc, uint64(gcConfidence)),
ready: make(chan struct{}),
} }
e.ready.Add(1)
go e.listenHeadChanges(ctx) go e.listenHeadChanges(ctx)
e.ready.Wait() // Wait for the first tipset to be seen or bail if shutting down
select {
// TODO: cleanup/gc goroutine case <-e.ready:
case <-ctx.Done():
}
return e return e
} }
@ -111,14 +114,22 @@ func (e *Events) listenHeadChangesOnce(ctx context.Context) error {
notifs, err := e.api.ChainNotify(ctx) notifs, err := e.api.ChainNotify(ctx)
if err != nil { if err != nil {
// TODO: retry // Retry is handled by caller
return xerrors.Errorf("listenHeadChanges ChainNotify call failed: %w", err) return xerrors.Errorf("listenHeadChanges ChainNotify call failed: %w", err)
} }
cur, ok := <-notifs // TODO: timeout? var cur []*api.HeadChange
var ok bool
// Wait for first tipset or bail
select {
case cur, ok = <-notifs:
if !ok { if !ok {
return xerrors.Errorf("notification channel closed") return xerrors.Errorf("notification channel closed")
} }
case <-ctx.Done():
return ctx.Err()
}
if len(cur) != 1 { if len(cur) != 1 {
return xerrors.Errorf("unexpected initial head notification length: %d", len(cur)) return xerrors.Errorf("unexpected initial head notification length: %d", len(cur))
@ -134,8 +145,8 @@ func (e *Events) listenHeadChangesOnce(ctx context.Context) error {
e.readyOnce.Do(func() { e.readyOnce.Do(func() {
e.lastTs = cur[0].Val e.lastTs = cur[0].Val
// Signal that we have seen first tipset
e.ready.Done() close(e.ready)
}) })
for notif := range notifs { for notif := range notifs {