From 8458966f02259ae642024b1ba1e2156638e4cd11 Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Thu, 23 Jun 2022 01:26:38 +0800 Subject: [PATCH] routing: remove the abstraction `shardHandler` This commit removes the unclear abstraction `shardHandler` that's used in our payment lifecycle. As we'll see in the following commits, `shardHandler` is an unnecessary layer and everything can be cleanly managed inside `paymentLifecycle`. --- routing/payment_lifecycle.go | 85 +++++++++++++++--------------------- routing/router.go | 20 +++++---- 2 files changed, 46 insertions(+), 59 deletions(-) diff --git a/routing/payment_lifecycle.go b/routing/payment_lifecycle.go index b2c3834f2..a07542f9a 100644 --- a/routing/payment_lifecycle.go +++ b/routing/payment_lifecycle.go @@ -30,6 +30,19 @@ type paymentLifecycle struct { shardTracker shards.ShardTracker timeoutChan <-chan time.Time currentHeight int32 + + // shardErrors is a channel where errors collected by calling + // collectResultAsync will be delivered. These results are meant to be + // inspected by calling waitForShard or checkShards, and the channel + // doesn't need to be initiated if the caller is using the sync + // collectResult directly. + // TODO(yy): delete. + shardErrors chan error + + // quit is closed to signal the sub goroutines of the payment lifecycle + // to stop. + quit chan struct{} + wg sync.WaitGroup } // newPaymentLifecycle initiates a new payment lifecycle and returns it. @@ -45,6 +58,8 @@ func newPaymentLifecycle(r *ChannelRouter, feeLimit lnwire.MilliSatoshi, paySession: paySession, shardTracker: shardTracker, currentHeight: currentHeight, + shardErrors: make(chan error), + quit: make(chan struct{}), } // If a timeout is specified, create a timeout channel. If no timeout is @@ -78,19 +93,10 @@ func (p *paymentLifecycle) calcFeeBudget( // resumePayment resumes the paymentLifecycle from the current state. func (p *paymentLifecycle) resumePayment() ([32]byte, *route.Route, error) { - shardHandler := &shardHandler{ - router: p.router, - identifier: p.identifier, - shardTracker: p.shardTracker, - shardErrors: make(chan error), - quit: make(chan struct{}), - paySession: p.paySession, - } - // When the payment lifecycle loop exits, we make sure to signal any - // sub goroutine of the shardHandler to exit, then wait for them to + // sub goroutine of the HTLC attempt to exit, then wait for them to // return. - defer shardHandler.stop() + defer p.stop() // If we had any existing attempts outstanding, we'll start by spinning // up goroutines that'll collect their results and deliver them to the @@ -106,7 +112,7 @@ func (p *paymentLifecycle) resumePayment() ([32]byte, *route.Route, error) { log.Infof("Resuming payment shard %v for payment %v", a.AttemptID, p.identifier) - shardHandler.collectResultAsync(&a) + p.collectResultAsync(&a) } // exitWithErr is a helper closure that logs and returns an error. @@ -122,7 +128,7 @@ lifecycle: for { // Start by quickly checking if there are any outcomes already // available to handle before we reevaluate our state. - if err := shardHandler.checkShards(); err != nil { + if err := p.checkShards(); err != nil { return exitWithErr(err) } @@ -187,7 +193,7 @@ lifecycle: // We still have outstanding shards, so wait for a new // outcome to be available before re-evaluating our // state. - if err := shardHandler.waitForShard(); err != nil { + if err := p.waitForShard(); err != nil { return exitWithErr(err) } continue lifecycle @@ -259,7 +265,7 @@ lifecycle: // We still have active shards, we'll wait for an // outcome to be available before retrying. - if err := shardHandler.waitForShard(); err != nil { + if err := p.waitForShard(); err != nil { return exitWithErr(err) } continue lifecycle @@ -272,7 +278,7 @@ lifecycle: lastShard := rt.ReceiverAmt() == ps.RemainingAmt // We found a route to try, launch a new shard. - attempt, outcome, err := shardHandler.launchShard(rt, lastShard) + attempt, outcome, err := p.launchShard(rt, lastShard) switch { // We may get a terminal error if we've processed a shard with // a terminal state (settled or permanent failure), while we @@ -298,7 +304,7 @@ lifecycle: // We must inspect the error to know whether it was // critical or not, to decide whether we should // continue trying. - err := shardHandler.handleSwitchErr( + err := p.handleSwitchErr( attempt, outcome.err, ) if err != nil { @@ -312,39 +318,18 @@ lifecycle: // Now that the shard was successfully sent, launch a go // routine that will handle its result when its back. - shardHandler.collectResultAsync(attempt) + p.collectResultAsync(attempt) } } -// shardHandler holds what is necessary to send and collect the result of -// shards. -type shardHandler struct { - identifier lntypes.Hash - router *ChannelRouter - shardTracker shards.ShardTracker - paySession PaymentSession - - // shardErrors is a channel where errors collected by calling - // collectResultAsync will be delivered. These results are meant to be - // inspected by calling waitForShard or checkShards, and the channel - // doesn't need to be initiated if the caller is using the sync - // collectResult directly. - shardErrors chan error - - // quit is closed to signal the sub goroutines of the payment lifecycle - // to stop. - quit chan struct{} - wg sync.WaitGroup -} - // stop signals any active shard goroutine to exit and waits for them to exit. -func (p *shardHandler) stop() { +func (p *paymentLifecycle) stop() { close(p.quit) p.wg.Wait() } // waitForShard blocks until any of the outstanding shards return. -func (p *shardHandler) waitForShard() error { +func (p *paymentLifecycle) waitForShard() error { select { case err := <-p.shardErrors: return err @@ -359,7 +344,7 @@ func (p *shardHandler) waitForShard() error { // checkShards is a non-blocking method that check if any shards has finished // their execution. -func (p *shardHandler) checkShards() error { +func (p *paymentLifecycle) checkShards() error { for { select { case err := <-p.shardErrors: @@ -400,7 +385,7 @@ type launchOutcome struct { // whether the attempt was successfully sent. If the launchOutcome wraps a // non-nil error, it means that the attempt was not sent onto the network, so // no result will be available in the future for it. -func (p *shardHandler) launchShard(rt *route.Route, +func (p *paymentLifecycle) launchShard(rt *route.Route, lastShard bool) (*channeldb.HTLCAttempt, *launchOutcome, error) { // Using the route received from the payment session, create a new @@ -455,7 +440,7 @@ type shardResult struct { // collectResultAsync launches a goroutine that will wait for the result of the // given HTLC attempt to be available then handle its result. It will fail the // payment with the control tower if a terminal error is encountered. -func (p *shardHandler) collectResultAsync(attempt *channeldb.HTLCAttempt) { +func (p *paymentLifecycle) collectResultAsync(attempt *channeldb.HTLCAttempt) { // errToSend is the error to be sent to sh.shardErrors. var errToSend error @@ -510,7 +495,7 @@ func (p *shardHandler) collectResultAsync(attempt *channeldb.HTLCAttempt) { // collectResult waits for the result for the given attempt to be available // from the Switch, then records the attempt outcome with the control tower. A // shardResult is returned, indicating the final outcome of this HTLC attempt. -func (p *shardHandler) collectResult(attempt *channeldb.HTLCAttempt) ( +func (p *paymentLifecycle) collectResult(attempt *channeldb.HTLCAttempt) ( *shardResult, error) { // We'll retrieve the hash specific to this shard from the @@ -632,7 +617,7 @@ func (p *shardHandler) collectResult(attempt *channeldb.HTLCAttempt) ( } // createNewPaymentAttempt creates a new payment attempt from the given route. -func (p *shardHandler) createNewPaymentAttempt(rt *route.Route, +func (p *paymentLifecycle) createNewPaymentAttempt(rt *route.Route, lastShard bool) (*channeldb.HTLCAttempt, error) { // Generate a new key to be used for this attempt. @@ -683,7 +668,7 @@ func (p *shardHandler) createNewPaymentAttempt(rt *route.Route, // sendAttempt attempts to send the current attempt to the switch to complete // the payment. If this attempt fails, then we'll continue on to the next // available route. -func (p *shardHandler) sendAttempt( +func (p *paymentLifecycle) sendAttempt( attempt *channeldb.HTLCAttempt) error { log.Tracef("Attempting to send payment %v (pid=%v), "+ @@ -744,7 +729,7 @@ func (p *shardHandler) sendAttempt( // the error type, the error is either the final outcome of the payment or we // need to continue with an alternative route. A final outcome is indicated by // a non-nil reason value. -func (p *shardHandler) handleSwitchErr(attempt *channeldb.HTLCAttempt, +func (p *paymentLifecycle) handleSwitchErr(attempt *channeldb.HTLCAttempt, sendErr error) error { internalErrorReason := channeldb.FailureReasonError @@ -841,7 +826,7 @@ func (p *shardHandler) handleSwitchErr(attempt *channeldb.HTLCAttempt, // handleFailureMessage tries to apply a channel update present in the failure // message if any. -func (p *shardHandler) handleFailureMessage(rt *route.Route, +func (p *paymentLifecycle) handleFailureMessage(rt *route.Route, errorSourceIdx int, failure lnwire.FailureMessage) error { if failure == nil { @@ -913,7 +898,7 @@ func (p *shardHandler) handleFailureMessage(rt *route.Route, } // failAttempt calls control tower to fail the current payment attempt. -func (p *shardHandler) failAttempt(attemptID uint64, +func (p *paymentLifecycle) failAttempt(attemptID uint64, sendError error) (*channeldb.HTLCAttempt, error) { log.Warnf("Attempt %v for payment %v failed: %v", attemptID, diff --git a/routing/router.go b/routing/router.go index fa255544c..2e679b4bb 100644 --- a/routing/router.go +++ b/routing/router.go @@ -2496,15 +2496,17 @@ func (r *ChannelRouter) sendToRoute(htlcHash lntypes.Hash, rt *route.Route, // shard we'll now launch. shardTracker := shards.NewSimpleShardTracker(htlcHash, nil) - // Launch a shard along the given route. - sh := &shardHandler{ - router: r, - identifier: paymentIdentifier, - shardTracker: shardTracker, - } + // Create a payment lifecycle using the given route with, + // - zero fee limit as we are not requesting routes. + // - nil payment session (since we already have a route). + // - no payment timeout. + // - no current block height. + p := newPaymentLifecycle( + r, 0, paymentIdentifier, nil, shardTracker, 0, 0, + ) var shardError error - attempt, outcome, err := sh.launchShard(rt, false) + attempt, outcome, err := p.launchShard(rt, false) // With SendToRoute, it can happen that the route exceeds protocol // constraints. Mark the payment as failed with an internal error. @@ -2536,7 +2538,7 @@ func (r *ChannelRouter) sendToRoute(htlcHash lntypes.Hash, rt *route.Route, // Shard successfully launched, wait for the result to be available. default: - result, err := sh.collectResult(attempt) + result, err := p.collectResult(attempt) if err != nil { return nil, err } @@ -2556,7 +2558,7 @@ func (r *ChannelRouter) sendToRoute(htlcHash lntypes.Hash, rt *route.Route, // the error to check if it maps into a terminal error code, if not use // a generic NO_ROUTE error. var failureReason *channeldb.FailureReason - err = sh.handleSwitchErr(attempt, shardError) + err = p.handleSwitchErr(attempt, shardError) switch { // If a non-terminal error is returned and `skipTempErr` is false, then