Merge pull request #472 from yacovm/nonblockingSubmit

Fail fast if request already exists in pool

Author: C0rWin

internal/bft/requestpool.go

@@ -22,6 +22,10 @@ const (
 	defaultRequestTimeout = 10 * time.Second // for unit tests only
 )
 
+var (
+	ErrReqAlreadyExists = fmt.Errorf("request already exists")
+)
+
 //go:generate mockery -dir . -name RequestTimeoutHandler -case underscore -output ./mocks/
 
 // RequestTimeoutHandler defines the methods called by request timeout timers created by time.AfterFunc.
@@ -46,7 +50,7 @@ type Pool struct {
 	inspector api.RequestInspector
 	options   PoolOptions
 
-	lock           sync.Mutex
+	lock           sync.RWMutex
 	fifo           *list.List
 	semaphore      *semaphore.Weighted
 	existMap       map[types.RequestInfo]*list.Element
@@ -138,6 +142,15 @@ func (rp *Pool) Submit(request []byte) error {
 		return errors.Errorf("pool closed, request rejected: %s", reqInfo)
 	}
 
+	rp.lock.RLock()
+	_, alreadyExists := rp.existMap[reqInfo]
+	rp.lock.RUnlock()
+
+	if alreadyExists {
+		rp.logger.Debugf("request %s already exists in the pool", reqInfo)
+		return ErrReqAlreadyExists
+	}
+
 	// do not wait for a semaphore with a lock, as it will prevent draining the pool.
 	if err := rp.semaphore.Acquire(context.Background(), 1); err != nil {
 		return errors.Wrapf(err, "acquiring semaphore for request: %s", reqInfo)
@@ -150,9 +163,9 @@ func (rp *Pool) Submit(request []byte) error {
 
 	if _, exist := rp.existMap[reqInfo]; exist {
 		rp.semaphore.Release(1)
-		errStr := fmt.Sprintf("request %s already exists in the pool", reqInfo)
+		errStr := fmt.Sprintf("request %s has been already added to the pool", reqInfo)
 		rp.logger.Debugf(errStr)
-		return errors.New(errStr)
+		return ErrReqAlreadyExists
 	}
 
 	to := time.AfterFunc(

internal/bft/requestpool_test.go

@@ -471,7 +471,7 @@ func TestReqPoolTimeout(t *testing.T) {
 
 		pool.RestartTimers()
 		err = pool.Submit(byteReq2)
-		assert.EqualError(t, err, "request {2 2} already exists in the pool")
+		assert.Equal(t, bft.ErrReqAlreadyExists, err)
 		pool.StopTimers()
 		pool.RestartTimers()