From 7e160a677d1590f97708a0d297f978a99977d398 Mon Sep 17 00:00:00 2001 From: obscuren Date: Wed, 6 May 2015 17:51:32 +0200 Subject: [PATCH 1/5] xeth, core, event/filter, rpc: new block and transaction filters --- core/filter.go | 6 +- event/filter/eth_filter.go | 4 +- rpc/api.go | 23 ++-- rpc/responses.go | 11 ++ xeth/xeth.go | 245 +++++++++++++++++++++++++------------ 5 files changed, 202 insertions(+), 87 deletions(-) diff --git a/core/filter.go b/core/filter.go index c10fb7eeb..2ca57da65 100644 --- a/core/filter.go +++ b/core/filter.go @@ -22,9 +22,9 @@ type Filter struct { max int topics [][]common.Hash - BlockCallback func(*types.Block, state.Logs) - PendingCallback func(*types.Transaction) - LogsCallback func(state.Logs) + BlockCallback func(*types.Block, state.Logs) + TransactionCallback func(*types.Transaction) + LogsCallback func(state.Logs) } // Create a new filter which uses a bloom filter on blocks to figure out whether a particular block diff --git a/event/filter/eth_filter.go b/event/filter/eth_filter.go index a1abe3418..b0d5078a2 100644 --- a/event/filter/eth_filter.go +++ b/event/filter/eth_filter.go @@ -88,8 +88,8 @@ out: case core.TxPreEvent: self.filterMu.RLock() for _, filter := range self.filters { - if filter.PendingCallback != nil { - filter.PendingCallback(event.Tx) + if filter.TransactionCallback != nil { + filter.TransactionCallback(event.Tx) } } self.filterMu.RUnlock() diff --git a/rpc/api.go b/rpc/api.go index b79a1306e..a3312075b 100644 --- a/rpc/api.go +++ b/rpc/api.go @@ -322,14 +322,13 @@ func (api *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) err return err } - id := api.xeth().RegisterFilter(args.Earliest, args.Latest, args.Skip, args.Max, args.Address, args.Topics) + id := api.xeth().NewLogFilter(args.Earliest, args.Latest, args.Skip, args.Max, args.Address, args.Topics) *reply = newHexNum(big.NewInt(int64(id)).Bytes()) + case "eth_newBlockFilter": - args := new(FilterStringArgs) - if err := json.Unmarshal(req.Params, &args); err != nil { - return err - } - *reply = newHexNum(api.xeth().NewFilterString(args.Word)) + *reply = newHexNum(api.xeth().NewBlockFilter()) + case "eth_transactionFilter": + *reply = newHexNum(api.xeth().NewTransactionFilter()) case "eth_uninstallFilter": args := new(FilterIdArgs) if err := json.Unmarshal(req.Params, &args); err != nil { @@ -341,7 +340,17 @@ func (api *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) err if err := json.Unmarshal(req.Params, &args); err != nil { return err } - *reply = NewLogsRes(api.xeth().FilterChanged(args.Id)) + + switch api.xeth().GetFilterType(args.Id) { + case xeth.BlockFilterTy: + *reply = NewHashesRes(api.xeth().BlockFilterChanged(args.Id)) + case xeth.TransactionFilterTy: + *reply = NewHashesRes(api.xeth().TransactionFilterChanged(args.Id)) + case xeth.LogFilterTy: + *reply = NewLogsRes(api.xeth().LogFilterChanged(args.Id)) + default: + *reply = []string{} // reply empty string slice + } case "eth_getFilterLogs": args := new(FilterIdArgs) if err := json.Unmarshal(req.Params, &args); err != nil { diff --git a/rpc/responses.go b/rpc/responses.go index 884b7e69b..9fdf60c02 100644 --- a/rpc/responses.go +++ b/rpc/responses.go @@ -3,6 +3,7 @@ package rpc import ( "encoding/json" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/types" ) @@ -303,3 +304,13 @@ func NewLogsRes(logs state.Logs) (ls []LogRes) { return } + +func NewHashesRes(hs []common.Hash) []string { + hashes := make([]string, len(hs)) + + for i, hash := range hs { + hashes[i] = hash.Hex() + } + + return hashes +} diff --git a/xeth/xeth.go b/xeth/xeth.go index ac59069d5..a0b936b57 100644 --- a/xeth/xeth.go +++ b/xeth/xeth.go @@ -29,6 +29,14 @@ var ( defaultGas = big.NewInt(90000) //500000 ) +// byte will be inferred +const ( + UnknownFilterTy = iota + BlockFilterTy + TransactionFilterTy + LogFilterTy +) + func DefaultGas() *big.Int { return new(big.Int).Set(defaultGas) } func DefaultGasPrice() *big.Int { return new(big.Int).Set(defaultGasPrice) } @@ -42,11 +50,17 @@ type XEth struct { quit chan struct{} filterManager *filter.FilterManager - logMut sync.RWMutex - logs map[int]*logFilter + logMu sync.RWMutex + logQueue map[int]*logQueue - messagesMut sync.RWMutex - messages map[int]*whisperFilter + blockMu sync.RWMutex + blockQueue map[int]*hashQueue + + transactionMu sync.RWMutex + transactionQueue map[int]*hashQueue + + messagesMu sync.RWMutex + messages map[int]*whisperFilter // regmut sync.Mutex // register map[string][]*interface{} // TODO improve return type @@ -59,14 +73,16 @@ type XEth struct { // confirms all transactions will be used. func New(eth *eth.Ethereum, frontend Frontend) *XEth { xeth := &XEth{ - backend: eth, - frontend: frontend, - whisper: NewWhisper(eth.Whisper()), - quit: make(chan struct{}), - filterManager: filter.NewFilterManager(eth.EventMux()), - logs: make(map[int]*logFilter), - messages: make(map[int]*whisperFilter), - agent: miner.NewRemoteAgent(), + backend: eth, + frontend: frontend, + whisper: NewWhisper(eth.Whisper()), + quit: make(chan struct{}), + filterManager: filter.NewFilterManager(eth.EventMux()), + logQueue: make(map[int]*logQueue), + blockQueue: make(map[int]*hashQueue), + transactionQueue: make(map[int]*hashQueue), + messages: make(map[int]*whisperFilter), + agent: miner.NewRemoteAgent(), } eth.Miner().Register(xeth.agent) @@ -87,23 +103,41 @@ done: for { select { case <-timer.C: - self.logMut.Lock() - self.messagesMut.Lock() - for id, filter := range self.logs { + self.logMu.Lock() + for id, filter := range self.logQueue { if time.Since(filter.timeout) > filterTickerTime { self.filterManager.UninstallFilter(id) - delete(self.logs, id) + delete(self.logQueue, id) } } + self.logMu.Unlock() + self.blockMu.Lock() + for id, filter := range self.blockQueue { + if time.Since(filter.timeout) > filterTickerTime { + self.filterManager.UninstallFilter(id) + delete(self.blockQueue, id) + } + } + self.blockMu.Unlock() + + self.transactionMu.Lock() + for id, filter := range self.transactionQueue { + if time.Since(filter.timeout) > filterTickerTime { + self.filterManager.UninstallFilter(id) + delete(self.transactionQueue, id) + } + } + self.transactionMu.Unlock() + + self.messagesMu.Lock() for id, filter := range self.messages { if time.Since(filter.activity()) > filterTickerTime { self.Whisper().Unwatch(id) delete(self.messages, id) } } - self.messagesMut.Unlock() - self.logMut.Unlock() + self.messagesMu.Unlock() case <-self.quit: break done } @@ -360,7 +394,32 @@ func (self *XEth) SecretToAddress(key string) string { return common.ToHex(pair.Address()) } -func (self *XEth) RegisterFilter(earliest, latest int64, skip, max int, address []string, topics [][]string) int { +func (self *XEth) UninstallFilter(id int) bool { + defer self.filterManager.UninstallFilter(id) + + if _, ok := self.logQueue[id]; ok { + self.logMu.Lock() + defer self.logMu.Unlock() + delete(self.logQueue, id) + return true + } + if _, ok := self.blockQueue[id]; ok { + self.blockMu.Lock() + defer self.blockMu.Unlock() + delete(self.blockQueue, id) + return true + } + if _, ok := self.transactionQueue[id]; ok { + self.transactionMu.Lock() + defer self.transactionMu.Unlock() + delete(self.transactionQueue, id) + return true + } + + return false +} + +func (self *XEth) NewLogFilter(earliest, latest int64, skip, max int, address []string, topics [][]string) int { var id int filter := core.NewFilter(self.backend) filter.SetEarliestBlock(earliest) @@ -370,71 +429,90 @@ func (self *XEth) RegisterFilter(earliest, latest int64, skip, max int, address filter.SetAddress(cAddress(address)) filter.SetTopics(cTopics(topics)) filter.LogsCallback = func(logs state.Logs) { - self.logMut.Lock() - defer self.logMut.Unlock() + self.logMu.Lock() + defer self.logMu.Unlock() - self.logs[id].add(logs...) + self.logQueue[id].add(logs...) } id = self.filterManager.InstallFilter(filter) - self.logs[id] = &logFilter{timeout: time.Now()} + self.logQueue[id] = &logQueue{timeout: time.Now()} return id } -func (self *XEth) UninstallFilter(id int) bool { - if _, ok := self.logs[id]; ok { - delete(self.logs, id) - self.filterManager.UninstallFilter(id) - return true - } - - return false -} - -func (self *XEth) NewFilterString(word string) int { +func (self *XEth) NewTransactionFilter() int { var id int filter := core.NewFilter(self.backend) + filter.TransactionCallback = func(tx *types.Transaction) { + self.transactionMu.Lock() + defer self.transactionMu.Unlock() - switch word { - case "pending": - filter.PendingCallback = func(tx *types.Transaction) { - self.logMut.Lock() - defer self.logMut.Unlock() - - self.logs[id].add(&state.Log{}) - } - case "latest": - filter.BlockCallback = func(block *types.Block, logs state.Logs) { - self.logMut.Lock() - defer self.logMut.Unlock() - - for _, log := range logs { - self.logs[id].add(log) - } - self.logs[id].add(&state.Log{}) - } + self.transactionQueue[id].add(tx.Hash()) } - id = self.filterManager.InstallFilter(filter) - self.logs[id] = &logFilter{timeout: time.Now()} - + self.transactionQueue[id] = &hashQueue{timeout: time.Now()} return id } -func (self *XEth) FilterChanged(id int) state.Logs { - self.logMut.Lock() - defer self.logMut.Unlock() +func (self *XEth) NewBlockFilter() int { + var id int + filter := core.NewFilter(self.backend) + filter.BlockCallback = func(block *types.Block, logs state.Logs) { + self.blockMu.Lock() + defer self.blockMu.Unlock() - if self.logs[id] != nil { - return self.logs[id].get() + self.blockQueue[id].add(block.Hash()) + } + id = self.filterManager.InstallFilter(filter) + self.blockQueue[id] = &hashQueue{timeout: time.Now()} + return id +} + +func (self *XEth) GetFilterType(id int) byte { + if _, ok := self.blockQueue[id]; ok { + return BlockFilterTy + } else if _, ok := self.transactionQueue[id]; ok { + return TransactionFilterTy + } else if _, ok := self.logQueue[id]; ok { + return LogFilterTy } + return UnknownFilterTy +} + +func (self *XEth) LogFilterChanged(id int) state.Logs { + self.logMu.Lock() + defer self.logMu.Unlock() + + if self.logQueue[id] != nil { + return self.logQueue[id].get() + } + return nil +} + +func (self *XEth) BlockFilterChanged(id int) []common.Hash { + self.blockMu.Lock() + defer self.blockMu.Unlock() + + if self.blockQueue[id] != nil { + return self.blockQueue[id].get() + } + return nil +} + +func (self *XEth) TransactionFilterChanged(id int) []common.Hash { + self.blockMu.Lock() + defer self.blockMu.Unlock() + + if self.blockQueue[id] != nil { + return self.transactionQueue[id].get() + } return nil } func (self *XEth) Logs(id int) state.Logs { - self.logMut.Lock() - defer self.logMut.Unlock() + self.logMu.Lock() + defer self.logMu.Unlock() filter := self.filterManager.GetFilter(id) if filter != nil { @@ -465,24 +543,24 @@ func (p *XEth) NewWhisperFilter(to, from string, topics [][]string) int { // Callback to delegate core whisper messages to this xeth filter callback := func(msg WhisperMessage) { - p.messagesMut.RLock() // Only read lock to the filter pool - defer p.messagesMut.RUnlock() + p.messagesMu.RLock() // Only read lock to the filter pool + defer p.messagesMu.RUnlock() p.messages[id].insert(msg) } // Initialize the core whisper filter and wrap into xeth id = p.Whisper().Watch(to, from, topics, callback) - p.messagesMut.Lock() + p.messagesMu.Lock() p.messages[id] = newWhisperFilter(id, p.Whisper()) - p.messagesMut.Unlock() + p.messagesMu.Unlock() return id } // UninstallWhisperFilter disables and removes an existing filter. func (p *XEth) UninstallWhisperFilter(id int) bool { - p.messagesMut.Lock() - defer p.messagesMut.Unlock() + p.messagesMu.Lock() + defer p.messagesMu.Unlock() if _, ok := p.messages[id]; ok { delete(p.messages, id) @@ -493,8 +571,8 @@ func (p *XEth) UninstallWhisperFilter(id int) bool { // WhisperMessages retrieves all the known messages that match a specific filter. func (self *XEth) WhisperMessages(id int) []WhisperMessage { - self.messagesMut.RLock() - defer self.messagesMut.RUnlock() + self.messagesMu.RLock() + defer self.messagesMu.RUnlock() if self.messages[id] != nil { return self.messages[id].messages() @@ -505,8 +583,8 @@ func (self *XEth) WhisperMessages(id int) []WhisperMessage { // WhisperMessagesChanged retrieves all the new messages matched by a filter // since the last retrieval func (self *XEth) WhisperMessagesChanged(id int) []WhisperMessage { - self.messagesMut.RLock() - defer self.messagesMut.RUnlock() + self.messagesMu.RLock() + defer self.messagesMu.RUnlock() if self.messages[id] != nil { return self.messages[id].retrieve() @@ -767,19 +845,36 @@ func (m callmsg) Gas() *big.Int { return m.gas } func (m callmsg) Value() *big.Int { return m.value } func (m callmsg) Data() []byte { return m.data } -type logFilter struct { +type logQueue struct { logs state.Logs timeout time.Time id int } -func (l *logFilter) add(logs ...*state.Log) { +func (l *logQueue) add(logs ...*state.Log) { l.logs = append(l.logs, logs...) } -func (l *logFilter) get() state.Logs { +func (l *logQueue) get() state.Logs { l.timeout = time.Now() tmp := l.logs l.logs = nil return tmp } + +type hashQueue struct { + hashes []common.Hash + timeout time.Time + id int +} + +func (l *hashQueue) add(hashes ...common.Hash) { + l.hashes = append(l.hashes, hashes...) +} + +func (l *hashQueue) get() []common.Hash { + l.timeout = time.Now() + tmp := l.hashes + l.hashes = nil + return tmp +} From b3c9b66f2958bb09ecffdb7a8064121e3a27b989 Mon Sep 17 00:00:00 2001 From: obscuren Date: Thu, 7 May 2015 13:56:19 +0200 Subject: [PATCH 2/5] rpc: eth_transactionFilter => eth_newPendingTransactionFilter --- rpc/api.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rpc/api.go b/rpc/api.go index a3312075b..f75ae42c4 100644 --- a/rpc/api.go +++ b/rpc/api.go @@ -327,7 +327,7 @@ func (api *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) err case "eth_newBlockFilter": *reply = newHexNum(api.xeth().NewBlockFilter()) - case "eth_transactionFilter": + case "eth_newPendingTransactionFilter": *reply = newHexNum(api.xeth().NewTransactionFilter()) case "eth_uninstallFilter": args := new(FilterIdArgs) From 258a7b9a936b206ee61d04a04cbb930d4d479e0d Mon Sep 17 00:00:00 2001 From: obscuren Date: Thu, 7 May 2015 17:20:57 +0200 Subject: [PATCH 3/5] xeth: check proper queue for pending transaction filter --- xeth/xeth.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/xeth/xeth.go b/xeth/xeth.go index a0b936b57..90f709aea 100644 --- a/xeth/xeth.go +++ b/xeth/xeth.go @@ -504,7 +504,7 @@ func (self *XEth) TransactionFilterChanged(id int) []common.Hash { self.blockMu.Lock() defer self.blockMu.Unlock() - if self.blockQueue[id] != nil { + if self.transactionQueue[id] != nil { return self.transactionQueue[id].get() } return nil From dcfecebe1fbd977e4c6bd95e2ba14dd4d26f9532 Mon Sep 17 00:00:00 2001 From: obscuren Date: Thu, 7 May 2015 17:27:17 +0200 Subject: [PATCH 4/5] core: get transaction by hash from transaction pool --- core/transaction_pool.go | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/core/transaction_pool.go b/core/transaction_pool.go index bac6b7f0b..6898a4bda 100644 --- a/core/transaction_pool.go +++ b/core/transaction_pool.go @@ -204,6 +204,27 @@ func (self *TxPool) AddTransactions(txs []*types.Transaction) { } } +// GetTransaction allows you to check the pending and queued transaction in the +// transaction pool. +// It has two stategies, first check the pool (map) then check the queue +func (tp *TxPool) GetTransaction(hash common.Hash) *types.Transaction { + // check the txs first + if tx, ok := tp.txs[hash]; ok { + return tx + } + + // check queue + for _, txs := range tp.queue { + for _, tx := range txs { + if tx.Hash() == hash { + return tx + } + } + } + + return nil +} + func (self *TxPool) GetTransactions() (txs types.Transactions) { self.mu.RLock() defer self.mu.RUnlock() From 60b5a94428abf57dc921347684508db44f0f1a04 Mon Sep 17 00:00:00 2001 From: obscuren Date: Thu, 7 May 2015 18:23:24 +0200 Subject: [PATCH 5/5] xeth: getTransactionByHash, try pool if db fails --- xeth/xeth.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/xeth/xeth.go b/xeth/xeth.go index 90f709aea..e5f068fd1 100644 --- a/xeth/xeth.go +++ b/xeth/xeth.go @@ -235,6 +235,8 @@ func (self *XEth) EthTransactionByHash(hash string) (tx *types.Transaction, blha data, _ := self.backend.ExtraDb().Get(common.FromHex(hash)) if len(data) != 0 { tx = types.NewTransactionFromBytes(data) + } else { // check pending transactions + tx = self.backend.TxPool().GetTransaction(common.HexToHash(hash)) } // meta