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.

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