Update on "sync and async torch.distributed.rpc for builtin operators"

mrshenli · mrshenli · commit 3c642e8786cf · 2019-08-05T14:39:39.000-07:00
Features: * sync and async RPC for builtin operators * RpcAgent API * ProcessGroupAgent implementation Goal: This is the first PR for #23110, and there will be many followup ones. So let's focus on the overall API and code structure. Details like efficiency and error handling can be improved in future PRs. * have a minimum working and testable RPC implementation. * make sure the RpcAgent API is sufficient for future ThriftAgent and TensorPipeAgent implementation * For tensor pipe implementation, it might allocate multiple underlying communication channels with different types, and might also use streaming serialization/deserialization for large tensors. To support this requirement, the current implementation only convert a BuiltinOp into a Message which contains a byte vector and a tensor table. It is up to the RpcAgent implementation to determine how it would like to serialize a Message object. * For ThriftAgent, as Thrift has it own request/response matching solution, the Message.id is no longer necessary. Hence the id can be dropped during serialization. All it needs to do is to pass the response Message object to the Future returned by send(...). * support blocking and non-blocking RequestCallback * blocking means the callback won't return before sending out the response * non-blocking can be achieved by enqueue the `(from, request, RpcAgent&)` tuple and use a different thread to process them. That is why there is an `RpcAgent&` arg in the param list. Differential Revision: [D15194693](https://our.internmc.facebook.com/intern/diff/D15194693/)
diff --git a/torch/csrc/distributed/rpc/ProcessGroupAgent.cpp b/torch/csrc/distributed/rpc/ProcessGroupAgent.cpp
@@ -77,13 +77,6 @@ ProcessGroupAgent::ProcessGroupAgent(
   listenerThread_ = std::thread(&ProcessGroupAgent::listenLoop, this);
 }
 
-ProcessGroupAgent::~ProcessGroupAgent() {
-  if (!stop_) {
-    AT_ERROR(stop_, "Must call ProcessGroupAgent::shutdown before destructor");
-  }
-}
-
-
 void ProcessGroupAgent::join() {
   // Every process i sends a SHUTDOWN message to process i + 1. This is
   // necessary for now because:
diff --git a/torch/csrc/distributed/rpc/ProcessGroupAgent.h b/torch/csrc/distributed/rpc/ProcessGroupAgent.h
@@ -30,8 +30,6 @@ class ProcessGroupAgent : public RpcAgent {
                     std::unordered_map<std::string, int> nameMap,
                     std::shared_ptr<c10d::ProcessGroup> pg);
 
-  ~ProcessGroupAgent() override;
-
   // This method wraps the destination information and the message into a
   // SendWork object, and put the SendWork into a queue. Another thread will
   // consume SendWork from the queue and send it out.
