mirror of
https://github.com/tendermint/tendermint.git
synced 2026-09-20 06:54:41 +00:00
Revert "delete everything" (includes everything non-go-crypto)
This reverts commit 96a3502
This commit is contained in:
Liamsi
@@ -0,0 +1,524 @@
|
||||
package mempool
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"container/list"
|
||||
"fmt"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/pkg/errors"
|
||||
|
||||
abci "github.com/tendermint/abci/types"
|
||||
auto "github.com/tendermint/tmlibs/autofile"
|
||||
"github.com/tendermint/tmlibs/clist"
|
||||
cmn "github.com/tendermint/tmlibs/common"
|
||||
"github.com/tendermint/tmlibs/log"
|
||||
|
||||
cfg "github.com/tendermint/tendermint/config"
|
||||
"github.com/tendermint/tendermint/proxy"
|
||||
"github.com/tendermint/tendermint/types"
|
||||
)
|
||||
|
||||
/*
|
||||
|
||||
The mempool pushes new txs onto the proxyAppConn.
|
||||
It gets a stream of (req, res) tuples from the proxy.
|
||||
The mempool stores good txs in a concurrent linked-list.
|
||||
|
||||
Multiple concurrent go-routines can traverse this linked-list
|
||||
safely by calling .NextWait() on each element.
|
||||
|
||||
So we have several go-routines:
|
||||
1. Consensus calling Update() and Reap() synchronously
|
||||
2. Many mempool reactor's peer routines calling CheckTx()
|
||||
3. Many mempool reactor's peer routines traversing the txs linked list
|
||||
4. Another goroutine calling GarbageCollectTxs() periodically
|
||||
|
||||
To manage these goroutines, there are three methods of locking.
|
||||
1. Mutations to the linked-list is protected by an internal mtx (CList is goroutine-safe)
|
||||
2. Mutations to the linked-list elements are atomic
|
||||
3. CheckTx() calls can be paused upon Update() and Reap(), protected by .proxyMtx
|
||||
|
||||
Garbage collection of old elements from mempool.txs is handlde via
|
||||
the DetachPrev() call, which makes old elements not reachable by
|
||||
peer broadcastTxRoutine() automatically garbage collected.
|
||||
|
||||
TODO: Better handle abci client errors. (make it automatically handle connection errors)
|
||||
|
||||
*/
|
||||
|
||||
var (
|
||||
// ErrTxInCache is returned to the client if we saw tx earlier
|
||||
ErrTxInCache = errors.New("Tx already exists in cache")
|
||||
|
||||
// ErrMempoolIsFull means Tendermint & an application can't handle that much load
|
||||
ErrMempoolIsFull = errors.New("Mempool is full")
|
||||
)
|
||||
|
||||
// Mempool is an ordered in-memory pool for transactions before they are proposed in a consensus
|
||||
// round. Transaction validity is checked using the CheckTx abci message before the transaction is
|
||||
// added to the pool. The Mempool uses a concurrent list structure for storing transactions that
|
||||
// can be efficiently accessed by multiple concurrent readers.
|
||||
type Mempool struct {
|
||||
config *cfg.MempoolConfig
|
||||
|
||||
proxyMtx sync.Mutex
|
||||
proxyAppConn proxy.AppConnMempool
|
||||
txs *clist.CList // concurrent linked-list of good txs
|
||||
counter int64 // simple incrementing counter
|
||||
height int64 // the last block Update()'d to
|
||||
rechecking int32 // for re-checking filtered txs on Update()
|
||||
recheckCursor *clist.CElement // next expected response
|
||||
recheckEnd *clist.CElement // re-checking stops here
|
||||
notifiedTxsAvailable bool
|
||||
txsAvailable chan int64 // fires the next height once for each height, when the mempool is not empty
|
||||
|
||||
// Keep a cache of already-seen txs.
|
||||
// This reduces the pressure on the proxyApp.
|
||||
cache *txCache
|
||||
|
||||
// A log of mempool txs
|
||||
wal *auto.AutoFile
|
||||
|
||||
logger log.Logger
|
||||
|
||||
metrics *Metrics
|
||||
}
|
||||
|
||||
// MempoolOption sets an optional parameter on the Mempool.
|
||||
type MempoolOption func(*Mempool)
|
||||
|
||||
// NewMempool returns a new Mempool with the given configuration and connection to an application.
|
||||
func NewMempool(
|
||||
config *cfg.MempoolConfig,
|
||||
proxyAppConn proxy.AppConnMempool,
|
||||
height int64,
|
||||
options ...MempoolOption,
|
||||
) *Mempool {
|
||||
mempool := &Mempool{
|
||||
config: config,
|
||||
proxyAppConn: proxyAppConn,
|
||||
txs: clist.New(),
|
||||
counter: 0,
|
||||
height: height,
|
||||
rechecking: 0,
|
||||
recheckCursor: nil,
|
||||
recheckEnd: nil,
|
||||
logger: log.NewNopLogger(),
|
||||
cache: newTxCache(config.CacheSize),
|
||||
metrics: NopMetrics(),
|
||||
}
|
||||
proxyAppConn.SetResponseCallback(mempool.resCb)
|
||||
for _, option := range options {
|
||||
option(mempool)
|
||||
}
|
||||
return mempool
|
||||
}
|
||||
|
||||
// EnableTxsAvailable initializes the TxsAvailable channel,
|
||||
// ensuring it will trigger once every height when transactions are available.
|
||||
// NOTE: not thread safe - should only be called once, on startup
|
||||
func (mem *Mempool) EnableTxsAvailable() {
|
||||
mem.txsAvailable = make(chan int64, 1)
|
||||
}
|
||||
|
||||
// SetLogger sets the Logger.
|
||||
func (mem *Mempool) SetLogger(l log.Logger) {
|
||||
mem.logger = l
|
||||
}
|
||||
|
||||
// WithMetrics sets the metrics.
|
||||
func WithMetrics(metrics *Metrics) MempoolOption {
|
||||
return func(mem *Mempool) { mem.metrics = metrics }
|
||||
}
|
||||
|
||||
// CloseWAL closes and discards the underlying WAL file.
|
||||
// Any further writes will not be relayed to disk.
|
||||
func (mem *Mempool) CloseWAL() bool {
|
||||
if mem == nil {
|
||||
return false
|
||||
}
|
||||
|
||||
mem.proxyMtx.Lock()
|
||||
defer mem.proxyMtx.Unlock()
|
||||
|
||||
if mem.wal == nil {
|
||||
return false
|
||||
}
|
||||
if err := mem.wal.Close(); err != nil && mem.logger != nil {
|
||||
mem.logger.Error("Mempool.CloseWAL", "err", err)
|
||||
}
|
||||
mem.wal = nil
|
||||
return true
|
||||
}
|
||||
|
||||
func (mem *Mempool) InitWAL() {
|
||||
walDir := mem.config.WalDir()
|
||||
if walDir != "" {
|
||||
err := cmn.EnsureDir(walDir, 0700)
|
||||
if err != nil {
|
||||
cmn.PanicSanity(errors.Wrap(err, "Error ensuring Mempool wal dir"))
|
||||
}
|
||||
af, err := auto.OpenAutoFile(walDir + "/wal")
|
||||
if err != nil {
|
||||
cmn.PanicSanity(errors.Wrap(err, "Error opening Mempool wal file"))
|
||||
}
|
||||
mem.wal = af
|
||||
}
|
||||
}
|
||||
|
||||
// Lock locks the mempool. The consensus must be able to hold lock to safely update.
|
||||
func (mem *Mempool) Lock() {
|
||||
mem.proxyMtx.Lock()
|
||||
}
|
||||
|
||||
// Unlock unlocks the mempool.
|
||||
func (mem *Mempool) Unlock() {
|
||||
mem.proxyMtx.Unlock()
|
||||
}
|
||||
|
||||
// Size returns the number of transactions in the mempool.
|
||||
func (mem *Mempool) Size() int {
|
||||
return mem.txs.Len()
|
||||
}
|
||||
|
||||
// Flushes the mempool connection to ensure async resCb calls are done e.g.
|
||||
// from CheckTx.
|
||||
func (mem *Mempool) FlushAppConn() error {
|
||||
return mem.proxyAppConn.FlushSync()
|
||||
}
|
||||
|
||||
// Flush removes all transactions from the mempool and cache
|
||||
func (mem *Mempool) Flush() {
|
||||
mem.proxyMtx.Lock()
|
||||
defer mem.proxyMtx.Unlock()
|
||||
|
||||
mem.cache.Reset()
|
||||
|
||||
for e := mem.txs.Front(); e != nil; e = e.Next() {
|
||||
mem.txs.Remove(e)
|
||||
e.DetachPrev()
|
||||
}
|
||||
}
|
||||
|
||||
// TxsFront returns the first transaction in the ordered list for peer
|
||||
// goroutines to call .NextWait() on.
|
||||
func (mem *Mempool) TxsFront() *clist.CElement {
|
||||
return mem.txs.Front()
|
||||
}
|
||||
|
||||
// TxsWaitChan returns a channel to wait on transactions. It will be closed
|
||||
// once the mempool is not empty (ie. the internal `mem.txs` has at least one
|
||||
// element)
|
||||
func (mem *Mempool) TxsWaitChan() <-chan struct{} {
|
||||
return mem.txs.WaitChan()
|
||||
}
|
||||
|
||||
// CheckTx executes a new transaction against the application to determine its validity
|
||||
// and whether it should be added to the mempool.
|
||||
// It blocks if we're waiting on Update() or Reap().
|
||||
// cb: A callback from the CheckTx command.
|
||||
// It gets called from another goroutine.
|
||||
// CONTRACT: Either cb will get called, or err returned.
|
||||
func (mem *Mempool) CheckTx(tx types.Tx, cb func(*abci.Response)) (err error) {
|
||||
mem.proxyMtx.Lock()
|
||||
defer mem.proxyMtx.Unlock()
|
||||
|
||||
if mem.Size() >= mem.config.Size {
|
||||
return ErrMempoolIsFull
|
||||
}
|
||||
|
||||
// CACHE
|
||||
if !mem.cache.Push(tx) {
|
||||
return ErrTxInCache
|
||||
}
|
||||
// END CACHE
|
||||
|
||||
// WAL
|
||||
if mem.wal != nil {
|
||||
// TODO: Notify administrators when WAL fails
|
||||
_, err := mem.wal.Write([]byte(tx))
|
||||
if err != nil {
|
||||
mem.logger.Error("Error writing to WAL", "err", err)
|
||||
}
|
||||
_, err = mem.wal.Write([]byte("\n"))
|
||||
if err != nil {
|
||||
mem.logger.Error("Error writing to WAL", "err", err)
|
||||
}
|
||||
}
|
||||
// END WAL
|
||||
|
||||
// NOTE: proxyAppConn may error if tx buffer is full
|
||||
if err = mem.proxyAppConn.Error(); err != nil {
|
||||
return err
|
||||
}
|
||||
reqRes := mem.proxyAppConn.CheckTxAsync(tx)
|
||||
if cb != nil {
|
||||
reqRes.SetCallback(cb)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// ABCI callback function
|
||||
func (mem *Mempool) resCb(req *abci.Request, res *abci.Response) {
|
||||
if mem.recheckCursor == nil {
|
||||
mem.resCbNormal(req, res)
|
||||
} else {
|
||||
mem.resCbRecheck(req, res)
|
||||
}
|
||||
mem.metrics.Size.Set(float64(mem.Size()))
|
||||
}
|
||||
|
||||
func (mem *Mempool) resCbNormal(req *abci.Request, res *abci.Response) {
|
||||
switch r := res.Value.(type) {
|
||||
case *abci.Response_CheckTx:
|
||||
tx := req.GetCheckTx().Tx
|
||||
if r.CheckTx.Code == abci.CodeTypeOK {
|
||||
mem.counter++
|
||||
memTx := &mempoolTx{
|
||||
counter: mem.counter,
|
||||
height: mem.height,
|
||||
tx: tx,
|
||||
}
|
||||
mem.txs.PushBack(memTx)
|
||||
mem.logger.Info("Added good transaction", "tx", fmt.Sprintf("%X", types.Tx(tx).Hash()), "res", r)
|
||||
mem.notifyTxsAvailable()
|
||||
} else {
|
||||
// ignore bad transaction
|
||||
mem.logger.Info("Rejected bad transaction", "tx", fmt.Sprintf("%X", types.Tx(tx).Hash()), "res", r)
|
||||
|
||||
// remove from cache (it might be good later)
|
||||
mem.cache.Remove(tx)
|
||||
}
|
||||
default:
|
||||
// ignore other messages
|
||||
}
|
||||
}
|
||||
|
||||
func (mem *Mempool) resCbRecheck(req *abci.Request, res *abci.Response) {
|
||||
switch r := res.Value.(type) {
|
||||
case *abci.Response_CheckTx:
|
||||
memTx := mem.recheckCursor.Value.(*mempoolTx)
|
||||
if !bytes.Equal(req.GetCheckTx().Tx, memTx.tx) {
|
||||
cmn.PanicSanity(cmn.Fmt("Unexpected tx response from proxy during recheck\n"+
|
||||
"Expected %X, got %X", r.CheckTx.Data, memTx.tx))
|
||||
}
|
||||
if r.CheckTx.Code == abci.CodeTypeOK {
|
||||
// Good, nothing to do.
|
||||
} else {
|
||||
// Tx became invalidated due to newly committed block.
|
||||
mem.txs.Remove(mem.recheckCursor)
|
||||
mem.recheckCursor.DetachPrev()
|
||||
|
||||
// remove from cache (it might be good later)
|
||||
mem.cache.Remove(req.GetCheckTx().Tx)
|
||||
}
|
||||
if mem.recheckCursor == mem.recheckEnd {
|
||||
mem.recheckCursor = nil
|
||||
} else {
|
||||
mem.recheckCursor = mem.recheckCursor.Next()
|
||||
}
|
||||
if mem.recheckCursor == nil {
|
||||
// Done!
|
||||
atomic.StoreInt32(&mem.rechecking, 0)
|
||||
mem.logger.Info("Done rechecking txs")
|
||||
|
||||
// incase the recheck removed all txs
|
||||
if mem.Size() > 0 {
|
||||
mem.notifyTxsAvailable()
|
||||
}
|
||||
}
|
||||
default:
|
||||
// ignore other messages
|
||||
}
|
||||
}
|
||||
|
||||
// TxsAvailable returns a channel which fires once for every height,
|
||||
// and only when transactions are available in the mempool.
|
||||
// NOTE: the returned channel may be nil if EnableTxsAvailable was not called.
|
||||
func (mem *Mempool) TxsAvailable() <-chan int64 {
|
||||
return mem.txsAvailable
|
||||
}
|
||||
|
||||
func (mem *Mempool) notifyTxsAvailable() {
|
||||
if mem.Size() == 0 {
|
||||
panic("notified txs available but mempool is empty!")
|
||||
}
|
||||
if mem.txsAvailable != nil && !mem.notifiedTxsAvailable {
|
||||
select {
|
||||
case mem.txsAvailable <- mem.height + 1:
|
||||
default:
|
||||
}
|
||||
|
||||
mem.notifiedTxsAvailable = true
|
||||
}
|
||||
}
|
||||
|
||||
// Reap returns a list of transactions currently in the mempool.
|
||||
// If maxTxs is -1, there is no cap on the number of returned transactions.
|
||||
func (mem *Mempool) Reap(maxTxs int) types.Txs {
|
||||
mem.proxyMtx.Lock()
|
||||
defer mem.proxyMtx.Unlock()
|
||||
|
||||
for atomic.LoadInt32(&mem.rechecking) > 0 {
|
||||
// TODO: Something better?
|
||||
time.Sleep(time.Millisecond * 10)
|
||||
}
|
||||
|
||||
txs := mem.collectTxs(maxTxs)
|
||||
return txs
|
||||
}
|
||||
|
||||
// maxTxs: -1 means uncapped, 0 means none
|
||||
func (mem *Mempool) collectTxs(maxTxs int) types.Txs {
|
||||
if maxTxs == 0 {
|
||||
return []types.Tx{}
|
||||
} else if maxTxs < 0 {
|
||||
maxTxs = mem.txs.Len()
|
||||
}
|
||||
txs := make([]types.Tx, 0, cmn.MinInt(mem.txs.Len(), maxTxs))
|
||||
for e := mem.txs.Front(); e != nil && len(txs) < maxTxs; e = e.Next() {
|
||||
memTx := e.Value.(*mempoolTx)
|
||||
txs = append(txs, memTx.tx)
|
||||
}
|
||||
return txs
|
||||
}
|
||||
|
||||
// Update informs the mempool that the given txs were committed and can be discarded.
|
||||
// NOTE: this should be called *after* block is committed by consensus.
|
||||
// NOTE: unsafe; Lock/Unlock must be managed by caller
|
||||
func (mem *Mempool) Update(height int64, txs types.Txs) error {
|
||||
// First, create a lookup map of txns in new txs.
|
||||
txsMap := make(map[string]struct{})
|
||||
for _, tx := range txs {
|
||||
txsMap[string(tx)] = struct{}{}
|
||||
}
|
||||
|
||||
// Set height
|
||||
mem.height = height
|
||||
mem.notifiedTxsAvailable = false
|
||||
|
||||
// Remove transactions that are already in txs.
|
||||
goodTxs := mem.filterTxs(txsMap)
|
||||
// Recheck mempool txs if any txs were committed in the block
|
||||
// NOTE/XXX: in some apps a tx could be invalidated due to EndBlock,
|
||||
// so we really still do need to recheck, but this is for debugging
|
||||
if mem.config.Recheck && (mem.config.RecheckEmpty || len(goodTxs) > 0) {
|
||||
mem.logger.Info("Recheck txs", "numtxs", len(goodTxs), "height", height)
|
||||
mem.recheckTxs(goodTxs)
|
||||
// At this point, mem.txs are being rechecked.
|
||||
// mem.recheckCursor re-scans mem.txs and possibly removes some txs.
|
||||
// Before mem.Reap(), we should wait for mem.recheckCursor to be nil.
|
||||
}
|
||||
mem.metrics.Size.Set(float64(mem.Size()))
|
||||
return nil
|
||||
}
|
||||
|
||||
func (mem *Mempool) filterTxs(blockTxsMap map[string]struct{}) []types.Tx {
|
||||
goodTxs := make([]types.Tx, 0, mem.txs.Len())
|
||||
for e := mem.txs.Front(); e != nil; e = e.Next() {
|
||||
memTx := e.Value.(*mempoolTx)
|
||||
// Remove the tx if it's alredy in a block.
|
||||
if _, ok := blockTxsMap[string(memTx.tx)]; ok {
|
||||
// remove from clist
|
||||
mem.txs.Remove(e)
|
||||
e.DetachPrev()
|
||||
|
||||
// NOTE: we don't remove committed txs from the cache.
|
||||
continue
|
||||
}
|
||||
// Good tx!
|
||||
goodTxs = append(goodTxs, memTx.tx)
|
||||
}
|
||||
return goodTxs
|
||||
}
|
||||
|
||||
// NOTE: pass in goodTxs because mem.txs can mutate concurrently.
|
||||
func (mem *Mempool) recheckTxs(goodTxs []types.Tx) {
|
||||
if len(goodTxs) == 0 {
|
||||
return
|
||||
}
|
||||
atomic.StoreInt32(&mem.rechecking, 1)
|
||||
mem.recheckCursor = mem.txs.Front()
|
||||
mem.recheckEnd = mem.txs.Back()
|
||||
|
||||
// Push txs to proxyAppConn
|
||||
// NOTE: resCb() may be called concurrently.
|
||||
for _, tx := range goodTxs {
|
||||
mem.proxyAppConn.CheckTxAsync(tx)
|
||||
}
|
||||
mem.proxyAppConn.FlushAsync()
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------
|
||||
|
||||
// mempoolTx is a transaction that successfully ran
|
||||
type mempoolTx struct {
|
||||
counter int64 // a simple incrementing counter
|
||||
height int64 // height that this tx had been validated in
|
||||
tx types.Tx //
|
||||
}
|
||||
|
||||
// Height returns the height for this transaction
|
||||
func (memTx *mempoolTx) Height() int64 {
|
||||
return atomic.LoadInt64(&memTx.height)
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------
|
||||
|
||||
// txCache maintains a cache of transactions.
|
||||
type txCache struct {
|
||||
mtx sync.Mutex
|
||||
size int
|
||||
map_ map[string]struct{}
|
||||
list *list.List // to remove oldest tx when cache gets too big
|
||||
}
|
||||
|
||||
// newTxCache returns a new txCache.
|
||||
func newTxCache(cacheSize int) *txCache {
|
||||
return &txCache{
|
||||
size: cacheSize,
|
||||
map_: make(map[string]struct{}, cacheSize),
|
||||
list: list.New(),
|
||||
}
|
||||
}
|
||||
|
||||
// Reset resets the txCache to empty.
|
||||
func (cache *txCache) Reset() {
|
||||
cache.mtx.Lock()
|
||||
cache.map_ = make(map[string]struct{}, cache.size)
|
||||
cache.list.Init()
|
||||
cache.mtx.Unlock()
|
||||
}
|
||||
|
||||
// Push adds the given tx to the txCache. It returns false if tx is already in the cache.
|
||||
func (cache *txCache) Push(tx types.Tx) bool {
|
||||
cache.mtx.Lock()
|
||||
defer cache.mtx.Unlock()
|
||||
|
||||
if _, exists := cache.map_[string(tx)]; exists {
|
||||
return false
|
||||
}
|
||||
|
||||
if cache.list.Len() >= cache.size {
|
||||
popped := cache.list.Front()
|
||||
poppedTx := popped.Value.(types.Tx)
|
||||
// NOTE: the tx may have already been removed from the map
|
||||
// but deleting a non-existent element is fine
|
||||
delete(cache.map_, string(poppedTx))
|
||||
cache.list.Remove(popped)
|
||||
}
|
||||
cache.map_[string(tx)] = struct{}{}
|
||||
cache.list.PushBack(tx)
|
||||
return true
|
||||
}
|
||||
|
||||
// Remove removes the given tx from the cache.
|
||||
func (cache *txCache) Remove(tx types.Tx) {
|
||||
cache.mtx.Lock()
|
||||
delete(cache.map_, string(tx))
|
||||
cache.mtx.Unlock()
|
||||
}
|
||||
@@ -0,0 +1,286 @@
|
||||
package mempool
|
||||
|
||||
import (
|
||||
"crypto/md5"
|
||||
"crypto/rand"
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/tendermint/abci/example/counter"
|
||||
"github.com/tendermint/abci/example/kvstore"
|
||||
abci "github.com/tendermint/abci/types"
|
||||
cmn "github.com/tendermint/tmlibs/common"
|
||||
"github.com/tendermint/tmlibs/log"
|
||||
|
||||
cfg "github.com/tendermint/tendermint/config"
|
||||
"github.com/tendermint/tendermint/proxy"
|
||||
"github.com/tendermint/tendermint/types"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func newMempoolWithApp(cc proxy.ClientCreator) *Mempool {
|
||||
config := cfg.ResetTestRoot("mempool_test")
|
||||
|
||||
appConnMem, _ := cc.NewABCIClient()
|
||||
appConnMem.SetLogger(log.TestingLogger().With("module", "abci-client", "connection", "mempool"))
|
||||
err := appConnMem.Start()
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
mempool := NewMempool(config.Mempool, appConnMem, 0)
|
||||
mempool.SetLogger(log.TestingLogger())
|
||||
return mempool
|
||||
}
|
||||
|
||||
func ensureNoFire(t *testing.T, ch <-chan int64, timeoutMS int) {
|
||||
timer := time.NewTimer(time.Duration(timeoutMS) * time.Millisecond)
|
||||
select {
|
||||
case <-ch:
|
||||
t.Fatal("Expected not to fire")
|
||||
case <-timer.C:
|
||||
}
|
||||
}
|
||||
|
||||
func ensureFire(t *testing.T, ch <-chan int64, timeoutMS int) {
|
||||
timer := time.NewTimer(time.Duration(timeoutMS) * time.Millisecond)
|
||||
select {
|
||||
case <-ch:
|
||||
case <-timer.C:
|
||||
t.Fatal("Expected to fire")
|
||||
}
|
||||
}
|
||||
|
||||
func checkTxs(t *testing.T, mempool *Mempool, count int) types.Txs {
|
||||
txs := make(types.Txs, count)
|
||||
for i := 0; i < count; i++ {
|
||||
txBytes := make([]byte, 20)
|
||||
txs[i] = txBytes
|
||||
_, err := rand.Read(txBytes)
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
if err := mempool.CheckTx(txBytes, nil); err != nil {
|
||||
t.Fatalf("Error after CheckTx: %v", err)
|
||||
}
|
||||
}
|
||||
return txs
|
||||
}
|
||||
|
||||
func TestTxsAvailable(t *testing.T) {
|
||||
app := kvstore.NewKVStoreApplication()
|
||||
cc := proxy.NewLocalClientCreator(app)
|
||||
mempool := newMempoolWithApp(cc)
|
||||
mempool.EnableTxsAvailable()
|
||||
|
||||
timeoutMS := 500
|
||||
|
||||
// with no txs, it shouldnt fire
|
||||
ensureNoFire(t, mempool.TxsAvailable(), timeoutMS)
|
||||
|
||||
// send a bunch of txs, it should only fire once
|
||||
txs := checkTxs(t, mempool, 100)
|
||||
ensureFire(t, mempool.TxsAvailable(), timeoutMS)
|
||||
ensureNoFire(t, mempool.TxsAvailable(), timeoutMS)
|
||||
|
||||
// call update with half the txs.
|
||||
// it should fire once now for the new height
|
||||
// since there are still txs left
|
||||
committedTxs, txs := txs[:50], txs[50:]
|
||||
if err := mempool.Update(1, committedTxs); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
ensureFire(t, mempool.TxsAvailable(), timeoutMS)
|
||||
ensureNoFire(t, mempool.TxsAvailable(), timeoutMS)
|
||||
|
||||
// send a bunch more txs. we already fired for this height so it shouldnt fire again
|
||||
moreTxs := checkTxs(t, mempool, 50)
|
||||
ensureNoFire(t, mempool.TxsAvailable(), timeoutMS)
|
||||
|
||||
// now call update with all the txs. it should not fire as there are no txs left
|
||||
committedTxs = append(txs, moreTxs...)
|
||||
if err := mempool.Update(2, committedTxs); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
ensureNoFire(t, mempool.TxsAvailable(), timeoutMS)
|
||||
|
||||
// send a bunch more txs, it should only fire once
|
||||
checkTxs(t, mempool, 100)
|
||||
ensureFire(t, mempool.TxsAvailable(), timeoutMS)
|
||||
ensureNoFire(t, mempool.TxsAvailable(), timeoutMS)
|
||||
}
|
||||
|
||||
func TestSerialReap(t *testing.T) {
|
||||
app := counter.NewCounterApplication(true)
|
||||
app.SetOption(abci.RequestSetOption{"serial", "on"})
|
||||
cc := proxy.NewLocalClientCreator(app)
|
||||
|
||||
mempool := newMempoolWithApp(cc)
|
||||
appConnCon, _ := cc.NewABCIClient()
|
||||
appConnCon.SetLogger(log.TestingLogger().With("module", "abci-client", "connection", "consensus"))
|
||||
err := appConnCon.Start()
|
||||
require.Nil(t, err)
|
||||
|
||||
cacheMap := make(map[string]struct{})
|
||||
deliverTxsRange := func(start, end int) {
|
||||
// Deliver some txs.
|
||||
for i := start; i < end; i++ {
|
||||
|
||||
// This will succeed
|
||||
txBytes := make([]byte, 8)
|
||||
binary.BigEndian.PutUint64(txBytes, uint64(i))
|
||||
err := mempool.CheckTx(txBytes, nil)
|
||||
_, cached := cacheMap[string(txBytes)]
|
||||
if cached {
|
||||
require.NotNil(t, err, "expected error for cached tx")
|
||||
} else {
|
||||
require.Nil(t, err, "expected no err for uncached tx")
|
||||
}
|
||||
cacheMap[string(txBytes)] = struct{}{}
|
||||
|
||||
// Duplicates are cached and should return error
|
||||
err = mempool.CheckTx(txBytes, nil)
|
||||
require.NotNil(t, err, "Expected error after CheckTx on duplicated tx")
|
||||
}
|
||||
}
|
||||
|
||||
reapCheck := func(exp int) {
|
||||
txs := mempool.Reap(-1)
|
||||
require.Equal(t, len(txs), exp, cmn.Fmt("Expected to reap %v txs but got %v", exp, len(txs)))
|
||||
}
|
||||
|
||||
updateRange := func(start, end int) {
|
||||
txs := make([]types.Tx, 0)
|
||||
for i := start; i < end; i++ {
|
||||
txBytes := make([]byte, 8)
|
||||
binary.BigEndian.PutUint64(txBytes, uint64(i))
|
||||
txs = append(txs, txBytes)
|
||||
}
|
||||
if err := mempool.Update(0, txs); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
}
|
||||
|
||||
commitRange := func(start, end int) {
|
||||
// Deliver some txs.
|
||||
for i := start; i < end; i++ {
|
||||
txBytes := make([]byte, 8)
|
||||
binary.BigEndian.PutUint64(txBytes, uint64(i))
|
||||
res, err := appConnCon.DeliverTxSync(txBytes)
|
||||
if err != nil {
|
||||
t.Errorf("Client error committing tx: %v", err)
|
||||
}
|
||||
if res.IsErr() {
|
||||
t.Errorf("Error committing tx. Code:%v result:%X log:%v",
|
||||
res.Code, res.Data, res.Log)
|
||||
}
|
||||
}
|
||||
res, err := appConnCon.CommitSync()
|
||||
if err != nil {
|
||||
t.Errorf("Client error committing: %v", err)
|
||||
}
|
||||
if len(res.Data) != 8 {
|
||||
t.Errorf("Error committing. Hash:%X", res.Data)
|
||||
}
|
||||
}
|
||||
|
||||
//----------------------------------------
|
||||
|
||||
// Deliver some txs.
|
||||
deliverTxsRange(0, 100)
|
||||
|
||||
// Reap the txs.
|
||||
reapCheck(100)
|
||||
|
||||
// Reap again. We should get the same amount
|
||||
reapCheck(100)
|
||||
|
||||
// Deliver 0 to 999, we should reap 900 new txs
|
||||
// because 100 were already counted.
|
||||
deliverTxsRange(0, 1000)
|
||||
|
||||
// Reap the txs.
|
||||
reapCheck(1000)
|
||||
|
||||
// Reap again. We should get the same amount
|
||||
reapCheck(1000)
|
||||
|
||||
// Commit from the conensus AppConn
|
||||
commitRange(0, 500)
|
||||
updateRange(0, 500)
|
||||
|
||||
// We should have 500 left.
|
||||
reapCheck(500)
|
||||
|
||||
// Deliver 100 invalid txs and 100 valid txs
|
||||
deliverTxsRange(900, 1100)
|
||||
|
||||
// We should have 600 now.
|
||||
reapCheck(600)
|
||||
}
|
||||
|
||||
func TestMempoolCloseWAL(t *testing.T) {
|
||||
// 1. Create the temporary directory for mempool and WAL testing.
|
||||
rootDir, err := ioutil.TempDir("", "mempool-test")
|
||||
require.Nil(t, err, "expecting successful tmpdir creation")
|
||||
defer os.RemoveAll(rootDir)
|
||||
|
||||
// 2. Ensure that it doesn't contain any elements -- Sanity check
|
||||
m1, err := filepath.Glob(filepath.Join(rootDir, "*"))
|
||||
require.Nil(t, err, "successful globbing expected")
|
||||
require.Equal(t, 0, len(m1), "no matches yet")
|
||||
|
||||
// 3. Create the mempool
|
||||
wcfg := cfg.DefaultMempoolConfig()
|
||||
wcfg.RootDir = rootDir
|
||||
app := kvstore.NewKVStoreApplication()
|
||||
cc := proxy.NewLocalClientCreator(app)
|
||||
appConnMem, _ := cc.NewABCIClient()
|
||||
mempool := NewMempool(wcfg, appConnMem, 10)
|
||||
mempool.InitWAL()
|
||||
|
||||
// 4. Ensure that the directory contains the WAL file
|
||||
m2, err := filepath.Glob(filepath.Join(rootDir, "*"))
|
||||
require.Nil(t, err, "successful globbing expected")
|
||||
require.Equal(t, 1, len(m2), "expecting the wal match in")
|
||||
|
||||
// 5. Write some contents to the WAL
|
||||
mempool.CheckTx(types.Tx([]byte("foo")), nil)
|
||||
walFilepath := mempool.wal.Path
|
||||
sum1 := checksumFile(walFilepath, t)
|
||||
|
||||
// 6. Sanity check to ensure that the written TX matches the expectation.
|
||||
require.Equal(t, sum1, checksumIt([]byte("foo\n")), "foo with a newline should be written")
|
||||
|
||||
// 7. Invoke CloseWAL() and ensure it discards the
|
||||
// WAL thus any other write won't go through.
|
||||
require.True(t, mempool.CloseWAL(), "CloseWAL should CloseWAL")
|
||||
mempool.CheckTx(types.Tx([]byte("bar")), nil)
|
||||
sum2 := checksumFile(walFilepath, t)
|
||||
require.Equal(t, sum1, sum2, "expected no change to the WAL after invoking CloseWAL() since it was discarded")
|
||||
|
||||
// 8. Second CloseWAL should do nothing
|
||||
require.False(t, mempool.CloseWAL(), "CloseWAL should CloseWAL")
|
||||
|
||||
// 9. Sanity check to ensure that the WAL file still exists
|
||||
m3, err := filepath.Glob(filepath.Join(rootDir, "*"))
|
||||
require.Nil(t, err, "successful globbing expected")
|
||||
require.Equal(t, 1, len(m3), "expecting the wal match in")
|
||||
}
|
||||
|
||||
func checksumIt(data []byte) string {
|
||||
h := md5.New()
|
||||
h.Write(data)
|
||||
return fmt.Sprintf("%x", h.Sum(nil))
|
||||
}
|
||||
|
||||
func checksumFile(p string, t *testing.T) string {
|
||||
data, err := ioutil.ReadFile(p)
|
||||
require.Nil(t, err, "expecting successful read of %q", p)
|
||||
return checksumIt(data)
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
package mempool
|
||||
|
||||
import (
|
||||
"github.com/go-kit/kit/metrics"
|
||||
"github.com/go-kit/kit/metrics/discard"
|
||||
|
||||
prometheus "github.com/go-kit/kit/metrics/prometheus"
|
||||
stdprometheus "github.com/prometheus/client_golang/prometheus"
|
||||
)
|
||||
|
||||
// Metrics contains metrics exposed by this package.
|
||||
// see MetricsProvider for descriptions.
|
||||
type Metrics struct {
|
||||
// Size of the mempool.
|
||||
Size metrics.Gauge
|
||||
}
|
||||
|
||||
// PrometheusMetrics returns Metrics build using Prometheus client library.
|
||||
func PrometheusMetrics() *Metrics {
|
||||
return &Metrics{
|
||||
Size: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
|
||||
Subsystem: "mempool",
|
||||
Name: "size",
|
||||
Help: "Size of the mempool (number of uncommitted transactions).",
|
||||
}, []string{}),
|
||||
}
|
||||
}
|
||||
|
||||
// NopMetrics returns no-op Metrics.
|
||||
func NopMetrics() *Metrics {
|
||||
return &Metrics{
|
||||
Size: discard.NewGauge(),
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,188 @@
|
||||
package mempool
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"reflect"
|
||||
"time"
|
||||
|
||||
abci "github.com/tendermint/abci/types"
|
||||
"github.com/tendermint/go-amino"
|
||||
"github.com/tendermint/tmlibs/clist"
|
||||
"github.com/tendermint/tmlibs/log"
|
||||
|
||||
cfg "github.com/tendermint/tendermint/config"
|
||||
"github.com/tendermint/tendermint/p2p"
|
||||
"github.com/tendermint/tendermint/types"
|
||||
)
|
||||
|
||||
const (
|
||||
MempoolChannel = byte(0x30)
|
||||
|
||||
maxMsgSize = 1048576 // 1MB TODO make it configurable
|
||||
peerCatchupSleepIntervalMS = 100 // If peer is behind, sleep this amount
|
||||
)
|
||||
|
||||
// MempoolReactor handles mempool tx broadcasting amongst peers.
|
||||
type MempoolReactor struct {
|
||||
p2p.BaseReactor
|
||||
config *cfg.MempoolConfig
|
||||
Mempool *Mempool
|
||||
}
|
||||
|
||||
// NewMempoolReactor returns a new MempoolReactor with the given config and mempool.
|
||||
func NewMempoolReactor(config *cfg.MempoolConfig, mempool *Mempool) *MempoolReactor {
|
||||
memR := &MempoolReactor{
|
||||
config: config,
|
||||
Mempool: mempool,
|
||||
}
|
||||
memR.BaseReactor = *p2p.NewBaseReactor("MempoolReactor", memR)
|
||||
return memR
|
||||
}
|
||||
|
||||
// SetLogger sets the Logger on the reactor and the underlying Mempool.
|
||||
func (memR *MempoolReactor) SetLogger(l log.Logger) {
|
||||
memR.Logger = l
|
||||
memR.Mempool.SetLogger(l)
|
||||
}
|
||||
|
||||
// GetChannels implements Reactor.
|
||||
// It returns the list of channels for this reactor.
|
||||
func (memR *MempoolReactor) GetChannels() []*p2p.ChannelDescriptor {
|
||||
return []*p2p.ChannelDescriptor{
|
||||
{
|
||||
ID: MempoolChannel,
|
||||
Priority: 5,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// AddPeer implements Reactor.
|
||||
// It starts a broadcast routine ensuring all txs are forwarded to the given peer.
|
||||
func (memR *MempoolReactor) AddPeer(peer p2p.Peer) {
|
||||
go memR.broadcastTxRoutine(peer)
|
||||
}
|
||||
|
||||
// RemovePeer implements Reactor.
|
||||
func (memR *MempoolReactor) RemovePeer(peer p2p.Peer, reason interface{}) {
|
||||
// broadcast routine checks if peer is gone and returns
|
||||
}
|
||||
|
||||
// Receive implements Reactor.
|
||||
// It adds any received transactions to the mempool.
|
||||
func (memR *MempoolReactor) Receive(chID byte, src p2p.Peer, msgBytes []byte) {
|
||||
msg, err := DecodeMessage(msgBytes)
|
||||
if err != nil {
|
||||
memR.Logger.Error("Error decoding message", "src", src, "chId", chID, "msg", msg, "err", err, "bytes", msgBytes)
|
||||
memR.Switch.StopPeerForError(src, err)
|
||||
return
|
||||
}
|
||||
memR.Logger.Debug("Receive", "src", src, "chId", chID, "msg", msg)
|
||||
|
||||
switch msg := msg.(type) {
|
||||
case *TxMessage:
|
||||
err := memR.Mempool.CheckTx(msg.Tx, nil)
|
||||
if err != nil {
|
||||
memR.Logger.Info("Could not check tx", "tx", msg.Tx, "err", err)
|
||||
}
|
||||
// broadcasting happens from go routines per peer
|
||||
default:
|
||||
memR.Logger.Error(fmt.Sprintf("Unknown message type %v", reflect.TypeOf(msg)))
|
||||
}
|
||||
}
|
||||
|
||||
// BroadcastTx is an alias for Mempool.CheckTx. Broadcasting itself happens in peer routines.
|
||||
func (memR *MempoolReactor) BroadcastTx(tx types.Tx, cb func(*abci.Response)) error {
|
||||
return memR.Mempool.CheckTx(tx, cb)
|
||||
}
|
||||
|
||||
// PeerState describes the state of a peer.
|
||||
type PeerState interface {
|
||||
GetHeight() int64
|
||||
}
|
||||
|
||||
// Send new mempool txs to peer.
|
||||
func (memR *MempoolReactor) broadcastTxRoutine(peer p2p.Peer) {
|
||||
if !memR.config.Broadcast {
|
||||
return
|
||||
}
|
||||
|
||||
var next *clist.CElement
|
||||
for {
|
||||
// This happens because the CElement we were looking at got garbage
|
||||
// collected (removed). That is, .NextWait() returned nil. Go ahead and
|
||||
// start from the beginning.
|
||||
if next == nil {
|
||||
select {
|
||||
case <-memR.Mempool.TxsWaitChan(): // Wait until a tx is available
|
||||
if next = memR.Mempool.TxsFront(); next == nil {
|
||||
continue
|
||||
}
|
||||
case <-peer.Quit():
|
||||
return
|
||||
case <-memR.Quit():
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
memTx := next.Value.(*mempoolTx)
|
||||
// make sure the peer is up to date
|
||||
height := memTx.Height()
|
||||
if peerState_i := peer.Get(types.PeerStateKey); peerState_i != nil {
|
||||
peerState := peerState_i.(PeerState)
|
||||
if peerState.GetHeight() < height-1 { // Allow for a lag of 1 block
|
||||
time.Sleep(peerCatchupSleepIntervalMS * time.Millisecond)
|
||||
continue
|
||||
}
|
||||
}
|
||||
// send memTx
|
||||
msg := &TxMessage{Tx: memTx.tx}
|
||||
success := peer.Send(MempoolChannel, cdc.MustMarshalBinaryBare(msg))
|
||||
if !success {
|
||||
time.Sleep(peerCatchupSleepIntervalMS * time.Millisecond)
|
||||
continue
|
||||
}
|
||||
|
||||
select {
|
||||
case <-next.NextWaitChan():
|
||||
// see the start of the for loop for nil check
|
||||
next = next.Next()
|
||||
case <-peer.Quit():
|
||||
return
|
||||
case <-memR.Quit():
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
//-----------------------------------------------------------------------------
|
||||
// Messages
|
||||
|
||||
// MempoolMessage is a message sent or received by the MempoolReactor.
|
||||
type MempoolMessage interface{}
|
||||
|
||||
func RegisterMempoolMessages(cdc *amino.Codec) {
|
||||
cdc.RegisterInterface((*MempoolMessage)(nil), nil)
|
||||
cdc.RegisterConcrete(&TxMessage{}, "tendermint/mempool/TxMessage", nil)
|
||||
}
|
||||
|
||||
// DecodeMessage decodes a byte-array into a MempoolMessage.
|
||||
func DecodeMessage(bz []byte) (msg MempoolMessage, err error) {
|
||||
if len(bz) > maxMsgSize {
|
||||
return msg, fmt.Errorf("Msg exceeds max size (%d > %d)",
|
||||
len(bz), maxMsgSize)
|
||||
}
|
||||
err = cdc.UnmarshalBinaryBare(bz, &msg)
|
||||
return
|
||||
}
|
||||
|
||||
//-------------------------------------
|
||||
|
||||
// TxMessage is a MempoolMessage containing a transaction.
|
||||
type TxMessage struct {
|
||||
Tx types.Tx
|
||||
}
|
||||
|
||||
// String returns a string representation of the TxMessage.
|
||||
func (m *TxMessage) String() string {
|
||||
return fmt.Sprintf("[TxMessage %v]", m.Tx)
|
||||
}
|
||||
@@ -0,0 +1,157 @@
|
||||
package mempool
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/fortytw2/leaktest"
|
||||
"github.com/pkg/errors"
|
||||
"github.com/stretchr/testify/assert"
|
||||
|
||||
"github.com/go-kit/kit/log/term"
|
||||
|
||||
"github.com/tendermint/abci/example/kvstore"
|
||||
"github.com/tendermint/tmlibs/log"
|
||||
|
||||
cfg "github.com/tendermint/tendermint/config"
|
||||
"github.com/tendermint/tendermint/p2p"
|
||||
"github.com/tendermint/tendermint/proxy"
|
||||
"github.com/tendermint/tendermint/types"
|
||||
)
|
||||
|
||||
// mempoolLogger is a TestingLogger which uses a different
|
||||
// color for each validator ("validator" key must exist).
|
||||
func mempoolLogger() log.Logger {
|
||||
return log.TestingLoggerWithColorFn(func(keyvals ...interface{}) term.FgBgColor {
|
||||
for i := 0; i < len(keyvals)-1; i += 2 {
|
||||
if keyvals[i] == "validator" {
|
||||
return term.FgBgColor{Fg: term.Color(uint8(keyvals[i+1].(int) + 1))}
|
||||
}
|
||||
}
|
||||
return term.FgBgColor{}
|
||||
})
|
||||
}
|
||||
|
||||
// connect N mempool reactors through N switches
|
||||
func makeAndConnectMempoolReactors(config *cfg.Config, N int) []*MempoolReactor {
|
||||
reactors := make([]*MempoolReactor, N)
|
||||
logger := mempoolLogger()
|
||||
for i := 0; i < N; i++ {
|
||||
app := kvstore.NewKVStoreApplication()
|
||||
cc := proxy.NewLocalClientCreator(app)
|
||||
mempool := newMempoolWithApp(cc)
|
||||
|
||||
reactors[i] = NewMempoolReactor(config.Mempool, mempool) // so we dont start the consensus states
|
||||
reactors[i].SetLogger(logger.With("validator", i))
|
||||
}
|
||||
|
||||
p2p.MakeConnectedSwitches(config.P2P, N, func(i int, s *p2p.Switch) *p2p.Switch {
|
||||
s.AddReactor("MEMPOOL", reactors[i])
|
||||
return s
|
||||
|
||||
}, p2p.Connect2Switches)
|
||||
return reactors
|
||||
}
|
||||
|
||||
// wait for all txs on all reactors
|
||||
func waitForTxs(t *testing.T, txs types.Txs, reactors []*MempoolReactor) {
|
||||
// wait for the txs in all mempools
|
||||
wg := new(sync.WaitGroup)
|
||||
for i := 0; i < len(reactors); i++ {
|
||||
wg.Add(1)
|
||||
go _waitForTxs(t, wg, txs, i, reactors)
|
||||
}
|
||||
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
wg.Wait()
|
||||
close(done)
|
||||
}()
|
||||
|
||||
timer := time.After(TIMEOUT)
|
||||
select {
|
||||
case <-timer:
|
||||
t.Fatal("Timed out waiting for txs")
|
||||
case <-done:
|
||||
}
|
||||
}
|
||||
|
||||
// wait for all txs on a single mempool
|
||||
func _waitForTxs(t *testing.T, wg *sync.WaitGroup, txs types.Txs, reactorIdx int, reactors []*MempoolReactor) {
|
||||
|
||||
mempool := reactors[reactorIdx].Mempool
|
||||
for mempool.Size() != len(txs) {
|
||||
time.Sleep(time.Millisecond * 100)
|
||||
}
|
||||
|
||||
reapedTxs := mempool.Reap(len(txs))
|
||||
for i, tx := range txs {
|
||||
assert.Equal(t, tx, reapedTxs[i], fmt.Sprintf("txs at index %d on reactor %d don't match: %v vs %v", i, reactorIdx, tx, reapedTxs[i]))
|
||||
}
|
||||
wg.Done()
|
||||
}
|
||||
|
||||
const (
|
||||
NUM_TXS = 1000
|
||||
TIMEOUT = 120 * time.Second // ridiculously high because CircleCI is slow
|
||||
)
|
||||
|
||||
func TestReactorBroadcastTxMessage(t *testing.T) {
|
||||
config := cfg.TestConfig()
|
||||
const N = 4
|
||||
reactors := makeAndConnectMempoolReactors(config, N)
|
||||
defer func() {
|
||||
for _, r := range reactors {
|
||||
r.Stop()
|
||||
}
|
||||
}()
|
||||
|
||||
// send a bunch of txs to the first reactor's mempool
|
||||
// and wait for them all to be received in the others
|
||||
txs := checkTxs(t, reactors[0].Mempool, NUM_TXS)
|
||||
waitForTxs(t, txs, reactors)
|
||||
}
|
||||
|
||||
func TestBroadcastTxForPeerStopsWhenPeerStops(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping test in short mode.")
|
||||
}
|
||||
|
||||
config := cfg.TestConfig()
|
||||
const N = 2
|
||||
reactors := makeAndConnectMempoolReactors(config, N)
|
||||
defer func() {
|
||||
for _, r := range reactors {
|
||||
r.Stop()
|
||||
}
|
||||
}()
|
||||
|
||||
// stop peer
|
||||
sw := reactors[1].Switch
|
||||
sw.StopPeerForError(sw.Peers().List()[0], errors.New("some reason"))
|
||||
|
||||
// check that we are not leaking any go-routines
|
||||
// i.e. broadcastTxRoutine finishes when peer is stopped
|
||||
leaktest.CheckTimeout(t, 10*time.Second)()
|
||||
}
|
||||
|
||||
func TestBroadcastTxForPeerStopsWhenReactorStops(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping test in short mode.")
|
||||
}
|
||||
|
||||
config := cfg.TestConfig()
|
||||
const N = 2
|
||||
reactors := makeAndConnectMempoolReactors(config, N)
|
||||
|
||||
// stop reactors
|
||||
for _, r := range reactors {
|
||||
r.Stop()
|
||||
}
|
||||
|
||||
// check that we are not leaking any go-routines
|
||||
// i.e. broadcastTxRoutine finishes when reactor is stopped
|
||||
leaktest.CheckTimeout(t, 10*time.Second)()
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
package mempool
|
||||
|
||||
import (
|
||||
"github.com/tendermint/go-amino"
|
||||
)
|
||||
|
||||
var cdc = amino.NewCodec()
|
||||
|
||||
func init() {
|
||||
RegisterMempoolMessages(cdc)
|
||||
}
|
||||
Reference in New Issue
Block a user