We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[tune] GPU utilization getting checked for cluster with no gpu
<!--Please include [tune], [rllib], [autoscaler] etc. in the issue title if relevant-->

### What is the problem?

Trivial issue, which does not actually impact performance as far as I can tell, but it seems if you launch a cluster with a configuration that sets gpus = 0, then there might be a check which avoids this error message...


```
(pid=4628) Exception in thread Thread-2: 
(pid=4628) Traceback (most recent call last):
(pid=4628)   File "/usr/lib/python3.6/threading.py", line 916, in _bootstrap_inner
(pid=4628)     self.run()
(pid=4628)   File "/home/ubuntu/algo/lib/python3.6/site-packages/ray/tune/utils/util.py", line 89, in run
(pid=4628)     self._read_utilization()
(pid=4628)   File "/home/ubuntu/algo/lib/python3.6/site-packages/ray/tune/utils/util.py", line 65, in _read_utilization
(pid=4628)     for gpu in GPUtil.getGPUs():
(pid=4628)   File "/home/ubuntu/algo/lib/python3.6/site-packages/GPUtil/GPUtil.py", line 102, in getGPUs
(pid=4628)     deviceIds = int(vals[i])
(pid=4628) ValueError: invalid literal for int() with base 10: "NVIDIA-SMI has failed because it couldn't communicate with the NVIDIA driver. Make  sure that the latest NVIDIA driver is installed and running."
```



*Ray version and other system information (Python version, TensorFlow version, OS):*
Ray 0.8.1, TF 2.1, RHEL 7.7

### Reproduction (REQUIRED)
Please provide a script that can be run to reproduce the issue. The script should have **no external library dependencies** (i.e., use fake or mock data / environments):

If we cannot run your script, we cannot fix your issue.

