[distributed] add support for rpc timeouts in process group agent #28392

rohan-varma · 2019-10-21T22:40:19Z

Stack from ghstack:

[distributed] add support for rpc timeouts in process group agent #28392 [distributed] add support for rpc timeouts in process group agent

Per #25531, we want to clean up futures when we detect that there are
failures/timeouts. As a first step, this diff adds a timeout parameter to process group agent (in miliseconds) that can be set by the user in Python, and also keeps track of timeouts by maintaining a map from timeout --> requests as suggested by @pritamdamania87. The map must stay in sync with the existing futures_ map, so we write/delete to this map only when the futures_ map is written/deleted.

Note that this diff does not implement the actual process of detecting/handling timed out RPCs. The next step will be to implement a background thread that polls this map periodically and marks futures that have timed out. After this we can also add relevant unit tests. There's still an open question about how we should handle futures that complete/return from the caller after the timeout.

Differential Revision: D18025163

timeouts Per #25531, we want to clean up futures when we detect that there are failures/timeouts. As a first step, this diff adds timers to the future object, provides functionality to check if a future is timed out, and allows specification of the timeout when initializing rpc. A future diff will check for these timeouts and mark the future completed with an exception indicating that it has timed out. Differential Revision: [D18025163](https://our.internmc.facebook.com/intern/diff/D18025163/) [ghstack-poisoned]

… to support future timeouts" timeouts** timeouts Per #25531, we want to clean up futures when we detect that there are failures/timeouts. As a first step, this diff adds timers to the future object, provides functionality to check if a future is timed out, and allows specification of the timeout when initializing rpc. A future diff will check for these timeouts and mark the future completed with an exception indicating that it has timed out. Differential Revision: [D18025163](https://our.internmc.facebook.com/intern/diff/D18025163/) [ghstack-poisoned]

timeouts Pull Request resolved: #28392 Per #25531, we want to clean up futures when we detect that there are failures/timeouts. As a first step, this diff adds timers to the future object, provides functionality to check if a future is timed out, and allows specification of the timeout when initializing rpc. A future diff will check for these timeouts and mark the future completed with an exception indicating that it has timed out. ghstack-source-id: 92337906 Differential Revision: [D18025163](https://our.internmc.facebook.com/intern/diff/D18025163/)

rohan-varma · 2019-10-22T00:12:16Z

test/rpc_test.py

+    @requires_process_group_agent("PROCESS_GROUP rpc backend specific test, skip")
+    def test_timeout_set(self):
+        rpc_timeout_seconds = rpc.get_rpc_timeout().seconds
+        self.assertEqual(rpc_timeout_seconds, 100)


I have the default timeout for all RPCs to be 100 seconds for now, but I'm not sure what a good default is. If it's too small, users could get time outs when in fact the operation was going to complete successfully.

Let's use 10s for now but also provide a way to specify infinite timeout. Some RPC frameworks use timeout=0 for this. We should document this as well.

rohan-varma · 2019-10-22T00:15:39Z

test/rpc_test.py

+            timeout_function,
+            args=(3,)
+        )
+        time.sleep(1)


this is not ideal to do in a test, but I'm not sure how to test that the timer works other than this.

zhaojuanmao · 2019-10-22T17:12:14Z

Could we add timeout outside of future class? e.g, in process group agent. One reason is that we will use IValue::Future soon, which does not have timeout setting.

rohan-varma · 2019-10-22T20:48:55Z

Could we add timeout outside of future class? e.g, in process group agent. One reason is that we will use IValue::Future soon, which does not have timeout setting.

We'd need to store some information indicating the (relative) time the future was started, so that we can check if the future has timed out. Would it be okay to have this as an additional structure in ProcessGroupAgent?

pritamdamania87 · 2019-10-23T00:50:39Z

The future itself should not have a timeout. ProcessGroupAgent should have a timeout on a request and mark the appropriate future with an "Operation timed out" exception.

pritamdamania87 · 2019-10-23T00:58:35Z

test/rpc_test.py

