We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[tune] Population-based training: broken when using keep_checkpoint_num
When using **population-based** training TUNE stops after some times throwing the following error:

`There are paused trials, but no more pending trials with sufficient resources.`

This is caused by not finding the latest checkpoint:

```
Failure # 1 (occurred at 2020-06-19_11-26-36)
Traceback (most recent call last):
  File "/usr/local/lib/python3.6/dist-packages/ray/tune/ray_trial_executor.py", line 294, in start_trial
    self._start_trial(trial, checkpoint, train=train)
  File "/usr/local/lib/python3.6/dist-packages/ray/tune/ray_trial_executor.py", line 235, in _start_trial
    self.restore(trial, checkpoint)
  File "/usr/local/lib/python3.6/dist-packages/ray/tune/ray_trial_executor.py", line 673, in restore
    data_dict = TrainableUtil.pickle_checkpoint(value)
  File "/usr/local/lib/python3.6/dist-packages/ray/tune/trainable.py", line 62, in pickle_checkpoint
    checkpoint_dir = TrainableUtil.find_checkpoint_dir(checkpoint_path)
  File "/usr/local/lib/python3.6/dist-packages/ray/tune/trainable.py", line 87, in find_checkpoint_dir
    raise FileNotFoundError("Path does not exist", checkpoint_path)
FileNotFoundError: [Errno Path does not exist] /content/TRASH_TUNE_PBT_oversampling_mimic_densenet121/TUNE_Model_0_2020-06-19_11-24-215xncry9c/checkpoint_6/
```

The error appears to be somewhat random since it only appears after quite some iterations

