forked from lightninglabs/neutrino
-
Notifications
You must be signed in to change notification settings - Fork 0
/
query.go
1501 lines (1291 loc) · 44.2 KB
/
query.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
// NOTE: THIS API IS UNSTABLE RIGHT NOW.
package neutrino
import (
"fmt"
"sync"
"sync/atomic"
"time"
"github.com/btcsuite/btcd/blockchain"
"github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/btcsuite/btcd/wire"
"github.com/btcsuite/btcutil"
"github.com/btcsuite/btcutil/gcs"
"github.com/btcsuite/btcutil/gcs/builder"
"github.com/davecgh/go-spew/spew"
"github.com/lightninglabs/neutrino/cache"
"github.com/lightninglabs/neutrino/filterdb"
"github.com/lightninglabs/neutrino/pushtx"
)
var (
// QueryTimeout specifies how long to wait for a peer to answer a
// query.
QueryTimeout = time.Second * 10
// QueryBatchTimout is the total time we'll wait for a batch fetch
// query to complete.
// TODO(halseth): instead use timeout since last received response?
QueryBatchTimeout = time.Second * 30
// QueryPeerCooldown is the time we'll wait before re-assigning a query
// to a peer that previously failed because of a timeout.
QueryPeerCooldown = time.Second * 5
// QueryNumRetries specifies how many times to retry sending a query to
// each peer before we've concluded we aren't going to get a valid
// response. This allows to make up for missed messages in some
// instances.
QueryNumRetries = 2
// QueryPeerConnectTimeout specifies how long to wait for the
// underlying chain service to connect to a peer before giving up
// on a query in case we don't have any peers.
QueryPeerConnectTimeout = time.Second * 30
// QueryEncoding specifies the default encoding (witness or not) for
// `getdata` and other similar messages.
QueryEncoding = wire.WitnessEncoding
)
// QueryAccess is an interface that gives acces to query a set of peers in
// different ways.
type QueryAccess interface {
queryAllPeers(
queryMsg wire.Message,
checkResponse func(sp *ServerPeer, resp wire.Message,
quit chan<- struct{}, peerQuit chan<- struct{}),
options ...QueryOption)
}
// A compile-time check to ensure that ChainService implements the
// QueryAccess interface.
var _ QueryAccess = (*ChainService)(nil)
// queries are a set of options that can be modified per-query, unlike global
// options.
//
// TODO: Make more query options that override global options.
type queryOptions struct {
// timeout lets the query know how long to wait for a peer to answer
// the query before moving onto the next peer.
timeout time.Duration
// numRetries tells the query how many times to retry asking each peer
// the query.
numRetries uint8
// peerConnectTimeout lets the query know how long to wait for the
// underlying chain service to connect to a peer before giving up
// on a query in case we don't have any peers.
peerConnectTimeout time.Duration
// encoding lets the query know which encoding to use when queueing
// messages to a peer.
encoding wire.MessageEncoding
// doneChan lets the query signal the caller when it's done, in case
// it's run in a goroutine.
doneChan chan<- struct{}
// persistToDisk indicates whether the filter should also be written
// to disk in addition to the memory cache. For "normal" wallets, they'll
// almost never need to re-match a filter once it's been fetched unless
// they're doing something like a key import.
persistToDisk bool
// optimisticBatch indicates whether we expect more calls to follow,
// and that we should attempt to batch more items with the query such
// that they can be cached, avoiding the extra round trip.
optimisticBatch optimisticBatchType
}
// optimisticBatchType is a type indicating the kind of batching we want to
// execute with a query.
type optimisticBatchType uint8
const (
// noBatch indicates no other than the specified item should be
// queried.
noBatch optimisticBatchType = iota
// forwardBatch is used to indicate we should also query for items
// following, as they most likely will be fetched next.
forwardBatch
// reverseBatch is used to indicate we should also query for items
// preceding, as they most likely will be fetched next.
reverseBatch
)
// QueryOption is a functional option argument to any of the network query
// methods, such as GetBlock and GetCFilter (when that resorts to a network
// query). These are always processed in order, with later options overriding
// earlier ones.
type QueryOption func(*queryOptions)
// defaultQueryOptions returns a queryOptions set to package-level defaults.
func defaultQueryOptions() *queryOptions {
return &queryOptions{
timeout: QueryTimeout,
numRetries: uint8(QueryNumRetries),
peerConnectTimeout: QueryPeerConnectTimeout,
encoding: QueryEncoding,
optimisticBatch: noBatch,
}
}
// applyQueryOptions updates a queryOptions set with functional options.
func (qo *queryOptions) applyQueryOptions(options ...QueryOption) {
for _, option := range options {
option(qo)
}
}
// Timeout is a query option that lets the query know how long to wait for each
// peer we ask the query to answer it before moving on.
func Timeout(timeout time.Duration) QueryOption {
return func(qo *queryOptions) {
qo.timeout = timeout
}
}
// NumRetries is a query option that lets the query know the maximum number of
// times each peer should be queried. The default is one.
func NumRetries(numRetries uint8) QueryOption {
return func(qo *queryOptions) {
qo.numRetries = numRetries
}
}
// PeerConnectTimeout is a query option that lets the query know how long to
// wait for the underlying chain service to connect to a peer before giving up
// on a query in case we don't have any peers.
func PeerConnectTimeout(timeout time.Duration) QueryOption {
return func(qo *queryOptions) {
qo.peerConnectTimeout = timeout
}
}
// Encoding is a query option that allows the caller to set a message encoding
// for the query messages.
func Encoding(encoding wire.MessageEncoding) QueryOption {
return func(qo *queryOptions) {
qo.encoding = encoding
}
}
// DoneChan allows the caller to pass a channel that will get closed when the
// query is finished.
func DoneChan(doneChan chan<- struct{}) QueryOption {
return func(qo *queryOptions) {
qo.doneChan = doneChan
}
}
// PersistToDisk allows the caller to tell that the filter should be kept
// on disk once it's found.
func PersistToDisk() QueryOption {
return func(qo *queryOptions) {
qo.persistToDisk = true
}
}
// OptimisticBatch allows the caller to tell that items following the requested
// one should be included in the query.
func OptimisticBatch() QueryOption {
return func(qo *queryOptions) {
qo.optimisticBatch = forwardBatch
}
}
// OptimisticReverseBatch allows the caller to tell that items preceding the
// requested one should be included in the query.
func OptimisticReverseBatch() QueryOption {
return func(qo *queryOptions) {
qo.optimisticBatch = reverseBatch
}
}
// queryState is an atomically updated per-query state for each query in a
// batch.
//
// State transitions are:
//
// * queryWaitSubmit->queryWaitResponse - send query to peer
// * queryWaitResponse->queryWaitSubmit - query timeout with no acceptable
// response
// * queryWaitResponse->queryAnswered - acceptable response to query received
type queryState uint32
const (
// Waiting to be submitted to a peer.
queryWaitSubmit queryState = iota
// Submitted to a peer, waiting for reply.
queryWaitResponse
// Valid reply received.
queryAnswered
)
// We provide 3 kinds of queries:
//
// * queryAllPeers allows a single query to be broadcast to all peers, and
// then waits for as many peers as possible to answer that query within
// a timeout. This allows for doing things like checking cfilter checkpoints.
//
// * queryPeers allows a single query to be passed to one peer at a time until
// the query is deemed answered. This is good for getting a single piece of
// data, such as a filter or a block.
//
// * queryBatch allows a batch of queries to be distributed among all peers,
// recirculating upon timeout.
//
// TODO(aakselrod): maybe abstract the query scheduler into a functional option
// and provide some presets (including the ones below) prior to factoring out
// the query API into its own package?
// queryChainServiceBatch is a helper function that sends a batch of queries to
// the entire pool of peers of the given ChainService, attempting to get them
// all answered unless the quit channel is closed. It continues to update its
// view of the connected peers in case peers connect or disconnect during the
// query. The package-level QueryTimeout parameter, overridable by the Timeout
// option, determines how long a peer waits for a query before moving onto the
// next one. The NumRetries option and the QueryNumRetries package-level
// variable are ignored; the query continues until it either completes or the
// passed quit channel is closed. For memory efficiency, we attempt to get
// responses as close to ordered as we can, so that the caller can cache as few
// responses as possible before committing to storage.
//
// TODO(aakselrod): support for more than one in-flight query per peer to
// reduce effects of latency.
func queryChainServiceBatch(
// s is the ChainService to use.
s *ChainService,
// queryMsgs is a slice of queries for which the caller wants responses.
queryMsgs []wire.Message,
// checkResponse is called for every received message to see if it
// answers the query message. It should return true if so.
checkResponse func(sp *ServerPeer, query wire.Message,
resp wire.Message) bool,
// queryQuit forces the query to end before it's complete.
queryQuit <-chan struct{},
// options takes functional options for executing the query.
options ...QueryOption) {
// Starting with the set of default options, we'll apply any specified
// functional options to the query.
qo := defaultQueryOptions()
qo.applyQueryOptions(options...)
// Shared state between this goroutine and the per-peer goroutines.
queryStates := make([]uint32, len(queryMsgs))
// subscription allows us to subscribe to notifications from peers.
msgChan := make(chan spMsg, len(queryMsgs))
subQuit := make(chan struct{})
subscription := spMsgSubscription{
msgChan: msgChan,
quitChan: subQuit,
}
defer close(subQuit)
// peerState holds a query message and an answer channel that get a
// notice when the query got a match. If it's the peer's match, the
// peer can mark the query a success and move on to the next query
// ahead of timeout.
type peerState struct {
msg wire.Message
matchSignal chan struct{}
}
// peerStates and its companion mutex allow the peer goroutines to
// tell the main goroutine what query they're currently working on, and
// for the main goroutine to signal the peer when a match has been
// received.
peerStates := make(map[string]peerState)
var mtxPeerStates sync.RWMutex
// allDone is a helper closure we'll use to determine whether we can
// exit due to all of our queries being answered.
allDone := func() bool {
for i := 0; i < len(queryStates); i++ {
if atomic.LoadUint32(&queryStates[i]) !=
uint32(queryAnswered) {
return false
}
}
return true
}
// allDoneSignal is a channel we'll close to signal to the main loop
// that all of the queries have been answered.
var allDoneOnce sync.Once
allDoneSignal := make(chan struct{})
peerGoroutine := func(sp *ServerPeer, quit <-chan struct{},
allDoneSignal chan struct{}) {
// Subscribe to messages from the peer.
sp.subscribeRecvMsg(subscription)
defer sp.unsubscribeRecvMsgs(subscription)
defer func() {
mtxPeerStates.Lock()
delete(peerStates, sp.Addr())
mtxPeerStates.Unlock()
}()
// Track the last query our peer failed to answer and skip over
// it for the next attempt. This helps prevent most instances
// of the same peer being asked for the same query every time.
firstUnfinished, handleQuery := 0, -1
for firstUnfinished < len(queryMsgs) {
select {
case <-queryQuit:
return
case <-s.quit:
return
case <-quit:
return
default:
}
handleQuery = -1
for i := firstUnfinished; i < len(queryMsgs); i++ {
// If this query is finished and we're at
// firstUnfinished, update firstUnfinished.
if i == firstUnfinished &&
atomic.LoadUint32(&queryStates[i]) ==
uint32(queryAnswered) {
firstUnfinished++
log.Tracef("Query #%v already answered, "+
"skipping", i)
continue
}
// We check to see if the query is waiting to
// be handled. If so, we mark it as being
// handled. If not, we move to the next one.
if !atomic.CompareAndSwapUint32(
&queryStates[i],
uint32(queryWaitSubmit),
uint32(queryWaitResponse),
) {
log.Tracef("Query #%v already being "+
"queried for, skipping", i)
continue
}
// The query is now marked as in-process. We
// begin to process it.
handleQuery = i
sp.QueueMessageWithEncoding(queryMsgs[i],
nil, qo.encoding)
break
}
// Regardless of whether we have a query or not, we
// need a timeout.
timeout := time.After(qo.timeout)
if handleQuery == -1 {
if firstUnfinished == len(queryMsgs) {
// We've now answered all the queries.
return
}
// We have nothing to work on but not all
// queries are answered yet. Wait for a query
// timeout, or a quit signal, then see if
// anything needs our help.
select {
case <-queryQuit:
return
case <-s.quit:
return
case <-quit:
return
case <-timeout:
if sp.Connected() {
continue
} else {
return
}
}
}
// We have a query we're working on.
matchSignal := make(chan struct{}, 1)
mtxPeerStates.Lock()
peerStates[sp.Addr()] = peerState{
msg: queryMsgs[handleQuery],
matchSignal: matchSignal,
}
mtxPeerStates.Unlock()
exiting := false
select {
case <-queryQuit:
exiting = true
case <-s.quit:
exiting = true
case <-quit:
exiting = true
case <-timeout:
// We failed, so set the query state back to
// zero and update our lastFailed state.
atomic.StoreUint32(
&queryStates[handleQuery], uint32(queryWaitSubmit),
)
// Delete the query from our peer states, to
// indicate we are no longer expecting a
// response.
mtxPeerStates.Lock()
delete(peerStates, sp.Addr())
mtxPeerStates.Unlock()
log.Tracef("Query for #%v failed, moving "+
"on: %v", handleQuery,
newLogClosure(func() string {
return spew.Sdump(queryMsgs[handleQuery])
}))
// To allow other peers to pick up this query,
// let the peer that just timed out wait a
// cooldown period before handing it the next
// query.
select {
case <-time.After(QueryPeerCooldown):
case <-queryQuit:
return
case <-s.quit:
return
case <-quit:
return
}
case _, ok := <-matchSignal:
if !ok {
exiting = true
break
}
log.Tracef("Query #%v answered, updating state",
handleQuery)
// We got a match signal so we can mark this
// query a success.
atomic.StoreUint32(&queryStates[handleQuery],
uint32(queryAnswered))
// If we're done answering all of our queries,
// we can exit now.
if allDone() {
allDoneOnce.Do(func() {
close(allDoneSignal)
})
return
}
}
// Before exiting the peer goroutine, reset the query
// state to ensure other peers can pick it up.
if exiting {
atomic.StoreUint32(
&queryStates[handleQuery], uint32(queryWaitSubmit),
)
// Delete the current state of the peer, so we
// won't process any lingering responses from
// it.
mtxPeerStates.Lock()
delete(peerStates, sp.Addr())
mtxPeerStates.Unlock()
return
}
}
}
// peerQuits holds per-peer quit channels so we can kill the goroutines
// when they disconnect.
peerQuits := make(map[string]chan struct{})
// Clean up on exit.
defer func() {
for _, quitChan := range peerQuits {
close(quitChan)
}
}()
for {
// Update our view of peers, starting new workers for new peers
// and removing disconnected/banned peers.
for _, peer := range s.Peers() {
sp := peer.Addr()
if _, ok := peerQuits[sp]; !ok && peer.Connected() {
peerQuits[sp] = make(chan struct{})
go peerGoroutine(
peer, peerQuits[sp], allDoneSignal,
)
}
}
for peer, quitChan := range peerQuits {
p := s.PeerByAddr(peer)
if p == nil || !p.Connected() {
close(quitChan)
delete(peerQuits, peer)
}
}
select {
case msg := <-msgChan:
mtxPeerStates.RLock()
curQuery, ok := peerStates[msg.sp.Addr()]
mtxPeerStates.RUnlock()
// Break if we didn't expect a response from this peer.
if !ok {
break
}
if checkResponse(msg.sp, curQuery.msg, msg.msg) {
select {
case <-queryQuit:
return
case <-s.quit:
return
case curQuery.matchSignal <- struct{}{}:
}
}
case <-time.After(qo.timeout):
case <-allDoneSignal:
return
case <-queryQuit:
return
case <-s.quit:
return
}
}
}
// queryAllPeers is a helper function that sends a query to all peers and waits
// for a timeout specified by the QueryTimeout package-level variable or the
// Timeout functional option. The NumRetries option is set to 1 by default
// unless overridden by the caller.
func (s *ChainService) queryAllPeers(
// queryMsg is the message to broadcast to all peers.
queryMsg wire.Message,
// checkResponse is called for every message within the timeout period.
// The quit channel lets the query know to terminate because the
// required response has been found. This is done by closing the
// channel. The peerQuit lets the query know to terminate the query for
// the peer which sent the response, allowing releasing resources for
// peers which respond quickly while continuing to wait for slower
// peers to respond and nonresponsive peers to time out.
checkResponse func(sp *ServerPeer, resp wire.Message,
quit chan<- struct{}, peerQuit chan<- struct{}),
// options takes functional options for executing the query.
options ...QueryOption) {
// Starting with the set of default options, we'll apply any specified
// functional options to the query.
qo := defaultQueryOptions()
qo.numRetries = 1
qo.applyQueryOptions(options...)
// This is done in a single-threaded query because the peerState is
// held in a single thread. This is the only part of the query
// framework that requires access to peerState, so it's done once per
// query.
peers := s.Peers()
// This will be shared state between the per-peer goroutines.
queryQuit := make(chan struct{})
allQuit := make(chan struct{})
var wg sync.WaitGroup
msgChan := make(chan spMsg)
subscription := spMsgSubscription{
msgChan: msgChan,
quitChan: allQuit,
}
// Now we start a goroutine for each peer which manages the peer's
// message subscription.
peerQuits := make(map[string]chan struct{})
for _, sp := range peers {
sp.subscribeRecvMsg(subscription)
wg.Add(1)
peerQuits[sp.Addr()] = make(chan struct{})
go func(sp *ServerPeer, peerQuit <-chan struct{}) {
defer wg.Done()
defer sp.unsubscribeRecvMsgs(subscription)
for i := uint8(0); i < qo.numRetries; i++ {
timeout := time.After(qo.timeout)
sp.QueueMessageWithEncoding(queryMsg,
nil, qo.encoding)
select {
case <-queryQuit:
return
case <-s.quit:
return
case <-peerQuit:
return
case <-timeout:
}
}
}(sp, peerQuits[sp.Addr()])
}
// This goroutine will wait until all of the peer-query goroutines have
// terminated, and then initiate a query shutdown.
go func() {
wg.Wait()
// Make sure our main goroutine and the subscription know to
// quit.
close(allQuit)
// Close the done channel, if any.
if qo.doneChan != nil {
close(qo.doneChan)
}
}()
// Loop for any messages sent to us via our subscription channel and
// check them for whether they satisfy the query. Break the loop when
// allQuit is closed.
checkResponses:
for {
select {
case <-queryQuit:
break checkResponses
case <-s.quit:
break checkResponses
case <-allQuit:
break checkResponses
// A message has arrived over the subscription channel, so we
// execute the checkResponses callback to see if this ends our
// query session.
case sm := <-msgChan:
// TODO: This will get stuck if checkResponse gets
// stuck. This is a caveat for callers that should be
// fixed before exposing this function for public use.
select {
case <-peerQuits[sm.sp.Addr()]:
default:
checkResponse(sm.sp, sm.msg, queryQuit,
peerQuits[sm.sp.Addr()])
}
}
}
}
// queryChainServicePeers is a helper function that sends a query to one or
// more peers of the given ChainService, and waits for an answer. The timeout
// for queries is set by the QueryTimeout package-level variable or the Timeout
// functional option.
func queryChainServicePeers(
// s is the ChainService to use.
s *ChainService,
// queryMsg is the message to send to each peer selected by selectPeer.
queryMsg wire.Message,
// checkResponse is called for every message within the timeout period.
// The quit channel lets the query know to terminate because the
// required response has been found. This is done by closing the
// channel.
checkResponse func(sp *ServerPeer, resp wire.Message,
quit chan<- struct{}),
// options takes functional options for executing the query.
options ...QueryOption) {
// Starting with the set of default options, we'll apply any specified
// functional options to the query.
qo := defaultQueryOptions()
qo.applyQueryOptions(options...)
// We get an initial view of our peers, to be updated each time a peer
// query times out.
queryPeer := s.blockManager.SyncPeer()
peerTries := make(map[string]uint8)
// This will be state used by the peer query goroutine.
queryQuit := make(chan struct{})
subQuit := make(chan struct{})
// Increase this number to be able to handle more queries at once as
// each channel gets results for all queries, otherwise messages can
// get mixed and there's a vicious cycle of retries causing a bigger
// message flood, more of which get missed.
msgChan := make(chan spMsg)
subscription := spMsgSubscription{
msgChan: msgChan,
quitChan: subQuit,
}
// Loop for any messages sent to us via our subscription channel and
// check them for whether they satisfy the query. Break the loop if
// it's time to quit.
peerTimeout := time.NewTimer(qo.timeout)
connectionTimeout := time.NewTimer(qo.peerConnectTimeout)
connectionTicker := connectionTimeout.C
if queryPeer != nil {
peerTries[queryPeer.Addr()]++
queryPeer.subscribeRecvMsg(subscription)
queryPeer.QueueMessageWithEncoding(queryMsg, nil, qo.encoding)
}
checkResponses:
for {
select {
case <-connectionTicker:
// When we time out, we're done.
if queryPeer != nil {
queryPeer.unsubscribeRecvMsgs(subscription)
}
break checkResponses
case <-queryQuit:
// Same when we get a quit signal.
if queryPeer != nil {
queryPeer.unsubscribeRecvMsgs(subscription)
}
break checkResponses
case <-s.quit:
// Same when chain server's quit is signaled.
if queryPeer != nil {
queryPeer.unsubscribeRecvMsgs(subscription)
}
break checkResponses
// A message has arrived over the subscription channel, so we
// execute the checkResponses callback to see if this ends our
// query session.
case sm := <-msgChan:
// TODO: This will get stuck if checkResponse gets
// stuck. This is a caveat for callers that should be
// fixed before exposing this function for public use.
checkResponse(sm.sp, sm.msg, queryQuit)
// Each time we receive a response from the current
// peer, we'll reset the main peer timeout as they're
// being responsive.
if !peerTimeout.Stop() {
select {
case <-peerTimeout.C:
default:
}
}
peerTimeout.Reset(qo.timeout)
// Also at this point, if the peerConnectTimeout is
// still active, then we can disable it, as we're
// receiving responses from the current peer.
if connectionTicker != nil && !connectionTimeout.Stop() {
select {
case <-connectionTimeout.C:
default:
}
}
connectionTicker = nil
// The current peer we're querying has failed to answer the
// query. Time to select a new peer and query it.
case <-peerTimeout.C:
if queryPeer != nil {
queryPeer.unsubscribeRecvMsgs(subscription)
}
queryPeer = nil
for _, peer := range s.Peers() {
// If the peer is no longer connected, we'll
// skip them.
if !peer.Connected() {
continue
}
// If we've yet to try this peer, we'll make
// sure to do so. If we've exceeded the number
// of tries we should retry this peer, then
// we'll skip them.
numTries, ok := peerTries[peer.Addr()]
if ok && numTries >= qo.numRetries {
continue
}
queryPeer = peer
// Found a peer we can query.
peerTries[queryPeer.Addr()]++
queryPeer.subscribeRecvMsg(subscription)
queryPeer.QueueMessageWithEncoding(
queryMsg, nil, qo.encoding,
)
break
}
// If at this point, we don't yet have a query peer,
// then we'll exit now as all the peers are exhausted.
if queryPeer == nil {
break checkResponses
}
}
}
// Close the subscription quit channel and the done channel, if any.
close(subQuit)
peerTimeout.Stop()
if qo.doneChan != nil {
close(qo.doneChan)
}
}
// getFilterFromCache returns a filter from ChainService's FilterCache if it
// exists, returning nil and error if it doesn't.
func (s *ChainService) getFilterFromCache(blockHash *chainhash.Hash,
filterType filterdb.FilterType) (*gcs.Filter, error) {
cacheKey := cache.FilterCacheKey{*blockHash, filterType}
filterValue, err := s.FilterCache.Get(cacheKey)
if err != nil {
return nil, err
}
return filterValue.(*cache.CacheableFilter).Filter, nil
}
// putFilterToCache inserts a given filter in ChainService's FilterCache.
func (s *ChainService) putFilterToCache(blockHash *chainhash.Hash,
filterType filterdb.FilterType, filter *gcs.Filter) (bool, error) {
cacheKey := cache.FilterCacheKey{*blockHash, filterType}
return s.FilterCache.Put(cacheKey, &cache.CacheableFilter{Filter: filter})
}
// cfiltersQuery is a struct that holds all the information necessary to
// perform batch GetCFilters request, and handle the responses.
type cfiltersQuery struct {
filterType wire.FilterType
startHeight int64
stopHeight int64
stopHash *chainhash.Hash
filterHeaders []chainhash.Hash
headerIndex map[chainhash.Hash]int
targetHash chainhash.Hash
filterChan chan *gcs.Filter
options []QueryOption
}
// queryMsg returns the wire message to perform this query.
func (q *cfiltersQuery) queryMsg() wire.Message {
return wire.NewMsgGetCFilters(
q.filterType, uint32(q.startHeight), q.stopHash,
)
}
// prepareCFiltersQuery creates a cfiltersQuery that can be used to fetch a
// CFilter fo the given block hash.
func (s *ChainService) prepareCFiltersQuery(blockHash chainhash.Hash,
filterType wire.FilterType, options ...QueryOption) (
*cfiltersQuery, error) {
_, height, err := s.BlockHeaders.FetchHeader(&blockHash)
if err != nil {
return nil, fmt.Errorf("unable to get header for start "+
"block=%v: %v", blockHash, err)
}
bestBlock, err := s.BestBlock()
if err != nil {
return nil, fmt.Errorf("unable to get best block: %v", err)
}
bestHeight := int64(bestBlock.Height)
qo := defaultQueryOptions()
qo.applyQueryOptions(options...)
// If the query specifies an optimistic batch we will attempt to fetch
// the maximum number of filters in anticipation of calls for the
// following or preceding filters.
var startHeight, stopHeight int64
switch qo.optimisticBatch {
// No batching, the start and stop height will be the same.
case noBatch:
startHeight = int64(height)
stopHeight = int64(height)
// Forward batch, fetch as many of the following filters as possible.
case forwardBatch:
startHeight = int64(height)
stopHeight = startHeight + wire.MaxGetCFiltersReqRange - 1
// We need a longer timeout, since we are going to receive more
// than a single response.
options = append(options, Timeout(QueryBatchTimeout))
// Reverse batch, fetch as many of the preceding filters as possible.
case reverseBatch:
stopHeight = int64(height)
startHeight = stopHeight - wire.MaxGetCFiltersReqRange + 1
// We need a longer timeout, since we are going to receive more
// than a single response.
options = append(options, Timeout(QueryBatchTimeout))
}
// Block 1 is the earliest one we can fetch.
if startHeight < 1 {
startHeight = 1
}
// If the stop height with the maximum batch size is above our best
// known block, then we use the best block height instead.
if stopHeight > bestHeight {
stopHeight = bestHeight
}
stopHash, err := s.GetBlockHash(stopHeight)
if err != nil {
return nil, fmt.Errorf("unable to get hash for "+
"stopHeight=%d: %v", stopHeight, err)
}
// In order to verify the authenticity of the received filters, we'll
// fetch the block headers and filter headers in the range
// [startHeight-1, stopHeight]. We go one below our startHeight since
// the hash of the previous block is needed for validation.
numFilters := uint32(stopHeight - startHeight + 1)
blockHeaders, _, err := s.BlockHeaders.FetchHeaderAncestors(
numFilters, stopHash,
)
if err != nil {
return nil, fmt.Errorf("unable to get %d block header "+
"ancestors for stopHash=%v: %v", numFilters,
stopHash, err)
}
if len(blockHeaders) != int(numFilters)+1 {
return nil, fmt.Errorf("expected %d block headers, got %d",
numFilters+1, len(blockHeaders))
}
filterHeaders, _, err := s.RegFilterHeaders.FetchHeaderAncestors(
numFilters, stopHash,
)