Skip to content

[TransferEngine] Add an optional coro_rpc P2P handshake - #5085

Open
staryxchen wants to merge 5 commits into
mainfrom
te-coro
Open

staryxchen wants to merge 5 commits into
mainfrom
te-coro

Conversation

@staryxchen

@staryxchen staryxchen commented Oct 30, 2026 •

Copy link
Copy Markdown
Collaborator

Description

In P2P mode, QP handshake, segment metadata, notify, and probe all enter the same handshake daemon. SocketHandShakePlugin serves each accepted connection on the accept thread: it reads the request, runs the callback, writes the reply, and waits for the peer to close. One stuck connection holds every handshake behind it. Once the listen backlog is full, new connect() calls time out.

#5051 hit this in a disaggregated prefill/decode deployment. A rank's handshake listener stopped serving. Peers first saw 5 s read timeouts, then connect() returned ETIMEDOUT after the backlog filled. Every transfer that needed a new connection to that rank failed until the process was restarted. The callback that blocked was an EFA passive handshake waiting on its endpoint lock. The same accept thread stalls in the same way for any of these:

  • A connection that sends nothing holds the thread for SO_RCVTIMEO (5 s). Port scans and health checks do this.
  • A peer that sends a frame header and then one byte every few seconds restarts SO_RCVTIMEO on each read. readFully() checks its 300 s budget only between reads.
  • A valid header followed by silence also holds the thread for 5 s.
  • A callback that blocks holds the thread for as long as it blocks.
  • An exception thrown while handling a request escapes the listener thread and calls std::terminate. jsoncpp throws when a field has the wrong JSON type, for example a notify whose "name" is an object. #1369 reported a Json::LogicError from this thread.

#4451 bounded the reply write with SO_SNDTIMEO. The serial accept loop is still there. #5051 keeps the handwritten server: the accept thread only calls accept, each connection gets a short-lived thread, and the daemon adds a connection cap and a 5 s read deadline. That still means maintaining the framing, the timeouts, and the thread lifetime. A fixed worker pool of 8 still stalls later handshakes when those 8 connections are slow.

This PR adds coro_rpc, which TENT already uses as its control plane, as an optional handshake. The default stays the existing socket server:

  • The accept loop stays inside coro_rpc. Connection and Metadata callbacks run on a dedicated pool of 2 threads owned by the handshake server, so a slow RDMA handshake does not block reads on other connections and does not start ylt's process-wide blocking pool (one thread per CPU). Probe and Notify stay on the IO threads.
  • One connection carries one in-flight RPC. Probe and Notify use another connection and do not queue behind a stuck Connection. Each peer has a hard cap of 128 in-flight handshakes and 128 idle connections (MC_HANDSHAKE_MAX_ACTIVE_PER_PEER, MC_HANDSHAKE_MAX_IDLE_PER_PEER). A further call fails immediately with ERR_TOO_MANY_REQUESTS. An idle connection is closed on return once the peer is already at the idle cap. A sweep about once a second drops connections that have been idle for MC_HANDSHAKE_IDLE_TTL_SECONDS (default 60), including peers that are never contacted again.
  • The client uses a process-wide pool of 2 IO threads, not the global executor sized to the CPU count.
  • The length in a request header is checked before the payload is read. An oversized header closes the connection.
  • An unresolved host returns ERR_DNS, the same code as the socket plugin. A refused connect stays ERR_SOCKET.
  • Exceptions thrown from callbacks stay inside the RPC boundary and become error returns. They do not abort the process.
  • The default stays socket, the length-prefixed JSON server. MC_HANDSHAKE_TRANSPORT=coro selects coro_rpc. Both peers must set the same value. An unset value keeps the framing already deployed, so a rolling upgrade does not mix the two protocols.
  • The implementation lives under mooncake-transfer-engine/src/coro_handshake/. include/handshake_rpc.h only exposes the factory functions.

This PR does not take these parts of #5051: refusing connections past a cap right after accept, a poll-based deadline that requires the whole read to finish within 5 s, or that PR's silent-connection, dribbling-byte, and malformed-request tests. The coro read timeout is a total budget for one request, 60 s by default (MC_HANDSHAKE_RPC_TIMEOUT_SECONDS), not a 5 s timer that resets per byte.

tebench returns an error when openSegment yields no descriptor, instead of dereferencing it. handshake_perf compares cold, warm, and concurrent latency. It is not a unit test. The change is about 2500 lines and has no separate RFC.