The error can be reproduced in this [colab notebook](https://colab.research.google.com/drive/1-o896bEUm7DTvS24Do0btlqbSHre49MH?usp=sharing). **It is not a COLAB related issue since the same problem arises on our own server.**

@richardliaw Is this related to #8772 ?

--- 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/checkpoint_manager.py`
Content:
```
1 # coding: utf-8
2 import heapq
3 import logging
4 
5 from ray.tune.result import TRAINING_ITERATION
6 
7 logger = logging.getLogger(__name__)
8 
9 
10 class Checkpoint:
11     """Describes a checkpoint of trial state.
12 
13     Checkpoint may be saved in different storage.
14 
15     Attributes:
16         storage (str): Storage type.
17         value (str): If storage==MEMORY, it is a Python object.
18             If storage==PERSISTENT, it is a path to persistent storage,
19             or a future that will be resolved to such a path.
20     """
21 
22     MEMORY = "memory"
23     PERSISTENT = "persistent"
24 
25     def __init__(self, storage, value, result=None):
26         self.storage = storage
27         self.value = value
28         self.result = result or {}
29 
30     @staticmethod
31     def from_object(value=None):
32         """Creates a checkpoint from a Python object."""
33         return Checkpoint(Checkpoint.MEMORY, value)
34 
35     @property
36     def is_ready(self):
37         """Returns whether the checkpoint is ready to be used for restoration.
38 
39         A PERSISTENT checkpoint is considered ready once its value is resolved
40         to an actual path. MEMORY checkpoints are always considered ready since
41         they are transient.
42         """
43         if self.storage == Checkpoint.PERSISTENT:
44             return isinstance(self.value, str)
45         return self.storage == Checkpoint.MEMORY
46 
47 
48 class QueueItem:
49     def __init__(self, priority, value):
50         self.priority = priority
51         self.value = value
52 
53     def __lt__(self, other):
54         return self.priority < other.priority
55 
56 
57 class CheckpointManager:
58     """Manages checkpoints on the driver for a trial."""
59 
60     def __init__(self, keep_checkpoints_num, checkpoint_score_attr, delete_fn):
61         """Initializes a new CheckpointManager.
62 
63         `newest_persistent_checkpoint` and `newest_memory_checkpoint` are
64         initialized to Checkpoint objects with values of None.
65 
66         Args:
67             keep_checkpoints_num (int): Keep at least this many checkpoints.
68             checkpoint_score_attr (str): Attribute to use to determine which
69                 checkpoints to keep.
70             delete_fn (function): Function that deletes checkpoints. Must be
71                 idempotent.
72         """
73         self.keep_checkpoints_num = keep_checkpoints_num or float("inf")
74         assert self.keep_checkpoints_num > 0, (
75             "keep_checkpoints_num must be greater than 0.")
76         self._checkpoint_score_desc = checkpoint_score_attr.startswith("min-")
77         if self._checkpoint_score_desc:
78             self._checkpoint_score_attr = checkpoint_score_attr[4:]
79         else:
80             self._checkpoint_score_attr = checkpoint_score_attr
81 
82         self.delete = delete_fn
83         self.newest_persistent_checkpoint = Checkpoint(Checkpoint.PERSISTENT,
84                                                        None)
85         self.newest_memory_checkpoint = Checkpoint(Checkpoint.MEMORY, None)
86         self._best_checkpoints = []
87         self._membership = set()
88 
89     @property
90     def newest_checkpoint(self):
91         """Returns the newest checkpoint (based on training iteration)."""
92         newest_checkpoint = max(
93             [self.newest_persistent_checkpoint, self.newest_memory_checkpoint],
94             key=lambda c: c.result.get(TRAINING_ITERATION, -1))
95         return newest_checkpoint
96 
97     def on_checkpoint(self, checkpoint):
98         """Starts tracking checkpoint metadata on checkpoint.
99 
100         Sets the newest checkpoint. For PERSISTENT checkpoints: Deletes
101         previous checkpoint as long as it isn't one of the best ones. Also
102         deletes the worst checkpoint if at capacity.
103 
104         Args:
105             checkpoint (Checkpoint): Trial state checkpoint.
106         """
107         if checkpoint.storage == Checkpoint.MEMORY:
108             self.newest_memory_checkpoint = checkpoint
109             return
110 
111         old_checkpoint = self.newest_persistent_checkpoint
112         self.newest_persistent_checkpoint = checkpoint
113 
114         # Remove the old checkpoint if it isn't one of the best ones.
115         if old_checkpoint.value and old_checkpoint not in self._membership:
116             self.delete(old_checkpoint)
117 
118         try:
119             queue_item = QueueItem(self._priority(checkpoint), checkpoint)
120         except KeyError:
121             logger.error("Result dict has no key: {}. "
122                          "checkpoint_score_attr must be set to a key in the "
123                          "result dict.".format(self._checkpoint_score_attr))
124             return
125 
126         if len(self._best_checkpoints) < self.keep_checkpoints_num:
127             heapq.heappush(self._best_checkpoints, queue_item)
128             self._membership.add(checkpoint)
129         elif queue_item.priority >= self._best_checkpoints[0].priority:
130             worst = heapq.heappushpop(self._best_checkpoints, queue_item).value
131             self._membership.add(checkpoint)
132             if worst in self._membership:
133                 self._membership.remove(worst)
134             # Don't delete the newest checkpoint. It will be deleted on the
135             # next on_checkpoint() call since it isn't in self._membership.
136             if worst != checkpoint:
137                 self.delete(worst)
138 
139     def best_checkpoints(self):
140         """Returns best PERSISTENT checkpoints, sorted by score."""
141         checkpoints = sorted(self._best_checkpoints, key=lambda c: c.priority)
142         return [queue_item.value for queue_item in checkpoints]
143 
144     def _priority(self, checkpoint):
145         priority = checkpoint.result[self._checkpoint_score_attr]
146         return -priority if self._checkpoint_score_desc else priority
147 
148     def __getstate__(self):
149         state = self.__dict__.copy()
150         # Avoid serializing lambda since it may capture cyclical dependencies.
151         state.pop("delete")
152         return state
153 
154     def __setstate__(self, state):
155         self.__dict__.update(state)
156         self.delete = None
157
```
--- 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/checkpoint_manager.py b/python/ray/tune/checkpoint_manager.py
--- a/python/ray/tune/checkpoint_manager.py
+++ b/python/ray/tune/checkpoint_manager.py
@@ -109,6 +109,10 @@
             return
 
         old_checkpoint = self.newest_persistent_checkpoint
+
+        if old_checkpoint.value == checkpoint.value:
+            return
+
         self.newest_persistent_checkpoint = checkpoint
 
         # Remove the old checkpoint if it isn't one of the best ones.

{"golden_diff": "diff --git a/python/ray/tune/checkpoint_manager.py b/python/ray/tune/checkpoint_manager.py\n--- a/python/ray/tune/checkpoint_manager.py\n+++ b/python/ray/tune/checkpoint_manager.py\n@@ -109,6 +109,10 @@\n             return\n \n         old_checkpoint = self.newest_persistent_checkpoint\n+\n+        if old_checkpoint.value == checkpoint.value:\n+            return\n+\n         self.newest_persistent_checkpoint = checkpoint\n \n         # Remove the old checkpoint if it isn't one of the best ones.\n", "issue": "[tune] Population-based training: broken when using keep_checkpoint_num\nWhen using **population-based** training TUNE stops after some times throwing the following error:\r\n\r\n`There are paused trials, but no more pending trials with sufficient resources.`\r\n\r\nThis is caused by not finding the latest checkpoint:\r\n\r\n```\r\nFailure # 1 (occurred at 2020-06-19_11-26-36)\r\nTraceback (most recent call last):\r\n  File \"/usr/local/lib/python3.6/dist-packages/ray/tune/ray_trial_executor.py\", line 294, in start_trial\r\n    self._start_trial(trial, checkpoint, train=train)\r\n  File \"/usr/local/lib/python3.6/dist-packages/ray/tune/ray_trial_executor.py\", line 235, in _start_trial\r\n    self.restore(trial, checkpoint)\r\n  File \"/usr/local/lib/python3.6/dist-packages/ray/tune/ray_trial_executor.py\", line 673, in restore\r\n    data_dict = TrainableUtil.pickle_checkpoint(value)\r\n  File \"/usr/local/lib/python3.6/dist-packages/ray/tune/trainable.py\", line 62, in pickle_checkpoint\r\n    checkpoint_dir = TrainableUtil.find_checkpoint_dir(checkpoint_path)\r\n  File \"/usr/local/lib/python3.6/dist-packages/ray/tune/trainable.py\", line 87, in find_checkpoint_dir\r\n    raise FileNotFoundError(\"Path does not exist\", checkpoint_path)\r\nFileNotFoundError: [Errno Path does not exist] /content/TRASH_TUNE_PBT_oversampling_mimic_densenet121/TUNE_Model_0_2020-06-19_11-24-215xncry9c/checkpoint_6/\r\n```\r\n\r\nThe error appears to be somewhat random since it only appears after quite some iterations\r\n\r\nThe error can be reproduced in this [colab notebook](https://colab.research.google.com/drive/1-o896bEUm7DTvS24Do0btlqbSHre49MH?usp=sharing). **It is not a COLAB related issue since the same problem arises on our own server.**\r\n\r\n@richardliaw Is this related to #8772 ?\n", "before_files": [{"content": "# coding: utf-8\nimport heapq\nimport logging\n\nfrom ray.tune.result import TRAINING_ITERATION\n\nlogger = logging.getLogger(__name__)\n\n\nclass Checkpoint:\n    \"\"\"Describes a checkpoint of trial state.\n\n    Checkpoint may be saved in different storage.\n\n    Attributes:\n        storage (str): Storage type.\n        value (str): If storage==MEMORY, it is a Python object.\n            If storage==PERSISTENT, it is a path to persistent storage,\n            or a future that will be resolved to such a path.\n    \"\"\"\n\n    MEMORY = \"memory\"\n    PERSISTENT = \"persistent\"\n\n    def __init__(self, storage, value, result=None):\n        self.storage = storage\n        self.value = value\n        self.result = result or {}\n\n    @staticmethod\n    def from_object(value=None):\n        \"\"\"Creates a checkpoint from a Python object.\"\"\"\n        return Checkpoint(Checkpoint.MEMORY, value)\n\n    @property\n    def is_ready(self):\n        \"\"\"Returns whether the checkpoint is ready to be used for restoration.\n\n        A PERSISTENT checkpoint is considered ready once its value is resolved\n        to an actual path. MEMORY checkpoints are always considered ready since\n        they are transient.\n        \"\"\"\n        if self.storage == Checkpoint.PERSISTENT:\n            return isinstance(self.value, str)\n        return self.storage == Checkpoint.MEMORY\n\n\nclass QueueItem:\n    def __init__(self, priority, value):\n        self.priority = priority\n        self.value = value\n\n    def __lt__(self, other):\n        return self.priority < other.priority\n\n\nclass CheckpointManager:\n    \"\"\"Manages checkpoints on the driver for a trial.\"\"\"\n\n    def __init__(self, keep_checkpoints_num, checkpoint_score_attr, delete_fn):\n        \"\"\"Initializes a new CheckpointManager.\n\n        `newest_persistent_checkpoint` and `newest_memory_checkpoint` are\n        initialized to Checkpoint objects with values of None.\n\n        Args:\n            keep_checkpoints_num (int): Keep at least this many checkpoints.\n            checkpoint_score_attr (str): Attribute to use to determine which\n                checkpoints to keep.\n            delete_fn (function): Function that deletes checkpoints. Must be\n                idempotent.\n        \"\"\"\n        self.keep_checkpoints_num = keep_checkpoints_num or float(\"inf\")\n        assert self.keep_checkpoints_num > 0, (\n            \"keep_checkpoints_num must be greater than 0.\")\n        self._checkpoint_score_desc = checkpoint_score_attr.startswith(\"min-\")\n        if self._checkpoint_score_desc:\n            self._checkpoint_score_attr = checkpoint_score_attr[4:]\n        else:\n            self._checkpoint_score_attr = checkpoint_score_attr\n\n        self.delete = delete_fn\n        self.newest_persistent_checkpoint = Checkpoint(Checkpoint.PERSISTENT,\n                                                       None)\n        self.newest_memory_checkpoint = Checkpoint(Checkpoint.MEMORY, None)\n        self._best_checkpoints = []\n        self._membership = set()\n\n    @property\n    def newest_checkpoint(self):\n        \"\"\"Returns the newest checkpoint (based on training iteration).\"\"\"\n        newest_checkpoint = max(\n            [self.newest_persistent_checkpoint, self.newest_memory_checkpoint],\n            key=lambda c: c.result.get(TRAINING_ITERATION, -1))\n        return newest_checkpoint\n\n    def on_checkpoint(self, checkpoint):\n        \"\"\"Starts tracking checkpoint metadata on checkpoint.\n\n        Sets the newest checkpoint. For PERSISTENT checkpoints: Deletes\n        previous checkpoint as long as it isn't one of the best ones. Also\n        deletes the worst checkpoint if at capacity.\n\n        Args:\n            checkpoint (Checkpoint): Trial state checkpoint.\n        \"\"\"\n        if checkpoint.storage == Checkpoint.MEMORY:\n            self.newest_memory_checkpoint = checkpoint\n            return\n\n        old_checkpoint = self.newest_persistent_checkpoint\n        self.newest_persistent_checkpoint = checkpoint\n\n        # Remove the old checkpoint if it isn't one of the best ones.\n        if old_checkpoint.value and old_checkpoint not in self._membership:\n            self.delete(old_checkpoint)\n\n        try:\n            queue_item = QueueItem(self._priority(checkpoint), checkpoint)\n        except KeyError:\n            logger.error(\"Result dict has no key: {}. \"\n                         \"checkpoint_score_attr must be set to a key in the \"\n                         \"result dict.\".format(self._checkpoint_score_attr))\n            return\n\n        if len(self._best_checkpoints) < self.keep_checkpoints_num:\n            heapq.heappush(self._best_checkpoints, queue_item)\n            self._membership.add(checkpoint)\n        elif queue_item.priority >= self._best_checkpoints[0].priority:\n            worst = heapq.heappushpop(self._best_checkpoints, queue_item).value\n            self._membership.add(checkpoint)\n            if worst in self._membership:\n                self._membership.remove(worst)\n            # Don't delete the newest checkpoint. It will be deleted on the\n            # next on_checkpoint() call since it isn't in self._membership.\n            if worst != checkpoint:\n                self.delete(worst)\n\n    def best_checkpoints(self):\n        \"\"\"Returns best PERSISTENT checkpoints, sorted by score.\"\"\"\n        checkpoints = sorted(self._best_checkpoints, key=lambda c: c.priority)\n        return [queue_item.value for queue_item in checkpoints]\n\n    def _priority(self, checkpoint):\n        priority = checkpoint.result[self._checkpoint_score_attr]\n        return -priority if self._checkpoint_score_desc else priority\n\n    def __getstate__(self):\n        state = self.__dict__.copy()\n        # Avoid serializing lambda since it may capture cyclical dependencies.\n        state.pop(\"delete\")\n        return state\n\n    def __setstate__(self, state):\n        self.__dict__.update(state)\n        self.delete = None\n", "path": "python/ray/tune/checkpoint_manager.py"}], "after_files": [{"content": "# coding: utf-8\nimport heapq\nimport logging\n\nfrom ray.tune.result import TRAINING_ITERATION\n\nlogger = logging.getLogger(__name__)\n\n\nclass Checkpoint:\n    \"\"\"Describes a checkpoint of trial state.\n\n    Checkpoint may be saved in different storage.\n\n    Attributes:\n        storage (str): Storage type.\n        value (str): If storage==MEMORY, it is a Python object.\n            If storage==PERSISTENT, it is a path to persistent storage,\n            or a future that will be resolved to such a path.\n    \"\"\"\n\n    MEMORY = \"memory\"\n    PERSISTENT = \"persistent\"\n\n    def __init__(self, storage, value, result=None):\n        self.storage = storage\n        self.value = value\n        self.result = result or {}\n\n    @staticmethod\n    def from_object(value=None):\n        \"\"\"Creates a checkpoint from a Python object.\"\"\"\n        return Checkpoint(Checkpoint.MEMORY, value)\n\n    @property\n    def is_ready(self):\n        \"\"\"Returns whether the checkpoint is ready to be used for restoration.\n\n        A PERSISTENT checkpoint is considered ready once its value is resolved\n        to an actual path. MEMORY checkpoints are always considered ready since\n        they are transient.\n        \"\"\"\n        if self.storage == Checkpoint.PERSISTENT:\n            return isinstance(self.value, str)\n        return self.storage == Checkpoint.MEMORY\n\n\nclass QueueItem:\n    def __init__(self, priority, value):\n        self.priority = priority\n        self.value = value\n\n    def __lt__(self, other):\n        return self.priority < other.priority\n\n\nclass CheckpointManager:\n    \"\"\"Manages checkpoints on the driver for a trial.\"\"\"\n\n    def __init__(self, keep_checkpoints_num, checkpoint_score_attr, delete_fn):\n        \"\"\"Initializes a new CheckpointManager.\n\n        `newest_persistent_checkpoint` and `newest_memory_checkpoint` are\n        initialized to Checkpoint objects with values of None.\n\n        Args:\n            keep_checkpoints_num (int): Keep at least this many checkpoints.\n            checkpoint_score_attr (str): Attribute to use to determine which\n                checkpoints to keep.\n            delete_fn (function): Function that deletes checkpoints. Must be\n                idempotent.\n        \"\"\"\n        self.keep_checkpoints_num = keep_checkpoints_num or float(\"inf\")\n        assert self.keep_checkpoints_num > 0, (\n            \"keep_checkpoints_num must be greater than 0.\")\n        self._checkpoint_score_desc = checkpoint_score_attr.startswith(\"min-\")\n        if self._checkpoint_score_desc:\n            self._checkpoint_score_attr = checkpoint_score_attr[4:]\n        else:\n            self._checkpoint_score_attr = checkpoint_score_attr\n\n        self.delete = delete_fn\n        self.newest_persistent_checkpoint = Checkpoint(Checkpoint.PERSISTENT,\n                                                       None)\n        self.newest_memory_checkpoint = Checkpoint(Checkpoint.MEMORY, None)\n        self._best_checkpoints = []\n        self._membership = set()\n\n    @property\n    def newest_checkpoint(self):\n        \"\"\"Returns the newest checkpoint (based on training iteration).\"\"\"\n        newest_checkpoint = max(\n            [self.newest_persistent_checkpoint, self.newest_memory_checkpoint],\n            key=lambda c: c.result.get(TRAINING_ITERATION, -1))\n        return newest_checkpoint\n\n    def on_checkpoint(self, checkpoint):\n        \"\"\"Starts tracking checkpoint metadata on checkpoint.\n\n        Sets the newest checkpoint. For PERSISTENT checkpoints: Deletes\n        previous checkpoint as long as it isn't one of the best ones. Also\n        deletes the worst checkpoint if at capacity.\n\n        Args:\n            checkpoint (Checkpoint): Trial state checkpoint.\n        \"\"\"\n        if checkpoint.storage == Checkpoint.MEMORY:\n            self.newest_memory_checkpoint = checkpoint\n            return\n\n        old_checkpoint = self.newest_persistent_checkpoint\n\n        if old_checkpoint.value == checkpoint.value:\n            return\n\n        self.newest_persistent_checkpoint = checkpoint\n\n        # Remove the old checkpoint if it isn't one of the best ones.\n        if old_checkpoint.value and old_checkpoint not in self._membership:\n            self.delete(old_checkpoint)\n\n        try:\n            queue_item = QueueItem(self._priority(checkpoint), checkpoint)\n        except KeyError:\n            logger.error(\"Result dict has no key: {}. \"\n                         \"checkpoint_score_attr must be set to a key in the \"\n                         \"result dict.\".format(self._checkpoint_score_attr))\n            return\n\n        if len(self._best_checkpoints) < self.keep_checkpoints_num:\n            heapq.heappush(self._best_checkpoints, queue_item)\n            self._membership.add(checkpoint)\n        elif queue_item.priority >= self._best_checkpoints[0].priority:\n            worst = heapq.heappushpop(self._best_checkpoints, queue_item).value\n            self._membership.add(checkpoint)\n            if worst in self._membership:\n                self._membership.remove(worst)\n            # Don't delete the newest checkpoint. It will be deleted on the\n            # next on_checkpoint() call since it isn't in self._membership.\n            if worst != checkpoint:\n                self.delete(worst)\n\n    def best_checkpoints(self):\n        \"\"\"Returns best PERSISTENT checkpoints, sorted by score.\"\"\"\n        checkpoints = sorted(self._best_checkpoints, key=lambda c: c.priority)\n        return [queue_item.value for queue_item in checkpoints]\n\n    def _priority(self, checkpoint):\n        priority = checkpoint.result[self._checkpoint_score_attr]\n        return -priority if self._checkpoint_score_desc else priority\n\n    def __getstate__(self):\n        state = self.__dict__.copy()\n        # Avoid serializing lambda since it may capture cyclical dependencies.\n        state.pop(\"delete\")\n        return state\n\n    def __setstate__(self, state):\n        self.__dict__.update(state)\n        self.delete = None\n", "path": "python/ray/tune/checkpoint_manager.py"}]}