Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Order mixclient and syncer shutdown #2463

Open
wants to merge 7 commits into
base: master
Choose a base branch
from
+71 −32
Open

Order mixclient and syncer shutdown #2463

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
7f35fe1
Order mixclient and syncer shutdown
jrick Jan 23, 2025
e77e577
block and return context error if wallet.Run returns nil
jrick Jan 24, 2025
007d62a
remove unnecessary wrapped context
jrick Jan 24, 2025
9a6e6f8
less stupid code
jrick Jan 24, 2025
27a89a9
unneeded comma
jrick Jan 24, 2025
498b9b3
back to wrapping
jrick Jan 24, 2025
1033335
this goroutine doesn't need to be part of the errgroup
jrick Jan 24, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 27 additions & 13 deletions chain/sync.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -542,9 +542,11 @@ func (s *Syncer) Run(ctx context.Context) (err error) {

params := s.wallet.ChainParams()

ntfnCtx, ntfnCtxCancel := context.WithCancel(context.Background())
defer ntfnCtxCancel()
s.notifier = &notifier{
syncer: s,
ctx: ctx,
ctx: ntfnCtx,
closed: make(chan struct{}),
}
addr, err := normalizeAddress(s.opts.Address, s.opts.DefaultPort)
Expand Down Expand Up @@ -589,12 +591,12 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
}
opts = append(opts, wsrpc.WithTLSConfig(tc))
}
client, err := wsrpc.Dial(ctx, addr, opts...)
wsClient, err := wsrpc.Dial(ctx, addr, opts...)
if err != nil {
return err
}
defer client.Close()
s.rpc = dcrd.New(client)
defer wsClient.Close()
s.rpc = dcrd.New(wsClient)

// Verify that the server is running on the expected network.
var netID wire.CurrencyNet
Expand Down Expand Up @@ -723,10 +725,27 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
return err
}

defer func() {
ntfnCtxCancel()

select {
case <-ctx.Done():
wsClient.Close()
default:
}

// Wait for notifications to finish before returning
<-s.notifier.closed
}()

// Ensure wallet.Run cleanly finishes/is canceled first when outer
// context is canceled.
walletCtx, walletCtxCancel := context.WithCancel(context.Background())
defer walletCtxCancel()
g.Go(func() error {
// Run wallet background goroutines (currently, this just runs
// mixclient).
return s.wallet.Run(ctx)
return s.wallet.Run(walletCtx)
})

