Waiting requests counting and reporting are added.

Author: dmitripikus

pkg/vllm-sim/defs.go

@@ -63,6 +63,8 @@ type VllmSimulator struct {
 	maxRunningReqs int64
 	// nRunningReqs ithe the number of inference requests that are currently being processed
 	nRunningReqs int64
+	// nWaitingReqs ithe the number of inference requests that are waiting to be processed
+	nWaitingReqs int64
 	// loraInfo is prometheus gauge
 	loraInfo *prometheus.GaugeVec
 	// runningRequests is prometheus gauge

pkg/vllm-sim/metrics.go

@@ -130,9 +130,16 @@ func (s *VllmSimulator) reportLoras() {
 		"").Set(float64(time.Now().Unix()))
 }
 
-// reportRequests sets information about running completion requests
-func (s *VllmSimulator) reportRequests() {
+// reportRunningRequests sets information about running completion requests
+func (s *VllmSimulator) reportRunningRequests() {
 	nRunningReqs := atomic.LoadInt64(&(s.nRunningReqs))
 	s.runningRequests.WithLabelValues(
 		s.model).Set(float64(nRunningReqs))
 }
+
+// reportWaitingRequests sets information about waiting completion requests
+func (s *VllmSimulator) reportWaitingRequests() {
+	nWaitingReqs := atomic.LoadInt64(&(s.nWaitingReqs))
+	s.waitingRequests.WithLabelValues(
+		s.model).Set(float64(nWaitingReqs))
+}

pkg/vllm-sim/simulator.go

@@ -41,7 +41,7 @@ import (
 func New(logger logr.Logger) *VllmSimulator {
 	return &VllmSimulator{
 		logger:  logger,
-		reqChan: make(chan *completionReqCtx),
+		reqChan: make(chan *completionReqCtx, 1000),
 	}
 }
 
@@ -267,6 +267,8 @@ func (s *VllmSimulator) handleCompletions(ctx *fasthttp.RequestCtx, isChatComple
 		wg:               &wg,
 	}
 	s.reqChan <- reqCtx
+	atomic.StoreInt64(&(s.nWaitingReqs), int64(len(s.reqChan)))
+	s.reportWaitingRequests()
 	wg.Wait()
 }
 
@@ -281,6 +283,9 @@ func (s *VllmSimulator) reqProcessingWorker(ctx context.Context, id int) {
 				s.logger.Info("reqProcessingWorker worker exiting: reqChan closed")
 				return
 			}
+			atomic.StoreInt64(&(s.nWaitingReqs), int64(len(s.reqChan)))
+			s.reportWaitingRequests()
+
 			req := reqCtx.completionReq
 			model := req.getModel()
 			if s.isLora(model) {
@@ -301,7 +306,7 @@ func (s *VllmSimulator) reqProcessingWorker(ctx context.Context, id int) {
 				s.reportLoras()
 			}
 			atomic.AddInt64(&(s.nRunningReqs), 1)
-			s.reportRequests()
+			s.reportRunningRequests()
 
 			responseTxt := req.createResponseText(s.mode)
 
@@ -319,7 +324,7 @@ func (s *VllmSimulator) reqProcessingWorker(ctx context.Context, id int) {
 func (s *VllmSimulator) responseSentCallback(model string) {
 
 	atomic.AddInt64(&(s.nRunningReqs), -1)
-	s.reportRequests()
+	s.reportRunningRequests()
 
 	if model == s.model {
 		// this is base model - do not continue