Suggested Merge Priority

  • P0 — Urgent: Critical security vulnerability, data loss/corruption, or widespread production outage requiring immediate action
  • P1 — High: Significant correctness or stability issue in existing functionality, or a regression blocking users without a practical workaround
  • P2 — Normal: Routine bug fix, validated performance/compatibility improvement, or new feature for a confirmed use case
  • P3 — Low: Experimental feature, optional enhancement, refactoring, or cleanup with no urgent user impact

Module

  • Transfer Engine (mooncake-transfer-engine)
  • Mooncake Store (mooncake-store)
  • Mooncake Conductor (mooncake-conductor)
  • Reshard (mooncake-reshard)
  • Mooncake EP (mooncake-ep)
  • Mooncake PG (mooncake-pg)
  • Integration (mooncake-integration)
  • P2P Store (mooncake-p2p-store)
  • Python Wheel (mooncake-wheel)
  • Common (mooncake-common)
  • Mooncake RL (mooncake-rl)
  • CI/CD
  • Docs
  • Other

Type of Change

  • Bug fix
  • New feature
  • Refactor
  • Breaking change
  • Documentation update
  • Performance improvement
  • Other

How Has This Been Tested?

Built with -DWITH_STORE=OFF -DUSE_CUDA=OFF -DUSE_ETCD=OFF against stock yalantinglibs 7801bc9. The numbers below are from the current tree after rebasing onto origin/main (522b716e). Loopback only. No RDMA run and no multi-node run.

Test commands:

./coro_handshake_test
./transfer_metadata_test --gtest_filter='-TransferMetadataTest.*'
MC_HANDSHAKE_TRANSPORT=coro ./transfer_metadata_test --gtest_filter='-TransferMetadataTest.*'
MC_FORCE_TCP=1 MC_HANDSHAKE_TRANSPORT=coro ./tebench --backend=classic --metadata_type=p2p ...
MC_FORCE_TCP=1 MC_HANDSHAKE_TRANSPORT=socket ./tebench --backend=classic --metadata_type=p2p ...
MC_HANDSHAKE_TRANSPORT=coro ./handshake_perf server
MC_HANDSHAKE_TRANSPORT=coro ./handshake_perf client <port> cold|warm|concurrent
MC_HANDSHAKE_TRANSPORT=socket ./handshake_perf server
MC_HANDSHAKE_TRANSPORT=socket ./handshake_perf client <port> cold|warm|concurrent

Test results:

  • Unit tests pass
  • Integration tests pass (if applicable)
  • Manual testing done (describe below)

coro_handshake_test: 23 passed. transfer_metadata_test: 23 passed with the default socket transport and 23 passed with MC_HANDSHAKE_TRANSPORT=coro. The etcd TransferMetadataTest fixture was not run because this build has USE_ETCD=OFF. tebench against a missing target (127.0.0.1:1) exits 1 instead of segfaulting.

tebench, loopback TCP, classic backend, P2P metadata, MC_FORCE_TCP=1, --check_consistency=true, 2 seconds per case. Both transports exited 0. Runs alternate socket then coro. 64KB at 1 thread moves around between runs; the two transports overlap.

Case socket coro
4KB, 1 thread 0.14 GB/s, 30 µs; 0.13 GB/s, 33 µs 0.10 GB/s, 41 µs; 0.10 GB/s, 42 µs
64KB, 1 thread 0.72 / 0.40 / 0.71 GB/s 0.76 / 0.57 / 0.88 GB/s
64KB, 2 threads 1.39 GB/s, 95 µs; 1.34 GB/s, 98 µs 1.38 GB/s, 95 µs; 1.91 GB/s, 69 µs

Handshake is a one-time cost here. After the segment is open, both sides use the same TCP data path. During these runs the coro target stayed at 9 threads. An earlier build posted Connection and Metadata callbacks to ylt's default blocking pool and the target grew to about 391 threads, one per CPU. That pool is no longer used.

handshake_perf, loopback, one run, zero failures. The socket plugin has no connection pool, so its cold and warm rows both open a new TCP connection per call. Cold is 100 calls. Warm is 1000 calls.

Cold (a new client each call):

Operation coro p50 coro p99 socket p50 socket p99
Connection 115 µs 491 µs 46 µs 65 µs
Metadata 108 µs 120 µs 45 µs 52 µs
Notify 94 µs 110 µs 45 µs 52 µs
Probe 91 µs 116 µs 44 µs 53 µs

A cold coro call builds a coro_rpc_client and opens a framed RPC connection. The socket path is one connect plus one write and one read. The Connection p99 of 491 µs is the cost of opening that first coro connection. An earlier cold Connection p99 of about 25 ms came from the first use of a process-wide executor sized to the CPU count. The client now uses a 2-thread pool, and that tail is gone.