// Request notifications for mixing messages.
Expand All @@ -739,18 +758,13 @@ func (s *Syncer) Run(ctx context.Context) (err error) {

log.Infof("Blockchain sync completed, wallet ready for general usage.")

// Wait for notifications to finish before returning
defer func() {
<-s.notifier.closed
}()

g.Go(func() error {
select {
case <-ctx.Done():
client.Close()
walletCtxCancel()
return ctx.Err()
case <-client.Done():
return client.Err()
case <-wsClient.Done():
return wsClient.Err()
}
})
return g.Wait()
Expand Down
9 changes: 7 additions & 2 deletions dcrwallet.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -534,9 +534,10 @@ func spvLoop(ctx context.Context, w *wallet.Wallet) {
for {
err := syncer.Run(ctx)
if done(ctx) {
loggers.SyncLog.Infof("SPV synchronization stopped")
return
}
log.Errorf("SPV synchronization ended: %v", err)
loggers.SyncLog.Errorf("SPV synchronization stopped: %v", err)
}
}

Expand Down Expand Up @@ -571,7 +572,11 @@ func rpcSyncLoop(ctx context.Context, w *wallet.Wallet) {
syncer := chain.NewSyncer(w, rpcOptions)
err := syncer.Run(ctx)
if err != nil {
loggers.SyncLog.Errorf("Wallet synchronization stopped: %v", err)
if errors.Is(err, context.Canceled) || ctx.Err() != nil {
loggers.SyncLog.Infof("RPC synchronization stopped")
return
}
loggers.SyncLog.Errorf("RPC synchronization stopped: %v", err)
select {
case <-ctx.Done():
return
Expand Down
2 changes: 2 additions & 0 deletions go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,3 +59,5 @@ require (
google.golang.org/genproto/googleapis/rpc v0.0.0-20241223144023-3abc09e42ca8 // indirect
lukechampine.com/blake3 v1.3.0 // indirect
)

replace github.com/decred/dcrd/mixing => github.com/jrick/dcrd/mixing v0.0.0-20250123211715-8a5ce5c2063a
4 changes: 2 additions & 2 deletions go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,8 +50,6 @@ github.com/decred/dcrd/gcs/v4 v4.1.0 h1:tpW7JW53yJZlgNwl/n2NL1b8NxHaIPRUyNuLMkB/
github.com/decred/dcrd/gcs/v4 v4.1.0/go.mod h1:nPTbGM/I3Ihe5KFvUmxZEqQP/jDZQjQ63+WEi/f4lqU=
github.com/decred/dcrd/hdkeychain/v3 v3.1.2 h1:x25WuuE7zM/20EynuVMyOhL0K8BwGBBsexGq8xTiHFA=
github.com/decred/dcrd/hdkeychain/v3 v3.1.2/go.mod h1:FnNJmZ7jqUDeAo6/c/xkQi5cuxh3EWtJeMmW6/Z8lcc=
github.com/decred/dcrd/mixing v0.4.2 h1:mpt2pNIFTI6L1hXrieAWJTQJv5t9WzHcNnhI+tnAG90=
github.com/decred/dcrd/mixing v0.4.2/go.mod h1:VF87lOn41kitgWVOwmXoB4qMYF7+bxItZXyw4JfW3EQ=
github.com/decred/dcrd/rpc/jsonrpc/types/v4 v4.3.0 h1:l0DnCcILTNrpy8APF3FLN312ChpkQaAuW30aC/RgBaw=
github.com/decred/dcrd/rpc/jsonrpc/types/v4 v4.3.0/go.mod h1:j+kkRPXPJB5S9VFOsx8SQLcU7PTFkPKRc1aCHN4ENzA=
github.com/decred/dcrd/rpcclient/v8 v8.0.1 h1:hd81e4w1KSqvPcozJlnz6XJfWKDNuahgooH/N5E8vOU=
Expand All @@ -78,6 +76,8 @@ github.com/jessevdk/go-flags v1.5.0 h1:1jKYvbxEjfUl0fmqTCOfonvskHHXMjBySTLW4y9LF
github.com/jessevdk/go-flags v1.5.0/go.mod h1:Fw0T6WPc1dYxT4mKEZRfG5kJhaTDP9pj1c2EWnYs/m4=
github.com/jrick/bitset v1.0.0 h1:Ws0PXV3PwXqWK2n7Vz6idCdrV/9OrBXgHEJi27ZB9Dw=
github.com/jrick/bitset v1.0.0/go.mod h1:ZOYB5Uvkla7wIEY4FEssPVi3IQXa02arznRaYaAEPe4=
github.com/jrick/dcrd/mixing v0.0.0-20250123211715-8a5ce5c2063a h1:wGTjDa+kmjKBkhnc8BACcU+OreLC+gxTFHLz5t+AFkw=
github.com/jrick/dcrd/mixing v0.0.0-20250123211715-8a5ce5c2063a/go.mod h1:VF87lOn41kitgWVOwmXoB4qMYF7+bxItZXyw4JfW3EQ=
github.com/jrick/logrotate v1.0.0 h1:lQ1bL/n9mBNeIXoTUoYRlK4dHuNJVofX9oWqBtPnSzI=
github.com/jrick/logrotate v1.0.0/go.mod h1:LNinyqDIJnpAur+b8yyulnQw/wDuN1+BYKlTRt3OuAQ=
github.com/jrick/wsrpc/v2 v2.3.8 h1:9vfM8o9g00HXQb/3D6+Y9Cy1uybjD7K1272vtdXXBps=
Expand Down
48 changes: 33 additions & 15 deletions spv/sync.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -321,7 +321,7 @@ func (s *Syncer) setRequiredHeight(tipHeight int32) {
}

// Run synchronizes the wallet, returning when synchronization fails or the
// context is cancelled.
// context is canceled.
func (s *Syncer) Run(ctx context.Context) (err error) {
s.doneMu.Lock()
s.done = make(chan struct{})
Expand Down Expand Up @@ -367,23 +367,23 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
}

// Start background handlers to read received messages from remote peers
g, ctx := errgroup.WithContext(ctx)
g.Go(func() error { return s.receiveGetData(ctx) })
g.Go(func() error { return s.receiveInv(ctx) })
g.Go(func() error { return s.receiveHeadersAnnouncements(ctx) })
g.Go(func() error { return s.receiveMixMsgs(ctx) })
g, gctx := errgroup.WithContext(context.Background())
g.Go(func() error { return s.receiveGetData(gctx) })
g.Go(func() error { return s.receiveInv(gctx) })
g.Go(func() error { return s.receiveHeadersAnnouncements(gctx) })
g.Go(func() error { return s.receiveMixMsgs(gctx) })
s.lp.AddHandledMessages(p2p.MaskGetData | p2p.MaskInv)

if len(s.persistentPeers) != 0 {
for i := range s.persistentPeers {
raddr := s.persistentPeers[i]
g.Go(func() error { return s.connectToPersistent(ctx, raddr) })
g.Go(func() error { return s.connectToPersistent(gctx, raddr) })
}
} else {
g.Go(func() error { return s.connectToCandidates(ctx) })
g.Go(func() error { return s.connectToCandidates(gctx) })
}

g.Go(func() error { return s.handleMempool(ctx) })
g.Go(func() error { return s.handleMempool(gctx) })

s.wallet.SetNetworkBackend(s)
defer s.wallet.SetNetworkBackend(nil)
Expand All @@ -392,7 +392,7 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
g.Go(func() error {
// First step: fetch missing CFilters.
progress := make(chan wallet.MissingCFilterProgress, 1)
go s.wallet.FetchMissingCFiltersWithProgress(ctx, s, progress)
go s.wallet.FetchMissingCFiltersWithProgress(gctx, s, progress)

log.Debugf("Fetching missing CFilters...")
s.fetchMissingCfiltersStart()
Expand All @@ -408,14 +408,14 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
// Next: fetch headers and cfilters up to mainchain tip.
s.fetchHeadersStart()
log.Debugf("Fetching headers and CFilters...")
err = s.initialSyncHeaders(ctx)
err = s.initialSyncHeaders(gctx)
if err != nil {
return err
}
s.fetchHeadersFinished()

// Finally: Perform the initial rescan over the received blocks.
err = s.initialSyncRescan(ctx)
err = s.initialSyncRescan(gctx)
if err != nil {
return err
}
Expand All @@ -425,10 +425,28 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
return nil
})

// Run wallet background goroutines (currently, this just runs
// mixclient).
// Ensure wallet.Run cleanly finishes/is canceled first when outer
// context is canceled.
walletCtx, walletCtxCancel := context.WithCancel(context.Background())
jrick marked this conversation as resolved.
Show resolved Hide resolved
go func() {
select {
case <-ctx.Done():
case <-gctx.Done():
}
walletCtxCancel()
}()
g.Go(func() error {
return s.wallet.Run(ctx)
// Run wallet background goroutines (currently, this just runs
// mixclient).
err := s.wallet.Run(walletCtx)
if err != nil {
return err
}

// If gctx has not yet been canceled, do so here now.
// walletCtx is canceled after either ctx or gctx is canceled.
<-walletCtx.Done()
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This doesn't look right to me yet.

In the case an error is returned from Run unrelated to the context being canceled, that error would be returned which would cause the group context to be canceled so everything would be fine.

However, when wallet.Run returns nil, walletCtx will never be canceled afaict. Thus, it would just hang here.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It will be cancelled by the goroutine started above this.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

https://go.dev/play/p/5VPuDP_Bbtw

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That is assuming there is an error in the goroutines in the group though. There won't necessarily be one.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

https://go.dev/play/p/sYabLXOyzoN

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

but they all return errors....

have you ran this and seen it hang?

Copy link
Member

@davecgh davecgh Jan 24, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I haven't seen any issues, but I'm just considering the edge cases. So long as there is a never a case where the grouped goroutines all exit without an error, then the code is fine.

My worry is just basically what I described. Namely, wallet.Run returns nil and then it just sits and waits, effectively until either there is an error in the group or the outer context is canceled. That part is correct.

However, it implies all of the goroutines in the group will be sitting there doing nothing after wallet.Run returns until eventually the outer context is canceled even though they really should be exiting after wallet.Run returns unconditionally so that syncer.Run returns without error versus sitting around in a defunct state (aka wallet.Run is no longer running, but all of the goroutines in the group are).

EDIT: So long as you're fine with that outcome, I'll approve it.

Copy link
Member

@davecgh davecgh Jan 24, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, after looking through more of the surrounding code, I guess it's just the poor naming of things that misled me and those other goroutines are supposed to remain active until the outer context is canceled regardless of whether or not the "wallet" is running. It'll still be processing p2p messages, fetching filters, etc.

I was completely misled by the naming to believe it was running the entire wallet (as it's named) as opposed to some optional additional background tasks.

return walletCtx.Err()
})

// Wait until cancellation or a handler errors.
Expand Down
Loading