chore(wp): fix comments

Signed-off-by: Gil Levkovich <69595609+glevkovich@users.noreply.github.com>

Author: glevkovich

src/facade/dragonfly_connection.cc

@@ -5,6 +5,7 @@
 
 #include "facade/dragonfly_connection.h"
 
+#include <absl/cleanup/cleanup.h>
 #include <absl/container/flat_hash_map.h>
 #include <absl/strings/escaping.h>
 #include <absl/strings/match.h>
@@ -1112,6 +1113,7 @@ void Connection::ConnectionFlow() {
     uint32_t prev_timeout{socket_->timeout()};
     static constexpr uint32_t kSocketDrainTimeoutMs{100};
     socket_->set_timeout(kSocketDrainTimeoutMs);
+    auto restore_timeout = absl::MakeCleanup([&] { socket_->set_timeout(prev_timeout); });
     while (!ec2) {
       // Discard any received data.
       io_buf_.Clear();
@@ -1120,7 +1122,6 @@ void Connection::ConnectionFlow() {
         break;  // Peer closed connection or timeout.
       }
     }
-    socket_->set_timeout(prev_timeout);
   }
 
   if (ec && !FiberSocketBase::IsConnClosed(ec)) {
@@ -1165,7 +1166,7 @@ Connection::ParserStatus Connection::ParseRedis(unsigned max_busy_cycles) {
       request_consumed_bytes_ = 0;
       bool has_more = consumed < read_buffer.size();
 
-      if (tl_traffic_logger.log_file && IsMain()) {
+      if (tl_traffic_logger.log_file && IsMain() /* log only on the main interface */) {
         LogTraffic(id_, has_more, *parsed_cmd_, service_->GetContextInfo(cc_.get()));
       }
 
@@ -1469,9 +1470,9 @@ void Connection::SquashPipeline() {
 
   local_stats_.cmds += result.processed;
   last_interaction_ = time(nullptr);
-  uint32_t dispatched = result.processed;
-  uint64_t before_flush = CycleClock::Now();
-  //
+  uint32_t num_dispatched_cmds = result.processed;
+  uint64_t flush_start_cycle_cnt = CycleClock::Now();
+
   // TODO: to investigate if always flushing will improve P99 latency because otherwise we
   // wait for the next batch to finish before fully flushing the current response.
   if (pending_pipeline_cmd_cnt_ == pipeline_count ||
@@ -1486,19 +1487,20 @@ void Connection::SquashPipeline() {
 
   if (result.account_in_stats) {
     stats_->pipeline_dispatch_calls++;
-    stats_->pipeline_dispatch_commands += dispatched;
-    stats_->pipeline_dispatch_flush_usec += CycleClock::ToUsec(CycleClock::Now() - before_flush);
+    stats_->pipeline_dispatch_commands += num_dispatched_cmds;
+    stats_->pipeline_dispatch_flush_usec +=
+        CycleClock::ToUsec(CycleClock::Now() - flush_start_cycle_cnt);
   }
 
   // update parsed_to_execute_ pointer
-  for (uint32_t i = 0; i < dispatched; ++i) {
+  for (uint32_t i = 0; i < num_dispatched_cmds; ++i) {
     if (parsed_to_execute_) {
       parsed_to_execute_ = parsed_to_execute_->next;
     }
   }
 
   auto* current{parsed_head_};
-  for (size_t i{}; i < dispatched; ++i) {
+  for (size_t i{}; i < num_dispatched_cmds; ++i) {
     DCHECK(current);
     auto* next{current->next};
 
@@ -1515,7 +1517,7 @@ void Connection::SquashPipeline() {
   }
 
   // If interrupted due to pause, fall back to regular dispatch
-  skip_next_squashing_ = (dispatched != pipeline_count);
+  skip_next_squashing_ = (num_dispatched_cmds != pipeline_count);
 }
 
 void Connection::ClearPipelinedMessages() {
@@ -1528,7 +1530,7 @@ void Connection::ClearPipelinedMessages() {
     FiberAtomicGuard guard;  // don't suspend when concluding to avoid getting new messages
     if (msg.IsControl())
       visit(async_op, msg.handle);  // to not miss checkpoints
-    RecycleIntrusiveMessage(std::move(msg));
+    UpdateIntrusiveMessageStats(std::move(msg));
   }
   dispatch_q_.clear();
 
@@ -1541,7 +1543,7 @@ void Connection::ClearPipelinedMessages() {
     if (stats_) {
       stats_->dispatch_queue_bytes -= curr->UsedMemory();
     }
-    RecyclePipelineCommand(curr);
+    UpdatePipelineCommandStats(curr);
     PipelineMessagePtr ptr(curr);  // reclaim ownership and destroy
   }
   parsed_head_ = nullptr;
@@ -1647,7 +1649,7 @@ void Connection::AsyncFiber() {
       }
 
       if (ShouldEndAsyncFiber(msg)) {
-        RecycleIntrusiveMessage(std::move(msg));
+        UpdateIntrusiveMessageStats(std::move(msg));
         CHECK(dispatch_q_.empty()) << DebugInfo();
         qbp.pipeline_cnd.notify_all();
         return;  // don't set conn closing flag
@@ -1656,7 +1658,7 @@ void Connection::AsyncFiber() {
       cc_->async_dispatch = true;
       std::visit(async_op, msg.handle);
       cc_->async_dispatch = false;
-      RecycleIntrusiveMessage(std::move(msg));
+      UpdateIntrusiveMessageStats(std::move(msg));
 
       // Ensure we check for more control messages before touching data pipeline messages.
       continue;
@@ -1665,12 +1667,17 @@ void Connection::AsyncFiber() {
     // 2. Handle Data Pipeline (Intrusive List) - only runs when dispatch_q_ is empty.
     DCHECK(dispatch_q_.empty());
     if (parsed_head_) {
+      // Ensure pointers are synchronized before deciding how to execute.
+      DCHECK_EQ(parsed_head_, parsed_to_execute_) << "Pointers diverged before dispatch decision.";
+
       bool squashing_enabled{squashing_threshold > 0};
       bool threshold_reached{pending_pipeline_cmd_cnt_ > squashing_threshold};
 
       if (squashing_enabled && threshold_reached && !skip_next_squashing_ &&
           !IsReplySizeOverLimit()) {
         SquashPipeline();
+        if (reply_builder_->GetError())
+          break;
       } else {
         auto replies_recorded_before{reply_builder_->RepliesRecorded()};
         if (!ExecuteRedisBatch())
@@ -1692,7 +1699,7 @@ void Connection::AsyncFiber() {
 
     if (subscriber_over_limit && stats_->dispatch_queue_subscriber_bytes < qbp.publish_buffer_limit)
       qbp.pubsub_ec.notify();
-  }
+  }  // while
 
   DCHECK(cc_->conn_closing || reply_builder_->GetError());
   cc_->conn_closing = true;
@@ -1721,8 +1728,8 @@ bool Connection::ExecuteRedisBatch() {
     if (cmd == parsed_head_) {
       // We've just executed the head of the pipeline queue. Advance the head and recycle command.
       parsed_head_ = parsed_to_execute_;
-      ReleaseParsedCommand(cmd, true);
     }
+    ReleaseParsedCommand(cmd, true);
 
     if (reply_builder_->GetError()) {
       return false;
@@ -1911,7 +1918,7 @@ void Connection::SendAsync(MessageHandle msg) {
   }
 }
 
-void Connection::RecycleIntrusiveMessage(MessageHandle msg) {
+void Connection::UpdateIntrusiveMessageStats(MessageHandle msg) {
   size_t used_mem = msg.UsedMemory();
 
   stats_->dispatch_queue_bytes -= used_mem;
@@ -1923,9 +1930,11 @@ void Connection::RecycleIntrusiveMessage(MessageHandle msg) {
     qbp.subscriber_bytes.fetch_sub(used_mem, memory_order_relaxed);
     stats_->dispatch_queue_subscriber_bytes -= used_mem;
   }
+
+  DCHECK_GE(stats_->dispatch_queue_bytes, stats_->dispatch_queue_subscriber_bytes);
 }
 
-void Connection::RecyclePipelineCommand(facade::ParsedCommand* cmd) {
+void Connection::UpdatePipelineCommandStats(facade::ParsedCommand* cmd) {
   DCHECK(cmd);
 
   if (cmd->parsed_cycle) {

src/facade/dragonfly_connection.h

@@ -301,8 +301,8 @@ class Connection : public util::Connection {
   io::Result<bool> CheckForHttpProto();
 
   // Executes pending Redis commands from the pipeline list.
-  // Returns early if the separate dispatch queue contains a control message,
-  // allowing the main loop to prioritize it.
+  // Returns true if the batch finished or was paused to prioritize a control message.
+  // Returns false if a fatal error occurred (e.g., socket closed or builder error).
   bool ExecuteRedisBatch();
 
   // Handles events from the dispatch queue.
@@ -315,8 +315,8 @@ class Connection : public util::Connection {
   void SendAsync(MessageHandle msg);
 
   // Updates memory stats and pooling, must be called for all used messages
-  void RecycleIntrusiveMessage(MessageHandle msg);
-  void RecyclePipelineCommand(facade::ParsedCommand* cmd);
+  void UpdateIntrusiveMessageStats(MessageHandle msg);
+  void UpdatePipelineCommandStats(facade::ParsedCommand* cmd);
 
   ParserStatus ParseRedis(unsigned max_busy_cycles);
   ParserStatus ParseMemcache();

src/facade/facade_types.h

@@ -141,13 +141,13 @@ inline std::string_view ArgS(ArgSlice args, size_t i) {
 }
 
 struct ConnectionStats {
-  size_t read_buf_capacity = 0;  // total capacity of input buffers
-  uint64_t dispatch_queue_entries =
-      0;  // total count of all pending messages (data + administrative messages)
+  size_t read_buf_capacity = 0;         // total capacity of input buffers
+  uint64_t dispatch_queue_entries = 0;  // total count of all pending
+                                        // messages (data + administrative messages)
   size_t dispatch_queue_bytes = 0;  // total size of memory used by all pending messages (pipeline
                                     // data queue + dispatch queue admin messages)
-  size_t dispatch_queue_subscriber_bytes =
-      0;  // total size of all publish messages (subset of dispatch_queue_bytes)
+  size_t dispatch_queue_subscriber_bytes = 0;  // total size of all publish messages
+                                               // (subset of dispatch_queue_bytes)
 
   size_t pipeline_cmd_cache_bytes = 0;