ProcessGroupBaby: use pipe for improved performance

[d4l3k]

This speeds up ProcessGroupBaby performance for small operations (4s -> 2.5s).

Optimizations:

• use Pipe instead of Queue: 4s -> 3s
• generate op_id in parent so _run_func is fully async with no synchronization with child process: 3s -> 2.5s

Test plan:

python torchft/checkpointing/pg_transport_bench.py --device cuda --inplace --total-size=16000 --chunk-size=4

With fix:

INFO:torchft.checkpointing.pg_transport:init_pg took 1.8898121230304241s
INFO:torchft.checkpointing.pg_transport:create state_dict took 0.024819843471050262s
INFO:torchft.checkpointing.pg_transport:preparing state_dict took 0.1422094851732254s
INFO:torchft.checkpointing.pg_transport:send pickle took 0.010462276637554169s
INFO:torchft.checkpointing.pg_transport:init_pg took 1.9368509724736214s
INFO:torchft.checkpointing.pg_transport:create state_dict took 0.02213948592543602s
INFO:torchft.checkpointing.pg_transport:send tensors took 2.2778245247900486s
INFO:torchft.checkpointing.pg_transport:send_checkpoint took 2.482279524207115s
INFO:torchft.checkpointing.pg_transport:recv_checkpoint took 2.641794554889202s

Baseline:

INFO:torchft.checkpointing.pg_transport:init_pg took 1.893109567463398s
INFO:torchft.checkpointing.pg_transport:create state_dict took 0.025067199021577835s
INFO:torchft.checkpointing.pg_transport:preparing state_dict took 0.1476067118346691s
INFO:torchft.checkpointing.pg_transport:init_pg took 1.9768157303333282s
INFO:torchft.checkpointing.pg_transport:create state_dict took 0.024319566786289215s
INFO:torchft.checkpointing.pg_transport:send pickle took 0.16672053188085556s
INFO:torchft.checkpointing.pg_transport:send tensors took 3.718041818588972s
INFO:torchft.checkpointing.pg_transport:send_checkpoint took 4.046096011996269s
INFO:torchft.checkpointing.pg_transport:recv_checkpoint took 4.189382903277874s

[H-Huang]

LGTM!

[H-Huang]

send doesnt need a timeout?

[d4l3k]

There's not an easy mechanism to handle timeouts on write and from my testing the messages are small enough that the messages effectively instantly send as they just copy into the kernel default buffer size

If the remote process exits this will fail with a BrokenPipe error as well so that case is covered