You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

425 lines
11 KiB

9 years ago
9 years ago
8 years ago
8 years ago
9 years ago
9 years ago
9 years ago
8 years ago
8 years ago
9 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
9 years ago
8 years ago
8 years ago
9 years ago
8 years ago
9 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
9 years ago
9 years ago
8 years ago
9 years ago
9 years ago
8 years ago
9 years ago
9 years ago
9 years ago
9 years ago
8 years ago
9 years ago
  1. package mempool
  2. import (
  3. "bytes"
  4. "container/list"
  5. "sync"
  6. "sync/atomic"
  7. "time"
  8. "github.com/pkg/errors"
  9. abci "github.com/tendermint/abci/types"
  10. auto "github.com/tendermint/tmlibs/autofile"
  11. "github.com/tendermint/tmlibs/clist"
  12. cmn "github.com/tendermint/tmlibs/common"
  13. "github.com/tendermint/tmlibs/log"
  14. cfg "github.com/tendermint/tendermint/config"
  15. "github.com/tendermint/tendermint/proxy"
  16. "github.com/tendermint/tendermint/types"
  17. )
  18. /*
  19. The mempool pushes new txs onto the proxyAppConn.
  20. It gets a stream of (req, res) tuples from the proxy.
  21. The memool stores good txs in a concurrent linked-list.
  22. Multiple concurrent go-routines can traverse this linked-list
  23. safely by calling .NextWait() on each element.
  24. So we have several go-routines:
  25. 1. Consensus calling Update() and Reap() synchronously
  26. 2. Many mempool reactor's peer routines calling CheckTx()
  27. 3. Many mempool reactor's peer routines traversing the txs linked list
  28. 4. Another goroutine calling GarbageCollectTxs() periodically
  29. To manage these goroutines, there are three methods of locking.
  30. 1. Mutations to the linked-list is protected by an internal mtx (CList is goroutine-safe)
  31. 2. Mutations to the linked-list elements are atomic
  32. 3. CheckTx() calls can be paused upon Update() and Reap(), protected by .proxyMtx
  33. Garbage collection of old elements from mempool.txs is handlde via
  34. the DetachPrev() call, which makes old elements not reachable by
  35. peer broadcastTxRoutine() automatically garbage collected.
  36. TODO: Better handle abci client errors. (make it automatically handle connection errors)
  37. */
  38. const cacheSize = 100000
  39. type Mempool struct {
  40. config *cfg.MempoolConfig
  41. proxyMtx sync.Mutex
  42. proxyAppConn proxy.AppConnMempool
  43. txs *clist.CList // concurrent linked-list of good txs
  44. counter int64 // simple incrementing counter
  45. height int // the last block Update()'d to
  46. rechecking int32 // for re-checking filtered txs on Update()
  47. recheckCursor *clist.CElement // next expected response
  48. recheckEnd *clist.CElement // re-checking stops here
  49. // Keep a cache of already-seen txs.
  50. // This reduces the pressure on the proxyApp.
  51. cache *txCache
  52. // A log of mempool txs
  53. wal *auto.AutoFile
  54. logger log.Logger
  55. }
  56. func NewMempool(config *cfg.MempoolConfig, proxyAppConn proxy.AppConnMempool) *Mempool {
  57. mempool := &Mempool{
  58. config: config,
  59. proxyAppConn: proxyAppConn,
  60. txs: clist.New(),
  61. counter: 0,
  62. height: 0,
  63. rechecking: 0,
  64. recheckCursor: nil,
  65. recheckEnd: nil,
  66. logger: log.NewNopLogger(),
  67. cache: newTxCache(cacheSize),
  68. }
  69. mempool.initWAL()
  70. proxyAppConn.SetResponseCallback(mempool.resCb)
  71. return mempool
  72. }
  73. // SetLogger allows you to set your own Logger.
  74. func (mem *Mempool) SetLogger(l log.Logger) {
  75. mem.logger = l
  76. }
  77. func (mem *Mempool) initWAL() {
  78. walDir := mem.config.WalDir()
  79. if walDir != "" {
  80. err := cmn.EnsureDir(walDir, 0700)
  81. if err != nil {
  82. cmn.PanicSanity(errors.Wrap(err, "Error ensuring Mempool wal dir"))
  83. }
  84. af, err := auto.OpenAutoFile(walDir + "/wal")
  85. if err != nil {
  86. cmn.PanicSanity(errors.Wrap(err, "Error opening Mempool wal file"))
  87. }
  88. mem.wal = af
  89. }
  90. }
  91. // consensus must be able to hold lock to safely update
  92. func (mem *Mempool) Lock() {
  93. mem.proxyMtx.Lock()
  94. }
  95. func (mem *Mempool) Unlock() {
  96. mem.proxyMtx.Unlock()
  97. }
  98. // Number of transactions in the mempool clist
  99. func (mem *Mempool) Size() int {
  100. return mem.txs.Len()
  101. }
  102. // Remove all transactions from mempool and cache
  103. func (mem *Mempool) Flush() {
  104. mem.proxyMtx.Lock()
  105. defer mem.proxyMtx.Unlock()
  106. mem.cache.Reset()
  107. for e := mem.txs.Front(); e != nil; e = e.Next() {
  108. mem.txs.Remove(e)
  109. e.DetachPrev()
  110. }
  111. }
  112. // Return the first element of mem.txs for peer goroutines to call .NextWait() on.
  113. // Blocks until txs has elements.
  114. func (mem *Mempool) TxsFrontWait() *clist.CElement {
  115. return mem.txs.FrontWait()
  116. }
  117. // Try a new transaction in the mempool.
  118. // Potentially blocking if we're blocking on Update() or Reap().
  119. // cb: A callback from the CheckTx command.
  120. // It gets called from another goroutine.
  121. // CONTRACT: Either cb will get called, or err returned.
  122. func (mem *Mempool) CheckTx(tx types.Tx, cb func(*abci.Response)) (err error) {
  123. mem.proxyMtx.Lock()
  124. defer mem.proxyMtx.Unlock()
  125. // CACHE
  126. if mem.cache.Exists(tx) {
  127. if cb != nil {
  128. cb(&abci.Response{
  129. Value: &abci.Response_CheckTx{
  130. &abci.ResponseCheckTx{
  131. Code: abci.CodeType_BadNonce, // TODO or duplicate tx
  132. Log: "Duplicate transaction (ignored)",
  133. },
  134. },
  135. })
  136. }
  137. return nil
  138. }
  139. mem.cache.Push(tx)
  140. // END CACHE
  141. // WAL
  142. if mem.wal != nil {
  143. // TODO: Notify administrators when WAL fails
  144. mem.wal.Write([]byte(tx))
  145. mem.wal.Write([]byte("\n"))
  146. }
  147. // END WAL
  148. // NOTE: proxyAppConn may error if tx buffer is full
  149. if err = mem.proxyAppConn.Error(); err != nil {
  150. return err
  151. }
  152. reqRes := mem.proxyAppConn.CheckTxAsync(tx)
  153. if cb != nil {
  154. reqRes.SetCallback(cb)
  155. }
  156. return nil
  157. }
  158. // ABCI callback function
  159. func (mem *Mempool) resCb(req *abci.Request, res *abci.Response) {
  160. if mem.recheckCursor == nil {
  161. mem.resCbNormal(req, res)
  162. } else {
  163. mem.resCbRecheck(req, res)
  164. }
  165. }
  166. func (mem *Mempool) resCbNormal(req *abci.Request, res *abci.Response) {
  167. switch r := res.Value.(type) {
  168. case *abci.Response_CheckTx:
  169. if r.CheckTx.Code == abci.CodeType_OK {
  170. mem.counter++
  171. memTx := &mempoolTx{
  172. counter: mem.counter,
  173. height: int64(mem.height),
  174. tx: req.GetCheckTx().Tx,
  175. }
  176. mem.txs.PushBack(memTx)
  177. } else {
  178. // ignore bad transaction
  179. mem.logger.Info("Bad Transaction", "res", r)
  180. // remove from cache (it might be good later)
  181. mem.cache.Remove(req.GetCheckTx().Tx)
  182. // TODO: handle other retcodes
  183. }
  184. default:
  185. // ignore other messages
  186. }
  187. }
  188. func (mem *Mempool) resCbRecheck(req *abci.Request, res *abci.Response) {
  189. switch r := res.Value.(type) {
  190. case *abci.Response_CheckTx:
  191. memTx := mem.recheckCursor.Value.(*mempoolTx)
  192. if !bytes.Equal(req.GetCheckTx().Tx, memTx.tx) {
  193. cmn.PanicSanity(cmn.Fmt("Unexpected tx response from proxy during recheck\n"+
  194. "Expected %X, got %X", r.CheckTx.Data, memTx.tx))
  195. }
  196. if r.CheckTx.Code == abci.CodeType_OK {
  197. // Good, nothing to do.
  198. } else {
  199. // Tx became invalidated due to newly committed block.
  200. mem.txs.Remove(mem.recheckCursor)
  201. mem.recheckCursor.DetachPrev()
  202. // remove from cache (it might be good later)
  203. mem.cache.Remove(req.GetCheckTx().Tx)
  204. }
  205. if mem.recheckCursor == mem.recheckEnd {
  206. mem.recheckCursor = nil
  207. } else {
  208. mem.recheckCursor = mem.recheckCursor.Next()
  209. }
  210. if mem.recheckCursor == nil {
  211. // Done!
  212. atomic.StoreInt32(&mem.rechecking, 0)
  213. mem.logger.Info("Done rechecking txs")
  214. }
  215. default:
  216. // ignore other messages
  217. }
  218. }
  219. // Get the valid transactions remaining
  220. // If maxTxs is -1, there is no cap on returned transactions.
  221. func (mem *Mempool) Reap(maxTxs int) types.Txs {
  222. mem.proxyMtx.Lock()
  223. defer mem.proxyMtx.Unlock()
  224. for atomic.LoadInt32(&mem.rechecking) > 0 {
  225. // TODO: Something better?
  226. time.Sleep(time.Millisecond * 10)
  227. }
  228. txs := mem.collectTxs(maxTxs)
  229. return txs
  230. }
  231. // maxTxs: -1 means uncapped, 0 means none
  232. func (mem *Mempool) collectTxs(maxTxs int) types.Txs {
  233. if maxTxs == 0 {
  234. return []types.Tx{}
  235. } else if maxTxs < 0 {
  236. maxTxs = mem.txs.Len()
  237. }
  238. txs := make([]types.Tx, 0, cmn.MinInt(mem.txs.Len(), maxTxs))
  239. for e := mem.txs.Front(); e != nil && len(txs) < maxTxs; e = e.Next() {
  240. memTx := e.Value.(*mempoolTx)
  241. txs = append(txs, memTx.tx)
  242. }
  243. return txs
  244. }
  245. // Tell mempool that these txs were committed.
  246. // Mempool will discard these txs.
  247. // NOTE: this should be called *after* block is committed by consensus.
  248. // NOTE: unsafe; Lock/Unlock must be managed by caller
  249. func (mem *Mempool) Update(height int, txs types.Txs) {
  250. // TODO: check err ?
  251. mem.proxyAppConn.FlushSync() // To flush async resCb calls e.g. from CheckTx
  252. // First, create a lookup map of txns in new txs.
  253. txsMap := make(map[string]struct{})
  254. for _, tx := range txs {
  255. txsMap[string(tx)] = struct{}{}
  256. }
  257. // Set height
  258. mem.height = height
  259. // Remove transactions that are already in txs.
  260. goodTxs := mem.filterTxs(txsMap)
  261. // Recheck mempool txs if any txs were committed in the block
  262. // NOTE/XXX: in some apps a tx could be invalidated due to EndBlock,
  263. // so we really still do need to recheck, but this is for debugging
  264. if mem.config.Recheck && (mem.config.RecheckEmpty || len(txs) > 0) {
  265. mem.logger.Info("Recheck txs", "numtxs", len(goodTxs))
  266. mem.recheckTxs(goodTxs)
  267. // At this point, mem.txs are being rechecked.
  268. // mem.recheckCursor re-scans mem.txs and possibly removes some txs.
  269. // Before mem.Reap(), we should wait for mem.recheckCursor to be nil.
  270. }
  271. }
  272. func (mem *Mempool) filterTxs(blockTxsMap map[string]struct{}) []types.Tx {
  273. goodTxs := make([]types.Tx, 0, mem.txs.Len())
  274. for e := mem.txs.Front(); e != nil; e = e.Next() {
  275. memTx := e.Value.(*mempoolTx)
  276. // Remove the tx if it's alredy in a block.
  277. if _, ok := blockTxsMap[string(memTx.tx)]; ok {
  278. // remove from clist
  279. mem.txs.Remove(e)
  280. e.DetachPrev()
  281. // NOTE: we don't remove committed txs from the cache.
  282. continue
  283. }
  284. // Good tx!
  285. goodTxs = append(goodTxs, memTx.tx)
  286. }
  287. return goodTxs
  288. }
  289. // NOTE: pass in goodTxs because mem.txs can mutate concurrently.
  290. func (mem *Mempool) recheckTxs(goodTxs []types.Tx) {
  291. if len(goodTxs) == 0 {
  292. return
  293. }
  294. atomic.StoreInt32(&mem.rechecking, 1)
  295. mem.recheckCursor = mem.txs.Front()
  296. mem.recheckEnd = mem.txs.Back()
  297. // Push txs to proxyAppConn
  298. // NOTE: resCb() may be called concurrently.
  299. for _, tx := range goodTxs {
  300. mem.proxyAppConn.CheckTxAsync(tx)
  301. }
  302. mem.proxyAppConn.FlushAsync()
  303. }
  304. //--------------------------------------------------------------------------------
  305. // A transaction that successfully ran
  306. type mempoolTx struct {
  307. counter int64 // a simple incrementing counter
  308. height int64 // height that this tx had been validated in
  309. tx types.Tx //
  310. }
  311. func (memTx *mempoolTx) Height() int {
  312. return int(atomic.LoadInt64(&memTx.height))
  313. }
  314. //--------------------------------------------------------------------------------
  315. type txCache struct {
  316. mtx sync.Mutex
  317. size int
  318. map_ map[string]struct{}
  319. list *list.List // to remove oldest tx when cache gets too big
  320. }
  321. func newTxCache(cacheSize int) *txCache {
  322. return &txCache{
  323. size: cacheSize,
  324. map_: make(map[string]struct{}, cacheSize),
  325. list: list.New(),
  326. }
  327. }
  328. func (cache *txCache) Reset() {
  329. cache.mtx.Lock()
  330. cache.map_ = make(map[string]struct{}, cacheSize)
  331. cache.list.Init()
  332. cache.mtx.Unlock()
  333. }
  334. func (cache *txCache) Exists(tx types.Tx) bool {
  335. cache.mtx.Lock()
  336. _, exists := cache.map_[string(tx)]
  337. cache.mtx.Unlock()
  338. return exists
  339. }
  340. // Returns false if tx is in cache.
  341. func (cache *txCache) Push(tx types.Tx) bool {
  342. cache.mtx.Lock()
  343. defer cache.mtx.Unlock()
  344. if _, exists := cache.map_[string(tx)]; exists {
  345. return false
  346. }
  347. if cache.list.Len() >= cache.size {
  348. popped := cache.list.Front()
  349. poppedTx := popped.Value.(types.Tx)
  350. // NOTE: the tx may have already been removed from the map
  351. // but deleting a non-existent element is fine
  352. delete(cache.map_, string(poppedTx))
  353. cache.list.Remove(popped)
  354. }
  355. cache.map_[string(tx)] = struct{}{}
  356. cache.list.PushBack(tx)
  357. return true
  358. }
  359. func (cache *txCache) Remove(tx types.Tx) {
  360. cache.mtx.Lock()
  361. delete(cache.map_, string(tx))
  362. cache.mtx.Unlock()
  363. }