diff --git a/core/tx_pool.go b/core/tx_pool.go index 65e076df9c..8d7aea690b 100644 --- a/core/tx_pool.go +++ b/core/tx_pool.go @@ -99,6 +99,7 @@ type TxPool struct { queue map[common.Address]*txList // Queued but non-processable transactions all map[common.Hash]*types.Transaction // All transactions to allow lookups beats map[common.Address]time.Time // Last heartbeat from each known account + txSeen map[common.Hash]time.Time // Time when a transaction enters in the pool wg sync.WaitGroup // for shutdown sync quit chan struct{} @@ -114,6 +115,7 @@ func NewTxPool(config *params.ChainConfig, eventMux *event.TypeMux, currentState queue: make(map[common.Address]*txList), all: make(map[common.Hash]*types.Transaction), beats: make(map[common.Address]time.Time), + txSeen: make(map[common.Hash]time.Time), eventMux: eventMux, currentState: currentStateFn, gasLimit: gasLimitFn, @@ -261,6 +263,11 @@ func (pool *TxPool) Pending() (map[common.Address]types.Transactions, error) { return pending, nil } +// Returns the time when the transaction was added to the pool +func (pool *TxPool) TxSeen(hash common.Hash) time.Time { + return pool.txSeen[hash] +} + // SetLocal marks a transaction as local, skipping gas price // check against local miner minimum in the future func (pool *TxPool) SetLocal(tx *types.Transaction) { @@ -370,9 +377,12 @@ func (pool *TxPool) enqueueTx(hash common.Hash, tx *types.Transaction) { // Discard any previous transaction and mark this if old != nil { delete(pool.all, old.Hash()) + delete(pool.txSeen, old.Hash()) + queuedReplaceCounter.Inc(1) } pool.all[hash] = tx + pool.txSeen[hash] = time.Now() } // promoteTx adds a transaction to the pending (processable) list of transactions. @@ -398,6 +408,9 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T pendingReplaceCounter.Inc(1) } pool.all[hash] = tx // Failsafe to work around direct pending inserts (tests) + if pool.txSeen[hash].IsZero() { + pool.txSeen[hash] = time.Now() + } // Set the potentially new pending nonce and notify any subsystems of the new tx pool.beats[addr] = time.Now() @@ -484,6 +497,7 @@ func (pool *TxPool) removeTx(hash common.Hash) { // Remove it from the list of known transactions delete(pool.all, hash) + delete(pool.txSeen, hash) // Remove the transaction from the pending lists and reset the account nonce if pending := pool.pending[addr]; pending != nil { @@ -526,6 +540,7 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB) { glog.Infof("Removed old queued transaction: %v", tx) } delete(pool.all, tx.Hash()) + delete(pool.txSeen, tx.Hash()) } // Drop all transactions that are too costly (low balance) drops, _ := list.Filter(state.GetBalance(addr)) @@ -534,6 +549,7 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB) { glog.Infof("Removed unpayable queued transaction: %v", tx) } delete(pool.all, tx.Hash()) + delete(pool.txSeen, tx.Hash()) queuedNofundsCounter.Inc(1) } // Gather all executable transactions and promote them @@ -549,6 +565,7 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB) { glog.Infof("Removed cap-exceeding queued transaction: %v", tx) } delete(pool.all, tx.Hash()) + delete(pool.txSeen, tx.Hash()) queuedRLCounter.Inc(1) } queued += uint64(list.Len()) @@ -663,6 +680,7 @@ func (pool *TxPool) demoteUnexecutables(state *state.StateDB) { glog.Infof("Removed old pending transaction: %v", tx) } delete(pool.all, tx.Hash()) + delete(pool.txSeen, tx.Hash()) } // Drop all transactions that are too costly (low balance), and queue any invalids back for later drops, invalids := list.Filter(state.GetBalance(addr)) @@ -671,6 +689,8 @@ func (pool *TxPool) demoteUnexecutables(state *state.StateDB) { glog.Infof("Removed unpayable pending transaction: %v", tx) } delete(pool.all, tx.Hash()) + delete(pool.txSeen, tx.Hash()) + pendingNofundsCounter.Inc(1) } for _, tx := range invalids { diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index e225b49965..589234a72e 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -69,6 +69,9 @@ type chainInsertFn func(types.Blocks) (int, error) // peerDropFn is a callback type for dropping a peer detected as malicious. type peerDropFn func(id string) +// calcDlyBstBlkFn is called to determine how long is delayed a propagation of a new block. +type calcDlyBstBlkFn func(block *types.Block) time.Duration + // announce is the hash notification of the availability of a new block in the // network. type announce struct { @@ -136,6 +139,7 @@ type Fetcher struct { chainHeight chainHeightFn // Retrieves the current chain's height insertChain chainInsertFn // Injects a batch of blocks into the chain dropPeer peerDropFn // Drops a peer for misbehaving + calcDlyBstBlk calcDlyBstBlkFn // Calculates the delay to broadcast a packet. // Testing hooks announceChangeHook func(common.Hash, bool) // Method to call upon adding or deleting a hash from the announce list @@ -146,7 +150,7 @@ type Fetcher struct { } // New creates a block fetcher to retrieve blocks based on hash announcements. -func New(getBlock blockRetrievalFn, validateBlock blockValidatorFn, broadcastBlock blockBroadcasterFn, chainHeight chainHeightFn, insertChain chainInsertFn, dropPeer peerDropFn) *Fetcher { +func New(getBlock blockRetrievalFn, validateBlock blockValidatorFn, broadcastBlock blockBroadcasterFn, chainHeight chainHeightFn, insertChain chainInsertFn, dropPeer peerDropFn, calcDlyBstBlk calcDlyBstBlkFn) *Fetcher { return &Fetcher{ notify: make(chan *announce), inject: make(chan *inject), @@ -166,6 +170,7 @@ func New(getBlock blockRetrievalFn, validateBlock blockValidatorFn, broadcastBlo getBlock: getBlock, validateBlock: validateBlock, broadcastBlock: broadcastBlock, + calcDlyBstBlk: calcDlyBstBlk, chainHeight: chainHeight, insertChain: insertChain, dropPeer: dropPeer, @@ -669,6 +674,9 @@ func (f *Fetcher) insert(peer string, block *types.Block) { case nil: // All ok, quickly propagate to our peers propBroadcastOutTimer.UpdateSince(block.ReceivedAt) + glog.V(logger.Debug).Infof("DelayEmpty: Before Delay Broadcast. #%d", block.NumberU64()) + time.Sleep(f.calcDlyBstBlk(block)); + glog.V(logger.Debug).Infof("DelayEmpty: After Delay Broadcast. #%d", block.NumberU64()) go f.broadcastBlock(block, true) case core.BlockFutureErr: diff --git a/eth/handler.go b/eth/handler.go index 771e69b8df..a4e7d78859 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -176,7 +176,8 @@ func NewProtocolManager(config *params.ChainConfig, fastSync bool, networkId int manager.setSynced() // Mark initial sync done on any fetcher import return manager.insertChain(blocks) } - manager.fetcher = fetcher.New(blockchain.GetBlockByHash, validator, manager.BroadcastBlock, heighter, inserter, manager.removePeer) + + manager.fetcher = fetcher.New(blockchain.GetBlockByHash, validator, manager.BroadcastBlock, heighter, inserter, manager.removePeer, manager.CalcDelayBroadcastBlock) if blockchain.Genesis().Hash().Hex() == defaultGenesisHash && networkId == 1 { glog.V(logger.Debug).Infoln("Bad Block Reporting is enabled") @@ -788,3 +789,102 @@ func (self *ProtocolManager) NodeInfo() *EthNodeInfo { Head: currentBlock.Hash(), } } + +func (self *ProtocolManager) CalcDelayBroadcastBlock(block *types.Block) time.Duration { + + TIME_TO_BE_CONSIDERED_OLD_TRANSACTION := 8 * time.Second + MAX_DELAY := 30 * time.Second + + getBlockTx := func(h1 common.Hash) *types.Transaction { + for _, tx := range block.Transactions() { + h2 := tx.Hash(); + if h1 == h2 { + return tx + } + } + return nil; + } + + hash := block.Hash() + pending, _ := self.txpool.Pending() + nIncludedTransactions := 0 + includedTransactionsGas := big.NewInt(0) + nNotIncludedOldTransactions := 0 + notIncludedOldTransactionsGas := big.NewInt(0) + totalTx := 0 + + includedTxs := make(map[common.Hash]*big.Int) + + for _, list := range pending { + for _, tx := range list { + + txHash := tx.Hash() + txSeen := self.txpool.TxSeen(txHash) + + glog.V(logger.Debug).Infof("DelayEmpty TxHash: %x Time Elapsed: %f ",txHash[:4], time.Since(txSeen).Seconds()) + + blockTx := getBlockTx(txHash) + + if blockTx != nil { + nIncludedTransactions ++; + includedTransactionsGas.Add(includedTransactionsGas , blockTx.Gas()); + } else { + if (time.Since(txSeen) > TIME_TO_BE_CONSIDERED_OLD_TRANSACTION) { + nNotIncludedOldTransactions ++ + notIncludedOldTransactionsGas.Add(notIncludedOldTransactionsGas, tx.Gas()) + includedTxs[txHash] = tx.Gas() + } + } + totalTx ++ + + } + } + + + missusedGas := new(big.Int).Sub( block.GasLimit() , includedTransactionsGas ); + + // Remove not included that does not fit in the block + for _, g := range includedTxs { + if g.Cmp(missusedGas) > 0 { + nNotIncludedOldTransactions -- + notIncludedOldTransactionsGas.Sub(notIncludedOldTransactionsGas, g) + glog.V(logger.Debug).Infof("DelayEmpty Big Tx not accounted") + } + } + + // If less of the 25% of the block is missusedGas just propagate fast. + // This percentage shuld be adjusted. + if new(big.Float).SetInt(missusedGas).Cmp( + new(big.Float).Mul( + new(big.Float).SetInt( block.GasLimit() ) , + new(big.Float).SetFloat64(0.4) )) < 0 { + return 0; + } + + factor1, _ := new(big.Float).Quo( + new(big.Float).SetInt(notIncludedOldTransactionsGas) , + + new(big.Float).Mul( + new(big.Float).SetInt( block.GasLimit() ), + new(big.Float).SetFloat64(0.75))).Float64() + + if factor1 > 1 { + factor1 = 1 + } + + + factor2, _ := new(big.Float).Quo( + new(big.Float).SetInt(missusedGas), + new(big.Float).SetInt( block.GasLimit() )).Float64() + + + delay := time.Duration( float64(MAX_DELAY.Nanoseconds()) * factor1 * factor2) * time.Nanosecond; + + glog.V(logger.Debug).Infof("DelayEmpty IncludedTxGas: %d, NotIncludedTxGas: %d",includedTransactionsGas.Int64(), notIncludedOldTransactionsGas.Int64()) + glog.V(logger.Debug).Infof("DelayEmpty TxBlock: %d, TxPending: %d TxIncluded: %d TxNotIncluded: %d",block.Transactions().Len(), totalTx, nIncludedTransactions, nNotIncludedOldTransactions) + glog.V(logger.Debug).Infof("DelayEmpty Missussed #%d Factor1: %f Factor2: %f",missusedGas.Int64(), factor1, factor2) + glog.V(logger.Debug).Infof("DelayEmpty block #%d [%x…] Delay: %f",block.NumberU64(), hash[:4], delay.Seconds()) + + return delay; + +} diff --git a/eth/protocol.go b/eth/protocol.go index 7d22b33dec..1e165bf913 100644 --- a/eth/protocol.go +++ b/eth/protocol.go @@ -20,6 +20,7 @@ import ( "fmt" "io" "math/big" + "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" @@ -103,6 +104,9 @@ type txPool interface { // Pending should return pending transactions. // The slice should be modifiable by the caller. Pending() (map[common.Address]types.Transactions, error) + + // Returns when the transaction was added to the pool + TxSeen(common.Hash) time.Time } // statusData is the network packet for the status message.