12313073348

Committed 13 Dec 2024 09:30AM UTC coverage: 58.666% (+1.2%) from 57.486%

Build # 12313073348

Build Type

Pull #9344

github

Committed by

ellemouton

Commit Message

htlcswitch+go.mod: use updated fn.ContextGuard

This commit updates the fn dep to the version containing the updates to
the ContextGuard implementation. Only the htlcswitch/link uses the guard
at the moment so this is updated to make use of the new implementation.

Pull Request Pull Request #9344: htlcswitch+go.mod: use updated fn.ContextGuard

Run Details

101 of 117 new or added lines in 4 files covered. (86.32%)

29 existing lines in 9 files now uncovered.

134589 of 229415 relevant lines covered (58.67%)

19278.57 hits per line

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

67.71

/lnwallet/sigpool.go

package lnwallet

import (
        "fmt"
        "sync"

        "github.com/btcsuite/btcd/btcec/v2"
        "github.com/btcsuite/btcd/wire"
        "github.com/lightningnetwork/lnd/input"
        "github.com/lightningnetwork/lnd/lnwire"
)

const (
        // jobBuffer is a constant the represents the buffer of jobs in the two
        // main queues. This allows clients avoid necessarily blocking when
        // submitting jobs into the queue.
        jobBuffer = 100

        // TODO(roasbeef): job buffer pool?
)

// VerifyJob is a job sent to the sigPool sig pool to verify a signature
// on a transaction. The items contained in the struct are necessary and
// sufficient to verify the full signature. The passed sigHash closure function
// should be set to a function that generates the relevant sighash.
//
// TODO(roasbeef): when we move to ecschnorr, make into batch signature
// verification using bos-coster (or pip?).
type VerifyJob struct {
        // PubKey is the public key that was used to generate the purported
        // valid signature. Note that with the current channel construction,
        // this public key will likely have been tweaked using the current per
        // commitment point for a particular commitment transactions.
        PubKey *btcec.PublicKey

        // Sig is the raw signature generated using the above public key.  This
        // is the signature to be verified.
        Sig input.Signature

        // SigHash is a function closure generates the sighashes that the
        // passed signature is known to have signed.
        SigHash func() ([]byte, error)

        // HtlcIndex is the index of the HTLC from the PoV of the remote
        // party's update log.
        HtlcIndex uint64

        // Cancel is a channel that is closed by the caller if they wish to
        // cancel all pending verification jobs part of a single batch. This
        // channel is closed in the case that a single signature in a batch has
        // been returned as invalid, as there is no need to verify the remainder
        // of the signatures.
        Cancel <-chan struct{}

        // ErrResp is the channel that the result of the signature verification
        // is to be sent over. In the see that the signature is valid, a nil
        // error will be passed. Otherwise, a concrete error detailing the
        // issue will be passed. This channel MUST be buffered.
        ErrResp chan *HtlcIndexErr
}

// HtlcIndexErr is a special type of error that also includes a pointer to the
// original validation job. This error message allows us to craft more detailed
// errors at upper layers.
type HtlcIndexErr struct {
        error

        *VerifyJob
}

// SignJob is a job sent to the sigPool sig pool to generate a valid
// signature according to the passed SignDescriptor for the passed transaction.
// Jobs are intended to be sent in batches in order to parallelize the job of
// generating signatures for a new commitment transaction.
type SignJob struct {
        // SignDesc is intended to be a full populated SignDescriptor which
        // encodes the necessary material (keys, witness script, etc) required
        // to generate a valid signature for the specified input.
        SignDesc input.SignDescriptor

        // Tx is the transaction to be signed. This is required to generate the
        // proper sighash for the input to be signed.
        Tx *wire.MsgTx

        // OutputIndex is the output index of the HTLC on the commitment
        // transaction being signed.
        OutputIndex int32

        // Cancel is a channel that is closed by the caller if they wish to
        // abandon all pending sign jobs part of a single batch. This should
        // never be closed by the validator.
        Cancel <-chan struct{}

        // Resp is the channel that the response to this particular SignJob
        // will be sent over. This channel MUST be buffered.
        //
        // TODO(roasbeef): actually need to allow caller to set, need to retain
        // order mark commit sig as special
        Resp chan SignJobResp
}

