mirrors/valkey
0
0 Fork 0
mirror of https://github.com/valkey-io/valkey.git synced 2026-10-28 08:24:42 +08:00
Code Activity
14,221 Commits 45 Branches 61 Tags
unstable
Commit Graph
3 Commits
Author SHA1 Message Date
Harkrishn Patro 2f3bcead22 Cluster Bus IO offload (#3438) 
Fixes: #3361

Cluster bus read, write, and **inbound** TLS accept work can run on the
shared
I/O thread pool, while all cluster state mutation stays on the main
thread. The
outbound TLS handshake (`SSL_connect`) is not offloaded — see Follow-Up.

Jobs flow through:
- `io_shared_inbox` (SPMC): main thread -> I/O workers
- `io_shared_outbox` (MPSC): I/O workers -> main thread

If the pool is inactive or enqueue fails, the caller falls back to
synchronous
I/O on the main thread and increments
`cluster_io_main_thread_fallbacks`.
Dispatch requires an already-active pool, so on a lightly loaded node
most
cluster bus I/O legitimately takes this path.

### Architecture

```text
+---------------------------+       +-------------------------------+       +---------------------------+
|       Main Thread         |       |        Shared Queues          |       |     I/O Thread Pool       |
|                           |       |                               |       |                           |
| clusterReadHandler        |------>| io_shared_inbox (SPMC)        |------>| clusterReadJob            |
| clusterWriteHandler       |       |                               |       | clusterWriteJob           |
| clusterAcceptHandler      |       |                               |       | clusterAcceptJob          |
|                           |       +-------------------------------+       +---------------------------+
|                           |
| processIOThreadsResponses |<------ io_shared_outbox (MPSC) <------ sendToMainThread(...)
| clusterHandle*Completion  |
| clusterProcessPacket      |
| freeClusterLink           |
+---------------------------+
```

### Key Design Decisions

1. Reuse the existing shared SPMC/MPSC queues; no cluster-specific
threading primitives.
2. I/O threads do transport work only: `connRead`, `connWrite`,
`connAccept`, framing, and result publication.
3. Main thread alone applies cluster packets and mutates cluster state.
4. Write offload uses a single canonical `send_msg_queue` plus a
snapshot boundary:
   `io_last_send_block`, `io_head_offset`, and `io_nodes_sent`.
5. Read offload publishes the complete-packet prefix via
`io_complete_bytes` and
   `io_complete_packets`.
6. Read completion drains the framed prefix in one pass, sliding each
packet to the
front and compacting the tail once. Worker jobs themselves are bounded:
reads at
   `RCVBUF_MAX_PREALLOC`, writes at `NET_MAX_WRITES_PER_EVENT`.
7. Link teardown is deferred with `io_refs` and `async_close`. For
links, `io_refs`
alone guards connection lifetime; the accept path has no link yet and
keeps a
   conn reference.
8. Accept offload is serialized per connection with
`CONN_FLAG_ACCEPT_OFFLOAD_PENDING`,
and `clusterConnAcceptHandler` is installed as `conn_handler` before
dispatch so
   every completion path finishes the accept.
9. `ConnectionOwnerKind` lets generic connection/TLS code distinguish
client-owned
   and cluster-owned connections safely.
10. Reads and writes are mutually exclusive per link. A one-shot yield
(`io_read_deferred`) stops a permanently backlogged send queue from
starving
    reads, since `CONN_FLAG_WRITE_BARRIER` fires writable first.

### Data Flow: Read Path

```text
[MAIN dispatch] clusterReadHandler
    |
    | drain any framed prefix
    | trySendClusterReadToIOThreads()
    v
[QUEUE] io_shared_inbox
    | JOB_REQ_CLUSTER_READ
    v
[IO worker] clusterReadJob
    | connRead() into rcvbuf
    | grow rcvbuf if needed
    | stop at RCVBUF_MAX_PREALLOC
    | frame complete packet prefix
    | set io_complete_bytes / packets
    | set io_result
    v
[QUEUE] io_shared_outbox
    | JOB_RES_CLUSTER_READ
    v
[MAIN completion] clusterHandleReadCompletion
    | set io_read_state = IDLE
    | io_refs--
    | reconcile memory
    | drain framed prefix, compact tail once
    | shrink rcvbuf around partial tail
```

If dispatch returns `C_ERR`, the main thread falls back to
`clusterReadHandler(conn)`.

## Data Flow: Write Path

```text
[MAIN] clusterSendMessage
    | append to send_msg_queue
    | arm write handler (WRITE_BARRIER)
    v
[MAIN dispatch] clusterWriteHandler   (on the writable event)
    | snapshot io_last_send_block / io_head_offset
    | trySendClusterWriteToIOThreads()
    v
[QUEUE] io_shared_inbox
    | JOB_REQ_CLUSTER_WRITE
    v
[IO worker] clusterWriteJob
    | write from queue head
    | stop at io_last_send_block
    | stop at NET_MAX_WRITES_PER_EVENT
    | set io_nodes_sent
    | set io_head_offset
    | set io_result
    v
[QUEUE] io_shared_outbox
    | JOB_RES_CLUSTER_WRITE
    v
[MAIN completion] clusterHandleWriteCompletion
    | set io_write_state = IDLE
    | io_refs--
    | pop sent nodes
    | apply head_msg_send_offset
    | reinstall handler or free on error
```

Important note:
- New messages appended while a write job is in flight stay on
`send_msg_queue`,
but the worker stops at `io_last_send_block`, so those new nodes are
picked up
by a later dispatch. Dispatch happens only on the writable event, so
messages
  queued within one event-loop iteration are batched into a single job.

If dispatch returns `C_ERR`, `clusterWriteHandler(conn)` falls back to
the synchronous `connWrite()` loop.

### Data Flow: Accept Path (TLS)

```text
[MAIN dispatch] clusterAcceptHandler
    |
    | create accepted conn
    | set owner kind
    | allow accept offload
    | install conn_handler
    | trySendClusterAcceptToIOThreads()
    v
[QUEUE] io_shared_inbox
    | JOB_REQ_CLUSTER_ACCEPT
    v
[IO worker] clusterAcceptJob
    | connAccept(conn, NULL)
    v
[QUEUE] io_shared_outbox
    | JOB_RES_CLUSTER_ACCEPT
    v
[MAIN completion] clusterHandleAcceptCompletion
    | clear ACCEPT_OFFLOAD_PENDING
    | finalize conn state / refs
    | connUpdateState() runs conn_handler
    |   (clusterConnAcceptHandler) once the
    |   handshake completes, or re-arms the
    |   TLS event if it needs another round
    | create clusterLink on success
```

Important note:
- TLS retries can re-enter the generic accept offload path.
Cluster-owned
connections are routed back to `trySendClusterAcceptToIOThreads`, and
the
pending flag ensures only one accept job is in flight for that
connection.
`conn_handler` is installed at accept time so a retry that cannot
re-offload
  still finishes the accept rather than orphaning the fd.

If dispatch returns `C_ERR`, the main thread falls back to
`connAccept(conn, clusterConnAcceptHandler)`.

### clusterLink I/O State

```text
IDLE --dispatch read-->  READ_PENDING  --read completion-->  IDLE
IDLE --dispatch write--> WRITE_PENDING --write completion--> IDLE

If freeClusterLink() sees io_refs > 0:
  detach link from node
  remove handlers
  set async_close = 1
  defer final free until the last completion drops io_refs to 0
```

Read and write jobs are mutually exclusive per link. If a read is
skipped because
a write is in flight, the next write dispatch yields one turn so the
read can
proceed.

### Observability

`CLUSTER INFO` reports, all counted on completion:

- `cluster_io_threaded_reads_processed`
- `cluster_io_threaded_writes_processed`
- `cluster_io_threaded_accepts_processed`
- `cluster_io_main_thread_fallbacks`

### Follow-Up

- Read completion applies every framed packet in one pass on the main
thread, so
  a large burst is unbounded main-thread work. A drain budget would need
  `io_complete_bytes` to survive across completions.
- Create separate queue for cluster io jobs to avoid contention with
client io
jobs. This will allow us to prioritize different type of jobs and drain
at
  different rate.
- Offload the outbound TLS handshake (accept). `connTLSAccept` has an
offload hook,
`connTLSConnect` does not. Not a copy of the accept path: the SPMC
request tag
space is full (`JOB_REQ_COUNT == 8`, 3-bit tags), `tls.c` needs a
connect-side
split mirroring `TLSAccept`/`TLSHandleAcceptResult`, and unlike accept a
`clusterLink` already exists so the job must join the
`io_refs`/`async_close`
  machinery.
- Move error classification to the completion — workers should record
raw
`nread`/`nwritten`/`errno` rather than re-deriving a verdict from the
shared
`conn->state`, as the client path already does. This removes the only
obstacle
  to full duplex on plain TCP links.
- Zero-copy drain: making `clusterProcessPacket` offset-aware removes
the last
  per-packet copy.

---------

Signed-off-by: Harkrishn Patro <h_patro@apple.com>
 2026-09-10 15:03:29 +01:00
Harkrishn Patro ead3dc6f29 Add design doc for I/O threads (#4022) 
Document the shared I/O threading infrastructure: SPMC inbox, MPSC
outbox, per-worker SPSC inboxes, job lifecycle, drain semantics, live
reconfiguration.

---------

Signed-off-by: Harkrishn Patro <h_patro@apple.com>
 2026-06-24 01:26:59 +01:00
Jacob Murphy 059a77c429 Add design-docs folder and README. (#3300) 
These are forked from the RFC instructions created by
@zuiderkwast and @hpatro in
https://github.com/valkey-io/valkey-rfc/pulls/1 and
https://github.com/valkey-io/valkey-rfc/pulls/6. It also includes the Atomic Slot Migration design to bootstrap the folder.

---------

Signed-off-by: Jacob Murphy <jkmurphy@google.com>
 2026-03-17 14:56:39 -07:00