From fc6fa6cb0c6ab023a9e93c38ca3e3b085421ee50 Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Mon, 12 Feb 2018 13:16:31 +0100 Subject: [PATCH] eth, les, eth/filter: paralellized log search in bloom filter matches --- accounts/abi/bind/backends/simulated.go | 32 +++++- eth/api_backend.go | 5 + eth/backend.go | 2 + eth/bloombits.go | 18 ++++ eth/filters/filter.go | 128 ++++++++++++++++++------ les/api_backend.go | 5 + les/backend.go | 2 + les/bloombits.go | 18 ++++ 8 files changed, 177 insertions(+), 33 deletions(-) diff --git a/accounts/abi/bind/backends/simulated.go b/accounts/abi/bind/backends/simulated.go index 400227c3a8..9ad23d1fc2 100644 --- a/accounts/abi/bind/backends/simulated.go +++ b/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 - bc *core.BlockChain + db ethdb.Database + 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 +} diff --git a/eth/api_backend.go b/eth/api_backend.go index a48815e0db..7935f5ffe1 100644 --- a/eth/api_backend.go +++ b/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 +} diff --git a/eth/backend.go b/eth/backend.go index 6a136182ab..b789fb5dff 100644 --- a/eth/backend.go +++ b/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), } diff --git a/eth/bloombits.go b/eth/bloombits.go index c7bb561402..122033fc5e 100644 --- a/eth/bloombits.go +++ b/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 ( diff --git a/eth/filters/filter.go b/eth/filters/filter.go index 071613ad7a..c7234f7956 100644 --- a/eth/filters/filter.go +++ b/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) - logs = append(logs, rest...) + f.unindexedLogs(ctx, tasks, end) + 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 - header, err := f.backend.HeaderByNumber(ctx, rpc.BlockNumber(number)) - if header == nil || err != nil { - return logs, err + task := &BlockFilterTask{ + blockNumber: number, + checkBloom: false, + filter: f, + context: ctx, + done: make(chan struct{}), } - found, err := f.checkMatches(ctx, header) - if err != nil { - return logs, err + select { + case f.backend.FilterTaskChannel() <- task: + 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) { - var logs []*types.Log - +func (f *Filter) unindexedLogs(ctx context.Context, tasks chan *BlockFilterTask, end uint64) error { for ; f.begin <= int64(end); f.begin++ { - header, err := f.backend.HeaderByNumber(ctx, rpc.BlockNumber(f.begin)) - if header == nil || err != nil { - return logs, err + task := &BlockFilterTask{ + blockNumber: uint64(f.begin), + checkBloom: true, + filter: f, + context: ctx, + done: make(chan struct{}), } - found, err := f.blockLogs(ctx, header) - if err != nil { - return logs, err + select { + case f.backend.FilterTaskChannel() <- task: + 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 } diff --git a/les/api_backend.go b/les/api_backend.go index 7531396235..4f3986da36 100644 --- a/les/api_backend.go +++ b/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 +} diff --git a/les/backend.go b/les/backend.go index cd99f8f813..721c517184 100644 --- a/les/backend.go +++ b/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), } diff --git a/les/bloombits.go b/les/bloombits.go index aea0fcd5f4..3b47cdc3d6 100644 --- a/les/bloombits.go +++ b/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() + } + } + }() + } }