diff --git a/accounts/abi/bind/backends/simulated.go b/accounts/abi/bind/backends/simulated.go index a28bd6dcdc..621de58685 100644 --- a/accounts/abi/bind/backends/simulated.go +++ b/accounts/abi/bind/backends/simulated.go @@ -461,7 +461,7 @@ func (fb *filterBackend) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscr return fb.bc.SubscribeLogsEvent(ch) } -func (fb *filterBackend) SubscribeTransactionEvent(ch chan<- *core.TransactionEvent) event.Subscription { +func (fb *filterBackend) SubscribeTransactionEvent(ch chan<- core.TransactionEvent) event.Subscription { return fb.bc.SubscribeTransactionEvent(ch) } diff --git a/core/blockchain.go b/core/blockchain.go index 44e3adf0ff..9e3b3d708d 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -1036,7 +1036,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks) (int, []interface{}, []*ty // acquiring. var ( stats = insertStats{startTime: mclock.Now()} - events = make([]interface{}, 0, len(chain)) + events = make([]interface{}, 0, 10*len(chain)) lastCanon *types.Block coalescedLogs []*types.Log ) @@ -1145,7 +1145,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks) (int, []interface{}, []*ty return i, events, coalescedLogs, err } // Process block using the parent state as reference point. - receipts, logs, usedGas, err := bc.processor.Process(block, state, bc.vmConfig) + receipts, logs, usedGas, txPostEvents, err := bc.processor.Process(block, state, bc.vmConfig) if err != nil { bc.reportBlock(block, receipts, err) return i, events, coalescedLogs, err @@ -1171,6 +1171,9 @@ func (bc *BlockChain) insertChain(chain types.Blocks) (int, []interface{}, []*ty coalescedLogs = append(coalescedLogs, logs...) blockInsertTimer.UpdateSince(bstart) events = append(events, ChainEvent{block, block.Hash(), logs}) + for _, txPostEvent := range txPostEvents { + events = append(events, txPostEvent) + } lastCanon = block // Only count canonical blocks for GC processing time @@ -1181,6 +1184,10 @@ func (bc *BlockChain) insertChain(chain types.Blocks) (int, []interface{}, []*ty common.PrettyDuration(time.Since(bstart)), "txs", len(block.Transactions()), "gas", block.GasUsed(), "uncles", len(block.Uncles())) blockInsertTimer.UpdateSince(bstart) + for _, txPostEvent := range txPostEvents { + txPostEvent.RetData.Removed = true + events = append(events, txPostEvent) + } events = append(events, ChainSideEvent{block}) } stats.processed++ @@ -1372,6 +1379,9 @@ func (bc *BlockChain) PostChainEvents(events []interface{}, logs []*types.Log) { case ChainSideEvent: bc.chainSideFeed.Send(ev) + + case TransactionEvent: + bc.txPostFeed.Send(ev) } } } @@ -1560,7 +1570,7 @@ func (bc *BlockChain) SubscribeChainSideEvent(ch chan<- ChainSideEvent) event.Su } // SubscribeTransactionEvent registers a subscription of SubscribeTransactionEvent. -func (bc *BlockChain) SubscribeTransactionEvent(ch chan<- *TransactionEvent) event.Subscription { +func (bc *BlockChain) SubscribeTransactionEvent(ch chan<- TransactionEvent) event.Subscription { return bc.scope.Track(bc.txPostFeed.Subscribe(ch)) } diff --git a/core/blockchain_test.go b/core/blockchain_test.go index 375823b57b..24e81a0cb9 100644 --- a/core/blockchain_test.go +++ b/core/blockchain_test.go @@ -117,7 +117,7 @@ func testBlockChainImport(chain types.Blocks, blockchain *BlockChain) error { if err != nil { return err } - receipts, _, usedGas, err := blockchain.Processor().Process(block, statedb, vm.Config{}) + receipts, _, usedGas, _, err := blockchain.Processor().Process(block, statedb, vm.Config{}) if err != nil { blockchain.reportBlock(block, receipts, err) return err diff --git a/core/chain_makers.go b/core/chain_makers.go index 31c9e3fb77..18a63a074b 100644 --- a/core/chain_makers.go +++ b/core/chain_makers.go @@ -98,7 +98,7 @@ func (b *BlockGen) AddTxWithChain(bc *BlockChain, tx *types.Transaction) { b.SetCoinbase(common.Address{}) } b.statedb.Prepare(tx.Hash(), common.Hash{}, len(b.txs)) - receipt, _, err := ApplyTransaction(b.config, bc, &b.header.Coinbase, b.gasPool, b.statedb, b.header, tx, &b.header.GasUsed, vm.Config{}) + receipt, _, _, err := ApplyTransaction(b.config, bc, &b.header.Coinbase, b.gasPool, b.statedb, b.header, tx, &b.header.GasUsed, vm.Config{}) if err != nil { panic(err) } diff --git a/core/events.go b/core/events.go index 7c1a683920..7f72fb4aac 100644 --- a/core/events.go +++ b/core/events.go @@ -38,7 +38,7 @@ type NewMinedBlockEvent struct{ Block *types.Block } // TransactionEvent is posted when a transaction completes execution type TransactionEvent struct { TxHash common.Hash - RetData types.ReturnData + RetData *types.ReturnData } // RemovedTransactionEvent is posted when a reorg happens diff --git a/core/state_processor.go b/core/state_processor.go index 812825d39a..eff57e9bf7 100644 --- a/core/state_processor.go +++ b/core/state_processor.go @@ -53,13 +53,15 @@ func NewStateProcessor(config *params.ChainConfig, bc *BlockChain, engine consen // Process returns the receipts and logs accumulated during the process and // returns the amount of gas that was used in the process. If any of the // transactions failed to execute due to insufficient gas it will return an error. -func (p *StateProcessor) Process(block *types.Block, statedb *state.StateDB, cfg vm.Config) (types.Receipts, []*types.Log, uint64, error) { +func (p *StateProcessor) Process(block *types.Block, statedb *state.StateDB, cfg vm.Config) (types.Receipts, []*types.Log, uint64, []TransactionEvent, error) { var ( - receipts types.Receipts - usedGas = new(uint64) - header = block.Header() - allLogs []*types.Log - gp = new(GasPool).AddGas(block.GasLimit()) + receipts types.Receipts + usedGas = new(uint64) + header = block.Header() + allLogs []*types.Log + gp = new(GasPool).AddGas(block.GasLimit()) + retData *types.ReturnData + txPostEvents []TransactionEvent ) // Mutate the the block and state according to any hard-fork specs if p.config.DAOForkSupport && p.config.DAOForkBlock != nil && p.config.DAOForkBlock.Cmp(block.Number()) == 0 { @@ -68,27 +70,29 @@ func (p *StateProcessor) Process(block *types.Block, statedb *state.StateDB, cfg // Iterate over and process the individual transactions for i, tx := range block.Transactions() { statedb.Prepare(tx.Hash(), block.Hash(), i) - receipt, _, err := ApplyTransaction(p.config, p.bc, nil, gp, statedb, header, tx, usedGas, cfg) + receipt, _, data, err := ApplyTransaction(p.config, p.bc, nil, gp, statedb, header, tx, usedGas, cfg) if err != nil { - return nil, nil, 0, err + return nil, nil, 0, nil, err } receipts = append(receipts, receipt) allLogs = append(allLogs, receipt.Logs...) + retData = &types.ReturnData{TxHash: tx.Hash(), Data: data} + txPostEvents = append(txPostEvents, TransactionEvent{TxHash: tx.Hash(), RetData: retData}) } // Finalize the block, applying any consensus engine specific extras (e.g. block rewards) p.engine.Finalize(p.bc, header, statedb, block.Transactions(), block.Uncles(), receipts) - return receipts, allLogs, *usedGas, nil + return receipts, allLogs, *usedGas, txPostEvents, nil } // ApplyTransaction attempts to apply a transaction to the given state database // and uses the input parameters for its environment. It returns the receipt // for the transaction, gas used and an error if the transaction failed, // indicating the block was invalid. -func ApplyTransaction(config *params.ChainConfig, bc *BlockChain, author *common.Address, gp *GasPool, statedb *state.StateDB, header *types.Header, tx *types.Transaction, usedGas *uint64, cfg vm.Config) (*types.Receipt, uint64, error) { +func ApplyTransaction(config *params.ChainConfig, bc *BlockChain, author *common.Address, gp *GasPool, statedb *state.StateDB, header *types.Header, tx *types.Transaction, usedGas *uint64, cfg vm.Config) (*types.Receipt, uint64, []byte, error) { msg, err := tx.AsMessage(types.MakeSigner(config, header.Number)) if err != nil { - return nil, 0, err + return nil, 0, nil, err } // Create a new context to be used in the EVM environment context := NewEVMContext(msg, header, bc, author) @@ -97,9 +101,9 @@ func ApplyTransaction(config *params.ChainConfig, bc *BlockChain, author *common vmenv := vm.NewEVM(context, statedb, config, cfg) // Apply the transaction to the current state (included in the env) - data, gas, failed, err := ApplyMessage(vmenv, msg, gp) + retData, gas, failed, err := ApplyMessage(vmenv, msg, gp) if err != nil { - return nil, 0, err + return nil, 0, nil, err } // Update the state with pending changes @@ -124,11 +128,5 @@ func ApplyTransaction(config *params.ChainConfig, bc *BlockChain, author *common receipt.Logs = statedb.GetLogs(tx.Hash()) receipt.Bloom = types.CreateBloom(types.Receipts{receipt}) - retData := types.ReturnData{TxHash: receipt.TxHash, Data: data, Removed: false} - txPostEvent := TransactionEvent{TxHash: receipt.TxHash, RetData: retData} - if bc != nil { // For some tests unrelated to tx events, this ptr can be nil; should always be non-nil in production - go bc.txPostFeed.Send(&txPostEvent) - } - - return receipt, gas, err + return receipt, gas, retData, err } diff --git a/core/types.go b/core/types.go index d0bbaf0aa7..d9679d149d 100644 --- a/core/types.go +++ b/core/types.go @@ -42,5 +42,5 @@ type Validator interface { // of gas used in the process and return an error if any of the internal rules // failed. type Processor interface { - Process(block *types.Block, statedb *state.StateDB, cfg vm.Config) (types.Receipts, []*types.Log, uint64, error) + Process(block *types.Block, statedb *state.StateDB, cfg vm.Config) (types.Receipts, []*types.Log, uint64, []TransactionEvent, error) } diff --git a/eth/api_backend.go b/eth/api_backend.go index d5ea49ce67..a6f7055445 100644 --- a/eth/api_backend.go +++ b/eth/api_backend.go @@ -144,7 +144,7 @@ func (b *EthApiBackend) SubscribeChainSideEvent(ch chan<- core.ChainSideEvent) e return b.eth.BlockChain().SubscribeChainSideEvent(ch) } -func (b *EthApiBackend) SubscribeTransactionEvent(ch chan<- *core.TransactionEvent) event.Subscription { +func (b *EthApiBackend) SubscribeTransactionEvent(ch chan<- core.TransactionEvent) event.Subscription { return b.eth.BlockChain().SubscribeTransactionEvent(ch) } diff --git a/eth/api_tracer.go b/eth/api_tracer.go index 07c4457bc3..6d86e50e03 100644 --- a/eth/api_tracer.go +++ b/eth/api_tracer.go @@ -274,7 +274,7 @@ func (api *PrivateDebugAPI) traceChain(ctx context.Context, start, end *types.Bl traced += uint64(len(txs)) } // Generate the next state snapshot fast without tracing - _, _, _, err := api.eth.blockchain.Processor().Process(block, statedb, vm.Config{}) + _, _, _, _, err := api.eth.blockchain.Processor().Process(block, statedb, vm.Config{}) if err != nil { failed = err break @@ -509,7 +509,7 @@ func (api *PrivateDebugAPI) computeStateDB(block *types.Block, reexec uint64) (* if block = api.eth.blockchain.GetBlockByNumber(block.NumberU64() + 1); block == nil { return nil, fmt.Errorf("block #%d not found", block.NumberU64()+1) } - _, _, _, err := api.eth.blockchain.Processor().Process(block, statedb, vm.Config{}) + _, _, _, _, err := api.eth.blockchain.Processor().Process(block, statedb, vm.Config{}) if err != nil { return nil, err } diff --git a/eth/filters/api.go b/eth/filters/api.go index 4bea2c5484..ec14fb6e6e 100644 --- a/eth/filters/api.go +++ b/eth/filters/api.go @@ -48,7 +48,7 @@ type filter struct { hashes []common.Hash crit FilterCriteria logs []*types.Log - retdata []types.ReturnData + retData []types.ReturnData s *Subscription // associated subscription in event system } @@ -239,16 +239,16 @@ func (api *PublicFilterAPI) NewReturnDataFilter() rpc.ID { ) api.filtersMu.Lock() - api.filters[retSub.ID] = &filter{typ: ReturnDataSubscription, deadline: time.NewTimer(deadline), retdata: make([]types.ReturnData, 0), s: retSub} + api.filters[retSub.ID] = &filter{typ: ReturnDataSubscription, deadline: time.NewTimer(deadline), retData: make([]types.ReturnData, 0), s: retSub} api.filtersMu.Unlock() go func() { for { select { - case retdata := <-retCh: + case retData := <-retCh: api.filtersMu.Lock() if f, found := api.filters[retSub.ID]; found { - f.retdata = append(f.retdata, *retdata) + f.retData = append(f.retData, *retData) } api.filtersMu.Unlock() case <-retSub.Err(): @@ -498,9 +498,9 @@ func (api *PublicFilterAPI) GetFilterChanges(id rpc.ID) (interface{}, error) { f.logs = nil return returnLogs(logs), nil case ReturnDataSubscription: - retdata := f.retdata - f.retdata = nil - return returnRetData(retdata), nil + retData := f.retData + f.retData = nil + return returnRetData(retData), nil } } diff --git a/eth/filters/filter.go b/eth/filters/filter.go index 085a38aeb5..a14b5849f1 100644 --- a/eth/filters/filter.go +++ b/eth/filters/filter.go @@ -37,7 +37,7 @@ type Backend interface { GetLogs(ctx context.Context, blockHash common.Hash) ([][]*types.Log, error) SubscribeTxPreEvent(chan<- core.TxPreEvent) event.Subscription - SubscribeTransactionEvent(chan<- *core.TransactionEvent) event.Subscription + SubscribeTransactionEvent(chan<- core.TransactionEvent) event.Subscription SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription diff --git a/eth/filters/filter_system.go b/eth/filters/filter_system.go index 097739fe46..2c991f6d8d 100644 --- a/eth/filters/filter_system.go +++ b/eth/filters/filter_system.go @@ -86,7 +86,7 @@ type subscription struct { logs chan []*types.Log hashes chan common.Hash headers chan *types.Header - retdata chan *types.ReturnData + retData chan *types.ReturnData installed chan struct{} // closed when the filter is installed err chan error // closed when the filter is uninstalled } @@ -341,7 +341,7 @@ func (es *EventSystem) broadcast(filters filterIndex, ev interface{}) { for _, f := range filters[PendingTransactionsSubscription] { f.hashes <- e.Tx.Hash() } - case *core.TransactionEvent: + case core.TransactionEvent: for _, f := range filters[ReturnDataSubscription] { f.retData <- e.RetData } @@ -443,7 +443,7 @@ func (es *EventSystem) eventLoop() { txPreCh = make(chan core.TxPreEvent, txPreChanSize) txPreSub = es.backend.SubscribeTxPreEvent(txPreCh) // Subscribe to TransactionEvent from applyTransaction - txCh = make(chan *core.TransactionEvent, txPostChanSize) + txCh = make(chan core.TransactionEvent, txPostChanSize) txSub = es.backend.SubscribeTransactionEvent(txCh) // Subscribe RemovedLogsEvent rmLogsCh = make(chan core.RemovedLogsEvent, rmLogsChanSize) diff --git a/eth/filters/filter_system_test.go b/eth/filters/filter_system_test.go index f1978f9609..3ce7de37ce 100644 --- a/eth/filters/filter_system_test.go +++ b/eth/filters/filter_system_test.go @@ -101,7 +101,7 @@ func (b *testBackend) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subsc return b.chainFeed.Subscribe(ch) } -func (b *testBackend) SubscribeTransactionEvent(ch chan<- *core.TransactionEvent) event.Subscription { +func (b *testBackend) SubscribeTransactionEvent(ch chan<- core.TransactionEvent) event.Subscription { return b.txPostFeed.Subscribe(ch) } diff --git a/internal/ethapi/backend.go b/internal/ethapi/backend.go index 40f05d2890..baafd92c92 100644 --- a/internal/ethapi/backend.go +++ b/internal/ethapi/backend.go @@ -57,7 +57,7 @@ type Backend interface { SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription SubscribeChainSideEvent(ch chan<- core.ChainSideEvent) event.Subscription - SubscribeTransactionEvent(ch chan<- *core.TransactionEvent) event.Subscription + SubscribeTransactionEvent(ch chan<- core.TransactionEvent) event.Subscription // TxPool API SendTx(ctx context.Context, signedTx *types.Transaction) error diff --git a/les/api_backend.go b/les/api_backend.go index 0cce15b142..c8bf925479 100644 --- a/les/api_backend.go +++ b/les/api_backend.go @@ -133,7 +133,7 @@ func (b *LesApiBackend) SubscribeTxPreEvent(ch chan<- core.TxPreEvent) event.Sub return b.eth.txPool.SubscribeTxPreEvent(ch) } -func (b *LesApiBackend) SubscribeTransactionEvent(ch chan<- *core.TransactionEvent) event.Subscription { +func (b *LesApiBackend) SubscribeTransactionEvent(ch chan<- core.TransactionEvent) event.Subscription { return b.eth.blockchain.SubscribeTransactionEvent(ch) } diff --git a/light/lightchain.go b/light/lightchain.go index 640ee3b954..4596c2f924 100644 --- a/light/lightchain.go +++ b/light/lightchain.go @@ -490,7 +490,7 @@ func (self *LightChain) SubscribeChainSideEvent(ch chan<- core.ChainSideEvent) e } // SubscribeTransactionEvent registers a subscription of TransactionEvent. -func (self *LightChain) SubscribeTransactionEvent(ch chan<- *core.TransactionEvent) event.Subscription { +func (self *LightChain) SubscribeTransactionEvent(ch chan<- core.TransactionEvent) event.Subscription { return self.scope.Track(self.txPostFeed.Subscribe(ch)) } diff --git a/miner/worker.go b/miner/worker.go index 546959fe96..cf9f251907 100644 --- a/miner/worker.go +++ b/miner/worker.go @@ -74,9 +74,10 @@ type Work struct { Block *types.Block // the new block - header *types.Header - txs []*types.Transaction - receipts []*types.Receipt + header *types.Header + txs []*types.Transaction + receipts []*types.Receipt + txPostEvents []core.TransactionEvent createdAt time.Time } @@ -320,12 +321,20 @@ func (self *worker) wait() { // implicit by posting ChainHeadEvent mustCommitNewWork = false } + // Broadcast the block and announce chain insertion event self.mux.Post(core.NewMinedBlockEvent{Block: block}) var ( events []interface{} logs = work.state.Logs() ) + + for _, txEv := range work.txPostEvents { + if stat == core.SideStatTy { + txEv.RetData.Removed = true + } + events = append(events, txEv) + } events = append(events, core.ChainEvent{Block: block, Hash: block.Hash(), Logs: logs}) if stat == core.CanonStatTy { events = append(events, core.ChainHeadEvent{Block: block}) @@ -425,7 +434,7 @@ func (self *worker) commitNewWork() { log.Error("Failed to prepare header for mining", "err", err) return } - // If we are care about TheDAO hard-fork check whether to override the extra-data or not + // If we care about TheDAO hard-fork, check whether to override the extra-data or not if daoBlock := self.config.DAOForkBlock; daoBlock != nil { // Check whether the block is among the fork extra-override range limit := new(big.Int).Add(daoBlock, params.DAOForkExtraRange) @@ -553,7 +562,7 @@ func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsB // Start executing the transaction env.state.Prepare(tx.Hash(), common.Hash{}, env.tcount) - err, logs := env.commitTransaction(tx, bc, coinbase, gp) + err, retData, logs := env.commitTransaction(tx, bc, coinbase, gp) switch err { case core.ErrGasLimitReached: // Pop the current out-of-gas transaction without shifting in the next from the account @@ -573,6 +582,8 @@ func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsB case nil: // Everything ok, collect the logs and shift in the next transaction from the same account coalescedLogs = append(coalescedLogs, logs...) + r := types.ReturnData{TxHash: tx.Hash(), Data: retData} + env.txPostEvents = append(env.txPostEvents, core.TransactionEvent{TxHash: r.TxHash, RetData: &r}) env.tcount++ txs.Shift() @@ -604,16 +615,16 @@ func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsB } } -func (env *Work) commitTransaction(tx *types.Transaction, bc *core.BlockChain, coinbase common.Address, gp *core.GasPool) (error, []*types.Log) { +func (env *Work) commitTransaction(tx *types.Transaction, bc *core.BlockChain, coinbase common.Address, gp *core.GasPool) (error, []byte, []*types.Log) { snap := env.state.Snapshot() - receipt, _, err := core.ApplyTransaction(env.config, bc, &coinbase, gp, env.state, env.header, tx, &env.header.GasUsed, vm.Config{}) + receipt, _, retData, err := core.ApplyTransaction(env.config, bc, &coinbase, gp, env.state, env.header, tx, &env.header.GasUsed, vm.Config{}) if err != nil { env.state.RevertToSnapshot(snap) - return err, nil + return err, nil, nil } env.txs = append(env.txs, tx) env.receipts = append(env.receipts, receipt) - return nil, receipt.Logs + return nil, retData, receipt.Logs }