From f99cc5a84070950f741699fecab67f07b921ba8e Mon Sep 17 00:00:00 2001 From: Bas van Kervel Date: Wed, 22 Jun 2016 09:34:33 +0200 Subject: [PATCH] eth/filter: refactor log filter system --- eth/filters/api.go | 596 ++++++--------------------- eth/filters/api_test.go | 19 +- eth/filters/filter.go | 109 +++-- eth/filters/filter_system.go | 642 ++++++++++++++++++++++++------ eth/filters/filter_system_test.go | 501 +++++++++++++++++++---- 5 files changed, 1138 insertions(+), 729 deletions(-) diff --git a/eth/filters/api.go b/eth/filters/api.go index 393019f8b0..dc4ec4fed5 100644 --- a/eth/filters/api.go +++ b/eth/filters/api.go @@ -17,283 +17,191 @@ package filters import ( - "crypto/rand" "encoding/hex" "encoding/json" "errors" "fmt" "sync" - "time" + + "golang.org/x/net/context" "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/core/types" - "github.com/ethereum/go-ethereum/core/vm" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/rpc" - - "golang.org/x/net/context" ) -var ( - filterTickerTime = 5 * time.Minute -) - -// byte will be inferred -const ( - unknownFilterTy = iota - blockFilterTy - transactionFilterTy - logFilterTy -) - -// 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. +// PublicFilterAPI allows clients to retrieve or get notified of data such as blocks, +// transactions and logs. type PublicFilterAPI struct { - mux *event.TypeMux + db ethdb.Database + m *Manager - quit chan struct{} - chainDb ethdb.Database - - filterManager *FilterSystem - - filterMapMu sync.RWMutex - filterMapping map[string]int // maps between filter internal filter identifiers and external filter identifiers - - logMu sync.RWMutex - logQueue map[int]*logQueue - - blockMu sync.RWMutex - blockQueue map[int]*hashQueue - - transactionMu sync.RWMutex - transactionQueue map[int]*hashQueue - - transactMu sync.Mutex + subscriptionsMu sync.Mutex + subscriptions map[string]FilterID // mapping between subscriptions and the underlying filters } // NewPublicFilterAPI returns a new PublicFilterAPI instance. +// It uses the given database to retrieve stored logs and the mux to listen for events. func NewPublicFilterAPI(chainDb ethdb.Database, mux *event.TypeMux) *PublicFilterAPI { - svc := &PublicFilterAPI{ - mux: mux, - chainDb: chainDb, - filterManager: NewFilterSystem(mux), - filterMapping: make(map[string]int), - logQueue: make(map[int]*logQueue), - blockQueue: make(map[int]*hashQueue), - transactionQueue: make(map[int]*hashQueue), + return &PublicFilterAPI{ + db: chainDb, + m: NewManager(mux), + subscriptions: make(map[string]FilterID), } - go svc.start() - return svc } -// Stop quits the work loop. -func (s *PublicFilterAPI) Stop() { - close(s.quit) +// NewBlockFilter creates a filter that returns hashes for new blocks. +// To check if the state has changed, call GetFilterChanges. +// +// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_newblockfilter +func (api *PublicFilterAPI) NewBlockFilter() (FilterID, error) { + return api.m.NewBlockFilter() } -// start the work loop, wait and process events. -func (s *PublicFilterAPI) start() { - timer := time.NewTicker(2 * time.Second) - defer timer.Stop() -done: - for { - select { - case <-timer.C: - s.logMu.Lock() - for id, filter := range s.logQueue { - if time.Since(filter.timeout) > filterTickerTime { - s.filterManager.Remove(id) - delete(s.logQueue, id) - } - } - s.logMu.Unlock() +// NewPendingTransactionFilter creates a filter, to notify when new pending transactions arrive. +// To check if the state has changed, call GetFilterChanges. +// +// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_newpendingtransactionfilter +func (api *PublicFilterAPI) NewPendingTransactionFilter() (FilterID, error) { + return api.m.NewPendingTransactionFilter() +} - s.blockMu.Lock() - for id, filter := range s.blockQueue { - if time.Since(filter.timeout) > filterTickerTime { - s.filterManager.Remove(id) - delete(s.blockQueue, id) - } - } - s.blockMu.Unlock() - - s.transactionMu.Lock() - for id, filter := range s.transactionQueue { - if time.Since(filter.timeout) > filterTickerTime { - s.filterManager.Remove(id) - delete(s.transactionQueue, id) - } - } - s.transactionMu.Unlock() - case <-s.quit: - break done - } +// GetFilterChanges returns new logs for the given filter since last time is was called. +// This can be used for polling. +// +// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getfilterchanges +func (api *PublicFilterAPI) GetFilterChanges(id FilterID) (interface{}, error) { + // filter might be deleted between retrieving filter type and calling the XXXFilterChanges + // this is not an issue, XXXFilterChanges will return an errFilterNotFound error. + switch api.m.FilterType(id) { + case BlockFilter: + hashes, err := api.m.GetBlockFilterChanges(id) + return toRPCHashes(hashes), err + case PendingTxFilter: + hashes, err := api.m.GetPendingTxFilterChanges(id) + return toRPCHashes(hashes), err + case LogFilter: + logs, err := api.m.GetLogFilterChanges(id) + return toRPCLogs(logs), err } + return nil, errFilterNotFound } -// NewBlockFilter create a new filter that returns blocks that are included into the canonical chain. -func (s *PublicFilterAPI) NewBlockFilter() (string, error) { - externalId, err := newFilterId() +// NewFilter creates a filter that can be used to fetch new logs. +// Note: as specification dictates, it can be used to fetch logs when the state changes. +// Older already created logs cannot be fetched using this method, use GetLogs. +// Therefore the fromBlock and toBlock parameters are ignored. +// +// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_newfilter +func (api *PublicFilterAPI) NewFilter(crit FilterCriteria) (FilterID, error) { + return api.m.NewLogFilter(crit, nil) +} + +// GetLogs returns logs matching the given criteria. +// +// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getlogs +func (api *PublicFilterAPI) GetLogs(crit FilterCriteria) []Log { + filter := New(api.db) + filter.SetBeginBlock(crit.FromBlock.Int64()) + filter.SetEndBlock(crit.ToBlock.Int64()) + filter.SetAddresses(crit.Addresses) + filter.SetTopics(crit.Topics) + + return toRPCLogs(filter.Find()) +} + +// GetFilterLogs return all logs that match for the given (log) filter. +// +// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getfilterlogs +func (api *PublicFilterAPI) GetFilterLogs(id FilterID) ([]Log, error) { + crit, err := api.m.GetLogFilterCriteria(id) if err != nil { - return "", err + return nil, err } - - s.blockMu.Lock() - filter := New(s.chainDb) - id, err := s.filterManager.Add(filter, ChainFilter) - if err != nil { - return "", err - } - - s.blockQueue[id] = &hashQueue{timeout: time.Now()} - - filter.BlockCallback = func(block *types.Block, logs vm.Logs) { - s.blockMu.Lock() - defer s.blockMu.Unlock() - - if queue := s.blockQueue[id]; queue != nil { - queue.add(block.Hash()) - } - } - - defer s.blockMu.Unlock() - - s.filterMapMu.Lock() - s.filterMapping[externalId] = id - s.filterMapMu.Unlock() - - return externalId, nil + return api.GetLogs(crit), nil } -// NewPendingTransactionFilter creates a filter that returns new pending transactions. -func (s *PublicFilterAPI) NewPendingTransactionFilter() (string, error) { - externalId, err := newFilterId() - if err != nil { - return "", err - } - - s.transactionMu.Lock() - defer s.transactionMu.Unlock() - - filter := New(s.chainDb) - id, err := s.filterManager.Add(filter, PendingTxFilter) - if err != nil { - return "", err - } - - s.transactionQueue[id] = &hashQueue{timeout: time.Now()} - - filter.TransactionCallback = func(tx *types.Transaction) { - s.transactionMu.Lock() - defer s.transactionMu.Unlock() - - if queue := s.transactionQueue[id]; queue != nil { - queue.add(tx.Hash()) - } - } - - s.filterMapMu.Lock() - s.filterMapping[externalId] = id - s.filterMapMu.Unlock() - - return externalId, nil -} - -// newLogFilter creates a new log filter. -func (s *PublicFilterAPI) newLogFilter(earliest, latest int64, addresses []common.Address, topics [][]common.Hash, callback func(log *vm.Log, removed bool)) (int, error) { - s.logMu.Lock() - defer s.logMu.Unlock() - - filter := New(s.chainDb) - id, err := s.filterManager.Add(filter, LogFilter) - if err != nil { - return 0, err - } - - s.logQueue[id] = &logQueue{timeout: time.Now()} - - filter.SetBeginBlock(earliest) - filter.SetEndBlock(latest) - filter.SetAddresses(addresses) - filter.SetTopics(topics) - filter.LogCallback = func(log *vm.Log, removed bool) { - if callback != nil { - callback(log, removed) - } else { - s.logMu.Lock() - defer s.logMu.Unlock() - if queue := s.logQueue[id]; queue != nil { - queue.add(vmlog{log, removed}) - } - } - } - - return id, nil -} - -// Logs creates a subscription that fires for all new log that match the given filter criteria. -func (s *PublicFilterAPI) Logs(ctx context.Context, args NewFilterArgs) (rpc.Subscription, error) { +// Logs creates a subscription that fires each time a log is created that matches the given criteria. +// Note, in case of chain reorganisations logs with the Removed true property can be returned indicating +// a previously sent log is reverted. +func (api *PublicFilterAPI) Logs(ctx context.Context, crit FilterCriteria) (subscription rpc.Subscription, err error) { notifier, supported := rpc.NotifierFromContext(ctx) if !supported { return nil, rpc.ErrNotificationsUnsupported } - var ( - externalId string - subscription rpc.Subscription - err error - ) - - if externalId, err = newFilterId(); err != nil { - return nil, err - } - - // uninstall filter when subscription is unsubscribed/cancelled - if subscription, err = notifier.NewSubscription(func(string) { - s.UninstallFilter(externalId) - }); err != nil { - return nil, err - } - - notifySubscriber := func(log *vm.Log, removed bool) { - rpcLog := toRPCLogs(vm.Logs{log}, removed) - if err := subscription.Notify(rpcLog); err != nil { - subscription.Cancel() + onUnsubscribe := func(subid string) { + api.subscriptionsMu.Lock() + defer api.subscriptionsMu.Unlock() + if fid, found := api.subscriptions[subid]; found { + delete(api.subscriptions, subid) + api.m.Uninstall(fid) // uninstall associated filter } } - // from and to block number are not used since subscriptions don't allow you to travel to "time" - var id int - if len(args.Addresses) > 0 { - id, err = s.newLogFilter(-1, -1, args.Addresses, args.Topics, notifySubscriber) - } else { - id, err = s.newLogFilter(-1, -1, nil, args.Topics, notifySubscriber) - } - + subscription, err = notifier.NewSubscription(onUnsubscribe) if err != nil { - subscription.Cancel() return nil, err } - s.filterMapMu.Lock() - s.filterMapping[externalId] = id - s.filterMapMu.Unlock() + send := func(id FilterID, logs []Log) { + rpcLogs := toRPCLogs(logs) + for _, l := range rpcLogs { + subscription.Notify(l) + } + } + + fid, err := api.m.NewLogFilterWithNoTimeout(crit, send) + if err != nil { + return nil, err + } + + api.subscriptionsMu.Lock() + api.subscriptions[subscription.ID()] = fid + api.subscriptionsMu.Unlock() return subscription, err } -// NewFilterArgs represents a request to create a new filter. -type NewFilterArgs struct { +// UninstallFilter deletes a filter by its id. If successful true is returned, +// otherwise false. +// +// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_uninstallfilter +func (api *PublicFilterAPI) UninstallFilter(id FilterID) bool { + return api.m.Uninstall(id) == nil +} + +// toRPCLogs is a helper that will return an empty slice of logs when nil is given. +// Otherwise the given logs are returned. This is necessary for the RPC interface. +func toRPCLogs(logs []Log) []Log { + if logs == nil { + return []Log{} + } + return logs +} + +// toRPCHashes is a helper that will return an empty hash array case the given hash +// array is nil, otherwise is will return the given hashes. The RPC interfaces defines +// that always an array is returned. +func toRPCHashes(hashes []common.Hash) []common.Hash { + if hashes == nil { + return []common.Hash{} + } + return hashes +} + +// FilterCriteria represents a request to create a new filter. +type FilterCriteria struct { FromBlock rpc.BlockNumber ToBlock rpc.BlockNumber Addresses []common.Address Topics [][]common.Hash } -// UnmarshalJSON sets *args fields with given data. -func (args *NewFilterArgs) UnmarshalJSON(data []byte) error { +// UnmarshalJSON sets *args fields with filter criteria JSON encoded in the given data. +func (args *FilterCriteria) UnmarshalJSON(data []byte) error { type input struct { From *rpc.BlockNumber `json:"fromBlock"` ToBlock *rpc.BlockNumber `json:"toBlock"` @@ -405,245 +313,3 @@ func (args *NewFilterArgs) UnmarshalJSON(data []byte) error { return nil } - -// NewFilter creates a new filter and returns the filter id. It can be uses to retrieve logs. -func (s *PublicFilterAPI) NewFilter(args NewFilterArgs) (string, error) { - externalId, err := newFilterId() - if err != nil { - return "", err - } - - var id int - if len(args.Addresses) > 0 { - id, err = s.newLogFilter(args.FromBlock.Int64(), args.ToBlock.Int64(), args.Addresses, args.Topics, nil) - } else { - id, err = s.newLogFilter(args.FromBlock.Int64(), args.ToBlock.Int64(), nil, args.Topics, nil) - } - if err != nil { - return "", err - } - - s.filterMapMu.Lock() - s.filterMapping[externalId] = id - s.filterMapMu.Unlock() - - return externalId, nil -} - -// GetLogs returns the logs matching the given argument. -func (s *PublicFilterAPI) GetLogs(args NewFilterArgs) []vmlog { - filter := New(s.chainDb) - filter.SetBeginBlock(args.FromBlock.Int64()) - filter.SetEndBlock(args.ToBlock.Int64()) - filter.SetAddresses(args.Addresses) - filter.SetTopics(args.Topics) - - return toRPCLogs(filter.Find(), false) -} - -// UninstallFilter removes the filter with the given filter id. -func (s *PublicFilterAPI) UninstallFilter(filterId string) bool { - s.filterMapMu.Lock() - defer s.filterMapMu.Unlock() - - id, ok := s.filterMapping[filterId] - if !ok { - return false - } - - defer s.filterManager.Remove(id) - delete(s.filterMapping, filterId) - - if _, ok := s.logQueue[id]; ok { - s.logMu.Lock() - defer s.logMu.Unlock() - delete(s.logQueue, id) - return true - } - if _, ok := s.blockQueue[id]; ok { - s.blockMu.Lock() - defer s.blockMu.Unlock() - delete(s.blockQueue, id) - return true - } - if _, ok := s.transactionQueue[id]; ok { - s.transactionMu.Lock() - defer s.transactionMu.Unlock() - delete(s.transactionQueue, id) - return true - } - - return false -} - -// getFilterType is a helper utility that determine the type of filter for the given filter id. -func (s *PublicFilterAPI) getFilterType(id int) byte { - if _, ok := s.blockQueue[id]; ok { - return blockFilterTy - } else if _, ok := s.transactionQueue[id]; ok { - return transactionFilterTy - } else if _, ok := s.logQueue[id]; ok { - return logFilterTy - } - - return unknownFilterTy -} - -// blockFilterChanged returns a collection of block hashes for the block filter with the given id. -func (s *PublicFilterAPI) blockFilterChanged(id int) []common.Hash { - s.blockMu.Lock() - defer s.blockMu.Unlock() - - if s.blockQueue[id] != nil { - return s.blockQueue[id].get() - } - return nil -} - -// transactionFilterChanged returns a collection of transaction hashes for the pending -// transaction filter with the given id. -func (s *PublicFilterAPI) transactionFilterChanged(id int) []common.Hash { - s.blockMu.Lock() - defer s.blockMu.Unlock() - - if s.transactionQueue[id] != nil { - return s.transactionQueue[id].get() - } - return nil -} - -// logFilterChanged returns a collection of logs for the log filter with the given id. -func (s *PublicFilterAPI) logFilterChanged(id int) []vmlog { - s.logMu.Lock() - defer s.logMu.Unlock() - - if s.logQueue[id] != nil { - return s.logQueue[id].get() - } - return nil -} - -// GetFilterLogs returns the logs for the filter with the given id. -func (s *PublicFilterAPI) GetFilterLogs(filterId string) []vmlog { - id, ok := s.filterMapping[filterId] - if !ok { - return toRPCLogs(nil, false) - } - - if filter := s.filterManager.Get(id); filter != nil { - return toRPCLogs(filter.Find(), false) - } - - return toRPCLogs(nil, false) -} - -// GetFilterChanges returns the logs for the filter with the given id since last time is was called. -// This can be used for polling. -func (s *PublicFilterAPI) GetFilterChanges(filterId string) interface{} { - s.filterMapMu.Lock() - id, ok := s.filterMapping[filterId] - s.filterMapMu.Unlock() - - if !ok { // filter not found - return []interface{}{} - } - - switch s.getFilterType(id) { - case blockFilterTy: - return returnHashes(s.blockFilterChanged(id)) - case transactionFilterTy: - return returnHashes(s.transactionFilterChanged(id)) - case logFilterTy: - return s.logFilterChanged(id) - } - - return []interface{}{} -} - -type vmlog struct { - *vm.Log - Removed bool `json:"removed"` -} - -type logQueue struct { - mu sync.Mutex - - logs []vmlog - timeout time.Time - id int -} - -func (l *logQueue) add(logs ...vmlog) { - l.mu.Lock() - defer l.mu.Unlock() - - l.logs = append(l.logs, logs...) -} - -func (l *logQueue) get() []vmlog { - l.mu.Lock() - defer l.mu.Unlock() - - l.timeout = time.Now() - tmp := l.logs - l.logs = nil - return tmp -} - -type hashQueue struct { - mu sync.Mutex - - hashes []common.Hash - timeout time.Time - id int -} - -func (l *hashQueue) add(hashes ...common.Hash) { - l.mu.Lock() - defer l.mu.Unlock() - - l.hashes = append(l.hashes, hashes...) -} - -func (l *hashQueue) get() []common.Hash { - l.mu.Lock() - defer l.mu.Unlock() - - l.timeout = time.Now() - tmp := l.hashes - l.hashes = nil - return tmp -} - -// newFilterId generates a new random filter identifier that can be exposed to the outer world. By publishing random -// identifiers it is not feasible for DApp's to guess filter id's for other DApp's and uninstall or poll for them -// causing the affected DApp to miss data. -func newFilterId() (string, error) { - var subid [16]byte - n, _ := rand.Read(subid[:]) - if n != 16 { - return "", errors.New("Unable to generate filter id") - } - return "0x" + hex.EncodeToString(subid[:]), nil -} - -// toRPCLogs is a helper that will convert a vm.Logs array to an structure which -// can hold additional information about the logs such as whether it was deleted. -// Additionally when nil is given it will by default instead create an empty slice -// instead. This is required by the RPC specification. -func toRPCLogs(logs vm.Logs, removed bool) []vmlog { - convertedLogs := make([]vmlog, len(logs)) - for i, log := range logs { - convertedLogs[i] = vmlog{Log: log, Removed: removed} - } - return convertedLogs -} - -// returnHashes is a helper that will return an empty hash array case the given hash array is nil, otherwise is will -// return the given hashes. The RPC interfaces defines that always an array is returned. -func returnHashes(hashes []common.Hash) []common.Hash { - if hashes == nil { - return []common.Hash{} - } - return hashes -} diff --git a/eth/filters/api_test.go b/eth/filters/api_test.go index 9e8edc2413..81c7105bb6 100644 --- a/eth/filters/api_test.go +++ b/eth/filters/api_test.go @@ -14,7 +14,7 @@ // You should have received a copy of the GNU Lesser General Public License // along with the go-ethereum library. If not, see . -package filters_test +package filters import ( "encoding/json" @@ -22,7 +22,6 @@ import ( "testing" "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/eth/filters" "github.com/ethereum/go-ethereum/rpc" ) @@ -39,7 +38,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) { ) // default values - var test0 filters.NewFilterArgs + var test0 FilterCriteria if err := json.Unmarshal([]byte("{}"), &test0); err != nil { t.Fatal(err) } @@ -57,7 +56,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) { } // from, to block number - var test1 filters.NewFilterArgs + var test1 FilterCriteria vector := fmt.Sprintf(`{"fromBlock":"0x%x","toBlock":"0x%x"}`, fromBlock, toBlock) if err := json.Unmarshal([]byte(vector), &test1); err != nil { t.Fatal(err) @@ -70,7 +69,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) { } // single address - var test2 filters.NewFilterArgs + var test2 FilterCriteria vector = fmt.Sprintf(`{"address": "%s"}`, address0.Hex()) if err := json.Unmarshal([]byte(vector), &test2); err != nil { t.Fatal(err) @@ -83,7 +82,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) { } // multiple address - var test3 filters.NewFilterArgs + var test3 FilterCriteria vector = fmt.Sprintf(`{"address": ["%s", "%s"]}`, address0.Hex(), address1.Hex()) if err := json.Unmarshal([]byte(vector), &test3); err != nil { t.Fatal(err) @@ -99,7 +98,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) { } // single topic - var test4 filters.NewFilterArgs + var test4 FilterCriteria vector = fmt.Sprintf(`{"topics": ["%s"]}`, topic0.Hex()) if err := json.Unmarshal([]byte(vector), &test4); err != nil { t.Fatal(err) @@ -115,7 +114,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) { } // test multiple "AND" topics - var test5 filters.NewFilterArgs + var test5 FilterCriteria vector = fmt.Sprintf(`{"topics": ["%s", "%s"]}`, topic0.Hex(), topic1.Hex()) if err := json.Unmarshal([]byte(vector), &test5); err != nil { t.Fatal(err) @@ -137,7 +136,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) { } // test optional topic - var test6 filters.NewFilterArgs + var test6 FilterCriteria vector = fmt.Sprintf(`{"topics": ["%s", null, "%s"]}`, topic0.Hex(), topic2.Hex()) if err := json.Unmarshal([]byte(vector), &test6); err != nil { t.Fatal(err) @@ -165,7 +164,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) { } // test OR topics - var test7 filters.NewFilterArgs + var test7 FilterCriteria vector = fmt.Sprintf(`{"topics": [["%s", "%s"], null, ["%s", null]]}`, topic0.Hex(), topic1.Hex(), topic2.Hex()) if err := json.Unmarshal([]byte(vector), &test7); err != nil { t.Fatal(err) diff --git a/eth/filters/filter.go b/eth/filters/filter.go index 995b588fb2..821bc6736e 100644 --- a/eth/filters/filter.go +++ b/eth/filters/filter.go @@ -23,15 +23,10 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" - "github.com/ethereum/go-ethereum/core/vm" "github.com/ethereum/go-ethereum/ethdb" ) -type AccountChange struct { - Address, StateAddress []byte -} - -// Filtering interface +// Filter can be used to retrieve and filter logs type Filter struct { created time.Time @@ -39,67 +34,67 @@ type Filter struct { begin, end int64 addresses []common.Address topics [][]common.Hash - - BlockCallback func(*types.Block, vm.Logs) - TransactionCallback func(*types.Transaction) - LogCallback func(*vm.Log, bool) } -// Create a new filter which uses a bloom filter on blocks to figure out whether a particular block -// is interesting or not. +// 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} } -// Set the earliest and latest block for filtering. +// SetBeginBlock sets the earliest block for filtering. // -1 = latest block (i.e., the current block) // hash = particular hash from-to -func (self *Filter) SetBeginBlock(begin int64) { - self.begin = begin +func (f *Filter) SetBeginBlock(begin int64) { + f.begin = begin } -func (self *Filter) SetEndBlock(end int64) { - self.end = end +// SetEndBlock sets the latest block for filtering. +func (f *Filter) SetEndBlock(end int64) { + f.end = end } -func (self *Filter) SetAddresses(addr []common.Address) { - self.addresses = addr +// SetAddresses matches only logs that are generated from addresses that are included +// in the given addresses. +func (f *Filter) SetAddresses(addr []common.Address) { + f.addresses = addr } -func (self *Filter) SetTopics(topics [][]common.Hash) { - self.topics = topics +// SetTopics matches only logs that have topics matching the given topics. +func (f *Filter) SetTopics(topics [][]common.Hash) { + f.topics = topics } -// Run filters logs with the current parameters set -func (self *Filter) Find() vm.Logs { - latestHash := core.GetHeadBlockHash(self.db) - latestBlock := core.GetBlock(self.db, latestHash, core.GetBlockNumber(self.db, latestHash)) - var beginBlockNo uint64 = uint64(self.begin) - if self.begin == -1 { +// Find 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)) + beginBlockNo := uint64(f.begin) + if f.begin == -1 { beginBlockNo = latestBlock.NumberU64() } - var endBlockNo uint64 = uint64(self.end) - if self.end == -1 { + endBlockNo := uint64(f.end) + if f.end == -1 { endBlockNo = latestBlock.NumberU64() } // 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(self.addresses) == 0 { - return self.getLogs(beginBlockNo, endBlockNo) + if len(f.addresses) == 0 { + return f.getLogs(beginBlockNo, endBlockNo) } - return self.mipFind(beginBlockNo, endBlockNo, 0) + return f.mipFind(beginBlockNo, endBlockNo, 0) } -func (self *Filter) mipFind(start, end uint64, depth int) (logs vm.Logs) { +func (f *Filter) mipFind(start, end uint64, depth int) (logs []Log) { level := core.MIPMapLevels[depth] // normalise numerator so we can work in level specific batches and // work with the proper range checks for num := start / level * level; num <= end; num += level { // find addresses in bloom filters - bloom := core.GetMipmapBloom(self.db, num, level) - for _, addr := range self.addresses { + bloom := core.GetMipmapBloom(f.db, num, level) + for _, addr := range f.addresses { if bloom.TestBytes(addr[:]) { // range check normalised values and make sure that // we're resolving the correct range instead of the @@ -107,9 +102,9 @@ func (self *Filter) mipFind(start, end uint64, depth int) (logs vm.Logs) { 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, self.getLogs(start, end)...) + logs = append(logs, f.getLogs(start, end)...) } else { - logs = append(logs, self.mipFind(start, end, depth+1)...) + logs = append(logs, f.mipFind(start, end, depth+1)...) } // break so we don't check the same range for each // possible address. Checks on multiple addresses @@ -122,29 +117,33 @@ func (self *Filter) mipFind(start, end uint64, depth int) (logs vm.Logs) { return logs } -func (self *Filter) getLogs(start, end uint64) (logs vm.Logs) { +func (f *Filter) getLogs(start, end uint64) (logs []Log) { var block *types.Block for i := start; i <= end; i++ { - hash := core.GetCanonicalHash(self.db, i) + hash := core.GetCanonicalHash(f.db, i) if hash != (common.Hash{}) { - block = core.GetBlock(self.db, hash, i) + block = core.GetBlock(f.db, hash, i) } else { // block not found return logs } // Use bloom filtering to see if this block is interesting given the // current parameters - if self.bloomFilter(block) { + if f.bloomFilter(block) { // Get the logs of the block var ( - receipts = core.GetBlockReceipts(self.db, block.Hash(), i) - unfiltered vm.Logs + receipts = core.GetBlockReceipts(f.db, block.Hash(), i) + unfiltered []Log ) for _, receipt := range receipts { - unfiltered = append(unfiltered, receipt.Logs...) + rl := make([]Log, len(receipt.Logs)) + for i, l := range receipt.Logs { + rl[i] = Log{l, false} + } + unfiltered = append(unfiltered, rl...) } - logs = append(logs, self.FilterLogs(unfiltered)...) + logs = append(logs, filterLogs(unfiltered, f.addresses, f.topics)...) } } @@ -161,26 +160,26 @@ func includes(addresses []common.Address, a common.Address) bool { return false } -func (self *Filter) FilterLogs(logs vm.Logs) vm.Logs { - var ret vm.Logs +func filterLogs(logs []Log, addresses []common.Address, topics [][]common.Hash) []Log { + var ret []Log // Filter the logs for interesting stuff Logs: for _, log := range logs { - if len(self.addresses) > 0 && !includes(self.addresses, log.Address) { + if len(addresses) > 0 && !includes(addresses, log.Address) { continue } - logTopics := make([]common.Hash, len(self.topics)) + logTopics := make([]common.Hash, len(topics)) copy(logTopics, log.Topics) // If the to filtered topics is greater than the amount of topics in // logs, skip. - if len(self.topics) > len(log.Topics) { + if len(topics) > len(log.Topics) { continue Logs } - for i, topics := range self.topics { + for i, topics := range topics { var match bool for _, topic := range topics { // common.Hash{} is a match all (wildcard) @@ -202,10 +201,10 @@ Logs: return ret } -func (self *Filter) bloomFilter(block *types.Block) bool { - if len(self.addresses) > 0 { +func (f *Filter) bloomFilter(block *types.Block) bool { + if len(f.addresses) > 0 { var included bool - for _, addr := range self.addresses { + for _, addr := range f.addresses { if types.BloomLookup(block.Bloom(), addr) { included = true break @@ -217,7 +216,7 @@ func (self *Filter) bloomFilter(block *types.Block) bool { } } - for _, sub := range self.topics { + for _, sub := range f.topics { var included bool for _, topic := range sub { if (topic == common.Hash{}) || types.BloomLookup(block.Bloom(), topic) { diff --git a/eth/filters/filter_system.go b/eth/filters/filter_system.go index 4343dfa21c..c2cf47a6b7 100644 --- a/eth/filters/filter_system.go +++ b/eth/filters/filter_system.go @@ -14,172 +14,576 @@ // You should have received a copy of the GNU Lesser General Public License // along with the go-ethereum library. If not, see . -// package filters implements an ethereum filtering system for block, +// Package filters implements an ethereum filtering system for block, // transactions and log events. package filters import ( + "bufio" + crand "crypto/rand" + "encoding/binary" + "encoding/hex" + "encoding/json" + "errors" "fmt" + "math/rand" "sync" "time" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/vm" "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" ) -// FilterType determines the type of filter and is used to put the filter in to -// the correct bucket when added. -type FilterType byte - const ( - ChainFilter FilterType = iota // new block events filter - PendingTxFilter // pending transaction filter - LogFilter // new or removed log filter - PendingLogFilter // pending log filter + // UnknownFilter indicates an unkown filter type + UnknownFilter Type = iota + // BlockFilter queries for new blocks + BlockFilter + // PendingTxFilter queries pending transactions + PendingTxFilter + // LogFilter queries for new or removed (chain reorg) logs + LogFilter + // PendingLogFilter queries for logs for the pending block + PendingLogFilter + + maxPendingHashes = 10240 // max buffer size of block/tx hashes before an filter is considered inactive and disabled + maxPendingLogs = 10240 // max buffer size of logs before an filter is considered inactive and disabled ) -// FilterSystem manages filters that filter specific events such as -// block, transaction and log events. The Filtering system can be used to listen -// for specific LOG events fired by the EVM (Ethereum Virtual Machine). -type FilterSystem struct { - filterMu sync.RWMutex - filterId int +var ( + errFilterNotFound = errors.New("filter not found") + errCouldNotGenerateFilterID = errors.New("unable to generate filter id") + errInvalidFilterID = errors.New("invalid filter id") + errUnableToUninstallFilter = errors.New("unable to uninstall filter, retry later") - chainFilters map[int]*Filter - pendingTxFilters map[int]*Filter - logFilters map[int]*Filter - pendingLogFilters map[int]*Filter + filterIDGenMu sync.Mutex + filterIDGen = filterIDGenerator() +) - // generic is an ugly hack for Get - generic map[int]*Filter +type logsCallback func(FilterID, []Log) - sub event.Subscription +// Type determines the kind of filter and is used to put the filter in to +// the correct bucket when added. +type Type byte + +// FilterID determines the type for a filter identifier. +type FilterID [16]byte + +// MarshalJSON serializes a FilterID into its JSON representation. +func (f FilterID) MarshalJSON() ([]byte, error) { + return json.Marshal(fmt.Sprintf("0x%x", f)) } -// NewFilterSystem returns a newly allocated filter manager -func NewFilterSystem(mux *event.TypeMux) *FilterSystem { - fs := &FilterSystem{ - chainFilters: make(map[int]*Filter), - pendingTxFilters: make(map[int]*Filter), - logFilters: make(map[int]*Filter), - pendingLogFilters: make(map[int]*Filter), - generic: make(map[int]*Filter), +// filterIDGenerator helper utility that generates a (pseudo) random sequence of bytes +// that are used to generate filter identifiers. +func filterIDGenerator() *rand.Rand { + if seed, err := binary.ReadVarint(bufio.NewReader(crand.Reader)); err == nil { + return rand.New(rand.NewSource(seed)) } - fs.sub = mux.Subscribe( + return rand.New(rand.NewSource(int64(time.Now().Nanosecond()))) +} + +// newID generates a filter identifier. +func newID() FilterID { + filterIDGenMu.Lock() + defer filterIDGenMu.Unlock() + + var id FilterID + for i := 0; i < len(id); i += 7 { + val := filterIDGen.Int63() + for j := 0; i+j < len(id) && j < 7; j++ { + id[i+j] = byte(val) + val >>= 8 + } + } + + return id +} + +// UnmarshalJSON parses a FilterID from its JSON representation. +func (f *FilterID) UnmarshalJSON(data []byte) error { + // `"0x...."` + if len(data) != 36 || data[0] != '"' || data[35] != '"' || data[1] != '0' || (data[2] != 'x' && data[2] != 'X') { + return errInvalidFilterID + } + + _, err := hex.Decode(f[:], data[3:35]) + return err +} + +// Log is a helper that can hold additional information about vm.Log +// necessary for the RPC interface. +type Log struct { + *vm.Log + Removed bool `json:"removed"` +} + +type filter struct { + ID FilterID + typ Type + created time.Time + lastUsed time.Time + canTimeout bool + hashes chan common.Hash // results in case filter type returns hashes + logsCrit FilterCriteria + lc logsCallback + logs chan []Log // results in case filter type returns logs +} + +// Manager listens for new events and offers a filter system to queury for +// events that match a set of criteria. +// +// Note, this system cannot be used to query past logs, it will only receive +// and handle events that are posted by the global event mux. Use a raw Filter +// to query for logs that are already stored. +type Manager struct { + sub event.Subscription + + install chan *filter // install filter for event notification + uninstall chan *filter // remove filter for event notification + + allMu sync.RWMutex + all map[FilterID]*filter // all installed filters +} + +// NewManager creates a new manager that listens for event on the given mux, +// parses and filters them. It uses the all map to retrieve filter changes. The +// work loop holds its own index that is used to forward events to filters. +// +// The returned manager has a loop that needs to be stopped with the Stop function +// or by stopping the given mux. +func NewManager(mux *event.TypeMux) *Manager { + sub := mux.Subscribe( core.PendingLogsEvent{}, core.RemovedLogsEvent{}, core.ChainEvent{}, core.TxPreEvent{}, vm.Logs(nil), ) - go fs.filterLoop() - return fs -} -// Stop quits the filter loop required for polling events -func (fs *FilterSystem) Stop() { - fs.sub.Unsubscribe() -} - -// Add adds a filter to the filter manager -func (fs *FilterSystem) Add(filter *Filter, filterType FilterType) (int, error) { - fs.filterMu.Lock() - defer fs.filterMu.Unlock() - - id := fs.filterId - filter.created = time.Now() - - switch filterType { - case ChainFilter: - fs.chainFilters[id] = filter - case PendingTxFilter: - fs.pendingTxFilters[id] = filter - case LogFilter: - fs.logFilters[id] = filter - case PendingLogFilter: - fs.pendingLogFilters[id] = filter - default: - return 0, fmt.Errorf("unknown filter type %v", filterType) + m := &Manager{ + sub: sub, + install: make(chan *filter), + uninstall: make(chan *filter, 1024), + all: make(map[FilterID]*filter), } - fs.generic[id] = filter - fs.filterId++ + go m.run() + + return m +} + +// Uninstall filter. If the given filter could not be found an error is returned. +func (m *Manager) Uninstall(id FilterID) error { + m.allMu.Lock() + + if f, found := m.all[id]; found { + delete(m.all, id) + m.allMu.Unlock() + + select { + case m.uninstall <- f: + return nil + default: + // can only happen when there are too many pending uninstall requests + return errUnableToUninstallFilter + } + } + + m.allMu.Unlock() + return errFilterNotFound +} + +// FilterType returns the filter type for the given id. +// If the filter could be found UnknownFilter is returned. +func (m *Manager) FilterType(id FilterID) Type { + m.allMu.RLock() + defer m.allMu.RUnlock() + + if f, found := m.all[id]; found { + return f.typ + } + return UnknownFilter +} + +// NewBlockFilter returns a filter identifier that can be used to get the hashes +// for new blocks. The given callback is optional. If nil is given block hashes +// are queued until fetched with GetBlockFilterChanges. +func (m *Manager) NewBlockFilter() (FilterID, error) { + id := newID() + + f := &filter{ + ID: id, + created: time.Now(), + lastUsed: time.Now(), + canTimeout: true, + typ: BlockFilter, + hashes: make(chan common.Hash, maxPendingHashes), + logs: make(chan []Log, maxPendingLogs), + } + + m.allMu.Lock() + m.all[id] = f + m.allMu.Unlock() + + m.install <- f return id, nil } -// Remove removes a filter by filter id -func (fs *FilterSystem) Remove(id int) { - fs.filterMu.Lock() - defer fs.filterMu.Unlock() +// GetBlockFilterChanges returns the data for the filter with the given id since +// last time it was called. +func (m *Manager) GetBlockFilterChanges(id FilterID) ([]common.Hash, error) { + m.allMu.RLock() + if f, found := m.all[id]; found && f.typ == BlockFilter { + m.allMu.RUnlock() - delete(fs.chainFilters, id) - delete(fs.pendingTxFilters, id) - delete(fs.logFilters, id) - delete(fs.pendingLogFilters, id) - delete(fs.generic, id) + // retieve hashes + f.lastUsed = time.Now() + hashes := make([]common.Hash, 0, len(f.hashes)) // prevent (most) allocs + for { + select { + case h := <-f.hashes: + hashes = append(hashes, h) + default: + return hashes, nil + } + } + } + + m.allMu.RUnlock() + return nil, errFilterNotFound } -func (fs *FilterSystem) Get(id int) *Filter { - fs.filterMu.RLock() - defer fs.filterMu.RUnlock() +// NewLogFilterWithNoTimeout returns a filter identifier that can be used to fetch +// logs matching the given criteria. The created filter will not timeout and the +// callee is expected to manually uninstall the filter. +func (m *Manager) NewLogFilterWithNoTimeout(crit FilterCriteria, cb logsCallback) (FilterID, error) { + id := newID() - return fs.generic[id] + f := &filter{ + ID: id, + created: time.Now(), + lastUsed: time.Now(), + canTimeout: false, + typ: LogFilter, + hashes: make(chan common.Hash, maxPendingHashes), + logs: make(chan []Log, maxPendingLogs), + lc: cb, + logsCrit: crit, + } + + m.allMu.Lock() + m.all[id] = f + m.allMu.Unlock() + + m.install <- f + + return id, nil } -// filterLoop waits for specific events from ethereum and fires their handlers -// when the filter matches the requirements. -func (fs *FilterSystem) filterLoop() { - for event := range fs.sub.Chan() { - switch ev := event.Data.(type) { - case core.ChainEvent: - fs.filterMu.RLock() - for _, filter := range fs.chainFilters { - if filter.BlockCallback != nil && !filter.created.After(event.Time) { - filter.BlockCallback(ev.Block, ev.Logs) - } - } - fs.filterMu.RUnlock() - case core.TxPreEvent: - fs.filterMu.RLock() - for _, filter := range fs.pendingTxFilters { - if filter.TransactionCallback != nil && !filter.created.After(event.Time) { - filter.TransactionCallback(ev.Tx) - } - } - fs.filterMu.RUnlock() +// NewLogFilter returns a filter identifier that can be used to fetch logs matching +// the given criteria. +func (m *Manager) NewLogFilter(crit FilterCriteria, cb logsCallback) (FilterID, error) { + id := newID() - case vm.Logs: - fs.filterMu.RLock() - for _, filter := range fs.logFilters { - if filter.LogCallback != nil && !filter.created.After(event.Time) { - for _, log := range filter.FilterLogs(ev) { - filter.LogCallback(log, false) - } + f := &filter{ + ID: id, + created: time.Now(), + lastUsed: time.Now(), + canTimeout: true, + typ: LogFilter, + hashes: make(chan common.Hash, maxPendingHashes), + logs: make(chan []Log, maxPendingLogs), + lc: cb, + logsCrit: crit, + } + + m.allMu.Lock() + m.all[id] = f + m.allMu.Unlock() + + m.install <- f + + return id, nil +} + +// GetLogFilterCriteria returns the filtering criteria for a filter, or an error in +// case the filter could not be found. +func (m *Manager) GetLogFilterCriteria(id FilterID) (FilterCriteria, error) { + m.allMu.RLock() + defer m.allMu.RUnlock() + + if f, found := m.all[id]; found { + return f.logsCrit, nil + } + + return FilterCriteria{}, errFilterNotFound +} + +// NewPendingLogFilter creates a filter that returns new pending logs that match the given criteria. +func (m *Manager) NewPendingLogFilter(crit FilterCriteria, cb logsCallback) (FilterID, error) { + id := newID() + + f := &filter{ + ID: id, + created: time.Now(), + lastUsed: time.Now(), + canTimeout: true, + typ: PendingLogFilter, + hashes: make(chan common.Hash, maxPendingHashes), + logs: make(chan []Log, maxPendingLogs), + lc: cb, + logsCrit: crit, + } + + m.allMu.Lock() + m.all[id] = f + m.allMu.Unlock() + + m.install <- f + + return id, nil +} + +// GetPendingLogFilterChanges returns logs for the pending block. +func (m *Manager) GetPendingLogFilterChanges(id FilterID) ([]Log, error) { + m.allMu.RLock() + defer m.allMu.RUnlock() + + if f, found := m.all[id]; found && f.typ == PendingLogFilter { + f.lastUsed = time.Now() + allLogs := make([]Log, 0, len(f.logs)) // prevent (most) allocs + for { + select { + case logs := <-f.logs: + allLogs = append(allLogs, logs...) + default: // available logs read + return allLogs, nil + } + } + } + + return nil, errFilterNotFound +} + +// GetLogFilterChanges returns all logs matching the criteria for the filter with the given filter id. +func (m *Manager) GetLogFilterChanges(id FilterID) ([]Log, error) { + m.allMu.RLock() + defer m.allMu.RUnlock() + + if f, found := m.all[id]; found && f.typ == LogFilter { + f.lastUsed = time.Now() + allLogs := make([]Log, 0, len(f.logs)) // prevent (most) allocs for the append + for { + select { + case logs := <-f.logs: + allLogs = append(allLogs, logs...) + default: // available logs read + return allLogs, nil + } + } + } + + return nil, errFilterNotFound +} + +// NewPendingTransactionFilter creates a filter that retrieves pending transactions. +func (m *Manager) NewPendingTransactionFilter() (FilterID, error) { + id := newID() + + f := &filter{ + ID: id, + created: time.Now(), + lastUsed: time.Now(), + canTimeout: true, + typ: PendingTxFilter, + hashes: make(chan common.Hash, maxPendingHashes), + logs: make(chan []Log, maxPendingLogs), + } + + m.allMu.Lock() + m.all[id] = f + m.allMu.Unlock() + + m.install <- f + + return id, nil +} + +// GetPendingTxFilterChanges returns hashes for pending transactions which are added since the last poll. +func (m *Manager) GetPendingTxFilterChanges(id FilterID) ([]common.Hash, error) { + m.allMu.RLock() + defer m.allMu.RUnlock() + + if f, found := m.all[id]; found && f.typ == PendingTxFilter { + f.lastUsed = time.Now() + hashes := make([]common.Hash, 0, len(f.hashes)) // prevent (most) allocs + for { + select { + case hash := <-f.hashes: + hashes = append(hashes, hash) + default: // read available tx hashes + return hashes, nil + } + } + } + + return nil, errFilterNotFound +} + +type filterIndex map[Type]map[FilterID]*filter + +// process an event and forward to filters that match the criteria. +func (m *Manager) process(filters filterIndex, ev *event.Event) { + var inactive []*filter + + logHandler := func(f *filter, logs []Log) { + if f.lc != nil && len(logs) > 0 { + f.lc(f.ID, logs) + } else if len(logs) > 0 { + select { + case f.logs <- logs: + return + default: // data queue full, disable filter + inactive = append(inactive, f) + } + } + } + + switch e := ev.Data.(type) { + case core.ChainEvent: + for _, f := range filters[BlockFilter] { + if ev.Time.After(f.created) { + select { + case f.hashes <- e.Hash: + continue + default: + // data queue full, disable filter + inactive = append(inactive, f) } } - fs.filterMu.RUnlock() - case core.RemovedLogsEvent: - fs.filterMu.RLock() - for _, filter := range fs.logFilters { - if filter.LogCallback != nil && !filter.created.After(event.Time) { - for _, removedLog := range filter.FilterLogs(ev.Logs) { - filter.LogCallback(removedLog, true) - } + } + case core.TxPreEvent: + for _, f := range filters[PendingTxFilter] { + if ev.Time.After(f.created) { + select { + case f.hashes <- e.Tx.Hash(): + continue + default: + // data queue full, disable filter + inactive = append(inactive, f) } } - fs.filterMu.RUnlock() - case core.PendingLogsEvent: - fs.filterMu.RLock() - for _, filter := range fs.pendingLogFilters { - if filter.LogCallback != nil && !filter.created.After(event.Time) { - for _, pendingLog := range ev.Logs { - filter.LogCallback(pendingLog, false) - } - } + } + case vm.Logs: + for _, f := range filters[LogFilter] { + if ev.Time.After(f.created) { + matchedLogs := filterLogs(convertLogs(e, false), f.logsCrit.Addresses, f.logsCrit.Topics) + logHandler(f, matchedLogs) } - fs.filterMu.RUnlock() + } + case core.RemovedLogsEvent: + for _, f := range filters[LogFilter] { + if ev.Time.After(f.created) { + matchedLogs := filterLogs(convertLogs(e.Logs, true), f.logsCrit.Addresses, f.logsCrit.Topics) + logHandler(f, matchedLogs) + } + } + case core.PendingLogsEvent: + for _, f := range filters[PendingLogFilter] { + if ev.Time.After(f.created) { + matchedLogs := filterLogs(convertLogs(e.Logs, false), f.logsCrit.Addresses, f.logsCrit.Topics) + logHandler(f, matchedLogs) + } + } + } + + m.allMu.Lock() + for _, f := range inactive { + delete(m.all, f.ID) + glog.Warningf("filter 0x%x uninstalled, queue full\n", f.ID) + } + m.allMu.Unlock() + + // remove filter for event listening, this must be run in a seperate go routine + // since timeout is called from the work loop and sending uninstall requests to the + // work loop from "itself" may deadlock when the uninstall channel is full. + go func() { + for _, f := range inactive { + m.uninstall <- f + } + }() +} + +// timeout uninstalls all filters that have not been used in the last 5 minutes. +func (m *Manager) timeout() { + deadline := time.Now().Add(-5 * time.Minute) + var inactive []*filter + m.allMu.Lock() + for _, f := range m.all { + if f.lastUsed.Before(deadline) && f.canTimeout { + delete(m.all, f.ID) // filter cannot be used from the external + inactive = append(inactive, f) + } + } + m.allMu.Unlock() + + // remove filter for event listening, this must be run in a seperate go routine + // since timeout is called from the work loop and sending uninstall requests to the + // work loop from "itself" may deadlock when the uninstall channel is full. + go func() { + for _, f := range inactive { + m.uninstall <- f // delete for the internal work loop + } + }() +} + +// run is the manager work loop. +// It will receive events and forwards them to installed filters. +// Inactive filters will be uninstalled. +func (m *Manager) run() { + index := make(filterIndex) + timeout := time.NewTicker(30 * time.Second) + + for { + select { + case f := <-m.install: + // lazy load + if _, found := index[f.typ]; !found { + index[f.typ] = make(map[FilterID]*filter) + } + index[f.typ][f.ID] = f + case f := <-m.uninstall: + close(f.hashes) + close(f.logs) + delete(index[f.typ], f.ID) + case ev, ok := <-m.sub.Chan(): + if !ok { + glog.V(logger.Debug).Infoln("filter manager stopped") + return + } + m.process(index, ev) + case <-timeout.C: + m.timeout() } } } + +// Stop the filter system. +func (m *Manager) Stop() { + m.sub.Unsubscribe() // end worker loop +} + +// convertLogs is a helper utility that converts vm.Logs to []filter.Log. +func convertLogs(in vm.Logs, removed bool) []Log { + logs := make([]Log, len(in)) + for i, l := range in { + logs[i] = Log{l, false} + } + return logs +} diff --git a/eth/filters/filter_system_test.go b/eth/filters/filter_system_test.go index 72824cb088..5cbf51a132 100644 --- a/eth/filters/filter_system_test.go +++ b/eth/filters/filter_system_test.go @@ -17,101 +17,442 @@ package filters import ( + "encoding/json" + "math/big" + "reflect" "testing" "time" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/vm" "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/rpc" ) -func TestCallbacks(t *testing.T) { - var ( - mux event.TypeMux - fs = NewFilterSystem(&mux) - blockDone = make(chan struct{}) - txDone = make(chan struct{}) - logDone = make(chan struct{}) - removedLogDone = make(chan struct{}) - pendingLogDone = make(chan struct{}) - ) +var ( + mux = new(event.TypeMux) + filterManager = NewManager(mux) +) - blockFilter := &Filter{ - BlockCallback: func(*types.Block, vm.Logs) { - close(blockDone) - }, - } - txFilter := &Filter{ - TransactionCallback: func(*types.Transaction) { - close(txDone) - }, - } - logFilter := &Filter{ - LogCallback: func(l *vm.Log, oob bool) { - if !oob { - close(logDone) - } - }, - } - removedLogFilter := &Filter{ - LogCallback: func(l *vm.Log, oob bool) { - if oob { - close(removedLogDone) - } - }, - } - pendingLogFilter := &Filter{ - LogCallback: func(*vm.Log, bool) { - close(pendingLogDone) - }, +// TestFilterIdSerialization tests if FilterIs is correct serialized and parsed. +func TestFilterIdSerialization(t *testing.T) { + t.Parallel() + + id := newID() + + serialized, err := json.Marshal(id) + if err != nil { + t.Fatal(err) } - fs.Add(blockFilter, ChainFilter) - fs.Add(txFilter, PendingTxFilter) - fs.Add(logFilter, LogFilter) - fs.Add(removedLogFilter, LogFilter) - fs.Add(pendingLogFilter, PendingLogFilter) - - mux.Post(core.ChainEvent{}) - mux.Post(core.TxPreEvent{}) - mux.Post(vm.Logs{&vm.Log{}}) - mux.Post(core.RemovedLogsEvent{Logs: vm.Logs{&vm.Log{}}}) - mux.Post(core.PendingLogsEvent{Logs: vm.Logs{&vm.Log{}}}) - - const dura = 5 * time.Second - failTimer := time.NewTimer(dura) - select { - case <-blockDone: - case <-failTimer.C: - t.Error("block filter failed to trigger (timeout)") + if len(serialized) != 36 { + t.Fatalf("invalid filter id length (%s), want %d, got %d", serialized, 36, len(serialized)) } - failTimer.Reset(dura) - select { - case <-txDone: - case <-failTimer.C: - t.Error("transaction filter failed to trigger (timeout)") + var filterID FilterID + if err := json.Unmarshal(serialized, &filterID); err != nil { + t.Fatal(err) } - failTimer.Reset(dura) - select { - case <-logDone: - case <-failTimer.C: - t.Error("log filter failed to trigger (timeout)") - } - - failTimer.Reset(dura) - select { - case <-removedLogDone: - case <-failTimer.C: - t.Error("removed log filter failed to trigger (timeout)") - } - - failTimer.Reset(dura) - select { - case <-pendingLogDone: - case <-failTimer.C: - t.Error("pending log filter failed to trigger (timeout)") + if filterID != id { + t.Errorf("invalid filter id, want %x, got %x", id, filterID) + } +} + +// TestBlockFilter tests if a block filter callback is called when core.ChainEvent are posted. +// It creates multiple filters: +// - one at the start and should receive all posted chain events and a second (blockHashes) +// - one that is created after a cutoff moment and uninstalled after a second cutoff moment (blockHashes[cutoff1:cutoff2]) +// - one that is created after the second cutoff moment (blockHashes[cutoff2:]) +func TestBlockFilter(t *testing.T) { + t.Parallel() + + var ( + blockHashes = []common.Hash{ + common.HexToHash("0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3"), + common.HexToHash("0x88e96d4537bea4d9c05d12549907b32561d3bf31f45aae734cdc119f13406cb6"), + common.HexToHash("0xb495a1d7e6663152ae92708da4843337b958146015a2802f4193a410044698c9"), + common.HexToHash("0x3d6122660cc824376f11ee842f83addc3525e2dd6756b9bcf0affa6aa88cf741"), + common.HexToHash("0x23adf5a3be0f5235b36941bcb29b62504278ec5b9cdfa277b992ba4a7a3cd3a2"), + common.HexToHash("0xf37c632d361e0a93f08ba29b1a2c708d9caa3ee19d1ee8d2a02612bffe49f0a9"), + common.HexToHash("0x1f1aed8e3694a067496c248e61879cda99b0709a1dfbacd0b693750df06b326e"), + common.HexToHash("0xe0c7c0b46e116b874354dce6f64b8581bd239186b03f30a978e3dc38656f723a"), + common.HexToHash("0x2ce94342df186bab4165c268c43ab982d360c9474f429fec5565adfc5d1f258b"), + common.HexToHash("0x997e47bf4cac509c627753c06385ac866641ec6f883734ff7944411000dc576e"), + } + + cutoff1 = 3 + cutoff2 = cutoff1 + 2 + + receivedHashes1 []common.Hash + receivedHashes2 []common.Hash + receivedHashes3 []common.Hash + ) + + // should receive all block hashes + fid1, err := filterManager.NewBlockFilter() + if err != nil { + t.Fatal(err) + } + + for _, h := range blockHashes[:cutoff1] { + ev := core.ChainEvent{Hash: h} + mux.Post(ev) + } + + // fid1 receives blockHashes[:] + for { + hashes, err := filterManager.GetBlockFilterChanges(fid1) + if err != nil { + t.Fatalf("unable to fetch block filter changes: %v", err) + } + receivedHashes1 = append(receivedHashes1, hashes...) + + if len(receivedHashes1) >= cutoff1 { + break + } + + time.Sleep(100 * time.Millisecond) + } + + // fid2 receives blockHashes[cutoff1:cutoff2] + fid2, err := filterManager.NewBlockFilter() + if err != nil { + t.Fatal(err) + } + + for _, h := range blockHashes[cutoff1:cutoff2] { + ev := core.ChainEvent{Hash: h} + mux.Post(ev) + } + + for len(receivedHashes1) < cutoff2 { + hashes, err := filterManager.GetBlockFilterChanges(fid1) + if err != nil { + t.Fatalf("unable to fetch block filter changes: %v", err) + } + receivedHashes1 = append(receivedHashes1, hashes...) + } + + receivedHashes2, err = filterManager.GetBlockFilterChanges(fid2) + if err != nil { + t.Fatalf("unable to fetch block filter changes: %v", err) + } + + // delete second filter to test we don't get any more events anymore for the filter + if err := filterManager.Uninstall(fid2); err != nil { + t.Fatalf("uninstall filter failed with %v", err) + } + if _, err = filterManager.GetBlockFilterChanges(fid2); err != errFilterNotFound { + t.Errorf(`expected filter not found error, got "%v"`, err) + } + + fid3, err := filterManager.NewBlockFilter() + if err != nil { + t.Fatal(err) + } + + for _, h := range blockHashes[cutoff2:] { + ev := core.ChainEvent{Hash: h} + mux.Post(ev) + } + + for len(receivedHashes1) < len(blockHashes) { + hashes, err := filterManager.GetBlockFilterChanges(fid1) + if err != nil { + t.Fatalf("unable to fetch block filter changes: %v", err) + } + receivedHashes1 = append(receivedHashes1, hashes...) + } + + receivedHashes3, err = filterManager.GetBlockFilterChanges(fid3) + if err != nil { + t.Fatalf("unable to fetch block filter changes: %v", err) + } + + if len(receivedHashes1) != len(blockHashes) { + t.Fatalf("received invalid number of block hashes, want %d, got %d", len(blockHashes), len(receivedHashes1)) + } + if len(receivedHashes2) != len(blockHashes[cutoff1:cutoff2]) { + t.Fatalf("received invalid number of block hashes, want %d, got %d", len(blockHashes[cutoff1:cutoff2]), len(receivedHashes2)) + } + if len(receivedHashes3) != len(blockHashes[cutoff2:]) { + t.Fatalf("received invalid number of block hashes, want %d, got %d", len(blockHashes[cutoff2:]), len(receivedHashes2)) + } + + // verify hashes + for i, h := range receivedHashes1 { + if blockHashes[i] != h { + t.Errorf("blockhash %d invalid, want %x, got %x", i, blockHashes[i], h) + } + } + for i, h := range receivedHashes2 { + if blockHashes[cutoff1+i] != h { + t.Errorf("blockhash %d invalid, want %x, got %x", i, blockHashes[cutoff1+i], h) + } + } + for i, h := range receivedHashes3 { + if blockHashes[cutoff2+i] != h { + t.Errorf("blockhash %d invalid, want %x, got %x", i, blockHashes[cutoff2+i], h) + } + } + + if err := filterManager.Uninstall(fid1); err != nil { + t.Errorf("unable to uninstall %x", fid1) + } + if err := filterManager.Uninstall(fid1); err != errFilterNotFound { + t.Errorf("expected double uninstall to fail, want %s, got %s", errFilterNotFound, err) + } +} + +// TestPendingTxFilter tests whether pending tx filters retrieve all pending transactions that are posted to the event mux. +func TestPendingTxFilter(t *testing.T) { + t.Parallel() + + var ( + transactions = []*types.Transaction{ + types.NewTransaction(0, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), new(big.Int), new(big.Int), nil), + types.NewTransaction(1, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), new(big.Int), new(big.Int), nil), + types.NewTransaction(2, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), new(big.Int), new(big.Int), nil), + types.NewTransaction(3, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), new(big.Int), new(big.Int), nil), + types.NewTransaction(4, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), new(big.Int), new(big.Int), nil), + } + + hashes []common.Hash + ) + + fid1, err := filterManager.NewPendingTransactionFilter() + if err != nil { + t.Fatal(err) + } + + for _, tx := range transactions { + ev := core.TxPreEvent{Tx: tx} + mux.Post(ev) + } + + for { + h, err := filterManager.GetPendingTxFilterChanges(fid1) + if err != nil { + t.Fatalf("unable to fetch pending transactions: %v", err) + } + hashes = append(hashes, h...) + + if len(hashes) >= len(transactions) { + break + } + + time.Sleep(100 * time.Millisecond) + } + + for i := range hashes { + if hashes[i] != transactions[i].Hash() { + t.Errorf("hashes[%d] invalid, want %x, got %x", i, transactions[i].Hash(), hashes[i]) + } + } +} + +// TestLogFilter tests whether log filters match the correct logs that are posted to the event mux. +func TestLogFilter(t *testing.T) { + t.Parallel() + + var ( + err error + + firstAddr = common.HexToAddress("0x1111111111111111111111111111111111111111") + secondAddr = common.HexToAddress("0x2222222222222222222222222222222222222222") + thirdAddress = common.HexToAddress("0x3333333333333333333333333333333333333333") + notUsedAddress = common.HexToAddress("0x9999999999999999999999999999999999999999") + firstTopic = common.HexToHash("0x1111111111111111111111111111111111111111111111111111111111111111") + secondTopic = common.HexToHash("0x2222222222222222222222222222222222222222222222222222222222222222") + notUsedTopic = common.HexToHash("0x9999999999999999999999999999999999999999999999999999999999999999") + + allLogs = vm.Logs{ + // Note, these are used for comparison of the test cases. + 0: vm.NewLog(firstAddr, []common.Hash{}, []byte(""), 0), + 1: vm.NewLog(firstAddr, []common.Hash{firstTopic}, []byte(""), 1), + 2: vm.NewLog(secondAddr, []common.Hash{firstTopic}, []byte(""), 1), + 3: vm.NewLog(thirdAddress, []common.Hash{secondTopic}, []byte(""), 2), + 4: vm.NewLog(thirdAddress, []common.Hash{secondTopic}, []byte(""), 3), + } + + testCases = []struct { + crit FilterCriteria + expected vm.Logs + id FilterID + }{ + // match all + 0: {FilterCriteria{}, allLogs, FilterID{}}, + // match none due to no matching addresses + 1: {FilterCriteria{Addresses: []common.Address{common.Address{}, notUsedAddress}, Topics: [][]common.Hash{allLogs[0].Topics}}, vm.Logs{}, FilterID{}}, + // match logs based on addresses, ignore topics + 2: {FilterCriteria{Addresses: []common.Address{firstAddr}}, allLogs[:2], FilterID{}}, + // match none due to no matching topics (match with address) + 3: {FilterCriteria{Addresses: []common.Address{secondAddr}, Topics: [][]common.Hash{[]common.Hash{notUsedTopic}}}, vm.Logs{}, FilterID{}}, + // match logs based on addresses and topics + 4: {FilterCriteria{Addresses: []common.Address{thirdAddress}, Topics: [][]common.Hash{[]common.Hash{firstTopic, secondTopic}}}, allLogs[3:5], FilterID{}}, + // match logs based on multiple addresses and "or" topics + 5: {FilterCriteria{Addresses: []common.Address{secondAddr, thirdAddress}, Topics: [][]common.Hash{[]common.Hash{firstTopic, secondTopic}}}, allLogs[2:5], FilterID{}}, + // block numbers are ignored for filters created with New***Filter, these return all logs that match the given criterias when the state changes + 6: {FilterCriteria{Addresses: []common.Address{firstAddr}, FromBlock: rpc.BlockNumber(1), ToBlock: rpc.BlockNumber(2)}, allLogs[:2], FilterID{}}, + } + ) + + // create all filters + for i := range testCases { + testCases[i].id, err = filterManager.NewLogFilter(testCases[i].crit, nil) + if err != nil { + t.Fatal(err) + } + } + + // raise events + if err = mux.Post(allLogs); err != nil { + t.Fatal(err) + } + + for i, tt := range testCases { + var fetched []Log + for { // fetch all expected logs + logs, err := filterManager.GetLogFilterChanges(tt.id) + if err != nil { + t.Errorf("unable to retrieve logs for case %d, %s", i, err) + break + } + + fetched = append(fetched, logs...) + if len(fetched) >= len(tt.expected) { + break + } + + time.Sleep(100 * time.Millisecond) + } + + if len(fetched) != len(tt.expected) { + t.Errorf("invalid number of logs for case %d, want %d log(s), got %d", i, len(tt.expected), len(fetched)) + return + } + + for l := range fetched { + if fetched[l].Removed { + t.Errorf("expected log not to be removed for log %d in case %d", l, i) + } + if !reflect.DeepEqual(fetched[l].Log, tt.expected[l]) { + t.Errorf("invalid log on index %d for case %d", l, i) + } + + } + } +} + +func TestPendingLogFilter(t *testing.T) { + t.Parallel() + + var ( + firstAddr = common.HexToAddress("0x1111111111111111111111111111111111111111") + secondAddr = common.HexToAddress("0x2222222222222222222222222222222222222222") + thirdAddress = common.HexToAddress("0x3333333333333333333333333333333333333333") + notUsedAddress = common.HexToAddress("0x9999999999999999999999999999999999999999") + firstTopic = common.HexToHash("0x1111111111111111111111111111111111111111111111111111111111111111") + secondTopic = common.HexToHash("0x2222222222222222222222222222222222222222222222222222222222222222") + thirdTopic = common.HexToHash("0x3333333333333333333333333333333333333333333333333333333333333333") + forthTopic = common.HexToHash("0x4444444444444444444444444444444444444444444444444444444444444444") + notUsedTopic = common.HexToHash("0x9999999999999999999999999999999999999999999999999999999999999999") + + allLogs = []core.PendingLogsEvent{ + 0: core.PendingLogsEvent{Logs: vm.Logs{vm.NewLog(firstAddr, []common.Hash{}, []byte(""), 0)}}, + 1: core.PendingLogsEvent{Logs: vm.Logs{vm.NewLog(firstAddr, []common.Hash{firstTopic}, []byte(""), 1)}}, + 2: core.PendingLogsEvent{Logs: vm.Logs{vm.NewLog(secondAddr, []common.Hash{firstTopic}, []byte(""), 2)}}, + 3: core.PendingLogsEvent{Logs: vm.Logs{vm.NewLog(thirdAddress, []common.Hash{secondTopic}, []byte(""), 3)}}, + 4: core.PendingLogsEvent{Logs: vm.Logs{vm.NewLog(thirdAddress, []common.Hash{secondTopic}, []byte(""), 4)}}, + 5: core.PendingLogsEvent{Logs: vm.Logs{ + vm.NewLog(thirdAddress, []common.Hash{firstTopic}, []byte(""), 5), + vm.NewLog(thirdAddress, []common.Hash{thirdTopic}, []byte(""), 5), + vm.NewLog(thirdAddress, []common.Hash{forthTopic}, []byte(""), 5), + vm.NewLog(firstAddr, []common.Hash{firstTopic}, []byte(""), 5), + }}, + } + + concatLogs = func(pl []core.PendingLogsEvent) vm.Logs { + var logs vm.Logs + for _, l := range pl { + logs = append(logs, l.Logs...) + } + return logs + } + + testCases = []struct { + crit FilterCriteria + expected vm.Logs + id FilterID + }{ + // match all + 0: {FilterCriteria{}, concatLogs(allLogs), FilterID{}}, + // match none due to no matching addresses + 1: {FilterCriteria{Addresses: []common.Address{common.Address{}, notUsedAddress}, Topics: [][]common.Hash{[]common.Hash{}}}, vm.Logs{}, FilterID{}}, + // match logs based on addresses, ignore topics + 2: {FilterCriteria{Addresses: []common.Address{firstAddr}}, append(concatLogs(allLogs[:2]), allLogs[5].Logs[3]), FilterID{}}, + // match none due to no matching topics (match with address) + 3: {FilterCriteria{Addresses: []common.Address{secondAddr}, Topics: [][]common.Hash{[]common.Hash{notUsedTopic}}}, vm.Logs{}, FilterID{}}, + // match logs based on addresses and topics + 4: {FilterCriteria{Addresses: []common.Address{thirdAddress}, Topics: [][]common.Hash{[]common.Hash{firstTopic, secondTopic}}}, append(concatLogs(allLogs[3:5]), allLogs[5].Logs[0]), FilterID{}}, + // match logs based on multiple addresses and "or" topics + 5: {FilterCriteria{Addresses: []common.Address{secondAddr, thirdAddress}, Topics: [][]common.Hash{[]common.Hash{firstTopic, secondTopic}}}, append(concatLogs(allLogs[2:5]), allLogs[5].Logs[0]), FilterID{}}, + // block numbers are ignored for filters created with New***Filter, these return all logs that match the given criterias when the state changes + 6: {FilterCriteria{Addresses: []common.Address{firstAddr}, FromBlock: rpc.BlockNumber(2), ToBlock: rpc.BlockNumber(3)}, append(concatLogs(allLogs[:2]), allLogs[5].Logs[3]), FilterID{}}, + // multiple pending logs, should match only 2 topics from the logs in block 5 + 7: {FilterCriteria{Addresses: []common.Address{thirdAddress}, Topics: [][]common.Hash{[]common.Hash{firstTopic, forthTopic}}}, vm.Logs{allLogs[5].Logs[0], allLogs[5].Logs[2]}, FilterID{}}, + } + + err error + ) + + // create all filters + for i := range testCases { + testCases[i].id, err = filterManager.NewPendingLogFilter(testCases[i].crit, nil) + if err != nil { + t.Fatal(err) + } + } + + // raise events + for _, l := range allLogs { + if err := mux.Post(l); err != nil { + t.Fatal(err) + } + } + + for i, tt := range testCases { + var fetched []Log + for { + logs, err := filterManager.GetPendingLogFilterChanges(tt.id) + if err != nil { + t.Errorf("unable to retrieve logs for case %d, %s", i, err) + break + } + fetched = append(fetched, logs...) + + if len(fetched) >= len(tt.expected) { + break + } + + time.Sleep(100 * time.Millisecond) + } + + if len(fetched) != len(tt.expected) { + t.Errorf("invalid number of logs for case %d, want %d log(s), got %d", i, len(tt.expected), len(fetched)) + continue + } + + for l := range fetched { + if fetched[l].Removed { + t.Errorf("expected log not to be removed for log %d in case %d", l, i) + } + if !reflect.DeepEqual(fetched[l].Log, tt.expected[l]) { + t.Errorf("invalid log on index %d for case %d", l, i) + } + } } }