Redesign pipelining support in Dragonfly

# Epic

Currently, the io/loop in dragonfly looks like this:

```mermaid
---
title: DflyConn Request-Response Loop
---
graph TD
A([socket.FiberRead]);
A -- Data Received --> C[cmd.dispatch];
C -- Response Generated --> D([socket.FiberWrite]);
D -- Write Complete --> A;
```

This loop runs in the context of the `DflyConn` fiber that manages the connection.
Unfortunately, it introduces at least 3 preemption points (`cmd.dispatch` has 1 or more), which in turn causes visible latency per command that aggregates for pipelines.
To address this we introduced an additional `Dispatch` fiber that is responsible for executing commands, while `DflyConn` reads and parses data from the socket. Both fibers communicate via `dispatch_q` using Producer/Consumer relationship.

The logic that decides whether use dispatch_q_ is in `Connection::DispatchSingle`. When `DflyConn` recognizes a possible pipeline (there is more data in the input buffer after parsing the current command) it switches to the following mode:

```mermaid
---
title: DflyConn Read/Dispatch Loop
---
graph TD
A([socket.FiberRead]);
A -- Data Received --> C["dispatch_q.push(cmd)"]
C --> A
```

```mermaid
---
title: DispatchFiber Loop
---
graph TD
D([dispatch_q.pop]) --> E[cmd.dispatch];
E -- Response Generated --> F([socket.FiberWrite]);
F -- Write Complete --> D;
```

so we introduced a fiber to speed-up reading from the socket while pipeline is executing, but `DispatchFiber` is still slow as `cmd.dispatch` preempts and issues one or more hops to shard threads.

To alleviate this we introduced `Connection::SquashPipeline()` and `MultiCommandSquasher` that can take a bunch of (simple) single shard commands and squash them into a single transaction, so that say 30 commands in the pipeline will be executed in a single hop. This works reasonably well but requires lots of cpu and still has not the most efficient io pattern as it must issue k commands and then wait for all of them to return and then issue a blocking reply to the socket.

To summarize:
1. We have a complicated two-fiber design to handle pipelines (and our of order replies like pubsub)
2. Our `command.dispatch` is synchronous, which warrants workarounds like `MultiCommandSquasher`
3. On the socket level we use `FiberRead` call, that requires a dedicated input buffer in advance. i.e. 1000 connections blocked on read, potentially reading a 64kb blob will require a 64MB of reserved space even if there are 0 in-flight requests.
4. `FiberWrite` call is fiber-blocking. Which means that if the socket is blocked on write - the whole flow is blocked. Moreover, we can not control the amount of pending writes, we can only close the connection externally to unblock the blocked fiber.
5. The pipelining efficiency is sub-optimal. Can be easily reproduced by load-testing a single connection `dfly_bench --ratio 0:1 --pipeline=30 -c 1 --proactor_threads=1 -n 2000000` on dragonfly vs valkey, for example.

## Suggestions

Each item is a task that requires careful execution and planning.

1. Stop using `FiberRead` and switch to OnRecv interface (https://github.com/romange/helio/pull/480). As `OnRecv` separates read notification from reading data, we can reduce reliance on per connection io_buf.
`OnRecv` can be called at any time when a fiber is preempted. Meaning that it will be able to read data during its other preemption points like `cmd.dispatch`.
2. Following (1) we can merge `DispatchFiber` and `DflyConn` together as now we do not need a dedicated fiber for reading from socket and adding a parsed command into dispatch_q.
3. Prepare for asynchronous cmd.dispatch executions that separate command execution from sending replies. We should remove assumptions in facade that cmd.dispatch is synchronous: introduce `ParsedCommand` on-heap entity that can be passed to `cmd.dispatch`. Similarly command arguments should always be allocated on head like we do today with dispatch_q. Basically we should make sure that if `cmd.dispatch` returns before command finishes executing, facade will work as expected.
4. Start using posix `socket.send` instead of socket.FiberWrite - the iouring model of socket.send is suboptimal, as the underlying send still needs to copy the passed iovec to networking buffers, therefore, calling `socket.send` that synchronously pushes data **if possible** gives us the most control: a) we can identify pending writes, b) we do not need to keep buffers if send succeeds and sends the iovec, c) it removes preemption point (though I must say the latency of that is almost negligible, at list for small responses.)
5. Devise a plan on how we can gradually rollout async implementation of commands. The goal is to deprecate MultiCommandSquasher for pipelines (and later for multi/exec transactions). I believe that changing 30-50 most common commands will be enough.

