diff --git a/core/blockchain.go b/core/blockchain.go
index 95ed06d8d6fbe6f3ce37ac2c8da04ead0cfb2f85..22dd617ad885b9d579d6a6a81e509ce074112899 100644
--- a/core/blockchain.go
+++ b/core/blockchain.go
@@ -1358,7 +1358,7 @@ func (self *BlockChain) reorg(oldBlock, newBlock *types.Block) error {
go self.eventMux.Post(RemovedTransactionEvent{diff})
}
if len(deletedLogs) > 0 {
- go self.eventMux.Post(RemovedLogEvent{deletedLogs})
+ go self.eventMux.Post(RemovedLogsEvent{deletedLogs})
}
return nil
diff --git a/core/blockchain_test.go b/core/blockchain_test.go
index b4ac1696a3219cab689af9b4ec8cc45f53331f58..1bb5f646dba97be3fe110ec6b909c54503d26fdf 100644
--- a/core/blockchain_test.go
+++ b/core/blockchain_test.go
@@ -982,7 +982,7 @@ func TestLogReorgs(t *testing.T) {
evmux := &event.TypeMux{}
blockchain, _ := NewBlockChain(db, FakePow{}, evmux)
- subs := evmux.Subscribe(RemovedLogEvent{})
+ subs := evmux.Subscribe(RemovedLogsEvent{})
chain, _ := GenerateChain(genesis, db, 2, func(i int, gen *BlockGen) {
if i == 1 {
tx, err := types.NewContractCreation(gen.TxNonce(addr1), new(big.Int), big.NewInt(1000000), new(big.Int), code).SignECDSA(key1)
@@ -1002,7 +1002,7 @@ func TestLogReorgs(t *testing.T) {
}
ev := <-subs.Chan()
- if len(ev.Data.(RemovedLogEvent).Logs) == 0 {
+ if len(ev.Data.(RemovedLogsEvent).Logs) == 0 {
t.Error("expected logs")
}
}
diff --git a/core/events.go b/core/events.go
index 1a760c71cfedcdd260380682d0774788e5e9ca07..c23206cad1741112cd01c2e628f579fdd386dd24 100644
--- a/core/events.go
+++ b/core/events.go
@@ -30,6 +30,11 @@ type TxPreEvent struct{ Tx *types.Transaction }
// TxPostEvent is posted when a transaction has been processed.
type TxPostEvent struct{ Tx *types.Transaction }
+// PendingLogsEvent is posted pre mining and notifies of pending logs.
+type PendingLogsEvent struct {
+ Logs vm.Logs
+}
+
// NewBlockEvent is posted when a block has been imported.
type NewBlockEvent struct{ Block *types.Block }
@@ -40,7 +45,7 @@ type NewMinedBlockEvent struct{ Block *types.Block }
type RemovedTransactionEvent struct{ Txs types.Transactions }
// RemovedLogEvent is posted when a reorg happens
-type RemovedLogEvent struct{ Logs vm.Logs }
+type RemovedLogsEvent struct{ Logs vm.Logs }
// ChainSplit is posted when a new head is detected
type ChainSplitEvent struct {
diff --git a/eth/filters/api.go b/eth/filters/api.go
index 148daa649bbc9d567f1b610c8ddc1b0d8c7da744..6cd184b808d13e08c41aebeb0cb518cfd6bc3e21 100644
--- a/eth/filters/api.go
+++ b/eth/filters/api.go
@@ -142,7 +142,11 @@ func (s *PublicFilterAPI) NewBlockFilter() (string, error) {
s.blockMu.Lock()
filter := New(s.chainDb)
- id := s.filterManager.Add(filter)
+ 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) {
@@ -174,7 +178,11 @@ func (s *PublicFilterAPI) NewPendingTransactionFilter() (string, error) {
defer s.transactionMu.Unlock()
filter := New(s.chainDb)
- id := s.filterManager.Add(filter)
+ 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) {
@@ -194,12 +202,16 @@ func (s *PublicFilterAPI) NewPendingTransactionFilter() (string, error) {
}
// newLogFilter creates a new log filter.
-func (s *PublicFilterAPI) newLogFilter(earliest, latest int64, addresses []common.Address, topics [][]common.Hash) int {
+func (s *PublicFilterAPI) newLogFilter(earliest, latest int64, addresses []common.Address, topics [][]common.Hash) (int, error) {
s.logMu.Lock()
defer s.logMu.Unlock()
filter := New(s.chainDb)
- id := s.filterManager.Add(filter)
+ id, err := s.filterManager.Add(filter, LogFilter)
+ if err != nil {
+ return 0, err
+ }
+
s.logQueue[id] = &logQueue{timeout: time.Now()}
filter.SetBeginBlock(earliest)
@@ -215,7 +227,7 @@ func (s *PublicFilterAPI) newLogFilter(earliest, latest int64, addresses []commo
}
}
- return id
+ return id, nil
}
// NewFilterArgs represents a request to create a new filter.
@@ -352,9 +364,12 @@ func (s *PublicFilterAPI) NewFilter(args NewFilterArgs) (string, error) {
var id int
if len(args.Addresses) > 0 {
- id = s.newLogFilter(args.FromBlock.Int64(), args.ToBlock.Int64(), args.Addresses, args.Topics)
+ id, err = s.newLogFilter(args.FromBlock.Int64(), args.ToBlock.Int64(), args.Addresses, args.Topics)
} else {
- id = s.newLogFilter(args.FromBlock.Int64(), args.ToBlock.Int64(), nil, args.Topics)
+ id, err = s.newLogFilter(args.FromBlock.Int64(), args.ToBlock.Int64(), nil, args.Topics)
+ }
+ if err != nil {
+ return "", err
}
s.filterMapMu.Lock()
diff --git a/eth/filters/filter.go b/eth/filters/filter.go
index 2c92d20b145020addb23d27834f3f0ae694e307e..96af93c4a3dd57187d2f001ca790dc3364a9036b 100644
--- a/eth/filters/filter.go
+++ b/eth/filters/filter.go
@@ -18,6 +18,7 @@ package filters
import (
"math"
+ "time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core"
@@ -32,6 +33,8 @@ type AccountChange struct {
// Filtering interface
type Filter struct {
+ created time.Time
+
db ethdb.Database
begin, end int64
addresses []common.Address
diff --git a/eth/filters/filter_system.go b/eth/filters/filter_system.go
index 04e58a08c54f443958e4221b33d0884fd763fd8d..b61a493b62ed528b8218924d9ca6fd1519ee95c5 100644
--- a/eth/filters/filter_system.go
+++ b/eth/filters/filter_system.go
@@ -19,6 +19,7 @@
package filters
import (
+ "fmt"
"sync"
"time"
@@ -27,26 +28,47 @@ import (
"github.com/ethereum/go-ethereum/event"
)
+// 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
+)
+
// 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
- filters map[int]*Filter
- created map[int]time.Time
- sub event.Subscription
+
+ chainFilters map[int]*Filter
+ pendingTxFilters map[int]*Filter
+ logFilters map[int]*Filter
+ pendingLogFilters map[int]*Filter
+
+ // generic is an ugly hack for Get
+ generic map[int]*Filter
+
+ sub event.Subscription
}
// NewFilterSystem returns a newly allocated filter manager
func NewFilterSystem(mux *event.TypeMux) *FilterSystem {
fs := &FilterSystem{
- filters: make(map[int]*Filter),
- created: make(map[int]time.Time),
+ 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),
}
fs.sub = mux.Subscribe(
- //core.PendingBlockEvent{},
- core.RemovedLogEvent{},
+ core.PendingLogsEvent{},
+ core.RemovedLogsEvent{},
core.ChainEvent{},
core.TxPreEvent{},
vm.Logs(nil),
@@ -61,15 +83,30 @@ func (fs *FilterSystem) Stop() {
}
// Add adds a filter to the filter manager
-func (fs *FilterSystem) Add(filter *Filter) (id int) {
+func (fs *FilterSystem) Add(filter *Filter, filterType FilterType) (int, error) {
fs.filterMu.Lock()
defer fs.filterMu.Unlock()
- id = fs.filterId
- fs.filters[id] = filter
- fs.created[id] = time.Now()
+
+ 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)
+ }
+ fs.generic[id] = filter
+
fs.filterId++
- return id
+ return id, nil
}
// Remove removes a filter by filter id
@@ -77,16 +114,18 @@ func (fs *FilterSystem) Remove(id int) {
fs.filterMu.Lock()
defer fs.filterMu.Unlock()
- delete(fs.filters, id)
- delete(fs.created, id)
+ delete(fs.chainFilters, id)
+ delete(fs.pendingTxFilters, id)
+ delete(fs.logFilters, id)
+ delete(fs.pendingLogFilters, id)
+ delete(fs.generic, id)
}
-// Get retrieves a filter installed using Add The filter may not be modified.
func (fs *FilterSystem) Get(id int) *Filter {
fs.filterMu.RLock()
defer fs.filterMu.RUnlock()
- return fs.filters[id]
+ return fs.generic[id]
}
// filterLoop waits for specific events from ethereum and fires their handlers
@@ -96,17 +135,16 @@ func (fs *FilterSystem) filterLoop() {
switch ev := event.Data.(type) {
case core.ChainEvent:
fs.filterMu.RLock()
- for id, filter := range fs.filters {
- if filter.BlockCallback != nil && !fs.created[id].After(event.Time) {
+ 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 id, filter := range fs.filters {
- if filter.TransactionCallback != nil && !fs.created[id].After(event.Time) {
+ for _, filter := range fs.pendingTxFilters {
+ if filter.TransactionCallback != nil && !filter.created.After(event.Time) {
filter.TransactionCallback(ev.Tx)
}
}
@@ -114,25 +152,34 @@ func (fs *FilterSystem) filterLoop() {
case vm.Logs:
fs.filterMu.RLock()
- for id, filter := range fs.filters {
- if filter.LogCallback != nil && !fs.created[id].After(event.Time) {
+ for _, filter := range fs.logFilters {
+ if filter.LogCallback != nil && !filter.created.After(event.Time) {
for _, log := range filter.FilterLogs(ev) {
filter.LogCallback(log, false)
}
}
}
fs.filterMu.RUnlock()
-
- case core.RemovedLogEvent:
+ case core.RemovedLogsEvent:
fs.filterMu.RLock()
- for id, filter := range fs.filters {
- if filter.LogCallback != nil && !fs.created[id].After(event.Time) {
+ for _, filter := range fs.logFilters {
+ if filter.LogCallback != nil && !filter.created.After(event.Time) {
for _, removedLog := range ev.Logs {
filter.LogCallback(removedLog, true)
}
}
}
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)
+ }
+ }
+ }
+ fs.filterMu.RUnlock()
}
}
}
diff --git a/eth/filters/filter_system_test.go b/eth/filters/filter_system_test.go
index 7ddeb02bc92399dfb6e68e15b70501bc97239efd..3ad7dd9cb6a30c0c250e5744fcd4886f7d9d19b4 100644
--- a/eth/filters/filter_system_test.go
+++ b/eth/filters/filter_system_test.go
@@ -18,6 +18,7 @@ func TestCallbacks(t *testing.T) {
txDone = make(chan struct{})
logDone = make(chan struct{})
removedLogDone = make(chan struct{})
+ pendingLogDone = make(chan struct{})
)
blockFilter := &Filter{
@@ -37,7 +38,6 @@ func TestCallbacks(t *testing.T) {
}
},
}
-
removedLogFilter := &Filter{
LogCallback: func(l *vm.Log, oob bool) {
if oob {
@@ -45,16 +45,23 @@ func TestCallbacks(t *testing.T) {
}
},
}
+ pendingLogFilter := &Filter{
+ LogCallback: func(*vm.Log, bool) {
+ close(pendingLogDone)
+ },
+ }
- fs.Add(blockFilter)
- fs.Add(txFilter)
- fs.Add(logFilter)
- fs.Add(removedLogFilter)
+ 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(core.RemovedLogEvent{vm.Logs{&vm.Log{}}})
mux.Post(vm.Logs{&vm.Log{}})
+ mux.Post(core.RemovedLogsEvent{vm.Logs{&vm.Log{}}})
+ mux.Post(core.PendingLogsEvent{vm.Logs{&vm.Log{}}})
const dura = 5 * time.Second
failTimer := time.NewTimer(dura)
@@ -84,4 +91,11 @@ func TestCallbacks(t *testing.T) {
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 (timout)")
+ }
}
diff --git a/miner/worker.go b/miner/worker.go
index 9c29d22504f8507d7926ebda59624fbdfdfe8d9a..81f7b16acad07ada1d294feeebff994713be436d 100644
--- a/miner/worker.go
+++ b/miner/worker.go
@@ -243,7 +243,7 @@ func (self *worker) update() {
// Apply transaction to the pending state if we're not mining
if atomic.LoadInt32(&self.mining) == 0 {
self.currentMu.Lock()
- self.current.commitTransactions(types.Transactions{ev.Tx}, self.gasPrice, self.chain)
+ self.current.commitTransactions(self.mux, types.Transactions{ev.Tx}, self.gasPrice, self.chain)
self.currentMu.Unlock()
}
}
@@ -529,7 +529,7 @@ func (self *worker) commitNewWork() {
transactions := append(singleTxOwner, multiTxOwner...)
*/
- work.commitTransactions(transactions, self.gasPrice, self.chain)
+ work.commitTransactions(self.mux, transactions, self.gasPrice, self.chain)
self.eth.TxPool().RemoveTransactions(work.lowGasTxs)
// compute uncles for the new block.
@@ -588,8 +588,10 @@ func (self *worker) commitUncle(work *Work, uncle *types.Header) error {
return nil
}
-func (env *Work) commitTransactions(transactions types.Transactions, gasPrice *big.Int, bc *core.BlockChain) {
+func (env *Work) commitTransactions(mux *event.TypeMux, transactions types.Transactions, gasPrice *big.Int, bc *core.BlockChain) {
gp := new(core.GasPool).AddGas(env.header.GasLimit)
+
+ var coalescedLogs vm.Logs
for _, tx := range transactions {
// We can skip err. It has already been validated in the tx pool
from, _ := tx.From()
@@ -627,7 +629,7 @@ func (env *Work) commitTransactions(transactions types.Transactions, gasPrice *b
env.state.StartRecord(tx.Hash(), common.Hash{}, 0)
- err := env.commitTransaction(tx, bc, gp)
+ err, logs := env.commitTransaction(tx, bc, gp)
switch {
case core.IsGasLimitErr(err):
// ignore the transactor so no nonce errors will be thrown for this account
@@ -643,20 +645,25 @@ func (env *Work) commitTransactions(transactions types.Transactions, gasPrice *b
}
default:
env.tcount++
+ coalescedLogs = append(coalescedLogs, logs...)
}
}
+ if len(coalescedLogs) > 0 {
+ go mux.Post(core.PendingLogsEvent{Logs: coalescedLogs})
+ }
}
-func (env *Work) commitTransaction(tx *types.Transaction, bc *core.BlockChain, gp *core.GasPool) error {
+func (env *Work) commitTransaction(tx *types.Transaction, bc *core.BlockChain, gp *core.GasPool) (error, vm.Logs) {
snap := env.state.Copy()
- receipt, _, _, err := core.ApplyTransaction(bc, gp, env.state, env.header, tx, env.header.GasUsed)
+ receipt, logs, _, err := core.ApplyTransaction(bc, gp, env.state, env.header, tx, env.header.GasUsed)
if err != nil {
env.state.Set(snap)
- return err
+ return err, nil
}
env.txs = append(env.txs, tx)
env.receipts = append(env.receipts, receipt)
- return nil
+
+ return nil, logs
}
// TODO: remove or use