From 593c025f1657f66dd74a8905c3a41cc01ea6b066 Mon Sep 17 00:00:00 2001 From: Aleksandr Bezobchuk Date: Fri, 3 Jun 2022 12:33:15 -0400 Subject: [PATCH] updates --- mempool/cache.go | 107 +++++++++++++++ mempool/clist_mempool.go | 114 ---------------- mempool/priority_queue.go | 159 +++++++++++++++++++++ mempool/tx.go | 281 ++++++++++++++++++++++++++++++++++++++ types/tx.go | 18 ++- 5 files changed, 561 insertions(+), 118 deletions(-) create mode 100644 mempool/cache.go create mode 100644 mempool/priority_queue.go create mode 100644 mempool/tx.go diff --git a/mempool/cache.go b/mempool/cache.go new file mode 100644 index 000000000..cdaefe762 --- /dev/null +++ b/mempool/cache.go @@ -0,0 +1,107 @@ +package mempool + +import ( + "container/list" + + tmsync "github.com/tendermint/tendermint/libs/sync" + "github.com/tendermint/tendermint/types" +) + +// TxCache defines an interface for raw transaction caching in a mempool. +// Currently, a TxCache does not allow direct reading or getting of transaction +// values. A TxCache is used primarily to push transactions and removing +// transactions. Pushing via Push returns a boolean telling the caller if the +// transaction already exists in the cache or not. +type TxCache interface { + // Reset resets the cache to an empty state. + Reset() + + // Push adds the given raw transaction to the cache and returns true if it was + // newly added. Otherwise, it returns false. + Push(tx types.Tx) bool + + // Remove removes the given raw transaction from the cache. + Remove(tx types.Tx) +} + +var _ TxCache = (*LRUTxCache)(nil) + +// LRUTxCache maintains a thread-safe LRU cache of raw transactions. The cache +// only stores the hash of the raw transaction. +type LRUTxCache struct { + mtx tmsync.Mutex + size int + cacheMap map[types.TxKey]*list.Element + list *list.List +} + +func NewLRUTxCache(cacheSize int) *LRUTxCache { + return &LRUTxCache{ + size: cacheSize, + cacheMap: make(map[types.TxKey]*list.Element, cacheSize), + list: list.New(), + } +} + +// GetList returns the underlying linked-list that backs the LRU cache. Note, +// this should be used for testing purposes only! +func (c *LRUTxCache) GetList() *list.List { + return c.list +} + +func (c *LRUTxCache) Reset() { + c.mtx.Lock() + defer c.mtx.Unlock() + + c.cacheMap = make(map[types.TxKey]*list.Element, c.size) + c.list.Init() +} + +func (c *LRUTxCache) Push(tx types.Tx) bool { + c.mtx.Lock() + defer c.mtx.Unlock() + + key := tx.Key() + + moved, ok := c.cacheMap[key] + if ok { + c.list.MoveToBack(moved) + return false + } + + if c.list.Len() >= c.size { + front := c.list.Front() + if front != nil { + frontKey := front.Value.(types.TxKey) + delete(c.cacheMap, frontKey) + c.list.Remove(front) + } + } + + e := c.list.PushBack(key) + c.cacheMap[key] = e + + return true +} + +func (c *LRUTxCache) Remove(tx types.Tx) { + c.mtx.Lock() + defer c.mtx.Unlock() + + key := tx.Key() + e := c.cacheMap[key] + delete(c.cacheMap, key) + + if e != nil { + c.list.Remove(e) + } +} + +// NopTxCache defines a no-op raw transaction cache. +type NopTxCache struct{} + +var _ TxCache = (*NopTxCache)(nil) + +func (NopTxCache) Reset() {} +func (NopTxCache) Push(types.Tx) bool { return true } +func (NopTxCache) Remove(types.Tx) {} diff --git a/mempool/clist_mempool.go b/mempool/clist_mempool.go index 3b3709588..fc2367a7c 100644 --- a/mempool/clist_mempool.go +++ b/mempool/clist_mempool.go @@ -2,7 +2,6 @@ package mempool import ( "bytes" - "container/list" "crypto/sha256" "fmt" "sync" @@ -648,116 +647,3 @@ func (mem *CListMempool) recheckTxs() { mem.proxyAppConn.FlushAsync() } - -//-------------------------------------------------------------------------------- - -// mempoolTx is a transaction that successfully ran -type mempoolTx struct { - height int64 // height that this tx had been validated in - gasWanted int64 // amount of gas this tx states it will require - tx types.Tx // - - // ids of peers who've sent us this tx (as a map for quick lookups). - // senders: PeerID -> bool - senders sync.Map -} - -// Height returns the height for this transaction -func (memTx *mempoolTx) Height() int64 { - return atomic.LoadInt64(&memTx.height) -} - -//-------------------------------------------------------------------------------- - -type txCache interface { - Reset() - Push(tx types.Tx) bool - Remove(tx types.Tx) -} - -// mapTxCache maintains a LRU cache of transactions. This only stores the hash -// of the tx, due to memory concerns. -type mapTxCache struct { - mtx tmsync.Mutex - size int - cacheMap map[[TxKeySize]byte]*list.Element - list *list.List -} - -var _ txCache = (*mapTxCache)(nil) - -// newMapTxCache returns a new mapTxCache. -func newMapTxCache(cacheSize int) *mapTxCache { - return &mapTxCache{ - size: cacheSize, - cacheMap: make(map[[TxKeySize]byte]*list.Element, cacheSize), - list: list.New(), - } -} - -// Reset resets the cache to an empty state. -func (cache *mapTxCache) Reset() { - cache.mtx.Lock() - cache.cacheMap = make(map[[TxKeySize]byte]*list.Element, cache.size) - cache.list.Init() - cache.mtx.Unlock() -} - -// Push adds the given tx to the cache and returns true. It returns -// false if tx is already in the cache. -func (cache *mapTxCache) Push(tx types.Tx) bool { - cache.mtx.Lock() - defer cache.mtx.Unlock() - - // Use the tx hash in the cache - txHash := TxKey(tx) - if moved, exists := cache.cacheMap[txHash]; exists { - cache.list.MoveToBack(moved) - return false - } - - if cache.list.Len() >= cache.size { - popped := cache.list.Front() - if popped != nil { - poppedTxHash := popped.Value.([TxKeySize]byte) - delete(cache.cacheMap, poppedTxHash) - cache.list.Remove(popped) - } - } - e := cache.list.PushBack(txHash) - cache.cacheMap[txHash] = e - return true -} - -// Remove removes the given tx from the cache. -func (cache *mapTxCache) Remove(tx types.Tx) { - cache.mtx.Lock() - txHash := TxKey(tx) - popped := cache.cacheMap[txHash] - delete(cache.cacheMap, txHash) - if popped != nil { - cache.list.Remove(popped) - } - - cache.mtx.Unlock() -} - -type nopTxCache struct{} - -var _ txCache = (*nopTxCache)(nil) - -func (nopTxCache) Reset() {} -func (nopTxCache) Push(types.Tx) bool { return true } -func (nopTxCache) Remove(types.Tx) {} - -//-------------------------------------------------------------------------------- - -// TxKey is the fixed length array hash used as the key in maps. -func TxKey(tx types.Tx) [TxKeySize]byte { - return sha256.Sum256(tx) -} - -// txID is a hash of the Tx. -func txID(tx []byte) []byte { - return types.Tx(tx).Hash() -} diff --git a/mempool/priority_queue.go b/mempool/priority_queue.go new file mode 100644 index 000000000..f381f23e8 --- /dev/null +++ b/mempool/priority_queue.go @@ -0,0 +1,159 @@ +package mempool + +import ( + "container/heap" + "sort" + + tmsync "github.com/tendermint/tendermint/libs/sync" +) + +var _ heap.Interface = (*TxPriorityQueue)(nil) + +// TxPriorityQueue defines a thread-safe priority queue for valid transactions. +type TxPriorityQueue struct { + mtx tmsync.RWMutex + txs []*WrappedTx +} + +func NewTxPriorityQueue() *TxPriorityQueue { + pq := &TxPriorityQueue{ + txs: make([]*WrappedTx, 0), + } + + heap.Init(pq) + + return pq +} + +// GetEvictableTxs attempts to find and return a list of *WrappedTx than can be +// evicted to make room for another *WrappedTx with higher priority. If no such +// list of *WrappedTx exists, nil will be returned. The returned list of *WrappedTx +// indicate that these transactions can be removed due to them being of lower +// priority and that their total sum in size allows room for the incoming +// transaction according to the mempool's configured limits. +func (pq *TxPriorityQueue) GetEvictableTxs(priority, txSize, totalSize, cap int64) []*WrappedTx { + pq.mtx.RLock() + defer pq.mtx.RUnlock() + + txs := make([]*WrappedTx, len(pq.txs)) + copy(txs, pq.txs) + + sort.Slice(txs, func(i, j int) bool { + return txs[i].priority < txs[j].priority + }) + + var ( + toEvict []*WrappedTx + i int + ) + + currSize := totalSize + + // Loop over all transactions in ascending priority order evaluating those + // that are only of less priority than the provided argument. We continue + // evaluating transactions until there is sufficient capacity for the new + // transaction (size) as defined by txSize. + for i < len(txs) && txs[i].priority < priority { + toEvict = append(toEvict, txs[i]) + currSize -= int64(txs[i].Size()) + + if currSize+txSize <= cap { + return toEvict + } + + i++ + } + + return nil +} + +// NumTxs returns the number of transactions in the priority queue. It is +// thread safe. +func (pq *TxPriorityQueue) NumTxs() int { + pq.mtx.RLock() + defer pq.mtx.RUnlock() + + return len(pq.txs) +} + +// RemoveTx removes a specific transaction from the priority queue. +func (pq *TxPriorityQueue) RemoveTx(tx *WrappedTx) { + pq.mtx.Lock() + defer pq.mtx.Unlock() + + if tx.heapIndex < len(pq.txs) { + heap.Remove(pq, tx.heapIndex) + } +} + +// PushTx adds a valid transaction to the priority queue. It is thread safe. +func (pq *TxPriorityQueue) PushTx(tx *WrappedTx) { + pq.mtx.Lock() + defer pq.mtx.Unlock() + + heap.Push(pq, tx) +} + +// PopTx removes the top priority transaction from the queue. It is thread safe. +func (pq *TxPriorityQueue) PopTx() *WrappedTx { + pq.mtx.Lock() + defer pq.mtx.Unlock() + + x := heap.Pop(pq) + if x != nil { + return x.(*WrappedTx) + } + + return nil +} + +// Push implements the Heap interface. +// +// NOTE: A caller should never call Push. Use PushTx instead. +func (pq *TxPriorityQueue) Push(x interface{}) { + n := len(pq.txs) + item := x.(*WrappedTx) + item.heapIndex = n + pq.txs = append(pq.txs, item) +} + +// Pop implements the Heap interface. +// +// NOTE: A caller should never call Pop. Use PopTx instead. +func (pq *TxPriorityQueue) Pop() interface{} { + old := pq.txs + n := len(old) + item := old[n-1] + old[n-1] = nil // avoid memory leak + item.heapIndex = -1 // for safety + pq.txs = old[0 : n-1] + return item +} + +// Len implements the Heap interface. +// +// NOTE: A caller should never call Len. Use NumTxs instead. +func (pq *TxPriorityQueue) Len() int { + return len(pq.txs) +} + +// Less implements the Heap interface. It returns true if the transaction at +// position i in the queue is of less priority than the transaction at position j. +func (pq *TxPriorityQueue) Less(i, j int) bool { + // If there exists two transactions with the same priority, consider the one + // that we saw the earliest as the higher priority transaction. + if pq.txs[i].priority == pq.txs[j].priority { + return pq.txs[i].timestamp.Before(pq.txs[j].timestamp) + } + + // We want Pop to give us the highest, not lowest, priority so we use greater + // than here. + return pq.txs[i].priority > pq.txs[j].priority +} + +// Swap implements the Heap interface. It swaps two transactions in the queue. +func (pq *TxPriorityQueue) Swap(i, j int) { + pq.txs[i], pq.txs[j] = pq.txs[j], pq.txs[i] + pq.txs[i].heapIndex = i + pq.txs[j].heapIndex = j +} diff --git a/mempool/tx.go b/mempool/tx.go new file mode 100644 index 000000000..f8d02f0e9 --- /dev/null +++ b/mempool/tx.go @@ -0,0 +1,281 @@ +package mempool + +import ( + "sort" + "time" + + "github.com/tendermint/tendermint/libs/clist" + tmsync "github.com/tendermint/tendermint/libs/sync" + "github.com/tendermint/tendermint/types" +) + +// WrappedTx defines a wrapper around a raw transaction with additional metadata +// that is used for indexing. +type WrappedTx struct { + // tx represents the raw binary transaction data + tx types.Tx + + // hash defines the transaction hash and the primary key used in the mempool + hash types.TxKey + + // height defines the height at which the transaction was validated at + height int64 + + // gasWanted defines the amount of gas the transaction sender requires + gasWanted int64 + + // priority defines the transaction's priority as specified by the application + // in the ResponseCheckTx response. + priority int64 + + // sender defines the transaction's sender as specified by the application in + // the ResponseCheckTx response. + sender string + + // timestamp is the time at which the node first received the transaction from + // a peer. It is used as a second dimension is prioritizing transactions when + // two transactions have the same priority. + timestamp time.Time + + // peers records a mapping of all peers that sent a given transaction + peers map[uint16]struct{} + + // heapIndex defines the index of the item in the heap + heapIndex int + + // gossipEl references the linked-list element in the gossip index + gossipEl *clist.CElement + + // removed marks the transaction as removed from the mempool. This is set + // during RemoveTx and is needed due to the fact that a given existing + // transaction in the mempool can be evicted when it is simultaneously having + // a reCheckTx callback executed. + removed bool +} + +func (wtx *WrappedTx) Size() int { + return len(wtx.tx) +} + +// TxStore implements a thread-safe mapping of valid transaction(s). +// +// NOTE: +// - Concurrent read-only access to a *WrappedTx object is OK. However, mutative +// access is not allowed. Regardless, it is not expected for the mempool to +// need mutative access. +type TxStore struct { + mtx tmsync.RWMutex + hashTxs map[types.TxKey]*WrappedTx // primary index + senderTxs map[string]*WrappedTx // sender is defined by the ABCI application +} + +func NewTxStore() *TxStore { + return &TxStore{ + senderTxs: make(map[string]*WrappedTx), + hashTxs: make(map[types.TxKey]*WrappedTx), + } +} + +// Size returns the total number of transactions in the store. +func (txs *TxStore) Size() int { + txs.mtx.RLock() + defer txs.mtx.RUnlock() + + return len(txs.hashTxs) +} + +// GetAllTxs returns all the transactions currently in the store. +func (txs *TxStore) GetAllTxs() []*WrappedTx { + txs.mtx.RLock() + defer txs.mtx.RUnlock() + + wTxs := make([]*WrappedTx, len(txs.hashTxs)) + i := 0 + for _, wtx := range txs.hashTxs { + wTxs[i] = wtx + i++ + } + + return wTxs +} + +// GetTxBySender returns a *WrappedTx by the transaction's sender property +// defined by the ABCI application. +func (txs *TxStore) GetTxBySender(sender string) *WrappedTx { + txs.mtx.RLock() + defer txs.mtx.RUnlock() + + return txs.senderTxs[sender] +} + +// GetTxByHash returns a *WrappedTx by the transaction's hash. +func (txs *TxStore) GetTxByHash(hash types.TxKey) *WrappedTx { + txs.mtx.RLock() + defer txs.mtx.RUnlock() + + return txs.hashTxs[hash] +} + +// IsTxRemoved returns true if a transaction by hash is marked as removed and +// false otherwise. +func (txs *TxStore) IsTxRemoved(hash types.TxKey) bool { + txs.mtx.RLock() + defer txs.mtx.RUnlock() + + wtx, ok := txs.hashTxs[hash] + if ok { + return wtx.removed + } + + return false +} + +// SetTx stores a *WrappedTx by it's hash. If the transaction also contains a +// non-empty sender, we additionally store the transaction by the sender as +// defined by the ABCI application. +func (txs *TxStore) SetTx(wtx *WrappedTx) { + txs.mtx.Lock() + defer txs.mtx.Unlock() + + if len(wtx.sender) > 0 { + txs.senderTxs[wtx.sender] = wtx + } + + txs.hashTxs[wtx.tx.Key()] = wtx +} + +// RemoveTx removes a *WrappedTx from the transaction store. It deletes all +// indexes of the transaction. +func (txs *TxStore) RemoveTx(wtx *WrappedTx) { + txs.mtx.Lock() + defer txs.mtx.Unlock() + + if len(wtx.sender) > 0 { + delete(txs.senderTxs, wtx.sender) + } + + delete(txs.hashTxs, wtx.tx.Key()) + wtx.removed = true +} + +// TxHasPeer returns true if a transaction by hash has a given peer ID and false +// otherwise. If the transaction does not exist, false is returned. +func (txs *TxStore) TxHasPeer(hash types.TxKey, peerID uint16) bool { + txs.mtx.RLock() + defer txs.mtx.RUnlock() + + wtx := txs.hashTxs[hash] + if wtx == nil { + return false + } + + _, ok := wtx.peers[peerID] + return ok +} + +// GetOrSetPeerByTxHash looks up a WrappedTx by transaction hash and adds the +// given peerID to the WrappedTx's set of peers that sent us this transaction. +// We return true if we've already recorded the given peer for this transaction +// and false otherwise. If the transaction does not exist by hash, we return +// (nil, false). +func (txs *TxStore) GetOrSetPeerByTxHash(hash types.TxKey, peerID uint16) (*WrappedTx, bool) { + txs.mtx.Lock() + defer txs.mtx.Unlock() + + wtx := txs.hashTxs[hash] + if wtx == nil { + return nil, false + } + + if wtx.peers == nil { + wtx.peers = make(map[uint16]struct{}) + } + + if _, ok := wtx.peers[peerID]; ok { + return wtx, true + } + + wtx.peers[peerID] = struct{}{} + return wtx, false +} + +// WrappedTxList implements a thread-safe list of *WrappedTx objects that can be +// used to build generic transaction indexes in the mempool. It accepts a +// comparator function, less(a, b *WrappedTx) bool, that compares two WrappedTx +// references which is used during Insert in order to determine sorted order. If +// less returns true, a <= b. +type WrappedTxList struct { + mtx tmsync.RWMutex + txs []*WrappedTx + less func(*WrappedTx, *WrappedTx) bool +} + +func NewWrappedTxList(less func(*WrappedTx, *WrappedTx) bool) *WrappedTxList { + return &WrappedTxList{ + txs: make([]*WrappedTx, 0), + less: less, + } +} + +// Size returns the number of WrappedTx objects in the list. +func (wtl *WrappedTxList) Size() int { + wtl.mtx.RLock() + defer wtl.mtx.RUnlock() + + return len(wtl.txs) +} + +// Reset resets the list of transactions to an empty list. +func (wtl *WrappedTxList) Reset() { + wtl.mtx.Lock() + defer wtl.mtx.Unlock() + + wtl.txs = make([]*WrappedTx, 0) +} + +// Insert inserts a WrappedTx reference into the sorted list based on the list's +// comparator function. +func (wtl *WrappedTxList) Insert(wtx *WrappedTx) { + wtl.mtx.Lock() + defer wtl.mtx.Unlock() + + i := sort.Search(len(wtl.txs), func(i int) bool { + return wtl.less(wtl.txs[i], wtx) + }) + + if i == len(wtl.txs) { + // insert at the end + wtl.txs = append(wtl.txs, wtx) + return + } + + // Make space for the inserted element by shifting values at the insertion + // index up one index. + // + // NOTE: The call to append does not allocate memory when cap(wtl.txs) > len(wtl.txs). + wtl.txs = append(wtl.txs[:i+1], wtl.txs[i:]...) + wtl.txs[i] = wtx +} + +// Remove attempts to remove a WrappedTx from the sorted list. +func (wtl *WrappedTxList) Remove(wtx *WrappedTx) { + wtl.mtx.Lock() + defer wtl.mtx.Unlock() + + i := sort.Search(len(wtl.txs), func(i int) bool { + return wtl.less(wtl.txs[i], wtx) + }) + + // Since the list is sorted, we evaluate all elements starting at i. Note, if + // the element does not exist, we may potentially evaluate the entire remainder + // of the list. However, a caller should not be expected to call Remove with a + // non-existing element. + for i < len(wtl.txs) { + if wtl.txs[i] == wtx { + wtl.txs = append(wtl.txs[:i], wtl.txs[i+1:]...) + return + } + + i++ + } +} diff --git a/types/tx.go b/types/tx.go index 92df92f13..37f86bec5 100644 --- a/types/tx.go +++ b/types/tx.go @@ -2,6 +2,7 @@ package types import ( "bytes" + "crypto/sha256" "errors" "fmt" @@ -11,16 +12,25 @@ import ( tmproto "github.com/tendermint/tendermint/proto/tendermint/types" ) -// Tx is an arbitrary byte array. -// NOTE: Tx has no types at this level, so when wire encoded it's just length-prefixed. -// Might we want types here ? -type Tx []byte +type ( + // Tx is an arbitrary byte array. + // NOTE: Tx has no types at this level, so when wire encoded it's just length-prefixed. + // Might we want types here ? + Tx []byte + + // TxKey is the fixed length array key used as an index. + TxKey [sha256.Size]byte +) // Hash computes the TMHASH hash of the wire encoded transaction. func (tx Tx) Hash() []byte { return tmhash.Sum(tx) } +func (tx Tx) Key() TxKey { + return sha256.Sum256(tx) +} + // String returns the hex-encoded transaction as a string. func (tx Tx) String() string { return fmt.Sprintf("Tx{%X}", []byte(tx))