diff options
Diffstat (limited to 'eth/filters')
-rw-r--r-- | eth/filters/api.go | 18 | ||||
-rw-r--r-- | eth/filters/filter.go | 77 | ||||
-rw-r--r-- | eth/filters/filter_system_test.go | 33 | ||||
-rw-r--r-- | eth/filters/filter_test.go | 36 |
4 files changed, 108 insertions, 56 deletions
diff --git a/eth/filters/api.go b/eth/filters/api.go index fa4bef283..834513262 100644 --- a/eth/filters/api.go +++ b/eth/filters/api.go @@ -52,6 +52,8 @@ type filter struct { // PublicFilterAPI offers support to create and manage filters. This will allow external clients to retrieve various // information related to the Ethereum protocol such als blocks, transactions and logs. type PublicFilterAPI struct { + backend Backend + useMipMap bool mux *event.TypeMux quit chan struct{} chainDb ethdb.Database @@ -316,7 +318,7 @@ func (api *PublicFilterAPI) NewFilter(crit FilterCriteria) rpc.ID { // GetLogs returns logs matching the given argument that are stored within the state. // // https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getlogs -func (api *PublicFilterAPI) GetLogs(crit FilterCriteria) []Log { +func (api *PublicFilterAPI) GetLogs(ctx context.Context, crit FilterCriteria) ([]Log, error) { if crit.FromBlock == nil { crit.FromBlock = big.NewInt(rpc.LatestBlockNumber.Int64()) } @@ -324,13 +326,14 @@ func (api *PublicFilterAPI) GetLogs(crit FilterCriteria) []Log { crit.ToBlock = big.NewInt(rpc.LatestBlockNumber.Int64()) } - filter := New(api.chainDb) + filter := New(api.backend, api.useMipMap) filter.SetBeginBlock(crit.FromBlock.Int64()) filter.SetEndBlock(crit.ToBlock.Int64()) filter.SetAddresses(crit.Addresses) filter.SetTopics(crit.Topics) - return returnLogs(filter.Find()) + logs, err := filter.Find(ctx) + return returnLogs(logs), err } // UninstallFilter removes the filter with the given filter id. @@ -354,22 +357,23 @@ func (api *PublicFilterAPI) UninstallFilter(id rpc.ID) bool { // If the filter could not be found an empty array of logs is returned. // // https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getfilterlogs -func (api *PublicFilterAPI) GetFilterLogs(id rpc.ID) []Log { +func (api *PublicFilterAPI) GetFilterLogs(ctx context.Context, id rpc.ID) ([]Log, error) { api.filtersMu.Lock() f, found := api.filters[id] api.filtersMu.Unlock() if !found || f.typ != LogsSubscription { - return []Log{} + return []Log{}, nil } - filter := New(api.chainDb) + filter := New(api.backend, api.useMipMap) filter.SetBeginBlock(f.crit.FromBlock.Int64()) filter.SetEndBlock(f.crit.ToBlock.Int64()) filter.SetAddresses(f.crit.Addresses) filter.SetTopics(f.crit.Topics) - return returnLogs(filter.Find()) + logs, err := filter.Find(ctx) + return returnLogs(logs), err } // GetFilterChanges returns the logs for the filter with the given id since diff --git a/eth/filters/filter.go b/eth/filters/filter.go index d181d0892..4004af300 100644 --- a/eth/filters/filter.go +++ b/eth/filters/filter.go @@ -24,10 +24,23 @@ import ( "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/rpc" + "golang.org/x/net/context" ) +type Backend interface { + ChainDb() ethdb.Database + EventMux() *event.TypeMux + HeaderByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*types.Header, error) + GetReceipts(ctx context.Context, blockHash common.Hash) (types.Receipts, error) +} + // Filter can be used to retrieve and filter logs type Filter struct { + backend Backend + useMipMap bool + created time.Time db ethdb.Database @@ -38,8 +51,12 @@ type Filter struct { // New creates a new filter which uses a bloom filter on blocks to figure out whether // a particular block is interesting or not. -func New(db ethdb.Database) *Filter { - return &Filter{db: db} +func New(backend Backend, useMipMap bool) *Filter { + return &Filter{ + backend: backend, + useMipMap: useMipMap, + db: backend.ChainDb(), + } } // SetBeginBlock sets the earliest block for filtering. @@ -66,30 +83,29 @@ func (f *Filter) SetTopics(topics [][]common.Hash) { } // Run filters logs with the current parameters set -func (f *Filter) Find() []Log { - latestHash := core.GetHeadBlockHash(f.db) - latestBlock := core.GetBlock(f.db, latestHash, core.GetBlockNumber(f.db, latestHash)) - if latestBlock == nil { - return []Log{} +func (f *Filter) Find(ctx context.Context) ([]Log, error) { + head, _ := f.backend.HeaderByNumber(ctx, rpc.LatestBlockNumber) + if head == nil { + return nil, nil } + headBlockNumber := head.Number.Uint64() var beginBlockNo uint64 = uint64(f.begin) if f.begin == -1 { - beginBlockNo = latestBlock.NumberU64() + beginBlockNo = headBlockNumber } - - endBlockNo := uint64(f.end) + var endBlockNo uint64 = uint64(f.end) if f.end == -1 { - endBlockNo = latestBlock.NumberU64() + endBlockNo = headBlockNumber } // if no addresses are present we can't make use of fast search which // uses the mipmap bloom filters to check for fast inclusion and uses // higher range probability in order to ensure at least a false positive - if len(f.addresses) == 0 { - return f.getLogs(beginBlockNo, endBlockNo) + if !f.useMipMap || len(f.addresses) == 0 { + return f.getLogs(ctx, beginBlockNo, endBlockNo) } - return f.mipFind(beginBlockNo, endBlockNo, 0) + return f.mipFind(beginBlockNo, endBlockNo, 0), nil } func (f *Filter) mipFind(start, end uint64, depth int) (logs []Log) { @@ -107,7 +123,8 @@ func (f *Filter) mipFind(start, end uint64, depth int) (logs []Log) { start := uint64(math.Max(float64(num), float64(start))) end := uint64(math.Min(float64(num+level-1), float64(end))) if depth+1 == len(core.MIPMapLevels) { - logs = append(logs, f.getLogs(start, end)...) + l, _ := f.getLogs(context.Background(), start, end) + logs = append(logs, l...) } else { logs = append(logs, f.mipFind(start, end, depth+1)...) } @@ -122,28 +139,22 @@ func (f *Filter) mipFind(start, end uint64, depth int) (logs []Log) { return logs } -func (f *Filter) getLogs(start, end uint64) (logs []Log) { - var block *types.Block - +func (f *Filter) getLogs(ctx context.Context, start, end uint64) (logs []Log, err error) { for i := start; i <= end; i++ { - hash := core.GetCanonicalHash(f.db, i) - if hash != (common.Hash{}) { - block = core.GetBlock(f.db, hash, i) - } else { // block not found - return logs - } - if block == nil { // block not found/written - return logs + header, err := f.backend.HeaderByNumber(ctx, rpc.BlockNumber(i)) + if header == nil || err != nil { + return logs, err } // Use bloom filtering to see if this block is interesting given the // current parameters - if f.bloomFilter(block) { + if f.bloomFilter(header.Bloom) { // Get the logs of the block - var ( - receipts = core.GetBlockReceipts(f.db, block.Hash(), i) - unfiltered []Log - ) + receipts, err := f.backend.GetReceipts(ctx, header.Hash()) + if err != nil { + return nil, err + } + var unfiltered []Log for _, receipt := range receipts { rl := make([]Log, len(receipt.Logs)) for i, l := range receipt.Logs { @@ -155,7 +166,7 @@ func (f *Filter) getLogs(start, end uint64) (logs []Log) { } } - return logs + return logs, nil } func includes(addresses []common.Address, a common.Address) bool { @@ -229,7 +240,7 @@ func bloomFilter(bloom types.Bloom, addresses []common.Address, topics [][]commo for _, sub := range topics { var included bool for _, topic := range sub { - if (topic == common.Hash{}) || types.BloomLookup(block.Bloom(), topic) { + if (topic == common.Hash{}) || types.BloomLookup(bloom, topic) { included = true break } diff --git a/eth/filters/filter_system_test.go b/eth/filters/filter_system_test.go index 1bd4d502d..48d6811c0 100644 --- a/eth/filters/filter_system_test.go +++ b/eth/filters/filter_system_test.go @@ -22,6 +22,8 @@ import ( "testing" "time" + "golang.org/x/net/context" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" @@ -38,6 +40,37 @@ var ( api = NewPublicFilterAPI(backend, false) ) +type testBackend struct { + mux *event.TypeMux + db ethdb.Database +} + +func (b *testBackend) ChainDb() ethdb.Database { + return b.db +} + +func (b *testBackend) EventMux() *event.TypeMux { + return b.mux +} + +func (b *testBackend) HeaderByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*types.Header, error) { + var hash common.Hash + var num uint64 + if blockNr == rpc.LatestBlockNumber { + hash = core.GetHeadBlockHash(b.db) + num = core.GetBlockNumber(b.db, hash) + } else { + num = uint64(blockNr) + hash = core.GetCanonicalHash(b.db, num) + } + return core.GetHeader(b.db, hash, num), nil +} + +func (b *testBackend) GetReceipts(ctx context.Context, blockHash common.Hash) (types.Receipts, error) { + num := core.GetBlockNumber(b.db, blockHash) + return core.GetBlockReceipts(b.db, blockHash, num), nil +} + // TestBlockSubscription tests if a block subscription returns block hashes for posted chain events. // It creates multiple subscriptions: // - one at the start and should receive all posted chain events and a second (blockHashes) diff --git a/eth/filters/filter_test.go b/eth/filters/filter_test.go index 7b714f5d5..e0b24046c 100644 --- a/eth/filters/filter_test.go +++ b/eth/filters/filter_test.go @@ -22,6 +22,8 @@ import ( "os" "testing" + "golang.org/x/net/context" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" @@ -48,6 +50,7 @@ func BenchmarkMipmaps(b *testing.B) { var ( db, _ = ethdb.NewLDBDatabase(dir, 0, 0) + backend = &testBackend{mux, db} key1, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") addr1 = crypto.PubkeyToAddress(key1.PublicKey) addr2 = common.BytesToAddress([]byte("jeff")) @@ -100,13 +103,13 @@ func BenchmarkMipmaps(b *testing.B) { } b.ResetTimer() - filter := New(db) + filter := New(backend, true) filter.SetAddresses([]common.Address{addr1, addr2, addr3, addr4}) filter.SetBeginBlock(0) filter.SetEndBlock(-1) for i := 0; i < b.N; i++ { - logs := filter.Find() + logs, _ := filter.Find(context.Background()) if len(logs) != 4 { b.Fatal("expected 4 log, got", len(logs)) } @@ -122,6 +125,7 @@ func TestFilters(t *testing.T) { var ( db, _ = ethdb.NewLDBDatabase(dir, 0, 0) + backend = &testBackend{mux, db} key1, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") addr = crypto.PubkeyToAddress(key1.PublicKey) @@ -201,23 +205,23 @@ func TestFilters(t *testing.T) { } } - filter := New(db) + filter := New(backend, true) filter.SetAddresses([]common.Address{addr}) filter.SetTopics([][]common.Hash{[]common.Hash{hash1, hash2, hash3, hash4}}) filter.SetBeginBlock(0) filter.SetEndBlock(-1) - logs := filter.Find() + logs, _ := filter.Find(context.Background()) if len(logs) != 4 { t.Error("expected 4 log, got", len(logs)) } - filter = New(db) + filter = New(backend, true) filter.SetAddresses([]common.Address{addr}) filter.SetTopics([][]common.Hash{[]common.Hash{hash3}}) filter.SetBeginBlock(900) filter.SetEndBlock(999) - logs = filter.Find() + logs, _ = filter.Find(context.Background()) if len(logs) != 1 { t.Error("expected 1 log, got", len(logs)) } @@ -225,12 +229,12 @@ func TestFilters(t *testing.T) { t.Errorf("expected log[0].Topics[0] to be %x, got %x", hash3, logs[0].Topics[0]) } - filter = New(db) + filter = New(backend, true) filter.SetAddresses([]common.Address{addr}) filter.SetTopics([][]common.Hash{[]common.Hash{hash3}}) filter.SetBeginBlock(990) filter.SetEndBlock(-1) - logs = filter.Find() + logs, _ = filter.Find(context.Background()) if len(logs) != 1 { t.Error("expected 1 log, got", len(logs)) } @@ -238,44 +242,44 @@ func TestFilters(t *testing.T) { t.Errorf("expected log[0].Topics[0] to be %x, got %x", hash3, logs[0].Topics[0]) } - filter = New(db) + filter = New(backend, true) filter.SetTopics([][]common.Hash{[]common.Hash{hash1, hash2}}) filter.SetBeginBlock(1) filter.SetEndBlock(10) - logs = filter.Find() + logs, _ = filter.Find(context.Background()) if len(logs) != 2 { t.Error("expected 2 log, got", len(logs)) } failHash := common.BytesToHash([]byte("fail")) - filter = New(db) + filter = New(backend, true) filter.SetTopics([][]common.Hash{[]common.Hash{failHash}}) filter.SetBeginBlock(0) filter.SetEndBlock(-1) - logs = filter.Find() + logs, _ = filter.Find(context.Background()) if len(logs) != 0 { t.Error("expected 0 log, got", len(logs)) } failAddr := common.BytesToAddress([]byte("failmenow")) - filter = New(db) + filter = New(backend, true) filter.SetAddresses([]common.Address{failAddr}) filter.SetBeginBlock(0) filter.SetEndBlock(-1) - logs = filter.Find() + logs, _ = filter.Find(context.Background()) if len(logs) != 0 { t.Error("expected 0 log, got", len(logs)) } - filter = New(db) + filter = New(backend, true) filter.SetTopics([][]common.Hash{[]common.Hash{failHash}, []common.Hash{hash1}}) filter.SetBeginBlock(0) filter.SetEndBlock(-1) - logs = filter.Find() + logs, _ = filter.Find(context.Background()) if len(logs) != 0 { t.Error("expected 0 log, got", len(logs)) } |