Create pool with timeout handler

Currently, the request pool has a SetTimeoutHandler method
which is never called in batcher tests, which leads
to null pointer panic when running with -count 5.

This is a by-product of the fact that one can create a pool without
a timeout handler.

I removed the said method and expanded the constructor so that
you have to pass in a timeout handler.

Signed-off-by: yacovm <yacovm@il.ibm.com>

Author: yacovm

internal/bft/batcher_test.go

@@ -11,11 +11,21 @@ import (
 	"time"
 
 	"github.com/SmartBFT-Go/consensus/internal/bft"
+	"github.com/SmartBFT-Go/consensus/internal/bft/mocks"
 	"github.com/SmartBFT-Go/consensus/pkg/types"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 	"go.uber.org/zap"
 )
 
+var (
+	noopTimeoutHandler = &mocks.RequestTimeoutHandler{}
+)
+
+func init() {
+	noopTimeoutHandler.On("OnRequestTimeout", mock.Anything, mock.Anything)
+}
+
 func TestBatcherBasic(t *testing.T) {
 	basicLog, err := zap.NewDevelopment()
 	assert.NoError(t, err)
@@ -25,7 +35,7 @@ func TestBatcherBasic(t *testing.T) {
 	byteReq1 := makeTestRequest("1", "1", "foo")
 	byteReq2 := makeTestRequest("2", "2", "foo")
 	byteReq3 := makeTestRequest("3", "3", "foo")
-	pool := bft.NewPool(log, insp, bft.PoolOptions{QueueSize: 3})
+	pool := bft.NewPool(log, insp, noopTimeoutHandler, bft.PoolOptions{QueueSize: 3})
 	err = pool.Submit(byteReq1)
 	assert.NoError(t, err)
 
@@ -95,7 +105,7 @@ func TestBatcherWhileSubmitting(t *testing.T) {
 	assert.NoError(t, err)
 	log := basicLog.Sugar()
 	insp := &testRequestInspector{}
-	pool := bft.NewPool(log, insp, bft.PoolOptions{QueueSize: 200})
+	pool := bft.NewPool(log, insp, noopTimeoutHandler, bft.PoolOptions{QueueSize: 200})
 
 	batcher := bft.NewBatchBuilder(pool, 100, 100*time.Second) // long time
 
@@ -136,7 +146,7 @@ func TestBatcherClose(t *testing.T) {
 	insp := &testRequestInspector{}
 
 	byteReq := makeTestRequest("1", "1", "foo")
-	pool := bft.NewPool(log, insp, bft.PoolOptions{QueueSize: 3})
+	pool := bft.NewPool(log, insp, noopTimeoutHandler, bft.PoolOptions{QueueSize: 3})
 	err = pool.Submit(byteReq)
 	assert.NoError(t, err)
 
@@ -170,7 +180,7 @@ func TestBatcherReset(t *testing.T) {
 	insp := &testRequestInspector{}
 
 	byteReq1 := makeTestRequest("1", "1", "foo")
-	pool := bft.NewPool(log, insp, bft.PoolOptions{QueueSize: 3})
+	pool := bft.NewPool(log, insp, noopTimeoutHandler, bft.PoolOptions{QueueSize: 3})
 	err = pool.Submit(byteReq1)
 	assert.NoError(t, err)
 

internal/bft/requestpool.go

@@ -19,7 +19,7 @@ import (
 )
 
 const (
-	DefaultRequestTimeout = 60000 * time.Millisecond
+	DefaultRequestTimeout = 10 * time.Second
 )
 
 //go:generate mockery -dir . -name RequestTimeoutHandler -case underscore -output ./mocks/
@@ -68,7 +68,7 @@ type PoolOptions struct {
 }
 
 // NewPool constructs new requests pool
-func NewPool(log api.Logger, inspector api.RequestInspector, options PoolOptions) *Pool {
+func NewPool(log api.Logger, inspector api.RequestInspector, th RequestTimeoutHandler, options PoolOptions) *Pool {
 	if options.RequestTimeout == 0 {
 		options.RequestTimeout = DefaultRequestTimeout
 	}
@@ -80,19 +80,16 @@ func NewPool(log api.Logger, inspector api.RequestInspector, options PoolOptions
 	}
 
 	return &Pool{
-		logger:    log,
-		inspector: inspector,
-		fifo:      list.New(),
-		semaphore: semaphore.NewWeighted(options.QueueSize),
-		existMap:  make(map[types.RequestInfo]*list.Element),
-		options:   options,
+		timeoutHandler: th,
+		logger:         log,
+		inspector:      inspector,
+		fifo:           list.New(),
+		semaphore:      semaphore.NewWeighted(options.QueueSize),
+		existMap:       make(map[types.RequestInfo]*list.Element),
+		options:        options,
 	}
 }
 
-func (rp *Pool) SetTimeoutHandler(handler RequestTimeoutHandler) {
-	rp.timeoutHandler = handler
-}
-
 func (rp *Pool) isStopped() bool {
 	rp.lock.Lock()
 	defer rp.lock.Unlock()

internal/bft/requestpool_test.go

@@ -36,8 +36,7 @@ func TestReqPoolBasic(t *testing.T) {
 	t.Run("create close", func(t *testing.T) {
 		timeoutHandler := &mocks.RequestTimeoutHandler{}
 
-		pool := bft.NewPool(log, insp, bft.PoolOptions{QueueSize: 3, RequestTimeout: time.Hour})
-		pool.SetTimeoutHandler(timeoutHandler)
+		pool := bft.NewPool(log, insp, timeoutHandler, bft.PoolOptions{QueueSize: 3, RequestTimeout: time.Hour})
 
 		assert.Equal(t, 0, pool.Size())
 		err = pool.Submit(byteReq1)
@@ -53,9 +52,8 @@ func TestReqPoolBasic(t *testing.T) {
 	t.Run("submit remove next", func(t *testing.T) {
 		timeoutHandler := &mocks.RequestTimeoutHandler{}
 
-		pool := bft.NewPool(log, insp, bft.PoolOptions{QueueSize: 3, RequestTimeout: time.Hour})
+		pool := bft.NewPool(log, insp, timeoutHandler, bft.PoolOptions{QueueSize: 3, RequestTimeout: time.Hour})
 		defer pool.Close()
-		pool.SetTimeoutHandler(timeoutHandler)
 
 		assert.Equal(t, 0, pool.Size())
 		err = pool.Submit(byteReq1)
@@ -152,9 +150,8 @@ func TestReqPoolCapacity(t *testing.T) {
 
 	t.Run("eventually submit", func(t *testing.T) {
 		timeoutHandler := &mocks.RequestTimeoutHandler{}
-		pool := bft.NewPool(log, insp, bft.PoolOptions{QueueSize: 50, RequestTimeout: time.Hour})
+		pool := bft.NewPool(log, insp, timeoutHandler, bft.PoolOptions{QueueSize: 50, RequestTimeout: time.Hour})
 		defer pool.Close()
-		pool.SetTimeoutHandler(timeoutHandler)
 
 		wg := sync.WaitGroup{}
 		wg.Add(2 * numReq)
@@ -188,9 +185,8 @@ func TestReqPoolCapacity(t *testing.T) {
 
 	t.Run("submit storm", func(t *testing.T) {
 		timeoutHandler := &mocks.RequestTimeoutHandler{}
-		pool := bft.NewPool(log, insp, bft.PoolOptions{QueueSize: 50, RequestTimeout: time.Hour})
+		pool := bft.NewPool(log, insp, timeoutHandler, bft.PoolOptions{QueueSize: 50, RequestTimeout: time.Hour})
 		defer pool.Close()
-		pool.SetTimeoutHandler(timeoutHandler)
 
 		wg := sync.WaitGroup{}
 		wg.Add(2 * numReq)
@@ -237,9 +233,8 @@ func TestReqPoolPrune(t *testing.T) {
 
 	byteReq1 := makeTestRequest("1", "1", "foo")
 	byteReq2 := makeTestRequest("2", "2", "bar")
-	pool := bft.NewPool(log, insp, bft.PoolOptions{QueueSize: 3, RequestTimeout: time.Hour})
+	pool := bft.NewPool(log, insp, timeoutHandler, bft.PoolOptions{QueueSize: 3, RequestTimeout: time.Hour})
 	defer pool.Close()
-	pool.SetTimeoutHandler(timeoutHandler)
 
 	assert.Equal(t, 0, pool.Size())
 
@@ -291,7 +286,7 @@ func TestReqPoolTimeout(t *testing.T) {
 			assert.Fail(t, "called OnAutoRemoveTimeout")
 		}).Return()
 
-		pool := bft.NewPool(log, insp,
+		pool := bft.NewPool(log, insp, timeoutHandler,
 			bft.PoolOptions{
 				QueueSize:         3,
 				RequestTimeout:    10 * time.Millisecond,
@@ -300,7 +295,6 @@ func TestReqPoolTimeout(t *testing.T) {
 			},
 		)
 		defer pool.Close()
-		pool.SetTimeoutHandler(timeoutHandler)
 
 		assert.Equal(t, 0, pool.Size())
 		err = pool.Submit(byteReq1)
@@ -336,7 +330,7 @@ func TestReqPoolTimeout(t *testing.T) {
 			assert.Fail(t, "called OnAutoRemoveTimeout")
 		}).Return()
 
-		pool := bft.NewPool(log, insp,
+		pool := bft.NewPool(log, insp, timeoutHandler,
 			bft.PoolOptions{
 				QueueSize:         3,
 				RequestTimeout:    10 * time.Millisecond,
@@ -345,7 +339,6 @@ func TestReqPoolTimeout(t *testing.T) {
 			},
 		)
 		defer pool.Close()
-		pool.SetTimeoutHandler(timeoutHandler)
 
 		assert.Equal(t, 0, pool.Size())
 		err = pool.Submit(byteReq1)
@@ -383,7 +376,7 @@ func TestReqPoolTimeout(t *testing.T) {
 			to3WG.Done()
 		}).Return()
 
-		pool := bft.NewPool(log, insp,
+		pool := bft.NewPool(log, insp, timeoutHandler,
 			bft.PoolOptions{
 				QueueSize:         3,
 				RequestTimeout:    10 * time.Millisecond,
@@ -392,7 +385,6 @@ func TestReqPoolTimeout(t *testing.T) {
 			},
 		)
 		defer pool.Close()
-		pool.SetTimeoutHandler(timeoutHandler)
 
 		assert.Equal(t, 0, pool.Size())
 		err = pool.Submit(byteReq1)
@@ -429,7 +421,7 @@ func TestReqPoolTimeout(t *testing.T) {
 			assert.Fail(t, "called OnAutoRemoveTimeout")
 		}).Return()
 
-		pool := bft.NewPool(log, insp,
+		pool := bft.NewPool(log, insp, timeoutHandler,
 			bft.PoolOptions{
 				QueueSize:         3,
 				RequestTimeout:    100 * time.Millisecond,
@@ -438,7 +430,6 @@ func TestReqPoolTimeout(t *testing.T) {
 			},
 		)
 		defer pool.Close()
-		pool.SetTimeoutHandler(timeoutHandler)
 		assert.Equal(t, 0, pool.Size())
 		err = pool.Submit(byteReq1)
 		assert.NoError(t, err)

pkg/consensus/consensus.go

@@ -64,16 +64,12 @@ func (c *Consensus) Start() {
 		LeaderFwdTimeout:  requestTimeout,
 		AutoRemoveTimeout: requestTimeout,
 	}
-	pool := algorithm.NewPool(c.Logger, c.RequestInspector, opts)
-	batchBuilder := algorithm.NewBatchBuilder(pool, c.BatchSize, c.BatchTimeout)
 
 	c.controller = &algorithm.Controller{
 		ProposerBuilder:  c,
 		WAL:              c.WAL,
 		ID:               c.SelfID,
 		N:                c.N,
-		Batcher:          batchBuilder,
-		RequestPool:      pool,
 		RequestTimeout:   requestTimeout,
 		Verifier:         c.Verifier,
 		Logger:           c.Logger,
@@ -86,7 +82,10 @@ func (c *Consensus) Start() {
 		RequestInspector: c.RequestInspector,
 	}
 
-	pool.SetTimeoutHandler(c.controller)
+	pool := algorithm.NewPool(c.Logger, c.RequestInspector, c.controller, opts)
+	batchBuilder := algorithm.NewBatchBuilder(pool, c.BatchSize, c.BatchTimeout)
+	c.controller.RequestPool = pool
+	c.controller.Batcher = batchBuilder
 
 	// If we delivered to the application proposal with sequence i,
 	// then we are expecting to be proposed a proposal with sequence i+1.