move TxPool reorg and events to background goroutine (#378)

* move TxPool reorg and events to background goroutine

* modify txpool handler test

blockchain/genesis.go

@@ -24,17 +24,16 @@ import (
 	"strings"
 	"time"
 
-	"github.com/fractalplatform/fractal/consensus/dpos"
-	"github.com/fractalplatform/fractal/snapshot"
-
 	"github.com/ethereum/go-ethereum/log"
 	am "github.com/fractalplatform/fractal/accountmanager"
 	at "github.com/fractalplatform/fractal/asset"
 	"github.com/fractalplatform/fractal/common"
+	"github.com/fractalplatform/fractal/consensus/dpos"
 	fm "github.com/fractalplatform/fractal/feemanager"
 	"github.com/fractalplatform/fractal/p2p/enode"
 	"github.com/fractalplatform/fractal/params"
 	"github.com/fractalplatform/fractal/rawdb"
+	"github.com/fractalplatform/fractal/snapshot"
 	"github.com/fractalplatform/fractal/state"
 	"github.com/fractalplatform/fractal/types"
 	"github.com/fractalplatform/fractal/utils/fdb"

rpcapi/p2p.go

@@ -132,13 +132,19 @@ func (api *PrivateP2pAPI) BadNodes() []string {
 }
 
 // AddBadNode add a bad node
-func (api *PrivateP2pAPI) AddBadNode(url string) error {
-	return api.b.AddBadNode(url)
+func (api *PrivateP2pAPI) AddBadNode(url string) (bool, error) {
+	if err := api.b.AddBadNode(url); err != nil {
+		return false, fmt.Errorf("invalid enode: %v", err)
+	}
+	return true, nil
 }
 
 // RemoveBadNode remove a bad node
-func (api *PrivateP2pAPI) RemoveBadNode(url string) error {
-	return api.b.RemoveBadNode(url)
+func (api *PrivateP2pAPI) RemoveBadNode(url string) (bool, error) {
+	if err := api.b.RemoveBadNode(url); err != nil {
+		return false, fmt.Errorf("invalid enode: %v", err)
+	}
+	return true, nil
 }
 
 // SelfNode return self enode url

txpool/accountset.go

@@ -24,14 +24,22 @@ import (
 // accountSet is simply a set of name to check for existence
 type accountSet struct {
 	accounts map[common.Name]struct{}
+	cache    *[]common.Name
 }
 
 // newAccountSet creates a new name set with an associated signer for sender
 // derivations.
-func newAccountSet(signer types.Signer) *accountSet {
-	return &accountSet{
-		accounts: make(map[common.Name]struct{}),
+func newAccountSet(signer types.Signer, names ...common.Name) *accountSet {
+	as := &accountSet{accounts: make(map[common.Name]struct{})}
+	for _, name := range names {
+		as.add(name)
 	}
+	return as
+}
+
+// addTx adds the sender of tx into the set.
+func (as *accountSet) addTx(tx *types.Transaction) {
+	as.add(tx.GetActions()[0].Sender())
 }
 
 // contains checks if a given name is contained within the set.
@@ -49,4 +57,26 @@ func (as *accountSet) containsName(tx *types.Transaction) bool {
 // add inserts a new name into the set to track.
 func (as *accountSet) add(name common.Name) {
 	as.accounts[name] = struct{}{}
+	as.cache = nil
+}
+
+// flatten returns the list of addresses within this set, also caching it for later
+// reuse. The returned slice should not be changed!
+func (as *accountSet) flatten() []common.Name {
+	if as.cache == nil {
+		accounts := make([]common.Name, 0, len(as.accounts))
+		for account := range as.accounts {
+			accounts = append(accounts, account)
+		}
+		as.cache = &accounts
+	}
+	return *as.cache
+}
+
+// merge adds all addresses from the 'other' set into 'as'.
+func (as *accountSet) merge(other *accountSet) {
+	for addr := range other.accounts {
+		as.accounts[addr] = struct{}{}
+	}
+	as.cache = nil
 }

txpool/config.go

@@ -96,6 +96,10 @@ func (config *Config) check() Config {
 		log.Warn("Sanitizing invalid txpool lifetime", "provided", conf.Lifetime, "updated", DefaultTxPoolConfig.Lifetime)
 		conf.Lifetime = DefaultTxPoolConfig.Lifetime
 	}
+	if conf.ResendTime < 1 {
+		log.Warn("Sanitizing invalid txpool resendtime", "provided", conf.ResendTime, "updated", DefaultTxPoolConfig.ResendTime)
+		conf.ResendTime = DefaultTxPoolConfig.ResendTime
+	}
 	if conf.RatioBroadcast < 1 {
 		log.Warn("Sanitizing invalid txpool ratiobroadcast", "provided", conf.RatioBroadcast, "updated", DefaultTxPoolConfig.RatioBroadcast)
 		conf.RatioBroadcast = DefaultTxPoolConfig.RatioBroadcast

txpool/handler_test.go

@@ -157,7 +157,7 @@ func TestP2PTxMsg(t *testing.T) {
 
 	event.SendTo(event.NewLocalStation("test", nil), nil, event.P2PTxMsg, txs)
 	for {
-		if pending, quened := pool.Stats(); pending > 0 || quened > 0 {
+		if pending, _ := pool.Stats(); pending > 0 {
 			break
 		}
 	}
@@ -172,7 +172,7 @@ func TestP2PTxMsg(t *testing.T) {
 
 	// trigger state change in the background
 	trigger = true
-	pool.lockedReset(nil, nil)
+	pool.requestReset(nil, nil)
 
 	_, err = pool.Pending()
 	if err != nil {

txpool/pricedlsit.go

@@ -49,9 +49,9 @@ func (l *txPricedList) Put(tx *types.Transaction) {
 // Removed notifies the prices transaction list that an old transaction dropped
 // from the pool. The list will just keep a counter of stale objects and update
 // the heap if a large enough ratio of transactions go stale.
-func (l *txPricedList) Removed() {
+func (l *txPricedList) Removed(count int) {
 	// Bump the stale counter, but exit if still too low (< 25%)
-	l.stales++
+	l.stales += count
 	if l.stales <= len(*l.items)/4 {
 		return
 	}

txpool/test_utils.go

@@ -49,6 +49,7 @@ func init() {
 		AccountQueue: 64,
 		GlobalQueue:  1024,
 		Lifetime:     3 * time.Hour,
+		ResendTime:   10 * time.Minute,
 		GasAssetID:   uint64(0),
 	}
 }