Merge pull request #769 from obscuren/develop

core: transaction queue
This commit is contained in:
Jeffrey Wilcke 2015-04-23 02:59:56 -07:00
commit 2f8809df40
12 changed files with 262 additions and 67 deletions

View File

@ -220,7 +220,7 @@ func (self *testFrontend) applyTxs() {
block := self.ethereum.ChainManager().NewBlock(cb) block := self.ethereum.ChainManager().NewBlock(cb)
coinbase := self.stateDb.GetStateObject(cb) coinbase := self.stateDb.GetStateObject(cb)
coinbase.SetGasPool(big.NewInt(10000000)) coinbase.SetGasPool(big.NewInt(10000000))
txs := self.ethereum.TxPool().GetTransactions() txs := self.ethereum.TxPool().GetQueuedTransactions()
for i := 0; i < len(txs); i++ { for i := 0; i < len(txs); i++ {
for _, tx := range txs { for _, tx := range txs {

View File

@ -258,7 +258,7 @@ func (sm *BlockProcessor) processWithParent(block, parent *types.Block) (logs st
state.Sync() state.Sync()
// Remove transactions from the pool // Remove transactions from the pool
sm.txpool.RemoveSet(block.Transactions()) sm.txpool.RemoveTransactions(block.Transactions())
// This puts transactions in a extra db for rpc // This puts transactions in a extra db for rpc
for i, tx := range block.Transactions() { for i, tx := range block.Transactions() {

View File

@ -108,7 +108,9 @@ func makeChain(bman *BlockProcessor, parent *types.Block, max int, db common.Dat
// Create a new chain manager starting from given block // Create a new chain manager starting from given block
// Effectively a fork factory // Effectively a fork factory
func newChainManager(block *types.Block, eventMux *event.TypeMux, db common.Database) *ChainManager { func newChainManager(block *types.Block, eventMux *event.TypeMux, db common.Database) *ChainManager {
bc := &ChainManager{blockDb: db, stateDb: db, genesisBlock: GenesisBlock(db), eventMux: eventMux} genesis := GenesisBlock(db)
bc := &ChainManager{blockDb: db, stateDb: db, genesisBlock: genesis, eventMux: eventMux}
bc.txState = state.ManageState(state.New(genesis.Root(), db))
bc.futureBlocks = NewBlockCache(1000) bc.futureBlocks = NewBlockCache(1000)
if block == nil { if block == nil {
bc.Reset() bc.Reset()

View File

@ -576,7 +576,7 @@ func (self *ChainManager) InsertChain(chain types.Blocks) error {
}) })
self.setTransState(state.New(block.Root(), self.stateDb)) self.setTransState(state.New(block.Root(), self.stateDb))
self.setTxState(state.New(block.Root(), self.stateDb)) self.txState.SetState(state.New(block.Root(), self.stateDb))
queue[i] = ChainEvent{block, logs} queue[i] = ChainEvent{block, logs}
queueEvent.canonicalCount++ queueEvent.canonicalCount++

View File

@ -81,7 +81,7 @@ func (err *NonceErr) Error() string {
} }
func NonceError(is, exp uint64) *NonceErr { func NonceError(is, exp uint64) *NonceErr {
return &NonceErr{Message: fmt.Sprintf("Transaction w/ invalid nonce (%d / %d)", is, exp), Is: is, Exp: exp} return &NonceErr{Message: fmt.Sprintf("Transaction w/ invalid nonce. tx=%d state=%d)", is, exp), Is: is, Exp: exp}
} }
func IsNonceErr(err error) bool { func IsNonceErr(err error) bool {

View File

@ -4,7 +4,9 @@ import (
"errors" "errors"
"fmt" "fmt"
"math/big" "math/big"
"sort"
"sync" "sync"
"time"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/state"
@ -17,7 +19,7 @@ import (
var ( var (
ErrInvalidSender = errors.New("Invalid sender") ErrInvalidSender = errors.New("Invalid sender")
ErrImpossibleNonce = errors.New("Impossible nonce") ErrNonce = errors.New("Nonce too low")
ErrNonExistentAccount = errors.New("Account does not exist") ErrNonExistentAccount = errors.New("Account does not exist")
ErrInsufficientFunds = errors.New("Insufficient funds") ErrInsufficientFunds = errors.New("Insufficient funds")
ErrIntrinsicGas = errors.New("Intrinsic gas too low") ErrIntrinsicGas = errors.New("Intrinsic gas too low")
@ -54,20 +56,43 @@ type TxPool struct {
txs map[common.Hash]*types.Transaction txs map[common.Hash]*types.Transaction
invalidHashes *set.Set invalidHashes *set.Set
queue map[common.Address]types.Transactions
subscribers []chan TxMsg subscribers []chan TxMsg
eventMux *event.TypeMux eventMux *event.TypeMux
} }
func NewTxPool(eventMux *event.TypeMux, currentStateFn stateFn) *TxPool { func NewTxPool(eventMux *event.TypeMux, currentStateFn stateFn) *TxPool {
return &TxPool{ txPool := &TxPool{
txs: make(map[common.Hash]*types.Transaction), txs: make(map[common.Hash]*types.Transaction),
queue: make(map[common.Address]types.Transactions),
queueChan: make(chan *types.Transaction, txPoolQueueSize), queueChan: make(chan *types.Transaction, txPoolQueueSize),
quit: make(chan bool), quit: make(chan bool),
eventMux: eventMux, eventMux: eventMux,
invalidHashes: set.New(), invalidHashes: set.New(),
currentState: currentStateFn, currentState: currentStateFn,
} }
return txPool
}
func (pool *TxPool) Start() {
// Queue timer will tick so we can attempt to move items from the queue to the
// main transaction pool.
queueTimer := time.NewTicker(300 * time.Millisecond)
// Removal timer will tick and attempt to remove bad transactions (account.nonce>tx.nonce)
removalTimer := time.NewTicker(1 * time.Second)
done:
for {
select {
case <-queueTimer.C:
pool.checkQueue()
case <-removalTimer.C:
pool.validatePool()
case <-pool.quit:
break done
}
}
} }
func (pool *TxPool) ValidateTransaction(tx *types.Transaction) error { func (pool *TxPool) ValidateTransaction(tx *types.Transaction) error {
@ -100,16 +125,12 @@ func (pool *TxPool) ValidateTransaction(tx *types.Transaction) error {
} }
if pool.currentState().GetNonce(from) > tx.Nonce() { if pool.currentState().GetNonce(from) > tx.Nonce() {
return ErrImpossibleNonce return ErrNonce
} }
return nil return nil
} }
func (self *TxPool) addTx(tx *types.Transaction) {
self.txs[tx.Hash()] = tx
}
func (self *TxPool) add(tx *types.Transaction) error { func (self *TxPool) add(tx *types.Transaction) error {
hash := tx.Hash() hash := tx.Hash()
@ -127,7 +148,7 @@ func (self *TxPool) add(tx *types.Transaction) error {
return err return err
} }
self.addTx(tx) self.queueTx(tx)
var toname string var toname string
if to := tx.To(); to != nil { if to := tx.To(); to != nil {
@ -144,9 +165,6 @@ func (self *TxPool) add(tx *types.Transaction) error {
glog.Infof("(t) %x => %s (%v) %x\n", from, toname, tx.Value, tx.Hash()) glog.Infof("(t) %x => %s (%v) %x\n", from, toname, tx.Value, tx.Hash())
} }
// Notify the subscribers
go self.eventMux.Post(TxPreEvent{tx})
return nil return nil
} }
@ -189,34 +207,108 @@ func (self *TxPool) GetTransactions() (txs types.Transactions) {
return return
} }
func (self *TxPool) RemoveSet(txs types.Transactions) { func (self *TxPool) GetQueuedTransactions() types.Transactions {
self.mu.RLock()
defer self.mu.RUnlock()
var txs types.Transactions
for _, ts := range self.queue {
txs = append(txs, ts...)
}
return txs
}
func (self *TxPool) RemoveTransactions(txs types.Transactions) {
self.mu.Lock() self.mu.Lock()
defer self.mu.Unlock() defer self.mu.Unlock()
for _, tx := range txs { for _, tx := range txs {
delete(self.txs, tx.Hash()) delete(self.txs, tx.Hash())
} }
} }
func (self *TxPool) InvalidateSet(hashes *set.Set) {
self.mu.Lock()
defer self.mu.Unlock()
hashes.Each(func(v interface{}) bool {
delete(self.txs, v.(common.Hash))
return true
})
self.invalidHashes.Merge(hashes)
}
func (pool *TxPool) Flush() { func (pool *TxPool) Flush() {
pool.txs = make(map[common.Hash]*types.Transaction) pool.txs = make(map[common.Hash]*types.Transaction)
} }
func (pool *TxPool) Start() {
}
func (pool *TxPool) Stop() { func (pool *TxPool) Stop() {
pool.Flush() pool.Flush()
close(pool.quit)
glog.V(logger.Info).Infoln("TX Pool stopped") glog.V(logger.Info).Infoln("TX Pool stopped")
} }
func (self *TxPool) queueTx(tx *types.Transaction) {
from, _ := tx.From()
self.queue[from] = append(self.queue[from], tx)
}
func (pool *TxPool) addTx(tx *types.Transaction) {
if _, ok := pool.txs[tx.Hash()]; !ok {
pool.txs[tx.Hash()] = tx
// Notify the subscribers. This event is posted in a goroutine
// because it's possible that somewhere during the post "Remove transaction"
// gets called which will then wait for the global tx pool lock and deadlock.
go pool.eventMux.Post(TxPreEvent{tx})
}
}
// check queue will attempt to insert
func (pool *TxPool) checkQueue() {
pool.mu.Lock()
defer pool.mu.Unlock()
statedb := pool.currentState()
for address, txs := range pool.queue {
sort.Sort(types.TxByNonce{txs})
var (
nonce = statedb.GetNonce(address)
start int
)
// Clean up the transactions first and determine the start of the nonces
for _, tx := range txs {
if tx.Nonce() >= nonce {
break
}
start++
}
pool.queue[address] = txs[start:]
// expected nonce
enonce := nonce
for _, tx := range pool.queue[address] {
// If the expected nonce does not match up with the next one
// (i.e. a nonce gap), we stop the loop
if enonce != tx.Nonce() {
break
}
enonce++
pool.addTx(tx)
}
//pool.queue[address] = txs[i:]
// delete the entire queue entry if it's empty. There's no need to keep it
if len(pool.queue[address]) == 0 {
delete(pool.queue, address)
}
}
}
func (pool *TxPool) validatePool() {
pool.mu.Lock()
defer pool.mu.Unlock()
statedb := pool.currentState()
for hash, tx := range pool.txs {
from, _ := tx.From()
if nonce := statedb.GetNonce(from); nonce > tx.Nonce() {
if glog.V(logger.Debug) {
glog.Infof("removed tx (%x) from pool due to nonce error. state=%d tx=%d\n", hash[:4], nonce, tx.Nonce())
}
delete(pool.txs, hash)
}
}
}

View File

@ -56,7 +56,57 @@ func TestInvalidTransactions(t *testing.T) {
tx.SignECDSA(key) tx.SignECDSA(key)
err = pool.Add(tx) err = pool.Add(tx)
if err != ErrImpossibleNonce { if err != ErrNonce {
t.Error("expected", ErrImpossibleNonce) t.Error("expected", ErrNonce)
}
}
func TestTransactionQueue(t *testing.T) {
pool, key := setupTxPool()
tx := transaction()
tx.SignECDSA(key)
from, _ := tx.From()
pool.currentState().AddBalance(from, big.NewInt(1))
pool.queueTx(tx)
pool.checkQueue()
if len(pool.txs) != 1 {
t.Error("expected valid txs to be 1 is", len(pool.txs))
}
tx = transaction()
tx.SignECDSA(key)
from, _ = tx.From()
pool.currentState().SetNonce(from, 10)
tx.SetNonce(1)
pool.queueTx(tx)
pool.checkQueue()
if _, ok := pool.txs[tx.Hash()]; ok {
t.Error("expected transaction to be in tx pool")
}
if len(pool.queue[from]) != 0 {
t.Error("expected transaction queue to be empty. is", len(pool.queue[from]))
}
pool, key = setupTxPool()
tx1, tx2, tx3 := transaction(), transaction(), transaction()
tx2.SetNonce(10)
tx3.SetNonce(11)
tx1.SignECDSA(key)
tx2.SignECDSA(key)
tx3.SignECDSA(key)
pool.queueTx(tx1)
pool.queueTx(tx2)
pool.queueTx(tx3)
from, _ = tx1.From()
pool.checkQueue()
if len(pool.txs) != 1 {
t.Error("expected tx pool to be 1 =")
}
if len(pool.queue[from]) != 3 {
t.Error("expected transaction queue to be empty. is", len(pool.queue[from]))
} }
} }

View File

@ -3,7 +3,6 @@ package eth
import ( import (
"crypto/ecdsa" "crypto/ecdsa"
"fmt" "fmt"
"math"
"path" "path"
"strings" "strings"
@ -136,11 +135,10 @@ type Ethereum struct {
protocolManager *ProtocolManager protocolManager *ProtocolManager
downloader *downloader.Downloader downloader *downloader.Downloader
net *p2p.Server net *p2p.Server
eventMux *event.TypeMux eventMux *event.TypeMux
txSub event.Subscription txSub event.Subscription
minedBlockSub event.Subscription miner *miner.Miner
miner *miner.Miner
// logger logger.LogSystem // logger logger.LogSystem
@ -222,7 +220,7 @@ func New(config *Config) (*Ethereum, error) {
eth.whisper = whisper.New() eth.whisper = whisper.New()
eth.shhVersionId = int(eth.whisper.Version()) eth.shhVersionId = int(eth.whisper.Version())
eth.miner = miner.New(eth, eth.pow, config.MinerThreads) eth.miner = miner.New(eth, eth.pow, config.MinerThreads)
eth.protocolManager = NewProtocolManager(config.ProtocolVersion, config.NetworkId, eth.txPool, eth.chainManager, eth.downloader) eth.protocolManager = NewProtocolManager(config.ProtocolVersion, config.NetworkId, eth.eventMux, eth.txPool, eth.chainManager, eth.downloader)
netprv, err := config.nodeKey() netprv, err := config.nodeKey()
if err != nil { if err != nil {
@ -379,7 +377,8 @@ func (s *Ethereum) Start() error {
} }
// Start services // Start services
s.txPool.Start() go s.txPool.Start()
s.protocolManager.Start()
if s.whisper != nil { if s.whisper != nil {
s.whisper.Start() s.whisper.Start()
@ -389,10 +388,6 @@ func (s *Ethereum) Start() error {
s.txSub = s.eventMux.Subscribe(core.TxPreEvent{}) s.txSub = s.eventMux.Subscribe(core.TxPreEvent{})
go s.txBroadcastLoop() go s.txBroadcastLoop()
// broadcast mined blocks
s.minedBlockSub = s.eventMux.Subscribe(core.NewMinedBlockEvent{})
go s.minedBroadcastLoop()
glog.V(logger.Info).Infoln("Server started") glog.V(logger.Info).Infoln("Server started")
return nil return nil
} }
@ -422,9 +417,9 @@ func (s *Ethereum) Stop() {
defer s.stateDb.Close() defer s.stateDb.Close()
defer s.extraDb.Close() defer s.extraDb.Close()
s.txSub.Unsubscribe() // quits txBroadcastLoop s.txSub.Unsubscribe() // quits txBroadcastLoop
s.minedBlockSub.Unsubscribe() // quits blockBroadcastLoop
s.protocolManager.Stop()
s.txPool.Stop() s.txPool.Stop()
s.eventMux.Stop() s.eventMux.Stop()
if s.whisper != nil { if s.whisper != nil {
@ -440,13 +435,10 @@ func (s *Ethereum) WaitForShutdown() {
<-s.shutdownChan <-s.shutdownChan
} }
// now tx broadcasting is taken out of txPool
// handled here via subscription, efficiency?
func (self *Ethereum) txBroadcastLoop() { func (self *Ethereum) txBroadcastLoop() {
// automatically stops if unsubscribe // automatically stops if unsubscribe
for obj := range self.txSub.Chan() { for obj := range self.txSub.Chan() {
event := obj.(core.TxPreEvent) event := obj.(core.TxPreEvent)
self.net.BroadcastLimited("eth", TxMsg, math.Sqrt, []*types.Transaction{event.Tx})
self.syncAccounts(event.Tx) self.syncAccounts(event.Tx)
} }
} }
@ -465,16 +457,6 @@ func (self *Ethereum) syncAccounts(tx *types.Transaction) {
} }
} }
func (self *Ethereum) minedBroadcastLoop() {
// automatically stops if unsubscribe
for obj := range self.minedBlockSub.Chan() {
switch ev := obj.(type) {
case core.NewMinedBlockEvent:
self.protocolManager.BroadcastBlock(ev.Block.Hash(), ev.Block)
}
}
}
func saveProtocolVersion(db common.Database, protov int) { func saveProtocolVersion(db common.Database, protov int) {
d, _ := db.Get([]byte("ProtocolVersion")) d, _ := db.Get([]byte("ProtocolVersion"))
protocolVersion := common.NewValue(d).Uint() protocolVersion := common.NewValue(d).Uint()

View File

@ -44,6 +44,7 @@ import (
"github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/eth/downloader" "github.com/ethereum/go-ethereum/eth/downloader"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/logger/glog"
"github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p"
@ -77,12 +78,17 @@ type ProtocolManager struct {
peers map[string]*peer peers map[string]*peer
SubProtocol p2p.Protocol SubProtocol p2p.Protocol
eventMux *event.TypeMux
txSub event.Subscription
minedBlockSub event.Subscription
} }
// NewProtocolManager returns a new ethereum sub protocol manager. The Ethereum sub protocol manages peers capable // NewProtocolManager returns a new ethereum sub protocol manager. The Ethereum sub protocol manages peers capable
// with the ethereum network. // with the ethereum network.
func NewProtocolManager(protocolVersion, networkId int, txpool txPool, chainman *core.ChainManager, downloader *downloader.Downloader) *ProtocolManager { func NewProtocolManager(protocolVersion, networkId int, mux *event.TypeMux, txpool txPool, chainman *core.ChainManager, downloader *downloader.Downloader) *ProtocolManager {
manager := &ProtocolManager{ manager := &ProtocolManager{
eventMux: mux,
txpool: txpool, txpool: txpool,
chainman: chainman, chainman: chainman,
downloader: downloader, downloader: downloader,
@ -105,6 +111,21 @@ func NewProtocolManager(protocolVersion, networkId int, txpool txPool, chainman
return manager return manager
} }
func (pm *ProtocolManager) Start() {
// broadcast transactions
pm.txSub = pm.eventMux.Subscribe(core.TxPreEvent{})
go pm.txBroadcastLoop()
// broadcast mined blocks
pm.minedBlockSub = pm.eventMux.Subscribe(core.NewMinedBlockEvent{})
go pm.minedBroadcastLoop()
}
func (pm *ProtocolManager) Stop() {
pm.txSub.Unsubscribe() // quits txBroadcastLoop
pm.minedBlockSub.Unsubscribe() // quits blockBroadcastLoop
}
func (pm *ProtocolManager) newPeer(pv, nv int, p *p2p.Peer, rw p2p.MsgReadWriter) *peer { func (pm *ProtocolManager) newPeer(pv, nv int, p *p2p.Peer, rw p2p.MsgReadWriter) *peer {
td, current, genesis := pm.chainman.Status() td, current, genesis := pm.chainman.Status()
@ -326,10 +347,51 @@ func (pm *ProtocolManager) BroadcastBlock(hash common.Hash, block *types.Block)
} }
} }
// Broadcast block to peer set // Broadcast block to peer set
// XXX due to the current shit state of the network disable the limit
peers = peers[:int(math.Sqrt(float64(len(peers))))] peers = peers[:int(math.Sqrt(float64(len(peers))))]
for _, peer := range peers { for _, peer := range peers {
peer.sendNewBlock(block) peer.sendNewBlock(block)
} }
glog.V(logger.Detail).Infoln("broadcast block to", len(peers), "peers") glog.V(logger.Detail).Infoln("broadcast block to", len(peers), "peers")
} }
// BroadcastTx will propagate the block to its connected peers. It will sort
// out which peers do not contain the block in their block set and will do a
// sqrt(peers) to determine the amount of peers we broadcast to.
func (pm *ProtocolManager) BroadcastTx(hash common.Hash, tx *types.Transaction) {
pm.pmu.Lock()
defer pm.pmu.Unlock()
// Find peers who don't know anything about the given hash. Peers that
// don't know about the hash will be a candidate for the broadcast loop
var peers []*peer
for _, peer := range pm.peers {
if !peer.txHashes.Has(hash) {
peers = append(peers, peer)
}
}
// Broadcast block to peer set
peers = peers[:int(math.Sqrt(float64(len(peers))))]
for _, peer := range peers {
peer.sendTransaction(tx)
}
glog.V(logger.Detail).Infoln("broadcast tx to", len(peers), "peers")
}
// Mined broadcast loop
func (self *ProtocolManager) minedBroadcastLoop() {
// automatically stops if unsubscribe
for obj := range self.minedBlockSub.Chan() {
switch ev := obj.(type) {
case core.NewMinedBlockEvent:
self.BroadcastBlock(ev.Block.Hash(), ev.Block)
}
}
}
func (self *ProtocolManager) txBroadcastLoop() {
// automatically stops if unsubscribe
for obj := range self.txSub.Chan() {
event := obj.(core.TxPreEvent)
self.BroadcastTx(event.Tx.Hash(), event.Tx)
}
}

View File

@ -86,6 +86,12 @@ func (p *peer) sendNewBlock(block *types.Block) error {
return p2p.Send(p.rw, NewBlockMsg, []interface{}{block, block.Td}) return p2p.Send(p.rw, NewBlockMsg, []interface{}{block, block.Td})
} }
func (p *peer) sendTransaction(tx *types.Transaction) error {
p.txHashes.Add(tx.Hash())
return p2p.Send(p.rw, TxMsg, []*types.Transaction{tx})
}
func (p *peer) requestHashes(from common.Hash) error { func (p *peer) requestHashes(from common.Hash) error {
glog.V(logger.Debug).Infof("[%s] fetching hashes (%d) %x...\n", p.id, maxHashes, from[:4]) glog.V(logger.Debug).Infof("[%s] fetching hashes (%d) %x...\n", p.id, maxHashes, from[:4])
return p2p.Send(p.rw, GetBlockHashesMsg, getBlockHashesMsgData{from, maxHashes}) return p2p.Send(p.rw, GetBlockHashesMsg, getBlockHashesMsgData{from, maxHashes})

View File

@ -258,7 +258,7 @@ func (self *worker) commitNewWork() {
tcount = 0 tcount = 0
ignoredTransactors = set.New() ignoredTransactors = set.New()
) )
//gasLimit:
for _, tx := range transactions { for _, tx := range transactions {
// We can skip err. It has already been validated in the tx pool // We can skip err. It has already been validated in the tx pool
from, _ := tx.From() from, _ := tx.From()
@ -290,13 +290,12 @@ func (self *worker) commitNewWork() {
// ignore the transactor so no nonce errors will be thrown for this account // ignore the transactor so no nonce errors will be thrown for this account
// next time the worker is run, they'll be picked up again. // next time the worker is run, they'll be picked up again.
ignoredTransactors.Add(from) ignoredTransactors.Add(from)
//glog.V(logger.Debug).Infof("Gas limit reached for block. %d TXs included in this block\n", i)
//break gasLimit glog.V(logger.Detail).Infof("Gas limit reached for (%x) in this block. Continue to try smaller txs\n", from[:4])
default: default:
tcount++ tcount++
} }
} }
//self.eth.TxPool().InvalidateSet(remove)
var ( var (
uncles []*types.Header uncles []*types.Header

View File

@ -682,9 +682,11 @@ func (self *XEth) Transact(fromStr, toStr, valueStr, gasStr, gasPriceStr, codeSt
if contractCreation { if contractCreation {
addr := core.AddressFromMessage(tx) addr := core.AddressFromMessage(tx)
glog.V(logger.Info).Infof("Contract addr %x\n", addr) glog.V(logger.Info).Infof("Tx(%x) created: %x\n", tx.Hash(), addr)
return core.AddressFromMessage(tx).Hex(), nil return core.AddressFromMessage(tx).Hex(), nil
} else {
glog.V(logger.Info).Infof("Tx(%x) to: %x\n", tx.Hash(), tx.To())
} }
return tx.Hash().Hex(), nil return tx.Hash().Hex(), nil
} }