diff --git a/core/bloombits/matcher.go b/core/bloombits/matcher.go index baab3888aa..32a660337f 100644 --- a/core/bloombits/matcher.go +++ b/core/bloombits/matcher.go @@ -18,6 +18,7 @@ package bloombits import ( "bytes" + "context" "errors" "math" "sort" @@ -61,6 +62,7 @@ type Retrieval struct { Sections []uint64 Bitsets [][]byte Error error + Context context.Context } // Matcher is a pipelined system of schedulers and logic matchers which perform @@ -138,7 +140,7 @@ func (m *Matcher) addScheduler(idx uint) { // Start starts the matching process and returns a stream of bloom matches in // a given range of blocks. If there are no more matches in the range, the result // channel is closed. -func (m *Matcher) Start(begin, end uint64, results chan uint64) (*MatcherSession, error) { +func (m *Matcher) Start(ctx context.Context, begin, end uint64, results chan uint64) (*MatcherSession, error) { // Make sure we're not creating concurrent sessions if atomic.SwapUint32(&m.running, 1) == 1 { return nil, errors.New("matcher already running") @@ -150,6 +152,7 @@ func (m *Matcher) Start(begin, end uint64, results chan uint64) (*MatcherSession matcher: m, quit: make(chan struct{}), kill: make(chan struct{}), + ctx: ctx, } for _, scheduler := range m.schedulers { scheduler.reset() @@ -505,6 +508,7 @@ type MatcherSession struct { quit chan struct{} // Quit channel to request pipeline termination kill chan struct{} // Term channel to signal non-graceful forced shutdown + ctx context.Context err error stopping bool lock sync.Mutex @@ -647,7 +651,7 @@ func (s *MatcherSession) Multiplex(batch int, wait time.Duration, mux chan chan case mux <- request: // Retrieval accepted, something must arrive before we're aborting - request <- &Retrieval{Bit: bit, Sections: sections} + request <- &Retrieval{Bit: bit, Sections: sections, Context: s.ctx} result := <-request if result.Error != nil { diff --git a/core/bloombits/matcher_test.go b/core/bloombits/matcher_test.go index 86db380a87..af6418e02b 100644 --- a/core/bloombits/matcher_test.go +++ b/core/bloombits/matcher_test.go @@ -17,6 +17,7 @@ package bloombits import ( + "context" "math/rand" "sync/atomic" "testing" @@ -144,7 +145,7 @@ func testMatcher(t *testing.T, filter [][]bloomIndexes, blocks uint64, intermitt quit := make(chan struct{}) matches := make(chan uint64, 16) - session, err := matcher.Start(0, blocks-1, matches) + session, err := matcher.Start(context.Background(), 0, blocks-1, matches) if err != nil { t.Fatalf("failed to stat matcher session: %v", err) } @@ -169,7 +170,7 @@ func testMatcher(t *testing.T, filter [][]bloomIndexes, blocks uint64, intermitt quit = make(chan struct{}) matches = make(chan uint64, 16) - session, err = matcher.Start(i+1, blocks-1, matches) + session, err = matcher.Start(context.Background(), i+1, blocks-1, matches) if err != nil { t.Fatalf("failed to stat matcher session: %v", err) } diff --git a/eth/filters/filter.go b/eth/filters/filter.go index 490fd0b906..d16af84ee8 100644 --- a/eth/filters/filter.go +++ b/eth/filters/filter.go @@ -135,7 +135,7 @@ func (f *Filter) indexedLogs(ctx context.Context, end uint64) ([]*types.Log, err // Create a matcher session and request servicing from the backend matches := make(chan uint64, 64) - session, err := f.matcher.Start(uint64(f.begin), end, matches) + session, err := f.matcher.Start(ctx, uint64(f.begin), end, matches) if err != nil { return nil, err } diff --git a/les/bloombits.go b/les/bloombits.go index 91f5383905..dff83d3491 100644 --- a/les/bloombits.go +++ b/les/bloombits.go @@ -17,7 +17,6 @@ package les import ( - "context" "time" "github.com/ethereum/go-ethereum/common/bitutil" @@ -54,10 +53,8 @@ func (eth *LightEthereum) startBloomHandlers() { case request := <-eth.bloomRequests: task := <-request - task.Bitsets = make([][]byte, len(task.Sections)) - - compVectors, err := light.GetBloomBits(context.Background(), eth.odr, task.Bit, task.Sections) + compVectors, err := light.GetBloomBits(task.Context, eth.odr, task.Bit, task.Sections) if err == nil { for i, _ := range task.Sections { if blob, err := bitutil.DecompressBytes(compVectors[i], int(light.BloomTrieFrequency/8)); err == nil {