htlcswitch: use fn.GoroutineManager #9140
Conversation
|
Important Review skippedAuto reviews are limited to specific labels. 🏷️ Labels to auto review (1)
Please check the settings in the CodeRabbit UI or the You can disable this status message by setting the Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
starius
force-pushed
the
goroutines
branch
2 times, most recently
from
8810118 to
88fbc4b
Compare
starius
force-pushed
the
goroutines
branch
2 times, most recently
from
8395cca to
e001027
Compare
starius
changed the title
|
@starius - I think these unit test failures are related to this PR - maybe take a look at fixing those up first & then re-ping reviewers when ready? |
htlcswitch/switch.go
Outdated
| }() | ||
| }) | ||
| if err != nil { | ||
| return |
There was a problem hiding this comment.
don't think this should return?
There was a problem hiding this comment.
Fixed, added a comment. Now this section looks like this:
// When this time ticks, then it indicates that we should
// collect all the forwarding events since the last internal,
// and write them out to our log.
case <-s.cfg.FwdEventTicker.Ticks():
// The error of Go is ignored: if it is shutting down,
// the loop will terminate on the next iteration, in
// s.gm.Done case.
_ = s.gm.Go(func(ctx context.Context) {
err := s.FlushForwardingEvents()
if err != nil {
log.Errorf("unable to flush "+
"forwarding events: %v", err)
}
})
I pushed branch reproduce-race to my fork. In that branch: |
starius
force-pushed
the
goroutines
branch
2 times, most recently
from
7cb95ef to
662c47b
Compare
Test failure was caused by extra call to s.Stop in |
There was a problem hiding this comment.
Thanks for the updates @starius!
Logic looks good, but I have some opinions about the API of the fn.Go call that I think is worth discussing before we merge. Would love to hear what @yyforyongyu & @ProofOfKeags think too.
|
What's the prio on this? I want to review but I need to balance with other stuff. |
Not critical. You can focus on P0 stuff, before addressing this. |
There was a problem hiding this comment.
Sorry a bit late in the game, but is there an issue page describing what the issue is?
I also don't understand the struct GoroutineManager - it looks like it's putting a mutex to guard the wait group operations?
My instinct is this is solving the wrong problem - we should always know when/where we call wg.Add and wg.Wait, if not, we should refactor our code so we always know when we cal wg.Add and wg.Wait. I guess other people have run into this issue before too.
This was requested in lightningnetwork#9140 (comment)
@yyforyongyu Thank you for the suggestion!
In this setup,
I agree that, ideally, the code should be refactored into an event-loop style, centralizing all goroutine launches and state changes within a single goroutine and using channels to transmit data to and from it. This approach aligns with the patterns we follow in other packages. However, implementing such a change would require significant time and extensive modifications to the package. What are your thoughts? |
|
I squashed the last commit (deeacc6), rebased and used GoroutineManager from fn v2. Fortunately fn v1 and fn v2 can be used simultaneously! |
There was a problem hiding this comment.
Thanks for the updates, I think things look good but i think we should change the API of the goroutine manager a bit more. See my suggestion here
htlcswitch/switch.go
Outdated
| @@ -836,7 +847,8 @@ func (s *Switch) logFwdErrs(num *int, wg *sync.WaitGroup, fwdChan chan error) { | |||
| log.Errorf("Unhandled error while reforwarding htlc "+ | |||
| "settle/fail over htlcswitch: %v", err) | |||
| } | |||
| case <-s.quit: | |||
|
|
|||
| case <-ctx.Done(): | |||
There was a problem hiding this comment.
something doesnt feel right here. It feels like we are mixing the use of caller ctx and quit channels. Here, they mean the same thing: so ie, why cant we just listen on s.gm.Done() here (ie, s.quit)? because this ctx that is now being passed in here is not coming from the caller of ForwardPackets and is instead coming from the creator of the the gm. I think the issue is stemming from the fact that we are passing a context to the constructor of the goroutine manager which is an anti-pattern. Im gonna see if I can work the goroutine manager a bit to work around this anti-pattern
There was a problem hiding this comment.
Thanks! I replaced ctx.Done() with s.gm.Done() here and also inside a goroutine launched by GetAttemptResult.
htlcswitch/switch.go
Outdated
| @@ -368,8 +370,11 @@ func New(cfg Config, currentHeight uint32) (*Switch, error) { | |||
| return nil, err | |||
| } | |||
|
|
|||
| gm := fn2.NewGoroutineManager(context.Background()) | |||
There was a problem hiding this comment.
it's an anti-pattern to pass a context into a constructor. I think we should try to avoid this as much as possible. I'll put up a suggested diff for the goroutine manager 👍
There was a problem hiding this comment.
Thanks! I updated fn dependency and used new API!
starius
force-pushed
the
goroutines
branch
2 times, most recently
from
c51f5ab to
1a18ed4
Compare
htlcswitch/switch.go
Outdated
| @@ -85,6 +86,9 @@ var ( | |||
| // fail payments if they increase our fee exposure. This is currently | |||
| // set to 500m msats. | |||
| DefaultMaxFeeExposure = lnwire.MilliSatoshi(500_000_000) | |||
|
|
|||
| // background is a shortcut for context.Background. | |||
| background = context.Background() | |||
There was a problem hiding this comment.
i dont think we should do this. Rather use a context.TODO() where needed
There was a problem hiding this comment.
if you rebase on top of #9344, then we can also add a context guard here and then we only need a single context.TODO() in Start()
There was a problem hiding this comment.
Fixed.
There are now 3 top-level methods left which use context.TODO() now:
- Start
- ForwardPackets
- GetAttemptResult
Probably they should get a context argument in the future and it will replace the context.TODO().
protofsm/state_machine.go
Outdated
| // background is a shortcut for context.Background. | ||
| background = context.Background() |
There was a problem hiding this comment.
we should not do this.
consider rebasing on top of #9342 which handles the bump to the correct fn version and handles updating the statemachine to thread contexts through correctly
There was a problem hiding this comment.
Done. This commit is not needed now.
htlcswitch/switch.go
Outdated
| var n *networkResult | ||
| select { | ||
| case n = <-nChan: | ||
| case <-s.quit: | ||
| case <-s.gm.Done(): |
There was a problem hiding this comment.
i think it is not great to refer to s.gm from inside a call-back that is called from s.gm (it screams "deadlock"). Rather just use the ctx provided to the callback which will be cancelled when the gm is shutdown (ie, when gm.Done() would have returned anyways)
htlcswitch/switch.go
Outdated
| // The error of Go is ignored: if it is shutting down, | ||
| // the loop will terminate on the next iteration, in | ||
| // s.gm.Done case. | ||
| _ = s.gm.Go(background, func(ctx context.Context) { |
There was a problem hiding this comment.
let htlcForwarder take a context and pass in a context in there from the goroutine which is starting it
htlcswitch/switch.go
Outdated
| @@ -3020,8 +3042,12 @@ func (s *Switch) handlePacketSettle(packet *htlcPacket) error { | |||
| // NOTE: `closeCircuit` modifies the state of `packet`. | |||
| if localHTLC { | |||
| // TODO(yy): remove the goroutine and send back the error here. | |||
| s.wg.Add(1) | |||
| go s.handleLocalResponse(packet) | |||
| ok := s.gm.Go(background, func(ctx context.Context) { | |||
There was a problem hiding this comment.
rather pass in a context to the calling func. Same for all the others
There was a problem hiding this comment.
Fixed.
Another instance is handlePacketFail.
|
@starius - those 2 PRs are in now so I think we can continue here |
starius
force-pushed
the
goroutines
branch
2 times, most recently
from
3f7a66f to
7ce33f8
Compare
Replaced the use of s.quit and s.wg with s.gm (GoroutineManager). This fixes a race condition between s.wg.Add(1) and s.wg.Wait(). Also added a test which used to fail under `-race` before this commit.
|
Squashed the commits. |
There was a problem hiding this comment.
I agree that, ideally, the code should be refactored into an event-loop style, centralizing all goroutine launches and state changes within a single goroutine and using channels to transmit data to and from it. This approach aligns with the patterns we follow in other packages. However, implementing such a change would require significant time and extensive modifications to the package. What are your thoughts?
Have you tried the event loop approach? At a glance I think we only need to add a new channel receiver attemptResultReq on Switch and read it in the main loop htlcForwader? Seems doable as the diff is small https://gist.github.com/yyforyongyu/7cf8d2e2586b2c38d197e05315b9d55d
The other approach is simply removing the wg.Add - why do we need it or am I missing anything here?
diff --git a/htlcswitch/switch.go b/htlcswitch/switch.go
index 720625f2c..5e11ce794 100644
--- a/htlcswitch/switch.go
+++ b/htlcswitch/switch.go
@@ -493,10 +493,7 @@ func (s *Switch) GetAttemptResult(attemptID uint64, paymentHash lntypes.Hash,
// Since the attempt was known, we can start a goroutine that can
// extract the result when it is available, and pass it on to the
// caller.
- s.wg.Add(1)
go func() {
- defer s.wg.Done()
-
var n *networkResult
select {
case n = <-nChan:
@@ -518,12 +515,15 @@ func (s *Switch) GetAttemptResult(attemptID uint64, paymentHash lntypes.Hash,
if err != nil {
e := fmt.Errorf("unable to extract result: %w", err)
log.Error(e)
- resultChan <- &PaymentResult{
- Error: e,
- }
+ fn.SendOrQuit(
+ resultChan, &PaymentResult{
+ Error: e,
+ }, s.quit,
+ )
return
}
- resultChan <- result
+
+ fn.SendOrQuit(resultChan, result, s.quit)
}()
return resultChan, nilI think we are more or less on the same page, as we know it's a temporary mitigation to the issue. And I wanna stress again about the wrong usage of wg.Add(1), as explained from this OG comment.
Or my question is this - now that we have the new fn.GoroutineManager, how are we gonna prevent future development from using it to cover the mistake that a wg.Add is called inside a goroutine?
| }() | ||
| }) | ||
| // The switch shutting down is signaled by closing the channel. | ||
| if !ok { |
There was a problem hiding this comment.
Why do we still need this check? Won't the line <-ctx.Done() be hit when it's shutting down?
There was a problem hiding this comment.
If GoroutineManager.Stop is called before the Go method (i.e., the switch is in the process of stopping), the Go method will return false without launching a new goroutine. In such cases, we should perform the same action as if it had stopped after launching the goroutine - specifically, closing resultChan. Failing to close the channel and simply returning it could cause the caller to get stuck indefinitely while waiting to receive from the channel.
| ok := s.gm.Go(context.TODO(), func(ctx context.Context) { | ||
| s.logFwdErrs(ctx, &numSent, &wg, fwdChan) | ||
| }) | ||
| if !ok { |
There was a problem hiding this comment.
same here - why do we need this check? I think s.logFwdErrs will listen on <-s.gm.Done(): and quit?
There was a problem hiding this comment.
This is a similar situation. We should handle it the same way if the goroutine manager is stopped before the Go method is executed.
If we remove |
|
@ProofOfKeags: review reminder |
Change Description
Replaced the use of
s.quitands.wgwiths.gm(GoroutineManager). WaitGroup is still needed to wait forhandleLocalResponse: if it was switched tos.gm, then it may skip running, which has unclear consequences. AfterhandleLocalResponseis changed to run without a goroutine, we can remove WaitGroup completely.This fixes a race condition between
s.wg.Add(1)ands.wg.Wait().Steps to Test
I added a test which used to fail under
-racebefore this commit.This test crashes with a data race if I undo the changes of implementation of switch.
Pull Request Checklist
Testing
Code Style and Documentation
[skip ci]in the commit message for small changes.