aboutsummaryrefslogtreecommitdiffstats
path: root/ethchain/transaction_pool.go
diff options
context:
space:
mode:
authorFelix Lange <fjl@twurst.com>2014-10-23 21:01:27 +0800
committerFelix Lange <fjl@twurst.com>2014-10-23 21:01:27 +0800
commit69baa465ea69ae60eed802445cf0132b9eb69934 (patch)
treeb09da7582b5c4850d4db13aee808f2fef2f97de0 /ethchain/transaction_pool.go
parent50fd46924900869e7210217c6a07979b544991c8 (diff)
parentfeef194829b07570e91873ed5d1e8cc51e8fa430 (diff)
downloadgo-tangerine-69baa465ea69ae60eed802445cf0132b9eb69934.tar.gz
go-tangerine-69baa465ea69ae60eed802445cf0132b9eb69934.tar.zst
go-tangerine-69baa465ea69ae60eed802445cf0132b9eb69934.zip
Merge eth-go repository into go-ethereum
mist, etheruem have been moved to cmd/
Diffstat (limited to 'ethchain/transaction_pool.go')
-rw-r--r--ethchain/transaction_pool.go224
1 files changed, 224 insertions, 0 deletions
diff --git a/ethchain/transaction_pool.go b/ethchain/transaction_pool.go
new file mode 100644
index 000000000..063688aa8
--- /dev/null
+++ b/ethchain/transaction_pool.go
@@ -0,0 +1,224 @@
+package ethchain
+
+import (
+ "bytes"
+ "container/list"
+ "fmt"
+ "math/big"
+ "sync"
+
+ "github.com/ethereum/go-ethereum/ethlog"
+ "github.com/ethereum/go-ethereum/ethstate"
+ "github.com/ethereum/go-ethereum/ethwire"
+)
+
+var txplogger = ethlog.NewLogger("TXP")
+
+const (
+ txPoolQueueSize = 50
+)
+
+type TxPoolHook chan *Transaction
+type TxMsgTy byte
+
+const (
+ TxPre = iota
+ TxPost
+
+ minGasPrice = 1000000
+)
+
+var MinGasPrice = big.NewInt(10000000000000)
+
+type TxMsg struct {
+ Tx *Transaction
+ Type TxMsgTy
+}
+
+func FindTx(pool *list.List, finder func(*Transaction, *list.Element) bool) *Transaction {
+ for e := pool.Front(); e != nil; e = e.Next() {
+ if tx, ok := e.Value.(*Transaction); ok {
+ if finder(tx, e) {
+ return tx
+ }
+ }
+ }
+
+ return nil
+}
+
+type TxProcessor interface {
+ ProcessTransaction(tx *Transaction)
+}
+
+// The tx pool a thread safe transaction pool handler. In order to
+// guarantee a non blocking pool we use a queue channel which can be
+// independently read without needing access to the actual pool. If the
+// pool is being drained or synced for whatever reason the transactions
+// will simple queue up and handled when the mutex is freed.
+type TxPool struct {
+ Ethereum EthManager
+ // The mutex for accessing the Tx pool.
+ mutex sync.Mutex
+ // Queueing channel for reading and writing incoming
+ // transactions to
+ queueChan chan *Transaction
+ // Quiting channel
+ quit chan bool
+ // The actual pool
+ pool *list.List
+
+ SecondaryProcessor TxProcessor
+
+ subscribers []chan TxMsg
+}
+
+func NewTxPool(ethereum EthManager) *TxPool {
+ return &TxPool{
+ pool: list.New(),
+ queueChan: make(chan *Transaction, txPoolQueueSize),
+ quit: make(chan bool),
+ Ethereum: ethereum,
+ }
+}
+
+// Blocking function. Don't use directly. Use QueueTransaction instead
+func (pool *TxPool) addTransaction(tx *Transaction) {
+ pool.mutex.Lock()
+ defer pool.mutex.Unlock()
+
+ pool.pool.PushBack(tx)
+
+ // Broadcast the transaction to the rest of the peers
+ pool.Ethereum.Broadcast(ethwire.MsgTxTy, []interface{}{tx.RlpData()})
+}
+
+func (pool *TxPool) ValidateTransaction(tx *Transaction) error {
+ // Get the last block so we can retrieve the sender and receiver from
+ // the merkle trie
+ block := pool.Ethereum.ChainManager().CurrentBlock
+ // Something has gone horribly wrong if this happens
+ if block == nil {
+ return fmt.Errorf("[TXPL] No last block on the block chain")
+ }
+
+ if len(tx.Recipient) != 0 && len(tx.Recipient) != 20 {
+ return fmt.Errorf("[TXPL] Invalid recipient. len = %d", len(tx.Recipient))
+ }
+
+ if tx.GasPrice.Cmp(MinGasPrice) < 0 {
+ return fmt.Errorf("Gas price to low. Require %v > Got %v", MinGasPrice, tx.GasPrice)
+ }
+
+ // Get the sender
+ //sender := pool.Ethereum.StateManager().procState.GetAccount(tx.Sender())
+ sender := pool.Ethereum.StateManager().CurrentState().GetAccount(tx.Sender())
+
+ totAmount := new(big.Int).Set(tx.Value)
+ // Make sure there's enough in the sender's account. Having insufficient
+ // funds won't invalidate this transaction but simple ignores it.
+ if sender.Balance().Cmp(totAmount) < 0 {
+ return fmt.Errorf("[TXPL] Insufficient amount in sender's (%x) account", tx.Sender())
+ }
+
+ if tx.IsContract() {
+ if tx.GasPrice.Cmp(big.NewInt(minGasPrice)) < 0 {
+ return fmt.Errorf("[TXPL] Gasprice too low, %s given should be at least %d.", tx.GasPrice, minGasPrice)
+ }
+ }
+
+ // Increment the nonce making each tx valid only once to prevent replay
+ // attacks
+
+ return nil
+}
+
+func (pool *TxPool) queueHandler() {
+out:
+ for {
+ select {
+ case tx := <-pool.queueChan:
+ hash := tx.Hash()
+ foundTx := FindTx(pool.pool, func(tx *Transaction, e *list.Element) bool {
+ return bytes.Compare(tx.Hash(), hash) == 0
+ })
+
+ if foundTx != nil {
+ break
+ }
+
+ // Validate the transaction
+ err := pool.ValidateTransaction(tx)
+ if err != nil {
+ txplogger.Debugln("Validating Tx failed", err)
+ } else {
+ // Call blocking version.
+ pool.addTransaction(tx)
+
+ tmp := make([]byte, 4)
+ copy(tmp, tx.Recipient)
+
+ txplogger.Debugf("(t) %x => %x (%v) %x\n", tx.Sender()[:4], tmp, tx.Value, tx.Hash())
+
+ // Notify the subscribers
+ pool.Ethereum.EventMux().Post(TxEvent{TxPre, tx})
+ }
+ case <-pool.quit:
+ break out
+ }
+ }
+}
+
+func (pool *TxPool) QueueTransaction(tx *Transaction) {
+ pool.queueChan <- tx
+}
+
+func (pool *TxPool) CurrentTransactions() []*Transaction {
+ pool.mutex.Lock()
+ defer pool.mutex.Unlock()
+
+ txList := make([]*Transaction, pool.pool.Len())
+ i := 0
+ for e := pool.pool.Front(); e != nil; e = e.Next() {
+ tx := e.Value.(*Transaction)
+
+ txList[i] = tx
+
+ i++
+ }
+
+ return txList
+}
+
+func (pool *TxPool) RemoveInvalid(state *ethstate.State) {
+ for e := pool.pool.Front(); e != nil; e = e.Next() {
+ tx := e.Value.(*Transaction)
+ sender := state.GetAccount(tx.Sender())
+ err := pool.ValidateTransaction(tx)
+ if err != nil || sender.Nonce >= tx.Nonce {
+ pool.pool.Remove(e)
+ }
+ }
+}
+
+func (pool *TxPool) Flush() []*Transaction {
+ txList := pool.CurrentTransactions()
+
+ // Recreate a new list all together
+ // XXX Is this the fastest way?
+ pool.pool = list.New()
+
+ return txList
+}
+
+func (pool *TxPool) Start() {
+ go pool.queueHandler()
+}
+
+func (pool *TxPool) Stop() {
+ close(pool.quit)
+
+ pool.Flush()
+
+ txplogger.Infoln("Stopped")
+}