Sorry about the delay.
rpc.api._wait_all_workers() is not supposed to be directly called by applications. I was only referencing it as one example to use rank 0 as a coordinator. The recommended way is doing sth like below:
futs = []
for i in array_rpc:
    futs.append(rpc.rpc_async(my_target, add_outlinks, args=(arr_send[i],source)))

...

for fut in futs:
    fut.wait()

Allow me some time, still reading the contents above.
Edits
I see you are already using future in the latest version.

Could you explain why? and could you please suggest some ways to decrease the excuse time? I run the program with UPC++, it take some minutes to finish (~ 5-6 minutes)

Given that the comp only takes 5s, it looks like pickle and comm takes the majority of the time? Curious, can you profile the time taken for one rpc_sync?
The communication pattern above looks like an allToAll. Is it guaranteed that the data sent from rank x to rank y will be of the same size. If so, you can use the allToAll API 1 (not in v1.5, only available on master now, will come to v1.6).
Regarding ways to speed up, one easy attempt would be increasing the number of threads. The program currently sets num_send_recv_threads=16. You could try 64. This could help speeding up deserialization on the server side. But the serializer on the sender side still occurs inline, we probably should also offload that to thread-pool in future releases.
I take that back. I think one major reason is that the pickle and execution both require Python GIL, and hence they won’t run in parallel on the server side. Can you try using TorchScript functions? You can find examples here 1 by searching for @torch.jit.script. Even if we use @torch.jit.script functions, the pickle/unpickle would still require GIL as those are Python objects, but the execution can run in parallel.