feat(scheduler): add back pressure for table dispatcher (#188)

* feat(scheduler): add back pressure for table dispatcher
moiot · Jul 15, 2019 · 7369fa9 · 7369fa9
1 parent aa2dcbe
commit 7369fa9
Showing 1 changed file with 12 additions and 2 deletions.
diff --git a/pkg/schedulers/batch_table_scheduler/batch_table_scheduler.go b/pkg/schedulers/batch_table_scheduler/batch_table_scheduler.go
@@ -512,8 +512,18 @@ func (scheduler *batchScheduler) startTableDispatcher(tableKey string) {
 			metrics.QueueLength.WithLabelValues(env.PipelineName, "table-latch", key).Set(float64(len(latches)))
 			metrics.QueueLength.WithLabelValues(env.PipelineName, "table-latch-ack", key).Set(float64(latchQLen))
 
-			if len(batch) >= scheduler.cfg.MaxBatchPerWorker || ((queueLen+latchQLen) == 0 && len(batch) > 0) {
-				flushFunc()
+			if len(batch) > 0 {
+				if len(batch) >= scheduler.cfg.MaxBatchPerWorker {
+					flushFunc()
+				} else if (queueLen + latchQLen) == 0 {
+					queueIdx := round % uint(scheduler.cfg.NrWorker)
+					round++
+					if len(scheduler.workerQueues[queueIdx]) < scheduler.cfg.QueueSize/2 {
+						flushFunc()
+					} else {
+						// worker queue has many items pending, try to accumulate message in the batch.
+					}
+				}
 			}
 		}