chore(wp): wp

[glevkovich]

draft - WIP

[Copilot]

Pull request overview

This PR refactors the command dispatch mechanism in Dragonfly's connection handling. It replaces the previous mixed approach (using both a dispatch queue and synchronous dispatch) with a unified pipeline queue based on an intrusive linked list for data commands, while keeping a separate dispatch queue for administrative/control messages.

Changes:

• Remove DispatchSingle() function and replace with ExecuteRedisBatch() for processing pipeline commands
• Separate pipeline command recycling (RecyclePipelineCommand()) from intrusive message recycling (RecycleIntrusiveMessage())
• Update SquashPipeline() to work with the new intrusive list structure
• Add timeout handling during connection drain to prevent hangs with non-responsive clients
• Improve logging and error messages for better debugging

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 6 comments.

File	Description
src/facade/facade_types.h	Update comments for ConnectionStats fields to clarify combined queue accounting
src/facade/dragonfly_connection.h	Remove PipelineMessagePtr from MessageHandle variant and update method signatures
src/facade/dragonfly_connection.cc	Major refactoring of dispatch logic, removing DispatchSingle and implementing ExecuteRedisBatch with intrusive list

Comments suppressed due to low confidence (1)

src/facade/dragonfly_connection.cc:1685

• Inconsistent backpressure logic: At line 1191, backpressure is checked using the combined count of both queues (parsed_cmd_q_len_ + dispatch_q_.size()), but at line 1684, only dispatch_q_.size() is checked. This inconsistency could lead to incorrect backpressure behavior. The check at line 1684 should also use the combined count for consistency.

    if (!qbp.IsPipelineBufferOverLimit(stats_->dispatch_queue_bytes, dispatch_q_.size()) ||
        dispatch_q_.empty()) {

[Copilot]

Pull request overview

Copilot reviewed 6 out of 6 changed files in this pull request and generated 2 comments.

Comments suppressed due to low confidence (1)

src/facade/dragonfly_connection.cc:1692

• Inconsistent backpressure notification logic. At line 1691, the notification check only considers dispatch_q_.size(), but the backpressure wait condition in ParseRedis (line 1199) checks parsed_cmd_q_len_ + dispatch_q_.size(). This means the parser could be waiting on backpressure for the total count (dispatch + pipeline), but this notification only checks if dispatch_q_ is below the limit. The notification should use the same total count calculation: parsed_cmd_q_len_ + dispatch_q_.size().

    if (!qbp.IsPipelineBufferOverLimit(stats_->dispatch_queue_bytes, dispatch_q_.size()) ||
        dispatch_q_.empty()) {

[Copilot]

Pull request overview

Copilot reviewed 6 out of 6 changed files in this pull request and generated 16 comments.

Comments suppressed due to low confidence (1)

src/facade/dragonfly_connection.cc:1691

• The condition variable notification logic on lines 1688-1691 checks if the pipeline is not over limit OR if dispatch_q_ is empty before notifying. However, this condition seems inverted - we should notify waiting parsers when we drop BELOW the limit, not when we're NOT over the limit OR queue is empty. The current logic will notify even if we're still over the limit as long as dispatch_q_ is empty. This could wake up parsers prematurely. Consider simplifying to only notify when actually below limits.

    if (!qbp.IsPipelineBufferOverLimit(stats_->dispatch_queue_bytes, dispatch_q_.size()) ||
        dispatch_q_.empty()) {
      qbp.pipeline_cnd.notify_all();  // very cheap if noone is waiting on it.
    }

[Copilot]

Copilot encountered an error and was unable to review this pull request. You can try again by re-requesting a review.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 3 comments.

Comments suppressed due to low confidence (1)

src/facade/dragonfly_connection.cc:1461

• The comment says 'in case the next dispatch is sync', but sync dispatch has been removed in this PR. This comment is now outdated and should be updated or removed.

    reply_builder_->SetBatchMode(false);  // in case the next dispatch is sync

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.

Comments suppressed due to low confidence (1)

src/facade/dragonfly_connection.cc:1669

• Backpressure notification check should consider total pending count (dispatch_q_.size() + parsed_cmd_q_len_) to be consistent with the backpressure blocking logic in ParseRedis (lines 1176-1189). Currently only checking dispatch_q_.size() which may cause premature notifications when the intrusive list still has many pending commands.

    if (!qbp.IsPipelineBufferOverLimit(stats_->dispatch_queue_bytes, dispatch_q_.size()) ||
        dispatch_q_.empty()) {
      qbp.pipeline_cnd.notify_all();  // very cheap if noone is waiting on it.