eth, les, eth/filter: paralellized log search in bloom filter matches

accounts/abi/bind/backends/simulated.go

@@ -75,7 +75,7 @@ func NewSimulatedBackend(alloc core.GenesisAlloc, gasLimit uint64) *SimulatedBac
 		database:   database,
 		blockchain: blockchain,
 		config:     genesis.Config,
-		events:     filters.NewEventSystem(new(event.TypeMux), &filterBackend{database, blockchain}, false),
+		events:     filters.NewEventSystem(new(event.TypeMux), &filterBackend{database, blockchain, nil}, false),
 	}
 	backend.rollback()
 	return backend
@@ -358,7 +358,27 @@ func (b *SimulatedBackend) FilterLogs(ctx context.Context, query ethereum.Filter
 			to = query.ToBlock.Int64()
 		}
 		// Construct the range filter
-		filter = filters.NewRangeFilter(&filterBackend{b.database, b.blockchain}, from, to, query.Addresses, query.Topics)
+		filterTasks := make(chan *filters.BlockFilterTask)
+		filter = filters.NewRangeFilter(&filterBackend{b.database, b.blockchain, filterTasks}, from, to, query.Addresses, query.Topics)
+
+		// start up a filter task service
+		const filterTaskServiceThreads = 16
+		stop := make(chan struct{})
+		defer close(stop)
+
+		for i := 0; i < filterTaskServiceThreads; i++ {
+			go func() {
+				for {
+					select {
+					case <-stop:
+						return
+
+					case task := <-filterTasks:
+						task.Do()
+					}
+				}
+			}()
+		}
 	}
 	// Run the filter and return all the logs
 	logs, err := filter.Logs(ctx)
@@ -441,8 +461,9 @@ func (m callmsg) Data() []byte         { return m.CallMsg.Data }
 // filterBackend implements filters.Backend to support filtering for logs without
 // taking bloom-bits acceleration structures into account.
 type filterBackend struct {
-	db ethdb.Database
+	db          ethdb.Database
-	bc *core.BlockChain
+	bc          *core.BlockChain
+	filterTasks chan *filters.BlockFilterTask
 }
 
 func (fb *filterBackend) ChainDb() ethdb.Database  { return fb.db }
@@ -503,3 +524,6 @@ func (fb *filterBackend) BloomStatus() (uint64, uint64) { return 4096, 0 }
 func (fb *filterBackend) ServiceFilter(ctx context.Context, ms *bloombits.MatcherSession) {
 	panic("not supported")
 }
+func (fb *filterBackend) FilterTaskChannel() chan *filters.BlockFilterTask {
+	return fb.filterTasks
+}

eth/api_backend.go

@@ -29,6 +29,7 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/core/vm"
 	"github.com/ethereum/go-ethereum/eth/downloader"
+	"github.com/ethereum/go-ethereum/eth/filters"
 	"github.com/ethereum/go-ethereum/eth/gasprice"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/event"
@@ -223,3 +224,7 @@ func (b *EthAPIBackend) ServiceFilter(ctx context.Context, session *bloombits.Ma
 		go session.Multiplex(bloomRetrievalBatch, bloomRetrievalWait, b.eth.bloomRequests)
 	}
 }
+
+func (b *EthApiBackend) FilterTaskChannel() chan *filters.BlockFilterTask {
+	return b.eth.filterTasks
+}

eth/backend.go

@@ -80,6 +80,7 @@ type Ethereum struct {
 	accountManager *accounts.Manager
 
 	bloomRequests chan chan *bloombits.Retrieval // Channel receiving bloom data retrieval requests
+	filterTasks   chan *filters.BlockFilterTask  // Channel receiving block filter tasks
 	bloomIndexer  *core.ChainIndexer             // Bloom indexer operating during block imports
 
 	APIBackend *EthAPIBackend
@@ -142,6 +143,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
 		gasPrice:       config.MinerGasPrice,
 		etherbase:      config.Etherbase,
 		bloomRequests:  make(chan chan *bloombits.Retrieval),
+		filterTasks:    make(chan *filters.BlockFilterTask),
 		bloomIndexer:   NewBloomIndexer(chainDb, params.BloomBitsBlocks, params.BloomConfirms),
 	}
 

eth/bloombits.go

@@ -34,6 +34,10 @@ const (
 	// instance to service bloombits lookups for all running filters.
 	bloomServiceThreads = 16
 
+	// filterTaskServiceThreads is the number of goroutines used globally by an Ethereum
+	// instance to service block filter tasks for all running filters.
+	filterTaskServiceThreads = 16
+
 	// bloomFilterThreads is the number of goroutines used locally per filter to
 	// multiplex requests onto the global servicing goroutines.
 	bloomFilterThreads = 3
@@ -77,6 +81,20 @@ func (eth *Ethereum) startBloomHandlers(sectionSize uint64) {
 			}
 		}()
 	}