+            args=(3,)
+        )
+        time.sleep(1)
+        self.assertTrue(fut.check_time_elapsed(timedelta(seconds=0)))


I don't think we should expose a method like check_time_elapsed to users.

pritamdamania87 · 2019-10-23T00:59:04Z

test/rpc_test.py

        self.assertEqual(fut.wait(), torch.ones(n, n) * 2)

+    @dist_init
+    def test_future_timer(self):


In the test we should call a method that takes more time than the timeout and verify that future.wait() throws a timed out exception.

pritamdamania87 · 2019-10-23T01:00:03Z

test/rpc_test.py

+    @requires_process_group_agent("PROCESS_GROUP rpc backend specific test, skip")
+    def test_timeout_set(self):
+        rpc_timeout_seconds = rpc.get_rpc_timeout().seconds
+        self.assertEqual(rpc_timeout_seconds, 100)


nit: make 100 a constant in the rpc_api.py that you can use elsewhere.

pritamdamania87 · 2019-10-23T01:00:23Z

torch/csrc/distributed/rpc/init.cpp

          py::arg("process_group"),
-          py::arg("num_send_recv_threads") = 4)
+          py::arg("num_send_recv_threads") = 4,
+          py::arg("future_timeout") = std::chrono::seconds(100))


nit: use a constant.

pritamdamania87 · 2019-10-23T01:00:58Z

torch/csrc/distributed/rpc/process_group_agent.cpp

      sendMutexes_(pg_->getSize()),
-      threadPool_(numSendRecvThreads) {
+      threadPool_(numSendRecvThreads),
+      futureTimeout_(futureTimeout) {


this should be called rpc timeout and not future timeout

pritamdamania87 · 2019-10-23T01:01:33Z

torch/csrc/distributed/rpc/process_group_agent.cpp

  return allWorkerInfo_[id];
 }

+const std::chrono::seconds& ProcessGroupAgent::getRpcTimeout() {


We should store millisecond precision in this class and not at seconds precision.

pritamdamania87 · 2019-10-23T01:01:56Z

torch/csrc/distributed/rpc/process_group_agent.h

      std::string workerName,
      std::shared_ptr<c10d::ProcessGroup> pg,
-      int numSendRecvThreads = 4);
+      int numSendRecvThreads = 4,


nit: not related to this PR, but make 4 a const too.

pritamdamania87 · 2019-10-23T01:04:49Z

torch/distributed/rpc/__init__.py

                            self_rank=-1,
                            init_method=None,
-                            num_send_recv_threads=4):
+                            num_send_recv_threads=4, rpc_timeout=100):


rpc_timeout should be specified by the user in milliseconds and not seconds. Some users might want to set it to 100ms for example.

Can we enforce that the user facing API only accepts timedelta as the timeout parameter? This would make the API much cleaner.

Also probably rename it to rpc_timeout_ms

torch/distributed/rpc/__init__.py

pritamdamania87 · 2019-10-23T01:07:35Z

test/rpc_test.py

+    @requires_process_group_agent("PROCESS_GROUP rpc backend specific test, skip")
+    def test_timeout_set(self):
+        rpc_timeout_seconds = rpc.get_rpc_timeout().seconds
+        self.assertEqual(rpc_timeout_seconds, 100)


Let's use 10s for now but also provide a way to specify infinite timeout. Some RPC frameworks use timeout=0 for this. We should document this as well.

pietern

Is the caller always responsible for timing out an operation?

pietern · 2019-10-24T13:07:11Z

torch/csrc/distributed/rpc/future_message.h

 #pragma once

 #include <torch/csrc/distributed/rpc/message.h>
+#include <chrono>


Include stdlib before everything else.

pietern · 2019-10-24T13:08:44Z

torch/csrc/distributed/rpc/future_message.h

  // TODO: make message_ an optional field, and get rid of UNKNOWN message type
  Message message_;
+  // clock to measure time elapsed since the future was created.
+  std::chrono::steady_clock clock_;


pietern · 2019-10-24T13:09:04Z