// SignJobResp is the response to a sign job. Both channels are to be read in
// order to ensure no unnecessary goroutine blocking occurs. Additionally, both
// channels should be buffered.
type SignJobResp struct {
        // Sig is the generated signature for a particular SignJob In the case
        // of an error during signature generation, then this value sent will
        // be nil.
        Sig lnwire.Sig

        // Err is the error that occurred when executing the specified
        // signature job. In the case that no error occurred, this value will
        // be nil.
        Err error
}

// SigPool is a struct that is meant to allow the current channel state
// machine to parallelize all signature generation and verification. This
// struct is needed as _each_ HTLC when creating a commitment transaction
// requires a signature, and similarly a receiver of a new commitment must
// verify all the HTLC signatures included within the CommitSig message. A pool
// of workers will be maintained by the sigPool. Batches of jobs (either
// to sign or verify) can be sent to the pool of workers which will
// asynchronously perform the specified job.
type SigPool struct {
        started sync.Once
        stopped sync.Once

        signer input.Signer

        verifyJobs chan VerifyJob
        signJobs   chan SignJob

        wg   sync.WaitGroup
        quit chan struct{}

        numWorkers int
}

// NewSigPool creates a new signature pool with the specified number of
// workers. The recommended parameter for the number of works is the number of
// physical CPU cores available on the target machine.
func NewSigPool(numWorkers int, signer input.Signer) *SigPool {
        return &SigPool{
                signer:     signer,
                numWorkers: numWorkers,
                verifyJobs: make(chan VerifyJob, jobBuffer),
                signJobs:   make(chan SignJob, jobBuffer),
                quit:       make(chan struct{}),
        }
}

// Start starts of all goroutines that the sigPool sig pool needs to
// carry out its duties.
func (s *SigPool) Start() error {
        s.started.Do(func() {
                walletLog.Info("SigPool starting")
                for i := 0; i < s.numWorkers; i++ {
                        s.wg.Add(1)
                        go s.poolWorker()
                }
        })
        return nil
}

// Stop signals any active workers carrying out jobs to exit so the sigPool can
// gracefully shutdown.
func (s *SigPool) Stop() error {
        s.stopped.Do(func() {
                close(s.quit)
                s.wg.Wait()
        })
        return nil
}

// poolWorker is the main worker goroutine within the sigPool sig pool.
// Individual batches are distributed amongst each of the active workers. The
// workers then execute the task based on the type of job, and return the
// result back to caller.
func (s *SigPool) poolWorker() {
        defer s.wg.Done()

        for {
                select {

                // We've just received a new signature job. Given the items
                // contained within the message, we'll craft a signature and
                // send the result along with a possible error back to the
                // caller.
                case sigMsg := <-s.signJobs:
                        rawSig, err := s.signer.SignOutputRaw(
                                sigMsg.Tx, &sigMsg.SignDesc,
                        )
                        if err != nil {
                                select {
                                case sigMsg.Resp <- SignJobResp{
                                        Sig: lnwire.Sig{},
                                        Err: err,
                                }:
                                        continue
                                case <-sigMsg.Cancel:
                                        continue
                                case <-s.quit:
                                        return
                                }
                        }

                        // Use the sig mapper to go from the input.Signature
                        // into the serialized lnwire.Sig that we'll send
                        // across the wire.
                        sig, err := lnwire.NewSigFromSignature(rawSig)

                        select {
                        case sigMsg.Resp <- SignJobResp{
                                Sig: sig,
                                Err: err,
                        }:
                        case <-sigMsg.Cancel:
                                continue
                        case <-s.quit:
                                return
                        }

                // We've just received a new verification job from the outside
                // world. We'll attempt to construct the sighash, parse the
                // signature, and finally verify the signature.
                case verifyMsg := <-s.verifyJobs:
                        sigHash, err := verifyMsg.SigHash()
                        if err != nil {
                                select {
                                case verifyMsg.ErrResp <- &HtlcIndexErr{
                                        error:     err,
                                        VerifyJob: &verifyMsg,
                                }:
                                        continue
                                case <-verifyMsg.Cancel:
                                        continue
                                }
                        }

                        rawSig := verifyMsg.Sig

                        if !rawSig.Verify(sigHash, verifyMsg.PubKey) {
                                err := fmt.Errorf("invalid signature "+
                                        "sighash: %x, sig: %x", sigHash,
                                        rawSig.Serialize())

                                select {
                                case verifyMsg.ErrResp <- &HtlcIndexErr{
                                        error:     err,
                                        VerifyJob: &verifyMsg,
                                }:
                                case <-verifyMsg.Cancel:
                                case <-s.quit:
                                        return
                                }
                        } else {
                                select {
                                case verifyMsg.ErrResp <- nil:
                                case <-verifyMsg.Cancel:
                                case <-s.quit:
                                        return
                                }
                        }

                // The sigPool sig pool is exiting, so we will as well.
                case <-s.quit:
                        return
                }
        }
}

