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