开启Geth node 和console
./build/bin/geth --datadir walker/ --miner.etherbase 0x106cdbb5029d186ae6f659b52fb83179fc153e6a --mine --miner.threads 1 --rpc --rpcvhosts "*" --rpcaddr 0.0.0.0 --rpcport 8545 --rpccorsdomain "*" --rpcapi "db,eth,net,web3,personal,debug"
在console转账
//解锁account
personal.unlockAccount("0x106cdbb5029d186ae6f659b52fb83179fc153e6a","walker")
true
//转账交易
eth.sendTransaction({from:'0x106cdbb5029d186ae6f659b52fb83179fc153e6a',to:'0x3b7bfcffd7a8c1047055edc58e2efa6eb589184f',value:web3.toWei(100,"ether")})
"0x21e5e3e9968d4bcf99402923b168bc172cadcc0fa4b27885d07ff52eac5a4f2b"
node处理流程
通过 console.Interactive() -> c.Evaluate(input)
-> simplechain/internal/ethapi.go
func (s *PublicTransactionPoolAPI) SendTransaction(ctx context.Context, args SendTxArgs) (common.Hash, error) {
// Look up the wallet containing the requested signer
account := accounts.Account{Address: args.From}
wallet, err := s.b.AccountManager().Find(account)
if err != nil {
return common.Hash{}, err
}
if args.Nonce == nil {
// Hold the addresse's mutex around signing to prevent concurrent assignment of
// the same nonce to multiple accounts.
s.nonceLock.LockAddr(args.From)
defer s.nonceLock.UnlockAddr(args.From)
}
// Set some sanity defaults and terminate on failure
if err := args.setDefaults(ctx, s.b); err != nil {
return common.Hash{}, err
}
// Assemble the transaction and sign with the wallet
tx := args.toTransaction()
var chainID *big.Int
if config := s.b.ChainConfig(); config.IsEIP155(s.b.CurrentBlock().Number()) {
chainID = config.ChainID
}
signed, err := wallet.SignTx(account, tx, chainID)
if err != nil {
return common.Hash{}, err
}
return submitTransaction(ctx, s.b, signed)
}
// submitTransaction is a helper function that submits tx to txPool and logs a message.
func submitTransaction(ctx context.Context, b Backend, tx *types.Transaction) (common.Hash, error) {
if err := b.SendTx(ctx, tx); err != nil {
return common.Hash{}, err
}
if tx.To() == nil {
signer := types.MakeSigner(b.ChainConfig(), b.CurrentBlock().Number())
from, err := types.Sender(signer, tx)
if err != nil {
return common.Hash{}, err
}
addr := crypto.CreateAddress(from, tx.Nonce())
log.Info("Submitted contract creation", "fullhash", tx.Hash().Hex(), "contract", addr.Hex())
} else {
log.Info("Submitted transaction", "fullhash", tx.Hash().Hex(), "recipient", tx.To())
}
return tx.Hash(), nil
}
sendTrx() -> simplechain/core/tx_pool.go
// addTx enqueues a single transaction into the pool if it is valid.
func (pool *TxPool) addTx(tx *types.Transaction, local bool) error {
pool.mu.Lock()
defer pool.mu.Unlock()
// Try to inject the transaction and update any state
replace, err := pool.add(tx, local)
if err != nil {
return err
}
// If we added a new transaction, run promotion checks and return
if !replace {
from, _ := types.Sender(pool.signer, tx) // already validated
pool.promoteExecutables([]common.Address{from})
}
return nil
}
以上代码流程比较清晰。至此node已经将交易放入交易池。
向交易池添加交易
1、添加交易TxPool.add(), add()方法用于将本地或远端的交易加入到交易池，这个方法的基本逻辑是：
1）检查交易是否收到过，重复接受的交易直接丢弃；
2）验证交易是否有效；
3）如果交易池满了，待插入的交易的价值比交易池中任意一个都低，则直接丢弃；
4）如果待插入的交易序号在pending列表中已经存在，且待插入的交易价值大于或等于原交易的110%，则替换原交易；
5）如果待插入的交易序号在pending列表中没有，则直接放入queue列表。如果对应的序号已经有交易了，则如果新交易的价值大于或等于原交易的110%，替换原交易；
注意：这里pool.config.GlobalSlots为所有可执行交易的总数，即pending列表总数，默认4096；pool.config.GlobalQueue为不可执行交易总数，即queue列表总数，默认1024；
// add validates a transaction and inserts it into the non-executable queue for
// later pending promotion and execution. If the transaction is a replacement for
// an already pending or queued one, it overwrites the previous and returns this
// so outer code doesn't uselessly call promote.
//
// If a newly added transaction is marked as local, its sending account will be
// whitelisted, preventing any associated transaction from being dropped out of
// the pool due to pricing constraints.
func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) {
// If the transaction is already known, discard it
hash := tx.Hash()
if pool.all.Get(hash) != nil {
log.Trace("Discarding already known transaction", "hash", hash)
return false, fmt.Errorf("known transaction: %x", hash)
}
// If the transaction fails basic validation, discard it
if err := pool.validateTx(tx, local); err != nil {
log.Trace("Discarding invalid transaction", "hash", hash, "err", err)
invalidTxCounter.Inc(1)
return false, err
}
// If the transaction pool is full, discard underpriced transactions
if uint64(pool.all.Count()) >= pool.config.GlobalSlots+pool.config.GlobalQueue {
// If the new transaction is underpriced, don't accept it
if !local && pool.priced.Underpriced(tx, pool.locals) {
log.Trace("Discarding underpriced transaction", "hash", hash, "price", tx.GasPrice())
underpricedTxCounter.Inc(1)
return false, ErrUnderpriced
}
// New transaction is better than our worse ones, make room for it
drop := pool.priced.Discard(pool.all.Count()-int(pool.config.GlobalSlots+pool.config.GlobalQueue-1), pool.locals)
for _, tx := range drop {
log.Trace("Discarding freshly underpriced transaction", "hash", tx.Hash(), "price", tx.GasPrice())
underpricedTxCounter.Inc(1)
pool.removeTx(tx.Hash(), false)
}
}
// If the transaction is replacing an already pending one, do directly
//如果插入pending已有的交易，必须交易价值大于或等于原交易的110%，方可替换
from, _ := types.Sender(pool.signer, tx) // already validated
if list := pool.pending[from]; list != nil && list.Overlaps(tx) {
// Nonce already pending, check if required price bump is met
inserted, old := list.Add(tx, pool.config.PriceBump)
if !inserted {
pendingDiscardCounter.Inc(1)
return false, ErrReplaceUnderpriced
}
// New transaction is better, replace old one
if old != nil {
pool.all.Remove(old.Hash())
pool.priced.Removed()
pendingReplaceCounter.Inc(1)
}
pool.all.Add(tx)
pool.priced.Put(tx)
pool.journalTx(from, tx)
log.Trace("Pooled new executable transaction", "hash", hash, "from", from, "to", tx.To())
// We've directly injected a replacement transaction, notify subsystems
//通知其他对交易池增加交易感兴趣的子系统：广播和矿工
go pool.txFeed.Send(NewTxsEvent{types.Transactions{tx}})
return old != nil, nil
}
// New transaction isn't replacing a pending one, push into queue
replace, err := pool.enqueueTx(hash, tx)
if err != nil {
return false, err
}
// Mark local addresses and journal local transactions
if local {
if !pool.locals.contains(from) {
log.Info("Setting new local account", "address", from)
pool.locals.add(from)
}
}
pool.journalTx(from, tx)
log.Trace("Pooled new future transaction", "hash", hash, "from", from, "to", tx.To())
return replace, nil
}
TxPool.add()的调用时机：
1）命令行发送交易
EthAPIBackend.SendTx
==> TxPool.AddLocal
==>Txpool.addTx
==>Txpool.add
2）收到远程节点广播的交易时
AddRemotes
==> TxPool.addRemotes
==> TxPool.addTxs
==> TxPool.addTxLocked
==> TxPool.add
3）交易池重新整理的过程
TxPool.reset
==> TxPool.addTxLocked
==> TxPool.add
这里的addLocal和addRemote的区别，其中有第二个参数来设定该交易是local还是remote，local的交易在打包时有优先权，在删除时有豁免权，还会以文件的形式保存在磁盘上。
func (pool *TxPool) AddLocal(tx *types.Transaction) error {
return pool.addTx(tx, !pool.config.NoLocals)
}
func (pool *TxPool) AddRemote(tx *types.Transaction) error {
return pool.addTx(tx, false)
}
交易加入queue列表：TxPool.enqueueTx
主要流程：
1）将交易插入queue中，如果待插入的交易序号在queue列表中已经有一个交易，那么待插入的交易价值大于原交易价值的110%，则替换原交易；
2）如果新交易替换成功，则从all列表中删除这个被替换的交易
3）更新all列表
// enqueueTx inserts a new transaction into the non-executable transaction queue.
//
// Note, this method assumes the pool lock is held!
func (pool *TxPool) enqueueTx(hash common.Hash, tx *types.Transaction) (bool, error) {
// Try to insert the transaction into the future queue
from, _ := types.Sender(pool.signer, tx) // already validated
if pool.queue[from] == nil {
pool.queue[from] = newTxList(false)
}
inserted, old := pool.queue[from].Add(tx, pool.config.PriceBump)
if !inserted {
// An older transaction was better, discard this
queuedDiscardCounter.Inc(1)
return false, ErrReplaceUnderpriced
}
// Discard any previous transaction and mark this
if old != nil {
pool.all.Remove(old.Hash())
pool.priced.Removed()
queuedReplaceCounter.Inc(1)
}
if pool.all.Get(hash) == nil {
pool.all.Add(tx)
pool.priced.Put(tx)
}
return old != nil, nil
}
