@ -5,6 +5,8 @@ import (
"math/big"
"time"
common2 "github.com/harmony-one/harmony/internal/common"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/rlp"
"github.com/harmony-one/harmony/block"
@ -14,7 +16,6 @@ import (
"github.com/harmony-one/harmony/core/state"
"github.com/harmony-one/harmony/core/types"
"github.com/harmony-one/harmony/core/vm"
shardingconfig "github.com/harmony-one/harmony/internal/configs/sharding"
"github.com/harmony-one/harmony/internal/ctxerror"
"github.com/harmony-one/harmony/internal/params"
"github.com/harmony-one/harmony/internal/utils"
@ -25,6 +26,8 @@ import (
// environment is the worker's current environment and holds all of the current state information.
type environment struct {
signer types . Signer
state * state . DB // apply state changes here
gasPool * core . GasPool // available gas used to pack transactions
@ -50,149 +53,97 @@ type Worker struct {
gasCeil uint64
}
// Returns a tuple where the first value is the txs sender account address,
// the second is the throttling result enum for the transaction of interest.
// Throttling happens based on the amount, frequency, etc.
func ( w * Worker ) throttleTxs ( selected types . Transactions , recentTxsStats types . RecentTxsStats , txsThrottleConfig * shardingconfig . TxsThrottleConfig , tx * types . Transaction ) ( common . Address , shardingconfig . TxThrottleFlag ) {
var sender common . Address
msg , err := tx . AsMessage ( types . MakeSigner ( w . config , w . chain . CurrentBlock ( ) . Epoch ( ) ) )
if err != nil {
utils . Logger ( ) . Error ( ) . Err ( err ) . Str ( "txId" , tx . Hash ( ) . Hex ( ) ) . Msg ( "Error when parsing tx into message" )
} else {
sender = msg . From ( )
}
// do not throttle transactions if disabled
if ! txsThrottleConfig . EnableTxnThrottling {
return sender , shardingconfig . TxSelect
}
// already selected max num txs
if len ( selected ) > txsThrottleConfig . MaxNumTxsPerBlockLimit {
utils . Logger ( ) . Info ( ) . Str ( "txId" , tx . Hash ( ) . Hex ( ) ) . Int ( "MaxNumTxsPerBlockLimit" , txsThrottleConfig . MaxNumTxsPerBlockLimit ) . Msg ( "Throttling tx with max num txs per block limit" )
return sender , shardingconfig . TxUnselect
}
// throttle a single sender sending too many transactions in one block
if tx . Value ( ) . Cmp ( txsThrottleConfig . MaxTxAmountLimit ) > 0 {
utils . Logger ( ) . Info ( ) . Str ( "txId" , tx . Hash ( ) . Hex ( ) ) . Uint64 ( "MaxTxAmountLimit" , txsThrottleConfig . MaxTxAmountLimit . Uint64 ( ) ) . Uint64 ( "txAmount" , tx . Value ( ) . Uint64 ( ) ) . Msg ( "Throttling tx with max amount limit" )
return sender , shardingconfig . TxInvalid
}
// throttle too large transaction
var numTxsPastHour uint64
for _ , blockTxsCounts := range recentTxsStats {
numTxsPastHour += blockTxsCounts [ sender ]
}
if numTxsPastHour >= txsThrottleConfig . MaxNumRecentTxsPerAccountLimit {
utils . Logger ( ) . Info ( ) . Str ( "txId" , tx . Hash ( ) . Hex ( ) ) . Uint64 ( "MaxNumRecentTxsPerAccountLimit" , txsThrottleConfig . MaxNumRecentTxsPerAccountLimit ) . Msg ( "Throttling tx with max txs per account in a single block limit" )
return sender , shardingconfig . TxInvalid
}
return sender , shardingconfig . TxSelect
}
// SelectTransactionsForNewBlock selects transactions for new block.
func ( w * Worker ) SelectTransactionsForNewBlock ( newBlockNum uint64 , txs types . Transactions , recentTxsStats types . RecentTxsStats , txsThrottleConfig * shardingconfig . TxsThrottleConfig , coinbase common . Address ) ( types . Transactions , types . Transactions , types . Transactions ) {
// CommitTransactions commits transactions for new block.
func ( w * Worker ) CommitTransactions ( pendingNormal map [ common . Address ] types . Transactions , pendingStaking staking . StakingTransactions , coinbase common . Address ) error {
if w . current . gasPool == nil {
w . current . gasPool = new ( core . GasPool ) . AddGas ( w . current . header . GasLimit ( ) )
}
selected := types . Transactions { }
unselected := types . Transactions { }
invalid := types . Transactions { }
for _ , tx := range txs {
if tx . ShardID ( ) != w . chain . ShardID ( ) {
invalid = append ( invalid , tx )
continue
}
txs := types . NewTransactionsByPriceAndNonce ( w . current . signer , pendingNormal )
var coalescedLogs [ ] * types . Log
// NORMAL
for {
// If we don't have enough gas for any further transactions then we're done
if w . current . gasPool . Gas ( ) < params . TxGas {
utils . Logger ( ) . Info ( ) . Str ( "Not enough gas for further transactions, have" , w . current . gasPool . String ( ) ) . Uint64 ( "want" , params . TxGas )
unselected = append ( unselected , tx )
utils . Logger ( ) . Info ( ) . Uint64 ( "have" , w . current . gasPool . Gas ( ) ) . Uint64 ( "want" , params . TxGas ) . Msg ( "Not enough gas for further transactions" )
break
}
// Retrieve the next transaction and abort if all done
tx := txs . Peek ( )
if tx == nil {
break
}
// Error may be ignored here. The error has already been checked
// during transaction acceptance is the transaction pool.
//
// We use the eip155 signer regardless of the current hf.
from , _ := types . Sender ( w . current . signer , tx )
// Check whether the tx is replay protected. If we're not in the EIP155 hf
// phase, start ignoring the sender until we do.
if tx . Protected ( ) && ! w . config . IsEIP155 ( w . current . header . Number ( ) ) {
utils . Logger ( ) . Info ( ) . Str ( "hash" , tx . Hash ( ) . Hex ( ) ) . Str ( "eip155Epoch" , w . config . EIP155Epoch . String ( ) ) . Msg ( "Ignoring reply protected transaction" )
txs . Pop ( )
continue
}
// Start executing the transaction
w . current . state . Prepare ( tx . Hash ( ) , common . Hash { } , len ( w . current . txs ) )
sender , flag := w . throttleTxs ( selected , recentTxsStats , txsThrottleConfig , tx )
switch flag {
case shardingconfig . TxUnselect :
unselected = append ( unselected , tx )
case shardingconfig . TxInvalid :
invalid = append ( invalid , tx )
case shardingconfig . TxSelect :
if tx . GasPrice ( ) . Uint64 ( ) == 0 {
invalid = append ( invalid , tx )
} else {
snap := w . current . state . Snapshot ( )
_ , err := w . commitTransaction ( tx , coinbase )
if err != nil {
w . current . state . RevertToSnapshot ( snap )
invalid = append ( invalid , tx )
utils . Logger ( ) . Error ( ) . Err ( err ) . Str ( "txId" , tx . Hash ( ) . Hex ( ) ) . Msg ( "Commit transaction error" )
} else {
selected = append ( selected , tx )
// handle the case when msg was not able to extracted from tx
if len ( sender . String ( ) ) > 0 {
recentTxsStats [ newBlockNum ] [ sender ] ++
}
}
}
if tx . ShardID ( ) != w . chain . ShardID ( ) {
txs . Shift ( )
continue
}
// log invalid or unselected txs
if flag == shardingconfig . TxUnselect || flag == shardingconfig . TxInvalid {
utils . Logger ( ) . Info ( ) . Str ( "txId" , tx . Hash ( ) . Hex ( ) ) . Str ( "txThrottleFlag" , flag . String ( ) ) . Msg ( "Transaction Throttle flag" )
logs , err := w . commitTransaction ( tx , coinbase )
sender , _ := common2 . AddressToBech32 ( from )
switch err {
case core . ErrGasLimitReached :
// Pop the current out-of-gas transaction without shifting in the next from the account
utils . Logger ( ) . Info ( ) . Str ( "sender" , sender ) . Msg ( "Gas limit exceeded for current block" )
txs . Pop ( )
case core . ErrNonceTooLow :
// New head notification data race between the transaction pool and miner, shift
utils . Logger ( ) . Info ( ) . Str ( "sender" , sender ) . Uint64 ( "nonce" , tx . Nonce ( ) ) . Msg ( "Skipping transaction with low nonce" )
txs . Shift ( )
case core . ErrNonceTooHigh :
// Reorg notification data race between the transaction pool and miner, skip account =
utils . Logger ( ) . Info ( ) . Str ( "sender" , sender ) . Uint64 ( "nonce" , tx . Nonce ( ) ) . Msg ( "Skipping account with high nonce" )
txs . Pop ( )
case nil :
// Everything ok, collect the logs and shift in the next transaction from the same account
coalescedLogs = append ( coalescedLogs , logs ... )
txs . Shift ( )
default :
// Strange error, discard the transaction and get the next in line (note, the
// nonce-too-high clause will prevent us from executing in vain).
utils . Logger ( ) . Info ( ) . Str ( "hash" , tx . Hash ( ) . Hex ( ) ) . AnErr ( "err" , err ) . Msg ( "Transaction failed, account skipped" )
txs . Shift ( )
}
utils . Logger ( ) . Info ( ) . Str ( "txId" , tx . Hash ( ) . Hex ( ) ) . Uint64 ( "txGasLimit" , tx . Gas ( ) ) . Msg ( "Transaction gas limit info" )
}
utils . Logger ( ) . Info ( ) . Uint64 ( "newBlockNum" , newBlockNum ) . Int ( "newTxns" , len ( selected ) ) . Uint64 ( "blockGasLimit" , w . current . header . GasLimit ( ) ) . Uint64 ( "blockGasUsed" , w . current . header . GasUsed ( ) ) . Msg ( "Block gas limit and usage info" )
return selected , unselected , invalid
}
// SelectStakingTransactionsForNewBlock selects staking transactions for new block.
func ( w * Worker ) SelectStakingTransactionsForNewBlock (
newBlockNum uint64 , txs staking . StakingTransactions ,
coinbase common . Address ) ( staking . StakingTransactions , staking . StakingTransactions , staking . StakingTransactions ) {
// only beaconchain process staking transaction
if w . chain . ShardID ( ) != shard . BeaconChainShardID {
utils . Logger ( ) . Warn ( ) . Msgf ( "Invalid shardID: %v" , w . chain . ShardID ( ) )
return nil , nil , nil
}
if w . current . gasPool == nil {
w . current . gasPool = new ( core . GasPool ) . AddGas ( w . current . header . GasLimit ( ) )
}
selected := staking . StakingTransactions { }
// TODO: chao add total gas fee checking when needed
unselected := staking . StakingTransactions { }
invalid := staking . StakingTransactions { }
for _ , tx := range txs {
snap := w . current . state . Snapshot ( )
_ , err := w . commitStakingTransaction ( tx , coinbase )
if err != nil {
w . current . state . RevertToSnapshot ( snap )
invalid = append ( invalid , tx )
utils . Logger ( ) . Error ( ) . Err ( err ) . Str ( "stakingTxId" , tx . Hash ( ) . Hex ( ) ) . Msg ( "Commit staking transaction error" )
} else {
selected = append ( selected , tx )
utils . Logger ( ) . Info ( ) . Str ( "stakingTxId" , tx . Hash ( ) . Hex ( ) ) . Uint64 ( "txGasLimit" , tx . Gas ( ) ) . Msg ( "StakingTransaction gas limit info" )
// STAKING - only beaconchain process staking transaction
if w . chain . ShardID ( ) == shard . BeaconChainShardID {
for _ , tx := range pendingStaking {
logs , err := w . commitStakingTransaction ( tx , coinbase )
if err != nil {
utils . Logger ( ) . Error ( ) . Err ( err ) . Str ( "stakingTxId" , tx . Hash ( ) . Hex ( ) ) . Msg ( "Commit staking transaction error" )
} else {
coalescedLogs = append ( coalescedLogs , logs ... )
utils . Logger ( ) . Info ( ) . Str ( "stakingTxId" , tx . Hash ( ) . Hex ( ) ) . Uint64 ( "txGasLimit" , tx . Gas ( ) ) . Msg ( "StakingTransaction gas limit info" )
}
}
}
utils . Logger ( ) . Info ( ) . Uint64 ( "newBlockNum" , newBlockNum ) . Uint64 ( "blockGasLimit" ,
w . current . header . GasLimit ( ) ) . Uint64 ( "blockGasUsed" ,
w . current . header . GasUsed ( ) ) . Msg ( "[SelectStakingTransaction] Block gas limit and usage info" )
utils . Logger ( ) . Info ( ) . Int ( "newTxns" , len ( w . current . txs ) ) . Uint64 ( "blockGasLimit" , w . current . header . GasLimit ( ) ) . Uint64 ( "blockGasUsed" , w . current . header . GasUsed ( ) ) . Msg ( "Block gas limit and usage info" )
return selected , unselected , invalid
return nil
}
func ( w * Worker ) commitStakingTransaction ( tx * staking . StakingTransaction , coinbase common . Address ) ( [ ] * types . Log , error ) {
@ -238,40 +189,6 @@ func (w *Worker) commitTransaction(tx *types.Transaction, coinbase common.Addres
return receipt . Logs , nil
}
// CommitTransactions commits transactions.
func ( w * Worker ) CommitTransactions ( txs types . Transactions , stakingTxns staking . StakingTransactions , coinbase common . Address ) error {
// Must update to the correct current state before processing potential txns
if err := w . UpdateCurrent ( coinbase ) ; err != nil {
utils . Logger ( ) . Error ( ) .
Err ( err ) .
Msg ( "Failed updating worker's state before committing txns" )
return err
}
if w . current . gasPool == nil {
w . current . gasPool = new ( core . GasPool ) . AddGas ( w . current . header . GasLimit ( ) )
}
for _ , tx := range txs {
snap := w . current . state . Snapshot ( )
_ , err := w . commitTransaction ( tx , coinbase )
if err != nil {
w . current . state . RevertToSnapshot ( snap )
return err
}
}
for _ , tx := range stakingTxns {
snap := w . current . state . Snapshot ( )
_ , err := w . commitStakingTransaction ( tx , coinbase )
if err != nil {
w . current . state . RevertToSnapshot ( snap )
return err
}
}
return nil
}
// CommitReceipts commits a list of already verified incoming cross shard receipts
func ( w * Worker ) CommitReceipts ( receiptsList [ ] * types . CXReceiptsProof ) error {
if w . current . gasPool == nil {
@ -322,6 +239,7 @@ func (w *Worker) makeCurrent(parent *types.Block, header *block.Header) error {
return err
}
env := & environment {
signer : types . NewEIP155Signer ( w . config . ChainID ) ,
state : state ,
header : header ,
}
Rongjian Lan
5 years ago
committed by
GitHub