- [ ] I have verified my script runs in a clean environment and reproduces the issue.
- [ ] I have verified the issue also occurs with the [latest wheels](https://ray.readthedocs.io/en/latest/installation.html).


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `python/ray/tune/utils/util.py`
Content:
```
1 import copy
2 import logging
3 import threading
4 import time
5 from collections import defaultdict
6 from threading import Thread
7 
8 import numpy as np
9 import ray
10 import psutil
11 
12 logger = logging.getLogger(__name__)
13 
14 try:
15     import GPUtil
16 except ImportError:
17     GPUtil = None
18 
19 _pinned_objects = []
20 PINNED_OBJECT_PREFIX = "ray.tune.PinnedObject:"
21 START_OF_TIME = time.time()
22 
23 
24 class UtilMonitor(Thread):
25     """Class for system usage utilization monitoring.
26 
27     It keeps track of CPU, RAM, GPU, VRAM usage (each gpu separately) by
28     pinging for information every x seconds in a separate thread.
29 
30     Requires psutil and GPUtil to be installed. Can be enabled with
31     tune.run(config={"log_sys_usage": True}).
32     """
33 
34     def __init__(self, start=True, delay=0.7):
35         self.stopped = True
36         if GPUtil is None and start:
37             logger.warning("Install gputil for GPU system monitoring.")
38 
39         if psutil is None and start:
40             logger.warning("Install psutil to monitor system performance.")
41 
42         if GPUtil is None and psutil is None:
43             return
44 
45         super(UtilMonitor, self).__init__()
46         self.delay = delay  # Time between calls to GPUtil
47         self.values = defaultdict(list)
48         self.lock = threading.Lock()
49         self.daemon = True
50         if start:
51             self.start()
52 
53     def _read_utilization(self):
54         with self.lock:
55             if psutil is not None:
56                 self.values["cpu_util_percent"].append(
57                     float(psutil.cpu_percent(interval=None)))
58                 self.values["ram_util_percent"].append(
59                     float(getattr(psutil.virtual_memory(), "percent")))
60             if GPUtil is not None:
61                 for gpu in GPUtil.getGPUs():
62                     self.values["gpu_util_percent" + str(gpu.id)].append(
63                         float(gpu.load))
64                     self.values["vram_util_percent" + str(gpu.id)].append(
65                         float(gpu.memoryUtil))
66 
67     def get_data(self):
68         if self.stopped:
69             return {}
70 
71         with self.lock:
72             ret_values = copy.deepcopy(self.values)
73             for key, val in self.values.items():
74                 del val[:]
75         return {
76             "perf": {
77                 k: np.mean(v)
78                 for k, v in ret_values.items() if len(v) > 0
79             }
80         }
81 
82     def run(self):
83         self.stopped = False
84         while not self.stopped:
85             self._read_utilization()
86             time.sleep(self.delay)
87 
88     def stop(self):
89         self.stopped = True
90 
91 
92 def pin_in_object_store(obj):
93     """Deprecated, use ray.put(value, weakref=False) instead."""
94 
95     obj_id = ray.put(obj, weakref=False)
96     _pinned_objects.append(obj_id)
97     return obj_id
98 
99 
100 def get_pinned_object(pinned_id):
101     """Deprecated."""
102 
103     return ray.get(pinned_id)
104 
105 
106 class warn_if_slow:
107     """Prints a warning if a given operation is slower than 100ms.
108 
109     Example:
110         >>> with warn_if_slow("some_operation"):
111         ...    ray.get(something)
112     """
113 
114     DEFAULT_THRESHOLD = 0.5
115 
116     def __init__(self, name, threshold=None):
117         self.name = name
118         self.threshold = threshold or self.DEFAULT_THRESHOLD
119         self.too_slow = False
120 
121     def __enter__(self):
122         self.start = time.time()
123         return self
124 
125     def __exit__(self, type, value, traceback):
126         now = time.time()
127         if now - self.start > self.threshold and now - START_OF_TIME > 60.0:
128             self.too_slow = True
129             logger.warning(
130                 "The `%s` operation took %s seconds to complete, "
131                 "which may be a performance bottleneck.", self.name,
132                 now - self.start)
133 
134 
135 def merge_dicts(d1, d2):
136     """
137     Args:
138         d1 (dict): Dict 1.
139         d2 (dict): Dict 2.
140 
141     Returns:
142          dict: A new dict that is d1 and d2 deep merged.
143     """
144     merged = copy.deepcopy(d1)
145     deep_update(merged, d2, True, [])
146     return merged
147 
148 
149 def deep_update(original,
150                 new_dict,
151                 new_keys_allowed=False,
152                 whitelist=None,
153                 override_all_if_type_changes=None):
154     """Updates original dict with values from new_dict recursively.
155 
156     If new key is introduced in new_dict, then if new_keys_allowed is not
157     True, an error will be thrown. Further, for sub-dicts, if the key is
158     in the whitelist, then new subkeys can be introduced.
159 
160     Args:
161         original (dict): Dictionary with default values.
162         new_dict (dict): Dictionary with values to be updated
163         new_keys_allowed (bool): Whether new keys are allowed.
164         whitelist (Optional[List[str]]): List of keys that correspond to dict
165             values where new subkeys can be introduced. This is only at the top
166             level.
167         override_all_if_type_changes(Optional[List[str]]): List of top level
168             keys with value=dict, for which we always simply override the
169             entire value (dict), iff the "type" key in that value dict changes.
170     """
171     whitelist = whitelist or []
172     override_all_if_type_changes = override_all_if_type_changes or []
173 
174     for k, value in new_dict.items():
175         if k not in original and not new_keys_allowed:
176             raise Exception("Unknown config parameter `{}` ".format(k))
177 
178         # Both orginal value and new one are dicts.
179         if isinstance(original.get(k), dict) and isinstance(value, dict):
180             # Check old type vs old one. If different, override entire value.
181             if k in override_all_if_type_changes and \
182                     "type" in value and "type" in original[k] and \
183                     value["type"] != original[k]["type"]:
184                 original[k] = value
185             # Whitelisted key -> ok to add new subkeys.
186             elif k in whitelist:
187                 deep_update(original[k], value, True)
188             # Non-whitelisted key.
189             else:
190                 deep_update(original[k], value, new_keys_allowed)
191         # Original value not a dict OR new value not a dict:
192         # Override entire value.
193         else:
194             original[k] = value
195     return original
196 
197 
198 def flatten_dict(dt, delimiter="/"):
199     dt = copy.deepcopy(dt)
200     while any(isinstance(v, dict) for v in dt.values()):
201         remove = []
202         add = {}
203         for key, value in dt.items():
204             if isinstance(value, dict):
205                 for subkey, v in value.items():
206                     add[delimiter.join([key, subkey])] = v
207                 remove.append(key)
208         dt.update(add)
209         for k in remove:
210             del dt[k]
211     return dt
212 
213 
214 def _to_pinnable(obj):
215     """Converts obj to a form that can be pinned in object store memory.
216 
217     Currently only numpy arrays are pinned in memory, if you have a strong
218     reference to the array value.
219     """
220 
221     return (obj, np.zeros(1))
222 
223 
224 def _from_pinnable(obj):
225     """Retrieve from _to_pinnable format."""
226 
227     return obj[0]
228 
229 
230 def validate_save_restore(trainable_cls,
231                           config=None,
232                           num_gpus=0,
233                           use_object_store=False):
234     """Helper method to check if your Trainable class will resume correctly.
235 
236     Args:
237         trainable_cls: Trainable class for evaluation.
238         config (dict): Config to pass to Trainable when testing.
239         num_gpus (int): GPU resources to allocate when testing.
240         use_object_store (bool): Whether to save and restore to Ray's object
241             store. Recommended to set this to True if planning to use
242             algorithms that pause training (i.e., PBT, HyperBand).
243     """
244     assert ray.is_initialized(), "Need Ray to be initialized."
245     remote_cls = ray.remote(num_gpus=num_gpus)(trainable_cls)
246     trainable_1 = remote_cls.remote(config=config)
247     trainable_2 = remote_cls.remote(config=config)
248 
249     from ray.tune.result import TRAINING_ITERATION
250 
251     for _ in range(3):
252         res = ray.get(trainable_1.train.remote())
253 
254     assert res.get(TRAINING_ITERATION), (
255         "Validation will not pass because it requires `training_iteration` "
256         "to be returned.")
257 
258     if use_object_store:
259         restore_check = trainable_2.restore_from_object.remote(
260             trainable_1.save_to_object.remote())
261         ray.get(restore_check)
262     else:
263         restore_check = ray.get(
264             trainable_2.restore.remote(trainable_1.save.remote()))
265 
266     res = ray.get(trainable_2.train.remote())
267     assert res[TRAINING_ITERATION] == 4
268 
269     res = ray.get(trainable_2.train.remote())
270     assert res[TRAINING_ITERATION] == 5
271     return True
272 
273 
274 if __name__ == "__main__":
275     ray.init()
276     X = pin_in_object_store("hello")
277     print(X)
278     result = get_pinned_object(X)
279     print(result)
280
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/python/ray/tune/utils/util.py b/python/ray/tune/utils/util.py
--- a/python/ray/tune/utils/util.py
+++ b/python/ray/tune/utils/util.py
@@ -58,7 +58,12 @@
                 self.values["ram_util_percent"].append(
                     float(getattr(psutil.virtual_memory(), "percent")))
             if GPUtil is not None:
-                for gpu in GPUtil.getGPUs():
+                gpu_list = []
+                try:
+                    gpu_list = GPUtil.getGPUs()
+                except Exception:
+                    logger.debug("GPUtil failed to retrieve GPUs.")
+                for gpu in gpu_list:
                     self.values["gpu_util_percent" + str(gpu.id)].append(
                         float(gpu.load))
                     self.values["vram_util_percent" + str(gpu.id)].append(

{"golden_diff": "diff --git a/python/ray/tune/utils/util.py b/python/ray/tune/utils/util.py\n--- a/python/ray/tune/utils/util.py\n+++ b/python/ray/tune/utils/util.py\n@@ -58,7 +58,12 @@\n                 self.values[\"ram_util_percent\"].append(\n                     float(getattr(psutil.virtual_memory(), \"percent\")))\n             if GPUtil is not None:\n-                for gpu in GPUtil.getGPUs():\n+                gpu_list = []\n+                try:\n+                    gpu_list = GPUtil.getGPUs()\n+                except Exception:\n+                    logger.debug(\"GPUtil failed to retrieve GPUs.\")\n+                for gpu in gpu_list:\n                     self.values[\"gpu_util_percent\" + str(gpu.id)].append(\n                         float(gpu.load))\n                     self.values[\"vram_util_percent\" + str(gpu.id)].append(\n", "issue": "[tune] GPU utilization getting checked for cluster with no gpu\n<!--Please include [tune], [rllib], [autoscaler] etc. in the issue title if relevant-->\r\n\r\n### What is the problem?\r\n\r\nTrivial issue, which does not actually impact performance as far as I can tell, but it seems if you launch a cluster with a configuration that sets gpus = 0, then there might be a check which avoids this error message...\r\n\r\n\r\n```\r\n(pid=4628) Exception in thread Thread-2: \r\n(pid=4628) Traceback (most recent call last):\r\n(pid=4628)   File \"/usr/lib/python3.6/threading.py\", line 916, in _bootstrap_inner\r\n(pid=4628)     self.run()\r\n(pid=4628)   File \"/home/ubuntu/algo/lib/python3.6/site-packages/ray/tune/utils/util.py\", line 89, in run\r\n(pid=4628)     self._read_utilization()\r\n(pid=4628)   File \"/home/ubuntu/algo/lib/python3.6/site-packages/ray/tune/utils/util.py\", line 65, in _read_utilization\r\n(pid=4628)     for gpu in GPUtil.getGPUs():\r\n(pid=4628)   File \"/home/ubuntu/algo/lib/python3.6/site-packages/GPUtil/GPUtil.py\", line 102, in getGPUs\r\n(pid=4628)     deviceIds = int(vals[i])\r\n(pid=4628) ValueError: invalid literal for int() with base 10: \"NVIDIA-SMI has failed because it couldn't communicate with the NVIDIA driver. Make  sure that the latest NVIDIA driver is installed and running.\"\r\n```\r\n\r\n\r\n\r\n*Ray version and other system information (Python version, TensorFlow version, OS):*\r\nRay 0.8.1, TF 2.1, RHEL 7.7\r\n\r\n### Reproduction (REQUIRED)\r\nPlease provide a script that can be run to reproduce the issue. The script should have **no external library dependencies** (i.e., use fake or mock data / environments):\r\n\r\nIf we cannot run your script, we cannot fix your issue.\r\n\r\n- [ ] I have verified my script runs in a clean environment and reproduces the issue.\r\n- [ ] I have verified the issue also occurs with the [latest wheels](https://ray.readthedocs.io/en/latest/installation.html).\r\n\n", "before_files": [{"content": "import copy\nimport logging\nimport threading\nimport time\nfrom collections import defaultdict\nfrom threading import Thread\n\nimport numpy as np\nimport ray\nimport psutil\n\nlogger = logging.getLogger(__name__)\n\ntry:\n    import GPUtil\nexcept ImportError:\n    GPUtil = None\n\n_pinned_objects = []\nPINNED_OBJECT_PREFIX = \"ray.tune.PinnedObject:\"\nSTART_OF_TIME = time.time()\n\n\nclass UtilMonitor(Thread):\n    \"\"\"Class for system usage utilization monitoring.\n\n    It keeps track of CPU, RAM, GPU, VRAM usage (each gpu separately) by\n    pinging for information every x seconds in a separate thread.\n\n    Requires psutil and GPUtil to be installed. Can be enabled with\n    tune.run(config={\"log_sys_usage\": True}).\n    \"\"\"\n\n    def __init__(self, start=True, delay=0.7):\n        self.stopped = True\n        if GPUtil is None and start:\n            logger.warning(\"Install gputil for GPU system monitoring.\")\n\n        if psutil is None and start:\n            logger.warning(\"Install psutil to monitor system performance.\")\n\n        if GPUtil is None and psutil is None:\n            return\n\n        super(UtilMonitor, self).__init__()\n        self.delay = delay  # Time between calls to GPUtil\n        self.values = defaultdict(list)\n        self.lock = threading.Lock()\n        self.daemon = True\n        if start:\n            self.start()\n\n    def _read_utilization(self):\n        with self.lock:\n            if psutil is not None:\n                self.values[\"cpu_util_percent\"].append(\n                    float(psutil.cpu_percent(interval=None)))\n                self.values[\"ram_util_percent\"].append(\n                    float(getattr(psutil.virtual_memory(), \"percent\")))\n            if GPUtil is not None:\n                for gpu in GPUtil.getGPUs():\n                    self.values[\"gpu_util_percent\" + str(gpu.id)].append(\n                        float(gpu.load))\n                    self.values[\"vram_util_percent\" + str(gpu.id)].append(\n                        float(gpu.memoryUtil))\n\n    def get_data(self):\n        if self.stopped:\n            return {}\n\n        with self.lock:\n            ret_values = copy.deepcopy(self.values)\n            for key, val in self.values.items():\n                del val[:]\n        return {\n            \"perf\": {\n                k: np.mean(v)\n                for k, v in ret_values.items() if len(v) > 0\n            }\n        }\n\n    def run(self):\n        self.stopped = False\n        while not self.stopped:\n            self._read_utilization()\n            time.sleep(self.delay)\n\n    def stop(self):\n        self.stopped = True\n\n\ndef pin_in_object_store(obj):\n    \"\"\"Deprecated, use ray.put(value, weakref=False) instead.\"\"\"\n\n    obj_id = ray.put(obj, weakref=False)\n    _pinned_objects.append(obj_id)\n    return obj_id\n\n\ndef get_pinned_object(pinned_id):\n    \"\"\"Deprecated.\"\"\"\n\n    return ray.get(pinned_id)\n\n\nclass warn_if_slow:\n    \"\"\"Prints a warning if a given operation is slower than 100ms.\n\n    Example:\n        >>> with warn_if_slow(\"some_operation\"):\n        ...    ray.get(something)\n    \"\"\"\n\n    DEFAULT_THRESHOLD = 0.5\n\n    def __init__(self, name, threshold=None):\n        self.name = name\n        self.threshold = threshold or self.DEFAULT_THRESHOLD\n        self.too_slow = False\n\n    def __enter__(self):\n        self.start = time.time()\n        return self\n\n    def __exit__(self, type, value, traceback):\n        now = time.time()\n        if now - self.start > self.threshold and now - START_OF_TIME > 60.0:\n            self.too_slow = True\n            logger.warning(\n                \"The `%s` operation took %s seconds to complete, \"\n                \"which may be a performance bottleneck.\", self.name,\n                now - self.start)\n\n\ndef merge_dicts(d1, d2):\n    \"\"\"\n    Args:\n        d1 (dict): Dict 1.\n        d2 (dict): Dict 2.\n\n    Returns:\n         dict: A new dict that is d1 and d2 deep merged.\n    \"\"\"\n    merged = copy.deepcopy(d1)\n    deep_update(merged, d2, True, [])\n    return merged\n\n\ndef deep_update(original,\n                new_dict,\n                new_keys_allowed=False,\n                whitelist=None,\n                override_all_if_type_changes=None):\n    \"\"\"Updates original dict with values from new_dict recursively.\n\n    If new key is introduced in new_dict, then if new_keys_allowed is not\n    True, an error will be thrown. Further, for sub-dicts, if the key is\n    in the whitelist, then new subkeys can be introduced.\n\n    Args:\n        original (dict): Dictionary with default values.\n        new_dict (dict): Dictionary with values to be updated\n        new_keys_allowed (bool): Whether new keys are allowed.\n        whitelist (Optional[List[str]]): List of keys that correspond to dict\n            values where new subkeys can be introduced. This is only at the top\n            level.\n        override_all_if_type_changes(Optional[List[str]]): List of top level\n            keys with value=dict, for which we always simply override the\n            entire value (dict), iff the \"type\" key in that value dict changes.\n    \"\"\"\n    whitelist = whitelist or []\n    override_all_if_type_changes = override_all_if_type_changes or []\n\n    for k, value in new_dict.items():\n        if k not in original and not new_keys_allowed:\n            raise Exception(\"Unknown config parameter `{}` \".format(k))\n\n        # Both orginal value and new one are dicts.\n        if isinstance(original.get(k), dict) and isinstance(value, dict):\n            # Check old type vs old one. If different, override entire value.\n            if k in override_all_if_type_changes and \\\n                    \"type\" in value and \"type\" in original[k] and \\\n                    value[\"type\"] != original[k][\"type\"]:\n                original[k] = value\n            # Whitelisted key -> ok to add new subkeys.\n            elif k in whitelist:\n                deep_update(original[k], value, True)\n            # Non-whitelisted key.\n            else:\n                deep_update(original[k], value, new_keys_allowed)\n        # Original value not a dict OR new value not a dict:\n        # Override entire value.\n        else:\n            original[k] = value\n    return original\n\n\ndef flatten_dict(dt, delimiter=\"/\"):\n    dt = copy.deepcopy(dt)\n    while any(isinstance(v, dict) for v in dt.values()):\n        remove = []\n        add = {}\n        for key, value in dt.items():\n            if isinstance(value, dict):\n                for subkey, v in value.items():\n                    add[delimiter.join([key, subkey])] = v\n                remove.append(key)\n        dt.update(add)\n        for k in remove:\n            del dt[k]\n    return dt\n\n\ndef _to_pinnable(obj):\n    \"\"\"Converts obj to a form that can be pinned in object store memory.\n\n    Currently only numpy arrays are pinned in memory, if you have a strong\n    reference to the array value.\n    \"\"\"\n\n    return (obj, np.zeros(1))\n\n\ndef _from_pinnable(obj):\n    \"\"\"Retrieve from _to_pinnable format.\"\"\"\n\n    return obj[0]\n\n\ndef validate_save_restore(trainable_cls,\n                          config=None,\n                          num_gpus=0,\n                          use_object_store=False):\n    \"\"\"Helper method to check if your Trainable class will resume correctly.\n\n    Args:\n        trainable_cls: Trainable class for evaluation.\n        config (dict): Config to pass to Trainable when testing.\n        num_gpus (int): GPU resources to allocate when testing.\n        use_object_store (bool): Whether to save and restore to Ray's object\n            store. Recommended to set this to True if planning to use\n            algorithms that pause training (i.e., PBT, HyperBand).\n    \"\"\"\n    assert ray.is_initialized(), \"Need Ray to be initialized.\"\n    remote_cls = ray.remote(num_gpus=num_gpus)(trainable_cls)\n    trainable_1 = remote_cls.remote(config=config)\n    trainable_2 = remote_cls.remote(config=config)\n\n    from ray.tune.result import TRAINING_ITERATION\n\n    for _ in range(3):\n        res = ray.get(trainable_1.train.remote())\n\n    assert res.get(TRAINING_ITERATION), (\n        \"Validation will not pass because it requires `training_iteration` \"\n        \"to be returned.\")\n\n    if use_object_store:\n        restore_check = trainable_2.restore_from_object.remote(\n            trainable_1.save_to_object.remote())\n        ray.get(restore_check)\n    else:\n        restore_check = ray.get(\n            trainable_2.restore.remote(trainable_1.save.remote()))\n\n    res = ray.get(trainable_2.train.remote())\n    assert res[TRAINING_ITERATION] == 4\n\n    res = ray.get(trainable_2.train.remote())\n    assert res[TRAINING_ITERATION] == 5\n    return True\n\n\nif __name__ == \"__main__\":\n    ray.init()\n    X = pin_in_object_store(\"hello\")\n    print(X)\n    result = get_pinned_object(X)\n    print(result)\n", "path": "python/ray/tune/utils/util.py"}], "after_files": [{"content": "import copy\nimport logging\nimport threading\nimport time\nfrom collections import defaultdict\nfrom threading import Thread\n\nimport numpy as np\nimport ray\nimport psutil\n\nlogger = logging.getLogger(__name__)\n\ntry:\n    import GPUtil\nexcept ImportError:\n    GPUtil = None\n\n_pinned_objects = []\nPINNED_OBJECT_PREFIX = \"ray.tune.PinnedObject:\"\nSTART_OF_TIME = time.time()\n\n\nclass UtilMonitor(Thread):\n    \"\"\"Class for system usage utilization monitoring.\n\n    It keeps track of CPU, RAM, GPU, VRAM usage (each gpu separately) by\n    pinging for information every x seconds in a separate thread.\n\n    Requires psutil and GPUtil to be installed. Can be enabled with\n    tune.run(config={\"log_sys_usage\": True}).\n    \"\"\"\n\n    def __init__(self, start=True, delay=0.7):\n        self.stopped = True\n        if GPUtil is None and start:\n            logger.warning(\"Install gputil for GPU system monitoring.\")\n\n        if psutil is None and start:\n            logger.warning(\"Install psutil to monitor system performance.\")\n\n        if GPUtil is None and psutil is None:\n            return\n\n        super(UtilMonitor, self).__init__()\n        self.delay = delay  # Time between calls to GPUtil\n        self.values = defaultdict(list)\n        self.lock = threading.Lock()\n        self.daemon = True\n        if start:\n            self.start()\n\n    def _read_utilization(self):\n        with self.lock:\n            if psutil is not None:\n                self.values[\"cpu_util_percent\"].append(\n                    float(psutil.cpu_percent(interval=None)))\n                self.values[\"ram_util_percent\"].append(\n                    float(getattr(psutil.virtual_memory(), \"percent\")))\n            if GPUtil is not None:\n                gpu_list = []\n                try:\n                    gpu_list = GPUtil.getGPUs()\n                except Exception:\n                    logger.debug(\"GPUtil failed to retrieve GPUs.\")\n                for gpu in gpu_list:\n                    self.values[\"gpu_util_percent\" + str(gpu.id)].append(\n                        float(gpu.load))\n                    self.values[\"vram_util_percent\" + str(gpu.id)].append(\n                        float(gpu.memoryUtil))\n\n    def get_data(self):\n        if self.stopped:\n            return {}\n\n        with self.lock:\n            ret_values = copy.deepcopy(self.values)\n            for key, val in self.values.items():\n                del val[:]\n        return {\n            \"perf\": {\n                k: np.mean(v)\n                for k, v in ret_values.items() if len(v) > 0\n            }\n        }\n\n    def run(self):\n        self.stopped = False\n        while not self.stopped:\n            self._read_utilization()\n            time.sleep(self.delay)\n\n    def stop(self):\n        self.stopped = True\n\n\ndef pin_in_object_store(obj):\n    \"\"\"Deprecated, use ray.put(value, weakref=False) instead.\"\"\"\n\n    obj_id = ray.put(obj, weakref=False)\n    _pinned_objects.append(obj_id)\n    return obj_id\n\n\ndef get_pinned_object(pinned_id):\n    \"\"\"Deprecated.\"\"\"\n\n    return ray.get(pinned_id)\n\n\nclass warn_if_slow:\n    \"\"\"Prints a warning if a given operation is slower than 100ms.\n\n    Example:\n        >>> with warn_if_slow(\"some_operation\"):\n        ...    ray.get(something)\n    \"\"\"\n\n    DEFAULT_THRESHOLD = 0.5\n\n    def __init__(self, name, threshold=None):\n        self.name = name\n        self.threshold = threshold or self.DEFAULT_THRESHOLD\n        self.too_slow = False\n\n    def __enter__(self):\n        self.start = time.time()\n        return self\n\n    def __exit__(self, type, value, traceback):\n        now = time.time()\n        if now - self.start > self.threshold and now - START_OF_TIME > 60.0:\n            self.too_slow = True\n            logger.warning(\n                \"The `%s` operation took %s seconds to complete, \"\n                \"which may be a performance bottleneck.\", self.name,\n                now - self.start)\n\n\ndef merge_dicts(d1, d2):\n    \"\"\"\n    Args:\n        d1 (dict): Dict 1.\n        d2 (dict): Dict 2.\n\n    Returns:\n         dict: A new dict that is d1 and d2 deep merged.\n    \"\"\"\n    merged = copy.deepcopy(d1)\n    deep_update(merged, d2, True, [])\n    return merged\n\n\ndef deep_update(original,\n                new_dict,\n                new_keys_allowed=False,\n                whitelist=None,\n                override_all_if_type_changes=None):\n    \"\"\"Updates original dict with values from new_dict recursively.\n\n    If new key is introduced in new_dict, then if new_keys_allowed is not\n    True, an error will be thrown. Further, for sub-dicts, if the key is\n    in the whitelist, then new subkeys can be introduced.\n\n    Args:\n        original (dict): Dictionary with default values.\n        new_dict (dict): Dictionary with values to be updated\n        new_keys_allowed (bool): Whether new keys are allowed.\n        whitelist (Optional[List[str]]): List of keys that correspond to dict\n            values where new subkeys can be introduced. This is only at the top\n            level.\n        override_all_if_type_changes(Optional[List[str]]): List of top level\n            keys with value=dict, for which we always simply override the\n            entire value (dict), iff the \"type\" key in that value dict changes.\n    \"\"\"\n    whitelist = whitelist or []\n    override_all_if_type_changes = override_all_if_type_changes or []\n\n    for k, value in new_dict.items():\n        if k not in original and not new_keys_allowed:\n            raise Exception(\"Unknown config parameter `{}` \".format(k))\n\n        # Both orginal value and new one are dicts.\n        if isinstance(original.get(k), dict) and isinstance(value, dict):\n            # Check old type vs old one. If different, override entire value.\n            if k in override_all_if_type_changes and \\\n                    \"type\" in value and \"type\" in original[k] and \\\n                    value[\"type\"] != original[k][\"type\"]:\n                original[k] = value\n            # Whitelisted key -> ok to add new subkeys.\n            elif k in whitelist:\n                deep_update(original[k], value, True)\n            # Non-whitelisted key.\n            else:\n                deep_update(original[k], value, new_keys_allowed)\n        # Original value not a dict OR new value not a dict:\n        # Override entire value.\n        else:\n            original[k] = value\n    return original\n\n\ndef flatten_dict(dt, delimiter=\"/\"):\n    dt = copy.deepcopy(dt)\n    while any(isinstance(v, dict) for v in dt.values()):\n        remove = []\n        add = {}\n        for key, value in dt.items():\n            if isinstance(value, dict):\n                for subkey, v in value.items():\n                    add[delimiter.join([key, subkey])] = v\n                remove.append(key)\n        dt.update(add)\n        for k in remove:\n            del dt[k]\n    return dt\n\n\ndef _to_pinnable(obj):\n    \"\"\"Converts obj to a form that can be pinned in object store memory.\n\n    Currently only numpy arrays are pinned in memory, if you have a strong\n    reference to the array value.\n    \"\"\"\n\n    return (obj, np.zeros(1))\n\n\ndef _from_pinnable(obj):\n    \"\"\"Retrieve from _to_pinnable format.\"\"\"\n\n    return obj[0]\n\n\ndef validate_save_restore(trainable_cls,\n                          config=None,\n                          num_gpus=0,\n                          use_object_store=False):\n    \"\"\"Helper method to check if your Trainable class will resume correctly.\n\n    Args:\n        trainable_cls: Trainable class for evaluation.\n        config (dict): Config to pass to Trainable when testing.\n        num_gpus (int): GPU resources to allocate when testing.\n        use_object_store (bool): Whether to save and restore to Ray's object\n            store. Recommended to set this to True if planning to use\n            algorithms that pause training (i.e., PBT, HyperBand).\n    \"\"\"\n    assert ray.is_initialized(), \"Need Ray to be initialized.\"\n    remote_cls = ray.remote(num_gpus=num_gpus)(trainable_cls)\n    trainable_1 = remote_cls.remote(config=config)\n    trainable_2 = remote_cls.remote(config=config)\n\n    from ray.tune.result import TRAINING_ITERATION\n\n    for _ in range(3):\n        res = ray.get(trainable_1.train.remote())\n\n    assert res.get(TRAINING_ITERATION), (\n        \"Validation will not pass because it requires `training_iteration` \"\n        \"to be returned.\")\n\n    if use_object_store:\n        restore_check = trainable_2.restore_from_object.remote(\n            trainable_1.save_to_object.remote())\n        ray.get(restore_check)\n    else:\n        restore_check = ray.get(\n            trainable_2.restore.remote(trainable_1.save.remote()))\n\n    res = ray.get(trainable_2.train.remote())\n    assert res[TRAINING_ITERATION] == 4\n\n    res = ray.get(trainable_2.train.remote())\n    assert res[TRAINING_ITERATION] == 5\n    return True\n\n\nif __name__ == \"__main__\":\n    ray.init()\n    X = pin_in_object_store(\"hello\")\n    print(X)\n    result = get_pinned_object(X)\n    print(result)\n", "path": "python/ray/tune/utils/util.py"}]}