13423774416

Committed 19 Feb 2025 10:39PM UTC coverage: 49.338% (-9.5%) from 58.794%

Build # 13423774416

Build Type

Pull #9484

github

Committed by

Abdulkbk

Commit Message

lnutils: add createdir util function

This utility function replaces repetitive logic patterns
throughout LND.

Pull Request Pull Request #9484: lnutils: add createDir util function

Run Details

0 of 13 new or added lines in 1 file covered. (0.0%)

27149 existing lines in 431 files now uncovered.

100732 of 204168 relevant lines covered (49.34%)

1.54 hits per line

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

80.6

/chainio/dispatcher.go

package chainio

import (
        "errors"
        "fmt"
        "sync"
        "sync/atomic"
        "time"

        "github.com/btcsuite/btclog/v2"
        "github.com/lightningnetwork/lnd/chainntnfs"
        "github.com/lightningnetwork/lnd/lnutils"
        "golang.org/x/sync/errgroup"
)

// DefaultProcessBlockTimeout is the timeout value used when waiting for one
// consumer to finish processing the new block epoch.
var DefaultProcessBlockTimeout = 60 * time.Second

// ErrProcessBlockTimeout is the error returned when a consumer takes too long
// to process the block.
var ErrProcessBlockTimeout = errors.New("process block timeout")

// BlockbeatDispatcher is a service that handles dispatching new blocks to
// `lnd`'s subsystems. During startup, subsystems that are block-driven should
// implement the `Consumer` interface and register themselves via
// `RegisterQueue`. When two subsystems are independent of each other, they
// should be registered in different queues so blocks are notified concurrently.
// Otherwise, when living in the same queue, the subsystems are notified of the
// new blocks sequentially, which means it's critical to understand the
// relationship of these systems to properly handle the order.
type BlockbeatDispatcher struct {
        wg sync.WaitGroup

        // notifier is used to receive new block epochs.
        notifier chainntnfs.ChainNotifier

        // beat is the latest blockbeat received.
        beat Blockbeat

        // consumerQueues is a map of consumers that will receive blocks. Its
        // key is a unique counter and its value is a queue of consumers. Each
        // queue is notified concurrently, and consumers in the same queue is
        // notified sequentially.
        consumerQueues map[uint32][]Consumer

        // counter is used to assign a unique id to each queue.
        counter atomic.Uint32

        // quit is used to signal the BlockbeatDispatcher to stop.
        quit chan struct{}
}

// NewBlockbeatDispatcher returns a new blockbeat dispatcher instance.
func NewBlockbeatDispatcher(n chainntnfs.ChainNotifier) *BlockbeatDispatcher {
        return &BlockbeatDispatcher{
                notifier:       n,
                quit:           make(chan struct{}),
                consumerQueues: make(map[uint32][]Consumer),
        }
}

// RegisterQueue takes a list of consumers and registers them in the same
// queue.
//
// NOTE: these consumers are notified sequentially.
func (b *BlockbeatDispatcher) RegisterQueue(consumers []Consumer) {
        qid := b.counter.Add(1)

        b.consumerQueues[qid] = append(b.consumerQueues[qid], consumers...)
        clog.Infof("Registered queue=%d with %d blockbeat consumers", qid,
                len(consumers))

        for _, c := range consumers {
                clog.Debugf("Consumer [%s] registered in queue %d", c.Name(),
                        qid)
        }
}

// Start starts the blockbeat dispatcher - it registers a block notification
// and monitors and dispatches new blocks in a goroutine. It will refuse to
// start if there are no registered consumers.
func (b *BlockbeatDispatcher) Start() error {
        // Make sure consumers are registered.
        if len(b.consumerQueues) == 0 {
                return fmt.Errorf("no consumers registered")
        }

        // Start listening to new block epochs. We should get a notification
        // with the current best block immediately.
        blockEpochs, err := b.notifier.RegisterBlockEpochNtfn(nil)
        if err != nil {
                return fmt.Errorf("register block epoch ntfn: %w", err)
        }

        clog.Infof("BlockbeatDispatcher is starting with %d consumer queues",
                len(b.consumerQueues))
        defer clog.Debug("BlockbeatDispatcher started")

        b.wg.Add(1)
        go b.dispatchBlocks(blockEpochs)

        return nil
}

