bound the mempool

Refs #345

config/config.go

@@ -335,6 +335,7 @@ type MempoolConfig struct {
 	RecheckEmpty bool   `mapstructure:"recheck_empty"`
 	Broadcast    bool   `mapstructure:"broadcast"`
 	WalPath      string `mapstructure:"wal_dir"`
+	Size         int    `mapstructure:"size"`
 	CacheSize    int    `mapstructure:"cache_size"`
 }
 
@@ -345,6 +346,7 @@ func DefaultMempoolConfig() *MempoolConfig {
 		RecheckEmpty: true,
 		Broadcast:    true,
 		WalPath:      filepath.Join(defaultDataDir, "mempool.wal"),
+		Size:         100000,
 		CacheSize:    100000,
 	}
 }

config/toml.go

@@ -179,6 +179,12 @@ recheck_empty = {{ .Mempool.RecheckEmpty }}
 broadcast = {{ .Mempool.Broadcast }}
 wal_dir = "{{ .Mempool.WalPath }}"
 
+# size of the mempool
+size = {{ .Mempool.Size }}
+
+# size of the cache (used to filter transactions we saw earlier)
+cache_size = {{ .Mempool.CacheSize }}
+
 ##### consensus configuration options #####
 [consensus]
 

docs/specification/configuration.rst

@@ -136,6 +136,12 @@ like the file below, however, double check by inspecting the
     broadcast = true
     wal_dir = "data/mempool.wal"
 
+    # size of the mempool
+    size = 100000
+
+    # size of the cache (used to filter transactions we saw earlier)
+    cache_size = 100000
+
     ##### consensus configuration options #####
     [consensus]
 

mempool/mempool.go

@@ -49,7 +49,13 @@ TODO: Better handle abci client errors. (make it automatically handle connection
 
 */
 
-var ErrTxInCache = errors.New("Tx already exists in cache")
+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
@@ -80,7 +86,6 @@ type Mempool struct {
 }
 
 // NewMempool returns a new Mempool with the given configuration and connection to an application.
-// TODO: Extract logger into arguments.
 func NewMempool(config *cfg.MempoolConfig, proxyAppConn proxy.AppConnMempool, height int64) *Mempool {
 	mempool := &Mempool{
 		config:        config,
@@ -202,6 +207,10 @@ 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