Bound blocking I/O on data plane transport sockets

[sanjeevrg89]

The H2H data plane sets TCP_NODELAY and socket buffer sizes and nothing
else, so every send(), recv(), and connect() on a data plane socket
blocks with no deadline. A peer that holds its connection open but stops
reading parks the sender inside writev() for as long as that peer lives.
BlockTransport runs those calls inline on a fixed pool of socket worker
threads, so one unresponsive peer also stalls transfers to every healthy
peer.

Set SO_SNDTIMEO, SO_RCVTIMEO, TCP_USER_TIMEOUT and TCP keepalive on both
the connect and the accept side, and bound connect() with a non-blocking
connect plus poll(). Two flags configure it:

--raiden_transport_io_timeout (default 5m)
--raiden_transport_connect_timeout (default 30s)

The defaults are ceilings on a stalled peer, not latency targets, so a
healthy transfer behaves exactly as before. Setting
raiden_transport_io_timeout to zero restores the previous unbounded
behavior in full.

The two mechanisms cover different failures. SO_SNDTIMEO and SO_RCVTIMEO
bound one blocking call, which catches a peer that stops reading.
Keepalive catches the opposite case, an idle pooled connection whose peer
vanished without sending FIN, where no data is in flight for a send
deadline to expire on. TCP_USER_TIMEOUT replaces the tcp_retries2 default
of roughly fifteen minutes for data already in flight.

SO_SNDTIMEO and SO_RCVTIMEO expire as EAGAIN or EWOULDBLOCK even on a
blocking socket, a case TcpSocketUtil previously treated as unreachable
via DCHECK(!WouldBlock(...)). Those loops now report a deadline as
DEADLINE_EXCEEDED and a kernel connection timeout as UNAVAILABLE, so a
caller can tell a stalled or dead peer from a protocol error. A connect
that runs out of time reports DEADLINE_EXCEEDED; every other connect
failure keeps the historical UNAVAILABLE that the transfer failure metric
labels on.

This bounds the failure, it does not retry it. A failed transfer still
fails its whole plan and recovery stays with the caller.

The control plane already does this in kv_cache/reshard/framed_rpc.cc.
This brings the data plane in line with it.

Validation

• A peer that accepts and never reads now fails the send with DEADLINE_EXCEEDED instead of blocking.
• ConnectToPeer returns a socket that is bounded and back in blocking mode, which send and recv assert on.
• A refused peer still reports UNAVAILABLE.
• A zero io_timeout leaves the socket untouched.
• Tested locally on Linux x86_64 host with GCC toolchain across the following test suites, all passing:
  • //tpu_sync/transport/lib/socket:util_test (PASSED in 7.2s)
  • //tpu_sync/transport/peregrine/src/internal/socket:tcp_socket_util_test (PASSED in 18.2s, verified SendReportsDeadlineExceededWhenPeerStopsReading non-hanging)
  • //tpu_sync/transport/lib:raw_buffer_transport_test (PASSED in 16.9s)
  • //tpu_sync/transport:block_transport_test (PASSED in 18.9s, verified telemetry assertions expecting error_code="UNAVAILABLE")
  • //tpu_sync/transport/peregrine/src/internal/socket:all (PASSED 6/6 tests)