diff --git a/accounts/abi/bind/backends/simulated.go b/accounts/abi/bind/backends/simulated.go
index b8c6814040a4..71b44b15c741 100644
--- a/accounts/abi/bind/backends/simulated.go
+++ b/accounts/abi/bind/backends/simulated.go
@@ -61,11 +61,13 @@ type SimulatedBackend struct {
database ethdb.Database // In memory database to store our testing data
blockchain *core.BlockChain // Ethereum blockchain to handle the consensus
- mu sync.Mutex
- pendingBlock *types.Block // Currently pending block that will be imported on request
- pendingState *state.StateDB // Currently pending state that will be the active on on request
+ mu sync.Mutex
+ pendingBlock *types.Block // Currently pending block that will be imported on request
+ pendingState *state.StateDB // Currently pending state that will be the active on request
+ pendingReceipts types.Receipts // Currently receipts for the pending block
- events *filters.EventSystem // Event system for filtering log events live
+ events *filters.EventSystem // for filtering log events live
+ filterSystem *filters.FilterSystem // for filtering database logs
config *params.ChainConfig
}
@@ -94,9 +96,7 @@ func SimulateWalletAddressAndSignFn() (common.Address, func(account accounts.Acc
// XDC simulated backend for testing purpose.
func NewXDCSimulatedBackend(alloc core.GenesisAlloc, gasLimit uint64, chainConfig *params.ChainConfig) *SimulatedBackend {
- // database := ethdb.NewMemDatabase()
database := rawdb.NewMemoryDatabase()
-
genesis := core.Genesis{
GasLimit: gasLimit, // need this big, support initial smart contract
Config: chainConfig,
@@ -126,8 +126,12 @@ func NewXDCSimulatedBackend(alloc core.GenesisAlloc, gasLimit uint64, chainConfi
database: database,
blockchain: blockchain,
config: genesis.Config,
- events: filters.NewEventSystem(new(event.TypeMux), &filterBackend{database, blockchain}, false),
}
+
+ filterBackend := &filterBackend{database, blockchain, backend}
+ backend.filterSystem = filters.NewFilterSystem(filterBackend, filters.Config{})
+ backend.events = filters.NewEventSystem(backend.filterSystem, false)
+
blockchain.Client = backend
backend.rollback()
return backend
@@ -146,8 +150,12 @@ func NewSimulatedBackend(alloc core.GenesisAlloc) *SimulatedBackend {
database: database,
blockchain: blockchain,
config: genesis.Config,
- events: filters.NewEventSystem(new(event.TypeMux), &filterBackend{database, blockchain}, false),
}
+
+ filterBackend := &filterBackend{database, blockchain, backend}
+ backend.filterSystem = filters.NewFilterSystem(filterBackend, filters.Config{})
+ backend.events = filters.NewEventSystem(backend.filterSystem, false)
+
backend.rollback()
return backend
}
@@ -421,7 +429,7 @@ func (b *SimulatedBackend) FilterLogs(ctx context.Context, query XDPoSChain.Filt
var filter *filters.Filter
if query.BlockHash != nil {
// Block filter requested, construct a single-shot filter
- filter = filters.NewBlockFilter(&filterBackend{b.database, b.blockchain}, *query.BlockHash, query.Addresses, query.Topics)
+ filter = b.filterSystem.NewBlockFilter(*query.BlockHash, query.Addresses, query.Topics)
} else {
// Initialize unset filter boundaried to run from genesis to chain head
from := int64(0)
@@ -433,7 +441,7 @@ func (b *SimulatedBackend) FilterLogs(ctx context.Context, query XDPoSChain.Filt
to = query.ToBlock.Int64()
}
// Construct the range filter
- filter = filters.NewRangeFilter(&filterBackend{b.database, b.blockchain}, from, to, query.Addresses, query.Topics)
+ filter = b.filterSystem.NewRangeFilter(from, to, query.Addresses, query.Topics)
}
// Run the filter and return all the logs
logs, err := filter.Logs(ctx)
@@ -523,8 +531,9 @@ func (m callMsg) AccessList() types.AccessList { return m.CallMsg.AccessList }
// filterBackend implements filters.Backend to support filtering for logs without
// taking bloom-bits acceleration structures into account.
type filterBackend struct {
- db ethdb.Database
- bc *core.BlockChain
+ db ethdb.Database
+ bc *core.BlockChain
+ backend *SimulatedBackend
}
func (fb *filterBackend) ChainDb() ethdb.Database { return fb.db }
@@ -545,35 +554,44 @@ func (fb *filterBackend) GetReceipts(ctx context.Context, hash common.Hash) (typ
return core.GetBlockReceipts(fb.db, hash, core.GetBlockNumber(fb.db, hash)), nil
}
-func (fb *filterBackend) GetLogs(ctx context.Context, hash common.Hash) ([][]*types.Log, error) {
- receipts := core.GetBlockReceipts(fb.db, hash, core.GetBlockNumber(fb.db, hash))
- if receipts == nil {
- return nil, nil
- }
- logs := make([][]*types.Log, len(receipts))
- for i, receipt := range receipts {
- logs[i] = receipt.Logs
- }
+func (fb *filterBackend) PendingBlockAndReceipts() (*types.Block, types.Receipts) {
+ return fb.backend.pendingBlock, fb.backend.pendingReceipts
+}
+
+func (fb *filterBackend) GetLogs(ctx context.Context, hash common.Hash, number uint64) ([][]*types.Log, error) {
+ logs := rawdb.ReadLogs(fb.db, hash, number)
return logs, nil
}
func (fb *filterBackend) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription {
- return event.NewSubscription(func(quit <-chan struct{}) error {
- <-quit
- return nil
- })
+ return nullSubscription()
}
+
func (fb *filterBackend) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription {
return fb.bc.SubscribeChainEvent(ch)
}
+
func (fb *filterBackend) SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription {
return fb.bc.SubscribeRemovedLogsEvent(ch)
}
+
func (fb *filterBackend) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription {
return fb.bc.SubscribeLogsEvent(ch)
}
+func (fb *filterBackend) SubscribePendingLogsEvent(ch chan<- []*types.Log) event.Subscription {
+ return nullSubscription()
+}
+
func (fb *filterBackend) BloomStatus() (uint64, uint64) { return 4096, 0 }
+
func (fb *filterBackend) ServiceFilter(ctx context.Context, ms *bloombits.MatcherSession) {
panic("not supported")
}
+
+func nullSubscription() event.Subscription {
+ return event.NewSubscription(func(quit <-chan struct{}) error {
+ <-quit
+ return nil
+ })
+}
diff --git a/cmd/XDC/main.go b/cmd/XDC/main.go
index 3a028fe180aa..ad8c4fd13a44 100644
--- a/cmd/XDC/main.go
+++ b/cmd/XDC/main.go
@@ -93,6 +93,7 @@ var (
//utils.CacheDatabaseFlag,
//utils.CacheGCFlag,
//utils.TrieCacheGenFlag,
+ utils.CacheLogSizeFlag,
utils.ListenPortFlag,
utils.MaxPeersFlag,
utils.MaxPendingPeersFlag,
diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go
index ae845a4469a4..589053f3ceb0 100644
--- a/cmd/utils/flags.go
+++ b/cmd/utils/flags.go
@@ -40,8 +40,10 @@ import (
"github.com/XinFinOrg/XDPoSChain/crypto"
"github.com/XinFinOrg/XDPoSChain/eth/downloader"
"github.com/XinFinOrg/XDPoSChain/eth/ethconfig"
+ "github.com/XinFinOrg/XDPoSChain/eth/filters"
"github.com/XinFinOrg/XDPoSChain/eth/gasprice"
"github.com/XinFinOrg/XDPoSChain/ethdb"
+ "github.com/XinFinOrg/XDPoSChain/internal/ethapi"
"github.com/XinFinOrg/XDPoSChain/log"
"github.com/XinFinOrg/XDPoSChain/metrics"
"github.com/XinFinOrg/XDPoSChain/metrics/exp"
@@ -52,6 +54,7 @@ import (
"github.com/XinFinOrg/XDPoSChain/p2p/nat"
"github.com/XinFinOrg/XDPoSChain/p2p/netutil"
"github.com/XinFinOrg/XDPoSChain/params"
+ "github.com/XinFinOrg/XDPoSChain/rpc"
whisper "github.com/XinFinOrg/XDPoSChain/whisper/whisperv6"
"gopkg.in/urfave/cli.v1"
)
@@ -313,6 +316,11 @@ var (
Usage: "Percentage of cache memory allowance to use for trie pruning",
Value: 25,
}
+ CacheLogSizeFlag = &cli.IntFlag{
+ Name: "cache.blocklogs",
+ Usage: "Size (in number of blocks) of the log cache for filtering",
+ Value: ethconfig.Defaults.FilterLogCacheSize,
+ }
// Miner settings
StakingEnabledFlag = cli.BoolFlag{
Name: "mine",
@@ -1208,6 +1216,9 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
if ctx.GlobalIsSet(GasPriceFlag.Name) {
cfg.GasPrice = GlobalBig(ctx, GasPriceFlag.Name)
}
+ if ctx.IsSet(CacheLogSizeFlag.Name) {
+ cfg.FilterLogCacheSize = ctx.Int(CacheLogSizeFlag.Name)
+ }
if ctx.GlobalIsSet(VMEnableDebugFlag.Name) {
// TODO(fjl): force-enable this in --dev mode
cfg.EnablePreimageRecording = ctx.GlobalBool(VMEnableDebugFlag.Name)
@@ -1407,6 +1418,19 @@ func WalkMatch(root, pattern string) ([]string, error) {
return matches, nil
}
+// RegisterFilterAPI adds the eth log filtering RPC API to the node.
+func RegisterFilterAPI(stack *node.Node, backend ethapi.Backend, ethcfg *ethconfig.Config) *filters.FilterSystem {
+ isLightClient := ethcfg.SyncMode == downloader.LightSync
+ filterSystem := filters.NewFilterSystem(backend, filters.Config{
+ LogCacheSize: ethcfg.FilterLogCacheSize,
+ })
+ stack.RegisterAPIs([]rpc.API{{
+ Namespace: "eth",
+ Service: filters.NewFilterAPI(filterSystem, isLightClient),
+ }})
+ return filterSystem
+}
+
func SetupMetrics(ctx *cli.Context) {
if metrics.Enabled {
log.Info("Enabling metrics collection")
diff --git a/cmd/utils/utils.go b/cmd/utils/utils.go
index 406d15ebfc83..a329fc8c4d2e 100644
--- a/cmd/utils/utils.go
+++ b/cmd/utils/utils.go
@@ -47,8 +47,7 @@ func RegisterShhService(stack *node.Node, cfg *whisper.Config) {
}
}
-// RegisterEthStatsService configures the Ethereum Stats daemon and adds it to
-// th egiven node.
+// RegisterEthStatsService configures the Ethereum Stats daemon and adds it to the node.
func RegisterEthStatsService(stack *node.Node, url string) {
if err := stack.Register(func(ctx *node.ServiceContext) (node.Service, error) {
// Retrieve both eth and les services
diff --git a/core/chain_makers.go b/core/chain_makers.go
index 348d68d1687a..15e87c88378d 100644
--- a/core/chain_makers.go
+++ b/core/chain_makers.go
@@ -113,6 +113,15 @@ func (b *BlockGen) AddTxWithChain(bc *BlockChain, tx *types.Transaction) {
}
}
+// AddUncheckedTx forcefully adds a transaction to the block without any
+// validation.
+//
+// AddUncheckedTx will cause consensus failures when used during real
+// chain processing. This is best used in conjunction with raw block insertion.
+func (b *BlockGen) AddUncheckedTx(tx *types.Transaction) {
+ b.txs = append(b.txs, tx)
+}
+
// Number returns the block number of the block being generated.
func (b *BlockGen) Number() *big.Int {
return new(big.Int).Set(b.header.Number)
diff --git a/core/events.go b/core/events.go
index 60dc8d7ddd36..bf7e7027e5c9 100644
--- a/core/events.go
+++ b/core/events.go
@@ -30,11 +30,6 @@ type OrderTxPreEvent struct{ Tx *types.OrderTransaction }
// LendingTxPreEvent is posted when a order transaction enters the order transaction pool.
type LendingTxPreEvent struct{ Tx *types.LendingTransaction }
-// PendingLogsEvent is posted pre mining and notifies of pending logs.
-type PendingLogsEvent struct {
- Logs []*types.Log
-}
-
// PendingStateEvent is posted pre mining and notifies of pending state changes.
type PendingStateEvent struct{}
diff --git a/core/rawdb/accessors_chain.go b/core/rawdb/accessors_chain.go
new file mode 100644
index 000000000000..04b30b2a2206
--- /dev/null
+++ b/core/rawdb/accessors_chain.go
@@ -0,0 +1,280 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see .
+
+package rawdb
+
+import (
+ "bytes"
+ "encoding/binary"
+ "errors"
+
+ "github.com/XinFinOrg/XDPoSChain/common"
+ "github.com/XinFinOrg/XDPoSChain/core/types"
+ "github.com/XinFinOrg/XDPoSChain/ethdb"
+ "github.com/XinFinOrg/XDPoSChain/log"
+ "github.com/XinFinOrg/XDPoSChain/params"
+ "github.com/XinFinOrg/XDPoSChain/rlp"
+)
+
+// ReadHeaderNumber returns the header number assigned to a hash.
+func ReadHeaderNumber(db ethdb.KeyValueReader, hash common.Hash) *uint64 {
+ data, _ := db.Get(headerNumberKey(hash))
+ if len(data) != 8 {
+ return nil
+ }
+ number := binary.BigEndian.Uint64(data)
+ return &number
+}
+
+// ReadBodyRLP retrieves the block body (transactions and uncles) in RLP encoding.
+func ReadBodyRLP(db ethdb.Reader, hash common.Hash, number uint64) rlp.RawValue {
+ // First try to look up the data in ancient database. Extra hash
+ // comparison is necessary since ancient database only maintains
+ // the canonical data.
+ data, _ := db.Ancient(freezerBodiesTable, number)
+ if len(data) > 0 {
+ h, _ := db.Ancient(freezerHashTable, number)
+ if common.BytesToHash(h) == hash {
+ return data
+ }
+ }
+ // Then try to look up the data in leveldb.
+ data, _ = db.Get(blockBodyKey(number, hash))
+ if len(data) > 0 {
+ return data
+ }
+ // In the background freezer is moving data from leveldb to flatten files.
+ // So during the first check for ancient db, the data is not yet in there,
+ // but when we reach into leveldb, the data was already moved. That would
+ // result in a not found error.
+ data, _ = db.Ancient(freezerBodiesTable, number)
+ if len(data) > 0 {
+ h, _ := db.Ancient(freezerHashTable, number)
+ if common.BytesToHash(h) == hash {
+ return data
+ }
+ }
+ return nil // Can't find the data anywhere.
+}
+
+// WriteBodyRLP stores an RLP encoded block body into the database.
+func WriteBodyRLP(db ethdb.KeyValueWriter, hash common.Hash, number uint64, rlp rlp.RawValue) {
+ if err := db.Put(blockBodyKey(number, hash), rlp); err != nil {
+ log.Crit("Failed to store block body", "err", err)
+ }
+}
+
+// ReadBody retrieves the block body corresponding to the hash.
+func ReadBody(db ethdb.Reader, hash common.Hash, number uint64) *types.Body {
+ data := ReadBodyRLP(db, hash, number)
+ if len(data) == 0 {
+ return nil
+ }
+ body := new(types.Body)
+ if err := rlp.Decode(bytes.NewReader(data), body); err != nil {
+ log.Error("Invalid block body RLP", "hash", hash, "err", err)
+ return nil
+ }
+ return body
+}
+
+// WriteBody stores a block body into the database.
+func WriteBody(db ethdb.KeyValueWriter, hash common.Hash, number uint64, body *types.Body) {
+ data, err := rlp.EncodeToBytes(body)
+ if err != nil {
+ log.Crit("Failed to RLP encode body", "err", err)
+ }
+ WriteBodyRLP(db, hash, number, data)
+}
+
+// ReadReceiptsRLP retrieves all the transaction receipts belonging to a block in RLP encoding.
+func ReadReceiptsRLP(db ethdb.Reader, hash common.Hash, number uint64) rlp.RawValue {
+ // First try to look up the data in ancient database. Extra hash
+ // comparison is necessary since ancient database only maintains
+ // the canonical data.
+ data, _ := db.Ancient(freezerReceiptTable, number)
+ if len(data) > 0 {
+ h, _ := db.Ancient(freezerHashTable, number)
+ if common.BytesToHash(h) == hash {
+ return data
+ }
+ }
+ // Then try to look up the data in leveldb.
+ data, _ = db.Get(blockReceiptsKey(number, hash))
+ if len(data) > 0 {
+ return data
+ }
+ // In the background freezer is moving data from leveldb to flatten files.
+ // So during the first check for ancient db, the data is not yet in there,
+ // but when we reach into leveldb, the data was already moved. That would
+ // result in a not found error.
+ data, _ = db.Ancient(freezerReceiptTable, number)
+ if len(data) > 0 {
+ h, _ := db.Ancient(freezerHashTable, number)
+ if common.BytesToHash(h) == hash {
+ return data
+ }
+ }
+ return nil // Can't find the data anywhere.
+}
+
+// ReadRawReceipts retrieves all the transaction receipts belonging to a block.
+// The receipt metadata fields are not guaranteed to be populated, so they
+// should not be used. Use ReadReceipts instead if the metadata is needed.
+func ReadRawReceipts(db ethdb.Reader, hash common.Hash, number uint64) types.Receipts {
+ // Retrieve the flattened receipt slice
+ data := ReadReceiptsRLP(db, hash, number)
+ if len(data) == 0 {
+ return nil
+ }
+ // Convert the receipts from their storage form to their internal representation
+ storageReceipts := []*types.ReceiptForStorage{}
+ if err := rlp.DecodeBytes(data, &storageReceipts); err != nil {
+ log.Error("Invalid receipt array RLP", "hash", hash, "err", err)
+ return nil
+ }
+ receipts := make(types.Receipts, len(storageReceipts))
+ for i, storageReceipt := range storageReceipts {
+ receipts[i] = (*types.Receipt)(storageReceipt)
+ }
+ return receipts
+}
+
+// ReadReceipts retrieves all the transaction receipts belonging to a block, including
+// its correspoinding metadata fields. If it is unable to populate these metadata
+// fields then nil is returned.
+//
+// The current implementation populates these metadata fields by reading the receipts'
+// corresponding block body, so if the block body is not found it will return nil even
+// if the receipt itself is stored.
+func ReadReceipts(db ethdb.Reader, hash common.Hash, number uint64, config *params.ChainConfig) types.Receipts {
+ // We're deriving many fields from the block body, retrieve beside the receipt
+ receipts := ReadRawReceipts(db, hash, number)
+ if receipts == nil {
+ return nil
+ }
+ body := ReadBody(db, hash, number)
+ if body == nil {
+ log.Error("Missing body but have receipt", "hash", hash, "number", number)
+ return nil
+ }
+ if err := receipts.DeriveFields(config, hash, number, body.Transactions); err != nil {
+ log.Error("Failed to derive block receipts fields", "hash", hash, "number", number, "err", err)
+ return nil
+ }
+ return receipts
+}
+
+// WriteReceipts stores all the transaction receipts belonging to a block.
+func WriteReceipts(db ethdb.KeyValueWriter, hash common.Hash, number uint64, receipts types.Receipts) {
+ // Convert the receipts into their storage form and serialize them
+ storageReceipts := make([]*types.ReceiptForStorage, len(receipts))
+ for i, receipt := range receipts {
+ storageReceipts[i] = (*types.ReceiptForStorage)(receipt)
+ }
+ bytes, err := rlp.EncodeToBytes(storageReceipts)
+ if err != nil {
+ log.Crit("Failed to encode block receipts", "err", err)
+ }
+ // Store the flattened receipt slice
+ if err := db.Put(blockReceiptsKey(number, hash), bytes); err != nil {
+ log.Crit("Failed to store block receipts", "err", err)
+ }
+}
+
+// storedReceiptRLP is the storage encoding of a receipt.
+// Re-definition in core/types/receipt.go.
+type storedReceiptRLP struct {
+ PostStateOrStatus []byte
+ CumulativeGasUsed uint64
+ Bloom types.Bloom
+ TxHash common.Hash
+ ContractAddress common.Address
+ Logs []*types.LogForStorage
+ GasUsed uint64
+}
+
+// ReceiptLogs is a barebone version of ReceiptForStorage which only keeps
+// the list of logs. When decoding a stored receipt into this object we
+// avoid creating the bloom filter.
+type receiptLogs struct {
+ Logs []*types.Log
+}
+
+// DecodeRLP implements rlp.Decoder.
+func (r *receiptLogs) DecodeRLP(s *rlp.Stream) error {
+ var stored storedReceiptRLP
+ if err := s.Decode(&stored); err != nil {
+ return err
+ }
+ r.Logs = make([]*types.Log, len(stored.Logs))
+ for i, log := range stored.Logs {
+ r.Logs[i] = (*types.Log)(log)
+ }
+ return nil
+}
+
+// DeriveLogFields fills the logs in receiptLogs with information such as block number, txhash, etc.
+func deriveLogFields(receipts []*receiptLogs, hash common.Hash, number uint64, txs types.Transactions) error {
+ logIndex := uint(0)
+ if len(txs) != len(receipts) {
+ return errors.New("transaction and receipt count mismatch")
+ }
+ for i := 0; i < len(receipts); i++ {
+ txHash := txs[i].Hash()
+ // The derived log fields can simply be set from the block and transaction
+ for j := 0; j < len(receipts[i].Logs); j++ {
+ receipts[i].Logs[j].BlockNumber = number
+ receipts[i].Logs[j].BlockHash = hash
+ receipts[i].Logs[j].TxHash = txHash
+ receipts[i].Logs[j].TxIndex = uint(i)
+ receipts[i].Logs[j].Index = logIndex
+ logIndex++
+ }
+ }
+ return nil
+}
+
+// ReadLogs retrieves the logs for all transactions in a block. The log fields
+// are populated with metadata. In case the receipts or the block body
+// are not found, a nil is returned.
+func ReadLogs(db ethdb.Reader, hash common.Hash, number uint64) [][]*types.Log {
+ // Retrieve the flattened receipt slice
+ data := ReadReceiptsRLP(db, hash, number)
+ if len(data) == 0 {
+ return nil
+ }
+ receipts := []*receiptLogs{}
+ if err := rlp.DecodeBytes(data, &receipts); err != nil {
+ log.Error("Invalid receipt array RLP", "hash", hash, "err", err)
+ return nil
+ }
+
+ body := ReadBody(db, hash, number)
+ if body == nil {
+ log.Error("Missing body but have receipt", "hash", hash, "number", number)
+ return nil
+ }
+ if err := deriveLogFields(receipts, hash, number, body.Transactions); err != nil {
+ log.Error("Failed to derive block receipts fields", "hash", hash, "number", number, "err", err)
+ return nil
+ }
+ logs := make([][]*types.Log, len(receipts))
+ for i, receipt := range receipts {
+ logs[i] = receipt.Logs
+ }
+ return logs
+}
diff --git a/core/rawdb/accessors_chain_test.go b/core/rawdb/accessors_chain_test.go
new file mode 100644
index 000000000000..1b6527dbd2e1
--- /dev/null
+++ b/core/rawdb/accessors_chain_test.go
@@ -0,0 +1,243 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see .
+
+package rawdb
+
+import (
+ "bytes"
+ "encoding/hex"
+ "io/ioutil"
+ "math/big"
+ "testing"
+
+ "github.com/XinFinOrg/XDPoSChain/common"
+ "github.com/XinFinOrg/XDPoSChain/core/types"
+ "github.com/XinFinOrg/XDPoSChain/params"
+ "github.com/XinFinOrg/XDPoSChain/rlp"
+)
+
+type fullLogRLP struct {
+ Address common.Address
+ Topics []common.Hash
+ Data []byte
+ BlockNumber uint64
+ TxHash common.Hash
+ TxIndex uint
+ BlockHash common.Hash
+ Index uint
+}
+
+func newFullLogRLP(l *types.Log) *fullLogRLP {
+ return &fullLogRLP{
+ Address: l.Address,
+ Topics: l.Topics,
+ Data: l.Data,
+ BlockNumber: l.BlockNumber,
+ TxHash: l.TxHash,
+ TxIndex: l.TxIndex,
+ BlockHash: l.BlockHash,
+ Index: l.Index,
+ }
+}
+
+// Tests that logs associated with a single block can be retrieved.
+func TestReadLogs(t *testing.T) {
+ db := NewMemoryDatabase()
+
+ // Create a live block since we need metadata to reconstruct the receipt
+ tx1 := types.NewTransaction(1, common.HexToAddress("0x1"), big.NewInt(1), 1, big.NewInt(1), nil)
+ tx2 := types.NewTransaction(2, common.HexToAddress("0x2"), big.NewInt(2), 2, big.NewInt(2), nil)
+
+ body := &types.Body{Transactions: types.Transactions{tx1, tx2}}
+
+ // Create the two receipts to manage afterwards
+ receipt1 := &types.Receipt{
+ Status: types.ReceiptStatusFailed,
+ CumulativeGasUsed: 1,
+ Logs: []*types.Log{
+ {Address: common.BytesToAddress([]byte{0x11})},
+ {Address: common.BytesToAddress([]byte{0x01, 0x11})},
+ },
+ TxHash: tx1.Hash(),
+ ContractAddress: common.BytesToAddress([]byte{0x01, 0x11, 0x11}),
+ GasUsed: 111111,
+ }
+ receipt1.Bloom = types.CreateBloom(types.Receipts{receipt1})
+
+ receipt2 := &types.Receipt{
+ PostState: common.Hash{2}.Bytes(),
+ CumulativeGasUsed: 2,
+ Logs: []*types.Log{
+ {Address: common.BytesToAddress([]byte{0x22})},
+ {Address: common.BytesToAddress([]byte{0x02, 0x22})},
+ },
+ TxHash: tx2.Hash(),
+ ContractAddress: common.BytesToAddress([]byte{0x02, 0x22, 0x22}),
+ GasUsed: 222222,
+ }
+ receipt2.Bloom = types.CreateBloom(types.Receipts{receipt2})
+ receipts := []*types.Receipt{receipt1, receipt2}
+
+ hash := common.BytesToHash([]byte{0x03, 0x14})
+ // Check that no receipt entries are in a pristine database
+ if rs := ReadReceipts(db, hash, 0, params.TestChainConfig); len(rs) != 0 {
+ t.Fatalf("non existent receipts returned: %v", rs)
+ }
+ // Insert the body that corresponds to the receipts
+ WriteBody(db, hash, 0, body)
+
+ // Insert the receipt slice into the database and check presence
+ WriteReceipts(db, hash, 0, receipts)
+
+ logs := ReadLogs(db, hash, 0)
+ if len(logs) == 0 {
+ t.Fatalf("no logs returned")
+ }
+ if have, want := len(logs), 2; have != want {
+ t.Fatalf("unexpected number of logs returned, have %d want %d", have, want)
+ }
+ if have, want := len(logs[0]), 2; have != want {
+ t.Fatalf("unexpected number of logs[0] returned, have %d want %d", have, want)
+ }
+ if have, want := len(logs[1]), 2; have != want {
+ t.Fatalf("unexpected number of logs[1] returned, have %d want %d", have, want)
+ }
+
+ // Fill in log fields so we can compare their rlp encoding
+ if err := types.Receipts(receipts).DeriveFields(params.TestChainConfig, hash, 0, body.Transactions); err != nil {
+ t.Fatal(err)
+ }
+ for i, pr := range receipts {
+ for j, pl := range pr.Logs {
+ rlpHave, err := rlp.EncodeToBytes(newFullLogRLP(logs[i][j]))
+ if err != nil {
+ t.Fatal(err)
+ }
+ rlpWant, err := rlp.EncodeToBytes(newFullLogRLP(pl))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !bytes.Equal(rlpHave, rlpWant) {
+ t.Fatalf("receipt #%d: receipt mismatch: have %s, want %s", i, hex.EncodeToString(rlpHave), hex.EncodeToString(rlpWant))
+ }
+ }
+ }
+}
+
+func TestDeriveLogFields(t *testing.T) {
+ // Create a few transactions to have receipts for
+ to2 := common.HexToAddress("0x2")
+ to3 := common.HexToAddress("0x3")
+ txs := types.Transactions{
+ types.NewTx(&types.LegacyTx{
+ Nonce: 1,
+ Value: big.NewInt(1),
+ Gas: 1,
+ GasPrice: big.NewInt(1),
+ }),
+ types.NewTx(&types.LegacyTx{
+ To: &to2,
+ Nonce: 2,
+ Value: big.NewInt(2),
+ Gas: 2,
+ GasPrice: big.NewInt(2),
+ }),
+ types.NewTx(&types.AccessListTx{
+ To: &to3,
+ Nonce: 3,
+ Value: big.NewInt(3),
+ Gas: 3,
+ GasPrice: big.NewInt(3),
+ }),
+ }
+ // Create the corresponding receipts
+ receipts := []*receiptLogs{
+ {
+ Logs: []*types.Log{
+ {Address: common.BytesToAddress([]byte{0x11})},
+ {Address: common.BytesToAddress([]byte{0x01, 0x11})},
+ },
+ },
+ {
+ Logs: []*types.Log{
+ {Address: common.BytesToAddress([]byte{0x22})},
+ {Address: common.BytesToAddress([]byte{0x02, 0x22})},
+ },
+ },
+ {
+ Logs: []*types.Log{
+ {Address: common.BytesToAddress([]byte{0x33})},
+ {Address: common.BytesToAddress([]byte{0x03, 0x33})},
+ },
+ },
+ }
+
+ // Derive log metadata fields
+ number := big.NewInt(1)
+ hash := common.BytesToHash([]byte{0x03, 0x14})
+ if err := deriveLogFields(receipts, hash, number.Uint64(), txs); err != nil {
+ t.Fatal(err)
+ }
+
+ // Iterate over all the computed fields and check that they're correct
+ logIndex := uint(0)
+ for i := range receipts {
+ for j := range receipts[i].Logs {
+ if receipts[i].Logs[j].BlockNumber != number.Uint64() {
+ t.Errorf("receipts[%d].Logs[%d].BlockNumber = %d, want %d", i, j, receipts[i].Logs[j].BlockNumber, number.Uint64())
+ }
+ if receipts[i].Logs[j].BlockHash != hash {
+ t.Errorf("receipts[%d].Logs[%d].BlockHash = %s, want %s", i, j, receipts[i].Logs[j].BlockHash.String(), hash.String())
+ }
+ if receipts[i].Logs[j].TxHash != txs[i].Hash() {
+ t.Errorf("receipts[%d].Logs[%d].TxHash = %s, want %s", i, j, receipts[i].Logs[j].TxHash.String(), txs[i].Hash().String())
+ }
+ if receipts[i].Logs[j].TxIndex != uint(i) {
+ t.Errorf("receipts[%d].Logs[%d].TransactionIndex = %d, want %d", i, j, receipts[i].Logs[j].TxIndex, i)
+ }
+ if receipts[i].Logs[j].Index != logIndex {
+ t.Errorf("receipts[%d].Logs[%d].Index = %d, want %d", i, j, receipts[i].Logs[j].Index, logIndex)
+ }
+ logIndex++
+ }
+ }
+}
+
+func BenchmarkDecodeRLPLogs(b *testing.B) {
+ // Encoded receipts from block 0x14ee094309fbe8f70b65f45ebcc08fb33f126942d97464aad5eb91cfd1e2d269
+ buf, err := ioutil.ReadFile("testdata/stored_receipts.bin")
+ if err != nil {
+ b.Fatal(err)
+ }
+ b.Run("ReceiptForStorage", func(b *testing.B) {
+ b.ReportAllocs()
+ var r []*types.ReceiptForStorage
+ for i := 0; i < b.N; i++ {
+ if err := rlp.DecodeBytes(buf, &r); err != nil {
+ b.Fatal(err)
+ }
+ }
+ })
+ b.Run("rlpLogs", func(b *testing.B) {
+ b.ReportAllocs()
+ var r []*receiptLogs
+ for i := 0; i < b.N; i++ {
+ if err := rlp.DecodeBytes(buf, &r); err != nil {
+ b.Fatal(err)
+ }
+ }
+ })
+}
diff --git a/core/rawdb/schema.go b/core/rawdb/schema.go
new file mode 100644
index 000000000000..d9d567e57828
--- /dev/null
+++ b/core/rawdb/schema.go
@@ -0,0 +1,66 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see .
+
+// Package rawdb contains a collection of low level database accessors.
+package rawdb
+
+import (
+ "encoding/binary"
+
+ "github.com/XinFinOrg/XDPoSChain/common"
+)
+
+// The fields below define the low level database schema prefixing.
+var (
+ // Data item prefixes (use single byte to avoid mixing data types, avoid `i`, used for indexes).
+ headerNumberPrefix = []byte("H") // headerNumberPrefix + hash -> num (uint64 big endian)
+ blockBodyPrefix = []byte("b") // blockBodyPrefix + num (uint64 big endian) + hash -> block body
+
+ blockReceiptsPrefix = []byte("r") // blockReceiptsPrefix + num (uint64 big endian) + hash -> block receipts
+)
+
+const (
+ // freezerHashTable indicates the name of the freezer canonical hash table.
+ freezerHashTable = "hashes"
+
+ // freezerBodiesTable indicates the name of the freezer block body table.
+ freezerBodiesTable = "bodies"
+
+ // freezerReceiptTable indicates the name of the freezer receipts table.
+ freezerReceiptTable = "receipts"
+)
+
+// encodeBlockNumber encodes a block number as big endian uint64
+func encodeBlockNumber(number uint64) []byte {
+ enc := make([]byte, 8)
+ binary.BigEndian.PutUint64(enc, number)
+ return enc
+}
+
+// headerNumberKey = headerNumberPrefix + hash
+func headerNumberKey(hash common.Hash) []byte {
+ return append(headerNumberPrefix, hash.Bytes()...)
+}
+
+// blockBodyKey = blockBodyPrefix + num (uint64 big endian) + hash
+func blockBodyKey(number uint64, hash common.Hash) []byte {
+ return append(append(blockBodyPrefix, encodeBlockNumber(number)...), hash.Bytes()...)
+}
+
+// blockReceiptsKey = blockReceiptsPrefix + num (uint64 big endian) + hash
+func blockReceiptsKey(number uint64, hash common.Hash) []byte {
+ return append(append(blockReceiptsPrefix, encodeBlockNumber(number)...), hash.Bytes()...)
+}
diff --git a/core/types/receipt.go b/core/types/receipt.go
index 9600ad1d794e..76507b521064 100644
--- a/core/types/receipt.go
+++ b/core/types/receipt.go
@@ -26,6 +26,8 @@ import (
"github.com/XinFinOrg/XDPoSChain/common"
"github.com/XinFinOrg/XDPoSChain/common/hexutil"
+ "github.com/XinFinOrg/XDPoSChain/crypto"
+ "github.com/XinFinOrg/XDPoSChain/params"
"github.com/XinFinOrg/XDPoSChain/rlp"
)
@@ -328,3 +330,47 @@ func (r Receipts) GetRlp(i int) []byte {
}
return bytes
}
+
+// DeriveFields fills the receipts with their computed fields based on consensus
+// data and contextual infos like containing block and transactions.
+func (r Receipts) DeriveFields(config *params.ChainConfig, hash common.Hash, number uint64, txs Transactions) error {
+ signer := MakeSigner(config, new(big.Int).SetUint64(number))
+
+ logIndex := uint(0)
+ if len(txs) != len(r) {
+ return errors.New("transaction and receipt count mismatch")
+ }
+ for i := 0; i < len(r); i++ {
+ // The transaction type and hash can be retrieved from the transaction itself
+ r[i].Type = txs[i].Type()
+ r[i].TxHash = txs[i].Hash()
+
+ // block location fields
+ r[i].BlockHash = hash
+ r[i].BlockNumber = new(big.Int).SetUint64(number)
+ r[i].TransactionIndex = uint(i)
+
+ // The contract address can be derived from the transaction itself
+ if txs[i].To() == nil {
+ // Deriving the signer is expensive, only do if it's actually needed
+ from, _ := Sender(signer, txs[i])
+ r[i].ContractAddress = crypto.CreateAddress(from, txs[i].Nonce())
+ }
+ // The used gas can be calculated based on previous r
+ if i == 0 {
+ r[i].GasUsed = r[i].CumulativeGasUsed
+ } else {
+ r[i].GasUsed = r[i].CumulativeGasUsed - r[i-1].CumulativeGasUsed
+ }
+ // The derived log fields can simply be set from the block and transaction
+ for j := 0; j < len(r[i].Logs); j++ {
+ r[i].Logs[j].BlockNumber = number
+ r[i].Logs[j].BlockHash = hash
+ r[i].Logs[j].TxHash = r[i].TxHash
+ r[i].Logs[j].TxIndex = uint(i)
+ r[i].Logs[j].Index = logIndex
+ logIndex++
+ }
+ }
+ return nil
+}
diff --git a/core/types/receipt_test.go b/core/types/receipt_test.go
index 82fec06c9667..1ca5a2864774 100644
--- a/core/types/receipt_test.go
+++ b/core/types/receipt_test.go
@@ -18,11 +18,14 @@ package types
import (
"bytes"
+ "math"
"math/big"
"reflect"
"testing"
"github.com/XinFinOrg/XDPoSChain/common"
+ "github.com/XinFinOrg/XDPoSChain/crypto"
+ "github.com/XinFinOrg/XDPoSChain/params"
"github.com/XinFinOrg/XDPoSChain/rlp"
)
@@ -141,6 +144,169 @@ func encodeAsV3StoredReceiptRLP(want *Receipt) ([]byte, error) {
return rlp.EncodeToBytes(stored)
}
+// Tests that receipt data can be correctly derived from the contextual infos
+func TestDeriveFields(t *testing.T) {
+ // Create a few transactions to have receipts for
+ to2 := common.HexToAddress("0x2")
+ to3 := common.HexToAddress("0x3")
+ txs := Transactions{
+ NewTx(&LegacyTx{
+ Nonce: 1,
+ Value: big.NewInt(1),
+ Gas: 1,
+ GasPrice: big.NewInt(1),
+ }),
+ NewTx(&LegacyTx{
+ To: &to2,
+ Nonce: 2,
+ Value: big.NewInt(2),
+ Gas: 2,
+ GasPrice: big.NewInt(2),
+ }),
+ NewTx(&AccessListTx{
+ To: &to3,
+ Nonce: 3,
+ Value: big.NewInt(3),
+ Gas: 3,
+ GasPrice: big.NewInt(3),
+ }),
+ }
+ // Create the corresponding receipts
+ receipts := Receipts{
+ &Receipt{
+ Status: ReceiptStatusFailed,
+ CumulativeGasUsed: 1,
+ Logs: []*Log{
+ {Address: common.BytesToAddress([]byte{0x11})},
+ {Address: common.BytesToAddress([]byte{0x01, 0x11})},
+ },
+ TxHash: txs[0].Hash(),
+ ContractAddress: common.BytesToAddress([]byte{0x01, 0x11, 0x11}),
+ GasUsed: 1,
+ },
+ &Receipt{
+ PostState: common.Hash{2}.Bytes(),
+ CumulativeGasUsed: 3,
+ Logs: []*Log{
+ {Address: common.BytesToAddress([]byte{0x22})},
+ {Address: common.BytesToAddress([]byte{0x02, 0x22})},
+ },
+ TxHash: txs[1].Hash(),
+ ContractAddress: common.BytesToAddress([]byte{0x02, 0x22, 0x22}),
+ GasUsed: 2,
+ },
+ &Receipt{
+ Type: AccessListTxType,
+ PostState: common.Hash{3}.Bytes(),
+ CumulativeGasUsed: 6,
+ Logs: []*Log{
+ {Address: common.BytesToAddress([]byte{0x33})},
+ {Address: common.BytesToAddress([]byte{0x03, 0x33})},
+ },
+ TxHash: txs[2].Hash(),
+ ContractAddress: common.BytesToAddress([]byte{0x03, 0x33, 0x33}),
+ GasUsed: 3,
+ },
+ }
+ // Clear all the computed fields and re-derive them
+ number := big.NewInt(1)
+ hash := common.BytesToHash([]byte{0x03, 0x14})
+
+ clearComputedFieldsOnReceipts(t, receipts)
+ if err := receipts.DeriveFields(params.TestChainConfig, hash, number.Uint64(), txs); err != nil {
+ t.Fatalf("DeriveFields(...) = %v, want ", err)
+ }
+ // Iterate over all the computed fields and check that they're correct
+ signer := MakeSigner(params.TestChainConfig, number)
+
+ logIndex := uint(0)
+ for i := range receipts {
+ if receipts[i].Type != txs[i].Type() {
+ t.Errorf("receipts[%d].Type = %d, want %d", i, receipts[i].Type, txs[i].Type())
+ }
+ if receipts[i].TxHash != txs[i].Hash() {
+ t.Errorf("receipts[%d].TxHash = %s, want %s", i, receipts[i].TxHash.String(), txs[i].Hash().String())
+ }
+ if receipts[i].BlockHash != hash {
+ t.Errorf("receipts[%d].BlockHash = %s, want %s", i, receipts[i].BlockHash.String(), hash.String())
+ }
+ if receipts[i].BlockNumber.Cmp(number) != 0 {
+ t.Errorf("receipts[%c].BlockNumber = %s, want %s", i, receipts[i].BlockNumber.String(), number.String())
+ }
+ if receipts[i].TransactionIndex != uint(i) {
+ t.Errorf("receipts[%d].TransactionIndex = %d, want %d", i, receipts[i].TransactionIndex, i)
+ }
+ if receipts[i].GasUsed != txs[i].Gas() {
+ t.Errorf("receipts[%d].GasUsed = %d, want %d", i, receipts[i].GasUsed, txs[i].Gas())
+ }
+ if txs[i].To() != nil && receipts[i].ContractAddress != (common.Address{}) {
+ t.Errorf("receipts[%d].ContractAddress = %s, want %s", i, receipts[i].ContractAddress.String(), (common.Address{}).String())
+ }
+ from, _ := Sender(signer, txs[i])
+ contractAddress := crypto.CreateAddress(from, txs[i].Nonce())
+ if txs[i].To() == nil && receipts[i].ContractAddress != contractAddress {
+ t.Errorf("receipts[%d].ContractAddress = %s, want %s", i, receipts[i].ContractAddress.String(), contractAddress.String())
+ }
+ for j := range receipts[i].Logs {
+ if receipts[i].Logs[j].BlockNumber != number.Uint64() {
+ t.Errorf("receipts[%d].Logs[%d].BlockNumber = %d, want %d", i, j, receipts[i].Logs[j].BlockNumber, number.Uint64())
+ }
+ if receipts[i].Logs[j].BlockHash != hash {
+ t.Errorf("receipts[%d].Logs[%d].BlockHash = %s, want %s", i, j, receipts[i].Logs[j].BlockHash.String(), hash.String())
+ }
+ if receipts[i].Logs[j].TxHash != txs[i].Hash() {
+ t.Errorf("receipts[%d].Logs[%d].TxHash = %s, want %s", i, j, receipts[i].Logs[j].TxHash.String(), txs[i].Hash().String())
+ }
+ if receipts[i].Logs[j].TxIndex != uint(i) {
+ t.Errorf("receipts[%d].Logs[%d].TransactionIndex = %d, want %d", i, j, receipts[i].Logs[j].TxIndex, i)
+ }
+ if receipts[i].Logs[j].Index != logIndex {
+ t.Errorf("receipts[%d].Logs[%d].Index = %d, want %d", i, j, receipts[i].Logs[j].Index, logIndex)
+ }
+ logIndex++
+ }
+ }
+}
+
+func clearComputedFieldsOnReceipts(t *testing.T, receipts Receipts) {
+ t.Helper()
+
+ for _, receipt := range receipts {
+ clearComputedFieldsOnReceipt(t, receipt)
+ }
+}
+
+func clearComputedFieldsOnReceipt(t *testing.T, receipt *Receipt) {
+ t.Helper()
+
+ receipt.TxHash = common.Hash{}
+ receipt.BlockHash = common.Hash{}
+ receipt.BlockNumber = big.NewInt(math.MaxUint32)
+ receipt.TransactionIndex = math.MaxUint32
+ receipt.ContractAddress = common.Address{}
+ receipt.GasUsed = 0
+
+ clearComputedFieldsOnLogs(t, receipt.Logs)
+}
+
+func clearComputedFieldsOnLogs(t *testing.T, logs []*Log) {
+ t.Helper()
+
+ for _, log := range logs {
+ clearComputedFieldsOnLog(t, log)
+ }
+}
+
+func clearComputedFieldsOnLog(t *testing.T, log *Log) {
+ t.Helper()
+
+ log.BlockNumber = math.MaxUint32
+ log.BlockHash = common.Hash{}
+ log.TxHash = common.Hash{}
+ log.TxIndex = math.MaxUint32
+ log.Index = math.MaxUint32
+}
+
// TestTypedReceiptEncodingDecoding reproduces a flaw that existed in the receipt
// rlp decoder, which failed due to a shadowing error.
func TestTypedReceiptEncodingDecoding(t *testing.T) {
diff --git a/eth/api_backend.go b/eth/api_backend.go
index 161ee5ddfb4f..b2aaf2fbe8b0 100644
--- a/eth/api_backend.go
+++ b/eth/api_backend.go
@@ -40,6 +40,7 @@ import (
"github.com/XinFinOrg/XDPoSChain/contracts"
"github.com/XinFinOrg/XDPoSChain/core"
"github.com/XinFinOrg/XDPoSChain/core/bloombits"
+ "github.com/XinFinOrg/XDPoSChain/core/rawdb"
"github.com/XinFinOrg/XDPoSChain/core/state"
stateDatabase "github.com/XinFinOrg/XDPoSChain/core/state"
"github.com/XinFinOrg/XDPoSChain/core/types"
@@ -180,6 +181,10 @@ func (b *EthApiBackend) BlockByNumberOrHash(ctx context.Context, blockNrOrHash r
return nil, errors.New("invalid arguments; neither block nor hash specified")
}
+func (b *EthApiBackend) PendingBlockAndReceipts() (*types.Block, types.Receipts) {
+ return b.eth.miner.PendingBlockAndReceipts()
+}
+
func (b *EthApiBackend) StateAndHeaderByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*state.StateDB, *types.Header, error) {
// Pending state is only known by the miner
if blockNr == rpc.PendingBlockNumber {
@@ -229,16 +234,8 @@ func (b *EthApiBackend) GetReceipts(ctx context.Context, blockHash common.Hash)
return core.GetBlockReceipts(b.eth.chainDb, blockHash, core.GetBlockNumber(b.eth.chainDb, blockHash)), nil
}
-func (b *EthApiBackend) GetLogs(ctx context.Context, blockHash common.Hash) ([][]*types.Log, error) {
- receipts := core.GetBlockReceipts(b.eth.chainDb, blockHash, core.GetBlockNumber(b.eth.chainDb, blockHash))
- if receipts == nil {
- return nil, nil
- }
- logs := make([][]*types.Log, len(receipts))
- for i, receipt := range receipts {
- logs[i] = receipt.Logs
- }
- return logs, nil
+func (b *EthApiBackend) GetLogs(ctx context.Context, hash common.Hash, number uint64) ([][]*types.Log, error) {
+ return rawdb.ReadLogs(b.eth.chainDb, hash, number), nil
}
func (b *EthApiBackend) GetTd(blockHash common.Hash) *big.Int {
@@ -259,6 +256,10 @@ func (b *EthApiBackend) SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEven
return b.eth.BlockChain().SubscribeRemovedLogsEvent(ch)
}
+func (b *EthApiBackend) SubscribePendingLogsEvent(ch chan<- []*types.Log) event.Subscription {
+ return b.eth.miner.SubscribePendingLogs(ch)
+}
+
func (b *EthApiBackend) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription {
return b.eth.BlockChain().SubscribeChainEvent(ch)
}
diff --git a/eth/backend.go b/eth/backend.go
index 253d280e046a..08c9ec33fa6e 100644
--- a/eth/backend.go
+++ b/eth/backend.go
@@ -400,7 +400,7 @@ func (s *Ethereum) APIs() []rpc.API {
}, {
Namespace: "eth",
Version: "1.0",
- Service: filters.NewPublicFilterAPI(s.ApiBackend, false),
+ Service: filters.NewFilterAPI(filters.NewFilterSystem(s.ApiBackend, filters.Config{LogCacheSize: s.config.FilterLogCacheSize}), true),
Public: true,
}, {
Namespace: "admin",
diff --git a/eth/ethconfig/config.go b/eth/ethconfig/config.go
index b246a9b332a6..73cc4e7cddbe 100644
--- a/eth/ethconfig/config.go
+++ b/eth/ethconfig/config.go
@@ -60,12 +60,13 @@ var Defaults = Config{
DatasetsInMem: 1,
DatasetsOnDisk: 2,
},
- NetworkId: 88,
- LightPeers: 100,
- DatabaseCache: 768,
- TrieCache: 256,
- TrieTimeout: 5 * time.Minute,
- GasPrice: big.NewInt(0.25 * params.Shannon),
+ NetworkId: 88,
+ LightPeers: 100,
+ DatabaseCache: 768,
+ TrieCache: 256,
+ TrieTimeout: 5 * time.Minute,
+ FilterLogCacheSize: 32,
+ GasPrice: big.NewInt(0.25 * params.Shannon),
TxPool: core.DefaultTxPoolConfig,
RPCGasCap: 25000000,
@@ -111,6 +112,9 @@ type Config struct {
TrieCache int
TrieTimeout time.Duration
+ // This is the number of blocks for which logs will be cached in the filter system.
+ FilterLogCacheSize int
+
// Mining-related options
Etherbase common.Address `toml:",omitempty"`
MinerThreads int `toml:",omitempty"`
diff --git a/eth/ethconfig/gen_config.go b/eth/ethconfig/gen_config.go
index d24b3251c05b..6b27542f1933 100644
--- a/eth/ethconfig/gen_config.go
+++ b/eth/ethconfig/gen_config.go
@@ -31,6 +31,7 @@ func (c Config) MarshalTOML() (interface{}, error) {
MinerThreads int `toml:",omitempty"`
ExtraData []byte `toml:",omitempty"`
GasPrice *big.Int
+ FilterLogCacheSize int
Ethash ethash.Config
TxPool core.TxPoolConfig
GPO gasprice.Config
@@ -55,6 +56,7 @@ func (c Config) MarshalTOML() (interface{}, error) {
enc.MinerThreads = c.MinerThreads
enc.ExtraData = c.ExtraData
enc.GasPrice = c.GasPrice
+ enc.FilterLogCacheSize = c.FilterLogCacheSize
enc.Ethash = c.Ethash
enc.TxPool = c.TxPool
enc.GPO = c.GPO
@@ -83,6 +85,7 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error {
MinerThreads *int `toml:",omitempty"`
ExtraData []byte `toml:",omitempty"`
GasPrice *big.Int
+ FilterLogCacheSize *int
Ethash *ethash.Config
TxPool *core.TxPoolConfig
GPO *gasprice.Config
@@ -140,6 +143,9 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error {
if dec.GasPrice != nil {
c.GasPrice = dec.GasPrice
}
+ if dec.FilterLogCacheSize != nil {
+ c.FilterLogCacheSize = *dec.FilterLogCacheSize
+ }
if dec.Ethash != nil {
c.Ethash = *dec.Ethash
}
diff --git a/eth/filters/api.go b/eth/filters/api.go
index 04256a109620..6196eb67cc18 100644
--- a/eth/filters/api.go
+++ b/eth/filters/api.go
@@ -30,7 +30,6 @@ import (
"github.com/XinFinOrg/XDPoSChain/common/hexutil"
"github.com/XinFinOrg/XDPoSChain/core/types"
"github.com/XinFinOrg/XDPoSChain/ethdb"
- "github.com/XinFinOrg/XDPoSChain/event"
"github.com/XinFinOrg/XDPoSChain/rpc"
)
@@ -41,10 +40,6 @@ var (
// The maximum number of topic criteria allowed, vm.LOG4 - vm.LOG0
const maxTopics = 4
-var (
- deadline = 5 * time.Minute // consider a filter inactive if it has not been polled for within deadline
-)
-
// filter is a helper struct that holds meta information over the filter type
// and associated subscription in the event system.
type filter struct {
@@ -56,49 +51,57 @@ type filter struct {
s *Subscription // associated subscription in event system
}
-// PublicFilterAPI offers support to create and manage filters. This will allow external clients to retrieve various
+// FilterAPI offers support to create and manage filters. This will allow external clients to retrieve various
// information related to the Ethereum protocol such als blocks, transactions and logs.
-type PublicFilterAPI struct {
- backend Backend
- mux *event.TypeMux
- quit chan struct{}
+type FilterAPI struct {
+ sys *FilterSystem
chainDb ethdb.Database
events *EventSystem
filtersMu sync.Mutex
filters map[rpc.ID]*filter
+ timeout time.Duration
}
-// NewPublicFilterAPI returns a new PublicFilterAPI instance.
-func NewPublicFilterAPI(backend Backend, lightMode bool) *PublicFilterAPI {
- api := &PublicFilterAPI{
- backend: backend,
- mux: backend.EventMux(),
- chainDb: backend.ChainDb(),
- events: NewEventSystem(backend.EventMux(), backend, lightMode),
+// NewFilterAPI returns a new FilterAPI instance.
+func NewFilterAPI(system *FilterSystem, lightMode bool) *FilterAPI {
+ api := &FilterAPI{
+ sys: system,
+ chainDb: system.backend.ChainDb(),
+ events: NewEventSystem(system, lightMode),
filters: make(map[rpc.ID]*filter),
+ timeout: system.cfg.Timeout,
}
- go api.timeoutLoop()
+ go api.timeoutLoop(system.cfg.Timeout)
return api
}
// timeoutLoop runs every 5 minutes and deletes filters that have not been recently used.
// Tt is started when the api is created.
-func (api *PublicFilterAPI) timeoutLoop() {
- ticker := time.NewTicker(5 * time.Minute)
+func (api *FilterAPI) timeoutLoop(timeout time.Duration) {
+ var toUninstall []*Subscription
+ ticker := time.NewTicker(timeout)
for {
<-ticker.C
api.filtersMu.Lock()
for id, f := range api.filters {
select {
case <-f.deadline.C:
- f.s.Unsubscribe()
+ toUninstall = append(toUninstall, f.s)
delete(api.filters, id)
default:
continue
}
}
api.filtersMu.Unlock()
+
+ // Unsubscribes are processed outside the lock to avoid the following scenario:
+ // event loop attempts broadcasting events to still active filters while
+ // Unsubscribe is waiting for it to process the uninstall request.
+ for _, s := range toUninstall {
+ s.Unsubscribe()
+ }
+ toUninstall = nil
}
}
@@ -109,14 +112,14 @@ func (api *PublicFilterAPI) timeoutLoop() {
// `eth_getFilterChanges` polling method that is also used for log filters.
//
// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_newpendingtransactionfilter
-func (api *PublicFilterAPI) NewPendingTransactionFilter() rpc.ID {
+func (api *FilterAPI) NewPendingTransactionFilter() rpc.ID {
var (
pendingTxs = make(chan []common.Hash)
pendingTxSub = api.events.SubscribePendingTxs(pendingTxs)
)
api.filtersMu.Lock()
- api.filters[pendingTxSub.ID] = &filter{typ: PendingTransactionsSubscription, deadline: time.NewTimer(deadline), hashes: make([]common.Hash, 0), s: pendingTxSub}
+ api.filters[pendingTxSub.ID] = &filter{typ: PendingTransactionsSubscription, deadline: time.NewTimer(api.timeout), hashes: make([]common.Hash, 0), s: pendingTxSub}
api.filtersMu.Unlock()
go func() {
@@ -142,7 +145,7 @@ func (api *PublicFilterAPI) NewPendingTransactionFilter() rpc.ID {
// NewPendingTransactions creates a subscription that is triggered each time a transaction
// enters the transaction pool and was signed from one of the transactions this nodes manages.
-func (api *PublicFilterAPI) NewPendingTransactions(ctx context.Context) (*rpc.Subscription, error) {
+func (api *FilterAPI) NewPendingTransactions(ctx context.Context) (*rpc.Subscription, error) {
notifier, supported := rpc.NotifierFromContext(ctx)
if !supported {
return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
@@ -179,14 +182,14 @@ func (api *PublicFilterAPI) NewPendingTransactions(ctx context.Context) (*rpc.Su
// It is part of the filter package since polling goes with eth_getFilterChanges.
//
// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_newblockfilter
-func (api *PublicFilterAPI) NewBlockFilter() rpc.ID {
+func (api *FilterAPI) NewBlockFilter() rpc.ID {
var (
headers = make(chan *types.Header)
headerSub = api.events.SubscribeNewHeads(headers)
)
api.filtersMu.Lock()
- api.filters[headerSub.ID] = &filter{typ: BlocksSubscription, deadline: time.NewTimer(deadline), hashes: make([]common.Hash, 0), s: headerSub}
+ api.filters[headerSub.ID] = &filter{typ: BlocksSubscription, deadline: time.NewTimer(api.timeout), hashes: make([]common.Hash, 0), s: headerSub}
api.filtersMu.Unlock()
go func() {
@@ -211,7 +214,7 @@ func (api *PublicFilterAPI) NewBlockFilter() rpc.ID {
}
// NewHeads send a notification each time a new (header) block is appended to the chain.
-func (api *PublicFilterAPI) NewHeads(ctx context.Context) (*rpc.Subscription, error) {
+func (api *FilterAPI) NewHeads(ctx context.Context) (*rpc.Subscription, error) {
notifier, supported := rpc.NotifierFromContext(ctx)
if !supported {
return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
@@ -241,7 +244,7 @@ func (api *PublicFilterAPI) NewHeads(ctx context.Context) (*rpc.Subscription, er
}
// Logs creates a subscription that fires for all new log that match the given filter criteria.
-func (api *PublicFilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc.Subscription, error) {
+func (api *FilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc.Subscription, error) {
notifier, supported := rpc.NotifierFromContext(ctx)
if !supported {
return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
@@ -295,7 +298,7 @@ type FilterCriteria ethereum.FilterQuery
// In case "fromBlock" > "toBlock" an error is returned.
//
// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_newfilter
-func (api *PublicFilterAPI) NewFilter(crit FilterCriteria) (rpc.ID, error) {
+func (api *FilterAPI) NewFilter(crit FilterCriteria) (rpc.ID, error) {
logs := make(chan []*types.Log)
logsSub, err := api.events.SubscribeLogs(ethereum.FilterQuery(crit), logs)
if err != nil {
@@ -303,7 +306,7 @@ func (api *PublicFilterAPI) NewFilter(crit FilterCriteria) (rpc.ID, error) {
}
api.filtersMu.Lock()
- api.filters[logsSub.ID] = &filter{typ: LogsSubscription, crit: crit, deadline: time.NewTimer(deadline), logs: make([]*types.Log, 0), s: logsSub}
+ api.filters[logsSub.ID] = &filter{typ: LogsSubscription, crit: crit, deadline: time.NewTimer(api.timeout), logs: make([]*types.Log, 0), s: logsSub}
api.filtersMu.Unlock()
go func() {
@@ -330,7 +333,7 @@ func (api *PublicFilterAPI) NewFilter(crit FilterCriteria) (rpc.ID, error) {
// GetLogs returns logs matching the given argument that are stored within the state.
//
// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getlogs
-func (api *PublicFilterAPI) GetLogs(ctx context.Context, crit FilterCriteria) ([]*types.Log, error) {
+func (api *FilterAPI) GetLogs(ctx context.Context, crit FilterCriteria) ([]*types.Log, error) {
if len(crit.Topics) > maxTopics {
return nil, errExceedMaxTopics
}
@@ -338,7 +341,7 @@ func (api *PublicFilterAPI) GetLogs(ctx context.Context, crit FilterCriteria) ([
var filter *Filter
if crit.BlockHash != nil {
// Block filter requested, construct a single-shot filter
- filter = NewBlockFilter(api.backend, *crit.BlockHash, crit.Addresses, crit.Topics)
+ filter = api.sys.NewBlockFilter(*crit.BlockHash, crit.Addresses, crit.Topics)
} else {
// Convert the RPC block numbers into internal representations
begin := rpc.LatestBlockNumber.Int64()
@@ -350,7 +353,7 @@ func (api *PublicFilterAPI) GetLogs(ctx context.Context, crit FilterCriteria) ([
end = crit.ToBlock.Int64()
}
// Construct the range filter
- filter = NewRangeFilter(api.backend, begin, end, crit.Addresses, crit.Topics)
+ filter = api.sys.NewRangeFilter(begin, end, crit.Addresses, crit.Topics)
}
// Run the filter and return all the logs
logs, err := filter.Logs(ctx)
@@ -363,7 +366,7 @@ func (api *PublicFilterAPI) GetLogs(ctx context.Context, crit FilterCriteria) ([
// UninstallFilter removes the filter with the given filter id.
//
// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_uninstallfilter
-func (api *PublicFilterAPI) UninstallFilter(id rpc.ID) bool {
+func (api *FilterAPI) UninstallFilter(id rpc.ID) bool {
api.filtersMu.Lock()
f, found := api.filters[id]
if found {
@@ -381,7 +384,7 @@ func (api *PublicFilterAPI) UninstallFilter(id rpc.ID) bool {
// If the filter could not be found an empty array of logs is returned.
//
// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getfilterlogs
-func (api *PublicFilterAPI) GetFilterLogs(ctx context.Context, id rpc.ID) ([]*types.Log, error) {
+func (api *FilterAPI) GetFilterLogs(ctx context.Context, id rpc.ID) ([]*types.Log, error) {
api.filtersMu.Lock()
f, found := api.filters[id]
api.filtersMu.Unlock()
@@ -393,7 +396,7 @@ func (api *PublicFilterAPI) GetFilterLogs(ctx context.Context, id rpc.ID) ([]*ty
var filter *Filter
if f.crit.BlockHash != nil {
// Block filter requested, construct a single-shot filter
- filter = NewBlockFilter(api.backend, *f.crit.BlockHash, f.crit.Addresses, f.crit.Topics)
+ filter = api.sys.NewBlockFilter(*f.crit.BlockHash, f.crit.Addresses, f.crit.Topics)
} else {
// Convert the RPC block numbers into internal representations
begin := rpc.LatestBlockNumber.Int64()
@@ -405,7 +408,7 @@ func (api *PublicFilterAPI) GetFilterLogs(ctx context.Context, id rpc.ID) ([]*ty
end = f.crit.ToBlock.Int64()
}
// Construct the range filter
- filter = NewRangeFilter(api.backend, begin, end, f.crit.Addresses, f.crit.Topics)
+ filter = api.sys.NewRangeFilter(begin, end, f.crit.Addresses, f.crit.Topics)
}
// Run the filter and return all the logs
logs, err := filter.Logs(ctx)
@@ -422,7 +425,7 @@ func (api *PublicFilterAPI) GetFilterLogs(ctx context.Context, id rpc.ID) ([]*ty
// (pending)Log filters return []Log.
//
// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getfilterchanges
-func (api *PublicFilterAPI) GetFilterChanges(id rpc.ID) (interface{}, error) {
+func (api *FilterAPI) GetFilterChanges(id rpc.ID) (interface{}, error) {
api.filtersMu.Lock()
defer api.filtersMu.Unlock()
@@ -432,14 +435,14 @@ func (api *PublicFilterAPI) GetFilterChanges(id rpc.ID) (interface{}, error) {
// receive timer value and reset timer
<-f.deadline.C
}
- f.deadline.Reset(deadline)
+ f.deadline.Reset(api.timeout)
switch f.typ {
case PendingTransactionsSubscription, BlocksSubscription:
hashes := f.hashes
f.hashes = nil
return returnHashes(hashes), nil
- case LogsSubscription:
+ case LogsSubscription, MinedAndPendingLogsSubscription:
logs := f.logs
f.logs = nil
return returnLogs(logs), nil
diff --git a/eth/filters/bench_test.go b/eth/filters/bench_test.go
index 98635b9de7b9..f465d67f7069 100644
--- a/eth/filters/bench_test.go
+++ b/eth/filters/bench_test.go
@@ -30,7 +30,6 @@ import (
"github.com/XinFinOrg/XDPoSChain/core/rawdb"
"github.com/XinFinOrg/XDPoSChain/core/types"
"github.com/XinFinOrg/XDPoSChain/ethdb"
- "github.com/XinFinOrg/XDPoSChain/event"
"github.com/XinFinOrg/XDPoSChain/node"
)
@@ -124,21 +123,25 @@ func benchmarkBloomBits(b *testing.B, sectionSize uint64) {
b.Log("Running filter benchmarks...")
start = time.Now()
- mux := new(event.TypeMux)
- var backend *testBackend
+
+ var (
+ backend *testBackend
+ sys *FilterSystem
+ )
for i := 0; i < benchFilterCnt; i++ {
if i%20 == 0 {
db.Close()
db, _ = rawdb.NewLevelDBDatabase(benchDataDir, 128, 1024, "")
- backend = &testBackend{mux, db, cnt, new(event.Feed), new(event.Feed), new(event.Feed), new(event.Feed)}
+ backend = &testBackend{db: db, sections: cnt}
+ sys = NewFilterSystem(backend, Config{})
}
var addr common.Address
addr[0] = byte(i)
addr[1] = byte(i / 256)
- filter := NewRangeFilter(backend, 0, int64(cnt*sectionSize-1), []common.Address{addr}, nil)
+ filter := sys.NewRangeFilter(0, int64(cnt*sectionSize-1), []common.Address{addr}, nil)
if _, err := filter.Logs(context.Background()); err != nil {
- b.Error("filter.Find error:", err)
+ b.Error("filter.Logs error:", err)
}
}
d = time.Since(start)
@@ -188,11 +191,11 @@ func BenchmarkNoBloomBits(b *testing.B) {
clearBloomBits(db)
+ _, sys := newTestFilterSystem(b, db, Config{})
+
b.Log("Running filter benchmarks...")
start := time.Now()
- mux := new(event.TypeMux)
- backend := &testBackend{mux, db, 0, new(event.Feed), new(event.Feed), new(event.Feed), new(event.Feed)}
- filter := NewRangeFilter(backend, 0, int64(headNum), []common.Address{{}}, nil)
+ filter := sys.NewRangeFilter(0, int64(headNum), []common.Address{{}}, nil)
filter.Logs(context.Background())
d := time.Since(start)
b.Log("Finished running filter benchmarks")
diff --git a/eth/filters/filter.go b/eth/filters/filter.go
index 4d47bafc0fc7..1fa21bb7e0d9 100644
--- a/eth/filters/filter.go
+++ b/eth/filters/filter.go
@@ -22,36 +22,15 @@ import (
"math/big"
"github.com/XinFinOrg/XDPoSChain/common"
- "github.com/XinFinOrg/XDPoSChain/core"
"github.com/XinFinOrg/XDPoSChain/core/bloombits"
"github.com/XinFinOrg/XDPoSChain/core/types"
- "github.com/XinFinOrg/XDPoSChain/ethdb"
- "github.com/XinFinOrg/XDPoSChain/event"
"github.com/XinFinOrg/XDPoSChain/rpc"
)
-type Backend interface {
- ChainDb() ethdb.Database
- EventMux() *event.TypeMux
- HeaderByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*types.Header, error)
- HeaderByHash(ctx context.Context, blockHash common.Hash) (*types.Header, error)
- GetReceipts(ctx context.Context, blockHash common.Hash) (types.Receipts, error)
- GetLogs(ctx context.Context, blockHash common.Hash) ([][]*types.Log, error)
-
- SubscribeNewTxsEvent(chan<- core.NewTxsEvent) event.Subscription
- SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription
- SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription
- SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription
-
- BloomStatus() (uint64, uint64)
- ServiceFilter(ctx context.Context, session *bloombits.MatcherSession)
-}
-
// Filter can be used to retrieve and filter logs.
type Filter struct {
- backend Backend
+ sys *FilterSystem
- db ethdb.Database
addresses []common.Address
topics [][]common.Hash
@@ -63,7 +42,7 @@ type Filter struct {
// NewRangeFilter creates a new filter which uses a bloom filter on blocks to
// figure out whether a particular block is interesting or not.
-func NewRangeFilter(backend Backend, begin, end int64, addresses []common.Address, topics [][]common.Hash) *Filter {
+func (sys *FilterSystem) NewRangeFilter(begin, end int64, addresses []common.Address, topics [][]common.Hash) *Filter {
// Flatten the address and topic filter clauses into a single bloombits filter
// system. Since the bloombits are not positional, nil topics are permitted,
// which get flattened into a nil byte slice.
@@ -82,10 +61,10 @@ func NewRangeFilter(backend Backend, begin, end int64, addresses []common.Addres
}
filters = append(filters, filter)
}
- size, _ := backend.BloomStatus()
+ size, _ := sys.backend.BloomStatus()
// Create a generic filter and convert it into a range filter
- filter := newFilter(backend, addresses, topics)
+ filter := newFilter(sys, addresses, topics)
filter.matcher = bloombits.NewMatcher(size, filters)
filter.begin = begin
@@ -96,21 +75,20 @@ func NewRangeFilter(backend Backend, begin, end int64, addresses []common.Addres
// NewBlockFilter creates a new filter which directly inspects the contents of
// a block to figure out whether it is interesting or not.
-func NewBlockFilter(backend Backend, block common.Hash, addresses []common.Address, topics [][]common.Hash) *Filter {
+func (sys *FilterSystem) NewBlockFilter(block common.Hash, addresses []common.Address, topics [][]common.Hash) *Filter {
// Create a generic filter and convert it into a block filter
- filter := newFilter(backend, addresses, topics)
+ filter := newFilter(sys, addresses, topics)
filter.block = block
return filter
}
// newFilter creates a generic filter that can either filter based on a block hash,
// or based on range queries. The search criteria needs to be explicitly set.
-func newFilter(backend Backend, addresses []common.Address, topics [][]common.Hash) *Filter {
+func newFilter(sys *FilterSystem, addresses []common.Address, topics [][]common.Hash) *Filter {
return &Filter{
- backend: backend,
+ sys: sys,
addresses: addresses,
topics: topics,
- db: backend.ChainDb(),
}
}
@@ -119,35 +97,44 @@ func newFilter(backend Backend, addresses []common.Address, topics [][]common.Ha
func (f *Filter) Logs(ctx context.Context) ([]*types.Log, error) {
// If we're doing singleton block filtering, execute and return
if f.block != (common.Hash{}) {
- header, err := f.backend.HeaderByHash(ctx, f.block)
+ header, err := f.sys.backend.HeaderByHash(ctx, f.block)
if err != nil {
return nil, err
}
if header == nil {
return nil, errors.New("unknown block")
}
- return f.blockLogs(ctx, header)
+ return f.blockLogs(ctx, header, false)
+ }
+ // Short-cut if all we care about is pending logs
+ if f.begin == rpc.PendingBlockNumber.Int64() {
+ if f.end != rpc.PendingBlockNumber.Int64() {
+ return nil, errors.New("invalid block range")
+ }
+ return f.pendingLogs()
}
// Figure out the limits of the filter range
- header, _ := f.backend.HeaderByNumber(ctx, rpc.LatestBlockNumber)
+ header, _ := f.sys.backend.HeaderByNumber(ctx, rpc.LatestBlockNumber)
if header == nil {
return nil, nil
}
- head := header.Number.Uint64()
-
- if f.begin == -1 {
+ var (
+ head = header.Number.Uint64()
+ end = uint64(f.end)
+ pending = f.end == rpc.PendingBlockNumber.Int64()
+ )
+ if f.begin == rpc.LatestBlockNumber.Int64() {
f.begin = int64(head)
}
- end := uint64(f.end)
- if f.end == -1 {
+ if f.end == rpc.LatestBlockNumber.Int64() || f.end == rpc.PendingBlockNumber.Int64() {
end = head
}
// Gather all indexed logs, and finish with non indexed ones
var (
- logs []*types.Log
- err error
+ logs []*types.Log
+ err error
+ size, sections = f.sys.backend.BloomStatus()
)
- size, sections := f.backend.BloomStatus()
if indexed := sections * size; indexed > uint64(f.begin) {
if indexed > end {
logs, err = f.indexedLogs(ctx, end)
@@ -160,6 +147,13 @@ func (f *Filter) Logs(ctx context.Context) ([]*types.Log, error) {
}
rest, err := f.unindexedLogs(ctx, end)
logs = append(logs, rest...)
+ if pending {
+ pendingLogs, err := f.pendingLogs()
+ if err != nil {
+ return nil, err
+ }
+ logs = append(logs, pendingLogs...)
+ }
return logs, err
}
@@ -175,7 +169,7 @@ func (f *Filter) indexedLogs(ctx context.Context, end uint64) ([]*types.Log, err
}
defer session.Close()
- f.backend.ServiceFilter(ctx, session)
+ f.sys.backend.ServiceFilter(ctx, session)
// Iterate over the matches until exhausted or context closed
var logs []*types.Log
@@ -194,11 +188,11 @@ func (f *Filter) indexedLogs(ctx context.Context, end uint64) ([]*types.Log, err
f.begin = int64(number) + 1
// Retrieve the suggested block and pull any truly matching logs
- header, err := f.backend.HeaderByNumber(ctx, rpc.BlockNumber(number))
+ header, err := f.sys.backend.HeaderByNumber(ctx, rpc.BlockNumber(number))
if header == nil || err != nil {
return logs, err
}
- found, err := f.checkMatches(ctx, header)
+ found, err := f.blockLogs(ctx, header, true)
if err != nil {
return logs, err
}
@@ -216,11 +210,11 @@ func (f *Filter) unindexedLogs(ctx context.Context, end uint64) ([]*types.Log, e
var logs []*types.Log
for ; f.begin <= int64(end); f.begin++ {
- header, err := f.backend.HeaderByNumber(ctx, rpc.BlockNumber(f.begin))
+ header, err := f.sys.backend.HeaderByNumber(ctx, rpc.BlockNumber(f.begin))
if header == nil || err != nil {
return logs, err
}
- found, err := f.blockLogs(ctx, header)
+ found, err := f.blockLogs(ctx, header, false)
if err != nil {
return logs, err
}
@@ -230,34 +224,34 @@ func (f *Filter) unindexedLogs(ctx context.Context, end uint64) ([]*types.Log, e
}
// blockLogs returns the logs matching the filter criteria within a single block.
-func (f *Filter) blockLogs(ctx context.Context, header *types.Header) (logs []*types.Log, err error) {
- if bloomFilter(header.Bloom, f.addresses, f.topics) {
- found, err := f.checkMatches(ctx, header)
+func (f *Filter) blockLogs(ctx context.Context, header *types.Header, skipBloom bool) ([]*types.Log, error) {
+ // Fast track: no filtering criteria
+ if len(f.addresses) == 0 && len(f.topics) == 0 {
+ list, err := f.sys.cachedGetLogs(ctx, header.Hash(), header.Number.Uint64())
if err != nil {
- return logs, err
+ return nil, err
}
- logs = append(logs, found...)
+ return flatten(list), nil
+ } else if skipBloom || bloomFilter(header.Bloom, f.addresses, f.topics) {
+ return f.checkMatches(ctx, header)
}
- return logs, nil
+ return nil, nil
}
// checkMatches checks if the receipts belonging to the given header contain any log events that
// match the filter criteria. This function is called when the bloom filter signals a potential match.
-func (f *Filter) checkMatches(ctx context.Context, header *types.Header) (logs []*types.Log, err error) {
- // Get the logs of the block
- logsList, err := f.backend.GetLogs(ctx, header.Hash())
+func (f *Filter) checkMatches(ctx context.Context, header *types.Header) ([]*types.Log, error) {
+ logsList, err := f.sys.cachedGetLogs(ctx, header.Hash(), header.Number.Uint64())
if err != nil {
return nil, err
}
- var unfiltered []*types.Log
- for _, logs := range logsList {
- unfiltered = append(unfiltered, logs...)
- }
- logs = filterLogs(unfiltered, nil, nil, f.addresses, f.topics)
+
+ unfiltered := flatten(logsList)
+ logs := filterLogs(unfiltered, nil, nil, f.addresses, f.topics)
if len(logs) > 0 {
// We have matching logs, check if we need to resolve full logs via the light client
if logs[0].TxHash == (common.Hash{}) {
- receipts, err := f.backend.GetReceipts(ctx, header.Hash())
+ receipts, err := f.sys.backend.GetReceipts(ctx, header.Hash())
if err != nil {
return nil, err
}
@@ -272,6 +266,19 @@ func (f *Filter) checkMatches(ctx context.Context, header *types.Header) (logs [
return nil, nil
}
+// pendingLogs returns the logs matching the filter criteria within the pending block.
+func (f *Filter) pendingLogs() ([]*types.Log, error) {
+ block, receipts := f.sys.backend.PendingBlockAndReceipts()
+ if bloomFilter(block.Bloom(), f.addresses, f.topics) {
+ var unfiltered []*types.Log
+ for _, r := range receipts {
+ unfiltered = append(unfiltered, r.Logs...)
+ }
+ return filterLogs(unfiltered, nil, nil, f.addresses, f.topics), nil
+ }
+ return nil, nil
+}
+
func includes(addresses []common.Address, a common.Address) bool {
for _, addr := range addresses {
if addr == a {
@@ -346,3 +353,11 @@ func bloomFilter(bloom types.Bloom, addresses []common.Address, topics [][]commo
}
return true
}
+
+func flatten(list [][]*types.Log) []*types.Log {
+ var flat []*types.Log
+ for _, logs := range list {
+ flat = append(flat, logs...)
+ }
+ return flat
+}
diff --git a/eth/filters/filter_system.go b/eth/filters/filter_system.go
index 7bf7db629711..78f77a89e116 100644
--- a/eth/filters/filter_system.go
+++ b/eth/filters/filter_system.go
@@ -21,18 +21,96 @@ package filters
import (
"context"
"errors"
+ "fmt"
"sync"
"time"
ethereum "github.com/XinFinOrg/XDPoSChain"
"github.com/XinFinOrg/XDPoSChain/common"
"github.com/XinFinOrg/XDPoSChain/core"
+ "github.com/XinFinOrg/XDPoSChain/core/bloombits"
"github.com/XinFinOrg/XDPoSChain/core/types"
+ "github.com/XinFinOrg/XDPoSChain/ethdb"
"github.com/XinFinOrg/XDPoSChain/event"
"github.com/XinFinOrg/XDPoSChain/log"
"github.com/XinFinOrg/XDPoSChain/rpc"
+ lru "github.com/hashicorp/golang-lru"
)
+// Config represents the configuration of the filter system.
+type Config struct {
+ LogCacheSize int // maximum number of cached blocks (default: 32)
+ Timeout time.Duration // how long filters stay active (default: 5min)
+}
+
+func (cfg Config) withDefaults() Config {
+ if cfg.Timeout == 0 {
+ cfg.Timeout = 5 * time.Minute
+ }
+ if cfg.LogCacheSize == 0 {
+ cfg.LogCacheSize = 32
+ }
+ return cfg
+}
+
+type Backend interface {
+ ChainDb() ethdb.Database
+ HeaderByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*types.Header, error)
+ HeaderByHash(ctx context.Context, blockHash common.Hash) (*types.Header, error)
+ GetReceipts(ctx context.Context, blockHash common.Hash) (types.Receipts, error)
+ GetLogs(ctx context.Context, blockHash common.Hash, number uint64) ([][]*types.Log, error)
+ PendingBlockAndReceipts() (*types.Block, types.Receipts)
+
+ SubscribeNewTxsEvent(chan<- core.NewTxsEvent) event.Subscription
+ SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription
+ SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription
+ SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription
+ SubscribePendingLogsEvent(ch chan<- []*types.Log) event.Subscription
+
+ BloomStatus() (uint64, uint64)
+ ServiceFilter(ctx context.Context, session *bloombits.MatcherSession)
+}
+
+// FilterSystem holds resources shared by all filters.
+type FilterSystem struct {
+ backend Backend
+ logsCache *lru.Cache
+ cfg *Config
+}
+
+// NewFilterSystem creates a filter system.
+func NewFilterSystem(backend Backend, config Config) *FilterSystem {
+ config = config.withDefaults()
+
+ cache, err := lru.New(config.LogCacheSize)
+ if err != nil {
+ panic(err)
+ }
+ return &FilterSystem{
+ backend: backend,
+ logsCache: cache,
+ cfg: &config,
+ }
+}
+
+// cachedGetLogs loads block logs from the backend and caches the result.
+func (sys *FilterSystem) cachedGetLogs(ctx context.Context, blockHash common.Hash, number uint64) ([][]*types.Log, error) {
+ cached, ok := sys.logsCache.Get(blockHash)
+ if ok {
+ return cached.([][]*types.Log), nil
+ }
+
+ logs, err := sys.backend.GetLogs(ctx, blockHash, number)
+ if err != nil {
+ return nil, err
+ }
+ if logs == nil {
+ return nil, fmt.Errorf("failed to get logs for block #%d (0x%s)", number, blockHash.TerminalString())
+ }
+ sys.logsCache.Add(blockHash, logs)
+ return logs, nil
+}
+
// Type determines the kind of filter and is used to put the filter in to
// the correct bucket when added.
type Type byte
@@ -67,10 +145,6 @@ const (
chainEvChanSize = 10
)
-var (
- ErrInvalidSubscriptionID = errors.New("invalid id")
-)
-
type subscription struct {
id rpc.ID
typ Type
@@ -86,25 +160,26 @@ type subscription struct {
// EventSystem creates subscriptions, processes events and broadcasts them to the
// subscription which match the subscription criteria.
type EventSystem struct {
- mux *event.TypeMux
backend Backend
+ sys *FilterSystem
lightMode bool
lastHead *types.Header
// Subscriptions
- txsSub event.Subscription // Subscription for new transaction event
- logsSub event.Subscription // Subscription for new log event
- rmLogsSub event.Subscription // Subscription for removed log event
- chainSub event.Subscription // Subscription for new chain event
- pendingLogSub *event.TypeMuxSubscription // Subscription for pending log event
+ txsSub event.Subscription // Subscription for new transaction event
+ logsSub event.Subscription // Subscription for new log event
+ rmLogsSub event.Subscription // Subscription for removed log event
+ pendingLogsSub event.Subscription // Subscription for pending log event
+ chainSub event.Subscription // Subscription for new chain event
// Channels
- install chan *subscription // install filter for event notification
- uninstall chan *subscription // remove filter for event notification
- txsCh chan core.NewTxsEvent // Channel to receive new transactions event
- logsCh chan []*types.Log // Channel to receive new log event
- rmLogsCh chan core.RemovedLogsEvent // Channel to receive removed log event
- chainCh chan core.ChainEvent // Channel to receive new chain event
+ install chan *subscription // install filter for event notification
+ uninstall chan *subscription // remove filter for event notification
+ txsCh chan core.NewTxsEvent // Channel to receive new transactions event
+ logsCh chan []*types.Log // Channel to receive new log event
+ pendingLogsCh chan []*types.Log // Channel to receive new log event
+ rmLogsCh chan core.RemovedLogsEvent // Channel to receive removed log event
+ chainCh chan core.ChainEvent // Channel to receive new chain event
}
// NewEventSystem creates a new manager that listens for event on the given mux,
@@ -113,17 +188,18 @@ type EventSystem struct {
//
// The returned manager has a loop that needs to be stopped with the Stop function
// or by stopping the given mux.
-func NewEventSystem(mux *event.TypeMux, backend Backend, lightMode bool) *EventSystem {
+func NewEventSystem(sys *FilterSystem, lightMode bool) *EventSystem {
m := &EventSystem{
- mux: mux,
- backend: backend,
- lightMode: lightMode,
- install: make(chan *subscription),
- uninstall: make(chan *subscription),
- txsCh: make(chan core.NewTxsEvent, txChanSize),
- logsCh: make(chan []*types.Log, logsChanSize),
- rmLogsCh: make(chan core.RemovedLogsEvent, rmLogsChanSize),
- chainCh: make(chan core.ChainEvent, chainEvChanSize),
+ sys: sys,
+ backend: sys.backend,
+ lightMode: lightMode,
+ install: make(chan *subscription),
+ uninstall: make(chan *subscription),
+ txsCh: make(chan core.NewTxsEvent, txChanSize),
+ logsCh: make(chan []*types.Log, logsChanSize),
+ rmLogsCh: make(chan core.RemovedLogsEvent, rmLogsChanSize),
+ pendingLogsCh: make(chan []*types.Log, logsChanSize),
+ chainCh: make(chan core.ChainEvent, chainEvChanSize),
}
// Subscribe events
@@ -131,12 +207,10 @@ func NewEventSystem(mux *event.TypeMux, backend Backend, lightMode bool) *EventS
m.logsSub = m.backend.SubscribeLogsEvent(m.logsCh)
m.rmLogsSub = m.backend.SubscribeRemovedLogsEvent(m.rmLogsCh)
m.chainSub = m.backend.SubscribeChainEvent(m.chainCh)
- // TODO(rjl493456442): use feed to subscribe pending log event
- m.pendingLogSub = m.mux.Subscribe(core.PendingLogsEvent{})
+ m.pendingLogsSub = m.backend.SubscribePendingLogsEvent(m.pendingLogsCh)
// Make sure none of the subscriptions are empty
- if m.txsSub == nil || m.logsSub == nil || m.rmLogsSub == nil || m.chainSub == nil ||
- m.pendingLogSub.Closed() {
+ if m.txsSub == nil || m.logsSub == nil || m.rmLogsSub == nil || m.chainSub == nil || m.pendingLogsSub == nil {
log.Crit("Subscribe for event system failed")
}
@@ -314,58 +388,61 @@ func (es *EventSystem) SubscribePendingTxs(hashes chan []common.Hash) *Subscript
type filterIndex map[Type]map[rpc.ID]*subscription
-// broadcast event to filters that match criteria.
-func (es *EventSystem) broadcast(filters filterIndex, ev interface{}) {
- if ev == nil {
+func (es *EventSystem) handleLogs(filters filterIndex, ev []*types.Log) {
+ if len(ev) == 0 {
+ return
+ }
+ for _, f := range filters[LogsSubscription] {
+ matchedLogs := filterLogs(ev, f.logsCrit.FromBlock, f.logsCrit.ToBlock, f.logsCrit.Addresses, f.logsCrit.Topics)
+ if len(matchedLogs) > 0 {
+ f.logs <- matchedLogs
+ }
+ }
+}
+
+func (es *EventSystem) handlePendingLogs(filters filterIndex, ev []*types.Log) {
+ if len(ev) == 0 {
return
}
+ for _, f := range filters[PendingLogsSubscription] {
+ matchedLogs := filterLogs(ev, nil, f.logsCrit.ToBlock, f.logsCrit.Addresses, f.logsCrit.Topics)
+ if len(matchedLogs) > 0 {
+ f.logs <- matchedLogs
+ }
+ }
+}
+
+func (es *EventSystem) handleRemovedLogs(filters filterIndex, ev core.RemovedLogsEvent) {
+ for _, f := range filters[LogsSubscription] {
+ matchedLogs := filterLogs(ev.Logs, f.logsCrit.FromBlock, f.logsCrit.ToBlock, f.logsCrit.Addresses, f.logsCrit.Topics)
+ if len(matchedLogs) > 0 {
+ f.logs <- matchedLogs
+ }
+ }
+}
+
+func (es *EventSystem) handleTxsEvent(filters filterIndex, ev core.NewTxsEvent) {
+ hashes := make([]common.Hash, 0, len(ev.Txs))
+ for _, tx := range ev.Txs {
+ hashes = append(hashes, tx.Hash())
+ }
+ for _, f := range filters[PendingTransactionsSubscription] {
+ f.hashes <- hashes
+ }
+}
- switch e := ev.(type) {
- case []*types.Log:
- if len(e) > 0 {
+func (es *EventSystem) handleChainEvent(filters filterIndex, ev core.ChainEvent) {
+ for _, f := range filters[BlocksSubscription] {
+ f.headers <- ev.Block.Header()
+ }
+ if es.lightMode && len(filters[LogsSubscription]) > 0 {
+ es.lightFilterNewHead(ev.Block.Header(), func(header *types.Header, remove bool) {
for _, f := range filters[LogsSubscription] {
- if matchedLogs := filterLogs(e, f.logsCrit.FromBlock, f.logsCrit.ToBlock, f.logsCrit.Addresses, f.logsCrit.Topics); len(matchedLogs) > 0 {
+ if matchedLogs := es.lightFilterLogs(header, f.logsCrit.Addresses, f.logsCrit.Topics, remove); len(matchedLogs) > 0 {
f.logs <- matchedLogs
}
}
- }
- case core.RemovedLogsEvent:
- for _, f := range filters[LogsSubscription] {
- if matchedLogs := filterLogs(e.Logs, f.logsCrit.FromBlock, f.logsCrit.ToBlock, f.logsCrit.Addresses, f.logsCrit.Topics); len(matchedLogs) > 0 {
- f.logs <- matchedLogs
- }
- }
- case *event.TypeMuxEvent:
- if muxe, ok := e.Data.(core.PendingLogsEvent); ok {
- for _, f := range filters[PendingLogsSubscription] {
- if e.Time.After(f.created) {
- if matchedLogs := filterLogs(muxe.Logs, nil, f.logsCrit.ToBlock, f.logsCrit.Addresses, f.logsCrit.Topics); len(matchedLogs) > 0 {
- f.logs <- matchedLogs
- }
- }
- }
- }
- case core.NewTxsEvent:
- hashes := make([]common.Hash, 0, len(e.Txs))
- for _, tx := range e.Txs {
- hashes = append(hashes, tx.Hash())
- }
- for _, f := range filters[PendingTransactionsSubscription] {
- f.hashes <- hashes
- }
- case core.ChainEvent:
- for _, f := range filters[BlocksSubscription] {
- f.headers <- e.Block.Header()
- }
- if es.lightMode && len(filters[LogsSubscription]) > 0 {
- es.lightFilterNewHead(e.Block.Header(), func(header *types.Header, remove bool) {
- for _, f := range filters[LogsSubscription] {
- if matchedLogs := es.lightFilterLogs(header, f.logsCrit.Addresses, f.logsCrit.Topics, remove); len(matchedLogs) > 0 {
- f.logs <- matchedLogs
- }
- }
- })
- }
+ })
}
}
@@ -408,7 +485,7 @@ func (es *EventSystem) lightFilterLogs(header *types.Header, addresses []common.
// Get the logs of the block
ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
defer cancel()
- logsList, err := es.backend.GetLogs(ctx, header.Hash())
+ logsList, err := es.sys.cachedGetLogs(ctx, header.Hash(), header.Number.Uint64())
if err != nil {
return nil
}
@@ -446,10 +523,10 @@ func (es *EventSystem) lightFilterLogs(header *types.Header, addresses []common.
func (es *EventSystem) eventLoop() {
// Ensure all subscriptions get cleaned up
defer func() {
- es.pendingLogSub.Unsubscribe()
es.txsSub.Unsubscribe()
es.logsSub.Unsubscribe()
es.rmLogsSub.Unsubscribe()
+ es.pendingLogsSub.Unsubscribe()
es.chainSub.Unsubscribe()
}()
@@ -460,20 +537,16 @@ func (es *EventSystem) eventLoop() {
for {
select {
- // Handle subscribed events
case ev := <-es.txsCh:
- es.broadcast(index, ev)
+ es.handleTxsEvent(index, ev)
case ev := <-es.logsCh:
- es.broadcast(index, ev)
+ es.handleLogs(index, ev)
case ev := <-es.rmLogsCh:
- es.broadcast(index, ev)
+ es.handleRemovedLogs(index, ev)
+ case ev := <-es.pendingLogsCh:
+ es.handlePendingLogs(index, ev)
case ev := <-es.chainCh:
- es.broadcast(index, ev)
- case ev, active := <-es.pendingLogSub.Chan():
- if !active { // system stopped
- return
- }
- es.broadcast(index, ev)
+ es.handleChainEvent(index, ev)
case f := <-es.install:
if f.typ == MinedAndPendingLogsSubscription {
diff --git a/eth/filters/filter_system_test.go b/eth/filters/filter_system_test.go
index fb4f7e7b8bc2..da36c01843b6 100644
--- a/eth/filters/filter_system_test.go
+++ b/eth/filters/filter_system_test.go
@@ -22,6 +22,7 @@ import (
"math/big"
"math/rand"
"reflect"
+ "runtime"
"testing"
"time"
@@ -39,23 +40,20 @@ import (
)
type testBackend struct {
- mux *event.TypeMux
- db ethdb.Database
- sections uint64
- txFeed *event.Feed
- rmLogsFeed *event.Feed
- logsFeed *event.Feed
- chainFeed *event.Feed
+ mux *event.TypeMux
+ db ethdb.Database
+ sections uint64
+ txFeed event.Feed
+ logsFeed event.Feed
+ rmLogsFeed event.Feed
+ pendingLogsFeed event.Feed
+ chainFeed event.Feed
}
func (b *testBackend) ChainDb() ethdb.Database {
return b.db
}
-func (b *testBackend) EventMux() *event.TypeMux {
- return b.mux
-}
-
func (b *testBackend) HeaderByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*types.Header, error) {
var hash common.Hash
var num uint64
@@ -79,17 +77,15 @@ func (b *testBackend) GetReceipts(ctx context.Context, blockHash common.Hash) (t
return core.GetBlockReceipts(b.db, blockHash, number), nil
}
-func (b *testBackend) GetLogs(ctx context.Context, blockHash common.Hash) ([][]*types.Log, error) {
- number := core.GetBlockNumber(b.db, blockHash)
- receipts := core.GetBlockReceipts(b.db, blockHash, number)
-
- logs := make([][]*types.Log, len(receipts))
- for i, receipt := range receipts {
- logs[i] = receipt.Logs
- }
+func (b *testBackend) GetLogs(ctx context.Context, hash common.Hash, number uint64) ([][]*types.Log, error) {
+ logs := rawdb.ReadLogs(b.db, hash, number)
return logs, nil
}
+func (b *testBackend) PendingBlockAndReceipts() (*types.Block, types.Receipts) {
+ return nil, nil
+}
+
func (b *testBackend) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription {
return b.txFeed.Subscribe(ch)
}
@@ -102,6 +98,10 @@ func (b *testBackend) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscript
return b.logsFeed.Subscribe(ch)
}
+func (b *testBackend) SubscribePendingLogsEvent(ch chan<- []*types.Log) event.Subscription {
+ return b.pendingLogsFeed.Subscribe(ch)
+}
+
func (b *testBackend) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription {
return b.chainFeed.Subscribe(ch)
}
@@ -137,6 +137,12 @@ func (b *testBackend) ServiceFilter(ctx context.Context, session *bloombits.Matc
}()
}
+func newTestFilterSystem(t testing.TB, db ethdb.Database, cfg Config) (*testBackend, *FilterSystem) {
+ backend := &testBackend{db: db}
+ sys := NewFilterSystem(backend, cfg)
+ return backend, sys
+}
+
// TestBlockSubscription tests if a block subscription returns block hashes for posted chain events.
// It creates multiple subscriptions:
// - one at the start and should receive all posted chain events and a second (blockHashes)
@@ -146,17 +152,12 @@ func TestBlockSubscription(t *testing.T) {
t.Parallel()
var (
- mux = new(event.TypeMux)
- db = rawdb.NewMemoryDatabase()
- txFeed = new(event.Feed)
- rmLogsFeed = new(event.Feed)
- logsFeed = new(event.Feed)
- chainFeed = new(event.Feed)
- backend = &testBackend{mux, db, 0, txFeed, rmLogsFeed, logsFeed, chainFeed}
- api = NewPublicFilterAPI(backend, false)
- genesis = new(core.Genesis).MustCommit(db)
- chain, _ = core.GenerateChain(params.TestChainConfig, genesis, ethash.NewFaker(), db, 10, func(i int, gen *core.BlockGen) {})
- chainEvents = []core.ChainEvent{}
+ db = rawdb.NewMemoryDatabase()
+ backend, sys = newTestFilterSystem(t, db, Config{})
+ api = NewFilterAPI(sys, false)
+ genesis = new(core.Genesis).MustCommit(db)
+ chain, _ = core.GenerateChain(params.TestChainConfig, genesis, ethash.NewFaker(), db, 10, func(i int, gen *core.BlockGen) {})
+ chainEvents = []core.ChainEvent{}
)
for _, blk := range chain {
@@ -191,7 +192,7 @@ func TestBlockSubscription(t *testing.T) {
time.Sleep(1 * time.Second)
for _, e := range chainEvents {
- chainFeed.Send(e)
+ backend.chainFeed.Send(e)
}
<-sub0.Err()
@@ -203,14 +204,9 @@ func TestPendingTxFilter(t *testing.T) {
t.Parallel()
var (
- mux = new(event.TypeMux)
- db = rawdb.NewMemoryDatabase()
- txFeed = new(event.Feed)
- rmLogsFeed = new(event.Feed)
- logsFeed = new(event.Feed)
- chainFeed = new(event.Feed)
- backend = &testBackend{mux, db, 0, txFeed, rmLogsFeed, logsFeed, chainFeed}
- api = NewPublicFilterAPI(backend, false)
+ db = rawdb.NewMemoryDatabase()
+ backend, sys = newTestFilterSystem(t, db, Config{})
+ api = NewFilterAPI(sys, false)
transactions = []*types.Transaction{
types.NewTransaction(0, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil),
@@ -226,7 +222,7 @@ func TestPendingTxFilter(t *testing.T) {
fid0 := api.NewPendingTransactionFilter()
time.Sleep(1 * time.Second)
- txFeed.Send(core.NewTxsEvent{Txs: transactions})
+ backend.txFeed.Send(core.NewTxsEvent{Txs: transactions})
timeout := time.Now().Add(1 * time.Second)
for {
@@ -263,14 +259,9 @@ func TestPendingTxFilter(t *testing.T) {
// If not it must return an error.
func TestLogFilterCreation(t *testing.T) {
var (
- mux = new(event.TypeMux)
- db = rawdb.NewMemoryDatabase()
- txFeed = new(event.Feed)
- rmLogsFeed = new(event.Feed)
- logsFeed = new(event.Feed)
- chainFeed = new(event.Feed)
- backend = &testBackend{mux, db, 0, txFeed, rmLogsFeed, logsFeed, chainFeed}
- api = NewPublicFilterAPI(backend, false)
+ db = rawdb.NewMemoryDatabase()
+ _, sys = newTestFilterSystem(t, db, Config{})
+ api = NewFilterAPI(sys, false)
testCases = []struct {
crit FilterCriteria
@@ -312,14 +303,9 @@ func TestInvalidLogFilterCreation(t *testing.T) {
t.Parallel()
var (
- mux = new(event.TypeMux)
- db = rawdb.NewMemoryDatabase()
- txFeed = new(event.Feed)
- rmLogsFeed = new(event.Feed)
- logsFeed = new(event.Feed)
- chainFeed = new(event.Feed)
- backend = &testBackend{mux, db, 0, txFeed, rmLogsFeed, logsFeed, chainFeed}
- api = NewPublicFilterAPI(backend, false)
+ db = rawdb.NewMemoryDatabase()
+ _, sys = newTestFilterSystem(t, db, Config{})
+ api = NewFilterAPI(sys, false)
)
// different situations where log filter creation should fail.
@@ -339,15 +325,10 @@ func TestInvalidLogFilterCreation(t *testing.T) {
func TestInvalidGetLogsRequest(t *testing.T) {
var (
- mux = new(event.TypeMux)
- db = rawdb.NewMemoryDatabase()
- txFeed = new(event.Feed)
- rmLogsFeed = new(event.Feed)
- logsFeed = new(event.Feed)
- chainFeed = new(event.Feed)
- backend = &testBackend{mux, db, 0, txFeed, rmLogsFeed, logsFeed, chainFeed}
- api = NewPublicFilterAPI(backend, false)
- blockHash = common.HexToHash("0x1111111111111111111111111111111111111111111111111111111111111111")
+ db = rawdb.NewMemoryDatabase()
+ _, sys = newTestFilterSystem(t, db, Config{})
+ api = NewFilterAPI(sys, false)
+ blockHash = common.HexToHash("0x1111111111111111111111111111111111111111111111111111111111111111")
)
// Reason: Cannot specify both BlockHash and FromBlock/ToBlock)
@@ -369,14 +350,9 @@ func TestLogFilter(t *testing.T) {
t.Parallel()
var (
- mux = new(event.TypeMux)
- db = rawdb.NewMemoryDatabase()
- txFeed = new(event.Feed)
- rmLogsFeed = new(event.Feed)
- logsFeed = new(event.Feed)
- chainFeed = new(event.Feed)
- backend = &testBackend{mux, db, 0, txFeed, rmLogsFeed, logsFeed, chainFeed}
- api = NewPublicFilterAPI(backend, false)
+ db = rawdb.NewMemoryDatabase()
+ backend, sys = newTestFilterSystem(t, db, Config{})
+ api = NewFilterAPI(sys, false)
firstAddr = common.HexToAddress("0x1111111111111111111111111111111111111111")
secondAddr = common.HexToAddress("0x2222222222222222222222222222222222222222")
@@ -386,7 +362,7 @@ func TestLogFilter(t *testing.T) {
secondTopic = common.HexToHash("0x2222222222222222222222222222222222222222222222222222222222222222")
notUsedTopic = common.HexToHash("0x9999999999999999999999999999999999999999999999999999999999999999")
- // posted twice, once as vm.Logs and once as core.PendingLogsEvent
+ // posted twice, once as regular logs and once as pending logs.
allLogs = []*types.Log{
{Address: firstAddr},
{Address: firstAddr, Topics: []common.Hash{firstTopic}, BlockNumber: 1},
@@ -439,11 +415,11 @@ func TestLogFilter(t *testing.T) {
// raise events
time.Sleep(1 * time.Second)
- if nsend := logsFeed.Send(allLogs); nsend == 0 {
- t.Fatal("Shoud have at least one subscription")
+ if nsend := backend.logsFeed.Send(allLogs); nsend == 0 {
+ t.Fatal("Logs event not delivered")
}
- if err := mux.Post(core.PendingLogsEvent{Logs: allLogs}); err != nil {
- t.Fatal(err)
+ if nsend := backend.pendingLogsFeed.Send(allLogs); nsend == 0 {
+ t.Fatal("Pending logs event not delivered")
}
for i, tt := range testCases {
@@ -488,14 +464,9 @@ func TestPendingLogsSubscription(t *testing.T) {
t.Parallel()
var (
- mux = new(event.TypeMux)
- db = rawdb.NewMemoryDatabase()
- txFeed = new(event.Feed)
- rmLogsFeed = new(event.Feed)
- logsFeed = new(event.Feed)
- chainFeed = new(event.Feed)
- backend = &testBackend{mux, db, 0, txFeed, rmLogsFeed, logsFeed, chainFeed}
- api = NewPublicFilterAPI(backend, false)
+ db = rawdb.NewMemoryDatabase()
+ backend, sys = newTestFilterSystem(t, db, Config{})
+ api = NewFilterAPI(sys, false)
firstAddr = common.HexToAddress("0x1111111111111111111111111111111111111111")
secondAddr = common.HexToAddress("0x2222222222222222222222222222222222222222")
@@ -507,26 +478,18 @@ func TestPendingLogsSubscription(t *testing.T) {
fourthTopic = common.HexToHash("0x4444444444444444444444444444444444444444444444444444444444444444")
notUsedTopic = common.HexToHash("0x9999999999999999999999999999999999999999999999999999999999999999")
- allLogs = []core.PendingLogsEvent{
- {Logs: []*types.Log{{Address: firstAddr, Topics: []common.Hash{}, BlockNumber: 0}}},
- {Logs: []*types.Log{{Address: firstAddr, Topics: []common.Hash{firstTopic}, BlockNumber: 1}}},
- {Logs: []*types.Log{{Address: secondAddr, Topics: []common.Hash{firstTopic}, BlockNumber: 2}}},
- {Logs: []*types.Log{{Address: thirdAddress, Topics: []common.Hash{secondTopic}, BlockNumber: 3}}},
- {Logs: []*types.Log{{Address: thirdAddress, Topics: []common.Hash{secondTopic}, BlockNumber: 4}}},
- {Logs: []*types.Log{
+ allLogs = [][]*types.Log{
+ {{Address: firstAddr, Topics: []common.Hash{}, BlockNumber: 0}},
+ {{Address: firstAddr, Topics: []common.Hash{firstTopic}, BlockNumber: 1}},
+ {{Address: secondAddr, Topics: []common.Hash{firstTopic}, BlockNumber: 2}},
+ {{Address: thirdAddress, Topics: []common.Hash{secondTopic}, BlockNumber: 3}},
+ {{Address: thirdAddress, Topics: []common.Hash{secondTopic}, BlockNumber: 4}},
+ {
{Address: thirdAddress, Topics: []common.Hash{firstTopic}, BlockNumber: 5},
{Address: thirdAddress, Topics: []common.Hash{thirdTopic}, BlockNumber: 5},
{Address: thirdAddress, Topics: []common.Hash{fourthTopic}, BlockNumber: 5},
{Address: firstAddr, Topics: []common.Hash{firstTopic}, BlockNumber: 5},
- }},
- }
-
- convertLogs = func(pl []core.PendingLogsEvent) []*types.Log {
- var logs []*types.Log
- for _, l := range pl {
- logs = append(logs, l.Logs...)
- }
- return logs
+ },
}
testCases = []struct {
@@ -536,21 +499,52 @@ func TestPendingLogsSubscription(t *testing.T) {
sub *Subscription
}{
// match all
- {ethereum.FilterQuery{}, convertLogs(allLogs), nil, nil},
+ {
+ ethereum.FilterQuery{}, flattenLogs(allLogs),
+ nil, nil,
+ },
// match none due to no matching addresses
- {ethereum.FilterQuery{Addresses: []common.Address{{}, notUsedAddress}, Topics: [][]common.Hash{nil}}, []*types.Log{}, nil, nil},
+ {
+ ethereum.FilterQuery{Addresses: []common.Address{{}, notUsedAddress}, Topics: [][]common.Hash{nil}},
+ nil,
+ nil, nil,
+ },
// match logs based on addresses, ignore topics
- {ethereum.FilterQuery{Addresses: []common.Address{firstAddr}}, append(convertLogs(allLogs[:2]), allLogs[5].Logs[3]), nil, nil},
+ {
+ ethereum.FilterQuery{Addresses: []common.Address{firstAddr}},
+ append(flattenLogs(allLogs[:2]), allLogs[5][3]),
+ nil, nil,
+ },
// match none due to no matching topics (match with address)
- {ethereum.FilterQuery{Addresses: []common.Address{secondAddr}, Topics: [][]common.Hash{{notUsedTopic}}}, []*types.Log{}, nil, nil},
+ {
+ ethereum.FilterQuery{Addresses: []common.Address{secondAddr}, Topics: [][]common.Hash{{notUsedTopic}}},
+ nil, nil, nil,
+ },
// match logs based on addresses and topics
- {ethereum.FilterQuery{Addresses: []common.Address{thirdAddress}, Topics: [][]common.Hash{{firstTopic, secondTopic}}}, append(convertLogs(allLogs[3:5]), allLogs[5].Logs[0]), nil, nil},
+ {
+ ethereum.FilterQuery{Addresses: []common.Address{thirdAddress}, Topics: [][]common.Hash{{firstTopic, secondTopic}}},
+ append(flattenLogs(allLogs[3:5]), allLogs[5][0]),
+ nil, nil,
+ },
// match logs based on multiple addresses and "or" topics
- {ethereum.FilterQuery{Addresses: []common.Address{secondAddr, thirdAddress}, Topics: [][]common.Hash{{firstTopic, secondTopic}}}, append(convertLogs(allLogs[2:5]), allLogs[5].Logs[0]), nil, nil},
+ {
+ ethereum.FilterQuery{Addresses: []common.Address{secondAddr, thirdAddress}, Topics: [][]common.Hash{{firstTopic, secondTopic}}},
+ append(flattenLogs(allLogs[2:5]), allLogs[5][0]),
+ nil,
+ nil,
+ },
// block numbers are ignored for filters created with New***Filter, these return all logs that match the given criteria when the state changes
- {ethereum.FilterQuery{Addresses: []common.Address{firstAddr}, FromBlock: big.NewInt(2), ToBlock: big.NewInt(3)}, append(convertLogs(allLogs[:2]), allLogs[5].Logs[3]), nil, nil},
+ {
+ ethereum.FilterQuery{Addresses: []common.Address{firstAddr}, FromBlock: big.NewInt(2), ToBlock: big.NewInt(3)},
+ append(flattenLogs(allLogs[:2]), allLogs[5][3]),
+ nil, nil,
+ },
// multiple pending logs, should match only 2 topics from the logs in block 5
- {ethereum.FilterQuery{Addresses: []common.Address{thirdAddress}, Topics: [][]common.Hash{{firstTopic, fourthTopic}}}, []*types.Log{allLogs[5].Logs[0], allLogs[5].Logs[2]}, nil, nil},
+ {
+ ethereum.FilterQuery{Addresses: []common.Address{thirdAddress}, Topics: [][]common.Hash{{firstTopic, fourthTopic}}},
+ []*types.Log{allLogs[5][0], allLogs[5][2]},
+ nil, nil,
+ },
}
)
@@ -593,10 +587,82 @@ func TestPendingLogsSubscription(t *testing.T) {
// raise events
time.Sleep(1 * time.Second)
- // allLogs are type of core.PendingLogsEvent
- for _, l := range allLogs {
- if err := mux.Post(l); err != nil {
- t.Fatal(err)
+ for _, ev := range allLogs {
+ backend.pendingLogsFeed.Send(ev)
+ }
+}
+
+// TestPendingTxFilterDeadlock tests if the event loop hangs when pending
+// txes arrive at the same time that one of multiple filters is timing out.
+// Please refer to #22131 for more details.
+func TestPendingTxFilterDeadlock(t *testing.T) {
+ t.Parallel()
+ timeout := 100 * time.Millisecond
+
+ var (
+ db = rawdb.NewMemoryDatabase()
+ backend, sys = newTestFilterSystem(t, db, Config{Timeout: timeout})
+ api = NewFilterAPI(sys, false)
+ done = make(chan struct{})
+ )
+
+ go func() {
+ // Bombard feed with txes until signal was received to stop
+ i := uint64(0)
+ for {
+ select {
+ case <-done:
+ return
+ default:
+ }
+
+ tx := types.NewTransaction(i, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil)
+ backend.txFeed.Send(core.NewTxsEvent{Txs: []*types.Transaction{tx}})
+ i++
}
+ }()
+
+ // Create a bunch of filters that will
+ // timeout either in 100ms or 200ms
+ fids := make([]rpc.ID, 20)
+ for i := 0; i < len(fids); i++ {
+ fid := api.NewPendingTransactionFilter()
+ fids[i] = fid
+ // Wait for at least one tx to arrive in filter
+ for {
+ hashes, err := api.GetFilterChanges(fid)
+ if err != nil {
+ t.Fatalf("Filter should exist: %v\n", err)
+ }
+ if len(hashes.([]common.Hash)) > 0 {
+ break
+ }
+ runtime.Gosched()
+ }
+ }
+
+ // Wait until filters have timed out
+ time.Sleep(3 * timeout)
+
+ // If tx loop doesn't consume `done` after a second
+ // it's hanging.
+ select {
+ case done <- struct{}{}:
+ // Check that all filters have been uninstalled
+ for _, fid := range fids {
+ if _, err := api.GetFilterChanges(fid); err == nil {
+ t.Errorf("Filter %s should have been uninstalled\n", fid)
+ }
+ }
+ case <-time.After(1 * time.Second):
+ t.Error("Tx sending loop hangs")
+ }
+}
+
+func flattenLogs(pl [][]*types.Log) []*types.Log {
+ var logs []*types.Log
+ for _, l := range pl {
+ logs = append(logs, l...)
}
+ return logs
}
diff --git a/eth/filters/filter_test.go b/eth/filters/filter_test.go
index 1b74d21df7e0..0945c661c290 100644
--- a/eth/filters/filter_test.go
+++ b/eth/filters/filter_test.go
@@ -29,7 +29,6 @@ import (
"github.com/XinFinOrg/XDPoSChain/core"
"github.com/XinFinOrg/XDPoSChain/core/types"
"github.com/XinFinOrg/XDPoSChain/crypto"
- "github.com/XinFinOrg/XDPoSChain/event"
"github.com/XinFinOrg/XDPoSChain/params"
)
@@ -50,18 +49,13 @@ func BenchmarkFilters(b *testing.B) {
defer os.RemoveAll(dir)
var (
- db, _ = rawdb.NewLevelDBDatabase(dir, 0, 0, "")
- mux = new(event.TypeMux)
- txFeed = new(event.Feed)
- rmLogsFeed = new(event.Feed)
- logsFeed = new(event.Feed)
- chainFeed = new(event.Feed)
- backend = &testBackend{mux, db, 0, txFeed, rmLogsFeed, logsFeed, chainFeed}
- key1, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
- addr1 = crypto.PubkeyToAddress(key1.PublicKey)
- addr2 = common.BytesToAddress([]byte("jeff"))
- addr3 = common.BytesToAddress([]byte("ethereum"))
- addr4 = common.BytesToAddress([]byte("random addresses please"))
+ db, _ = rawdb.NewLevelDBDatabase(dir, 0, 0, "")
+ _, sys = newTestFilterSystem(b, db, Config{})
+ key1, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
+ addr1 = crypto.PubkeyToAddress(key1.PublicKey)
+ addr2 = common.BytesToAddress([]byte("jeff"))
+ addr3 = common.BytesToAddress([]byte("ethereum"))
+ addr4 = common.BytesToAddress([]byte("random addresses please"))
)
defer db.Close()
@@ -97,7 +91,7 @@ func BenchmarkFilters(b *testing.B) {
}
b.ResetTimer()
- filter := NewRangeFilter(backend, 0, -1, []common.Address{addr1, addr2, addr3, addr4}, nil)
+ filter := sys.NewRangeFilter(0, -1, []common.Address{addr1, addr2, addr3, addr4}, nil)
for i := 0; i < b.N; i++ {
logs, _ := filter.Logs(context.Background())
@@ -115,15 +109,10 @@ func TestFilters(t *testing.T) {
defer os.RemoveAll(dir)
var (
- db, _ = rawdb.NewLevelDBDatabase(dir, 0, 0, "")
- mux = new(event.TypeMux)
- txFeed = new(event.Feed)
- rmLogsFeed = new(event.Feed)
- logsFeed = new(event.Feed)
- chainFeed = new(event.Feed)
- backend = &testBackend{mux, db, 0, txFeed, rmLogsFeed, logsFeed, chainFeed}
- key1, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
- addr = crypto.PubkeyToAddress(key1.PublicKey)
+ db, _ = rawdb.NewLevelDBDatabase(dir, 0, 0, "")
+ _, sys = newTestFilterSystem(t, db, Config{})
+ key1, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
+ addr = crypto.PubkeyToAddress(key1.PublicKey)
hash1 = common.BytesToHash([]byte("topic1"))
hash2 = common.BytesToHash([]byte("topic2"))
@@ -144,6 +133,7 @@ func TestFilters(t *testing.T) {
},
}
gen.AddUncheckedReceipt(receipt)
+ gen.AddUncheckedTx(types.NewTransaction(1, common.HexToAddress("0x1"), big.NewInt(1), 1, big.NewInt(2100), nil))
case 2:
receipt := types.NewReceipt(nil, false, 0)
receipt.Logs = []*types.Log{
@@ -153,6 +143,7 @@ func TestFilters(t *testing.T) {
},
}
gen.AddUncheckedReceipt(receipt)
+ gen.AddUncheckedTx(types.NewTransaction(2, common.HexToAddress("0x2"), big.NewInt(2), 2, big.NewInt(2100), nil))
case 998:
receipt := types.NewReceipt(nil, false, 0)
receipt.Logs = []*types.Log{
@@ -162,6 +153,7 @@ func TestFilters(t *testing.T) {
},
}
gen.AddUncheckedReceipt(receipt)
+ gen.AddUncheckedTx(types.NewTransaction(998, common.HexToAddress("0x998"), big.NewInt(998), 998, big.NewInt(2100), nil))
case 999:
receipt := types.NewReceipt(nil, false, 0)
receipt.Logs = []*types.Log{
@@ -171,6 +163,7 @@ func TestFilters(t *testing.T) {
},
}
gen.AddUncheckedReceipt(receipt)
+ gen.AddUncheckedTx(types.NewTransaction(999, common.HexToAddress("0x999"), big.NewInt(999), 999, big.NewInt(2100), nil))
}
})
for i, block := range chain {
@@ -186,14 +179,14 @@ func TestFilters(t *testing.T) {
}
}
- filter := NewRangeFilter(backend, 0, -1, []common.Address{addr}, [][]common.Hash{{hash1, hash2, hash3, hash4}})
+ filter := sys.NewRangeFilter(0, -1, []common.Address{addr}, [][]common.Hash{{hash1, hash2, hash3, hash4}})
logs, _ := filter.Logs(context.Background())
if len(logs) != 4 {
t.Error("expected 4 log, got", len(logs))
}
- filter = NewRangeFilter(backend, 900, 999, []common.Address{addr}, [][]common.Hash{{hash3}})
+ filter = sys.NewRangeFilter(900, 999, []common.Address{addr}, [][]common.Hash{{hash3}})
logs, _ = filter.Logs(context.Background())
if len(logs) != 1 {
t.Error("expected 1 log, got", len(logs))
@@ -202,7 +195,7 @@ func TestFilters(t *testing.T) {
t.Errorf("expected log[0].Topics[0] to be %x, got %x", hash3, logs[0].Topics[0])
}
- filter = NewRangeFilter(backend, 990, -1, []common.Address{addr}, [][]common.Hash{{hash3}})
+ filter = sys.NewRangeFilter(990, -1, []common.Address{addr}, [][]common.Hash{{hash3}})
logs, _ = filter.Logs(context.Background())
if len(logs) != 1 {
t.Error("expected 1 log, got", len(logs))
@@ -211,7 +204,7 @@ func TestFilters(t *testing.T) {
t.Errorf("expected log[0].Topics[0] to be %x, got %x", hash3, logs[0].Topics[0])
}
- filter = NewRangeFilter(backend, 1, 10, nil, [][]common.Hash{{hash1, hash2}})
+ filter = sys.NewRangeFilter(1, 10, nil, [][]common.Hash{{hash1, hash2}})
logs, _ = filter.Logs(context.Background())
if len(logs) != 2 {
@@ -219,7 +212,7 @@ func TestFilters(t *testing.T) {
}
failHash := common.BytesToHash([]byte("fail"))
- filter = NewRangeFilter(backend, 0, -1, nil, [][]common.Hash{{failHash}})
+ filter = sys.NewRangeFilter(0, -1, nil, [][]common.Hash{{failHash}})
logs, _ = filter.Logs(context.Background())
if len(logs) != 0 {
@@ -227,14 +220,14 @@ func TestFilters(t *testing.T) {
}
failAddr := common.BytesToAddress([]byte("failmenow"))
- filter = NewRangeFilter(backend, 0, -1, []common.Address{failAddr}, nil)
+ filter = sys.NewRangeFilter(0, -1, []common.Address{failAddr}, nil)
logs, _ = filter.Logs(context.Background())
if len(logs) != 0 {
t.Error("expected 0 log, got", len(logs))
}
- filter = NewRangeFilter(backend, 0, -1, nil, [][]common.Hash{{failHash}, {hash1}})
+ filter = sys.NewRangeFilter(0, -1, nil, [][]common.Hash{{failHash}, {hash1}})
logs, _ = filter.Logs(context.Background())
if len(logs) != 0 {
diff --git a/internal/ethapi/backend.go b/internal/ethapi/backend.go
index d862977d2c5f..ad09ad95c00c 100644
--- a/internal/ethapi/backend.go
+++ b/internal/ethapi/backend.go
@@ -35,6 +35,7 @@ import (
"github.com/XinFinOrg/XDPoSChain/core/types"
"github.com/XinFinOrg/XDPoSChain/core/vm"
"github.com/XinFinOrg/XDPoSChain/eth/downloader"
+ "github.com/XinFinOrg/XDPoSChain/eth/filters"
"github.com/XinFinOrg/XDPoSChain/ethdb"
"github.com/XinFinOrg/XDPoSChain/event"
"github.com/XinFinOrg/XDPoSChain/params"
@@ -49,9 +50,8 @@ type Backend interface {
ProtocolVersion() int
SuggestPrice(ctx context.Context) (*big.Int, error)
ChainDb() ethdb.Database
- EventMux() *event.TypeMux
AccountManager() *accounts.Manager
- RPCGasCap() uint64 // global gas cap for eth_call over rpc: DoS protection
+ RPCGasCap() uint64 // global gas cap for eth_call over rpc: DoS protection
RPCTxFeeCap() float64 // global tx fee cap for all transaction related APIs
XDCxService() *XDCx.XDCX
LendingService() *XDCxlending.Lending
@@ -67,6 +67,7 @@ type Backend interface {
StateAndHeaderByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*state.StateDB, *types.Header, error)
StateAndHeaderByNumberOrHash(ctx context.Context, blockNrOrHash rpc.BlockNumberOrHash) (*state.StateDB, *types.Header, error)
GetBlock(ctx context.Context, blockHash common.Hash) (*types.Block, error)
+ PendingBlockAndReceipts() (*types.Block, types.Receipts)
GetReceipts(ctx context.Context, blockHash common.Hash) (types.Receipts, error)
GetTd(blockHash common.Hash) *big.Int
GetEVM(ctx context.Context, msg core.Message, state *state.StateDB, XDCxState *tradingstate.TradingStateDB, header *types.Header, vmConfig *vm.Config) (*vm.EVM, func() error, error)
@@ -88,6 +89,7 @@ type Backend interface {
OrderTxPoolContent() (map[common.Address]types.OrderTransactions, map[common.Address]types.OrderTransactions)
OrderStats() (pending int, queued int)
SendLendingTx(ctx context.Context, signedTx *types.LendingTransaction) error
+ SubscribePendingLogsEvent(ch chan<- []*types.Log) event.Subscription
ChainConfig() *params.ChainConfig
CurrentBlock() *types.Block
@@ -102,6 +104,10 @@ type Backend interface {
GetBlocksHashCache(blockNr uint64) []common.Hash
AreTwoBlockSamePath(newBlock common.Hash, oldBlock common.Hash) bool
GetOrderNonce(address common.Hash) (uint64, error)
+
+ // eth/filters needs to be initialized from this backend type, so methods needed by
+ // it must also be included here.
+ filters.Backend
}
func GetAPIs(apiBackend Backend, chainReader consensus.ChainReader) []rpc.API {
diff --git a/les/api_backend.go b/les/api_backend.go
index 3e93efe9601f..680d81f18ed4 100644
--- a/les/api_backend.go
+++ b/les/api_backend.go
@@ -130,6 +130,10 @@ func (b *LesApiBackend) BlockByNumberOrHash(ctx context.Context, blockNrOrHash r
return nil, errors.New("invalid arguments; neither block nor hash specified")
}
+func (b *LesApiBackend) PendingBlockAndReceipts() (*types.Block, types.Receipts) {
+ return nil, nil
+}
+
func (b *LesApiBackend) StateAndHeaderByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*state.StateDB, *types.Header, error) {
header, err := b.HeaderByNumber(ctx, blockNr)
if header == nil || err != nil {
@@ -163,8 +167,8 @@ func (b *LesApiBackend) GetReceipts(ctx context.Context, blockHash common.Hash)
return light.GetBlockReceipts(ctx, b.eth.odr, blockHash, core.GetBlockNumber(b.eth.chainDb, blockHash))
}
-func (b *LesApiBackend) GetLogs(ctx context.Context, blockHash common.Hash) ([][]*types.Log, error) {
- return light.GetBlockLogs(ctx, b.eth.odr, blockHash, core.GetBlockNumber(b.eth.chainDb, blockHash))
+func (b *LesApiBackend) GetLogs(ctx context.Context, hash common.Hash, number uint64) ([][]*types.Log, error) {
+ return light.GetBlockLogs(ctx, b.eth.odr, hash, number)
}
func (b *LesApiBackend) GetTd(blockHash common.Hash) *big.Int {
@@ -241,6 +245,13 @@ func (b *LesApiBackend) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscri
return b.eth.blockchain.SubscribeLogsEvent(ch)
}
+func (b *LesApiBackend) SubscribePendingLogsEvent(ch chan<- []*types.Log) event.Subscription {
+ return event.NewSubscription(func(quit <-chan struct{}) error {
+ <-quit
+ return nil
+ })
+}
+
func (b *LesApiBackend) SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription {
return b.eth.blockchain.SubscribeRemovedLogsEvent(ch)
}
@@ -261,10 +272,6 @@ func (b *LesApiBackend) ChainDb() ethdb.Database {
return b.eth.chainDb
}
-func (b *LesApiBackend) EventMux() *event.TypeMux {
- return b.eth.eventMux
-}
-
func (b *LesApiBackend) AccountManager() *accounts.Manager {
return b.eth.accountManager
}
diff --git a/les/backend.go b/les/backend.go
index 304dfc7bd331..fe3ffa80b5cb 100644
--- a/les/backend.go
+++ b/les/backend.go
@@ -190,7 +190,7 @@ func (s *LightEthereum) APIs() []rpc.API {
}, {
Namespace: "eth",
Version: "1.0",
- Service: filters.NewPublicFilterAPI(s.ApiBackend, true),
+ Service: filters.NewFilterAPI(filters.NewFilterSystem(s.ApiBackend, filters.Config{LogCacheSize: s.config.FilterLogCacheSize}), true),
Public: true,
}, {
Namespace: "net",
diff --git a/miner/miner.go b/miner/miner.go
index 4a9d34b9f7fa..835f0f014b74 100644
--- a/miner/miner.go
+++ b/miner/miner.go
@@ -178,7 +178,18 @@ func (self *Miner) PendingBlock() *types.Block {
return self.worker.pendingBlock()
}
+// PendingBlockAndReceipts returns the currently pending block and corresponding receipts.
+func (miner *Miner) PendingBlockAndReceipts() (*types.Block, types.Receipts) {
+ return miner.worker.pendingBlockAndReceipts()
+}
+
func (self *Miner) SetEtherbase(addr common.Address) {
self.coinbase = addr
self.worker.setEtherbase(addr)
}
+
+// SubscribePendingLogs starts delivering logs from pending transactions
+// to the given channel.
+func (self *Miner) SubscribePendingLogs(ch chan<- []*types.Log) event.Subscription {
+ return self.worker.pendingLogsFeed.Subscribe(ch)
+}
diff --git a/miner/worker.go b/miner/worker.go
index a97f55ceabbb..050191720eea 100644
--- a/miner/worker.go
+++ b/miner/worker.go
@@ -104,6 +104,9 @@ type worker struct {
mu sync.Mutex
+ // Feeds
+ pendingLogsFeed event.Feed
+
// update loop
mux *event.TypeMux
txsCh chan core.NewTxsEvent
@@ -125,6 +128,10 @@ type worker struct {
coinbase common.Address
extra []byte
+ snapshotMu sync.RWMutex // The lock used to protect the block snapshot and state snapshot
+ snapshotBlock *types.Block
+ snapshotReceipts types.Receipts
+
currentMu sync.Mutex
current *Work
@@ -216,6 +223,14 @@ func (self *worker) pendingBlock() *types.Block {
return self.current.Block
}
+// pendingBlockAndReceipts returns pending block and corresponding receipts.
+func (w *worker) pendingBlockAndReceipts() (*types.Block, types.Receipts) {
+ // return a snapshot to avoid contention on currentMu mutex
+ w.snapshotMu.RLock()
+ defer w.snapshotMu.RUnlock()
+ return w.snapshotBlock, w.snapshotReceipts
+}
+
func (self *worker) start() {
self.mu.Lock()
defer self.mu.Unlock()
@@ -322,7 +337,7 @@ func (self *worker) update() {
}
feeCapacity := state.GetTRC21FeeCapacityFromState(self.current.state)
txset, specialTxs := types.NewTransactionsByPriceAndNonce(self.current.signer, txs, nil, feeCapacity)
- self.current.commitTransactions(self.mux, feeCapacity, txset, specialTxs, self.chain, self.coinbase)
+ self.current.commitTransactions(self.mux, feeCapacity, txset, specialTxs, self.chain, self.coinbase, &self.pendingLogsFeed)
self.currentMu.Unlock()
} else {
// If we're mining, but nothing is being processed, wake on new transactions
@@ -781,7 +796,7 @@ func (self *worker) commitNewWork() {
specialTxs = append(specialTxs, txStateRoot)
}
}
- work.commitTransactions(self.mux, feeCapacity, txs, specialTxs, self.chain, self.coinbase)
+ work.commitTransactions(self.mux, feeCapacity, txs, specialTxs, self.chain, self.coinbase, &self.pendingLogsFeed)
// compute uncles for the new block.
var (
uncles []*types.Header
@@ -801,7 +816,7 @@ func (self *worker) commitNewWork() {
self.push(work)
}
-func (env *Work) commitTransactions(mux *event.TypeMux, balanceFee map[common.Address]*big.Int, txs *types.TransactionsByPriceAndNonce, specialTxs types.Transactions, bc *core.BlockChain, coinbase common.Address) {
+func (env *Work) commitTransactions(mux *event.TypeMux, balanceFee map[common.Address]*big.Int, txs *types.TransactionsByPriceAndNonce, specialTxs types.Transactions, bc *core.BlockChain, coinbase common.Address, pendingLogsFeed *event.Feed) {
gp := new(core.GasPool).AddGas(env.header.GasLimit)
balanceUpdated := map[common.Address]*big.Int{}
totalFeeUsed := big.NewInt(0)
@@ -1039,10 +1054,7 @@ func (env *Work) commitTransactions(mux *event.TypeMux, balanceFee map[common.Ad
}
go func(logs []*types.Log, tcount int) {
if len(logs) > 0 {
- err := mux.Post(core.PendingLogsEvent{Logs: logs})
- if err != nil {
- log.Warn("[commitTransactions] Error when sending PendingLogsEvent", "LogLength", len(logs))
- }
+ pendingLogsFeed.Send(logs)
}
if tcount > 0 {
err := mux.Post(core.PendingStateEvent{})
diff --git a/node/node.go b/node/node.go
index c54328913742..a9f767bf28f1 100644
--- a/node/node.go
+++ b/node/node.go
@@ -48,6 +48,7 @@ type Node struct {
serverConfig p2p.Config
server *p2p.Server // Currently running P2P networking layer
+ state int // Tracks state of node lifecycle
serviceFuncs []ServiceConstructor // Service constructors (in dependency order)
services map[reflect.Type]Service // Currently running services
@@ -74,6 +75,10 @@ type Node struct {
log log.Logger
}
+const (
+ initializingState = iota
+)
+
// New creates a new P2P node, ready for protocol registration.
func New(conf *Config) (*Node, error) {
// Copy config and resolve the datadir so future changes to the current
@@ -302,6 +307,17 @@ func (n *Node) stopInProc() {
}
}
+// RegisterAPIs registers the APIs a service provides on the node.
+func (n *Node) RegisterAPIs(apis []rpc.API) {
+ n.lock.Lock()
+ defer n.lock.Unlock()
+
+ if n.state != initializingState {
+ panic("can't register APIs on running/stopped node")
+ }
+ n.rpcAPIs = append(n.rpcAPIs, apis...)
+}
+
// startIPC initializes and starts the IPC RPC endpoint.
func (n *Node) startIPC(apis []rpc.API) error {
// Short circuit if the IPC endpoint isn't being exposed