accounts, eth, les: blockhash based filtering on all code paths

This commit is contained in:
Péter Szilágyi 2018-07-12 17:36:07 +03:00
parent 96339daf40
commit e1f1d3085c
No known key found for this signature in database
GPG Key ID: E9AE538CEDF8293D
8 changed files with 136 additions and 72 deletions

View File

@ -324,18 +324,24 @@ func (b *SimulatedBackend) SendTransaction(ctx context.Context, tx *types.Transa
// //
// TODO(karalabe): Deprecate when the subscription one can return past data too. // TODO(karalabe): Deprecate when the subscription one can return past data too.
func (b *SimulatedBackend) FilterLogs(ctx context.Context, query ethereum.FilterQuery) ([]types.Log, error) { func (b *SimulatedBackend) FilterLogs(ctx context.Context, query ethereum.FilterQuery) ([]types.Log, error) {
// Initialize unset filter boundaried to run from genesis to chain head var filter *filters.Filter
from := int64(0) if query.BlockHash != nil {
if query.FromBlock != nil { // Block filter requested, construct a single-shot filter
from = query.FromBlock.Int64() filter = filters.NewBlockFilter(&filterBackend{b.database, b.blockchain}, *query.BlockHash, query.Addresses, query.Topics)
} else {
// Initialize unset filter boundaried to run from genesis to chain head
from := int64(0)
if query.FromBlock != nil {
from = query.FromBlock.Int64()
}
to := int64(-1)
if query.ToBlock != nil {
to = query.ToBlock.Int64()
}
// Construct the range filter
filter = filters.NewRangeFilter(&filterBackend{b.database, b.blockchain}, from, to, query.Addresses, query.Topics)
} }
to := int64(-1) // Run the filter and return all the logs
if query.ToBlock != nil {
to = query.ToBlock.Int64()
}
// Construct and execute the filter
filter := filters.New(&filterBackend{b.database, b.blockchain}, from, to, query.Addresses, query.Topics)
logs, err := filter.Logs(ctx) logs, err := filter.Logs(ctx)
if err != nil { if err != nil {
return nil, err return nil, err
@ -430,6 +436,10 @@ func (fb *filterBackend) HeaderByNumber(ctx context.Context, block rpc.BlockNumb
return fb.bc.GetHeaderByNumber(uint64(block.Int64())), nil return fb.bc.GetHeaderByNumber(uint64(block.Int64())), nil
} }
func (fb *filterBackend) HeaderByHash(ctx context.Context, hash common.Hash) (*types.Header, error) {
return fb.bc.GetHeaderByHash(hash), nil
}
func (fb *filterBackend) GetReceipts(ctx context.Context, hash common.Hash) (types.Receipts, error) { func (fb *filterBackend) GetReceipts(ctx context.Context, hash common.Hash) (types.Receipts, error) {
number := rawdb.ReadHeaderNumber(fb.db, hash) number := rawdb.ReadHeaderNumber(fb.db, hash)
if number == nil { if number == nil {

View File

@ -70,6 +70,10 @@ func (b *EthAPIBackend) HeaderByNumber(ctx context.Context, blockNr rpc.BlockNum
return b.eth.blockchain.GetHeaderByNumber(uint64(blockNr)), nil return b.eth.blockchain.GetHeaderByNumber(uint64(blockNr)), nil
} }
func (b *EthAPIBackend) HeaderByHash(ctx context.Context, hash common.Hash) (*types.Header, error) {
return b.eth.blockchain.GetHeaderByHash(hash), nil
}
func (b *EthAPIBackend) BlockByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*types.Block, error) { func (b *EthAPIBackend) BlockByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*types.Block, error) {
// Pending block is only known by the miner // Pending block is only known by the miner
if blockNr == rpc.PendingBlockNumber { if blockNr == rpc.PendingBlockNumber {

View File

@ -28,7 +28,6 @@ import (
ethereum "github.com/ethereum/go-ethereum" ethereum "github.com/ethereum/go-ethereum"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil" "github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/core/rawdb"
"github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/event"
@ -325,36 +324,26 @@ func (api *PublicFilterAPI) NewFilter(crit FilterCriteria) (rpc.ID, error) {
// //
// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getlogs // https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getlogs
func (api *PublicFilterAPI) GetLogs(ctx context.Context, crit FilterCriteria) ([]*types.Log, error) { func (api *PublicFilterAPI) GetLogs(ctx context.Context, crit FilterCriteria) ([]*types.Log, error) {
var ( var filter *Filter
fromBlock int64
toBlock int64
)
if crit.BlockHash != nil { if crit.BlockHash != nil {
// look up block number from block hash // Block filter requested, construct a single-shot filter
if block := rawdb.ReadHeaderNumber(api.chainDb, *crit.BlockHash); block != nil { filter = NewBlockFilter(api.backend, *crit.BlockHash, crit.Addresses, crit.Topics)
// verify block is part of canonical chain
if canonical := rawdb.ReadCanonicalHash(api.chainDb, *block); canonical != *crit.BlockHash {
return nil, fmt.Errorf("Block with hash %s was removed from canonical chain", crit.BlockHash.Hex())
}
fromBlock = int64(*block)
toBlock = fromBlock
} else {
return nil, fmt.Errorf("Block with hash %s was not found", crit.BlockHash.Hex())
}
} else { } else {
// Convert the RPC block numbers into internal representations // Convert the RPC block numbers into internal representations
var (
begin int64
end int64
)
if crit.FromBlock == nil { if crit.FromBlock == nil {
fromBlock = int64(rpc.LatestBlockNumber) begin = int64(rpc.LatestBlockNumber)
} }
if crit.ToBlock == nil { if crit.ToBlock == nil {
toBlock = int64(rpc.LatestBlockNumber) end = int64(rpc.LatestBlockNumber)
} }
// Construct the range filter
filter = NewRangeFilter(api.backend, begin, end, crit.Addresses, crit.Topics)
} }
// Run the filter and return all the logs
// Create and run the filter to get all the logs
filter := New(api.backend, fromBlock, toBlock, crit.Addresses, crit.Topics)
logs, err := filter.Logs(ctx) logs, err := filter.Logs(ctx)
if err != nil { if err != nil {
return nil, err return nil, err
@ -392,17 +381,24 @@ func (api *PublicFilterAPI) GetFilterLogs(ctx context.Context, id rpc.ID) ([]*ty
return nil, fmt.Errorf("filter not found") return nil, fmt.Errorf("filter not found")
} }
begin := rpc.LatestBlockNumber.Int64() var filter *Filter
if f.crit.FromBlock != nil { if f.crit.BlockHash != nil {
begin = f.crit.FromBlock.Int64() // Block filter requested, construct a single-shot filter
filter = NewBlockFilter(api.backend, *f.crit.BlockHash, f.crit.Addresses, f.crit.Topics)
} else {
// Convert the RPC block numbers into internal representations
begin := rpc.LatestBlockNumber.Int64()
if f.crit.FromBlock != nil {
begin = f.crit.FromBlock.Int64()
}
end := rpc.LatestBlockNumber.Int64()
if f.crit.ToBlock != nil {
end = f.crit.ToBlock.Int64()
}
// Construct the range filter
filter = NewRangeFilter(api.backend, begin, end, f.crit.Addresses, f.crit.Topics)
} }
end := rpc.LatestBlockNumber.Int64() // Run the filter and return all the logs
if f.crit.ToBlock != nil {
end = f.crit.ToBlock.Int64()
}
// Create and run the filter to get all the logs
filter := New(api.backend, begin, end, f.crit.Addresses, f.crit.Topics)
logs, err := filter.Logs(ctx) logs, err := filter.Logs(ctx)
if err != nil { if err != nil {
return nil, err return nil, err

View File

@ -135,7 +135,7 @@ func benchmarkBloomBits(b *testing.B, sectionSize uint64) {
var addr common.Address var addr common.Address
addr[0] = byte(i) addr[0] = byte(i)
addr[1] = byte(i / 256) addr[1] = byte(i / 256)
filter := New(backend, 0, int64(cnt*sectionSize-1), []common.Address{addr}, nil) filter := NewRangeFilter(backend, 0, int64(cnt*sectionSize-1), []common.Address{addr}, nil)
if _, err := filter.Logs(context.Background()); err != nil { if _, err := filter.Logs(context.Background()); err != nil {
b.Error("filter.Find error:", err) b.Error("filter.Find error:", err)
} }
@ -192,7 +192,7 @@ func BenchmarkNoBloomBits(b *testing.B) {
start := time.Now() start := time.Now()
mux := new(event.TypeMux) mux := new(event.TypeMux)
backend := &testBackend{mux, db, 0, new(event.Feed), new(event.Feed), new(event.Feed), new(event.Feed)} backend := &testBackend{mux, db, 0, new(event.Feed), new(event.Feed), new(event.Feed), new(event.Feed)}
filter := New(backend, 0, int64(*headNum), []common.Address{{}}, nil) filter := NewRangeFilter(backend, 0, int64(*headNum), []common.Address{{}}, nil)
filter.Logs(context.Background()) filter.Logs(context.Background())
d := time.Since(start) d := time.Since(start)
fmt.Println("Finished running filter benchmarks") fmt.Println("Finished running filter benchmarks")

View File

@ -18,6 +18,7 @@ package filters
import ( import (
"context" "context"
"errors"
"math/big" "math/big"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
@ -33,6 +34,7 @@ type Backend interface {
ChainDb() ethdb.Database ChainDb() ethdb.Database
EventMux() *event.TypeMux EventMux() *event.TypeMux
HeaderByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*types.Header, error) HeaderByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*types.Header, error)
HeaderByHash(ctx context.Context, blockHash common.Hash) (*types.Header, error)
GetReceipts(ctx context.Context, blockHash common.Hash) (types.Receipts, error) GetReceipts(ctx context.Context, blockHash common.Hash) (types.Receipts, error)
GetLogs(ctx context.Context, blockHash common.Hash) ([][]*types.Log, error) GetLogs(ctx context.Context, blockHash common.Hash) ([][]*types.Log, error)
@ -49,17 +51,19 @@ type Backend interface {
type Filter struct { type Filter struct {
backend Backend backend Backend
db ethdb.Database db ethdb.Database
begin, end int64 addresses []common.Address
addresses []common.Address topics [][]common.Hash
topics [][]common.Hash
block common.Hash // Block hash if filtering a single block
begin, end int64 // Range interval if filtering multiple blocks
matcher *bloombits.Matcher matcher *bloombits.Matcher
} }
// New creates a new filter which uses a bloom filter on blocks to figure out whether // NewRangeFilter creates a new filter which uses a bloom filter on blocks to
// a particular block is interesting or not. // figure out whether a particular block is interesting or not.
func New(backend Backend, begin, end int64, addresses []common.Address, topics [][]common.Hash) *Filter { func NewRangeFilter(backend Backend, begin, end int64, addresses []common.Address, topics [][]common.Hash) *Filter {
// Flatten the address and topic filter clauses into a single bloombits filter // Flatten the address and topic filter clauses into a single bloombits filter
// system. Since the bloombits are not positional, nil topics are permitted, // system. Since the bloombits are not positional, nil topics are permitted,
// which get flattened into a nil byte slice. // which get flattened into a nil byte slice.
@ -78,23 +82,52 @@ func New(backend Backend, begin, end int64, addresses []common.Address, topics [
} }
filters = append(filters, filter) filters = append(filters, filter)
} }
// Assemble and return the filter
size, _ := backend.BloomStatus() size, _ := backend.BloomStatus()
// Create a generic filter and convert it into a range filter
filter := newFilter(backend, addresses, topics)
filter.matcher = bloombits.NewMatcher(size, filters)
filter.begin = begin
filter.end = end
return filter
}
// NewBlockFilter creates a new filter which directly inspects the contents of
// a block to figure out whether it is interesting or not.
func NewBlockFilter(backend Backend, block common.Hash, addresses []common.Address, topics [][]common.Hash) *Filter {
// Create a generic filter and convert it into a block filter
filter := newFilter(backend, addresses, topics)
filter.block = block
return filter
}
// newFilter creates a generic filter that can either filter based on a block hash,
// or based on range queries. The search criteria needs to be explicitly set.
func newFilter(backend Backend, addresses []common.Address, topics [][]common.Hash) *Filter {
return &Filter{ return &Filter{
backend: backend, backend: backend,
begin: begin,
end: end,
addresses: addresses, addresses: addresses,
topics: topics, topics: topics,
db: backend.ChainDb(), db: backend.ChainDb(),
matcher: bloombits.NewMatcher(size, filters),
} }
} }
// Logs searches the blockchain for matching log entries, returning all from the // Logs searches the blockchain for matching log entries, returning all from the
// first block that contains matches, updating the start of the filter accordingly. // first block that contains matches, updating the start of the filter accordingly.
func (f *Filter) Logs(ctx context.Context) ([]*types.Log, error) { func (f *Filter) Logs(ctx context.Context) ([]*types.Log, error) {
// If we're doing singleton block filtering, execute and return
if f.block != (common.Hash{}) {
header, err := f.backend.HeaderByHash(ctx, f.block)
if err != nil {
return nil, err
}
if header == nil {
return nil, errors.New("unknown block")
}
return f.blockLogs(ctx, header)
}
// Figure out the limits of the filter range // Figure out the limits of the filter range
header, _ := f.backend.HeaderByNumber(ctx, rpc.LatestBlockNumber) header, _ := f.backend.HeaderByNumber(ctx, rpc.LatestBlockNumber)
if header == nil { if header == nil {
@ -187,13 +220,23 @@ func (f *Filter) unindexedLogs(ctx context.Context, end uint64) ([]*types.Log, e
if header == nil || err != nil { if header == nil || err != nil {
return logs, err return logs, err
} }
if bloomFilter(header.Bloom, f.addresses, f.topics) { found, err := f.blockLogs(ctx, header)
found, err := f.checkMatches(ctx, header) if err != nil {
if err != nil { return logs, err
return logs, err
}
logs = append(logs, found...)
} }
logs = append(logs, found...)
}
return logs, nil
}
// blockLogs returns the logs matching the filter criteria within a single block.
func (f *Filter) blockLogs(ctx context.Context, header *types.Header) (logs []*types.Log, err error) {
if bloomFilter(header.Bloom, f.addresses, f.topics) {
found, err := f.checkMatches(ctx, header)
if err != nil {
return logs, err
}
logs = append(logs, found...)
} }
return logs, nil return logs, nil
} }

View File

@ -75,6 +75,14 @@ func (b *testBackend) HeaderByNumber(ctx context.Context, blockNr rpc.BlockNumbe
return rawdb.ReadHeader(b.db, hash, num), nil return rawdb.ReadHeader(b.db, hash, num), nil
} }
func (b *testBackend) HeaderByHash(ctx context.Context, hash common.Hash) (*types.Header, error) {
number := rawdb.ReadHeaderNumber(b.db, hash)
if number == nil {
return nil, nil
}
return rawdb.ReadHeader(b.db, hash, *number), nil
}
func (b *testBackend) GetReceipts(ctx context.Context, hash common.Hash) (types.Receipts, error) { func (b *testBackend) GetReceipts(ctx context.Context, hash common.Hash) (types.Receipts, error) {
if number := rawdb.ReadHeaderNumber(b.db, hash); number != nil { if number := rawdb.ReadHeaderNumber(b.db, hash); number != nil {
return rawdb.ReadReceipts(b.db, hash, *number), nil return rawdb.ReadReceipts(b.db, hash, *number), nil

View File

@ -92,7 +92,7 @@ func BenchmarkFilters(b *testing.B) {
} }
b.ResetTimer() b.ResetTimer()
filter := New(backend, 0, -1, []common.Address{addr1, addr2, addr3, addr4}, nil) filter := NewRangeFilter(backend, 0, -1, []common.Address{addr1, addr2, addr3, addr4}, nil)
for i := 0; i < b.N; i++ { for i := 0; i < b.N; i++ {
logs, _ := filter.Logs(context.Background()) logs, _ := filter.Logs(context.Background())
@ -175,14 +175,14 @@ func TestFilters(t *testing.T) {
rawdb.WriteReceipts(db, block.Hash(), block.NumberU64(), receipts[i]) rawdb.WriteReceipts(db, block.Hash(), block.NumberU64(), receipts[i])
} }
filter := New(backend, 0, -1, []common.Address{addr}, [][]common.Hash{{hash1, hash2, hash3, hash4}}) filter := NewRangeFilter(backend, 0, -1, []common.Address{addr}, [][]common.Hash{{hash1, hash2, hash3, hash4}})
logs, _ := filter.Logs(context.Background()) logs, _ := filter.Logs(context.Background())
if len(logs) != 4 { if len(logs) != 4 {
t.Error("expected 4 log, got", len(logs)) t.Error("expected 4 log, got", len(logs))
} }
filter = New(backend, 900, 999, []common.Address{addr}, [][]common.Hash{{hash3}}) filter = NewRangeFilter(backend, 900, 999, []common.Address{addr}, [][]common.Hash{{hash3}})
logs, _ = filter.Logs(context.Background()) logs, _ = filter.Logs(context.Background())
if len(logs) != 1 { if len(logs) != 1 {
t.Error("expected 1 log, got", len(logs)) t.Error("expected 1 log, got", len(logs))
@ -191,7 +191,7 @@ func TestFilters(t *testing.T) {
t.Errorf("expected log[0].Topics[0] to be %x, got %x", hash3, logs[0].Topics[0]) t.Errorf("expected log[0].Topics[0] to be %x, got %x", hash3, logs[0].Topics[0])
} }
filter = New(backend, 990, -1, []common.Address{addr}, [][]common.Hash{{hash3}}) filter = NewRangeFilter(backend, 990, -1, []common.Address{addr}, [][]common.Hash{{hash3}})
logs, _ = filter.Logs(context.Background()) logs, _ = filter.Logs(context.Background())
if len(logs) != 1 { if len(logs) != 1 {
t.Error("expected 1 log, got", len(logs)) t.Error("expected 1 log, got", len(logs))
@ -200,7 +200,7 @@ func TestFilters(t *testing.T) {
t.Errorf("expected log[0].Topics[0] to be %x, got %x", hash3, logs[0].Topics[0]) t.Errorf("expected log[0].Topics[0] to be %x, got %x", hash3, logs[0].Topics[0])
} }
filter = New(backend, 1, 10, nil, [][]common.Hash{{hash1, hash2}}) filter = NewRangeFilter(backend, 1, 10, nil, [][]common.Hash{{hash1, hash2}})
logs, _ = filter.Logs(context.Background()) logs, _ = filter.Logs(context.Background())
if len(logs) != 2 { if len(logs) != 2 {
@ -208,7 +208,7 @@ func TestFilters(t *testing.T) {
} }
failHash := common.BytesToHash([]byte("fail")) failHash := common.BytesToHash([]byte("fail"))
filter = New(backend, 0, -1, nil, [][]common.Hash{{failHash}}) filter = NewRangeFilter(backend, 0, -1, nil, [][]common.Hash{{failHash}})
logs, _ = filter.Logs(context.Background()) logs, _ = filter.Logs(context.Background())
if len(logs) != 0 { if len(logs) != 0 {
@ -216,14 +216,14 @@ func TestFilters(t *testing.T) {
} }
failAddr := common.BytesToAddress([]byte("failmenow")) failAddr := common.BytesToAddress([]byte("failmenow"))
filter = New(backend, 0, -1, []common.Address{failAddr}, nil) filter = NewRangeFilter(backend, 0, -1, []common.Address{failAddr}, nil)
logs, _ = filter.Logs(context.Background()) logs, _ = filter.Logs(context.Background())
if len(logs) != 0 { if len(logs) != 0 {
t.Error("expected 0 log, got", len(logs)) t.Error("expected 0 log, got", len(logs))
} }
filter = New(backend, 0, -1, nil, [][]common.Hash{{failHash}, {hash1}}) filter = NewRangeFilter(backend, 0, -1, nil, [][]common.Hash{{failHash}, {hash1}})
logs, _ = filter.Logs(context.Background()) logs, _ = filter.Logs(context.Background())
if len(logs) != 0 { if len(logs) != 0 {

View File

@ -60,10 +60,13 @@ func (b *LesApiBackend) HeaderByNumber(ctx context.Context, blockNr rpc.BlockNum
if blockNr == rpc.LatestBlockNumber || blockNr == rpc.PendingBlockNumber { if blockNr == rpc.LatestBlockNumber || blockNr == rpc.PendingBlockNumber {
return b.eth.blockchain.CurrentHeader(), nil return b.eth.blockchain.CurrentHeader(), nil
} }
return b.eth.blockchain.GetHeaderByNumberOdr(ctx, uint64(blockNr)) return b.eth.blockchain.GetHeaderByNumberOdr(ctx, uint64(blockNr))
} }
func (b *LesApiBackend) HeaderByHash(ctx context.Context, hash common.Hash) (*types.Header, error) {
return b.eth.blockchain.GetHeaderByHash(hash), nil
}
func (b *LesApiBackend) BlockByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*types.Block, error) { func (b *LesApiBackend) BlockByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*types.Block, error) {
header, err := b.HeaderByNumber(ctx, blockNr) header, err := b.HeaderByNumber(ctx, blockNr)
if header == nil || err != nil { if header == nil || err != nil {