core: separate transaction journal from pool

core/tx_journal.go

@@ -0,0 +1,150 @@
+// Copyright 2017 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 <http://www.gnu.org/licenses/>.
+
+package core
+
+import (
+	"errors"
+	"io"
+	"os"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/rlp"
+)
+
+// errNoActiveJournal is returned if a transaction is attempted to be inserted
+// into the journal, but no such file is currently open.
+var errNoActiveJournal = errors.New("no active journal")
+
+// txJournal is a rotating log of transactions with the aim of storing locally
+// created transactions to allow non-executed ones to survive node restarts.
+type txJournal struct {
+	path   string         // Filesystem path to store the transactions at
+	writer io.WriteCloser // Output stream to write new transactions into
+}
+
+// newTxJournal creates a new transaction journal to
+func newTxJournal(path string) *txJournal {
+	return &txJournal{
+		path: path,
+	}
+}
+
+// load parses a transaction journal dump from disk, loading its contents into
+// the specified pool.
+func (journal *txJournal) load(add func(*types.Transaction) error) error {
+	// Skip the parsing if the journal file doens't exist at all
+	if _, err := os.Stat(journal.path); os.IsNotExist(err) {
+		return nil
+	}
+	// Open the journal for loading any past transactions
+	input, err := os.Open(journal.path)
+	if err != nil {
+		return err
+	}
+	defer input.Close()
+
+	// Inject all transactions from the journal into the pool
+	stream := rlp.NewStream(input, 0)
+	total, dropped := 0, 0
+
+	var failure error
+	for {
+		// Parse the next transaction and terminate on error
+		tx := new(types.Transaction)
+		if err = stream.Decode(tx); err != nil {
+			if err != io.EOF {
+				failure = err
+			}
+			break
+		}
+		// Import the transaction and bump the appropriate progress counters
+		total++
+		if err = add(tx); err != nil {
+			log.Debug("Failed to add journaled transaction", "err", err)
+			dropped++
+			continue
+		}
+	}
+	log.Info("Loaded local transaction journal", "transactions", total, "dropped", dropped)
+
+	return failure
+}
+
+// insert adds the specified transaction to the local disk journal.
+func (journal *txJournal) insert(tx *types.Transaction) error {
+	if journal.writer == nil {
+		return errNoActiveJournal
+	}
+	if err := rlp.Encode(journal.writer, tx); err != nil {
+		return err
+	}
+	return nil
+}
+
+// rotate regenerates the transaction journal based on the current contents of
+// the transaction pool.
+func (journal *txJournal) rotate(all map[common.Address]types.Transactions) error {
+	// Close the current journal (if any is open)
+	if journal.writer != nil {
+		if err := journal.writer.Close(); err != nil {
+			return err
+		}
+		journal.writer = nil
+	}
+	// Generate a new journal with the contents of the current pool
+	replacement, err := os.OpenFile(journal.path+".new", os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0755)
+	if err != nil {
+		return err
+	}
+	journaled := 0
+	for _, txs := range all {
+		for _, tx := range txs {
+			if err = rlp.Encode(replacement, tx); err != nil {
+				replacement.Close()
+				return err
+			}
+		}
+		journaled += len(txs)
+	}
+	replacement.Close()
+
+	// Replace the live journal with the newly generated one
+	if err = os.Rename(journal.path+".new", journal.path); err != nil {
+		return err
+	}
+	sink, err := os.OpenFile(journal.path, os.O_WRONLY|os.O_APPEND, 0755)
+	if err != nil {
+		return err
+	}
+	journal.writer = sink
+	log.Info("Regenerated local transaction journal", "transactions", journaled, "accounts", len(all))
+
+	return nil
+}
+
+// close flushes the transaction journal contents to disk and closes the file.
+func (journal *txJournal) close() error {
+	var err error
+
+	if journal.writer != nil {
+		err = journal.writer.Close()
+		journal.writer = nil
+	}
+	return err
+}

core/tx_pool.go

@@ -19,9 +19,7 @@ package core
 import (
 	"errors"
 	"fmt"
-	"io"
 	"math/big"
-	"os"
 	"sort"
 	"sync"
 	"time"
@@ -33,7 +31,6 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethereum/go-ethereum/params"
-	"github.com/ethereum/go-ethereum/rlp"
 	"gopkg.in/karalabe/cookiejar.v2/collections/prque"
 )
 
