adding a simple request pool

Signed-off-by: Hagar Meir <[email protected]>

Author: HagarMeir

internal/bft/controller.go

@@ -66,11 +66,6 @@ type Signer interface {
 	SignProposal(types.Proposal) *types.Signature
 }
 
-//go:generate mockery -dir . -name RequestPool -case underscore -output ./mocks/
-type RequestPool interface {
-	Submit(request []byte)
-}
-
 //go:generate mockery -dir . -name Batcher -case underscore -output ./mocks/
 type Batcher interface {
 	NextBatch() [][]byte
@@ -120,15 +115,18 @@ func (c *Controller) leaderID() uint64 {
 }
 
 func (c *Controller) computeQuorum() int {
-	f := int(math.Floor((float64(c.N) - 1.0) / 3.0))
+	f := int((int(c.N) - 1) / 3)
 	q := int(math.Ceil((float64(c.N) + float64(f) + 1) / 2.0))
 	c.Logger.Debugf("The number of nodes (N) is %d, F is %d, and the quorum size is %d", c.N, f, q)
 	return q
 }
 
 // SubmitRequest submits a request to go through consensus
 func (c *Controller) SubmitRequest(request []byte) {
-	c.RequestPool.Submit(request)
+	err := c.RequestPool.Submit(request)
+	if err != nil {
+		c.Logger.Warnf("Request %v was not submitted, error: %v", request, err)
+	}
 }
 
 // ProcessMessages dispatches the incoming message to the required component

internal/bft/mocks/request_inspector.go

@@ -0,0 +1,97 @@
+// Copyright IBM Corp. All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+package bft
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/SmartBFT-Go/consensus/pkg/types"
+	"golang.org/x/sync/semaphore"
+)
+
+//go:generate mockery -dir . -name RequestInspector -case underscore -output ./mocks/
+type RequestInspector interface {
+	RequestID(req []byte) types.RequestInfo
+}
+
+type RequestPool struct {
+	Log              Logger
+	RequestInspector RequestInspector
+	queue            []Request
+	semaphore        *semaphore.Weighted
+	lock             sync.RWMutex
+	QueueSize        int64
+}
+
+type Request struct {
+	ID       string
+	Request  []byte
+	ClientID string
+}
+
+func (rp *RequestPool) Start() {
+	rp.queue = make([]Request, 0)
+	rp.semaphore = semaphore.NewWeighted(rp.QueueSize)
+}
+
+// Submit a request into the pool, returns an error when request is already in the pool
+func (rp *RequestPool) Submit(request []byte) error {
+	reqInfo := rp.RequestInspector.RequestID(request)
+	req := Request{
+		ID:       reqInfo.ID,
+		Request:  request,
+		ClientID: reqInfo.ClientID,
+	}
+	if err := rp.semaphore.Acquire(context.Background(), 1); err != nil {
+		return fmt.Errorf("error in acquiring semaphore: %v", err)
+	}
+	rp.lock.Lock()
+	defer rp.lock.Unlock()
+	for _, existingReq := range rp.queue {
+		if existingReq.ClientID == reqInfo.ClientID && existingReq.ID == reqInfo.ID {
+			rp.semaphore.Release(1)
+			err := fmt.Sprintf("a request with ID %v and client ID %v already exists in the pool", reqInfo.ID, reqInfo.ClientID)
+			rp.Log.Errorf(err)
+			return fmt.Errorf(err)
+		}
+	}
+	rp.queue = append(rp.queue, req)
+	return nil
+}
+
+// NextRequests return the next requests to be batched
+func (rp *RequestPool) NextRequests(n int) []Request {
+	rp.lock.RLock()
+	defer rp.lock.RUnlock()
+	if len(rp.queue) <= n {
+		return rp.queue
+	}
+	return rp.queue[:n]
+}
+
+// RemoveRequest removes the given request from the pool
+func (rp *RequestPool) RemoveRequest(request Request) error {
+	rp.lock.Lock()
+	defer rp.lock.Unlock()
+	removed := false
+	for i, existingReq := range rp.queue {
+		if existingReq.ClientID == request.ClientID && existingReq.ID == request.ID {
+			rp.Log.Infof("Removed request %v from request pool", request)
+			rp.queue = append(rp.queue[:i], rp.queue[i+1:]...)
+			removed = true
+			rp.semaphore.Release(1)
+			break
+		}
+	}
+	if !removed {
+		err := fmt.Sprintf("Request %v is not in the pool at remove time", request)
+		rp.Log.Warnf(err)
+		return fmt.Errorf(err)
+	}
+	return nil
+}

internal/bft/requestpool.go

