avoid throughput quantized

Author: linkdata

operation.go

@@ -73,25 +73,25 @@ func (op *Operation) run(ch chan<- int64) {
 
 	if stopCh != nil {
 		for {
-			var limitCh chan<- int64
 			var todo int64
-			var batch int64
 			if limit := op.Limit.Load(); limit > 0 {
 				carry += limit
 				todo = carry / secparts
 				carry = carry % secparts
 				todo += op.avail.Swap(0)
-				if todo > 0 {
-					limitCh = ch
-					batch = min(batchsize, todo)
-				}
 			} else {
 				carry = 0
 			}
 			waitCh := op.WaitCh()
 
 		partialsecond:
 			for {
+				var limitCh chan<- int64
+				var batch int64
+				if todo > 0 {
+					limitCh = ch
+					batch = min(batchsize, todo)
+				}
 				select {
 				case <-stopCh:
 					return
@@ -100,15 +100,15 @@ func (op *Operation) run(ch chan<- int64) {
 				case limitCh <- batch:
 					todo -= batch
 					todo += op.avail.Swap(0)
-					if todo < batch {
-						<-waitCh
-						break partialsecond
-					}
 				case <-waitCh:
 					break partialsecond
 				}
 			}
 
+			// Preserve any budget left in this slice for the next one.
+			if todo > 0 {
+				op.avail.Add(todo)
+			}
 			count := op.count.Swap(0)
 			op.Count.Add(count)
 			counts[seccount] = count

operation_test.go

@@ -164,7 +164,7 @@ func TestOperation_read_rate_low(t *testing.T) {
 			t.Log(l.Reads.Limit.Load())
 			t.Error(elapsed)
 		}
-		if rate < 900 || rate > 1000 {
+		if rate < 900 || rate > 1010 {
 			t.Error(rate)
 		}
 	})
@@ -239,6 +239,38 @@ func TestOperation_read_rate_high(t *testing.T) {
 	})
 }
 
+func TestOperation_read_rate_nonBatchMultiple(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		const limit = 50000
+		ticker := NewTicker()
+		defer ticker.Stop()
+		l := ticker.NewLimiter(limit)
+		defer l.Stop()
+
+		r := bytes.NewReader(make([]byte, limit))
+		buf := make([]byte, limit)
+
+		now := time.Now()
+		n, err := l.Reads.io(r.Read, buf)
+		elapsed := time.Since(now)
+		<-l.WaitCh()
+		synctest.Wait()
+
+		if n != limit {
+			t.Fatal(n)
+		}
+		if err != nil {
+			t.Fatal(err)
+		}
+		if elapsed < time.Millisecond*900 || elapsed > time.Millisecond*1300 {
+			t.Fatal(elapsed)
+		}
+		if rate := l.Reads.Rate.Load(); rate < limit-(limit/10) {
+			t.Fatalf("rate too low: got %d want at least %d", rate, limit-(limit/10))
+		}
+	})
+}
+
 func TestOperation_write_rate(t *testing.T) {
 	synctest.Test(t, func(t *testing.T) {
 		ticker := NewTicker()