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