core: fixed races in the txpool

core/tx_list.go

@@ -601,6 +601,7 @@ func (l *txPricedList) Discard(slots int, force bool) (types.Transactions, bool)
 func (l *txPricedList) Reheap() {
 	l.mu.Lock()
 	defer l.mu.Unlock()
+	l.reheap()
 }
 
 // reheap forcibly rebuilds the heap based on the current remote transaction set.

core/tx_pool.go

@@ -266,6 +266,7 @@ type TxPool struct {
 	reorgDoneCh        chan chan struct{}
 	reorgShutdownCh    chan struct{}  // requests shutdown of scheduleReorgLoop
 	wg                 sync.WaitGroup // tracks loop, scheduleReorgLoop
+	initDoneCh         chan struct{}  // is closed once the pool is initialized (for tests)
 
 	changesSinceReorg int // A counter for how many drops we've performed in-between reorg.
 }
@@ -297,6 +298,7 @@ func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, chain block
 		reorgDoneCh:        make(chan chan struct{}),
 		reorgShutdownCh:    make(chan struct{}),
 		updateBlockchainCh: make(chan blockChain),
+		initDoneCh:         make(chan struct{}),
 		gasPrice:           new(big.Int).SetUint64(config.PriceLimit),
 	}
 	pool.locals = newAccountSet(pool.signer)
@@ -350,6 +352,8 @@ func (pool *TxPool) loop() {
 	defer evict.Stop()
 	defer journal.Stop()
 
+	// Notify tests that the init phase is done
+	close(pool.initDoneCh)
 	for {
 		select {
 		// Handle ChainHeadEvent

core/tx_pool_test.go

@@ -124,7 +124,8 @@ func setupTxPoolWithConfig(config *params.ChainConfig) (*TxPool, *ecdsa.PrivateK
 	key, _ := crypto.GenerateKey()
 	pool := NewTxPool(testTxPoolConfig, config, blockchain)
 
-	pool.chainHeadCh <- ChainHeadEvent{}
+	// wait for the pool to initialize
+	<-pool.initDoneCh
 	return pool, key
 }
 
@@ -428,9 +429,7 @@ func TestTransactionChainFork(t *testing.T) {
 		statedb, _ := state.New(common.Hash{}, state.NewDatabase(rawdb.NewMemoryDatabase()), nil)
 		statedb.AddBalance(addr, big.NewInt(100000000000000))
 
-		pool.mu.Lock()
 		pool.chain = &testBlockChain{statedb, 1000000, new(event.Feed)}
-		pool.mu.Unlock()
 		<-pool.requestReset(nil, nil)
 	}
 	resetState()
@@ -459,9 +458,7 @@ func TestTransactionDoubleNonce(t *testing.T) {
 		statedb, _ := state.New(common.Hash{}, state.NewDatabase(rawdb.NewMemoryDatabase()), nil)
 		statedb.AddBalance(addr, big.NewInt(100000000000000))
 
-		pool.mu.Lock()
 		pool.chain = &testBlockChain{statedb, 1000000, new(event.Feed)}
-		pool.mu.Unlock()
 		<-pool.requestReset(nil, nil)
 	}
 	resetState()