@@ -0,0 +1,137 @@
+// Copyright IBM Corp. All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+package bft_test
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+
+	"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"
+	"go.uber.org/zap"
+)
+
+func TestReqPoolBasic(t *testing.T) {
+	basicLog, err := zap.NewDevelopment()
+	assert.NoError(t, err)
+	log := basicLog.Sugar()
+	insp := &mocks.RequestInspector{}
+	byteReq1 := []byte{1}
+	insp.On("RequestID", byteReq1).Return(types.RequestInfo{ID: "1", ClientID: "1"})
+	pool := bft.RequestPool{
+		Log:              log,
+		RequestInspector: insp,
+		QueueSize:        3,
+	}
+	pool.Start()
+	err = pool.Submit(byteReq1)
+	assert.NoError(t, err)
+	req1 := bft.Request{
+		ID:       "1",
+		ClientID: "1",
+		Request:  byteReq1,
+	}
+	err = pool.Submit(byteReq1)
+	assert.Error(t, err)
+	err = pool.RemoveRequest(req1)
+	assert.NoError(t, err)
+	err = pool.Submit(byteReq1)
+	assert.NoError(t, err)
+	err = pool.RemoveRequest(req1)
+	assert.NoError(t, err)
+
+	byteReq2 := []byte{2}
+	insp.On("RequestID", byteReq2).Return(types.RequestInfo{ID: "2", ClientID: "2"})
+	err = pool.Submit(byteReq2)
+	assert.NoError(t, err)
+	err = pool.Submit(byteReq1)
+	assert.NoError(t, err)
+	err = pool.Submit(byteReq1)
+	assert.Error(t, err)
+	err = pool.Submit(byteReq2)
+	assert.Error(t, err)
+	err = pool.RemoveRequest(req1)
+	assert.NoError(t, err)
+	err = pool.Submit(byteReq1)
+	assert.NoError(t, err)
+	req2 := bft.Request{
+		ID:       "2",
+		ClientID: "2",
+		Request:  byteReq2,
+	}
+	err = pool.RemoveRequest(req2)
+	assert.NoError(t, err)
+	err = pool.Submit(byteReq2)
+	assert.NoError(t, err)
+
+	byteReq3 := []byte{3}
+	insp.On("RequestID", byteReq3).Return(types.RequestInfo{ID: "3", ClientID: "3"})
+	err = pool.Submit(byteReq3)
+	assert.NoError(t, err)
+
+	next := pool.NextRequests(4)
+	assert.Equal(t, "1", next[0].ID)
+	assert.Equal(t, "2", next[1].ID)
+	assert.Equal(t, "3", next[2].ID)
+	assert.Len(t, next, 3)
+
+	err = pool.RemoveRequest(req2)
+	assert.NoError(t, err)
+
+	next = pool.NextRequests(4)
+	assert.Equal(t, "1", next[0].ID)
+	assert.Equal(t, "3", next[1].ID)
+	assert.Len(t, next, 2)
+
+	next = pool.NextRequests(1)
+	assert.Equal(t, "1", next[0].ID)
+	assert.Len(t, next, 1)
+
+}
+
+func TestEventuallySubmit(t *testing.T) {
+	n := 100
+	basicLog, err := zap.NewDevelopment()
+	assert.NoError(t, err)
+	log := basicLog.Sugar()
+	insp := &mocks.RequestInspector{}
+	pool := bft.RequestPool{
+		Log:              log,
+		RequestInspector: insp,
+		QueueSize:        50,
+	}
+	pool.Start()
+	wg := sync.WaitGroup{}
+	wg.Add(2 * n)
+	for i := 0; i < n; i++ {
+		go func(i int) {
+			byteReq := []byte{byte(i)}
+			str := fmt.Sprintf("%d", i)
+			insp.On("RequestID", byteReq).Return(types.RequestInfo{ID: str, ClientID: str})
+			err := pool.Submit(byteReq)
+			assert.NoError(t, err)
+			wg.Done()
+		}(i)
+		go func(i int) {
+			byteReq := []byte{byte(i)}
+			str := fmt.Sprintf("%d", i)
+			req := bft.Request{
+				ID:       str,
+				ClientID: str,
+				Request:  byteReq,
+			}
+			err := pool.RemoveRequest(req)
+			for err != nil {
+				err = pool.RemoveRequest(req)
+			}
+			wg.Done()
+		}(i)
+	}
+	wg.Wait()
+}

internal/bft/requestpool_test.go

@@ -42,10 +42,6 @@ func (c *Consensus) Sync() (protos.ViewMetadata, uint64) {
 	panic("implement me")
 }
 
-func (c *Consensus) Submit(request []byte) {
-	panic("implement me")
-}
-
 func (c *Consensus) BatchRemainder(remainder [][]byte) {
 	panic("implement me")
 }
@@ -69,10 +65,12 @@ type Future interface {
 }
 
 func (c *Consensus) Start() Future {
+	pool := algorithm.RequestPool{}
+	// TODO use request pool in example
 	c.controller = &algorithm.Controller{
 		ID:              c.SelfID,
 		N:               4,
-		RequestPool:     c,
+		RequestPool:     pool,
 		Batcher:         c,
 		Verifier:        c.Verifier,
 		Logger:          c.Logger,