aboutsummaryrefslogtreecommitdiffstats
path: root/core
diff options
context:
space:
mode:
authorWei-Ning Huang <w@dexon.org>2018-10-31 18:21:55 +0800
committerWei-Ning Huang <w@dexon.org>2019-04-09 21:32:52 +0800
commit2b99b89e5af0cf9feca3ab532629f8fab3858edc (patch)
treef02dbbb15fac9703d6d29230a89b0bab572c23ca /core
parent48cd626abe5169e87e6a7f264f4f6cfda71c8e1f (diff)
downloaddexon-2b99b89e5af0cf9feca3ab532629f8fab3858edc.tar
dexon-2b99b89e5af0cf9feca3ab532629f8fab3858edc.tar.gz
dexon-2b99b89e5af0cf9feca3ab532629f8fab3858edc.tar.bz2
dexon-2b99b89e5af0cf9feca3ab532629f8fab3858edc.tar.lz
dexon-2b99b89e5af0cf9feca3ab532629f8fab3858edc.tar.xz
dexon-2b99b89e5af0cf9feca3ab532629f8fab3858edc.tar.zst
dexon-2b99b89e5af0cf9feca3ab532629f8fab3858edc.zip
core: tx_pool: remove transactions on BlockConfirmed event
Diffstat (limited to 'core')
-rw-r--r--core/blockchain.go26
-rw-r--r--core/events.go2
-rw-r--r--core/tx_pool.go56
-rw-r--r--core/tx_pool_test.go10
4 files changed, 53 insertions, 41 deletions
diff --git a/core/blockchain.go b/core/blockchain.go
index 0e86f2b59..264617857 100644
--- a/core/blockchain.go
+++ b/core/blockchain.go
@@ -104,14 +104,15 @@ type BlockChain struct {
triegc *prque.Prque // Priority queue mapping block numbers to tries to gc
gcproc time.Duration // Accumulates canonical block processing for trie dumping
- hc *HeaderChain
- rmLogsFeed event.Feed
- chainFeed event.Feed
- chainSideFeed event.Feed
- chainHeadFeed event.Feed
- logsFeed event.Feed
- scope event.SubscriptionScope
- genesisBlock *types.Block
+ hc *HeaderChain
+ rmLogsFeed event.Feed
+ chainFeed event.Feed
+ chainSideFeed event.Feed
+ chainHeadFeed event.Feed
+ blockConfirmedFeed event.Feed
+ logsFeed event.Feed
+ scope event.SubscriptionScope
+ genesisBlock *types.Block
mu sync.RWMutex // global mutex for locking chain operations
chainmu sync.RWMutex // blockchain insertion lock
@@ -1610,6 +1611,7 @@ func (bc *BlockChain) processPendingBlock(block *types.Block, witness *coreTypes
// add into pending blocks
bc.addPendingBlock(newPendingBlock, receipts)
+ events = append(events, BlockConfirmedEvent{newPendingBlock})
// start insert available pending blocks into db
for pendingHeight := bc.CurrentBlock().NumberU64() + 1; pendingHeight <= witness.Height; pendingHeight++ {
@@ -1852,6 +1854,9 @@ func (bc *BlockChain) PostChainEvents(events []interface{}, logs []*types.Log) {
case ChainHeadEvent:
bc.chainHeadFeed.Send(ev)
+ case BlockConfirmedEvent:
+ bc.blockConfirmedFeed.Send(ev)
+
case ChainSideEvent:
bc.chainSideFeed.Send(ev)
}
@@ -2044,6 +2049,11 @@ func (bc *BlockChain) SubscribeChainHeadEvent(ch chan<- ChainHeadEvent) event.Su
return bc.scope.Track(bc.chainHeadFeed.Subscribe(ch))
}
+// SubscribeBlockConfirmedEvent registers a subscription of ChainHeadEvent.
+func (bc *BlockChain) SubscribeBlockConfirmedEvent(ch chan<- BlockConfirmedEvent) event.Subscription {
+ return bc.scope.Track(bc.blockConfirmedFeed.Subscribe(ch))
+}
+
// SubscribeChainSideEvent registers a subscription of ChainSideEvent.
func (bc *BlockChain) SubscribeChainSideEvent(ch chan<- ChainSideEvent) event.Subscription {
return bc.scope.Track(bc.chainSideFeed.Subscribe(ch))
diff --git a/core/events.go b/core/events.go
index e76aa4784..1231daa37 100644
--- a/core/events.go
+++ b/core/events.go
@@ -47,6 +47,8 @@ type ChainSideEvent struct {
type ChainHeadEvent struct{ Block *types.Block }
+type BlockConfirmedEvent struct{ Block *types.Block }
+
type NewNotarySetEvent struct {
Round uint64
Pubkeys map[string]struct{} // pubkeys in hex format
diff --git a/core/tx_pool.go b/core/tx_pool.go
index eccf82e93..622c8ce9d 100644
--- a/core/tx_pool.go
+++ b/core/tx_pool.go
@@ -36,8 +36,8 @@ import (
)
const (
- // chainHeadChanSize is the size of channel listening to ChainHeadEvent.
- chainHeadChanSize = 10
+ // blockConfirmedChanSize is the size of channel listening to BlockConfirmedEvent.
+ blockConfirmedChanSize = 10
)
var (
@@ -118,7 +118,7 @@ type blockChain interface {
GetBlock(hash common.Hash, number uint64) *types.Block
StateAt(root common.Hash) (*state.StateDB, error)
- SubscribeChainHeadEvent(ch chan<- ChainHeadEvent) event.Subscription
+ SubscribeBlockConfirmedEvent(ch chan<- BlockConfirmedEvent) event.Subscription
}
// TxPoolConfig are the configuration parameters of the transaction pool.
@@ -203,16 +203,16 @@ func (config *TxPoolConfig) sanitize() TxPoolConfig {
// current state) and future transactions. Transactions move between those
// two states over time as they are received and processed.
type TxPool struct {
- config TxPoolConfig
- chainconfig *params.ChainConfig
- chain blockChain
- gasPrice *big.Int
- txFeed event.Feed
- scope event.SubscriptionScope
- chainHeadCh chan ChainHeadEvent
- chainHeadSub event.Subscription
- signer types.Signer
- mu sync.RWMutex
+ config TxPoolConfig
+ chainconfig *params.ChainConfig
+ chain blockChain
+ gasPrice *big.Int
+ txFeed event.Feed
+ scope event.SubscriptionScope
+ blockConfirmedCh chan BlockConfirmedEvent
+ blockConfirmedSub event.Subscription
+ signer types.Signer
+ mu sync.RWMutex
currentState *state.StateDB // Current state in the blockchain head
pendingState *state.ManagedState // Pending state tracking virtual nonces
@@ -240,16 +240,16 @@ func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, chain block
// Create the transaction pool with its initial settings
pool := &TxPool{
- config: config,
- chainconfig: chainconfig,
- chain: chain,
- signer: types.NewEIP155Signer(chainconfig.ChainID),
- pending: make(map[common.Address]*txList),
- queue: make(map[common.Address]*txList),
- beats: make(map[common.Address]time.Time),
- all: newTxLookup(),
- chainHeadCh: make(chan ChainHeadEvent, chainHeadChanSize),
- gasPrice: new(big.Int).SetUint64(config.PriceLimit),
+ config: config,
+ chainconfig: chainconfig,
+ chain: chain,
+ signer: types.NewEIP155Signer(chainconfig.ChainID),
+ pending: make(map[common.Address]*txList),
+ queue: make(map[common.Address]*txList),
+ beats: make(map[common.Address]time.Time),
+ all: newTxLookup(),
+ blockConfirmedCh: make(chan BlockConfirmedEvent, blockConfirmedChanSize),
+ gasPrice: new(big.Int).SetUint64(config.PriceLimit),
}
pool.locals = newAccountSet(pool.signer)
for _, addr := range config.Locals {
@@ -271,7 +271,7 @@ func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, chain block
}
}
// Subscribe events from blockchain
- pool.chainHeadSub = pool.chain.SubscribeChainHeadEvent(pool.chainHeadCh)
+ pool.blockConfirmedSub = pool.chain.SubscribeBlockConfirmedEvent(pool.blockConfirmedCh)
// Start the event loop and return
pool.wg.Add(1)
@@ -304,8 +304,8 @@ func (pool *TxPool) loop() {
// Keep waiting for and reacting to the various events
for {
select {
- // Handle ChainHeadEvent
- case ev := <-pool.chainHeadCh:
+ // Handle BlockConfirmedEvent
+ case ev := <-pool.blockConfirmedCh:
if ev.Block != nil {
pool.mu.Lock()
if pool.chainconfig.IsHomestead(ev.Block.Number()) {
@@ -317,7 +317,7 @@ func (pool *TxPool) loop() {
pool.mu.Unlock()
}
// Be unsubscribed due to system stopped
- case <-pool.chainHeadSub.Err():
+ case <-pool.blockConfirmedSub.Err():
return
// Handle stats reporting ticks
@@ -461,7 +461,7 @@ func (pool *TxPool) Stop() {
pool.scope.Close()
// Unsubscribe subscriptions registered from blockchain
- pool.chainHeadSub.Unsubscribe()
+ pool.blockConfirmedSub.Unsubscribe()
pool.wg.Wait()
if pool.journal != nil {
diff --git a/core/tx_pool_test.go b/core/tx_pool_test.go
index f43ba9e58..0ef926d1d 100644
--- a/core/tx_pool_test.go
+++ b/core/tx_pool_test.go
@@ -45,9 +45,9 @@ func init() {
}
type testBlockChain struct {
- statedb *state.StateDB
- gasLimit uint64
- chainHeadFeed *event.Feed
+ statedb *state.StateDB
+ gasLimit uint64
+ blockConfirmedFeed *event.Feed
}
func (bc *testBlockChain) CurrentBlock() *types.Block {
@@ -64,8 +64,8 @@ func (bc *testBlockChain) StateAt(common.Hash) (*state.StateDB, error) {
return bc.statedb, nil
}
-func (bc *testBlockChain) SubscribeChainHeadEvent(ch chan<- ChainHeadEvent) event.Subscription {
- return bc.chainHeadFeed.Subscribe(ch)
+func (bc *testBlockChain) SubscribeBlockConfirmedEvent(ch chan<- BlockConfirmedEvent) event.Subscription {
+ return bc.blockConfirmedFeed.Subscribe(ch)
}
func transaction(nonce uint64, gaslimit uint64, key *ecdsa.PrivateKey) *types.Transaction {