+
+	for i := 0; i < filterTaskServiceThreads; i++ {
+		go func() {
+			for {
+				select {
+				case <-eth.shutdownChan:
+					return
+
+				case task := <-eth.filterTasks:
+					task.Do()
+				}
+			}
+		}()
+	}
 }
 
 const (

eth/filters/filter.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"math/big"
+	"sync"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core"
@@ -45,6 +46,41 @@ type Backend interface {
 
 	BloomStatus() (uint64, uint64)
 	ServiceFilter(ctx context.Context, session *bloombits.MatcherSession)
+	FilterTaskChannel() chan *BlockFilterTask
+}
+
+// BlockFilterTask represents a request for log filtering task assignments for a given
+// block number, or a response for such a request.
+// It can also have the actual results set to be used as a delivery data struct.
+//
+// The contest and error fields are used by the light client to terminate matching
+// early if an error is enountered on some path of the pipeline.
+type BlockFilterTask struct {
+	blockNumber uint64
+	checkBloom  bool
+	filter      *Filter
+	context     context.Context
+
+	logs []*types.Log
+	err  error
+	done chan struct{}
+}
+
+func (t *BlockFilterTask) Do() {
+	defer close(t.done)
+
+	header, err := t.filter.backend.HeaderByNumber(t.context, rpc.BlockNumber(t.blockNumber))
+	if err != nil {
+		t.err = err
+		return
+	}
+	if header == nil {
+		t.err = errors.New("Header not found")
+		return
+	}
+	if !t.checkBloom || bloomFilter(header.Bloom, t.filter.addresses, t.filter.topics) {
+		t.logs, t.err = t.filter.checkMatches(t.context, header)
+	}
 }
 
 // Filter can be used to retrieve and filter logs.
@@ -146,40 +182,60 @@ func (f *Filter) Logs(ctx context.Context) ([]*types.Log, error) {
 	var (
 		logs []*types.Log
 		err  error
+		wg   sync.WaitGroup
 	)
+	tasks := make(chan *BlockFilterTask, 64)
+	wg.Add(1)
+	go func() {
+		for task := range tasks {
+			if err == nil {
+				select {
+				case <-task.done:
+				case <-ctx.Done():
+					err = ctx.Err()
+					continue
+				}
+				if len(task.logs) != 0 {
+					logs = append(logs, task.logs...)
+				}
+				err = task.err
+			}
+		}
+		wg.Done()
+	}()
+
 	size, sections := f.backend.BloomStatus()
 	if indexed := sections * size; indexed > uint64(f.begin) {
 		if indexed > end {
-			logs, err = f.indexedLogs(ctx, end)
+			f.indexedLogs(ctx, tasks, end)
 		} else {
-			logs, err = f.indexedLogs(ctx, indexed-1)
+			f.indexedLogs(ctx, tasks, indexed-1)
 		}
 		if err != nil {
 			return logs, err
 		}
 	}
-	rest, err := f.unindexedLogs(ctx, end)
+	f.unindexedLogs(ctx, tasks, end)
-	logs = append(logs, rest...)
+	close(tasks)
+	wg.Wait()
 	return logs, err
 }
 
 // indexedLogs returns the logs matching the filter criteria based on the bloom
 // bits indexed available locally or via the network.
-func (f *Filter) indexedLogs(ctx context.Context, end uint64) ([]*types.Log, error) {
+func (f *Filter) indexedLogs(ctx context.Context, tasks chan *BlockFilterTask, end uint64) error {
 	// Create a matcher session and request servicing from the backend
 	matches := make(chan uint64, 64)
 
 	session, err := f.matcher.Start(ctx, uint64(f.begin), end, matches)
 	if err != nil {
-		return nil, err
+		return err
 	}
 	defer session.Close()
 
 	f.backend.ServiceFilter(ctx, session)
 
 	// Iterate over the matches until exhausted or context closed
-	var logs []*types.Log
-
 	for {
 		select {
 		case number, ok := <-matches:
@@ -189,44 +245,58 @@ func (f *Filter) indexedLogs(ctx context.Context, end uint64) ([]*types.Log, err
 				if err == nil {
 					f.begin = int64(end) + 1
 				}
-				return logs, err
+				return err
 			}
 			f.begin = int64(number) + 1
 
-			// Retrieve the suggested block and pull any truly matching logs
+			task := &BlockFilterTask{
-			header, err := f.backend.HeaderByNumber(ctx, rpc.BlockNumber(number))
+				blockNumber: number,
-			if header == nil || err != nil {
+				checkBloom:  false,
-				return logs, err
+				filter:      f,
+				context:     ctx,
+				done:        make(chan struct{}),
 			}
-			found, err := f.checkMatches(ctx, header)
+			select {
-			if err != nil {
+			case f.backend.FilterTaskChannel() <- task:
-				return logs, err
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+			select {
+			case tasks <- task:
+			case <-ctx.Done():
+				return ctx.Err()
 			}
-			logs = append(logs, found...)
 
 		case <-ctx.Done():
-			return logs, ctx.Err()
+			return ctx.Err()
 		}
 	}
 }
 
 // indexedLogs returns the logs matching the filter criteria based on raw block
 // iteration and bloom matching.
-func (f *Filter) unindexedLogs(ctx context.Context, end uint64) ([]*types.Log, error) {
+func (f *Filter) unindexedLogs(ctx context.Context, tasks chan *BlockFilterTask, end uint64) error {
-	var logs []*types.Log
-
 	for ; f.begin <= int64(end); f.begin++ {
-		header, err := f.backend.HeaderByNumber(ctx, rpc.BlockNumber(f.begin))
+		task := &BlockFilterTask{
-		if header == nil || err != nil {
+			blockNumber: uint64(f.begin),
-			return logs, err
+			checkBloom:  true,
+			filter:      f,
+			context:     ctx,
+			done:        make(chan struct{}),
 		}
-		found, err := f.blockLogs(ctx, header)
+		select {
-		if err != nil {
+		case f.backend.FilterTaskChannel() <- task:
-			return logs, err
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+		select {
+		case tasks <- task:
+		case <-ctx.Done():
+			return ctx.Err()
 		}
 		logs = append(logs, found...)
 	}
-	return logs, nil
+	return nil
 }
 
 // blockLogs returns the logs matching the filter criteria within a single block.
@@ -236,7 +306,7 @@ func (f *Filter) blockLogs(ctx context.Context, header *types.Header) (logs []*t
 		if err != nil {
 			return logs, err
 		}
-		logs = append(logs, found...)
+		logs = found
 	}
 	return logs, nil
 }

les/api_backend.go

@@ -30,6 +30,7 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/core/vm"
 	"github.com/ethereum/go-ethereum/eth/downloader"
+	"github.com/ethereum/go-ethereum/eth/filters"
 	"github.com/ethereum/go-ethereum/eth/gasprice"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/event"
@@ -200,3 +201,7 @@ func (b *LesApiBackend) ServiceFilter(ctx context.Context, session *bloombits.Ma
 		go session.Multiplex(bloomRetrievalBatch, bloomRetrievalWait, b.eth.bloomRequests)
 	}
 }
+
+func (b *LesApiBackend) FilterTaskChannel() chan *filters.BlockFilterTask {
+	return b.eth.filterTasks
+}

les/backend.go

@@ -63,6 +63,7 @@ type LightEthereum struct {
 	retriever  *retrieveManager
 
 	bloomRequests chan chan *bloombits.Retrieval // Channel receiving bloom data retrieval requests
+	filterTasks   chan *filters.BlockFilterTask  // Channel receiving block filter tasks
 	bloomIndexer  *core.ChainIndexer
 
 	ApiBackend *LesApiBackend
@@ -106,6 +107,7 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) {
 		shutdownChan:   make(chan bool),
 		networkId:      config.NetworkId,
 		bloomRequests:  make(chan chan *bloombits.Retrieval),
+		filterTasks:    make(chan *filters.BlockFilterTask),
 		bloomIndexer:   eth.NewBloomIndexer(chainDb, params.BloomBitsBlocksClient, params.HelperTrieConfirmations),
 	}
 

les/bloombits.go

@@ -28,6 +28,10 @@ const (
 	// instance to service bloombits lookups for all running filters.
 	bloomServiceThreads = 16
 
+	// filterTaskServiceThreads is the number of goroutines used globally by an Ethereum
+	// instance to service block filter tasks for all running filters.
+	filterTaskServiceThreads = 16
+
 	// bloomFilterThreads is the number of goroutines used locally per filter to
 	// multiplex requests onto the global servicing goroutines.
 	bloomFilterThreads = 3
@@ -71,4 +75,18 @@ func (eth *LightEthereum) startBloomHandlers(sectionSize uint64) {
 			}
 		}()
 	}
+
+	for i := 0; i < filterTaskServiceThreads; i++ {
+		go func() {
+			for {
+				select {
+				case <-eth.shutdownChan:
+					return
+
+				case task := <-eth.filterTasks:
+					task.Do()
+				}
+			}
+		}()
+	}
 }