// Stop shuts down the blockbeat dispatcher.
func (b *BlockbeatDispatcher) Stop() {
        clog.Info("BlockbeatDispatcher is stopping")
        defer clog.Debug("BlockbeatDispatcher stopped")

        // Signal the dispatchBlocks goroutine to stop.
        close(b.quit)
        b.wg.Wait()
}

func (b *BlockbeatDispatcher) log() btclog.Logger {
        return b.beat.logger()
}

// dispatchBlocks listens to new block epoch and dispatches it to all the
// consumers. Each queue is notified concurrently, and the consumers in the
// same queue are notified sequentially.
//
// NOTE: Must be run as a goroutine.
func (b *BlockbeatDispatcher) dispatchBlocks(
        blockEpochs *chainntnfs.BlockEpochEvent) {

        defer b.wg.Done()
        defer blockEpochs.Cancel()

        for {
                select {
                case blockEpoch, ok := <-blockEpochs.Epochs:
                        if !ok {
                                clog.Debugf("Block epoch channel closed")

                                return
                        }

                        // Log a separator so it's easier to identify when a
                        // new block arrives for subsystems.
                        clog.Debugf("%v", lnutils.NewSeparatorClosure())

                        clog.Infof("Received new block %v at height %d, "+
                                "notifying consumers...", blockEpoch.Hash,
                                blockEpoch.Height)

                        // Record the time it takes the consumer to process
                        // this block.
                        start := time.Now()

                        // Update the current block epoch.
                        b.beat = NewBeat(*blockEpoch)

                        // Notify all consumers.
                        err := b.notifyQueues()
                        if err != nil {
                                b.log().Errorf("Notify block failed: %v", err)
                        }

                        b.log().Infof("Notified all consumers on new block "+
                                "in %v", time.Since(start))

                case <-b.quit:
                        b.log().Debugf("BlockbeatDispatcher quit signal " +
                                "received")

                        return
                }
        }
}

// notifyQueues notifies each queue concurrently about the latest block epoch.
func (b *BlockbeatDispatcher) notifyQueues() error {
        // errChans is a map of channels that will be used to receive errors
        // returned from notifying the consumers.
        errChans := make(map[uint32]chan error, len(b.consumerQueues))

        // Notify each queue in goroutines.
        for qid, consumers := range b.consumerQueues {
                b.log().Debugf("Notifying queue=%d with %d consumers", qid,
                        len(consumers))

                // Create a signal chan.
                errChan := make(chan error, 1)
                errChans[qid] = errChan

                // Notify each queue concurrently.
                go func(qid uint32, c []Consumer, beat Blockbeat) {
                        // Notify each consumer in this queue sequentially.
                        errChan <- DispatchSequential(beat, c)
                }(qid, consumers, b.beat)
        }

        // Wait for all consumers in each queue to finish.
        for qid, errChan := range errChans {
                select {
                case err := <-errChan:
                        if err != nil {
                                return fmt.Errorf("queue=%d got err: %w", qid,
                                        err)
                        }

                        b.log().Debugf("Notified queue=%d", qid)

                case <-b.quit:
                        b.log().Debugf("BlockbeatDispatcher quit signal " +
                                "received, exit notifyQueues")

                        return nil
                }
        }

        return nil
}

// DispatchSequential takes a list of consumers and notify them about the new
// epoch sequentially. It requires the consumer to finish processing the block
// within the specified time, otherwise a timeout error is returned.
func DispatchSequential(b Blockbeat, consumers []Consumer) error {
        for _, c := range consumers {
                // Send the beat to the consumer.
                err := notifyAndWait(b, c, DefaultProcessBlockTimeout)
                if err != nil {
                        b.logger().Errorf("Failed to process block: %v", err)

                        return err
                }
        }

        return nil
}

// DispatchConcurrent notifies each consumer concurrently about the blockbeat.
// It requires the consumer to finish processing the block within the specified
// time, otherwise a timeout error is returned.
func DispatchConcurrent(b Blockbeat, consumers []Consumer) error {
        eg := &errgroup.Group{}

        // Notify each queue in goroutines.
        for _, c := range consumers {
                // Notify each consumer concurrently.
                eg.Go(func() error {
                        // Send the beat to the consumer.
                        err := notifyAndWait(b, c, DefaultProcessBlockTimeout)

                        // Exit early if there's no error.
                        if err == nil {
                                return nil
                        }

                        b.logger().Errorf("Consumer=%v failed to process "+
                                "block: %v", c.Name(), err)

                        return err
                })
        }

        // Wait for all consumers in each queue to finish.
        if err := eg.Wait(); err != nil {
                return err
        }

        return nil
}