@@ -172,8 +169,8 @@ type TxPool struct {
 	signer       types.Signer
 	mu           sync.RWMutex
 
-	locals  *accountSet    // Set of local transaction to exepmt from evicion rules
-	journal io.WriteCloser // Journal of local transaction to back up to disk
+	locals  *accountSet // Set of local transaction to exepmt from evicion rules
+	journal *txJournal  // Journal of local transaction to back up to disk
 
 	pending map[common.Address]*txList         // All currently processable transactions
 	queue   map[common.Address]*txList         // Queued but non-processable transactions
@@ -212,8 +209,16 @@ func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, eventMux *e
 	pool.priced = newTxPricedList(&pool.all)
 	pool.resetState()
 
-	if err := pool.rotateJournal(); err != nil {
-		log.Warn("Failed to load transaction journal", "err", err)
+	// If local transactions and journaling is enabled, load from disk
+	if !config.NoLocals && config.Journal != "" {
+		pool.journal = newTxJournal(config.Journal)
+
+		if err := pool.journal.load(pool.AddLocal); err != nil {
+			log.Warn("Failed to load transaction journal", "err", err)
+		}
+		if err := pool.journal.rotate(pool.local()); err != nil {
+			log.Warn("Failed to rotate transaction journal", "err", err)
+		}
 	}
 	// Start the event loop and return
 	pool.wg.Add(1)
@@ -294,11 +299,11 @@ func (pool *TxPool) loop() {
 
 		// Handle local transaction journal rotation
 		case <-journal.C:
-			pool.mu.Lock()
-			if err := pool.rotateJournal(); err != nil {
-				log.Warn("Failed to rotate local tx journal", "err", err)
+			if pool.journal != nil {
+				if err := pool.journal.rotate(pool.local()); err != nil {
+					log.Warn("Failed to rotate local tx journal", "err", err)
+				}
 			}
-			pool.mu.Unlock()
 		}
 	}
 }
@@ -333,7 +338,7 @@ func (pool *TxPool) Stop() {
 	pool.wg.Wait()
 
 	if pool.journal != nil {
-		pool.journal.Close()
+		pool.journal.close()
 	}
 	log.Info("Transaction pool stopped")
 }
@@ -421,6 +426,22 @@ func (pool *TxPool) Pending() (map[common.Address]types.Transactions, error) {
 	return pending, nil
 }
 
+// local retrieves all currently known local transactions, groupped by origin
+// account and sorted by nonce. The returned transaction set is a copy and can be
+// freely modified by calling code.
+func (pool *TxPool) local() map[common.Address]types.Transactions {
+	txs := make(map[common.Address]types.Transactions)
+	for addr := range pool.locals.accounts {
+		if pending := pool.pending[addr]; pending != nil {
+			txs[addr] = append(txs[addr], pending.Flatten()...)
+		}
+		if queued := pool.queue[addr]; queued != nil {
+			txs[addr] = append(txs[addr], queued.Flatten()...)
+		}
+	}
+	return txs
+}
+
 // validateTx checks whether a transaction is valid according to the consensus
 // rules and adheres to some heuristic limits of the local node (price and size).
 func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error {
@@ -574,7 +595,7 @@ func (pool *TxPool) journalTx(from common.Address, tx *types.Transaction) {
 	if pool.journal == nil || !pool.locals.contains(from) {
 		return
 	}
-	if err := rlp.Encode(pool.journal, tx); err != nil {
+	if err := pool.journal.insert(tx); err != nil {
 		log.Warn("Failed to journal local transaction", "err", err)
 	}
 }
@@ -974,107 +995,6 @@ func (pool *TxPool) demoteUnexecutables(state *state.StateDB) {
 	}
 }
 
-// rotateJournal regenerates the local transaction journal based on the current
-// contents of the transaction pool. If the pool doesn't have a journal opened,
-// it will first pull in the contents of any existing one from disk.
-func (pool *TxPool) rotateJournal() error {
-	// If local transactions or journaling are disabled, skip
-	if pool.config.NoLocals || pool.config.Journal == "" {
-		return nil
-	}
-	// If there's no journal open (first run), parse any existing one
-	if pool.journal == nil {
-		// Skip the parsing if the journal file doens't exist at all
-		if _, err := os.Stat(pool.config.Journal); !os.IsNotExist(err) {
-			// Open the journal for loading any past transactions
-			journal, err := os.Open(pool.config.Journal)
-			if err != nil {
-				return err
-			}
-			// Inject all transactions from the journal into the pool
-			stream := rlp.NewStream(journal, 0)
-			total, dropped := 0, 0
-
-			for {
-				tx := new(types.Transaction)
-				if err := stream.Decode(tx); err != nil {
-					if err == io.EOF {
-						break
-					}
-					return err
-				}
-				total++
-				if err := pool.addTx(tx, true); err != nil {
-					log.Debug("Failed to add journaled transaction", "err", err)
-					dropped++
-					continue
-				}
-			}
-			log.Info("Loaded local transaction journal", "transactions", total, "dropped", dropped)
-
-			// All transactions added, set the active journal
-			pool.journal = journal
-		}
-	}
-	// Close the current journal (if any is open)
-	if pool.journal != nil {
-		if err := pool.journal.Close(); err != nil {
-			return err
-		}
-		pool.journal = nil
-	}
-	// Generate a new journal with the contents of the current pool
-	replacement, err := os.OpenFile(pool.config.Journal+".new", os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0755)
-	if err != nil {
-		return err
-	}
-	accounts, journaled := 0, 0
-	for addr := range pool.locals.accounts {
-		// Flatten and serialize all the local executable transactions
-		pending := pool.pending[addr]
-		if pending != nil {
-			txs := pending.Flatten()
-			for _, tx := range txs {
-				if err = rlp.Encode(replacement, tx); err != nil {
-					replacement.Close()
-					return err
-				}
-			}
-			journaled += len(txs)
-		}
-		// Flatten and serialize all the local non-executable transactions
-		queued := pool.queue[addr]
-		if queued != nil {
-			txs := queued.Flatten()
-			for _, tx := range txs {
-				if err = rlp.Encode(replacement, tx); err != nil {
-					replacement.Close()
-					return err
-				}
-			}
-			journaled += len(txs)
-		}
-		// Bump the address counter if we did have transactions
-		if pending != nil || queued != nil {
-			accounts++
-		}
-	}
-	replacement.Close()
-
-	// Replace the live journal with the newly generated one
-	if err = os.Rename(pool.config.Journal+".new", pool.config.Journal); err != nil {
-		return err
-	}
-	journal, err := os.OpenFile(pool.config.Journal, os.O_WRONLY|os.O_APPEND, 0755)
-	if err != nil {
-		return err
-	}
-	pool.journal = journal
-	log.Info("Regenerated local transaction journal", "transactions", journaled, "accounts", accounts)
-
-	return nil
-}
-
 // addressByHeartbeat is an account address tagged with its last activity timestamp.
 type addressByHeartbeat struct {
 	address   common.Address