The POC that demonstrates the feasibility of this approach is located here:
https://github.com/romange/midi-redis/tree/Pr2

---

## Roadmap & Milestones

To tackle this safely and iteratively, we are breaking the V2 rollout into milestones. A detailed optimization plan is tracked internally (`dragonflydb/dataplane-private#214`).

*(Note: We aim to complete Milestone 3 for the 1.40 release. The completed tasks listed below highlight major architectural milestones but do not represent an exhaustive list of all minor commits).*

### Milestone 1: Memcached on IoLoopV2 (Completed)
**Goal:** Prove the single-fiber event loop architecture in production using the Memcached protocol.
- [x] **Enable IoLoopV2 for Memcached by default (#6700):** Flipped the experimental flag, activating the new async I/O loop. Achieved significantly improved throughput via async command execution and deferred replies for major string commands.

### Milestone 2: RESP IoLoopV2 - Correctness & Functional Parity (Current)
**Goal:** Ensure correctness, single-fiber async safety, starvation prevention, and baseline performance for RESP connections before enabling by default.
- [x] **Protocol-specific configuration (#7424):** Replaced the experimental flag with `enable_memcache_io_loop_v2` and `enable_resp_io_loop_v2`.
- [x] **Backpressure handling & fiber parking (#7018):** Park the IoLoopV2 fiber safely via `io_event_.await()` when pipeline queues exceed limits.
- [x] **Connection Migration (#7143):** Enabled `CLIENT MIGRATE` for V2, safely re-arming `io_uring` multishot post-migration.
- [x] **Aggressive Syscall Reduction (#7213):** Removed unconditional flushing, deferring flushes to explicit event boundaries. Resulted in a ~94% syscall reduction under heavy pipeline saturation.
- [x] **Control-path starvation prevention (#7234):** Bounded `dispatch_q_` processing so PubSub floods can no longer starve data-path command execution.
- [x] Baseline PubSub Performance: Wakeup batching to eliminate the 2x syscall regression (#7437 pending)
- [x] Cloud Benchmark: Async-supported Commands & PubSub (V1 vs V2) - https://github.com/dragonflydb/dragonfly/issues/7443
- [x] Cloud Benchmark: Commands without async dispatch support e.g., ZADD (V1 vs V2) - **https://github.com/dragonflydb/dragonfly/issues/7443**
- [x] **Implement subscriber-side reply batching in ProcessControlMessages (PubSub)**: https://github.com/dragonflydb/dataplane-private/issues/232
### Milestone 3: RESP IoLoopV2 - Core Performance Optimization and Async Command Coverage Expansion (Target: v1.40)

**Goal:** Implement clear, well-scoped performance wins to eliminate the remaining sequential bottlenecks.
- [x] SquashPipeline:** Port V1's aggressive sync-command squashing to V2.
- [x] **Eager Parsing:** Parse eagerly in `NotifyOnRecv` using a shared `IoBuf` to ensure the fiber wakes to a full batch of commands. Research/Benchmarks: https://github.com/dragonflydb/dataplane-private/issues/242
- [ ] **Epoch Yielding:** Implement smart yielding heuristics to allow pipelines to coalesce under load (depends on Eager Parsing).
- [ ] **PubSub batching:** Implement subscriber-side reply batching and deferred fan-out to close the remaining p=1 PubSub performance gaps.
- [ ] **Async Command Coverage Expansion (Parallelizable):** Port remaining sync-bound command families to the coroutine-based async dispatch path (`SetAsyncHandler` / `SupportsAsync()`). While basic commands (like `SET`, `GET`, `DEL`, `MGET`, and `INCR`) are already optimized, the rest currently fall back to sequential head-of-line blocking. Expanding async coverage unlocks V2's ability to process these commands concurrently across shards without fiber blocking. This can be done as a parallel work (one engineer per command family) categorized by traffic frequency:
* **P0 (High Traffic Keyspace):** `MSET`, `EXISTS`, `EXPIRE`, `TTL`, `PTTL`, `PERSIST`, `PEXPIRE`, `EXPIREAT`, `PEXPIREAT`.
* **P1 (Hashes, Lists, and Sets):** `HSET`/`HGET`/`HMSET`/`HMGET`/`HDEL`/`HGETALL`/`HINCRBY`/`HEXISTS`/`HLEN`, `LPUSH`/`RPUSH`/`LPOP`/`RPOP`/`LRANGE`/`LLEN`/`LINDEX`, `SADD`/`SREM`/`SMEMBERS`/`SISMEMBER`/`SCARD`.
* **P2 (Sorted Sets):** `ZADD`/`ZRANGE`/`ZRANGEBYSCORE`/`ZINCRBY`/`ZREM`/`ZCARD`/`ZSCORE`/`ZCOUNT`.
* **P3 (Generic Keyspace):** `TYPE`, `RENAME`, `EXPIRETIME`, `PEXPIRETIME`, `SRANDMEMBER`.
- [ ] **Async Command Coverage Expansion (Parallelizable):** Port remaining sync-bound command families to the coroutine-based async dispatch path (`SetAsyncHandler` / `SupportsAsync()`). While basic commands (like `SET`, `GET`, `DEL`, `MGET`, and `INCR`) are already optimized, the rest currently fall back to sequential head-of-line blocking. Expanding async coverage unlocks V2's ability to process these commands concurrently across shards without fiber blocking. This can be done as parallel work (one engineer per command family) categorized by traffic frequency:

* **P0 (High Traffic Keyspace):** `MSET`, `EXISTS`, `EXPIRE`, `TTL`, `PTTL`, `PERSIST`, `PEXPIRE`, `EXPIREAT`, `PEXPIREAT`.
* **P1 (Hashes, Lists, and Sets):** `HSET`/`HGET`/`HMSET`/`HMGET`/`HDEL`/`HGETALL`/`HINCRBY`/`HEXISTS`/`HLEN`, `LPUSH`/`RPUSH`/`LPOP`/`RPOP`/`LRANGE`/`LLEN`/`LINDEX`, `SADD`/`SREM`/`SMEMBERS`/`SISMEMBER`/`SCARD`.
* **P2 (Sorted Sets):** `ZADD`/`ZRANGE`/`ZRANGEBYSCORE`/`ZINCRBY`/`ZREM`/`ZCARD`/`ZSCORE`/`ZCOUNT`.
* **P3 (Generic Keyspace):** `TYPE`, `RENAME`, `EXPIRETIME`, `PEXPIRETIME`, `SRANDMEMBER`.
* **P4 (Single-Key Commands, bulk sweep):** All remaining single-key commands across all families (e.g., `OBJECT`, `TOUCH`, etc.). These hit only one shard so `CanRunInlined()` may already skip the cross-thread hop — the throughput benefit is smaller than P0–P2, but porting them unifies the execution model and removes edge-case sync fallbacks. Suitable for a single bulk PR per family.
* **Out of Scope:** Blocking commands (`BLPOP`, `BRPOP`, `WAIT`, `WAITAOF`), admin/config (`CONFIG`, `DEBUG`, `COMMAND`, `INFO`), scripting (`EVAL`, `EVALSHA`), and cluster/replication commands. These have fundamentally synchronous semantics or no throughput relevance.
* *Validation:* Benchmarking an enabled command with V2 at deep pipelines (p=500) should show immediate cross-shard parallelism gains over the original sync path.

### Milestone 4: TLS, Buffer Rings, & Conditional Optimization
**Goal:** Address remaining high-complexity tasks and architectural gaps. Priorities and scope will be dictated entirely by the Milestone 2 benchmark results.
- **TLS Support:** Add full TLS compatibility for IoLoopV2.
- **io_uring Buffer Rings:** Integrate kernel-provided buffer pools to reduce per-connection memory overhead.
- **Investigate Heuristics:** Evaluate redundant wakeup deduplication and soft backpressure limits.

This is a broad architectural epic rather than a single change. Start with Connection::DispatchSingle, SquashPipeline, and MultiCommandSquasher, then review the OnRecv proposal and the async-handler references; run the listed dfly_bench pipeline command for a baseline. Done depends on selecting and completing one roadmap milestone with its stated correctness and benchmark criteria.

Written by the indexing model from the issue text.