seedhammer/dcrd
seedhammer/dcrd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

indexers: fix indexer wait for sync.

Browse Source 
This fixes an issue with indexers wait for sync where the indexer gets
synced before a sync subcsriber is created. The sync subscriber is left
idling, waitingfor the next update which blocks the caller. The index
subscriber has been updated to periodically update sync subcribers
of all subscribed indexers to it as a result.
This commit is contained in:
Donald Adu-Poku 2022-03-24 23:31:16 +00:00 committed by Dave Collins
parent 6ec3707f98
commit fd038e36ce
9 changed files with 154 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
blockchain/indexers/addrindex.go
View File

@ -643,7 +643,7 @@ func (idx *AddrIndex) Init(ctx context.Context, chainParams *chaincfg.Params) er
}
// Recover the address index and its dependents to the main chain if needed.
if err := recover(ctx, idx); err != nil {
if err := recoverIndex(ctx, idx); err != nil {
return err
}
@ -702,12 +702,22 @@ func (idx *AddrIndex) IndexSubscription() *IndexSubscription {
// Subscribers returns all client channels waiting for the next index update.
//
// This is part of the Indexer interface.
// Deprecated: This will be removed in the next major version bump.
func (idx *AddrIndex) Subscribers() map[chan bool]struct{} {
idx.mtx.Lock()
defer idx.mtx.Unlock()
return idx.subscribers
}
// NotifySyncSubscribers signals subscribers of an index sync update.
//
// This is part of the Indexer interface.
func (idx *AddrIndex) NotifySyncSubscribers() {
idx.mtx.Lock()
notifySyncSubscribers(idx.subscribers)
idx.mtx.Unlock()
}
// WaitForSync subscribes clients for the next index sync update.
//
// This is part of the Indexer interface.

6
blockchain/indexers/addrindex_test.go
View File

@ -456,12 +456,16 @@ func TestAddrIndexAsync(t *testing.T) {
bk5a := addBlock(t, chain, &g, "bk5a")
// Resubscribe the indexes.
subber.mtx.Lock()
err = addrIdx.sub.stop()
subber.mtx.Unlock()
if err != nil {
t.Fatal(err)
}
subber.mtx.Lock()
err = txIdx.sub.stop()
subber.mtx.Unlock()
if err != nil {
t.Fatal(err)
}
@ -619,7 +623,9 @@ func TestAddrIndexAsync(t *testing.T) {
}
// Drop the address index and resubscribe.
subber.mtx.Lock()
err = addrIdx.sub.stop()
subber.mtx.Unlock()
if err != nil {
t.Fatal(err)
}

30
blockchain/indexers/common.go
View File

@ -143,7 +143,12 @@ type Indexer interface {
WaitForSync() chan bool
// Subscribers returns all client channels waiting for the next index update.
// Deprecated: This will be removed in the next major version bump.
Subscribers() map[chan bool]struct{}
// NotifySyncSubscribers signals subscribers of an index sync update.
// This should only be called when an index is synced.
NotifySyncSubscribers()
}
// IndexDropper provides a method to remove an index from the database. Indexers
@ -460,9 +465,18 @@ func tip(db database.DB, key []byte) (int64, *chainhash.Hash, error) {
return int64(height), hash, err
}
// recover reverts the index to a block on the main chain by repeatedly
// notifySyncSubscribers signals provided subscribers the index subcribed to
// is synced.
func notifySyncSubscribers(subscribers map[chan bool]struct{}) {
for sub := range subscribers {
close(sub)
delete(subscribers, sub)
}
}
// recoverIndex reverts the index to a block on the main chain by repeatedly
// disconnecting the index tip if it is not on the main chain.
func recover(ctx context.Context, idx Indexer) error {
func recoverIndex(ctx context.Context, idx Indexer) error {
// Fetch the current tip for the index.
height, hash, err := idx.Tip()
if err != nil {
@ -670,13 +684,6 @@ func upgradeIndex(ctx context.Context, indexer Indexer, genesisHash *chainhash.H
// maybeNotifySubscribers updates subscribers the index is synced when
// the tip is identical to the chain tip.
func maybeNotifySubscribers(ctx context.Context, indexer Indexer) error {
subs := indexer.Subscribers()
// Exit immediately if the index has no subscribers.
if len(subs) == 0 {
return nil
}
if interruptRequested(ctx) {
return indexerError(ErrInterruptRequested, interruptMsg)
}
@ -689,10 +696,7 @@ func maybeNotifySubscribers(ctx context.Context, indexer Indexer) error {
}
if tipHeight == bestHeight && *bestHash == *tipHash {
for sub := range subs {
close(sub)
delete(subs, sub)
}
indexer.NotifySyncSubscribers()
}
return nil

12
blockchain/indexers/existsaddrindex.go
View File

@ -129,7 +129,7 @@ func (idx *ExistsAddrIndex) Init(ctx context.Context, chainParams *chaincfg.Para
// Recover the exists address index and its dependents to the main
// chain if needed.
if err := recover(ctx, idx); err != nil {
if err := recoverIndex(ctx, idx); err != nil {
return err
}
@ -197,12 +197,22 @@ func (idx *ExistsAddrIndex) IndexSubscription() *IndexSubscription {
// Subscribers returns all client channels waiting for the next index update.
//
// This is part of the Indexer interface.
// Deprecated: This will be removed in the next major version bump.
func (idx *ExistsAddrIndex) Subscribers() map[chan bool]struct{} {
idx.mtx.Lock()
defer idx.mtx.Unlock()
return idx.subscribers
}
// NotifySyncSubscribers signals subscribers of an index sync update.
//
// This is part of the Indexer interface.
func (idx *ExistsAddrIndex) NotifySyncSubscribers() {
idx.mtx.Lock()
notifySyncSubscribers(idx.subscribers)
idx.mtx.Unlock()
}
// WaitForSync subscribes clients for the next index sync update.
//
// This is part of the Indexer interface.

4
blockchain/indexers/existsaddrindex_test.go
View File

@ -128,7 +128,9 @@ func TestExistsAddrIndexAsync(t *testing.T) {
bk5a := addBlock(t, chain, &g, "bk5a")
// Resubscribe the index.
subber.mtx.Lock()
err = idx.sub.stop()
subber.mtx.Unlock()
if err != nil {
t.Fatal(err)
}
@ -223,7 +225,9 @@ func TestExistsAddrIndexAsync(t *testing.T) {
}
// Resubscribe the index.
subber.mtx.Lock()
err = idx.sub.stop()
subber.mtx.Unlock()
if err != nil {
t.Fatal(err)
}

90
blockchain/indexers/indexsubscriber.go
View File

@ -9,6 +9,7 @@ import (
"fmt"
"sync"
"sync/atomic"
"time"
"github.com/decred/dcrd/blockchain/v5/internal/progresslog"
"github.com/decred/dcrd/database/v3"
@ -34,6 +35,10 @@ var (
// noPrereqs indicates no index prerequisites.
noPrereqs = "none"
// syncUpdateInterval is the time between periodically checking indexes
// and notifying their synchronization subscribers if synced.
syncUpdateInterval = time.Millisecond * 500
)
// IndexNtfn represents an index notification detailing a block connection
@ -80,6 +85,8 @@ func newIndexSubscription(subber *IndexSubscriber, indexer Indexer, prereq strin
// stop prevents any future index updates from being delivered and
// unsubscribes the associated subscription.
//
// This must be called with the index subscriber mutex held for writes.
func (s *IndexSubscription) stop() error {
// If the subscription has a prerequisite, find it and remove the
@ -109,9 +116,7 @@ func (s *IndexSubscription) stop() error {
// If the subscription is independent, remove it from the
// index subscriber's subscriptions.
s.mtx.Lock()
delete(s.subscriber.subscriptions, s.id)
s.mtx.Unlock()
return nil
}
@ -125,6 +130,7 @@ type IndexSubscriber struct {
mtx sync.Mutex
ctx context.Context
cancel context.CancelFunc
wg sync.WaitGroup
quit chan struct{}
}
@ -348,14 +354,60 @@ func (s *IndexSubscriber) CatchUp(ctx context.Context, db database.DB, queryer C
return nil
}
// Run relays index notifications to subscribed indexes.
// handleSyncSubscribers updates index sync subscribers when a subscribed
// indexer is fully synced.
//
// This should be run as a goroutine.
func (s *IndexSubscriber) Run(ctx context.Context) {
func (s *IndexSubscriber) handleSyncSubscribers(ctx context.Context) {
ticker := time.NewTicker(syncUpdateInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
s.wg.Done()
return
case <-ticker.C:
s.mtx.Lock()
for _, sub := range s.subscriptions {
err := maybeNotifySubscribers(ctx, sub.idx)
if err != nil {
log.Errorf("unable to notify sync subscribers: %v", err)
s.cancel()
}
}
s.mtx.Unlock()
}
}
}
// handleIndexUpdates relays updates to subscribed indexers.
func (s *IndexSubscriber) handleIndexUpdates(ctx context.Context) {
for {
select {
case <-ctx.Done():
close(s.quit)
// Stop all updates to subscribed indexes and terminate their
// processes.
s.mtx.Lock()
for _, sub := range s.subscriptions {
err := sub.stop()
if err != nil {
log.Error("unable to stop index subscription: %v", err)
}
}
s.mtx.Unlock()
s.cancel()
s.wg.Done()
return
case ntfn := <-s.c:
// Relay the index update to subscribed indexes.
s.mtx.Lock()
for _, sub := range s.subscriptions {
err := updateIndex(ctx, sub.idx, &ntfn)
if err != nil {
@ -364,27 +416,23 @@ func (s *IndexSubscriber) Run(ctx context.Context) {
break
}
}
s.mtx.Unlock()
if ntfn.Done != nil {
close(ntfn.Done)
}
case <-ctx.Done():
log.Infof("Index subscriber shutting down")
close(s.quit)
// Stop all updates to subscribed indexes and terminate their
// processes.
for _, sub := range s.subscriptions {
err := sub.stop()
if err != nil {
log.Error("unable to stop index subscription: %v", err)
}
}
s.cancel()
return
}
}
}
// Run relays index notifications to subscribed indexes.
//
// This should be run as a goroutine.
func (s *IndexSubscriber) Run(ctx context.Context) {
s.wg.Add(2)
go s.handleIndexUpdates(ctx)
go s.handleSyncSubscribers(ctx)
s.wg.Wait()
log.Infof("Index subscriber shutting down")
}

22
blockchain/indexers/indexsubscriber_test.go
View File

@ -7,6 +7,7 @@ package indexers
import (
"context"
"testing"
"time"
"github.com/decred/dcrd/blockchain/v5/chaingen"
"github.com/decred/dcrd/chaincfg/v3"
@ -37,7 +38,6 @@ func TestIndexSubscriberAsync(t *testing.T) {
defer pCancel()
subber := NewIndexSubscriber(ctx)
go subber.Run(ctx)
err = AddIndexSpendConsumers(db, chain)
if err != nil {
@ -64,6 +64,8 @@ func TestIndexSubscriberAsync(t *testing.T) {
t.Fatal(err)
}
go subber.Run(ctx)
// Ensure all indexes through their prerequisite/dependency relationships
// are synced to the current chain tip (bk3).
addrIdxTipHeight, addrIdxTipHash, err := addrIdx.Tip()
@ -110,6 +112,22 @@ func TestIndexSubscriberAsync(t *testing.T) {
existsAddrIdxTipHash)
}
// Create 3 sync subscribers for the tx index.
for idx := 0; idx < 3; idx++ {
txIdx.WaitForSync()
}
// Wait for the sync subscriber handler to run.
time.Sleep(time.Millisecond * 750)
// Ensure the tx index no longer has sync subscribers.
txIdx.mtx.Lock()
if len(txIdx.subscribers) > 0 {
txIdx.mtx.Unlock()
t.Fatalf("expected no sync subscribers for the tx index")
}
txIdx.mtx.Unlock()
// Ensure the address index remains in sync with the main chain when new
// blocks are connected.
bk4 := addBlock(t, chain, &g, "bk4")
@ -168,7 +186,9 @@ func TestIndexSubscriberAsync(t *testing.T) {
}
// Ensure stopping a prequisite subscription stops its dependency as well.
subber.mtx.Lock()
err = txIdx.sub.stop()
subber.mtx.Unlock()
if err != nil {
t.Fatal(err)
}

12
blockchain/indexers/txindex.go
View File

@ -369,7 +369,7 @@ func (idx *TxIndex) Init(ctx context.Context, chainParams *chaincfg.Params) erro
}
// Recover the tx index and its dependents to the main chain if needed.
if err := recover(ctx, idx); err != nil {
if err := recoverIndex(ctx, idx); err != nil {
return err
}
@ -482,12 +482,22 @@ func (idx *TxIndex) IndexSubscription() *IndexSubscription {
// Subscribers returns all client channels waiting for the next index update.
//
// This is part of the Indexer interface.
// Deprecated: This will be removed in the next major version bump.
func (idx *TxIndex) Subscribers() map[chan bool]struct{} {
idx.mtx.Lock()
defer idx.mtx.Unlock()
return idx.subscribers
}
// NotifySyncSubscribers signals subscribers of an index sync update.
//
// This is part of the Indexer interface.
func (idx *TxIndex) NotifySyncSubscribers() {
idx.mtx.Lock()
notifySyncSubscribers(idx.subscribers)
idx.mtx.Unlock()
}
// WaitForSync subscribes clients for the next index sync update.
//
// This is part of the Indexer interface.

4
blockchain/indexers/txindex_test.go
View File

@ -434,7 +434,9 @@ func TestTxIndexAsync(t *testing.T) {
bk5a := addBlock(t, chain, &g, "bk5a")
// Resubscribe the index.
subber.mtx.Lock()
err = idx.sub.stop()
subber.mtx.Unlock()
if err != nil {
t.Fatal(err)
}
@ -528,7 +530,9 @@ func TestTxIndexAsync(t *testing.T) {
}
// Resubscribe the index.
subber.mtx.Lock()
err = idx.sub.stop()
subber.mtx.Unlock()
if err != nil {
t.Fatal(err)
}