pubsub: simplify and improve batch sizing, especially for low message rates

Author: vangent

pubsub/benchmark_test.go

@@ -159,6 +159,13 @@ func TestReceivePerformance(t *testing.T) {
 				return d
 			},
 		},
+		{
+			description: "intermittent",
+			receiveProfile: func(_ bool, maxMessages int) (int, time.Duration) {
+				n := rand.Int() % 2
+				return n, 250 * time.Millisecond
+			},
+		},
 	}
 
 	for _, test := range tests {

pubsub/pubsub.go

@@ -386,8 +386,7 @@ type Subscription struct {
 	unreportedAckErr error             // permanent error from background SendAcks that hasn't been returned to the user yet
 	waitc            chan struct{}     // for goroutines waiting on ReceiveBatch
 	runningBatchSize float64           // running number of messages to request via ReceiveBatch
-	throughputStart  time.Time         // start time for throughput measurement, or the zero Time if queue is empty
-	throughputEnd    time.Time         // end time for throughput measurement, or the zero Time if queue is not empty
+	throughputStart  time.Time         // start time for throughput measurement
 	throughputCount  int               // number of msgs given out via Receive since throughputStart
 
 	// Used in tests.
@@ -471,14 +470,11 @@ func (s *Subscription) updateBatchSize() int {
 	} else {
 		// Update s.runningBatchSize based on throughput since our last time here,
 		// as measured by the ratio of the number of messages returned to elapsed
-		// time when there were messages available in the queue.
-		if s.throughputEnd.IsZero() {
-			s.throughputEnd = now
-		}
-		elapsed := s.throughputEnd.Sub(s.throughputStart)
-		if elapsed == 0 {
-			// Avoid divide-by-zero.
-			elapsed = 1 * time.Millisecond
+		// time.
+		elapsed := now.Sub(s.throughputStart)
+		if elapsed < 100*time.Millisecond {
+			// Avoid divide-by-zero and huge numbers.
+			elapsed = 100 * time.Millisecond
 		}
 		msgsPerSec := float64(s.throughputCount) / elapsed.Seconds()
 
@@ -500,13 +496,7 @@ func (s *Subscription) updateBatchSize() int {
 	}
 
 	// Reset throughput measurement markers.
-	if len(s.q) > 0 {
-		s.throughputStart = now
-	} else {
-		// Will get set to non-zero value when we receive some messages.
-		s.throughputStart = time.Time{}
-	}
-	s.throughputEnd = time.Time{}
+	s.throughputStart = now
 	s.throughputCount = 0
 
 	// Using Ceil guarantees at least one message.
@@ -569,6 +559,7 @@ func (s *Subscription) Receive(ctx context.Context) (_ *Message, err error) {
 			// waiting goroutines, by closing s.waitc.
 			s.waitc = make(chan struct{})
 			batchSize := s.updateBatchSize()
+			// log.Printf("BATCH SIZE %d", batchSize)
 
 			go func() {
 				if s.preReceiveBatchHook != nil {
@@ -583,12 +574,6 @@ func (s *Subscription) Receive(ctx context.Context) (_ *Message, err error) {
 				} else if len(msgs) > 0 {
 					s.q = append(s.q, msgs...)
 				}
-				// Set the start time for measuring throughput even if we didn't get
-				// any messages; this allows batch size to decay over time if there
-				// aren't any message available.
-				if s.throughputStart.IsZero() {
-					s.throughputStart = time.Now()
-				}
 				close(s.waitc)
 				s.waitc = nil
 			}()
@@ -638,10 +623,6 @@ func (s *Subscription) Receive(ctx context.Context) (_ *Message, err error) {
 			})
 			return m2, nil
 		}
-		// No messages are available. Close the interval for throughput measurement.
-		if s.throughputEnd.IsZero() && !s.throughputStart.IsZero() && s.throughputCount > 0 {
-			s.throughputEnd = time.Now()
-		}
 		// A call to ReceiveBatch must be in flight. Wait for it.
 		waitc := s.waitc
 		s.mu.Unlock()