This repository has been archived by the owner on Feb 3, 2020. It is now read-only.
forked from lightninglabs/neutrino
-
Notifications
You must be signed in to change notification settings - Fork 2
/
query.go
1379 lines (1188 loc) · 41.1 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/wakiyamap/monad/blockchain"
"github.com/wakiyamap/monad/chaincfg/chainhash"
"github.com/wakiyamap/monad/wire"
"github.com/wakiyamap/monautil"
"github.com/wakiyamap/monautil/gcs"
"github.com/wakiyamap/monautil/gcs/builder"
"github.com/davecgh/go-spew/spew"
"github.com/wakiyamap/neutrino/cache"
"github.com/wakiyamap/neutrino/filterdb"
"github.com/wakiyamap/neutrino/pushtx"
)
var (
// QueryTimeout specifies how long to wait for a peer to answer a
// query.
QueryTimeout = time.Second * 3
// 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
// 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
)
// 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)
// peerStates and its companion mutex allow the peer goroutines to
// tell the main goroutine what query they're currently working on.
peerStates := make(map[string]wire.Message)
var mtxPeerStates sync.RWMutex
peerGoroutine := func(sp *ServerPeer, quit <-chan struct{},
matchSignal <-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.
mtxPeerStates.Lock()
peerStates[sp.Addr()] = queryMsgs[handleQuery]
mtxPeerStates.Unlock()
select {
case <-queryQuit:
return
case <-s.quit:
return
case <-quit:
return
case <-timeout:
// We failed, so set the query state back to
// zero and update our lastFailed state.
atomic.StoreUint32(
&queryStates[handleQuery], uint32(queryWaitSubmit),
)
log.Tracef("Query for #%v failed, moving "+
"on: %v", handleQuery,
newLogClosure(func() string {
return spew.Sdump(queryMsgs[handleQuery])
}))
case <-matchSignal:
// We got a match signal so we can mark this
// query a success.
atomic.StoreUint32(&queryStates[handleQuery],
uint32(queryAnswered))
log.Tracef("Query #%v answered, updating state",
handleQuery)
}
}
}
// peerQuits holds per-peer quit channels so we can kill the goroutines
// when they disconnect.
peerQuits := make(map[string]chan struct{})
// matchSignals holds per-peer answer channels that get a notice that
// 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.
matchSignals := 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{})
matchSignals[sp] = make(chan struct{})
go peerGoroutine(
peer, peerQuits[sp], matchSignals[sp],
)
}
}
for peer, quitChan := range peerQuits {
p := s.PeerByAddr(peer)
if p == nil || !p.Connected() {
close(quitChan)
close(matchSignals[peer])
delete(peerQuits, peer)
delete(matchSignals, peer)
}
}
select {
case msg := <-msgChan:
mtxPeerStates.RLock()
curQuery := peerStates[msg.sp.Addr()]
mtxPeerStates.RUnlock()
if checkResponse(msg.sp, curQuery, msg.msg) {
select {
case <-queryQuit:
return
case <-s.quit:
return
case matchSignals[msg.sp.Addr()] <- struct{}{}:
}
}
case <-time.After(qo.timeout):
// Check if we're done; if so, quit.
allDone := true
for i := 0; i < len(queryStates); i++ {
if atomic.LoadUint32(&queryStates[i]) !=
uint32(queryAnswered) {
allDone = false
}
}
if allDone {
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.NewTicker(qo.timeout)
timeout := time.After(qo.peerConnectTimeout)
if queryPeer != nil {
peerTries[queryPeer.Addr()]++
queryPeer.subscribeRecvMsg(subscription)
queryPeer.QueueMessageWithEncoding(queryMsg, nil, qo.encoding)
}
checkResponses:
for {
select {
case <-timeout:
// 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)
// 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,
)
if err != nil {
return nil, fmt.Errorf("unable to get %d filter header "+
"ancestors for stopHash=%v: %v", numFilters, stopHash,
err)
}
if len(filterHeaders) != int(numFilters)+1 {
return nil, fmt.Errorf("expected %d filter headers, got %d",
numFilters+1, len(filterHeaders))
}
// We create a header index such that we can easily index into our
// header slices for a given block hash in the received response,
// without consulting the database. This also keeps track of which
// blocks we are still awaiting a response for. We start at index=1, as
// 0 is for the block startHeight-1, which is only needed for
// validation.
headerIndex := make(map[chainhash.Hash]int, len(blockHeaders)-1)
for i := 1; i < len(blockHeaders); i++ {
block := blockHeaders[i]
headerIndex[block.BlockHash()] = i
}
// We'll immediately respond to the caller with the requested filter
// when it is received, so we make a channel to notify on when it's
// ready.
filterChan := make(chan *gcs.Filter, 1)
return &cfiltersQuery{
filterType: filterType,
startHeight: startHeight,
stopHeight: stopHeight,
stopHash: stopHash,
filterHeaders: filterHeaders,
headerIndex: headerIndex,
targetHash: blockHash,
filterChan: filterChan,
options: options,
}, nil
}
// handleCFiltersRespons is called every time we receive a response for the
// GetCFilters request.
func (s *ChainService) handleCFiltersResponse(q *cfiltersQuery,
resp wire.Message, quit chan<- struct{}) {
// We're only interested in "cfilter" messages.
response, ok := resp.(*wire.MsgCFilter)
if !ok {
return
}
// If the response doesn't match our request, ignore this message.
if q.filterType != response.FilterType {
return
}
// If this filter is for a block not in our index, we can ignore it, as
// we either already got it, or it is out of our queried range.
i, ok := q.headerIndex[response.BlockHash]
if !ok {
return
}
gotFilter, err := gcs.FromNBytes(
builder.DefaultP, builder.DefaultM, response.Data,
)
if err != nil {
// Malformed filter data. We can ignore this message.
return
}
// Now that we have a proper filter, ensure that re-calculating the
// filter header hash for the header _after_ the filter in the chain
// checks out. If not, we can ignore this response.
curHeader := q.filterHeaders[i]
prevHeader := q.filterHeaders[i-1]
gotHeader, err := builder.MakeHeaderForFilter(
gotFilter, prevHeader,
)
if err != nil {
return
}
if gotHeader != curHeader {
return
}
// At this point, the filter matches what we know about it and we
// declare it sane. If this is the filter requested initially, send it
// to the caller immediately.
if response.BlockHash == q.targetHash {
q.filterChan <- gotFilter
}
// Put the filter in the cache and persistToDisk if the caller
// requested it.
// TODO(halseth): for an LRU we could take care to insert the next
// height filter last.
dbFilterType := filterdb.RegularFilter
evict, err := s.putFilterToCache(
&response.BlockHash, dbFilterType, gotFilter,
)
if err != nil {
log.Warnf("Couldn't write filter to cache: %v", err)
}
// TODO(halseth): dynamically increase/decrease the batch size to match
// our cache capacity.
numFilters := q.stopHeight - q.startHeight + 1
if evict && s.FilterCache.Len() < int(numFilters) {
log.Debugf("Items evicted from the cache with less "+
"than %d elements. Consider increasing the "+
"cache size...", numFilters)
}