diff --git a/core/tx_journal.go b/core/tx_journal.go new file mode 100644 index 0000000000..94a9ff9b8c --- /dev/null +++ b/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 . + +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 +} diff --git a/core/tx_pool.go b/core/tx_pool.go index 13bc398341..16f7742650 100644 --- a/core/tx_pool.go +++ b/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