chore: fix pr comments 5

Signed-off-by: Gil Levkovich <[email protected]>

Author: glevkovich

src/facade/dragonfly_connection.cc

@@ -1113,6 +1113,12 @@ void Connection::ConnectionFlow() {
 
 void Connection::DispatchSingle(bool has_more, absl::FunctionRef<void()> invoke_cb,
                                 absl::FunctionRef<void()> enqueue_cmd_cb) {
+  // Unconditional return when closing:
+  // else, non-throttled connections skip the check below and enqueue data even if they are closing.
+  // No one will read that data anyway.
+  if (cc_->conn_closing)
+    return;
+
   bool optimize_for_async = has_more;
   QueueBackpressure& qbp = GetQueueBackpressure();
   size_t global_pipeline_bytes = qbp.pipeline_bytes.load(std::memory_order_relaxed);
@@ -1150,9 +1156,7 @@ void Connection::DispatchSingle(bool has_more, absl::FunctionRef<void()> invoke_
 
   // Dispatch async if we're handling a pipeline or if we can't dispatch sync.
   if (optimize_for_async || !can_dispatch_sync) {
-    if (!cc_->conn_closing) {
-      LaunchAsyncFiberIfNeeded();
-    }
+    LaunchAsyncFiberIfNeeded();
     enqueue_cmd_cb();
   } else {
     ShrinkPipelinePool();  // Gradually release pipeline request pool.

src/server/server_family.cc

@@ -1558,7 +1558,8 @@ void PrintPrometheusMetrics(uint64_t uptime, const Metrics& m, DflyCmd* dfly_cmd
                     &resp->body());
   AppendMetricWithoutLabels("blocked_clients", "", conn_stats.num_blocked_clients,
                             MetricType::GAUGE, &resp->body());
-  AppendMetricWithoutLabels("pipeline_queue_length", "", conn_stats.dispatch_queue_entries,
+  AppendMetricWithoutLabels("pipeline_queue_length", "",
+                            conn_stats.dispatch_queue_entries + conn_stats.pipeline_queue_entries,
                             MetricType::GAUGE, &resp->body());
   AppendMetricWithoutLabels("send_delay_seconds", "",
                             double(GetDelayMs(m.oldest_pending_send_ts)) / 1000.0,
@@ -1770,6 +1771,8 @@ void PrintPrometheusMetrics(uint64_t uptime, const Metrics& m, DflyCmd* dfly_cmd
                               MetricType::GAUGE, &resp->body());
     AppendMetricWithoutLabels("dispatch_queue_bytes", "", conn_stats.dispatch_queue_bytes,
                               MetricType::GAUGE, &resp->body());
+    AppendMetricWithoutLabels("pipeline_queue_bytes", "", conn_stats.pipeline_queue_bytes,
+                              MetricType::GAUGE, &resp->body());
     AppendMetricWithoutLabels("pipeline_cmd_cache_bytes", "", conn_stats.pipeline_cmd_cache_bytes,
                               MetricType::GAUGE, &resp->body());
   }