From 8dfa464f05cc713f03a35228491b96c892ca3b58 Mon Sep 17 00:00:00 2001 From: Anton Kaliaev Date: Mon, 4 Feb 2019 15:54:48 +0400 Subject: [PATCH] bound mempool memory usage Closes #3079 --- CHANGELOG_PENDING.md | 1 + config/config.go | 6 ++++++ config/toml.go | 3 +++ docs/tendermint-core/configuration.md | 3 +++ mempool/mempool.go | 17 +++++++++++++++-- mempool/mempool_test.go | 22 ++++++++++++++++++++++ 6 files changed, 50 insertions(+), 2 deletions(-) diff --git a/CHANGELOG_PENDING.md b/CHANGELOG_PENDING.md index daa42654c..e5cf12279 100644 --- a/CHANGELOG_PENDING.md +++ b/CHANGELOG_PENDING.md @@ -18,6 +18,7 @@ Special thanks to external contributors on this release: * P2P Protocol ### FEATURES: +- [mempool] \#3079 bound mempool memory usage (`mempool.max_bytes` is set to 1GB by default; see config.toml) ### IMPROVEMENTS: - [tools] add go-deadlock tool to help detect deadlocks diff --git a/config/config.go b/config/config.go index fc50d7c61..81a14fadc 100644 --- a/config/config.go +++ b/config/config.go @@ -536,6 +536,8 @@ type MempoolConfig struct { WalPath string `mapstructure:"wal_dir"` Size int `mapstructure:"size"` CacheSize int `mapstructure:"cache_size"` + // Maximum size of all txs in the mempool in bytes + MaxBytes int `mapstructure:"max_bytes"` } // DefaultMempoolConfig returns a default configuration for the Tendermint mempool @@ -548,6 +550,7 @@ func DefaultMempoolConfig() *MempoolConfig { // ABCI Recheck Size: 5000, CacheSize: 10000, + MaxBytes: 1000000000, // 1GB } } @@ -577,6 +580,9 @@ func (cfg *MempoolConfig) ValidateBasic() error { if cfg.CacheSize < 0 { return errors.New("cache_size can't be negative") } + if cfg.MaxBytes <= 0 { + return errors.New("max_bytes must be a positive number") + } return nil } diff --git a/config/toml.go b/config/toml.go index e842e9e37..f28eb1275 100644 --- a/config/toml.go +++ b/config/toml.go @@ -240,6 +240,9 @@ size = {{ .Mempool.Size }} # size of the cache (used to filter transactions we saw earlier) cache_size = {{ .Mempool.CacheSize }} +# maximum size of all txs in the mempool in bytes +max_bytes = {{ .Mempool.MaxBytes }} + ##### consensus configuration options ##### [consensus] diff --git a/docs/tendermint-core/configuration.md b/docs/tendermint-core/configuration.md index 0d9a58c4b..ffdafb728 100644 --- a/docs/tendermint-core/configuration.md +++ b/docs/tendermint-core/configuration.md @@ -189,6 +189,9 @@ size = 5000 # size of the cache (used to filter transactions we saw earlier) cache_size = 10000 +# maximum size of all txs in the mempool in bytes +max_bytes = 1000000000 + ##### consensus configuration options ##### [consensus] diff --git a/mempool/mempool.go b/mempool/mempool.go index 8550f2f81..4c8999c98 100644 --- a/mempool/mempool.go +++ b/mempool/mempool.go @@ -147,6 +147,8 @@ type Mempool struct { preCheck PreCheckFunc postCheck PostCheckFunc + sizeBytes int64 // size of all txs in the mempool in bytes + // Keep a cache of already-seen txs. // This reduces the pressure on the proxyApp. cache txCache @@ -265,8 +267,13 @@ func (mem *Mempool) Size() int { return mem.txs.Len() } -// Flushes the mempool connection to ensure async resCb calls are done e.g. -// from CheckTx. +// SizeBytes returns the size of all txs in the mempool in bytes. +func (mem *Mempool) SizeBytes() int64 { + return atomic.LoadInt64(&mem.sizeBytes) +} + +// FlushAppConn flushes the mempool connection to ensure async resCb calls are +// done e.g. from CheckTx. func (mem *Mempool) FlushAppConn() error { return mem.proxyAppConn.FlushSync() } @@ -280,6 +287,7 @@ func (mem *Mempool) Flush() { for e := mem.txs.Front(); e != nil; e = e.Next() { mem.txs.Remove(e) + _ = atomic.SwapInt64(&mem.sizeBytes, 0) e.DetachPrev() } } @@ -310,6 +318,8 @@ func (mem *Mempool) CheckTx(tx types.Tx, cb func(*abci.Response)) (err error) { if mem.Size() >= mem.config.Size { return ErrMempoolIsFull + } else if int64(len(tx))+mem.SizeBytes() > int64(mem.config.MaxBytes) { + return ErrMempoolIsFull } // The size of the corresponding amino-encoded TxMessage @@ -383,6 +393,7 @@ func (mem *Mempool) resCbNormal(req *abci.Request, res *abci.Response) { tx: tx, } mem.txs.PushBack(memTx) + atomic.AddInt64(&mem.sizeBytes, int64(len(tx))) mem.logger.Info("Added good transaction", "tx", TxID(tx), "res", r, @@ -424,6 +435,7 @@ func (mem *Mempool) resCbRecheck(req *abci.Request, res *abci.Response) { // Tx became invalidated due to newly committed block. mem.logger.Info("Tx is no longer valid", "tx", TxID(tx), "res", r, "err", postCheckErr) mem.txs.Remove(mem.recheckCursor) + atomic.AddInt64(&mem.sizeBytes, int64(-len(tx))) mem.recheckCursor.DetachPrev() // remove from cache (it might be good later) @@ -597,6 +609,7 @@ func (mem *Mempool) removeTxs(txs types.Txs) []types.Tx { if _, ok := txsMap[string(memTx.tx)]; ok { // remove from clist mem.txs.Remove(e) + atomic.AddInt64(&mem.sizeBytes, int64(-len(memTx.tx))) e.DetachPrev() // NOTE: we don't remove committed txs from the cache. diff --git a/mempool/mempool_test.go b/mempool/mempool_test.go index 9d21e734f..6c40df3a7 100644 --- a/mempool/mempool_test.go +++ b/mempool/mempool_test.go @@ -450,6 +450,28 @@ func TestMempoolMaxMsgSize(t *testing.T) { } +func TestMempoolSizeBytes(t *testing.T) { + app := kvstore.NewKVStoreApplication() + cc := proxy.NewLocalClientCreator(app) + mempool := newMempoolWithApp(cc) + + assert.EqualValues(t, 0, mempool.SizeBytes()) + + err := mempool.CheckTx([]byte{0x01}, nil) + require.NoError(t, err) + assert.EqualValues(t, 1, mempool.SizeBytes()) + + mempool.Update(1, []types.Tx{[]byte{0x01}}, nil, nil) + assert.EqualValues(t, 0, mempool.SizeBytes()) + + err = mempool.CheckTx([]byte{0x02, 0x03}, nil) + require.NoError(t, err) + assert.EqualValues(t, 2, mempool.SizeBytes()) + + mempool.Flush() + assert.EqualValues(t, 0, mempool.SizeBytes()) +} + func checksumIt(data []byte) string { h := md5.New() h.Write(data)