12583319996

Committed 02 Jan 2025 01:38PM UTC coverage: 57.522% (-1.1%) from 58.598%

Build # 12583319996

Build Type

Pull #9361

github

Committed by

starius

Commit Message

fn/ContextGuard: use context.AfterFunc to wait

Simplifies context cancellation handling by using context.AfterFunc instead of a
goroutine to wait for context cancellation. This approach avoids the overhead of
a goroutine during the waiting period.

For ctxQuitUnsafe, since g.quit is closed only in the Quit method (which also
cancels all associated contexts), waiting on context cancellation ensures the
same behavior without unnecessary dependency on g.quit.

Added a test to ensure that the Create method does not launch any goroutines.

Pull Request Pull Request #9361: fn: optimize context guard

Run Details

102587 of 178344 relevant lines covered (57.52%)

24734.33 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/channel_notifier.go

package lnd

import (
        "fmt"

        "github.com/btcsuite/btcd/wire"
        "github.com/lightningnetwork/lnd/chanbackup"
        "github.com/lightningnetwork/lnd/channeldb"
        "github.com/lightningnetwork/lnd/channelnotifier"
)

// channelNotifier is an implementation of the chanbackup.ChannelNotifier
// interface using the existing channelnotifier.ChannelNotifier struct. This
// implementation allows us to satisfy all the dependencies of the
// chanbackup.SubSwapper struct.
type channelNotifier struct {
        // chanNotifier is the based channel notifier that we'll proxy requests
        // from.
        chanNotifier *channelnotifier.ChannelNotifier

        // addrs is an implementation of the addrSource interface that allows
        // us to get the latest set of addresses for a given node. We'll need
        // this to be able to create an SCB for new channels.
        addrs channeldb.AddrSource
}

// SubscribeChans requests a new channel subscription relative to the initial
// set of known channels. We use the knownChans as a synchronization point to
// ensure that the chanbackup.SubSwapper does not miss any channel open or
// close events in the period between when it's created, and when it requests
// the channel subscription.
//
// NOTE: This is part of the chanbackup.ChannelNotifier interface.
func (c *channelNotifier) SubscribeChans(startingChans map[wire.OutPoint]struct{}) (
        *chanbackup.ChannelSubscription, error) {

        ltndLog.Infof("Channel backup proxy channel notifier starting")

        // TODO(roasbeef): read existing set of chans and diff

        quit := make(chan struct{})
        chanUpdates := make(chan chanbackup.ChannelEvent, 1)

        // sendChanOpenUpdate is a closure that sends a ChannelEvent to the
        // chanUpdates channel to inform subscribers about new pending or
        // confirmed channels.
        sendChanOpenUpdate := func(newOrPendingChan *channeldb.OpenChannel) {
                _, nodeAddrs, err := c.addrs.AddrsForNode(
                        newOrPendingChan.IdentityPub,
                )
                if err != nil {
                        pub := newOrPendingChan.IdentityPub
                        ltndLog.Errorf("unable to fetch addrs for %x: %v",
                                pub.SerializeCompressed(), err)
                }

                chanEvent := chanbackup.ChannelEvent{
                        NewChans: []chanbackup.ChannelWithAddrs{
                                {
                                        OpenChannel: newOrPendingChan,
                                        Addrs:       nodeAddrs,
                                },
                        },
                }

                select {
                case chanUpdates <- chanEvent:
                case <-quit:
                        return
                }
        }

        // In order to adhere to the interface, we'll proxy the events from the
        // channel notifier to the sub-swapper in a format it understands.
        go func() {
                // First, we'll subscribe to the primary channel notifier so we can
                // obtain events for new opened/closed channels.
                chanSubscription, err := c.chanNotifier.SubscribeChannelEvents()
                if err != nil {
                        panic(fmt.Sprintf("unable to subscribe to chans: %v",
                                err))
                }

                defer chanSubscription.Cancel()

                for {
                        select {

                        // A new event has been sent by the chanNotifier, we'll
                        // filter out the events we actually care about and
                        // send them to the sub-swapper.
                        case e := <-chanSubscription.Updates():
                                // TODO(roasbeef): batch dispatch ntnfs

                                switch event := e.(type) {
                                // A new channel has been opened and is still
                                // pending. We can still create a backup, even
                                // if the final channel ID is not yet available.
                                case channelnotifier.PendingOpenChannelEvent:
                                        pendingChan := event.PendingChannel
                                        sendChanOpenUpdate(pendingChan)

                                // A new channel has been confirmed, we'll
                                // obtain the node address, then send to the
                                // sub-swapper.
                                case channelnotifier.OpenChannelEvent:
                                        sendChanOpenUpdate(event.Channel)

                                // An existing channel has been closed, we'll
                                // send only the chanPoint of the closed
                                // channel to the sub-swapper.
                                case channelnotifier.ClosedChannelEvent:
                                        chanPoint := event.CloseSummary.ChanPoint
                                        closeType := event.CloseSummary.CloseType

                                        // Because we see the contract as closed
                                        // once our local force close TX
                                        // confirms, the channel arbitrator
                                        // already fires on this event. But
                                        // because our funds can be in limbo for
                                        // up to 2 weeks worst case we don't
                                        // want to remove the crucial info we
                                        // need for sweeping that time locked
                                        // output before we've actually done so.
                                        if closeType == channeldb.LocalForceClose {
                                                ltndLog.Debugf("Channel %v "+
                                                        "was force closed by "+
                                                        "us, not removing "+
                                                        "from channel backup "+
                                                        "until fully resolved",
                                                        chanPoint)

                                                continue
                                        }

                                        chanEvent := chanbackup.ChannelEvent{
                                                ClosedChans: []wire.OutPoint{
                                                        chanPoint,
                                                },
                                        }

                                        select {
                                        case chanUpdates <- chanEvent:
                                        case <-quit:
                                                return
                                        }

                                // A channel was fully resolved on chain. This
                                // should only really interest us if it was a
                                // locally force closed channel where we didn't
                                // remove the channel already when the close
                                // event was fired.
                                case channelnotifier.FullyResolvedChannelEvent:
                                        chanEvent := chanbackup.ChannelEvent{
                                                ClosedChans: []wire.OutPoint{
                                                        *event.ChannelPoint,
                                                },
                                        }

                                        select {
                                        case chanUpdates <- chanEvent:
                                        case <-quit:
                                                return
                                        }
                                }

                        // The cancel method has been called, signalling us to
                        // exit
                        case <-quit:
                                return
                        }
                }
        }()

        return &chanbackup.ChannelSubscription{
                ChanUpdates: chanUpdates,
                Cancel: func() {
                        close(quit)
                },
        }, nil
}

