From 14a2f42f3700640f191e0095b50a266d2a919b38 Mon Sep 17 00:00:00 2001 From: obscuren Date: Thu, 19 Mar 2015 16:19:54 +0100 Subject: fixed chain event. Closes #529 --- core/block_processor.go | 20 ++++++++------------ core/chain_makers.go | 4 ++-- core/chain_manager.go | 10 +++++----- core/events.go | 29 +++++++++++++++++++++++------ core/filter.go | 4 ++-- core/types/common.go | 8 ++++++-- event/filter/eth_filter.go | 6 +++--- rpc/api.go | 15 +++++++++------ 8 files changed, 58 insertions(+), 38 deletions(-) diff --git a/core/block_processor.go b/core/block_processor.go index f67d6d006..62c9c92a6 100644 --- a/core/block_processor.go +++ b/core/block_processor.go @@ -7,8 +7,8 @@ import ( "sync" "time" - "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/pow" @@ -17,10 +17,6 @@ import ( "gopkg.in/fatih/set.v0" ) -type PendingBlockEvent struct { - Block *types.Block -} - var statelogger = logger.NewLogger("BLOCK") type BlockProcessor struct { @@ -137,7 +133,7 @@ func (self *BlockProcessor) ApplyTransactions(coinbase *state.StateObject, state block.Header().GasUsed = totalUsedGas if transientProcess { - go self.eventMux.Post(PendingBlockEvent{block}) + go self.eventMux.Post(PendingBlockEvent{block, statedb.Logs()}) } return receipts, handled, unhandled, erroneous, err @@ -146,25 +142,25 @@ func (self *BlockProcessor) ApplyTransactions(coinbase *state.StateObject, state // Process block will attempt to process the given block's transactions and applies them // on top of the block's parent state (given it exists) and will return wether it was // successful or not. -func (sm *BlockProcessor) Process(block *types.Block) (td *big.Int, err error) { +func (sm *BlockProcessor) Process(block *types.Block) (td *big.Int, logs state.Logs, err error) { // Processing a blocks may never happen simultaneously sm.mutex.Lock() defer sm.mutex.Unlock() header := block.Header() if sm.bc.HasBlock(header.Hash()) { - return nil, &KnownBlockError{header.Number, header.Hash()} + return nil, nil, &KnownBlockError{header.Number, header.Hash()} } if !sm.bc.HasBlock(header.ParentHash) { - return nil, ParentError(header.ParentHash) + return nil, nil, ParentError(header.ParentHash) } parent := sm.bc.GetBlock(header.ParentHash) return sm.processWithParent(block, parent) } -func (sm *BlockProcessor) processWithParent(block, parent *types.Block) (td *big.Int, err error) { +func (sm *BlockProcessor) processWithParent(block, parent *types.Block) (td *big.Int, logs state.Logs, err error) { sm.lastAttemptedBlock = block // Create a new state based on the parent's root (e.g., create copy) @@ -177,7 +173,7 @@ func (sm *BlockProcessor) processWithParent(block, parent *types.Block) (td *big // There can be at most two uncles if len(block.Uncles()) > 2 { - return nil, ValidationError("Block can only contain one uncle (contained %v)", len(block.Uncles())) + return nil, nil, ValidationError("Block can only contain one uncle (contained %v)", len(block.Uncles())) } receipts, err := sm.TransitionState(state, parent, block, false) @@ -236,7 +232,7 @@ func (sm *BlockProcessor) processWithParent(block, parent *types.Block) (td *big chainlogger.Infof("processed block #%d (%x...)\n", header.Number, block.Hash()[0:4]) - return td, nil + return td, state.Logs(), nil } // Validates the current block. Returns an error if the block was invalid, diff --git a/core/chain_makers.go b/core/chain_makers.go index 59c297dbe..d800dee34 100644 --- a/core/chain_makers.go +++ b/core/chain_makers.go @@ -4,8 +4,8 @@ import ( "fmt" "math/big" - "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/pow" "github.com/ethereum/go-ethereum/state" @@ -93,7 +93,7 @@ func makeChain(bman *BlockProcessor, parent *types.Block, max int, db common.Dat blocks := make(types.Blocks, max) for i := 0; i < max; i++ { block := makeBlock(bman, parent, i, db, seed) - td, err := bman.processWithParent(block, parent) + td, _, err := bman.processWithParent(block, parent) if err != nil { fmt.Println("process with parent failed", err) panic(err) diff --git a/core/chain_manager.go b/core/chain_manager.go index ff91b0427..88518e62b 100644 --- a/core/chain_manager.go +++ b/core/chain_manager.go @@ -6,8 +6,8 @@ import ( "math/big" "sync" - "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/rlp" @@ -411,7 +411,7 @@ func (self *ChainManager) InsertChain(chain types.Blocks) error { for i, block := range chain { // Call in to the block processor and check for errors. It's likely that if one block fails // all others will fail too (unless a known block is returned). - td, err := self.processor.Process(block) + td, logs, err := self.processor.Process(block) if err != nil { if IsKnownBlockErr(err) { continue @@ -437,7 +437,7 @@ func (self *ChainManager) InsertChain(chain types.Blocks) error { if block.Header().Number.Cmp(new(big.Int).Add(cblock.Header().Number, common.Big1)) < 0 { chainlogger.Infof("Split detected. New head #%v (%x) TD=%v, was #%v (%x) TD=%v\n", block.Header().Number, block.Hash()[:4], td, cblock.Header().Number, cblock.Hash()[:4], self.td) - queue[i] = ChainSplitEvent{block} + queue[i] = ChainSplitEvent{block, logs} queueEvent.splitCount++ } @@ -456,10 +456,10 @@ func (self *ChainManager) InsertChain(chain types.Blocks) error { self.setTransState(state.New(block.Root(), self.stateDb)) self.setTxState(state.New(block.Root(), self.stateDb)) - queue[i] = ChainEvent{block} + queue[i] = ChainEvent{block, logs} queueEvent.canonicalCount++ } else { - queue[i] = ChainSideEvent{block} + queue[i] = ChainSideEvent{block, logs} queueEvent.sideCount++ } } diff --git a/core/events.go b/core/events.go index 23678ef60..8c5fb592a 100644 --- a/core/events.go +++ b/core/events.go @@ -1,6 +1,9 @@ package core -import "github.com/ethereum/go-ethereum/core/types" +import ( + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/state" +) // TxPreEvent is posted when a transaction enters the transaction pool. type TxPreEvent struct{ Tx *types.Transaction } @@ -15,11 +18,25 @@ type NewBlockEvent struct{ Block *types.Block } type NewMinedBlockEvent struct{ Block *types.Block } // ChainSplit is posted when a new head is detected -type ChainSplitEvent struct{ Block *types.Block } - -type ChainEvent struct{ Block *types.Block } - -type ChainSideEvent struct{ Block *types.Block } +type ChainSplitEvent struct { + Block *types.Block + Logs state.Logs +} + +type ChainEvent struct { + Block *types.Block + Logs state.Logs +} + +type ChainSideEvent struct { + Block *types.Block + Logs state.Logs +} + +type PendingBlockEvent struct { + Block *types.Block + Logs state.Logs +} type ChainHeadEvent struct{ Block *types.Block } diff --git a/core/filter.go b/core/filter.go index d58aa8d7c..487e82902 100644 --- a/core/filter.go +++ b/core/filter.go @@ -33,8 +33,8 @@ type Filter struct { max int topics [][][]byte - BlockCallback func(*types.Block) - PendingCallback func(*types.Block) + BlockCallback func(*types.Block, state.Logs) + PendingCallback func(*types.Block, state.Logs) LogsCallback func(state.Logs) } diff --git a/core/types/common.go b/core/types/common.go index 795374959..def7936a5 100644 --- a/core/types/common.go +++ b/core/types/common.go @@ -1,7 +1,11 @@ package types -import "math/big" +import ( + "math/big" + + "github.com/ethereum/go-ethereum/state" +) type BlockProcessor interface { - Process(*Block) (*big.Int, error) + Process(*Block) (*big.Int, state.Logs, error) } diff --git a/event/filter/eth_filter.go b/event/filter/eth_filter.go index 4ba66a7e0..cb75d7e1a 100644 --- a/event/filter/eth_filter.go +++ b/event/filter/eth_filter.go @@ -63,7 +63,7 @@ func (self *FilterManager) filterLoop() { // Subscribe to events events := self.eventMux.Subscribe( core.PendingBlockEvent{}, - //core.ChainEvent{}, + core.ChainEvent{}, state.Logs(nil)) out: @@ -77,7 +77,7 @@ out: self.filterMu.RLock() for _, filter := range self.filters { if filter.BlockCallback != nil { - filter.BlockCallback(event.Block) + filter.BlockCallback(event.Block, event.Logs) } } self.filterMu.RUnlock() @@ -86,7 +86,7 @@ out: self.filterMu.RLock() for _, filter := range self.filters { if filter.PendingCallback != nil { - filter.PendingCallback(event.Block) + filter.PendingCallback(event.Block, event.Logs) } } self.filterMu.RUnlock() diff --git a/rpc/api.go b/rpc/api.go index 5d4e235d9..a0c12f837 100644 --- a/rpc/api.go +++ b/rpc/api.go @@ -86,7 +86,7 @@ func (self *EthereumApi) getStateWithNum(num int64) *xeth.State { } func (self *EthereumApi) start() { - timer := time.NewTicker(filterTickerTime) + timer := time.NewTicker(2 * time.Second) done: for { select { @@ -94,20 +94,20 @@ done: self.logMut.Lock() self.messagesMut.Lock() for id, filter := range self.logs { - if time.Since(filter.timeout) > 20*time.Second { + if time.Since(filter.timeout) > filterTickerTime { self.filterManager.UninstallFilter(id) delete(self.logs, id) } } for id, filter := range self.messages { - if time.Since(filter.timeout) > 20*time.Second { + if time.Since(filter.timeout) > filterTickerTime { self.xeth().Whisper().Unwatch(id) delete(self.messages, id) } } - self.logMut.Unlock() self.messagesMut.Unlock() + self.logMut.Unlock() case <-self.quit: break done } @@ -180,10 +180,13 @@ func (self *EthereumApi) NewFilterString(args *FilterStringArgs, reply *interfac var id int filter := core.NewFilter(self.xeth().Backend()) - callback := func(block *types.Block) { + callback := 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.StateLog{}) } @@ -483,7 +486,7 @@ func (p *EthereumApi) GetBlockUncleCountByNumber(blocknum int64) (int64, error) func (p *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) error { // Spec at https://github.com/ethereum/wiki/wiki/Generic-JSON-RPC - rpclogger.Debugf("%s %s", req.Method, req.Params) + rpclogger.Infof("%s %s", req.Method, req.Params) switch req.Method { case "web3_sha3": args := new(Sha3Args) -- cgit v1.2.3