wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-09-09 11:32:43 +08:00
Code Issues Packages Projects Releases Wiki Activity

[autoscaler] Small fixes for local cluster usability (#4864)

Browse Source
This commit is contained in:
Richard Liaw
2019-07-06 21:55:18 -07:00
committed by Eric Liang
parent 1798d4f077
commit 6a14f1a540
2 changed files with 22 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/ray/autoscaler/local/node_provider.py
+15 -4
View File
@@ -28,10 +28,15 @@ class ClusterState(object):
with self.file_lock:
if os.path.exists(self.save_path):
workers = json.loads(open(self.save_path).read())
head_config = workers.get(provider_config["head_ip"])
if not head_config or head_config.get(
"tags", {}).get(TAG_RAY_NODE_TYPE) != "head":
workers = {}
logger.info("Head IP changed - recreating cluster.")
else:
workers = {}
logger.info("ClusterState: "
"Loaded cluster state: {}".format(workers))
"Loaded cluster state: {}".format(list(workers)))
for worker_ip in provider_config["worker_ips"]:
if worker_ip not in workers:
workers[worker_ip] = {
@@ -55,8 +60,8 @@ class ClusterState(object):
TAG_RAY_NODE_TYPE] == "head"
assert len(workers) == len(provider_config["worker_ips"]) + 1
with open(self.save_path, "w") as f:
logger.info("ClusterState: "
"Writing cluster state: {}".format(workers))
logger.debug("ClusterState: "
"Writing cluster state: {}".format(workers))
f.write(json.dumps(workers))
def get(self):
@@ -74,11 +79,17 @@ class ClusterState(object):
workers[worker_id] = info
with open(self.save_path, "w") as f:
logger.info("ClusterState: "
"Writing cluster state: {}".format(workers))
"Writing cluster state: {}".format(
list(workers)))
f.write(json.dumps(workers))
class LocalNodeProvider(NodeProvider):
"""NodeProvider for private/local clusters.
`node_id` is overloaded to also be `node_ip` in this class.
"""
def __init__(self, provider_config, cluster_name):
NodeProvider.__init__(self, provider_config, cluster_name)
self.state = ClusterState("/tmp/cluster-{}.lock".format(cluster_name),