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

mrshenli · mrshenli · commit c5cb2a163a8b · 2019-07-24T19:02:04.000-07:00
Stack from [ghstack](https://github.com/ezyang/ghstack): * **#23228 sync and async torch.distributed.rpc for builtin operators** Features: * sync and async RPC for builtin operators * RpcAgent API * ProcessGroupAgent implementation Goal: * have a minimum working and testable RPC implementation for #23110 * 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
@@ -90,7 +90,7 @@ void ProcessGroupAgent::shutdown() {
   // 2. A GLOO process cannot send message to itself. (there is an ongoing
   //    effort to fix this problem).
   int dst = (pg_->getRank() + 1) % pg_->getSize();
-  enqueue(SendWork(dst, std::move(Message({}, {}, MessageType::SHUTDOWN))));
+  enqueue(SendWork(dst, Message({}, {}, MessageType::SHUTDOWN)));
   std::unique_lock<std::mutex> lock(sendQueueMutex_);
   workConsumeCV_.wait(lock, [&] { return sendQueue_.empty(); });
   stop_ = true;
diff --git a/torch/csrc/distributed/rpc/init.cpp b/torch/csrc/distributed/rpc/init.cpp
@@ -21,7 +21,7 @@ namespace {
 template <typename T>
 using shared_ptr_class_ = py::class_<T, std::shared_ptr<T>>;
 
-PyObject* rpc_init(PyObject* _unused) {
+PyObject* rpc_init(PyObject* /* unused */) {
   auto dist_module = THPObjectPtr(PyImport_ImportModule("torch.distributed"));
   if (!dist_module) {
     throw python_error();
