Skip to content

Commit

Permalink
core: separate transaction journal from pool
Browse files Browse the repository at this point in the history
  • Loading branch information
karalabe committed Jul 12, 2017
1 parent a117d37 commit bb2cdf8
Show file tree
Hide file tree
Showing 2 changed files with 184 additions and 114 deletions.
150 changes: 150 additions & 0 deletions core/tx_journal.go
Original file line number Diff line number Diff line change
@@ -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
}
148 changes: 34 additions & 114 deletions core/tx_pool.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,7 @@ package core
import (
"errors"
"fmt"
"io"
"math/big"
"os"
"sort"
"sync"
"time"
Expand All @@ -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"
)

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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()
}
}
}
Expand Down Expand Up @@ -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")
}
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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)
}
}
Expand Down Expand Up @@ -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
Expand Down

0 comments on commit bb2cdf8

Please sign in to comment.