Merge 0486400874 into 5f7826270c

core/tx_pool.go

@@ -97,6 +97,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{}
@@ -112,6 +113,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,
@@ -259,6 +261,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) {
@@ -361,9 +368,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.
@@ -389,6 +399,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()
@@ -475,6 +488,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 {
@@ -517,6 +531,7 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB) {
 				return fmt.Sprintf("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))
@@ -525,6 +540,7 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB) {
 				return fmt.Sprintf("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
@@ -540,6 +556,7 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB) {
 				return fmt.Sprintf("Removed cap-exceeding queued transaction: %v", tx)
 			}})
 			delete(pool.all, tx.Hash())
+			delete(pool.txSeen, tx.Hash())
 			queuedRLCounter.Inc(1)
 		}
 		queued += uint64(list.Len())
@@ -654,6 +671,7 @@ func (pool *TxPool) demoteUnexecutables(state *state.StateDB) {
 				return fmt.Sprintf("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))
@@ -662,6 +680,8 @@ func (pool *TxPool) demoteUnexecutables(state *state.StateDB) {
 				return fmt.Sprintf("Removed unpayable pending transaction: %v", tx)
 			}})
 			delete(pool.all, tx.Hash())
+			delete(pool.txSeen, tx.Hash())
+
 			pendingNofundsCounter.Inc(1)
 		}
 		for _, tx := range invalids {

eth/fetcher/fetcher.go

@@ -68,6 +68,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 {
@@ -135,6 +138,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
@@ -145,7 +149,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),
@@ -165,6 +169,7 @@ func New(getBlock blockRetrievalFn, validateBlock blockValidatorFn, broadcastBlo
 		getBlock:       getBlock,
 		validateBlock:  validateBlock,
 		broadcastBlock: broadcastBlock,
+		calcDlyBstBlk:  calcDlyBstBlk,
 		chainHeight:    chainHeight,
 		insertChain:    insertChain,
 		dropPeer:       dropPeer,
@@ -672,6 +677,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:

eth/handler.go

@@ -175,7 +175,8 @@ func NewProtocolManager(config *params.ChainConfig, fastSync bool, networkId int
 		atomic.StoreUint32(&manager.synced, 1) // 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 {
 		log.Debug(fmt.Sprint("Bad Block Reporting is enabled"))
@@ -765,3 +766,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;
+
+}

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.