// SubmitSignBatch submits a batch of signature jobs to the sigPool.  The
// response and cancel channels for each of the SignJob's are expected to be
// fully populated, as the response for each job will be sent over the
// response channel within the job itself.
func (s *SigPool) SubmitSignBatch(signJobs []SignJob) {
        for _, job := range signJobs {
                select {
                case s.signJobs <- job:
                case <-job.Cancel:
                        // TODO(roasbeef): return error?
                case <-s.quit:
                        return
                }
        }
}

// SubmitVerifyBatch submits a batch of verification jobs to the sigPool. For
// each job submitted, an error will be passed into the returned channel
// denoting if signature verification was valid or not. The passed cancelChan
// allows the caller to cancel all pending jobs in the case that they wish to
// bail early.
func (s *SigPool) SubmitVerifyBatch(verifyJobs []VerifyJob,
        cancelChan chan struct{}) <-chan *HtlcIndexErr {

        errChan := make(chan *HtlcIndexErr, len(verifyJobs))

        for _, job := range verifyJobs {
                job.Cancel = cancelChan
                job.ErrResp = errChan

                select {
                case s.verifyJobs <- job:
                case <-job.Cancel:
                        return errChan
                }
        }

        return errChan
}

1	package lnwallet
2
3	import (
4	"fmt"
5	"sync"
6
7	"github.com/btcsuite/btcd/btcec/v2"
8	"github.com/btcsuite/btcd/wire"
9	"github.com/lightningnetwork/lnd/input"
10	"github.com/lightningnetwork/lnd/lnwire"
11	)
12
13	const (
14	// jobBuffer is a constant the represents the buffer of jobs in the two
15	// main queues. This allows clients avoid necessarily blocking when
16	// submitting jobs into the queue.
17	jobBuffer = 100
18
19	// TODO(roasbeef): job buffer pool?
20	)
21
22	// VerifyJob is a job sent to the sigPool sig pool to verify a signature
23	// on a transaction. The items contained in the struct are necessary and
24	// sufficient to verify the full signature. The passed sigHash closure function
25	// should be set to a function that generates the relevant sighash.
26	//
27	// TODO(roasbeef): when we move to ecschnorr, make into batch signature
28	// verification using bos-coster (or pip?).
29	type VerifyJob struct {
30	// PubKey is the public key that was used to generate the purported
31	// valid signature. Note that with the current channel construction,
32	// this public key will likely have been tweaked using the current per
33	// commitment point for a particular commitment transactions.
34	PubKey *btcec.PublicKey
35
36	// Sig is the raw signature generated using the above public key. This
37	// is the signature to be verified.
38	Sig input.Signature
39
40	// SigHash is a function closure generates the sighashes that the
41	// passed signature is known to have signed.
42	SigHash func() ([]byte, error)
43
44	// HtlcIndex is the index of the HTLC from the PoV of the remote
45	// party's update log.
46	HtlcIndex uint64
47
48	// Cancel is a channel that is closed by the caller if they wish to
49	// cancel all pending verification jobs part of a single batch. This
50	// channel is closed in the case that a single signature in a batch has
51	// been returned as invalid, as there is no need to verify the remainder
52	// of the signatures.
53	Cancel <-chan struct{}
54
55	// ErrResp is the channel that the result of the signature verification
56	// is to be sent over. In the see that the signature is valid, a nil
57	// error will be passed. Otherwise, a concrete error detailing the
58	// issue will be passed. This channel MUST be buffered.
59	ErrResp chan *HtlcIndexErr
60	}
61
62	// HtlcIndexErr is a special type of error that also includes a pointer to the
63	// original validation job. This error message allows us to craft more detailed
64	// errors at upper layers.
65	type HtlcIndexErr struct {
66	error
67
68	*VerifyJob
69	}
70
71	// SignJob is a job sent to the sigPool sig pool to generate a valid
72	// signature according to the passed SignDescriptor for the passed transaction.
73	// Jobs are intended to be sent in batches in order to parallelize the job of
74	// generating signatures for a new commitment transaction.
75	type SignJob struct {
76	// SignDesc is intended to be a full populated SignDescriptor which
77	// encodes the necessary material (keys, witness script, etc) required
78	// to generate a valid signature for the specified input.
79	SignDesc input.SignDescriptor
80
81	// Tx is the transaction to be signed. This is required to generate the
82	// proper sighash for the input to be signed.
83	Tx *wire.MsgTx
84
85	// OutputIndex is the output index of the HTLC on the commitment
86	// transaction being signed.
87	OutputIndex int32
88
89	// Cancel is a channel that is closed by the caller if they wish to
90	// abandon all pending sign jobs part of a single batch. This should
91	// never be closed by the validator.
92	Cancel <-chan struct{}
93
94	// Resp is the channel that the response to this particular SignJob
95	// will be sent over. This channel MUST be buffered.
96	//
97	// TODO(roasbeef): actually need to allow caller to set, need to retain
98	// order mark commit sig as special
99	Resp chan SignJobResp
100	}
101
102	// SignJobResp is the response to a sign job. Both channels are to be read in
103	// order to ensure no unnecessary goroutine blocking occurs. Additionally, both
104	// channels should be buffered.
105	type SignJobResp struct {
106	// Sig is the generated signature for a particular SignJob In the case
107	// of an error during signature generation, then this value sent will
108	// be nil.
109	Sig lnwire.Sig
110
111	// Err is the error that occurred when executing the specified
112	// signature job. In the case that no error occurred, this value will
113	// be nil.
114	Err error
115	}
116
117	// SigPool is a struct that is meant to allow the current channel state
118	// machine to parallelize all signature generation and verification. This
119	// struct is needed as _each_ HTLC when creating a commitment transaction
120	// requires a signature, and similarly a receiver of a new commitment must
121	// verify all the HTLC signatures included within the CommitSig message. A pool
122	// of workers will be maintained by the sigPool. Batches of jobs (either
123	// to sign or verify) can be sent to the pool of workers which will
124	// asynchronously perform the specified job.
125	type SigPool struct {
126	started sync.Once
127	stopped sync.Once
128
129	signer input.Signer
130
131	verifyJobs chan VerifyJob
132	signJobs chan SignJob
133
134	wg sync.WaitGroup
135	quit chan struct{}
136
137	numWorkers int
138	}
139
140	// NewSigPool creates a new signature pool with the specified number of
141	// workers. The recommended parameter for the number of works is the number of
142	// physical CPU cores available on the target machine.
143	func NewSigPool(numWorkers int, signer input.Signer) *SigPool {	632✔
144	return &SigPool{	632✔
145	signer: signer,	632✔
146	numWorkers: numWorkers,	632✔
147	verifyJobs: make(chan VerifyJob, jobBuffer),	632✔
148	signJobs: make(chan SignJob, jobBuffer),	632✔
149	quit: make(chan struct{}),	632✔
150	}	632✔
151	}	632✔
152
153	// Start starts of all goroutines that the sigPool sig pool needs to
154	// carry out its duties.
155	func (s *SigPool) Start() error {	632✔
156	s.started.Do(func() {	1,264✔
157	walletLog.Info("SigPool starting")	632✔
158	for i := 0; i < s.numWorkers; i++ {	1,942✔
159	s.wg.Add(1)	1,310✔
160	go s.poolWorker()	1,310✔
161	}	1,310✔
162	})
163	return nil	632✔
164	}
165
166	// Stop signals any active workers carrying out jobs to exit so the sigPool can
167	// gracefully shutdown.
168	func (s *SigPool) Stop() error {	406✔
169	s.stopped.Do(func() {	812✔
170	close(s.quit)	406✔
171	s.wg.Wait()	406✔
172	})	406✔
173	return nil	406✔
174	}
175
176	// poolWorker is the main worker goroutine within the sigPool sig pool.
177	// Individual batches are distributed amongst each of the active workers. The
178	// workers then execute the task based on the type of job, and return the
179	// result back to caller.
180	func (s *SigPool) poolWorker() {	1,310✔
181	defer s.wg.Done()	1,310✔
182		1,310✔
183	for {	10,490✔
184	select {	9,180✔
185
186	// We've just received a new signature job. Given the items
187	// contained within the message, we'll craft a signature and
188	// send the result along with a possible error back to the
189	// caller.
190	case sigMsg := <-s.signJobs:	3,966✔
191	rawSig, err := s.signer.SignOutputRaw(	3,966✔
192	sigMsg.Tx, &sigMsg.SignDesc,	3,966✔
193	)	3,966✔
194	if err != nil {	3,966✔
195	select {	×
196	case sigMsg.Resp <- SignJobResp{
197	Sig: lnwire.Sig{},
198	Err: err,
199	}:	×
200	continue	×
201	case <-sigMsg.Cancel:	×
202	continue	×
203	case <-s.quit:	×
204	return	×
205	}
206	}
207
208	// Use the sig mapper to go from the input.Signature
209	// into the serialized lnwire.Sig that we'll send
210	// across the wire.
211	sig, err := lnwire.NewSigFromSignature(rawSig)	3,966✔
212		3,966✔
213	select {	3,966✔
214	case sigMsg.Resp <- SignJobResp{
215	Sig: sig,
216	Err: err,
217	}:	3,965✔
UNCOV 218	case <-sigMsg.Cancel:	×
UNCOV 219	continue	×
220	case <-s.quit:	1✔
221	return	1✔
222	}
223
224	// We've just received a new verification job from the outside
225	// world. We'll attempt to construct the sighash, parse the
226	// signature, and finally verify the signature.
227	case verifyMsg := <-s.verifyJobs:	3,913✔
228	sigHash, err := verifyMsg.SigHash()	3,913✔
229	if err != nil {	3,913✔
230	select {	×
231	case verifyMsg.ErrResp <- &HtlcIndexErr{
232	error: err,
233	VerifyJob: &verifyMsg,
234	}:	×
235	continue	×
236	case <-verifyMsg.Cancel:	×
237	continue	×
238	}
239	}
240
241	rawSig := verifyMsg.Sig	3,913✔
242		3,913✔
243	if !rawSig.Verify(sigHash, verifyMsg.PubKey) {	3,913✔
244	err := fmt.Errorf("invalid signature "+	×
245	"sighash: %x, sig: %x", sigHash,	×
246	rawSig.Serialize())	×
247		×
248	select {	×
249	case verifyMsg.ErrResp <- &HtlcIndexErr{
250	error: err,
251	VerifyJob: &verifyMsg,
252	}:	×
253	case <-verifyMsg.Cancel:	×
254	case <-s.quit:	×
255	return	×
256	}
257	} else {	3,913✔
258	select {	3,913✔
259	case verifyMsg.ErrResp <- nil:	3,913✔
260	case <-verifyMsg.Cancel:	×
261	case <-s.quit:	×
262	return	×
263	}
264	}
265
266	// The sigPool sig pool is exiting, so we will as well.
267	case <-s.quit:	405✔
268	return	405✔
269	}
270	}
271	}
272
273	// SubmitSignBatch submits a batch of signature jobs to the sigPool. The
274	// response and cancel channels for each of the SignJob's are expected to be
275	// fully populated, as the response for each job will be sent over the
276	// response channel within the job itself.
277	func (s *SigPool) SubmitSignBatch(signJobs []SignJob) {	2,068✔
278	for _, job := range signJobs {	6,044✔
279	select {	3,976✔
280	case s.signJobs <- job:	3,976✔
281	case <-job.Cancel:	×
282	// TODO(roasbeef): return error?
283	case <-s.quit:	×
284	return	×
285	}
286	}
287	}
288
289	// SubmitVerifyBatch submits a batch of verification jobs to the sigPool. For
290	// each job submitted, an error will be passed into the returned channel
291	// denoting if signature verification was valid or not. The passed cancelChan
292	// allows the caller to cancel all pending jobs in the case that they wish to
293	// bail early.
294	func (s *SigPool) SubmitVerifyBatch(verifyJobs []VerifyJob,
295	cancelChan chan struct{}) <-chan *HtlcIndexErr {	2,040✔
296		2,040✔
297	errChan := make(chan *HtlcIndexErr, len(verifyJobs))	2,040✔
298		2,040✔
299	for _, job := range verifyJobs {	5,953✔
300	job.Cancel = cancelChan	3,913✔
301	job.ErrResp = errChan	3,913✔
302		3,913✔
303	select {	3,913✔
304	case s.verifyJobs <- job:	3,913✔
305	case <-job.Cancel:	×
306	return errChan	×
307	}
308	}
309
310	return errChan	2,040✔
311	}

lightningnetwork / lnd / 12313073348

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