mirror of
https://github.com/wassname/ray.git
synced 2026-07-11 15:50:37 +08:00
[tune] Refactor syncer (#6496)
* Refactor syncer and log_sync. * Fix documentation. * Remove delete from api * Rename to get_node_syncer
This commit is contained in:
Ujval Misra
committed by
Richard Liaw
Richard Liaw
parent
7a24144bfd
commit
81197e47c7
@@ -1,107 +0,0 @@
|
||||
from __future__ import absolute_import
|
||||
from __future__ import division
|
||||
from __future__ import print_function
|
||||
|
||||
import distutils.spawn
|
||||
import logging
|
||||
|
||||
try: # py3
|
||||
from shlex import quote
|
||||
except ImportError: # py2
|
||||
from pipes import quote
|
||||
|
||||
import ray
|
||||
from ray.tune.cluster_info import get_ssh_key, get_ssh_user
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
_log_sync_warned = False
|
||||
|
||||
|
||||
def log_sync_template(options=""):
|
||||
"""Syncs the local_dir between driver and worker if possible.
|
||||
|
||||
Requires ray cluster to be started with the autoscaler. Also requires
|
||||
rsync to be installed.
|
||||
|
||||
Args:
|
||||
options (str): Addtional rsync options.
|
||||
|
||||
Returns:
|
||||
Sync template with source and target parameters.
|
||||
"""
|
||||
if not distutils.spawn.find_executable("rsync"):
|
||||
logger.error("Log sync requires rsync to be installed.")
|
||||
return
|
||||
global _log_sync_warned
|
||||
ssh_key = get_ssh_key()
|
||||
if ssh_key is None:
|
||||
if not _log_sync_warned:
|
||||
logger.debug("Log sync requires cluster to be setup with "
|
||||
"`ray up`.")
|
||||
_log_sync_warned = True
|
||||
return
|
||||
|
||||
rsh = "ssh -i {ssh_key} -o ConnectTimeout=120s -o StrictHostKeyChecking=no"
|
||||
rsh = rsh.format(ssh_key=quote(ssh_key))
|
||||
template = """rsync {options} -savz -e "{rsh}" {{source}} {{target}}"""
|
||||
return template.format(options=options, rsh=rsh)
|
||||
|
||||
|
||||
class NodeSyncMixin(object):
|
||||
# TODO(ujvl): Refactor this code.
|
||||
"""Mixin for syncing files to/from a remote dir to a local dir."""
|
||||
|
||||
def __init__(self):
|
||||
assert hasattr(self, "_remote_dir"), "Mixin not mixed with Syncer."
|
||||
self.local_ip = ray.services.get_node_ip_address()
|
||||
self.worker_ip = None
|
||||
|
||||
def set_worker_ip(self, worker_ip):
|
||||
"""Set the worker ip to sync logs from."""
|
||||
self.worker_ip = worker_ip
|
||||
|
||||
def has_remote_target(self):
|
||||
"""Returns whether the Syncer has a remote target."""
|
||||
if not self.worker_ip:
|
||||
logger.debug("Worker IP unknown, skipping log sync for %s",
|
||||
self._local_dir)
|
||||
return False
|
||||
if self.worker_ip == self.local_ip:
|
||||
logger.debug("Worker IP is local IP, skipping log sync for %s",
|
||||
self._local_dir)
|
||||
return False
|
||||
return True
|
||||
|
||||
def sync_up_if_needed(self):
|
||||
if not self.has_remote_target():
|
||||
return True
|
||||
super(NodeSyncMixin, self).sync_up()
|
||||
|
||||
def sync_down_if_needed(self):
|
||||
if not self.has_remote_target():
|
||||
return True
|
||||
super(NodeSyncMixin, self).sync_down()
|
||||
|
||||
def sync_down(self):
|
||||
if not self.has_remote_target():
|
||||
return True
|
||||
return super(NodeSyncMixin, self).sync_down()
|
||||
|
||||
def sync_up(self):
|
||||
if not self.has_remote_target():
|
||||
return True
|
||||
return super(NodeSyncMixin, self).sync_up()
|
||||
|
||||
@property
|
||||
def _remote_path(self):
|
||||
ssh_user = get_ssh_user()
|
||||
global _log_sync_warned
|
||||
if not self.has_remote_target():
|
||||
return
|
||||
if ssh_user is None:
|
||||
if not _log_sync_warned:
|
||||
logger.error("Log sync requires cluster to be setup with "
|
||||
"`ray up`.")
|
||||
_log_sync_warned = True
|
||||
return
|
||||
return "{}@{}:{}/".format(ssh_user, self.worker_ip, self._remote_dir)
|
||||
@@ -14,7 +14,7 @@ import numpy as np
|
||||
|
||||
import ray.cloudpickle as cloudpickle
|
||||
from ray.tune.util import flatten_dict
|
||||
from ray.tune.syncer import get_log_syncer
|
||||
from ray.tune.syncer import get_node_syncer
|
||||
from ray.tune.result import (NODE_IP, TRAINING_ITERATION, TIME_TOTAL_S,
|
||||
TIMESTEPS_TOTAL, EXPR_PARAM_FILE,
|
||||
EXPR_PARAM_PICKLE_FILE, EXPR_PROGRESS_FILE,
|
||||
@@ -385,7 +385,7 @@ class UnifiedLogger(Logger):
|
||||
loggers (list): List of logger creators. Defaults to CSV, Tensorboard,
|
||||
and JSON loggers.
|
||||
sync_function (func|str): Optional function for syncer to run.
|
||||
See ray/python/ray/tune/log_sync.py
|
||||
See ray/python/ray/tune/syncer.py
|
||||
"""
|
||||
|
||||
def __init__(self,
|
||||
@@ -411,7 +411,7 @@ class UnifiedLogger(Logger):
|
||||
except Exception as exc:
|
||||
logger.warning("Could not instantiate %s: %s.", cls.__name__,
|
||||
str(exc))
|
||||
self._log_syncer = get_log_syncer(
|
||||
self._log_syncer = get_node_syncer(
|
||||
self.logdir,
|
||||
remote_dir=self.logdir,
|
||||
sync_function=self._sync_function)
|
||||
|
||||
@@ -0,0 +1,203 @@
|
||||
from __future__ import absolute_import
|
||||
from __future__ import division
|
||||
from __future__ import print_function
|
||||
|
||||
import distutils
|
||||
import distutils.spawn
|
||||
import logging
|
||||
import subprocess
|
||||
import tempfile
|
||||
import types
|
||||
|
||||
try: # py3
|
||||
from shlex import quote
|
||||
except ImportError: # py2
|
||||
from pipes import quote
|
||||
from ray.tune.error import TuneError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
S3_PREFIX = "s3://"
|
||||
GS_PREFIX = "gs://"
|
||||
ALLOWED_REMOTE_PREFIXES = (S3_PREFIX, GS_PREFIX)
|
||||
|
||||
noop_template = ": {target}" # noop in bash
|
||||
|
||||
|
||||
def noop(*args):
|
||||
return
|
||||
|
||||
|
||||
def get_sync_client(sync_function):
|
||||
"""Returns a sync client.
|
||||
|
||||
Args:
|
||||
sync_function (str|function): Sync function.
|
||||
|
||||
Raises:
|
||||
ValueError if sync_function is malformed.
|
||||
"""
|
||||
if isinstance(sync_function, types.FunctionType):
|
||||
client_cls = FunctionBasedClient
|
||||
elif isinstance(sync_function, str):
|
||||
client_cls = CommandBasedClient
|
||||
else:
|
||||
raise ValueError("Sync function {} must be string or function".format(
|
||||
sync_function))
|
||||
return client_cls(sync_function, sync_function)
|
||||
|
||||
|
||||
def get_cloud_sync_client(remote_path):
|
||||
"""Returns a CommandBasedClient that can sync to/from remote storage.
|
||||
|
||||
Args:
|
||||
remote_path (str): Path to remote storage (S3 or GS).
|
||||
|
||||
Raises:
|
||||
ValueError if malformed remote_dir.
|
||||
"""
|
||||
if remote_path.startswith(S3_PREFIX):
|
||||
if not distutils.spawn.find_executable("aws"):
|
||||
raise ValueError(
|
||||
"Upload uri starting with '{}' requires awscli tool"
|
||||
" to be installed".format(S3_PREFIX))
|
||||
template = "aws s3 sync {source} {target}"
|
||||
elif remote_path.startswith(GS_PREFIX):
|
||||
if not distutils.spawn.find_executable("gsutil"):
|
||||
raise ValueError(
|
||||
"Upload uri starting with '{}' requires gsutil tool"
|
||||
" to be installed".format(GS_PREFIX))
|
||||
template = "gsutil rsync -r {source} {target}"
|
||||
else:
|
||||
raise ValueError("Upload uri must start with one of: {}"
|
||||
"".format(ALLOWED_REMOTE_PREFIXES))
|
||||
return CommandBasedClient(template, template)
|
||||
|
||||
|
||||
class SyncClient(object):
|
||||
"""Client interface for interacting with remote storage options."""
|
||||
|
||||
def sync_up(self, source, target):
|
||||
"""Syncs up from source to target.
|
||||
|
||||
Args:
|
||||
source (str): Source path.
|
||||
target (str): Target path.
|
||||
|
||||
Returns:
|
||||
True if sync initiation successful, False otherwise.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def sync_down(self, source, target):
|
||||
"""Syncs down from source to target.
|
||||
|
||||
Args:
|
||||
source (str): Source path.
|
||||
target (str): Target path.
|
||||
|
||||
Returns:
|
||||
True if sync initiation successful, False otherwise.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def wait(self):
|
||||
"""Waits for current sync to complete, if asynchronously started."""
|
||||
pass
|
||||
|
||||
def reset(self):
|
||||
"""Resets state."""
|
||||
pass
|
||||
|
||||
|
||||
class FunctionBasedClient(SyncClient):
|
||||
def __init__(self, sync_up_func, sync_down_func):
|
||||
self.sync_up_func = sync_up_func
|
||||
self.sync_down_func = sync_down_func
|
||||
|
||||
def sync_up(self, source, target):
|
||||
self.sync_up_func(source, target)
|
||||
return True
|
||||
|
||||
def sync_down(self, source, target):
|
||||
self.sync_down_func(source, target)
|
||||
return True
|
||||
|
||||
|
||||
NOOP = FunctionBasedClient(noop, noop)
|
||||
|
||||
|
||||
class CommandBasedClient(SyncClient):
|
||||
def __init__(self, sync_up_template, sync_down_template):
|
||||
"""Syncs between two directories with the given command.
|
||||
|
||||
Arguments:
|
||||
sync_up_template (str): A runnable string template; needs to
|
||||
include replacement fields '{source}' and '{target}'.
|
||||
sync_down_template (str): A runnable string template; needs to
|
||||
include replacement fields '{source}' and '{target}'.
|
||||
"""
|
||||
self._validate_sync_string(sync_up_template)
|
||||
self._validate_sync_string(sync_down_template)
|
||||
self.sync_up_template = sync_up_template
|
||||
self.sync_down_template = sync_down_template
|
||||
self.logfile = None
|
||||
self.cmd_process = None
|
||||
|
||||
def set_logdir(self, logdir):
|
||||
"""Sets the directory to log sync execution output in.
|
||||
|
||||
Args:
|
||||
logdir (str): Log directory.
|
||||
"""
|
||||
self.logfile = tempfile.NamedTemporaryFile(
|
||||
prefix="log_sync", dir=logdir, suffix=".log", delete=False)
|
||||
|
||||
def sync_up(self, source, target):
|
||||
return self._execute(self.sync_up_template, source, target)
|
||||
|
||||
def sync_down(self, source, target):
|
||||
return self._execute(self.sync_down_template, source, target)
|
||||
|
||||
def wait(self):
|
||||
if self.cmd_process:
|
||||
_, error_msg = self.cmd_process.communicate()
|
||||
error_msg = error_msg.decode("ascii")
|
||||
code = self.cmd_process.returncode
|
||||
self.cmd_process = None
|
||||
if code != 0:
|
||||
raise TuneError("Sync error ({}): {}".format(code, error_msg))
|
||||
|
||||
def reset(self):
|
||||
if self.is_running:
|
||||
logger.warning("Sync process still running but resetting anyways.")
|
||||
self.cmd_process = None
|
||||
|
||||
@property
|
||||
def is_running(self):
|
||||
"""Returns whether a sync process is running."""
|
||||
if self.cmd_process:
|
||||
self.cmd_process.poll()
|
||||
return self.cmd_process.returncode is None
|
||||
return False
|
||||
|
||||
def _execute(self, sync_template, source, target):
|
||||
"""Executes sync_template on source and target."""
|
||||
if self.is_running:
|
||||
logger.warning("Last sync client cmd still in progress, skipping.")
|
||||
return False
|
||||
final_cmd = sync_template.format(
|
||||
source=quote(source), target=quote(target))
|
||||
logger.debug("Running sync: {}".format(final_cmd))
|
||||
self.cmd_process = subprocess.Popen(
|
||||
final_cmd, shell=True, stderr=subprocess.PIPE, stdout=self.logfile)
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def _validate_sync_string(sync_string):
|
||||
if not isinstance(sync_string, str):
|
||||
raise ValueError("{} is not a string.".format(sync_string))
|
||||
if "{source}" not in sync_string:
|
||||
raise ValueError("Sync template missing '{source}'.")
|
||||
if "{target}" not in sync_string:
|
||||
raise ValueError("Sync template missing '{target}'.")
|
||||
+110
-169
@@ -5,26 +5,23 @@ from __future__ import print_function
|
||||
import distutils
|
||||
import logging
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
import time
|
||||
import types
|
||||
|
||||
try: # py3
|
||||
from shlex import quote
|
||||
except ImportError: # py2
|
||||
from pipes import quote
|
||||
|
||||
from ray.tune.error import TuneError
|
||||
from ray.tune.log_sync import log_sync_template, NodeSyncMixin
|
||||
from ray import services
|
||||
from ray.tune.cluster_info import get_ssh_key, get_ssh_user
|
||||
from ray.tune.sync_client import (CommandBasedClient, get_sync_client,
|
||||
get_cloud_sync_client, NOOP)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
S3_PREFIX = "s3://"
|
||||
GS_PREFIX = "gs://"
|
||||
ALLOWED_REMOTE_PREFIXES = (S3_PREFIX, GS_PREFIX)
|
||||
SYNC_PERIOD = 300
|
||||
|
||||
_log_sync_warned = False
|
||||
_syncers = {}
|
||||
|
||||
|
||||
@@ -33,127 +30,34 @@ def wait_for_sync():
|
||||
syncer.wait()
|
||||
|
||||
|
||||
class SyncClient(object):
|
||||
def sync_up(self, source, target):
|
||||
"""Sync up from source to target.
|
||||
def log_sync_template(options=""):
|
||||
"""Template enabling syncs between driver and worker when possible.
|
||||
Requires ray cluster to be started with the autoscaler. Also requires
|
||||
rsync to be installed.
|
||||
|
||||
Args:
|
||||
source (str): Source path.
|
||||
target (str): Target path.
|
||||
Args:
|
||||
options (str): Additional rsync options.
|
||||
|
||||
Returns:
|
||||
True if sync initiation successful, False otherwise.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
Returns:
|
||||
Sync template with source and target parameters. None if rsync
|
||||
unavailable.
|
||||
"""
|
||||
if not distutils.spawn.find_executable("rsync"):
|
||||
logger.error("Log sync requires rsync to be installed.")
|
||||
return None
|
||||
global _log_sync_warned
|
||||
ssh_key = get_ssh_key()
|
||||
if ssh_key is None:
|
||||
if not _log_sync_warned:
|
||||
logger.debug("Log sync requires cluster to be setup with "
|
||||
"`ray up`.")
|
||||
_log_sync_warned = True
|
||||
return None
|
||||
|
||||
def sync_down(self, source, target):
|
||||
"""Sync down from source to target.
|
||||
|
||||
Args:
|
||||
source (str): Source path.
|
||||
target (str): Target path.
|
||||
|
||||
Returns:
|
||||
True if sync initiation successful, False otherwise.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def wait(self):
|
||||
"""Wait for current sync to complete, if asynchronously started."""
|
||||
pass
|
||||
|
||||
def reset(self):
|
||||
"""Resets state."""
|
||||
pass
|
||||
|
||||
|
||||
class FunctionBasedClient(SyncClient):
|
||||
def __init__(self, sync_up_func, sync_down_func):
|
||||
self.sync_up_func = sync_up_func
|
||||
self.sync_down_func = sync_down_func
|
||||
|
||||
def sync_up(self, source, target):
|
||||
self.sync_up_func(source, target)
|
||||
return True
|
||||
|
||||
def sync_down(self, source, target):
|
||||
self.sync_down_func(source, target)
|
||||
return True
|
||||
|
||||
|
||||
class CommandBasedClient(SyncClient):
|
||||
def __init__(self, sync_up_template, sync_down_template):
|
||||
"""Syncs between two directories with the given command.
|
||||
|
||||
Arguments:
|
||||
sync_up_template (str): A runnable string template; needs to
|
||||
include replacement fields '{source}' and '{target}'.
|
||||
sync_down_template (str): A runnable string template; needs to
|
||||
include replacement fields '{source}' and '{target}'.
|
||||
"""
|
||||
if not isinstance(sync_up_template, str):
|
||||
raise ValueError("{} is not a string.".format(sync_up_template))
|
||||
if not isinstance(sync_down_template, str):
|
||||
raise ValueError("{} is not a string.".format(sync_down_template))
|
||||
self._validate_sync_string(sync_up_template)
|
||||
self._validate_sync_string(sync_down_template)
|
||||
self.sync_up_template = sync_up_template
|
||||
self.sync_down_template = sync_down_template
|
||||
self.logfile = None
|
||||
self.sync_process = None
|
||||
|
||||
def set_logdir(self, logdir):
|
||||
"""Sets the directory to log sync execution output in.
|
||||
|
||||
Args:
|
||||
logdir: Log directory.
|
||||
"""
|
||||
self.logfile = tempfile.NamedTemporaryFile(
|
||||
prefix="log_sync", dir=logdir, suffix=".log", delete=False)
|
||||
|
||||
def sync_up(self, source, target):
|
||||
return self.execute(self.sync_up_template, source, target)
|
||||
|
||||
def sync_down(self, source, target):
|
||||
return self.execute(self.sync_down_template, source, target)
|
||||
|
||||
def execute(self, sync_template, source, target):
|
||||
"""Executes sync_template on source and target."""
|
||||
if self.sync_process:
|
||||
self.sync_process.poll()
|
||||
if self.sync_process.returncode is None:
|
||||
logger.warning("Last sync is still in progress, skipping.")
|
||||
return False
|
||||
final_cmd = sync_template.format(
|
||||
source=quote(source), target=quote(target))
|
||||
logger.debug("Running sync: {}".format(final_cmd))
|
||||
self.sync_process = subprocess.Popen(
|
||||
final_cmd, shell=True, stderr=subprocess.PIPE, stdout=self.logfile)
|
||||
return True
|
||||
|
||||
def wait(self):
|
||||
if self.sync_process:
|
||||
_, error_msg = self.sync_process.communicate()
|
||||
error_msg = error_msg.decode("ascii")
|
||||
code = self.sync_process.returncode
|
||||
self.sync_process = None
|
||||
if code != 0:
|
||||
raise TuneError("Sync error ({}): {}".format(code, error_msg))
|
||||
|
||||
def reset(self):
|
||||
if self.sync_process:
|
||||
logger.warning("Sync process still running but resetting anyways.")
|
||||
self.sync_process = None
|
||||
|
||||
@staticmethod
|
||||
def _validate_sync_string(sync_string):
|
||||
if "{source}" not in sync_string:
|
||||
raise ValueError("Sync template missing '{source}'.")
|
||||
if "{target}" not in sync_string:
|
||||
raise ValueError("Sync template missing '{target}'.")
|
||||
|
||||
|
||||
NOOP = FunctionBasedClient(lambda s, t: None, lambda s, t: None)
|
||||
rsh = "ssh -i {ssh_key} -o ConnectTimeout=120s -o StrictHostKeyChecking=no"
|
||||
rsh = rsh.format(ssh_key=quote(ssh_key))
|
||||
template = "rsync {options} -savz -e {rsh} {{source}} {{target}}"
|
||||
return template.format(options=options, rsh=quote(rsh))
|
||||
|
||||
|
||||
class Syncer(object):
|
||||
@@ -234,6 +138,76 @@ class Syncer(object):
|
||||
return self._remote_dir
|
||||
|
||||
|
||||
class NodeSyncer(Syncer):
|
||||
"""Syncer for syncing files to/from a remote dir to a local dir."""
|
||||
|
||||
def __init__(self, local_dir, remote_dir, sync_client):
|
||||
self.local_ip = services.get_node_ip_address()
|
||||
self.worker_ip = None
|
||||
super(NodeSyncer, self).__init__(local_dir, remote_dir, sync_client)
|
||||
|
||||
def set_worker_ip(self, worker_ip):
|
||||
"""Sets the worker IP to sync logs from."""
|
||||
self.worker_ip = worker_ip
|
||||
|
||||
def has_remote_target(self):
|
||||
"""Returns whether the Syncer has a remote target."""
|
||||
if not self.worker_ip:
|
||||
logger.debug("Worker IP unknown, skipping sync for %s",
|
||||
self._local_dir)
|
||||
return False
|
||||
if self.worker_ip == self.local_ip:
|
||||
logger.debug("Worker IP is local IP, skipping sync for %s",
|
||||
self._local_dir)
|
||||
return False
|
||||
return True
|
||||
|
||||
def sync_up_if_needed(self):
|
||||
if not self.has_remote_target():
|
||||
return True
|
||||
return super(NodeSyncer, self).sync_up_if_needed()
|
||||
|
||||
def sync_down_if_needed(self):
|
||||
if not self.has_remote_target():
|
||||
return True
|
||||
return super(NodeSyncer, self).sync_down_if_needed()
|
||||
|
||||
def sync_up_to_new_location(self, worker_ip):
|
||||
if worker_ip != self.worker_ip:
|
||||
logger.debug("Setting new worker IP to %s", worker_ip)
|
||||
self.set_worker_ip(worker_ip)
|
||||
self.reset()
|
||||
if not self.sync_up():
|
||||
logger.warning(
|
||||
"Sync up to new location skipped. This should not occur.")
|
||||
else:
|
||||
logger.warning("Sync attempted to same IP %s.", worker_ip)
|
||||
|
||||
def sync_up(self):
|
||||
if not self.has_remote_target():
|
||||
return True
|
||||
return super(NodeSyncer, self).sync_up()
|
||||
|
||||
def sync_down(self):
|
||||
if not self.has_remote_target():
|
||||
return True
|
||||
return super(NodeSyncer, self).sync_down()
|
||||
|
||||
@property
|
||||
def _remote_path(self):
|
||||
ssh_user = get_ssh_user()
|
||||
global _log_sync_warned
|
||||
if not self.has_remote_target():
|
||||
return None
|
||||
if ssh_user is None:
|
||||
if not _log_sync_warned:
|
||||
logger.error("Syncer requires cluster to be setup with "
|
||||
"`ray up`.")
|
||||
_log_sync_warned = True
|
||||
return None
|
||||
return "{}@{}:{}/".format(ssh_user, self.worker_ip, self._remote_dir)
|
||||
|
||||
|
||||
def get_cloud_syncer(local_dir, remote_dir=None, sync_function=None):
|
||||
"""Returns a Syncer.
|
||||
|
||||
@@ -247,6 +221,9 @@ def get_cloud_syncer(local_dir, remote_dir=None, sync_function=None):
|
||||
remote_dir. If string, then it must be a string template for
|
||||
syncer to run. If not provided, it defaults
|
||||
to standard S3 or gsutil sync commands.
|
||||
|
||||
Raises:
|
||||
ValueError if malformed remote_dir.
|
||||
"""
|
||||
key = (local_dir, remote_dir)
|
||||
|
||||
@@ -257,37 +234,18 @@ def get_cloud_syncer(local_dir, remote_dir=None, sync_function=None):
|
||||
_syncers[key] = Syncer(local_dir, remote_dir, NOOP)
|
||||
return _syncers[key]
|
||||
|
||||
client = _get_sync_client(sync_function)
|
||||
client = get_sync_client(sync_function)
|
||||
|
||||
if client:
|
||||
_syncers[key] = Syncer(local_dir, remote_dir, client)
|
||||
return _syncers[key]
|
||||
|
||||
if remote_dir.startswith(S3_PREFIX):
|
||||
if not distutils.spawn.find_executable("aws"):
|
||||
raise TuneError(
|
||||
"Upload uri starting with '{}' requires awscli tool"
|
||||
" to be installed".format(S3_PREFIX))
|
||||
template = "aws s3 sync {source} {target}"
|
||||
s3_client = CommandBasedClient(template, template)
|
||||
_syncers[key] = Syncer(local_dir, remote_dir, s3_client)
|
||||
elif remote_dir.startswith(GS_PREFIX):
|
||||
if not distutils.spawn.find_executable("gsutil"):
|
||||
raise TuneError(
|
||||
"Upload uri starting with '{}' requires gsutil tool"
|
||||
" to be installed".format(GS_PREFIX))
|
||||
template = "gsutil rsync -r {source} {target}"
|
||||
gs_client = CommandBasedClient(template, template)
|
||||
_syncers[key] = Syncer(local_dir, remote_dir, gs_client)
|
||||
else:
|
||||
raise TuneError("Upload uri must start with one of: {}"
|
||||
"".format(ALLOWED_REMOTE_PREFIXES))
|
||||
|
||||
sync_client = get_cloud_sync_client(remote_dir)
|
||||
_syncers[key] = Syncer(local_dir, remote_dir, sync_client)
|
||||
return _syncers[key]
|
||||
|
||||
|
||||
def get_log_syncer(local_dir, remote_dir=None, sync_function=None):
|
||||
"""Returns a log Syncer.
|
||||
def get_node_syncer(local_dir, remote_dir=None, sync_function=None):
|
||||
"""Returns a NodeSyncer.
|
||||
|
||||
Args:
|
||||
local_dir (str): Source directory for syncing.
|
||||
@@ -303,7 +261,7 @@ def get_log_syncer(local_dir, remote_dir=None, sync_function=None):
|
||||
elif not remote_dir:
|
||||
sync_client = NOOP
|
||||
elif sync_function:
|
||||
sync_client = _get_sync_client(sync_function)
|
||||
sync_client = get_sync_client(sync_function)
|
||||
else:
|
||||
sync_up = log_sync_template()
|
||||
sync_down = log_sync_template(options="--remove-source-files")
|
||||
@@ -313,22 +271,5 @@ def get_log_syncer(local_dir, remote_dir=None, sync_function=None):
|
||||
else:
|
||||
sync_client = NOOP
|
||||
|
||||
class MixedSyncer(NodeSyncMixin, Syncer):
|
||||
def __init__(self, *args, **kwargs):
|
||||
Syncer.__init__(self, *args, **kwargs)
|
||||
NodeSyncMixin.__init__(self)
|
||||
|
||||
_syncers[key] = MixedSyncer(local_dir, remote_dir, sync_client)
|
||||
_syncers[key] = NodeSyncer(local_dir, remote_dir, sync_client)
|
||||
return _syncers[key]
|
||||
|
||||
|
||||
def _get_sync_client(sync_function):
|
||||
if not sync_function:
|
||||
return None
|
||||
if isinstance(sync_function, types.FunctionType):
|
||||
return FunctionBasedClient(sync_function, sync_function)
|
||||
elif isinstance(sync_function, str):
|
||||
return CommandBasedClient(sync_function, sync_function)
|
||||
else:
|
||||
raise ValueError("Sync function {} must be string or function".format(
|
||||
sync_function))
|
||||
|
||||
@@ -30,7 +30,7 @@ class TestSyncFunctionality(unittest.TestCase):
|
||||
ray.shutdown()
|
||||
_register_all() # re-register the evicted objects
|
||||
|
||||
@patch("ray.tune.syncer.S3_PREFIX", "test")
|
||||
@patch("ray.tune.sync_client.S3_PREFIX", "test")
|
||||
def testNoUploadDir(self):
|
||||
"""No Upload Dir is given."""
|
||||
with self.assertRaises(AssertionError):
|
||||
@@ -45,7 +45,7 @@ class TestSyncFunctionality(unittest.TestCase):
|
||||
"sync_to_cloud": "echo {source} {target}"
|
||||
}).trials
|
||||
|
||||
@patch("ray.tune.syncer.S3_PREFIX", "test")
|
||||
@patch("ray.tune.sync_client.S3_PREFIX", "test")
|
||||
def testCloudProperString(self):
|
||||
with self.assertRaises(ValueError):
|
||||
[trial] = tune.run(
|
||||
@@ -120,7 +120,7 @@ class TestSyncFunctionality(unittest.TestCase):
|
||||
"sync_to_driver": "ls {source}"
|
||||
}).trials
|
||||
|
||||
with patch.object(CommandBasedClient, "execute") as mock_fn:
|
||||
with patch.object(CommandBasedClient, "_execute") as mock_fn:
|
||||
with patch("ray.services.get_node_ip_address") as mock_sync:
|
||||
mock_sync.return_value = "0.0.0.0"
|
||||
[trial] = tune.run(
|
||||
@@ -198,7 +198,7 @@ class TestSyncFunctionality(unittest.TestCase):
|
||||
def sync_func(source, target):
|
||||
pass
|
||||
|
||||
with patch.object(CommandBasedClient, "execute") as mock_sync:
|
||||
with patch.object(CommandBasedClient, "_execute") as mock_sync:
|
||||
[trial] = tune.run(
|
||||
"__fake",
|
||||
name="foo",
|
||||
|
||||
Reference in New Issue
Block a user