// notifyAndWait sends the blockbeat to the specified consumer. It requires the
// consumer to finish processing the block within the specified time, otherwise
// a timeout error is returned.
func notifyAndWait(b Blockbeat, c Consumer, timeout time.Duration) error {
        b.logger().Debugf("Waiting for consumer[%s] to process it", c.Name())

        // Record the time it takes the consumer to process this block.
        start := time.Now()

        errChan := make(chan error, 1)
        go func() {
                errChan <- c.ProcessBlock(b)
        }()

        // We expect the consumer to finish processing this block under 30s,
        // otherwise a timeout error is returned.
        select {
        case err := <-errChan:
                if err == nil {
                        break
                }

                return fmt.Errorf("%s got err in ProcessBlock: %w", c.Name(),
                        err)

        case <-time.After(timeout):
                return fmt.Errorf("consumer %s: %w", c.Name(),
                        ErrProcessBlockTimeout)
        }

        b.logger().Debugf("Consumer[%s] processed block in %v", c.Name(),
                time.Since(start))

        return nil
}

1	package chainio
2
3	import (
4	"errors"
5	"fmt"
6	"sync"
7	"sync/atomic"
8	"time"
9
10	"github.com/btcsuite/btclog/v2"
11	"github.com/lightningnetwork/lnd/chainntnfs"
12	"github.com/lightningnetwork/lnd/lnutils"
13	"golang.org/x/sync/errgroup"
14	)
15
16	// DefaultProcessBlockTimeout is the timeout value used when waiting for one
17	// consumer to finish processing the new block epoch.
18	var DefaultProcessBlockTimeout = 60 * time.Second
19
20	// ErrProcessBlockTimeout is the error returned when a consumer takes too long
21	// to process the block.
22	var ErrProcessBlockTimeout = errors.New("process block timeout")
23
24	// BlockbeatDispatcher is a service that handles dispatching new blocks to
25	// `lnd`'s subsystems. During startup, subsystems that are block-driven should
26	// implement the `Consumer` interface and register themselves via
27	// `RegisterQueue`. When two subsystems are independent of each other, they
28	// should be registered in different queues so blocks are notified concurrently.
29	// Otherwise, when living in the same queue, the subsystems are notified of the
30	// new blocks sequentially, which means it's critical to understand the
31	// relationship of these systems to properly handle the order.
32	type BlockbeatDispatcher struct {
33	wg sync.WaitGroup
34
35	// notifier is used to receive new block epochs.
36	notifier chainntnfs.ChainNotifier
37
38	// beat is the latest blockbeat received.
39	beat Blockbeat
40
41	// consumerQueues is a map of consumers that will receive blocks. Its
42	// key is a unique counter and its value is a queue of consumers. Each
43	// queue is notified concurrently, and consumers in the same queue is
44	// notified sequentially.
45	consumerQueues map[uint32][]Consumer
46
47	// counter is used to assign a unique id to each queue.
48	counter atomic.Uint32
49
50	// quit is used to signal the BlockbeatDispatcher to stop.
51	quit chan struct{}
52	}
53
54	// NewBlockbeatDispatcher returns a new blockbeat dispatcher instance.
55	func NewBlockbeatDispatcher(n chainntnfs.ChainNotifier) *BlockbeatDispatcher {
56	return &BlockbeatDispatcher{
57	notifier: n,
58	quit: make(chan struct{}),
59	consumerQueues: make(map[uint32][]Consumer),
60	}
61	}
62
63	// RegisterQueue takes a list of consumers and registers them in the same
64	// queue.
65	//
66	// NOTE: these consumers are notified sequentially.
67	func (b *BlockbeatDispatcher) RegisterQueue(consumers []Consumer) {
68	qid := b.counter.Add(1)	3✔
69		3✔
70	b.consumerQueues[qid] = append(b.consumerQueues[qid], consumers...)	3✔
71	clog.Infof("Registered queue=%d with %d blockbeat consumers", qid,	3✔
72	len(consumers))	3✔
73
74	for _, c := range consumers {
75	clog.Debugf("Consumer [%s] registered in queue %d", c.Name(),	3✔
76	qid)	3✔
77	}	3✔
78	}	3✔
79		3✔
80	// Start starts the blockbeat dispatcher - it registers a block notification	3✔
81	// and monitors and dispatches new blocks in a goroutine. It will refuse to	3✔
82	// start if there are no registered consumers.	3✔
83	func (b *BlockbeatDispatcher) Start() error {
84	// Make sure consumers are registered.
85	if len(b.consumerQueues) == 0 {
86	return fmt.Errorf("no consumers registered")
87	}
88		3✔
89	// Start listening to new block epochs. We should get a notification	3✔
90	// with the current best block immediately.	3✔
91	blockEpochs, err := b.notifier.RegisterBlockEpochNtfn(nil)	3✔
92	if err != nil {	3✔
93	return fmt.Errorf("register block epoch ntfn: %w", err)	3✔
94	}	3✔
95		6✔
96	clog.Infof("BlockbeatDispatcher is starting with %d consumer queues",	3✔
97	len(b.consumerQueues))	3✔
98	defer clog.Debug("BlockbeatDispatcher started")	3✔
99
100	b.wg.Add(1)
101	go b.dispatchBlocks(blockEpochs)
102
103	return nil
104	}	3✔
105		3✔
106	// Stop shuts down the blockbeat dispatcher.	3✔
UNCOV 107	func (b *BlockbeatDispatcher) Stop() {	×
UNCOV 108	clog.Info("BlockbeatDispatcher is stopping")	×
109	defer clog.Debug("BlockbeatDispatcher stopped")
110
111	// Signal the dispatchBlocks goroutine to stop.
112	close(b.quit)	3✔
113	b.wg.Wait()	3✔
UNCOV 114	}	×
UNCOV 115		×
116	func (b *BlockbeatDispatcher) log() btclog.Logger {
117	return b.beat.logger()	3✔
118	}	3✔
119		3✔
120	// dispatchBlocks listens to new block epoch and dispatches it to all the	3✔
121	// consumers. Each queue is notified concurrently, and the consumers in the	3✔
122	// same queue are notified sequentially.	3✔
123	//	3✔
124	// NOTE: Must be run as a goroutine.	3✔
125	func (b *BlockbeatDispatcher) dispatchBlocks(
126	blockEpochs *chainntnfs.BlockEpochEvent) {
127
128	defer b.wg.Done()	3✔
129	defer blockEpochs.Cancel()	3✔
130		3✔
131	for {	3✔
132	select {	3✔
133	case blockEpoch, ok := <-blockEpochs.Epochs:	3✔
134	if !ok {	3✔
135	clog.Debugf("Block epoch channel closed")	3✔
136
137	return	3✔
138	}	3✔
139		3✔
140	// Log a separator so it's easier to identify when a
141	// new block arrives for subsystems.
142	clog.Debugf("%v", lnutils.NewSeparatorClosure())
143
144	clog.Infof("Received new block %v at height %d, "+
145	"notifying consumers...", blockEpoch.Hash,
146	blockEpoch.Height)
147		3✔
148	// Record the time it takes the consumer to process	3✔
149	// this block.	3✔
150	start := time.Now()	3✔
151		3✔
152	// Update the current block epoch.	6✔
153	b.beat = NewBeat(*blockEpoch)	3✔
154		3✔
155	// Notify all consumers.	3✔
156	err := b.notifyQueues()	×
157	if err != nil {	×
158	b.log().Errorf("Notify block failed: %v", err)	×
159	}	×
160
161	b.log().Infof("Notified all consumers on new block "+
162	"in %v", time.Since(start))
163		3✔
164	case <-b.quit:	3✔
165	b.log().Debugf("BlockbeatDispatcher quit signal " +	3✔
166	"received")	3✔
167		3✔
168	return	3✔
169	}	3✔
170	}	3✔
171	}	3✔
172		3✔
173	// notifyQueues notifies each queue concurrently about the latest block epoch.	3✔
174	func (b *BlockbeatDispatcher) notifyQueues() error {	3✔
175	// errChans is a map of channels that will be used to receive errors	3✔
176	// returned from notifying the consumers.	3✔
177	errChans := make(map[uint32]chan error, len(b.consumerQueues))	3✔
178		3✔
179	// Notify each queue in goroutines.	×
180	for qid, consumers := range b.consumerQueues {	×
181	b.log().Debugf("Notifying queue=%d with %d consumers", qid,
182	len(consumers))	3✔
183		3✔
184	// Create a signal chan.
185	errChan := make(chan error, 1)
186	errChans[qid] = errChan
187		3✔
188	// Notify each queue concurrently.	3✔
189	go func(qid uint32, c []Consumer, beat Blockbeat) {	3✔
190	// Notify each consumer in this queue sequentially.	3✔
191	errChan <- DispatchSequential(beat, c)	6✔
192	}(qid, consumers, b.beat)	3✔
193	}	3✔
194
195	// Wait for all consumers in each queue to finish.	3✔
196	for qid, errChan := range errChans {
197	select {	3✔
198	case err := <-errChan:	3✔
199	if err != nil {	3✔
200	return fmt.Errorf("queue=%d got err: %w", qid,	3✔
201	err)	3✔
202	}
203
204	b.log().Debugf("Notified queue=%d", qid)
205
206	case <-b.quit:
207	b.log().Debugf("BlockbeatDispatcher quit signal " +
208	"received, exit notifyQueues")	3✔
209		3✔
210	return nil	3✔
211	}	3✔
212	}	3✔
213
UNCOV 214	return nil	×
UNCOV 215	}	×
UNCOV 216		×
217	// DispatchSequential takes a list of consumers and notify them about the new
218	// epoch sequentially. It requires the consumer to finish processing the block
219	// within the specified time, otherwise a timeout error is returned.	3✔
220	func DispatchSequential(b Blockbeat, consumers []Consumer) error {	3✔
221	for _, c := range consumers {	3✔
222	// Send the beat to the consumer.	3✔
223	err := notifyAndWait(b, c, DefaultProcessBlockTimeout)
224	if err != nil {	×
225	b.logger().Errorf("Failed to process block: %v", err)	×
226		×
227	return err
228	}
229	}
230
231	return nil	3✔
232	}	3✔
233		3✔
234	// DispatchConcurrent notifies each consumer concurrently about the blockbeat.	3✔
235	// It requires the consumer to finish processing the block within the specified	3✔
236	// time, otherwise a timeout error is returned.	3✔
237	func DispatchConcurrent(b Blockbeat, consumers []Consumer) error {	6✔
238	eg := &errgroup.Group{}	3✔
239		3✔
240	// Notify each queue in goroutines.	3✔
241	for _, c := range consumers {	3✔
242	// Notify each consumer concurrently.	3✔
243	eg.Go(func() error {	3✔
244	// Send the beat to the consumer.	3✔
245	err := notifyAndWait(b, c, DefaultProcessBlockTimeout)	3✔
246		6✔
247	// Exit early if there's no error.	3✔
248	if err == nil {	3✔
249	return nil	3✔
250	}
251
252	b.logger().Errorf("Consumer=%v failed to process "+
253	"block: %v", c.Name(), err)	6✔
254		3✔
255	return err	3✔
256	})	3✔
UNCOV 257	}	×
UNCOV 258		×
UNCOV 259	// Wait for all consumers in each queue to finish.	×
260	if err := eg.Wait(); err != nil {
261	return err	3✔
262	}
263		×
264	return nil	×
265	}	×
266		×
267	// notifyAndWait sends the blockbeat to the specified consumer. It requires the	×
268	// consumer to finish processing the block within the specified time, otherwise
269	// a timeout error is returned.
270	func notifyAndWait(b Blockbeat, c Consumer, timeout time.Duration) error {
271	b.logger().Debugf("Waiting for consumer[%s] to process it", c.Name())	3✔
272
273	// Record the time it takes the consumer to process this block.
274	start := time.Now()
275
276	errChan := make(chan error, 1)
277	go func() {	3✔
278	errChan <- c.ProcessBlock(b)	6✔
279	}()	3✔
280		3✔
281	// We expect the consumer to finish processing this block under 30s,	3✔
UNCOV 282	// otherwise a timeout error is returned.	×
UNCOV 283	select {	×
UNCOV 284	case err := <-errChan:	×
UNCOV 285	if err == nil {	×
286	break
287	}
288		3✔
289	return fmt.Errorf("%s got err in ProcessBlock: %w", c.Name(),
290	err)
291
292	case <-time.After(timeout):
293	return fmt.Errorf("consumer %s: %w", c.Name(),
294	ErrProcessBlockTimeout)	3✔
295	}	3✔
296		3✔
297	b.logger().Debugf("Consumer[%s] processed block in %v", c.Name(),	3✔
298	time.Since(start))	6✔
299		3✔
300	return nil	6✔
301	}	3✔

lightningnetwork / lnd / 13423774416

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