protocol/txpool.go

   1 package protocol
   2
   3 import (
   4         "errors"
   5         "sync"
   6         "sync/atomic"
   7         "time"
   8
   9         "github.com/golang/groupcache/lru"
  10
  11         "github.com/bytom/blockchain/txdb/storage"
  12         "github.com/bytom/consensus"
  13         "github.com/bytom/protocol/bc"
  14         "github.com/bytom/protocol/bc/legacy"
  15         "github.com/bytom/protocol/state"
  16         log "github.com/sirupsen/logrus"
  17 )
  18
  19 var (
  20         maxCachedErrTxs = 1000
  21         maxNewTxChSize  = 1000
  22         maxNewTxNum     = 10000
  23         // ErrTransactionNotExist is the pre-defined error message
  24         ErrTransactionNotExist = errors.New("transaction are not existed in the mempool")
  25         ErrPoolIsFull          = errors.New("transaction pool reach the max number")
  26 )
  27
  28 // TxDesc store tx and related info for mining strategy
  29 type TxDesc struct {
  30         Tx       *legacy.Tx
  31         Added    time.Time
  32         Height   uint64
  33         Weight   uint64
  34         Fee      uint64
  35         FeePerKB uint64
  36 }
  37
  38 // TxPool is use for store the unconfirmed transaction
  39 type TxPool struct {
  40         lastUpdated int64
  41         mtx         sync.RWMutex
  42         pool        map[bc.Hash]*TxDesc
  43         utxo        map[bc.Hash]bc.Hash
  44         errCache    *lru.Cache
  45         newTxCh     chan *legacy.Tx
  46 }
  47
  48 // NewTxPool init a new TxPool
  49 func NewTxPool() *TxPool {
  50         return &TxPool{
  51                 lastUpdated: time.Now().Unix(),
  52                 pool:        make(map[bc.Hash]*TxDesc),
  53                 utxo:        make(map[bc.Hash]bc.Hash),
  54                 errCache:    lru.New(maxCachedErrTxs),
  55                 newTxCh:     make(chan *legacy.Tx, maxNewTxChSize),
  56         }
  57 }
  58
  59 // GetNewTxCh return a unconfirmed transaction feed channel
  60 func (mp *TxPool) GetNewTxCh() chan *legacy.Tx {
  61         return mp.newTxCh
  62 }
  63
  64 // AddTransaction add a verified transaction to pool
  65 func (mp *TxPool) AddTransaction(tx *legacy.Tx, gasOnlyTx bool, height, fee uint64) (*TxDesc, error) {
  66         mp.mtx.Lock()
  67         defer mp.mtx.Unlock()
  68
  69         if len(mp.pool) >= maxNewTxNum {
  70                 return nil, ErrPoolIsFull
  71         }
  72
  73         txD := &TxDesc{
  74                 Tx:       tx,
  75                 Added:    time.Now(),
  76                 Weight:   tx.TxData.SerializedSize,
  77                 Height:   height,
  78                 Fee:      fee,
  79                 FeePerKB: fee * 1000 / tx.TxHeader.SerializedSize,
  80         }
  81
  82         mp.pool[tx.Tx.ID] = txD
  83         atomic.StoreInt64(&mp.lastUpdated, time.Now().Unix())
  84
  85         for _, id := range tx.TxHeader.ResultIds {
  86                 output, err := tx.Output(*id)
  87                 if err != nil {
  88                         // error due to it's a retirement, utxo doesn't care this output type so skip it
  89                         continue
  90                 }
  91                 if !gasOnlyTx || *output.Source.Value.AssetId == *consensus.BTMAssetID {
  92                         mp.utxo[*id] = tx.Tx.ID
  93                 }
  94         }
  95
  96         mp.newTxCh <- tx
  97         log.WithField("tx_id", tx.Tx.ID).Info("Add tx to mempool")
  98         return txD, nil
  99 }
 100
 101 // AddErrCache add a failed transaction record to lru cache
 102 func (mp *TxPool) AddErrCache(txHash *bc.Hash, err error) {
 103         mp.mtx.Lock()
 104         defer mp.mtx.Unlock()
 105
 106         mp.errCache.Add(txHash, err)
 107 }
 108
 109 // GetErrCache return the error of the transaction
 110 func (mp *TxPool) GetErrCache(txHash *bc.Hash) error {
 111         mp.mtx.Lock()
 112         defer mp.mtx.Unlock()
 113
 114         v, ok := mp.errCache.Get(txHash)
 115         if !ok {
 116                 return nil
 117         }
 118         return v.(error)
 119 }
 120
 121 // RemoveTransaction remove a transaction from the pool
 122 func (mp *TxPool) RemoveTransaction(txHash *bc.Hash) {
 123         mp.mtx.Lock()
 124         defer mp.mtx.Unlock()
 125
 126         txD, ok := mp.pool[*txHash]
 127         if !ok {
 128                 return
 129         }
 130
 131         for _, output := range txD.Tx.TxHeader.ResultIds {
 132                 delete(mp.utxo, *output)
 133         }
 134         delete(mp.pool, *txHash)
 135         atomic.StoreInt64(&mp.lastUpdated, time.Now().Unix())
 136
 137         log.WithField("tx_id", txHash).Info("remove tx from mempool")
 138 }
 139
 140 // GetTransaction return the TxDesc by hash
 141 func (mp *TxPool) GetTransaction(txHash *bc.Hash) (*TxDesc, error) {
 142         mp.mtx.RLock()
 143         defer mp.mtx.RUnlock()
 144
 145         if txD, ok := mp.pool[*txHash]; ok {
 146                 return txD, nil
 147         }
 148
 149         return nil, ErrTransactionNotExist
 150 }
 151
 152 // GetTransactions return all the transactions in the pool
 153 func (mp *TxPool) GetTransactions() []*TxDesc {
 154         mp.mtx.RLock()
 155         defer mp.mtx.RUnlock()
 156
 157         txDs := make([]*TxDesc, len(mp.pool))
 158         i := 0
 159         for _, desc := range mp.pool {
 160                 txDs[i] = desc
 161                 i++
 162         }
 163         return txDs
 164 }
 165
 166 // GetTransactionUTXO return unconfirmed utxo
 167 func (mp *TxPool) GetTransactionUTXO(tx *bc.Tx) *state.UtxoViewpoint {
 168         mp.mtx.RLock()
 169         defer mp.mtx.RUnlock()
 170
 171         view := state.NewUtxoViewpoint()
 172         for _, prevout := range tx.SpentOutputIDs {
 173                 if _, ok := mp.utxo[prevout]; ok {
 174                         view.Entries[prevout] = storage.NewUtxoEntry(false, 0, false)
 175                 }
 176         }
 177         return view
 178 }
 179
 180 // IsTransactionInPool check wheather a transaction in pool or not
 181 func (mp *TxPool) IsTransactionInPool(txHash *bc.Hash) bool {
 182         mp.mtx.RLock()
 183         defer mp.mtx.RUnlock()
 184
 185         if _, ok := mp.pool[*txHash]; ok {
 186                 return true
 187         }
 188         return false
 189 }
 190
 191 // IsTransactionInErrCache check wheather a transaction in errCache or not
 192 func (mp *TxPool) IsTransactionInErrCache(txHash *bc.Hash) bool {
 193         mp.mtx.RLock()
 194         defer mp.mtx.RUnlock()
 195
 196         _, ok := mp.errCache.Get(txHash)
 197         return ok
 198 }
 199
 200 // HaveTransaction IsTransactionInErrCache check is  transaction in errCache or pool
 201 func (mp *TxPool) HaveTransaction(txHash *bc.Hash) bool {
 202         return mp.IsTransactionInPool(txHash) || mp.IsTransactionInErrCache(txHash)
 203 }
 204
 205 // Count return number of transcation in pool
 206 func (mp *TxPool) Count() int {
 207         mp.mtx.RLock()
 208         defer mp.mtx.RUnlock()
 209
 210         count := len(mp.pool)
 211         return count
 212 }