torch/csrc/distributed/rpc/future_message.h

 // TODO: consider using ivalue::Future.
 struct TORCH_API FutureMessage final {
 public:
+  FutureMessage();


satgera · 2019-10-24T21:51:15Z

torch/csrc/distributed/rpc/future_message.cpp

+  if (completed_) {
+    return false;
+  }
+  const auto now = std::chrono::high_resolution_clock::now();


Please use std::chrono::steady_clock every place in this diff from what I recall high_resolution_clock isn't guaranteed to be monotonic and may be more expensive than steady_clock.

…set timeout when initializing rpc" Per #25531, we want to clean up futures when we detect that there are failures/timeouts. As a first step, this diff adds timers to the future object, provides functionality to check if a future is timed out, and allows specification of the timeout when initializing rpc. In the next diff, we will add a separate thread to ProcessGroupAgent to clean up these futures if they have timed out. Differential Revision: [D18025163](https://our.internmc.facebook.com/intern/diff/D18025163/) [ghstack-poisoned]

… when initializing rpc Pull Request resolved: #28392 Per #25531, we want to clean up futures when we detect that there are failures/timeouts. As a first step, this diff adds timers to the future object, provides functionality to check if a future is timed out, and allows specification of the timeout when initializing rpc. A future diff will check for these timeouts and mark the future completed with an exception indicating that it has timed out. ghstack-source-id: 92601334 Differential Revision: [D18025163](https://our.internmc.facebook.com/intern/diff/D18025163/)

…set timeout when initializing rpc" Per #25531, we want to clean up futures when we detect that there are failures/timeouts. As a first step, this diff adds timers to the future object, provides functionality to check if a future is timed out, and allows specification of the timeout when initializing rpc. In the next diff, we will add a separate thread to ProcessGroupAgent to clean up these futures if they have timed out. Differential Revision: [D18025163](https://our.internmc.facebook.com/intern/diff/D18025163/) [ghstack-poisoned]

… when initializing rpc Pull Request resolved: #28392 Per #25531, we want to clean up futures when we detect that there are failures/timeouts. As a first step, this diff adds timers to the future object, provides functionality to check if a future is timed out, and allows specification of the timeout when initializing rpc. A future diff will check for these timeouts and mark the future completed with an exception indicating that it has timed out. ghstack-source-id: 92603029 Differential Revision: [D18025163](https://our.internmc.facebook.com/intern/diff/D18025163/)

…set timeout when initializing rpc" Per #25531, we want to clean up futures when we detect that there are failures/timeouts. As a first step, this diff adds timers to the future object, provides functionality to check if a future is timed out, and allows specification of the timeout when initializing rpc. In the next diff, we will add a separate thread to ProcessGroupAgent to clean up these futures if they have timed out. Differential Revision: [D18025163](https://our.internmc.facebook.com/intern/diff/D18025163/) [ghstack-poisoned]

… when initializing rpc Pull Request resolved: #28392 Per #25531, we want to clean up futures when we detect that there are failures/timeouts. As a first step, this diff adds timers to the future object, provides functionality to check if a future is timed out, and allows specification of the timeout when initializing rpc. A future diff will check for these timeouts and mark the future completed with an exception indicating that it has timed out. ghstack-source-id: 92607009 Differential Revision: [D18025163](https://our.internmc.facebook.com/intern/diff/D18025163/)

…set timeout when initializing rpc" Per #25531, we want to clean up futures when we detect that there are failures/timeouts. As a first step, this diff adds timers to the future object, provides functionality to check if a future is timed out, and allows specification of the timeout when initializing rpc. In the next diff, we will add a separate thread to ProcessGroupAgent to clean up these futures if they have timed out. Differential Revision: [D18025163](https://our.internmc.facebook.com/intern/diff/D18025163/) [ghstack-poisoned]

…s group agent" Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. Differential Revision: [D18434215](https://our.internmc.facebook.com/intern/diff/D18434215/) [ghstack-poisoned]

Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. Differential Revision: [D18434215](https://our.internmc.facebook.com/intern/diff/D18434215/) [ghstack-poisoned]

…s group agent" Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. Differential Revision: [D18434215](https://our.internmc.facebook.com/intern/diff/D18434215/) [ghstack-poisoned]

Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. Differential Revision: [D18434215](https://our.internmc.facebook.com/intern/diff/D18434215/) [ghstack-poisoned]

…s group agent" Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. The existing `futureTimeouts_` map is reworked so that the key is now the time (in ms) that the RPC would timeout, instead of when the RPC began. The reason for this is that it allows us to order futures which have different timeouts, which is a feature that we'd want to support in the future. A helper function `getRPCEndTime` can be used to retrieve the key into this map. Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. Differential Revision: [D18434215](https://our.internmc.facebook.com/intern/diff/D18434215/) [ghstack-poisoned]

Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. The existing `futureTimeouts_` map is reworked so that the key is now the time (in ms) that the RPC would timeout, instead of when the RPC began. The reason for this is that it allows us to order futures which have different timeouts, which is a feature that we'd want to support in the future. A helper function `getRPCEndTime` can be used to retrieve the key into this map. Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. Differential Revision: [D18434215](https://our.internmc.facebook.com/intern/diff/D18434215/) [ghstack-poisoned]

Pull Request resolved: #29601 Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs at a pre-set interval, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. ghstack-source-id: 94086158 Differential Revision: [D18434215](https://our.internmc.facebook.com/intern/diff/D18434215/)

…s group agent" Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. The existing `futureTimeouts_` map is reworked so that the key is now the time (in ms) that the RPC would timeout, instead of when the RPC began. The reason for this is that it allows us to order futures which have different timeouts, which is a feature that we'd want to support in the future. A helper function `getRPCEndTime` can be used to retrieve the key into this map. Since we'd like to be able to quickly stop checking for timed out RPCs after looking at the first entry in the `futureTimeouts_` map, we cannot key by beginning time (an RPC with a much smaller timeout may later be created) or by the timeout itself (since an RPC with a large timeout would not be at the beginning, even if it is about to time out). Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. Differential Revision: [D18434215](https://our.internmc.facebook.com/intern/diff/D18434215/) [ghstack-poisoned]

Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. The existing `futureTimeouts_` map is reworked so that the key is now the time (in ms) that the RPC would timeout, instead of when the RPC began. The reason for this is that it allows us to order futures which have different timeouts, which is a feature that we'd want to support in the future. A helper function `getRPCEndTime` can be used to retrieve the key into this map. Since we'd like to be able to quickly stop checking for timed out RPCs after looking at the first entry in the `futureTimeouts_` map, we cannot key by beginning time (an RPC with a much smaller timeout may later be created) or by the timeout itself (since an RPC with a large timeout would not be at the beginning, even if it is about to time out). Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. Differential Revision: [D18434215](https://our.internmc.facebook.com/intern/diff/D18434215/) [ghstack-poisoned]

…s group agent" Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. The existing `futureTimeouts_` map is reworked so that the key is now the time (in ms) that the RPC would timeout, instead of when the RPC began. The reason for this is that it allows us to order futures which have different timeouts, which is a feature that we'd want to support in the future. A helper function `getRPCEndTime` can be used to retrieve the key into this map. Since we'd like to be able to quickly stop checking for timed out RPCs after looking at the first entry in the `futureTimeouts_` map, we cannot key by beginning time (an RPC with a much smaller timeout may later be created) or by the timeout itself (since an RPC with a large timeout would not be at the beginning, even if it is about to time out). Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. Differential Revision: [D18434215](https://our.internmc.facebook.com/intern/diff/D18434215/) [ghstack-poisoned]

Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. The existing `futureTimeouts_` map is reworked so that the key is now the time (in ms) that the RPC would timeout, instead of when the RPC began. The reason for this is that it allows us to order futures which have different timeouts, which is a feature that we'd want to support in the future. A helper function `getRPCEndTime` can be used to retrieve the key into this map. Since we'd like to be able to quickly stop checking for timed out RPCs after looking at the first entry in the `futureTimeouts_` map, we cannot key by beginning time (an RPC with a much smaller timeout may later be created) or by the timeout itself (since an RPC with a large timeout would not be at the beginning, even if it is about to time out). Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. Differential Revision: [D18434215](https://our.internmc.facebook.com/intern/diff/D18434215/) [ghstack-poisoned]

Pull Request resolved: #29601 Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs at a pre-set interval, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. ghstack-source-id: 94097468 Differential Revision: [D18434215](https://our.internmc.facebook.com/intern/diff/D18434215/)

…s group agent" Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. The existing `futureTimeouts_` map is reworked so that the key is now the time (in ms) that the RPC would timeout, instead of when the RPC began. The reason for this is that it allows us to order futures which have different timeouts, which is a feature that we'd want to support in the future. A helper function `getRPCEndTime` can be used to retrieve the key into this map. Since we'd like to be able to quickly stop checking for timed out RPCs after looking at the first entry in the `futureTimeouts_` map, we cannot key by beginning time (an RPC with a much smaller timeout may later be created) or by the timeout itself (since an RPC with a large timeout would not be at the beginning, even if it is about to time out). Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. Differential Revision: [D18434215](https://our.internmc.facebook.com/intern/diff/D18434215/) [ghstack-poisoned]

Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. The existing `futureTimeouts_` map is reworked so that the key is now the time (in ms) that the RPC would timeout, instead of when the RPC began. The reason for this is that it allows us to order futures which have different timeouts, which is a feature that we'd want to support in the future. A helper function `getRPCEndTime` can be used to retrieve the key into this map. Since we'd like to be able to quickly stop checking for timed out RPCs after looking at the first entry in the `futureTimeouts_` map, we cannot key by beginning time (an RPC with a much smaller timeout may later be created) or by the timeout itself (since an RPC with a large timeout would not be at the beginning, even if it is about to time out). Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. Differential Revision: [D18434215](https://our.internmc.facebook.com/intern/diff/D18434215/) [ghstack-poisoned]

Pull Request resolved: #29601 Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs at a pre-set interval, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. ghstack-source-id: 94109478 Differential Revision: [D18434215](https://our.internmc.facebook.com/intern/diff/D18434215/)

…s group agent" Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. The existing `futureTimeouts_` map is reworked so that the key is now the time (in ms) that the RPC would timeout, instead of when the RPC began. The reason for this is that it allows us to order futures which have different timeouts, which is a feature that we'd want to support in the future. A helper function `getRPCEndTime` can be used to retrieve the key into this map. Since we'd like to be able to quickly stop checking for timed out RPCs after looking at the first entry in the `futureTimeouts_` map, we cannot key by beginning time (an RPC with a much smaller timeout may later be created) or by the timeout itself (since an RPC with a large timeout would not be at the beginning, even if it is about to time out). Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. Differential Revision: [D18434215](https://our.internmc.facebook.com/intern/diff/D18434215/) [ghstack-poisoned]

Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. The existing `futureTimeouts_` map is reworked so that the key is now the time (in ms) that the RPC would timeout, instead of when the RPC began. The reason for this is that it allows us to order futures which have different timeouts, which is a feature that we'd want to support in the future. A helper function `getRPCEndTime` can be used to retrieve the key into this map. Since we'd like to be able to quickly stop checking for timed out RPCs after looking at the first entry in the `futureTimeouts_` map, we cannot key by beginning time (an RPC with a much smaller timeout may later be created) or by the timeout itself (since an RPC with a large timeout would not be at the beginning, even if it is about to time out). Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. Differential Revision: [D18434215](https://our.internmc.facebook.com/intern/diff/D18434215/) [ghstack-poisoned]

Pull Request resolved: #29601 Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs at a pre-set interval, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. ghstack-source-id: 94167479 Differential Revision: [D18434215](https://our.internmc.facebook.com/intern/diff/D18434215/)

…s group agent" Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. The existing `futureTimeouts_` map is reworked so that the key is now the time (in ms) that the RPC would timeout, instead of when the RPC began. The reason for this is that it allows us to order futures which have different timeouts, which is a feature that we'd want to support in the future. A helper function `getRPCEndTime` can be used to retrieve the key into this map. Since we'd like to be able to quickly stop checking for timed out RPCs after looking at the first entry in the `futureTimeouts_` map, we cannot key by beginning time (an RPC with a much smaller timeout may later be created) or by the timeout itself (since an RPC with a large timeout would not be at the beginning, even if it is about to time out). Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. Differential Revision: [D18434215](https://our.internmc.facebook.com/intern/diff/D18434215/) [ghstack-poisoned]

Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. The existing `futureTimeouts_` map is reworked so that the key is now the time (in ms) that the RPC would timeout, instead of when the RPC began. The reason for this is that it allows us to order futures which have different timeouts, which is a feature that we'd want to support in the future. A helper function `getRPCEndTime` can be used to retrieve the key into this map. Since we'd like to be able to quickly stop checking for timed out RPCs after looking at the first entry in the `futureTimeouts_` map, we cannot key by beginning time (an RPC with a much smaller timeout may later be created) or by the timeout itself (since an RPC with a large timeout would not be at the beginning, even if it is about to time out). Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. Differential Revision: [D18434215](https://our.internmc.facebook.com/intern/diff/D18434215/) [ghstack-poisoned]

Pull Request resolved: #29601 Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs at a pre-set interval, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. ghstack-source-id: 94175131 Differential Revision: [D18434215](https://our.internmc.facebook.com/intern/diff/D18434215/)

Summary: Pull Request resolved: #29601 Follow up from #28392. Adds a background thread to `ProcessGroupAgent` that polls for timed out RPCs at a pre-set interval, and marks them as completed with a timeout exception if they have timed out. Also deletes the futures from the corresponding maps `futures_` and `futureTimeouts`. Unit tests are added to ensure that timed out RPCs are appropriately cleaned up. Also adds a `shutdown` variable to process group agent to control the shutting down of this background thread, which can eventually be extended to use for controlling a clean shutdown of process group agent. ghstack-source-id: 94175131 Test Plan: Added unit tests Differential Revision: D18434215 fbshipit-source-id: c48abdb8759fe1447200ec66bb9d4b1c50ec4535

rohan-varma requested review from mrshenli and pietern as code owners October 21, 2019 22:40

rohan-varma changed the title ~~[distributed] add timers to future and timeout to rpc to support future timeouts~~ [WIP][distributed] add timers to future and timeout to rpc to support future timeouts Oct 21, 2019

rohan-varma requested a review from apaszke as a code owner October 22, 2019 00:07

rohan-varma changed the title ~~[WIP][distributed] add timers to future and timeout to rpc to support future timeouts~~ [distributed] add timeout for RPC futures, and ability to set timeout when initializing rpc Oct 22, 2019

rohan-varma commented Oct 22, 2019

View reviewed changes

rohan-varma requested review from pritamdamania87 and zhaojuanmao October 22, 2019 00:16

pritamdamania87 suggested changes Oct 23, 2019

View reviewed changes

pietern reviewed Oct 24, 2019

View reviewed changes

satgera reviewed Oct 24, 2019

View reviewed changes

pietern mentioned this pull request Nov 20, 2019

[RPC] ProcessGroupAgent should clean up futures on failures/timeouts #25531

Closed

rohan-varma mentioned this pull request Jan 28, 2020

[RFC] Support per-RPC timeouts in RPC layer. #32686

Closed

mruberry added the Merged label Oct 28, 2020

[distributed] add support for rpc timeouts in process group agent #28392

[distributed] add support for rpc timeouts in process group agent #28392

Uh oh!

Conversation

rohan-varma commented Oct 21, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zhaojuanmao commented Oct 22, 2019

Uh oh!

rohan-varma commented Oct 22, 2019

Uh oh!

pritamdamania87 commented Oct 23, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pietern left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

satgera Oct 24, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

9 participants

rohan-varma commented Oct 21, 2019 •

edited

Loading

satgera Oct 24, 2019 •

edited

Loading