protocol/mempool.go

   1 package protocol
   2
   3 import (
   4         "errors"
   5         "sync"
   6         "sync/atomic"
   7         "time"
   8
   9         "github.com/bytom/protocol/bc"
  10         "github.com/bytom/protocol/bc/legacy"
  11         "github.com/golang/groupcache/lru"
  12 )
  13
  14 var (
  15         maxCachedErrTxs = 1000
  16         maxNewTxChSize  = 1000
  17         // ErrTransactionNotExist is the pre-defined error message
  18         ErrTransactionNotExist = errors.New("transaction are not existed in the mempool")
  19 )
  20
  21 // TxDesc store tx and related info for mining strategy
  22 type TxDesc struct {
  23         Tx       *legacy.Tx
  24         Added    time.Time
  25         Height   uint64
  26         Weight   uint64
  27         Fee      uint64
  28         FeePerKB uint64
  29 }
  30
  31 // TxPool is use for store the unconfirmed transaction
  32 type TxPool struct {
  33         lastUpdated int64
  34         mtx         sync.RWMutex
  35         pool        map[bc.Hash]*TxDesc
  36         errCache    *lru.Cache
  37         newTxCh     chan *legacy.Tx
  38 }
  39
  40 // NewTxPool init a new TxPool
  41 func NewTxPool() *TxPool {
  42         return &TxPool{
  43                 lastUpdated: time.Now().Unix(),
  44                 pool:        make(map[bc.Hash]*TxDesc),
  45                 errCache:    lru.New(maxCachedErrTxs),
  46                 newTxCh:     make(chan *legacy.Tx, maxNewTxChSize),
  47         }
  48 }
  49
  50 // GetNewTxCh return a unconfirmed transaction feed channel
  51 func (mp *TxPool) GetNewTxCh() chan *legacy.Tx {
  52         return mp.newTxCh
  53 }
  54
  55 // AddTransaction add a verified transaction to pool
  56 func (mp *TxPool) AddTransaction(tx *legacy.Tx, height, fee uint64) *TxDesc {
  57         txD := &TxDesc{
  58                 Tx:       tx,
  59                 Added:    time.Now(),
  60                 Weight:   tx.TxData.SerializedSize,
  61                 Height:   height,
  62                 Fee:      fee,
  63                 FeePerKB: fee * 1000 / tx.TxHeader.SerializedSize,
  64         }
  65
  66         mp.mtx.Lock()
  67         defer mp.mtx.Unlock()
  68
  69         mp.pool[tx.Tx.ID] = txD
  70         atomic.StoreInt64(&mp.lastUpdated, time.Now().Unix())
  71
  72         mp.newTxCh <- tx
  73         return txD
  74 }
  75
  76 // AddErrCache add a failed transaction record to lru cache
  77 func (mp *TxPool) AddErrCache(txHash *bc.Hash, err error) {
  78         mp.mtx.Lock()
  79         defer mp.mtx.Unlock()
  80
  81         mp.errCache.Add(txHash, err)
  82 }
  83
  84 // GetErrCache return the error of the transaction
  85 func (mp *TxPool) GetErrCache(txHash *bc.Hash) error {
  86         mp.mtx.Lock()
  87         defer mp.mtx.Unlock()
  88
  89         v, ok := mp.errCache.Get(txHash)
  90         if !ok {
  91                 return nil
  92         }
  93         return v.(error)
  94 }
  95
  96 // RemoveTransaction remove a transaction from the pool
  97 func (mp *TxPool) RemoveTransaction(txHash *bc.Hash) {
  98         mp.mtx.Lock()
  99         defer mp.mtx.Unlock()
 100
 101         if _, ok := mp.pool[*txHash]; ok {
 102                 delete(mp.pool, *txHash)
 103                 atomic.StoreInt64(&mp.lastUpdated, time.Now().Unix())
 104         }
 105 }
 106
 107 // GetTransaction return the TxDesc by hash
 108 func (mp *TxPool) GetTransaction(txHash *bc.Hash) (*TxDesc, error) {
 109         mp.mtx.RLock()
 110         defer mp.mtx.RUnlock()
 111
 112         if txD, ok := mp.pool[*txHash]; ok {
 113                 return txD, nil
 114         }
 115
 116         return nil, ErrTransactionNotExist
 117 }
 118
 119 // GetTransactions return all the transactions in the pool
 120 func (mp *TxPool) GetTransactions() []*TxDesc {
 121         mp.mtx.RLock()
 122         defer mp.mtx.RUnlock()
 123
 124         txDs := make([]*TxDesc, len(mp.pool))
 125         i := 0
 126         for _, desc := range mp.pool {
 127                 txDs[i] = desc
 128                 i++
 129         }
 130         return txDs
 131 }
 132
 133 // IsTransactionInPool check wheather a transaction in pool or not
 134 func (mp *TxPool) IsTransactionInPool(txHash *bc.Hash) bool {
 135         mp.mtx.RLock()
 136         defer mp.mtx.RUnlock()
 137
 138         if _, ok := mp.pool[*txHash]; ok {
 139                 return true
 140         }
 141         return false
 142 }
 143
 144 // IsTransactionInErrCache check wheather a transaction in errCache or not
 145 func (mp *TxPool) IsTransactionInErrCache(txHash *bc.Hash) bool {
 146         mp.mtx.RLock()
 147         defer mp.mtx.RUnlock()
 148
 149         _, ok := mp.errCache.Get(txHash)
 150         return ok
 151 }
 152
 153 // HaveTransaction IsTransactionInErrCache check is  transaction in errCache or pool
 154 func (mp *TxPool) HaveTransaction(txHash *bc.Hash) bool {
 155         return mp.IsTransactionInPool(txHash) || mp.IsTransactionInErrCache(txHash)
 156 }
 157
 158 // Count return number of transcation in pool
 159 func (mp *TxPool) Count() int {
 160         mp.mtx.RLock()
 161         defer mp.mtx.RUnlock()
 162
 163         count := len(mp.pool)
 164         return count
 165 }