core,eth,eth/filters,miner: send all TransactionEvent's together after block is done.

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)
 }
 

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))
 }
 

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

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)
 	}

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

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())
+		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}
+	return receipt, gas, retData, err
-	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
 }

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)
 }

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)
 }
 

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
 		}

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
+			retData := f.retData
-			f.retdata = nil
+			f.retData = nil
-			return returnRetData(retdata), nil
+			return returnRetData(retData), nil
 		}
 	}
 

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

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)

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)
 }
 

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

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)
 }
 

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))
 }
 

miner/worker.go

@@ -77,6 +77,7 @@ type Work struct {
 	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
 }