address review comments

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

Author: HagarMeir

node/batcher/batcher_builder.go

@@ -127,6 +127,7 @@ func createMemPool(b *Batcher, config *node_config.BatcherNodeConfig) MemPool {
 		BatchMaxSize:          config.BatchMaxSize,
 		BatchMaxSizeBytes:     config.BatchMaxBytes,
 		RequestMaxBytes:       config.RequestMaxBytes,
+		BatchTimeout:          config.BatchCreationTimeout,
 		SubmitTimeout:         config.SubmitTimeout,
 		FirstStrikeThreshold:  config.FirstStrikeThreshold,
 		SecondStrikeThreshold: config.SecondStrikeThreshold,

request/pool.go

@@ -62,6 +62,7 @@ type PoolOptions struct {
 	BatchMaxSize          uint32
 	BatchMaxSizeBytes     uint32
 	RequestMaxBytes       uint64
+	BatchTimeout          time.Duration
 	SubmitTimeout         time.Duration
 	FirstStrikeThreshold  time.Duration
 	SecondStrikeThreshold time.Duration
@@ -91,6 +92,17 @@ func (rp *Pool) start() {
 	rp.pending.Start()
 }
 
+func (rp *Pool) addRandomnessToFirstStrike() time.Duration {
+	if rp.options.BatchTimeout.Milliseconds() == 0 {
+		return time.Duration(0) // no randomness
+	}
+	return time.Duration(randRange(int(2*rp.options.BatchTimeout.Milliseconds()), int(-2*rp.options.BatchTimeout.Milliseconds()))) * time.Millisecond
+}
+
+func randRange(max, min int) int {
+	return rand.Intn(max-min) + min
+}
+
 func (rp *Pool) createPendingStore() *PendingStore {
 	return &PendingStore{
 		Inspector:             rp.inspector,
@@ -100,12 +112,12 @@ func (rp *Pool) createPendingStore() *PendingStore {
 		StartTime:             time.Now(),
 		Logger:                rp.logger,
 		SecondStrikeThreshold: rp.options.SecondStrikeThreshold,
-		FirstStrikeThreshold:  rp.options.FirstStrikeThreshold,
+		FirstStrikeThreshold:  rp.options.FirstStrikeThreshold + rp.addRandomnessToFirstStrike(),
 		OnDelete: func(key string) {
 			rp.semaphore.Release(1)
 			atomic.AddInt64(&rp.size, -1)
 		},
-		Epoch:                time.Duration(rand.Intn(50))*time.Millisecond + rp.options.FirstStrikeThreshold/10,
+		Epoch:                rp.options.FirstStrikeThreshold / 10,
 		FirstStrikeCallback:  rp.striker.OnFirstStrikeTimeout,
 		SecondStrikeCallback: rp.striker.OnSecondStrikeTimeout,
 	}

request/pool_test.go

@@ -146,6 +146,8 @@ func TestRestartPool(t *testing.T) {
 		SubmitTimeout:         time.Second * 10,
 	}, &striker{})
 
+	assert.Equal(t, 5*time.Second, pool.pending.FirstStrikeThreshold)
+
 	pool.Restart(true)
 
 	count := 100
@@ -214,6 +216,8 @@ func TestBasicBatching(t *testing.T) {
 		SubmitTimeout:         time.Second * 10,
 	}, &striker{})
 
+	assert.Equal(t, 5*time.Second, pool.pending.FirstStrikeThreshold)
+
 	pool.Restart(true)
 
 	ctx, cancel1 := context.WithTimeout(context.Background(), time.Second)
@@ -253,8 +257,13 @@ func TestBasicBatching(t *testing.T) {
 		RequestMaxBytes:       100 * 1024,
 		AutoRemoveTimeout:     time.Second * 10,
 		SubmitTimeout:         time.Second * 10,
+		BatchTimeout:          time.Second,
 	}, &striker{})
 
+	t.Logf("First strike with random is %f seconds\n", pool.pending.FirstStrikeThreshold.Seconds())
+	assert.GreaterOrEqual(t, 7*time.Second, pool.pending.FirstStrikeThreshold)
+	assert.LessOrEqual(t, 3*time.Second, pool.pending.FirstStrikeThreshold)
+
 	pool.Restart(true)
 
 	assert.NoError(t, pool.Submit(byteReq4))