core/bloombits: added context to Retrieval

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 {

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)
 				}

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
 	}

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 {