Warm (one client):

Operation coro p50 coro p99 socket p50 socket p99
Connection 52 µs 60 µs 43 µs 51 µs
Metadata 52 µs 59 µs 43 µs 50 µs
Notify 38 µs 46 µs 43 µs 50 µs
Probe 36 µs 44 µs 42 µs 49 µs

Notify and Probe run on the IO thread and reuse a pooled connection, so they beat the socket path. Connection and Metadata are posted onto the handshake server's 2-thread pool. That hop is what keeps them a little above Notify and Probe, and a little above the socket path. The hop is there so a long RDMA handshake does not block reads on that connection.

Concurrent Probe (200 calls per worker):

Workers coro p50 coro p99 coro calls/s socket p50 socket p99 socket calls/s
1 39 µs 49 µs 23k 44 µs 57 µs 22k
8 66 µs 105 µs 112k 186 µs 217 µs 42k
32 213 µs 282 µs 142k 1086 µs 1160 µs 29k

One pooled connection carries one in-flight RPC, so a slow Connection cannot head-of-line block a Probe. These concurrent numbers sit in the same band as the previous three-run median (32 workers, coro p99 about 300 µs, about 140k calls/s). An idle cap of 2 closed the extra connections after each burst and pushed that p99 to about 16 ms. The cap is now 128. The socket server still handles one new TCP connection at a time on its accept thread.

Checklist

  • I have performed a self-review of my own code
  • I have formatted my code using ./scripts/code_format.sh
  • I have run pre-commit on the files changed in this PR and all hooks pass
  • I have updated the documentation (if applicable)
  • I have added tests to prove my changes are effective
  • For changes >500 LOC: I have filed an RFC issue

AI Assistance Disclosure

  • No AI tools were used
  • AI tools were used (specify below)

Cursor wrote the coro handshake server, the tests, handshake_perf, and the docs update, and ran the local tests. The human submitter needs to review and defend every changed line.

@github-actions github-actions Bot added documentation Improvements or additions to documentation run-ci Transfer Engine size/L 1001+ changed lines (additions + deletions) labels Oct 30, 2026
@staryxchen staryxchen changed the title [TransferEngine] Default the P2P handshake to coro_rpc [TransferEngine] Add an optional coro_rpc P2P handshake Oct 30, 2026
@staryxchen
staryxchen marked this pull request as draft October 30, 2026 06:03
@staryxchen
staryxchen marked this pull request as ready for review October 30, 2026 08:25

@UNIDY2002 UNIDY2002 left a comment

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks! We tested the coro-based P2P handshake with Mooncake PG, including normal and failure recovery paths. It works as expected and appears compatible with PG.

@staryxchen

Copy link
Copy Markdown
Collaborator Author

Thanks! We tested the coro-based P2P handshake with Mooncake PG, including normal and failure recovery paths. It works as expected and appears compatible with PG.

Thanks for the feedback. After this lands, we would like PG to switch to the coro handshake. It is the more robust: a slow callback no longer blocks every other handshake on the accept thread.

staryxchen and others added 5 commits November 3, 2026 07:18
openSegment can return no descriptor when the peer is down. Sorting that
null descriptor crashed the initiator.

Signed-off-by: staryxchen <staryxchen@tencent.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
Socket stays the default framing. Set MC_HANDSHAKE_TRANSPORT=coro on
both peers to serve Connection, Metadata, Notify, and Probe with
coro_rpc. The server bounds frame size, in-flight bytes, and per-peer
connections, and an unresolved host returns ERR_DNS like the socket plugin.
Connection and Metadata callbacks run on a dedicated pool of 2 threads.

Signed-off-by: staryxchen <staryxchen@tencent.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
The PR format check rewrapped several assertions in
coro_handshake_test.cpp.

Signed-off-by: staryxchen <staryxchen@tencent.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
hipify compiles copies of the .cpp files from the build tree. The
sibling headers stay in the source tree, so that directory is now on
the include path.

Signed-off-by: staryxchen <staryxchen@tencent.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
The upgraded ylt checks a response header before allocating, and can
refuse a socket before constructing a connection. The handshake client
sets both limits to MC_HANDSHAKE_MAX_LENGTH. The server closes a new
socket once MC_HANDSHAKE_MAX_CONNECTIONS live connections are open.

Signed-off-by: staryxchen <staryxchen@tencent.com>
Co-authored-by: Cursor <cursoragent@cursor.com>

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

documentation Improvements or additions to documentation run-ci size/L 1001+ changed lines (additions + deletions) Transfer Engine

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants