rpc: direct server-to-server tensor transfer for a layer split

[danielhanchen]

What this does

A layer split over several RPC devices used to move every hidden state through the host that runs
the scheduler. ggml_backend_sched_compute_splits has no direct path between two RPC buffers on
different endpoints (ggml_backend_rpc_buffer_cpy_tensor returns false as soon as the sockets
differ), so each stage boundary fell back to ggml_backend_tensor_copy: a GET_TENSOR into a
host allocation on the coordinator, then a SET_TENSOR back out. Two transfers, a synchronize,
and the hidden state of every boundary crossing the coordinator's memory.

With this change the coordinator still issues every control command, but for a boundary between
two remote servers it issues one command instead of two and the data goes straight from one
server to the other. P stages form a ring: the coordinator sends the embeddings to stage 1,
stage 1 sends its output to stage 2, and so on back to the device that holds the output layer.

Protocol

Minor version 2 -> 3. Every existing command byte and struct is unchanged, so old clients and
servers interoperate for everything else, and the new commands are only used after the HELLO
version check reports minor >= 3 on both endpoints.

• RPC_CMD_COPY_TENSOR_TO is sent to the server that holds the source tensor. Payload is
  | src rpc_tensor | dst rpc_tensor | size | endpoint_len | destination endpoint |. The source
  server validates the source region exactly as RPC_CMD_GET_TENSOR does (deserialize_tensor
  plus the buffer range check), reads it, opens or reuses a connection to the destination
  endpoint using the same connect and HELLO negotiation a client uses (so RDMA when both rails
  allow it and TCP otherwise), and pushes the data as an ordinary RPC_CMD_SET_TENSOR. Nothing
  about the destination pointer is trusted on the source side: the destination applies its own
  deserialize_tensor and buffer range checks, the same ones it applies to a client.
• RPC_CMD_PEER_BARRIER is an empty request with a one byte response. A connection is served
  strictly in order, so its response proves the SET_TENSOR that preceded it has completed.
  The source server waits for it before answering the client, which is the ordering guarantee:
  the destination cannot compute before the write has landed, because the scheduler only sends
  RPC_CMD_GRAPH_COMPUTE to the destination after RPC_CMD_COPY_TENSOR_TO has returned.

Connections to other servers are cached per destination endpoint and closed when the client that
asked for them disconnects. Recoverable failures (destination unreachable, destination too old,
write rejected) come back as result = 0 and the client falls back to the previous path; only a
malformed request or an out of bounds source closes the connection.

Serving several connections at once is what this needs, so a server now runs one thread per
connection over a shared buffer registry and a shared execution mutex; a session still owns and
frees only the buffers it allocated. The execution mutex is released before the peer connection
is used, so a ring of servers cannot deadlock on each other.

Client side, the RPC backend's cpy_tensor_async takes the RPC to RPC case when the two
endpoints differ. Two devices of one server keep the server-local RPC_CMD_COPY_TENSOR, and
every other combination returns false so the previous path runs unchanged. GGML_RPC_P2P=0
forces the old path for A/B.

Commands and bytes per decode step

Client side counters (GGML_RPC_STATS=1), decode window, Qwen3.8-27B UD-Q4_K_XL, 32 rows,
three stages (--device RPC0,RPC1,CUDA0 -sm layer --tensor-split 1,1,1), so one RPC to RPC
boundary per step. The hidden state is 32 x 5120 x 4 B = 655360 B.

per decode step	hub path	direct path
GET_TENSOR	2.00	1.00
SET_TENSOR	12.00	11.00
COPY_TENSOR_TO	-	1.00
total commands	16.00	15.00
request bytes from the coordinator	1 382 840	727 488
response bytes to the coordinator	~1 310 720	~655 360

One GET_TENSOR plus one SET_TENSOR become one COPY_TENSOR_TO per boundary, and 1.31 MB per
step stops crossing the coordinator: 2.69 MB of RPC traffic per step becomes 1.38 MB, a 49
percent reduction. For P stages the saving is (P - 2) boundaries per step, so it grows with the
number of nodes while the coordinator's work stays flat.

CPU-only harness (tests/cpu/rpc_p2p.sh, three local rpc-servers on the CPU backend,
--device RPC0,RPC1,RPC2 -sm layer, the process host doing the embedding lookup), single stream,
exact per-step counts over the decode window:

per decode step, 3 RPC devices	hub	direct
GET_TENSOR	2.01	0.00
SET_TENSOR	18.99	17.01
COPY_TENSOR_TO	-	1.99
SET_TENSOR bytes	33 375	16 240

Two RPC to RPC boundaries, two COPY_TENSOR_TO, and the SET_TENSOR payload halves.

Correctness

Greedy output over five prompts, 48 tokens, temperature 0, top_k 1, on the CPU-only harness:
md5 177dc61e0703eba3bdaf7bf1131f0458 for all four arms - three stage and two stage,
GGML_RPC_P2P=0 and GGML_RPC_P2P=1 - which is the same md5 the existing RPC harnesses record.
tools/server/tests: 368 passed, 6 skipped, 199 deselected.

Numbers on two GB10 nodes

Qwen3.8-27B UD-Q4_K_XL, 32 concurrent closed loop clients, npp 128 / ntg 256, 64 requests,
-c 16384 (16386 for the three group cells), --cache-ram 0, -fa on, -t 6. Local CUDA0 holds
the last third and the output layer; two ggml-rpc-server processes on the peer (ports 50052 and
50053, one GPU between them) hold the first two thirds, --tensor-split 1,1,1.

cell	arm	position in round	tok/s	TPOT ms	TTFT med s	util local	util peer	clocks.sm local / peer	tmax local / peer
h1_a	hub, N=1	1	101.52	292.0	6.55	34.1	58.5	2401 / 2445	55 / 69
d1_a	direct, N=1	2	99.06	298.4	5.98	32.9	59.6	2398 / 1804	58 / 68
h3_a	hub, 3 groups	3	87.92	342.7	5.45	46.4	92.4	2396 / 2466	60 / 74
d3_a	direct, 3 groups	4	75.54	344.7	6.31	42.7	89.0	2392 / 1856	63 / 75
d3_b	direct, 3 groups	1	83.38	359.4	6.47	43.8	91.8	2393 / 1856	65 / 65
h3_b	hub, 3 groups	2	77.31	355.3	7.22	45.4	90.3	2392 / 1690	67 / 58
d1_b	direct, N=1	3	96.98	302.4	6.33	31.3	60.2	2392 / 1690	63 / 56
h1_b	hub, N=1	4	98.18	300.7	6.33	30.6	59.6	2392 / 1690	61 / 54

Two stage reference on the same binaries, --device RPC0,CUDA0, one rpc-server on the peer:

cell	tok/s	TPOT ms	TTFT med s	util local	util peer	clocks.sm local / peer
N=1	101.13	293.8	5.34	46.1	47.6	2389 / 2438
N=2	143.22	208.1	4.92	85.0	84.4	2400 / 2268

Reading these honestly: throughput does not move. The peer's mean SM clock falls monotonically
through each round (2466, 1856, 1690 MHz), so the first cell of a round is always the fastest one,
and the ranking of the two arms flips when the order is reversed: hub is ahead by 16 percent when
it runs first (round a) and behind by 8 percent when the direct arm runs first (round b). Both
arms are inside that band, so this pair and this topology do not separate them on tok/s. That is
expected here, because the two remote stages share one GPU on one node: the boundary the change
removes from the coordinator becomes a loopback transfer on the peer, and nothing is freed on the
one link that matters. What is unambiguous is the protocol accounting above, which is what scales:
for P stages on P nodes the change removes (P - 2) hidden state round trips per step from the
coordinator's link and memory, so the coordinator stops being a per-boundary relay and the wire
cost of a stage boundary no longer depends on where the scheduler runs.

Also visible in the table: three stages over the same two GPUs is worse than two stages
(143.2 tok/s at two stages with two groups against 77 to 88 at three stages with three groups),
so a third stage should only be added with a third node.

Known limitation, not fixed here: RPC_CMD_COPY_TENSOR_TO blocks the source server's connection
thread for the whole transfer plus the barrier, so with several pipeline groups over one
connection the groups serialise behind it. Moving the transfer to a per-destination worker and
letting the destination enforce the ordering with a sequence number, instead of the client
waiting for the acknowledgement, would remove that; it needs a second command on the destination
and another measurement window.

Non-RPC workloads

Every line of this change is in ggml/src/ggml-rpc/:

 ggml/include/ggml-rpc.h         |   2 +-
 ggml/src/ggml-rpc/ggml-rpc.cpp  | 325 +++++++++++++++++++++++++++++++-----
 ggml/src/ggml-rpc/transport.cpp |   2 +-

ggml-backend.cpp, the scheduler, and the CUDA, Metal, Vulkan, CPU and HIP backends are not
touched, so a build with -DGGML_RPC=OFF contains none of it and libggml-base and libllama
gain no new behaviour when the RPC backend is not loaded. Checks on one GB10:

• -DGGML_RPC=OFF -DGGML_CUDA=ON configures and builds clean.

• test-backend-ops -b CUDA0: 13572/13572 tests passed, OK, on this branch and on the base, same
  count.

• Greedy generation on one GPU with no --rpc, three prompts, 64 tokens, temperature 0, top_k 1:
  md5 1f99f8da109f7a456073901232e1f014 on this branch and on the base, byte identical.

• llama-batched-bench on one GPU with no --rpc, npp 512, ntg 128, npl 1 / 8 / 32, run as a
  base / new / base bracket in one window (whole run tok/s):

  npl	base a	new	base b
  1	55.32	53.72	53.75
  8	232.26	212.29	211.87
  32	347.84	339.19	338.14

  The first run of the bracket is the fastest in every row (a cold GPU at the start of the
  window); the new binary matches the second base run to within 0.2 percent at npl 1 and 8 and
  sits between the two base runs at npl 32.