// A compile-time constraint to ensure channelNotifier implements
// chanbackup.ChannelNotifier.
var _ chanbackup.ChannelNotifier = (*channelNotifier)(nil)

1	package lnd
2
3	import (
4	"fmt"
5
6	"github.com/btcsuite/btcd/wire"
7	"github.com/lightningnetwork/lnd/chanbackup"
8	"github.com/lightningnetwork/lnd/channeldb"
9	"github.com/lightningnetwork/lnd/channelnotifier"
10	)
11
12	// channelNotifier is an implementation of the chanbackup.ChannelNotifier
13	// interface using the existing channelnotifier.ChannelNotifier struct. This
14	// implementation allows us to satisfy all the dependencies of the
15	// chanbackup.SubSwapper struct.
16	type channelNotifier struct {
17	// chanNotifier is the based channel notifier that we'll proxy requests
18	// from.
19	chanNotifier *channelnotifier.ChannelNotifier
20
21	// addrs is an implementation of the addrSource interface that allows
22	// us to get the latest set of addresses for a given node. We'll need
23	// this to be able to create an SCB for new channels.
24	addrs channeldb.AddrSource
25	}
26
27	// SubscribeChans requests a new channel subscription relative to the initial
28	// set of known channels. We use the knownChans as a synchronization point to
29	// ensure that the chanbackup.SubSwapper does not miss any channel open or
30	// close events in the period between when it's created, and when it requests
31	// the channel subscription.
32	//
33	// NOTE: This is part of the chanbackup.ChannelNotifier interface.
34	func (c *channelNotifier) SubscribeChans(startingChans map[wire.OutPoint]struct{}) (
35	*chanbackup.ChannelSubscription, error) {	×
36		×
37	ltndLog.Infof("Channel backup proxy channel notifier starting")	×
38		×
39	// TODO(roasbeef): read existing set of chans and diff	×
40		×
41	quit := make(chan struct{})	×
42	chanUpdates := make(chan chanbackup.ChannelEvent, 1)	×
43		×
44	// sendChanOpenUpdate is a closure that sends a ChannelEvent to the	×
45	// chanUpdates channel to inform subscribers about new pending or	×
46	// confirmed channels.	×
47	sendChanOpenUpdate := func(newOrPendingChan *channeldb.OpenChannel) {	×
48	_, nodeAddrs, err := c.addrs.AddrsForNode(	×
49	newOrPendingChan.IdentityPub,	×
50	)	×
51	if err != nil {	×
52	pub := newOrPendingChan.IdentityPub	×
53	ltndLog.Errorf("unable to fetch addrs for %x: %v",	×
54	pub.SerializeCompressed(), err)	×
55	}	×
56
57	chanEvent := chanbackup.ChannelEvent{	×
58	NewChans: []chanbackup.ChannelWithAddrs{	×
59	{	×
60	OpenChannel: newOrPendingChan,	×
61	Addrs: nodeAddrs,	×
62	},	×
63	},	×
64	}	×
65		×
66	select {	×
67	case chanUpdates <- chanEvent:	×
68	case <-quit:	×
69	return	×
70	}
71	}
72
73	// In order to adhere to the interface, we'll proxy the events from the
74	// channel notifier to the sub-swapper in a format it understands.
75	go func() {	×
76	// First, we'll subscribe to the primary channel notifier so we can	×
77	// obtain events for new opened/closed channels.	×
78	chanSubscription, err := c.chanNotifier.SubscribeChannelEvents()	×
79	if err != nil {	×
80	panic(fmt.Sprintf("unable to subscribe to chans: %v",	×
81	err))	×
82	}
83
84	defer chanSubscription.Cancel()	×
85		×
86	for {	×
87	select {	×
88
89	// A new event has been sent by the chanNotifier, we'll
90	// filter out the events we actually care about and
91	// send them to the sub-swapper.
92	case e := <-chanSubscription.Updates():	×
93	// TODO(roasbeef): batch dispatch ntnfs	×
94		×
95	switch event := e.(type) {	×
96	// A new channel has been opened and is still
97	// pending. We can still create a backup, even
98	// if the final channel ID is not yet available.
99	case channelnotifier.PendingOpenChannelEvent:	×
100	pendingChan := event.PendingChannel	×
101	sendChanOpenUpdate(pendingChan)	×
102
103	// A new channel has been confirmed, we'll
104	// obtain the node address, then send to the
105	// sub-swapper.
106	case channelnotifier.OpenChannelEvent:	×
107	sendChanOpenUpdate(event.Channel)	×
108
109	// An existing channel has been closed, we'll
110	// send only the chanPoint of the closed
111	// channel to the sub-swapper.
112	case channelnotifier.ClosedChannelEvent:	×
113	chanPoint := event.CloseSummary.ChanPoint	×
114	closeType := event.CloseSummary.CloseType	×
115		×
116	// Because we see the contract as closed	×
117	// once our local force close TX	×
118	// confirms, the channel arbitrator	×
119	// already fires on this event. But	×
120	// because our funds can be in limbo for	×
121	// up to 2 weeks worst case we don't	×
122	// want to remove the crucial info we	×
123	// need for sweeping that time locked	×
124	// output before we've actually done so.	×
125	if closeType == channeldb.LocalForceClose {	×
126	ltndLog.Debugf("Channel %v "+	×
127	"was force closed by "+	×
128	"us, not removing "+	×
129	"from channel backup "+	×
130	"until fully resolved",	×
131	chanPoint)	×
132		×
133	continue	×
134	}
135
136	chanEvent := chanbackup.ChannelEvent{	×
137	ClosedChans: []wire.OutPoint{	×
138	chanPoint,	×
139	},	×
140	}	×
141		×
142	select {	×
143	case chanUpdates <- chanEvent:	×
144	case <-quit:	×
145	return	×
146	}
147
148	// A channel was fully resolved on chain. This
149	// should only really interest us if it was a
150	// locally force closed channel where we didn't
151	// remove the channel already when the close
152	// event was fired.
153	case channelnotifier.FullyResolvedChannelEvent:	×
154	chanEvent := chanbackup.ChannelEvent{	×
155	ClosedChans: []wire.OutPoint{	×
156	*event.ChannelPoint,	×
157	},	×
158	}	×
159		×
160	select {	×
161	case chanUpdates <- chanEvent:	×
162	case <-quit:	×
163	return	×
164	}
165	}
166
167	// The cancel method has been called, signalling us to
168	// exit
169	case <-quit:	×
170	return	×
171	}
172	}
173	}()
174
175	return &chanbackup.ChannelSubscription{	×
176	ChanUpdates: chanUpdates,	×
177	Cancel: func() {	×
178	close(quit)	×
179	},	×
180	}, nil
181	}
182
183	// A compile-time constraint to ensure channelNotifier implements
184	// chanbackup.ChannelNotifier.
185	var _ chanbackup.ChannelNotifier = (*channelNotifier)(nil)

lightningnetwork / lnd / 12583319996

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous