From 17203241a5c027895d354c9a4a2fe04dbf6126eb Mon Sep 17 00:00:00 2001 From: Dacheng Li Date: Sat, 14 Nov 2020 22:00:35 +0000 Subject: [PATCH 01/14] example file --- examples/image_classifier.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/examples/image_classifier.py b/examples/image_classifier.py index edb9bb2..61a073e 100644 --- a/examples/image_classifier.py +++ b/examples/image_classifier.py @@ -5,10 +5,16 @@ ############################################################ # Step 1: Construct AutoDist with ResourceSpec from autodist import AutoDist -filepath = os.path.join(os.path.dirname(__file__), 'resource_spec.yml') +filepath = os.path.join(adaptdl.env.share_path(),'resource_spec.yml') autodist = AutoDist(resource_spec_file=filepath) +import adaptdl +import adaptdl.torch as adl + ############################################################ +adaptdl.torch.write_config() + + fashion_mnist = tf.keras.datasets.fashion_mnist (train_images, train_labels), (test_images, test_labels) = fashion_mnist.load_data() train_images = train_images[:512, :, :, None] From 954cc80c9059ef48db3edd22402b63a272b832d2 Mon Sep 17 00:00:00 2001 From: Dacheng Li Date: Tue, 17 Nov 2020 00:38:00 +0000 Subject: [PATCH 02/14] scratch --- autodist/autodist.py | 8 ++- autodist/cluster.py | 107 ++++++++++++++++++++++++++++++++++++- autodist/resource_spec.py | 4 ++ examples/benchmark/bert.py | 9 +++- 4 files changed, 123 insertions(+), 5 deletions(-) diff --git a/autodist/autodist.py b/autodist/autodist.py index 69a8598..04024d0 100644 --- a/autodist/autodist.py +++ b/autodist/autodist.py @@ -36,6 +36,7 @@ from autodist.strategy import base from autodist.strategy.ps_lb_strategy import PSLoadBalancing from autodist.utils import logging +import adaptdl.collective as collective IS_AUTODIST_WORKER = bool(ENV.AUTODIST_WORKER.val) IS_AUTODIST_CHIEF = not IS_AUTODIST_WORKER @@ -122,9 +123,12 @@ def _setup(self, strategy): if IS_AUTODIST_CHIEF: # we should only have one single coordinator for one single AutoDist() instance scope, # even though we could have multiple strategies. + collective.initialize() self._coordinator = Coordinator(strategy=strategy, cluster=self._cluster) - self._cluster.start() - self._coordinator.launch_clients() + self._cluster.start_chief() + self._coordinator.launch_clients_chief() + else: + collective.initialize() logging.info('Current PID {} belongs to address {}'.format(os.getpid(), self._cluster.get_local_address())) diff --git a/autodist/cluster.py b/autodist/cluster.py index 77f65d0..e555a7f 100644 --- a/autodist/cluster.py +++ b/autodist/cluster.py @@ -165,7 +165,7 @@ def start(self): """ # pylint: disable=import-outside-toplevel from autodist.utils import server_starter - + assert False # atexit registration should be placed # - before the beginning of the start # (to ensure the clean termination if the start fails in its half way); and @@ -178,7 +178,59 @@ def start(self): envs = ['{}={}'.format(k, v) for k, v in envs.items()] module_name = server_starter.__name__ module_file = server_starter.__file__ + print(self.cluster_spec) + for job_name, tasks in self.cluster_spec.items(): + for task_index, full_address in enumerate(tasks): + address = full_address.split(':')[0] + args = ['--job_name=%s' % job_name, '--task_index=%d' % task_index, + '--cpu_device_num=%d' % len(self._cpu_devices[address])] + if address in self._gpu_devices: + envs_cuda = [] + else: + envs_cuda = ['CUDA_VISIBLE_DEVICES=""'] + if self.is_chief(address): + json.dump(self.cluster_spec, open(os.path.join(DEFAULT_WORKING_DIR, 'cluster_spec.json'), 'w+')) + cmd = envs + envs_cuda + [sys.executable, '-m', module_name] + args + # pylint: disable=subprocess-popen-preexec-fn + proc = subprocess.Popen(' '.join(cmd), shell=True, preexec_fn=os.setsid) + self.subprocesses.append(proc) + # The above line immediately follows the Popen + # to ensure no gap for termination failure due to the empty proc list. + logging.debug('$ local tf.server started at {}: job_name={} task_index={}'.format( + full_address, job_name, task_index + )) + else: # remote + self.remote_pre_start_tf_server(address, tf_server_starter_filepath=module_file) + file = os.path.join(DEFAULT_WORKING_DIR, os.path.basename(module_file)) + bash = envs + envs_cuda + ['python', '-u', file] + args + logging.info("Launching tf.server on %s" % address) + proc = self.remote_exec(bash, hostname=address) + # The above line immediately follows the Popen + # to ensure no gap for termination failure due to the empty proc list. + self.subprocesses.append(proc) + # pylint: disable=too-many-locals + def start_chief(self): + """ + Start tf.servers on all nodes. + + Note that this only runs (and only should run) on the chief node. + """ + # pylint: disable=import-outside-toplevel + from autodist.utils import server_starter + # atexit registration should be placed + # - before the beginning of the start + # (to ensure the clean termination if the start fails in its half way); and + # - at the same module as the start + # (to follow the python assumption that + # lower level modules will normally be imported + # before higher level modules and thus must be cleaned up later). + atexit.register(self.terminate) + envs = {ENV.AUTODIST_MIN_LOG_LEVEL.name: 'ERROR'} + envs = ['{}={}'.format(k, v) for k, v in envs.items()] + module_name = server_starter.__name__ + module_file = server_starter.__file__ + print(self.cluster_spec) for job_name, tasks in self.cluster_spec.items(): for task_index, full_address in enumerate(tasks): address = full_address.split(':')[0] @@ -209,6 +261,58 @@ def start(self): # to ensure no gap for termination failure due to the empty proc list. self.subprocesses.append(proc) + # pylint: disable=too-many-locals + def start_worker(self): + """ + Start tf.servers on all nodes. + + Note that this only runs (and only should run) on the chief node. + """ + # pylint: disable=import-outside-toplevel + from autodist.utils import server_starter + + # atexit registration should be placed + # - before the beginning of the start + # (to ensure the clean termination if the start fails in its half way); and + # - at the same module as the start + # (to follow the python assumption that + # lower level modules will normally be imported + # before higher level modules and thus must be cleaned up later). + atexit.register(self.terminate) + envs = {ENV.AUTODIST_MIN_LOG_LEVEL.name: 'ERROR'} + envs = ['{}={}'.format(k, v) for k, v in envs.items()] + module_name = server_starter.__name__ + module_file = server_starter.__file__ + print(self.cluster_spec) + for job_name, tasks in self.cluster_spec.items(): + for task_index, full_address in enumerate(tasks): + address = full_address.split(':')[0] + args = ['--job_name=%s' % job_name, '--task_index=%d' % task_index, + '--cpu_device_num=%d' % len(self._cpu_devices[address])] + if address in self._gpu_devices: + envs_cuda = [] + else: + envs_cuda = ['CUDA_VISIBLE_DEVICES=""'] + if self.is_chief(address): + json.dump(self.cluster_spec, open(os.path.join(DEFAULT_WORKING_DIR, 'cluster_spec.json'), 'w+')) + cmd = envs + envs_cuda + [sys.executable, '-m', module_name] + args + # pylint: disable=subprocess-popen-preexec-fn + proc = subprocess.Popen(' '.join(cmd), shell=True, preexec_fn=os.setsid) + self.subprocesses.append(proc) + # The above line immediately follows the Popen + # to ensure no gap for termination failure due to the empty proc list. + logging.debug('$ local tf.server started at {}: job_name={} task_index={}'.format( + full_address, job_name, task_index + )) + else: # remote + self.remote_pre_start_tf_server(address, tf_server_starter_filepath=module_file) + file = os.path.join(DEFAULT_WORKING_DIR, os.path.basename(module_file)) + bash = envs + envs_cuda + ['python', '-u', file] + args + logging.info("Launching tf.server on %s" % address) + proc = self.remote_exec(bash, hostname=address) + # The above line immediately follows the Popen + # to ensure no gap for termination failure due to the empty proc list. + self.subprocesses.append(proc) def terminate(self): """Terminate.""" logging.debug('Terminating cluster...') @@ -273,6 +377,7 @@ class SSHCluster(Cluster): def __init__(self, resource_spec): self._ssh_conf = resource_spec.ssh_config_map + print(self._ssh_conf) super().__init__(resource_spec) @contextlib.contextmanager diff --git a/autodist/resource_spec.py b/autodist/resource_spec.py index 5945aef..d4664c6 100644 --- a/autodist/resource_spec.py +++ b/autodist/resource_spec.py @@ -184,6 +184,7 @@ def _from_resource_info(self, resource_file=None): def _parse_node(self, node, num_nodes): host_address = node['address'] + print(host_address) if is_loopback_address(host_address) and num_nodes > 1: raise ValueError("Can't (currently) use a loopback address when there are multiple nodes.") if node.get('chief') or num_nodes == 1: @@ -193,6 +194,7 @@ def _parse_node(self, node, num_nodes): logging.info("Chief: %s" % host_address) self.__chief_address = host_address host_cpu = DeviceSpec(host_address, device_index=0) + print(host_cpu) self._add_device(host_cpu) # handle any other CPUs when GPU is unavailable if len(node.get('gpus', [])) == 0: @@ -201,9 +203,11 @@ def _parse_node(self, node, num_nodes): self._add_device(cpu) # handle GPUs for gpu_index in set(sorted(node.get('gpus', []))): + print(gpu_index) gpu = DeviceSpec(host_address, host_cpu, DeviceType.GPU, gpu_index) self._add_device(gpu) self.__ssh_group[host_address] = node.get('ssh_config') + print(node.get('ssh_config')) if self.__ssh_group[host_address] is None and self.__chief_address != host_address: raise ValueError("Need to define SSH groups for all non-chief nodes.") # handle network bandwidth (optional) diff --git a/examples/benchmark/bert.py b/examples/benchmark/bert.py index 8caa995..51a10fa 100644 --- a/examples/benchmark/bert.py +++ b/examples/benchmark/bert.py @@ -36,6 +36,11 @@ from utils import common_flags from utils import input_pipeline from utils import bert_utils +import adaptdl.torch as adl +#import adaptdl +import adaptdl.env as env +import adaptdl.torch as adl +adl.write_config() ######################################################################### # Import AutoDist and Strategy @@ -197,8 +202,8 @@ def main(_): else: os.environ['AUTODIST_PATCH_TF'] = 'False' resource_spec_file = os.path.join( - os.path.dirname(__file__), - '../resource_spec.yml') + env.share_path(), + 'resource_spec.yml') if FLAGS.autodist_strategy == 'PS': strategy = AutoDist( From 4b55c2dfcf71d96620cc37f6fd11f5743f7450a0 Mon Sep 17 00:00:00 2001 From: Dacheng Li Date: Fri, 20 Nov 2020 06:52:07 +0000 Subject: [PATCH 03/14] testing --- autodist/autodist.py | 2 + autodist/cluster.py | 112 +++++++++++++++++++++++++++------------- autodist/coordinator.py | 103 ++++++++++++++++++++++++++++++++++-- 3 files changed, 176 insertions(+), 41 deletions(-) diff --git a/autodist/autodist.py b/autodist/autodist.py index 04024d0..28defcb 100644 --- a/autodist/autodist.py +++ b/autodist/autodist.py @@ -129,6 +129,8 @@ def _setup(self, strategy): self._coordinator.launch_clients_chief() else: collective.initialize() + self._cluster.start_worker() + self._coordinator.launch_clients_worker() logging.info('Current PID {} belongs to address {}'.format(os.getpid(), self._cluster.get_local_address())) diff --git a/autodist/cluster.py b/autodist/cluster.py index e555a7f..e21a6cb 100644 --- a/autodist/cluster.py +++ b/autodist/cluster.py @@ -40,10 +40,11 @@ from abc import ABCMeta, abstractmethod import paramiko - +import adaptdl.collective as collective from autodist.const import DEFAULT_PORT_RANGE, DEFAULT_WORKING_DIR, ENV from autodist.resource_spec import ResourceSpec from autodist.utils import logging +import socket warnings.filterwarnings(action='ignore', module=paramiko.__name__) @@ -230,7 +231,6 @@ def start_chief(self): envs = ['{}={}'.format(k, v) for k, v in envs.items()] module_name = server_starter.__name__ module_file = server_starter.__file__ - print(self.cluster_spec) for job_name, tasks in self.cluster_spec.items(): for task_index, full_address in enumerate(tasks): address = full_address.split(':')[0] @@ -254,12 +254,12 @@ def start_chief(self): else: # remote self.remote_pre_start_tf_server(address, tf_server_starter_filepath=module_file) file = os.path.join(DEFAULT_WORKING_DIR, os.path.basename(module_file)) - bash = envs + envs_cuda + ['python', '-u', file] + args - logging.info("Launching tf.server on %s" % address) - proc = self.remote_exec(bash, hostname=address) + # bash = envs + envs_cuda + ['python', '-u', file] + args + # logging.info("Launching tf.server on %s" % address) + # proc = self.remote_exec(bash, hostname=address) # The above line immediately follows the Popen # to ensure no gap for termination failure due to the empty proc list. - self.subprocesses.append(proc) + # self.subprocesses.append(proc) # pylint: disable=too-many-locals def start_worker(self): @@ -283,43 +283,35 @@ def start_worker(self): envs = ['{}={}'.format(k, v) for k, v in envs.items()] module_name = server_starter.__name__ module_file = server_starter.__file__ - print(self.cluster_spec) for job_name, tasks in self.cluster_spec.items(): for task_index, full_address in enumerate(tasks): address = full_address.split(':')[0] + if socket.gethostname() != address: + continue args = ['--job_name=%s' % job_name, '--task_index=%d' % task_index, '--cpu_device_num=%d' % len(self._cpu_devices[address])] if address in self._gpu_devices: envs_cuda = [] else: envs_cuda = ['CUDA_VISIBLE_DEVICES=""'] - if self.is_chief(address): - json.dump(self.cluster_spec, open(os.path.join(DEFAULT_WORKING_DIR, 'cluster_spec.json'), 'w+')) - cmd = envs + envs_cuda + [sys.executable, '-m', module_name] + args - # pylint: disable=subprocess-popen-preexec-fn - proc = subprocess.Popen(' '.join(cmd), shell=True, preexec_fn=os.setsid) - self.subprocesses.append(proc) + assert not self.is_chief(address): + self.remote_pre_start_tf_server(address, tf_server_starter_filepath=module_file, chief=False) + file = os.path.join(DEFAULT_WORKING_DIR, os.path.basename(module_file)) + bash = envs + envs_cuda + ['python', '-u', file] + args + logging.info("Launching tf.server on %s" % address) + proc = self.local_exec(bash, address) # The above line immediately follows the Popen # to ensure no gap for termination failure due to the empty proc list. - logging.debug('$ local tf.server started at {}: job_name={} task_index={}'.format( - full_address, job_name, task_index - )) - else: # remote - self.remote_pre_start_tf_server(address, tf_server_starter_filepath=module_file) - file = os.path.join(DEFAULT_WORKING_DIR, os.path.basename(module_file)) - bash = envs + envs_cuda + ['python', '-u', file] + args - logging.info("Launching tf.server on %s" % address) - proc = self.remote_exec(bash, hostname=address) - # The above line immediately follows the Popen - # to ensure no gap for termination failure due to the empty proc list. - self.subprocesses.append(proc) + self.subprocesses.append(proc) + assert len(self.subprocesses) <= 1 + def terminate(self): """Terminate.""" logging.debug('Terminating cluster...') for p in self.subprocesses: os.killpg(os.getpgid(p.pid), signal.SIGTERM) - def remote_pre_start_tf_server(self, hostname, tf_server_starter_filepath, working_dir=DEFAULT_WORKING_DIR): + def remote_pre_start_tf_server(self, hostname, tf_server_starter_filepath, working_dir=DEFAULT_WORKING_DIR, chief): """ Prepare to start a TensorFlow server remotely. @@ -329,11 +321,12 @@ def remote_pre_start_tf_server(self, hostname, tf_server_starter_filepath, worki working_dir (str): remote working directory """ logging.info("Copying necessary files to %s" % hostname) - self.remote_copy(local_path=tf_server_starter_filepath, remote_path=working_dir, hostname=hostname) + self.remote_copy(local_path=tf_server_starter_filepath, remote_path=working_dir, hostname=hostname, chief) self.remote_file_write( remote_path=os.path.join(working_dir, 'cluster_spec.json'), data=json.dumps(self.cluster_spec), hostname=hostname, + chief ) @abstractmethod @@ -377,7 +370,6 @@ class SSHCluster(Cluster): def __init__(self, resource_spec): self._ssh_conf = resource_spec.ssh_config_map - print(self._ssh_conf) super().__init__(resource_spec) @contextlib.contextmanager @@ -391,6 +383,7 @@ def _get_ssh_client(self, hostname): Returns: Yields a Paramiko SSHClient. """ + assert False ssh_config = self._ssh_conf[hostname] client = paramiko.SSHClient() client.load_system_host_keys() @@ -410,6 +403,7 @@ def _get_sftp_client(self, hostname): Returns: Yields a Paramiko SFTPClient. """ + assert False ssh_config = self._ssh_conf[hostname] t = paramiko.Transport((hostname, ssh_config.port)) t.connect(username=ssh_config.username, pkey=ssh_config.pkey) @@ -429,6 +423,7 @@ def remote_exec(self, args, hostname): Returns: Process: process handle """ + assert False cmd_list = [] ssh_config = self._ssh_conf[hostname] if ssh_config.python_venv: @@ -449,7 +444,27 @@ def remote_exec(self, args, hostname): proc = subprocess.Popen(remote_cmd, shell=True, preexec_fn=os.setsid) return proc - def remote_file_write(self, remote_path, data, hostname): + def local_exec(self, args, hostname): + cmd_list = [] + ssh_config = self._ssh_conf[hostname] + if ssh_config.python_venv: + cmd_list.append('%s;' % ssh_config.python_venv) + if ssh_config.env: + cmd_list.extend(['%s=%s' % (k, v) for k, v in ssh_config.env.items()]) + full_cmd = ' '.join(cmd_list + args) + + remote_cmd = '\'bash -c "{}"\' Date: Sun, 22 Nov 2020 22:52:12 +0000 Subject: [PATCH 04/14] finished launching --- autodist/autodist.py | 23 +++++--- autodist/cluster.py | 77 ++++++++++++++----------- autodist/const.py | 18 +++++- autodist/coordinator.py | 97 ++++++++++++-------------------- autodist/resource_spec.py | 3 - autodist/utils/server_starter.py | 3 + examples/benchmark/bert.py | 1 - examples/image_classifier.py | 10 ++-- 8 files changed, 120 insertions(+), 112 deletions(-) diff --git a/autodist/autodist.py b/autodist/autodist.py index 28defcb..f76efa9 100644 --- a/autodist/autodist.py +++ b/autodist/autodist.py @@ -37,10 +37,10 @@ from autodist.strategy.ps_lb_strategy import PSLoadBalancing from autodist.utils import logging import adaptdl.collective as collective - +import socket IS_AUTODIST_WORKER = bool(ENV.AUTODIST_WORKER.val) +logging.info(f"is worker: {IS_AUTODIST_WORKER}") IS_AUTODIST_CHIEF = not IS_AUTODIST_WORKER - _DEFAULT_AUTODIST = {} @@ -98,37 +98,43 @@ def build_strategy(self): """ return self._strategy_builder.build(self._original_graph_item, self._resource_spec) - def _build_or_load_strategy(self): + def _build_or_load_strategy(self, load=False): self._original_graph_item.prepare() + import socket + print(socket.gethostbyname(socket.gethostname()), IS_AUTODIST_CHIEF) if IS_AUTODIST_CHIEF: s = self.build_strategy() s.serialize() else: + if not load: + return + #return None strategy_id = ENV.AUTODIST_STRATEGY_ID.val + print(strategy_id) assert strategy_id s = base.Strategy.deserialize(strategy_id) return s def _compile_strategy(self, strategy): - logging.debug('Raw strategy: %s' % strategy) + #logging.debug('Raw strategy: %s' % strategy) device_resolver = DeviceResolver(self._cluster) compiled_strategy = base.StrategyCompiler(self._original_graph_item) \ .set_device_resolver(device_resolver.resolve_to_device_str) \ .compile(strategy) - logging.info('Compiled strategy: %s' % compiled_strategy) + #logging.info('Compiled strategy: %s' % compiled_strategy) return compiled_strategy def _setup(self, strategy): """Prepare for the execution.""" + logging.info(f"{socket.gethostname()} is chief: {IS_AUTODIST_CHIEF}") if IS_AUTODIST_CHIEF: # we should only have one single coordinator for one single AutoDist() instance scope, # even though we could have multiple strategies. - collective.initialize() self._coordinator = Coordinator(strategy=strategy, cluster=self._cluster) self._cluster.start_chief() self._coordinator.launch_clients_chief() - else: - collective.initialize() + else: + self._coordinator = Coordinator(strategy=strategy, cluster=self._cluster) self._cluster.start_worker() self._coordinator.launch_clients_worker() logging.info('Current PID {} belongs to address {}'.format(os.getpid(), self._cluster.get_local_address())) @@ -145,6 +151,7 @@ def _initialize_graph(self): def _build(self): strategy = self._build_or_load_strategy() self._setup(strategy) # Put it before transforming to allow multiple works to transform concurrently + strategy = self._build_or_load_strategy(load=True) compiled_strategy = self._compile_strategy(strategy) graph_transformer = GraphTransformer( compiled_strategy=compiled_strategy, diff --git a/autodist/cluster.py b/autodist/cluster.py index e21a6cb..1d63908 100644 --- a/autodist/cluster.py +++ b/autodist/cluster.py @@ -136,7 +136,9 @@ def get_local_address(self): Returns: str: Worker ip or chief address by default. """ - return ENV.AUTODIST_WORKER.val or self._chief + hostname = socket.gethostname() + local_ip = socket.gethostbyname(hostname) + return local_ip #ENV.AUTODIST_WORKER.val or self._chief def get_local_worker_task_index(self): """ @@ -145,7 +147,11 @@ def get_local_worker_task_index(self): Returns: int: Task index """ - return [i for i, a in enumerate(self._full_addresses) if self.get_local_address() in a][0] + logging.info(f"full address {self._full_addresses}") + logging.info(f"local address {self.get_local_address()}") + return_ = [i for i, a in enumerate(self._full_addresses) if self.get_local_address() in a][0] + logging.info(f"returning {return_}") + return return_ def get_local_session_target(self): """ @@ -203,7 +209,7 @@ def start(self): else: # remote self.remote_pre_start_tf_server(address, tf_server_starter_filepath=module_file) file = os.path.join(DEFAULT_WORKING_DIR, os.path.basename(module_file)) - bash = envs + envs_cuda + ['python', '-u', file] + args + #bash = envs + envs_cuda + ['python', '-u', file] + args logging.info("Launching tf.server on %s" % address) proc = self.remote_exec(bash, hostname=address) # The above line immediately follows the Popen @@ -226,7 +232,7 @@ def start_chief(self): # (to follow the python assumption that # lower level modules will normally be imported # before higher level modules and thus must be cleaned up later). - atexit.register(self.terminate) + #atexit.register(self.terminate) envs = {ENV.AUTODIST_MIN_LOG_LEVEL.name: 'ERROR'} envs = ['{}={}'.format(k, v) for k, v in envs.items()] module_name = server_starter.__name__ @@ -244,6 +250,7 @@ def start_chief(self): json.dump(self.cluster_spec, open(os.path.join(DEFAULT_WORKING_DIR, 'cluster_spec.json'), 'w+')) cmd = envs + envs_cuda + [sys.executable, '-m', module_name] + args # pylint: disable=subprocess-popen-preexec-fn + logging.info("cmd at chief: %s", cmd) proc = subprocess.Popen(' '.join(cmd), shell=True, preexec_fn=os.setsid) self.subprocesses.append(proc) # The above line immediately follows the Popen @@ -251,9 +258,9 @@ def start_chief(self): logging.debug('$ local tf.server started at {}: job_name={} task_index={}'.format( full_address, job_name, task_index )) - else: # remote - self.remote_pre_start_tf_server(address, tf_server_starter_filepath=module_file) - file = os.path.join(DEFAULT_WORKING_DIR, os.path.basename(module_file)) + # else: # remote + self.remote_pre_start_tf_server(None, tf_server_starter_filepath=module_file,chief=True) + file = os.path.join(DEFAULT_WORKING_DIR, os.path.basename(module_file)) # bash = envs + envs_cuda + ['python', '-u', file] + args # logging.info("Launching tf.server on %s" % address) # proc = self.remote_exec(bash, hostname=address) @@ -278,7 +285,7 @@ def start_worker(self): # (to follow the python assumption that # lower level modules will normally be imported # before higher level modules and thus must be cleaned up later). - atexit.register(self.terminate) + #atexit.register(self.terminate) envs = {ENV.AUTODIST_MIN_LOG_LEVEL.name: 'ERROR'} envs = ['{}={}'.format(k, v) for k, v in envs.items()] module_name = server_starter.__name__ @@ -286,7 +293,9 @@ def start_worker(self): for job_name, tasks in self.cluster_spec.items(): for task_index, full_address in enumerate(tasks): address = full_address.split(':')[0] - if socket.gethostname() != address: + hostname = socket.gethostname() + local_ip = socket.gethostbyname(hostname) + if local_ip != address: continue args = ['--job_name=%s' % job_name, '--task_index=%d' % task_index, '--cpu_device_num=%d' % len(self._cpu_devices[address])] @@ -294,12 +303,14 @@ def start_worker(self): envs_cuda = [] else: envs_cuda = ['CUDA_VISIBLE_DEVICES=""'] - assert not self.is_chief(address): + assert not self.is_chief(address) self.remote_pre_start_tf_server(address, tf_server_starter_filepath=module_file, chief=False) - file = os.path.join(DEFAULT_WORKING_DIR, os.path.basename(module_file)) - bash = envs + envs_cuda + ['python', '-u', file] + args + #file = os.path.join(DEFAULT_WORKING_DIR, os.path.basename(module_file)) + #bash = envs + envs_cuda + ['python', '-u', file] + args + + cmd = envs + envs_cuda + [sys.executable, '-m', module_name] + args logging.info("Launching tf.server on %s" % address) - proc = self.local_exec(bash, address) + proc = self.local_exec(cmd, address) # The above line immediately follows the Popen # to ensure no gap for termination failure due to the empty proc list. self.subprocesses.append(proc) @@ -311,7 +322,7 @@ def terminate(self): for p in self.subprocesses: os.killpg(os.getpgid(p.pid), signal.SIGTERM) - def remote_pre_start_tf_server(self, hostname, tf_server_starter_filepath, working_dir=DEFAULT_WORKING_DIR, chief): + def remote_pre_start_tf_server(self, hostname, tf_server_starter_filepath, chief, working_dir=DEFAULT_WORKING_DIR): """ Prepare to start a TensorFlow server remotely. @@ -320,13 +331,13 @@ def remote_pre_start_tf_server(self, hostname, tf_server_starter_filepath, worki tf_server_starter_filepath (str): local starter file path working_dir (str): remote working directory """ - logging.info("Copying necessary files to %s" % hostname) - self.remote_copy(local_path=tf_server_starter_filepath, remote_path=working_dir, hostname=hostname, chief) + #logging.info("Copying necessary files to %s" % hostname) + self.remote_copy(local_path=tf_server_starter_filepath, remote_path=working_dir, hostname=hostname, chief=chief) self.remote_file_write( remote_path=os.path.join(working_dir, 'cluster_spec.json'), data=json.dumps(self.cluster_spec), hostname=hostname, - chief + chief=chief ) @abstractmethod @@ -445,24 +456,26 @@ def remote_exec(self, args, hostname): return proc def local_exec(self, args, hostname): - cmd_list = [] - ssh_config = self._ssh_conf[hostname] - if ssh_config.python_venv: - cmd_list.append('%s;' % ssh_config.python_venv) - if ssh_config.env: - cmd_list.extend(['%s=%s' % (k, v) for k, v in ssh_config.env.items()]) - full_cmd = ' '.join(cmd_list + args) + #cmd_list = [] + #ssh_config = self._ssh_conf[hostname] + #if ssh_config.python_venv: + # cmd_list.append('%s;' % ssh_config.python_venv) + #if ssh_config.env: + # cmd_list.extend(['%s=%s' % (k, v) for k, v in ssh_config.env.items()]) + full_cmd = ' '.join(args) - remote_cmd = '\'bash -c "{}"\' 1: raise ValueError("Can't (currently) use a loopback address when there are multiple nodes.") if node.get('chief') or num_nodes == 1: @@ -194,7 +193,6 @@ def _parse_node(self, node, num_nodes): logging.info("Chief: %s" % host_address) self.__chief_address = host_address host_cpu = DeviceSpec(host_address, device_index=0) - print(host_cpu) self._add_device(host_cpu) # handle any other CPUs when GPU is unavailable if len(node.get('gpus', [])) == 0: @@ -203,7 +201,6 @@ def _parse_node(self, node, num_nodes): self._add_device(cpu) # handle GPUs for gpu_index in set(sorted(node.get('gpus', []))): - print(gpu_index) gpu = DeviceSpec(host_address, host_cpu, DeviceType.GPU, gpu_index) self._add_device(gpu) self.__ssh_group[host_address] = node.get('ssh_config') diff --git a/autodist/utils/server_starter.py b/autodist/utils/server_starter.py index 922c645..081811b 100644 --- a/autodist/utils/server_starter.py +++ b/autodist/utils/server_starter.py @@ -56,6 +56,7 @@ def gen_server(cluster_spec, job_name: str, task_index: int, cpu_device_num: int cpu_device_num: The number of CPU devices """ _clean_stale_servers() + logging.info("cleaned server.") # TODO: The following config should be less hard coded ad based on strategy experimental = config_pb2.ConfigProto.Experimental( @@ -87,9 +88,11 @@ def start_server(cluster_spec, job_name: str, task_index: int, cpu_device_num: i """ s = gen_server(cluster_spec, job_name, task_index, cpu_device_num) s.join() + logging.info("joined server.") if __name__ == '__main__': + logging.info("started joining") parser = argparse.ArgumentParser() parser.add_argument( "--job_name", diff --git a/examples/benchmark/bert.py b/examples/benchmark/bert.py index 51a10fa..d9f1ce4 100644 --- a/examples/benchmark/bert.py +++ b/examples/benchmark/bert.py @@ -37,7 +37,6 @@ from utils import input_pipeline from utils import bert_utils import adaptdl.torch as adl -#import adaptdl import adaptdl.env as env import adaptdl.torch as adl adl.write_config() diff --git a/examples/image_classifier.py b/examples/image_classifier.py index 61a073e..ce39725 100644 --- a/examples/image_classifier.py +++ b/examples/image_classifier.py @@ -5,14 +5,14 @@ ############################################################ # Step 1: Construct AutoDist with ResourceSpec from autodist import AutoDist -filepath = os.path.join(adaptdl.env.share_path(),'resource_spec.yml') -autodist = AutoDist(resource_spec_file=filepath) -import adaptdl -import adaptdl.torch as adl +import adaptdl.torch as adl +import adaptdl.env as env +filepath = os.path.join(env.share_path(),'resource_spec.yml') +autodist = AutoDist(resource_spec_file=filepath) ############################################################ -adaptdl.torch.write_config() +adl.write_config() fashion_mnist = tf.keras.datasets.fashion_mnist From a912b40dbe873a8fd2c97757877b2db2acaa7e6f Mon Sep 17 00:00:00 2001 From: Dacheng Li Date: Sun, 6 Dec 2020 21:51:04 +0000 Subject: [PATCH 05/14] added from adaptdl --- examples/benchmark/bert.py | 5 +- examples/benchmark/utils/misc/gen_yml.py | 60 ++++++++++++++++++++++++ 2 files changed, 63 insertions(+), 2 deletions(-) create mode 100644 examples/benchmark/utils/misc/gen_yml.py diff --git a/examples/benchmark/bert.py b/examples/benchmark/bert.py index d9f1ce4..562d8af 100644 --- a/examples/benchmark/bert.py +++ b/examples/benchmark/bert.py @@ -29,7 +29,7 @@ from absl import logging from utils.logs import logger -from utils.misc import keras_utils +from utils.misc import keras_utils, gen_yml from utils import bert_modeling as modeling from utils import bert_models @@ -39,7 +39,8 @@ import adaptdl.torch as adl import adaptdl.env as env import adaptdl.torch as adl -adl.write_config() + +gen_yml.generate() ######################################################################### # Import AutoDist and Strategy diff --git a/examples/benchmark/utils/misc/gen_yml.py b/examples/benchmark/utils/misc/gen_yml.py new file mode 100644 index 0000000..d636f5a --- /dev/null +++ b/examples/benchmark/utils/misc/gen_yml.py @@ -0,0 +1,60 @@ +# Copyright 2020 Petuum, Inc. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import logging +import requests +import adaptdl.collective as collective +import os +import getpass +import adaptdl.env as env +logging.basicConfig(level=logging.INFO) +LOG = logging.getLogger(__name__) +LOG.setLevel(logging.INFO) + +def generate(): + url = env.supervisor_url() + if url: + key = env.job_id() + group = env.num_restarts() + while True: + response = requests.get(url=f"{url}/discover_gpu/{key}/{group}") + if response.status_code != 408: # Timeout. + break + response.raise_for_status() + master_addr = response.json()[0][0] + else: + raise ValueError("supervisor url not found.") + # write to the share path + path = os.path.join(env.share_path(), "resource_spec.yml") + LOG.info(f"writing to {path}") + + f = open(path, "w") + f.write("nodes: \n") + num_nodes = len(response.json()) + for i in range(num_nodes): + f.write(f" - address: {response.json()[i][0]} \n") + f.write(f" gpus: {list(range(response.json()[i][1]))} \n") + if i == 0: # chief + master_addr = response.json()[i][0] + f.write(" chief: true \n") + else: + f.write(" ssh_config: conf \n") + f.write("ssh: \n") + f.write(" conf: \n") + f.write(f" username: '{getpass.getuser()}' \n") + f.write(" key_file: '/root/.ssh/id_rsa' \n") + f.close() + # Initialize collective module. + master_port = env.master_port() + collective.initialize(master_addr, master_port) From e472cdec75d156d8ca96d64bafaeff5f24b83c40 Mon Sep 17 00:00:00 2001 From: Dacheng Li Date: Mon, 7 Dec 2020 04:38:44 +0000 Subject: [PATCH 06/14] add back original functionality --- autodist/autodist.py | 45 +++--- autodist/cluster.py | 171 +++++++++++------------ autodist/const.py | 32 +++-- autodist/coordinator.py | 71 ++++------ examples/benchmark/utils/misc/gen_yml.py | 3 +- examples/image_classifier.py | 8 +- 6 files changed, 155 insertions(+), 175 deletions(-) diff --git a/autodist/autodist.py b/autodist/autodist.py index f76efa9..c6fb6de 100644 --- a/autodist/autodist.py +++ b/autodist/autodist.py @@ -23,7 +23,7 @@ from tensorflow.python.ops import array_ops from tensorflow.python.util import tf_contextlib -from autodist.cluster import Cluster, SSHCluster +from autodist.cluster import Cluster, SSHCluster, ADAPTDLCluster from autodist.const import ENV from autodist.coordinator import Coordinator from autodist.graph_item import GraphItem @@ -36,11 +36,13 @@ from autodist.strategy import base from autodist.strategy.ps_lb_strategy import PSLoadBalancing from autodist.utils import logging + import adaptdl.collective as collective import socket IS_AUTODIST_WORKER = bool(ENV.AUTODIST_WORKER.val) -logging.info(f"is worker: {IS_AUTODIST_WORKER}") IS_AUTODIST_CHIEF = not IS_AUTODIST_WORKER +IS_ADAPTDL = bool(ENV.ADAPTDL.val) +logging.info(f"is chief: {IS_AUTODIST_CHIEF}, is from adaptdl: {IS_ADAPTDL}") _DEFAULT_AUTODIST = {} @@ -75,7 +77,10 @@ def __init__(self, resource_spec_file, strategy_builder=None): self._remapper = None self._built = None # Ref to the built GraphDef - self._cluster: Cluster = SSHCluster(self._resource_spec) # which can be also defined with strategy + if IS_ADAPTDL: + self._cluster: Cluster = ADAPTDLCluster(self._resource_spec) + else: + self._cluster: Cluster = SSHCluster(self._resource_spec) # which can be also defined with strategy self._coordinator: Coordinator @tf_contextlib.contextmanager @@ -100,17 +105,13 @@ def build_strategy(self): def _build_or_load_strategy(self, load=False): self._original_graph_item.prepare() - import socket - print(socket.gethostbyname(socket.gethostname()), IS_AUTODIST_CHIEF) if IS_AUTODIST_CHIEF: s = self.build_strategy() s.serialize() else: - if not load: + if IS_ADAPTDL and not load: return - #return None strategy_id = ENV.AUTODIST_STRATEGY_ID.val - print(strategy_id) assert strategy_id s = base.Strategy.deserialize(strategy_id) return s @@ -126,17 +127,22 @@ def _compile_strategy(self, strategy): def _setup(self, strategy): """Prepare for the execution.""" - logging.info(f"{socket.gethostname()} is chief: {IS_AUTODIST_CHIEF}") - if IS_AUTODIST_CHIEF: - # we should only have one single coordinator for one single AutoDist() instance scope, - # even though we could have multiple strategies. - self._coordinator = Coordinator(strategy=strategy, cluster=self._cluster) - self._cluster.start_chief() - self._coordinator.launch_clients_chief() + if not IS_ADAPTDL: + if IS_AUTODIST_CHIEF: + # we should only have one single coordinator for one single AutoDist() instance scope, + # even though we could have multiple strategies. + self._coordinator = Coordinator(strategy=strategy, cluster=self._cluster) + self._cluster.start() + self._coordinator.launch_clients() else: - self._coordinator = Coordinator(strategy=strategy, cluster=self._cluster) - self._cluster.start_worker() - self._coordinator.launch_clients_worker() + if IS_AUTODIST_CHIEF: + self._coordinator = Coordinator(strategy=strategy, cluster=self._cluster) + self._cluster.start_chief() + self._coordinator.launch_clients_chief() + else: + self._coordinator = Coordinator(strategy=strategy, cluster=self._cluster) + self._cluster.start_worker() + self._coordinator.launch_clients_worker() logging.info('Current PID {} belongs to address {}'.format(os.getpid(), self._cluster.get_local_address())) @@ -151,7 +157,8 @@ def _initialize_graph(self): def _build(self): strategy = self._build_or_load_strategy() self._setup(strategy) # Put it before transforming to allow multiple works to transform concurrently - strategy = self._build_or_load_strategy(load=True) + if IS_ADAPTDL: + strategy = self._build_or_load_strategy(load=True) compiled_strategy = self._compile_strategy(strategy) graph_transformer = GraphTransformer( compiled_strategy=compiled_strategy, diff --git a/autodist/cluster.py b/autodist/cluster.py index 1d63908..831e2f3 100644 --- a/autodist/cluster.py +++ b/autodist/cluster.py @@ -40,14 +40,15 @@ from abc import ABCMeta, abstractmethod import paramiko + import adaptdl.collective as collective +import socket from autodist.const import DEFAULT_PORT_RANGE, DEFAULT_WORKING_DIR, ENV from autodist.resource_spec import ResourceSpec from autodist.utils import logging -import socket warnings.filterwarnings(action='ignore', module=paramiko.__name__) - +IS_ADAPTDL = bool(ENV.ADAPTDL.val) class Cluster(metaclass=ABCMeta): """Cluster manager for TensorFlow servers.""" @@ -136,9 +137,12 @@ def get_local_address(self): Returns: str: Worker ip or chief address by default. """ - hostname = socket.gethostname() - local_ip = socket.gethostbyname(hostname) - return local_ip #ENV.AUTODIST_WORKER.val or self._chief + if IS_ADAPTDL: + hostname = socket.gethostname() + local_ip = socket.gethostbyname(hostname) + return local_ip + + return ENV.AUTODIST_WORKER.val or self._chief def get_local_worker_task_index(self): """ @@ -147,11 +151,14 @@ def get_local_worker_task_index(self): Returns: int: Task index """ - logging.info(f"full address {self._full_addresses}") - logging.info(f"local address {self.get_local_address()}") - return_ = [i for i, a in enumerate(self._full_addresses) if self.get_local_address() in a][0] - logging.info(f"returning {return_}") - return return_ + if IS_ADAPTDL: + logging.info(f"full address {self._full_addresses}") + logging.info(f"local address {self.get_local_address()}") + return_ = [i for i, a in enumerate(self._full_addresses) if self.get_local_address() in a][0] + logging.info(f"returning {return_}") + return return_ + + return [i for i, a in enumerate(self._full_addresses) if self.get_local_address() in a][0] def get_local_session_target(self): """ @@ -172,7 +179,7 @@ def start(self): """ # pylint: disable=import-outside-toplevel from autodist.utils import server_starter - assert False + # atexit registration should be placed # - before the beginning of the start # (to ensure the clean termination if the start fails in its half way); and @@ -185,7 +192,7 @@ def start(self): envs = ['{}={}'.format(k, v) for k, v in envs.items()] module_name = server_starter.__name__ module_file = server_starter.__file__ - print(self.cluster_spec) + for job_name, tasks in self.cluster_spec.items(): for task_index, full_address in enumerate(tasks): address = full_address.split(':')[0] @@ -207,9 +214,9 @@ def start(self): full_address, job_name, task_index )) else: # remote - self.remote_pre_start_tf_server(address, tf_server_starter_filepath=module_file) + self.remote_pre_start_tf_server(address, tf_server_starter_filepath=module_file,chief=None) file = os.path.join(DEFAULT_WORKING_DIR, os.path.basename(module_file)) - #bash = envs + envs_cuda + ['python', '-u', file] + args + bash = envs + envs_cuda + ['python', '-u', file] + args logging.info("Launching tf.server on %s" % address) proc = self.remote_exec(bash, hostname=address) # The above line immediately follows the Popen @@ -219,20 +226,10 @@ def start(self): # pylint: disable=too-many-locals def start_chief(self): """ - Start tf.servers on all nodes. - - Note that this only runs (and only should run) on the chief node. + Start tf.servers on all nodes. AdaptDL version. Run on chief. """ # pylint: disable=import-outside-toplevel from autodist.utils import server_starter - # atexit registration should be placed - # - before the beginning of the start - # (to ensure the clean termination if the start fails in its half way); and - # - at the same module as the start - # (to follow the python assumption that - # lower level modules will normally be imported - # before higher level modules and thus must be cleaned up later). - #atexit.register(self.terminate) envs = {ENV.AUTODIST_MIN_LOG_LEVEL.name: 'ERROR'} envs = ['{}={}'.format(k, v) for k, v in envs.items()] module_name = server_starter.__name__ @@ -253,39 +250,19 @@ def start_chief(self): logging.info("cmd at chief: %s", cmd) proc = subprocess.Popen(' '.join(cmd), shell=True, preexec_fn=os.setsid) self.subprocesses.append(proc) - # The above line immediately follows the Popen - # to ensure no gap for termination failure due to the empty proc list. logging.debug('$ local tf.server started at {}: job_name={} task_index={}'.format( full_address, job_name, task_index )) - # else: # remote self.remote_pre_start_tf_server(None, tf_server_starter_filepath=module_file,chief=True) file = os.path.join(DEFAULT_WORKING_DIR, os.path.basename(module_file)) - # bash = envs + envs_cuda + ['python', '-u', file] + args - # logging.info("Launching tf.server on %s" % address) - # proc = self.remote_exec(bash, hostname=address) - # The above line immediately follows the Popen - # to ensure no gap for termination failure due to the empty proc list. - # self.subprocesses.append(proc) # pylint: disable=too-many-locals def start_worker(self): """ - Start tf.servers on all nodes. - - Note that this only runs (and only should run) on the chief node. + Start tf.servers on all nodes. AdaptDL version. Run on non-chief. """ # pylint: disable=import-outside-toplevel from autodist.utils import server_starter - - # atexit registration should be placed - # - before the beginning of the start - # (to ensure the clean termination if the start fails in its half way); and - # - at the same module as the start - # (to follow the python assumption that - # lower level modules will normally be imported - # before higher level modules and thus must be cleaned up later). - #atexit.register(self.terminate) envs = {ENV.AUTODIST_MIN_LOG_LEVEL.name: 'ERROR'} envs = ['{}={}'.format(k, v) for k, v in envs.items()] module_name = server_starter.__name__ @@ -305,14 +282,10 @@ def start_worker(self): envs_cuda = ['CUDA_VISIBLE_DEVICES=""'] assert not self.is_chief(address) self.remote_pre_start_tf_server(address, tf_server_starter_filepath=module_file, chief=False) - #file = os.path.join(DEFAULT_WORKING_DIR, os.path.basename(module_file)) - #bash = envs + envs_cuda + ['python', '-u', file] + args - + cmd = envs + envs_cuda + [sys.executable, '-m', module_name] + args logging.info("Launching tf.server on %s" % address) proc = self.local_exec(cmd, address) - # The above line immediately follows the Popen - # to ensure no gap for termination failure due to the empty proc list. self.subprocesses.append(proc) assert len(self.subprocesses) <= 1 @@ -331,14 +304,24 @@ def remote_pre_start_tf_server(self, hostname, tf_server_starter_filepath, chief tf_server_starter_filepath (str): local starter file path working_dir (str): remote working directory """ - #logging.info("Copying necessary files to %s" % hostname) - self.remote_copy(local_path=tf_server_starter_filepath, remote_path=working_dir, hostname=hostname, chief=chief) - self.remote_file_write( - remote_path=os.path.join(working_dir, 'cluster_spec.json'), - data=json.dumps(self.cluster_spec), - hostname=hostname, - chief=chief - ) + logging.info("Copying necessary files to %s" % hostname) + if IS_ADAPTDL: + self.remote_copy(local_path=tf_server_starter_filepath, remote_path=working_dir, hostname=hostname, chief=chief) + self.remote_file_write( + remote_path=os.path.join(working_dir, 'cluster_spec.json'), + data=json.dumps(self.cluster_spec), + hostname=hostname, + chief=chief + ) + else: + assert chief is None + self.remote_copy(local_path=tf_server_starter_filepath, remote_path=working_dir, hostname=hostname, chief=None) + self.remote_file_write( + remote_path=os.path.join(working_dir, 'cluster_spec.json'), + data=json.dumps(self.cluster_spec), + hostname=hostname, + chief=None + ) @abstractmethod def remote_exec(self, args, hostname): @@ -394,12 +377,12 @@ def _get_ssh_client(self, hostname): Returns: Yields a Paramiko SSHClient. """ - assert False ssh_config = self._ssh_conf[hostname] client = paramiko.SSHClient() client.load_system_host_keys() client.set_missing_host_key_policy(paramiko.WarningPolicy) client.connect(hostname=hostname, port=ssh_config.port, username=ssh_config.username, pkey=ssh_config.pkey) + assert not IS_ADAPTDL yield client client.close() @@ -414,7 +397,6 @@ def _get_sftp_client(self, hostname): Returns: Yields a Paramiko SFTPClient. """ - assert False ssh_config = self._ssh_conf[hostname] t = paramiko.Transport((hostname, ssh_config.port)) t.connect(username=ssh_config.username, pkey=ssh_config.pkey) @@ -434,7 +416,6 @@ def remote_exec(self, args, hostname): Returns: Process: process handle """ - assert False cmd_list = [] ssh_config = self._ssh_conf[hostname] if ssh_config.python_venv: @@ -455,28 +436,46 @@ def remote_exec(self, args, hostname): proc = subprocess.Popen(remote_cmd, shell=True, preexec_fn=os.setsid) return proc - def local_exec(self, args, hostname): - #cmd_list = [] - #ssh_config = self._ssh_conf[hostname] - #if ssh_config.python_venv: - # cmd_list.append('%s;' % ssh_config.python_venv) - #if ssh_config.env: - # cmd_list.extend(['%s=%s' % (k, v) for k, v in ssh_config.env.items()]) - full_cmd = ' '.join(args) + def remote_file_write(self, remote_path, data, hostname): + """ + Write a remote file. - logging.info(full_cmd) - #remote_cmd = '\'bash -c "{}"\' Date: Mon, 7 Dec 2020 05:15:46 +0000 Subject: [PATCH 07/14] minor fix --- autodist/const.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/autodist/const.py b/autodist/const.py index 3770fd3..4643cb6 100644 --- a/autodist/const.py +++ b/autodist/const.py @@ -87,7 +87,7 @@ class ENV(Enum): def val(self): """Return the output of the lambda on the system's value in the environment.""" # pylint: disable=invalid-envvar-value, unpacking-non-sequence - if self.name == "AUTODIST_WORKER" and self.ADAPTDL.val(): + if self.name == "AUTODIST_WORKER" and self.ADAPTDL.val: return self.val_autodist_worker _, default_fn = self.value return default_fn(os.getenv(self.name)) From 39a15c934e0add87648a9a2c23a43f63388c378f Mon Sep 17 00:00:00 2001 From: Dacheng Li Date: Tue, 8 Dec 2020 17:16:18 +0000 Subject: [PATCH 08/14] tested single machine, testing distributed original. --- autodist/autodist.py | 4 +- autodist/cluster.py | 76 +++++-- autodist/const.py | 10 +- autodist/coordinator.py | 41 ++-- examples/benchmark/bert.py | 11 +- examples/benchmark/bert_with_adaptdl.py | 265 ++++++++++++++++++++++++ 6 files changed, 345 insertions(+), 62 deletions(-) create mode 100644 examples/benchmark/bert_with_adaptdl.py diff --git a/autodist/autodist.py b/autodist/autodist.py index c6fb6de..dd3920f 100644 --- a/autodist/autodist.py +++ b/autodist/autodist.py @@ -37,8 +37,6 @@ from autodist.strategy.ps_lb_strategy import PSLoadBalancing from autodist.utils import logging -import adaptdl.collective as collective -import socket IS_AUTODIST_WORKER = bool(ENV.AUTODIST_WORKER.val) IS_AUTODIST_CHIEF = not IS_AUTODIST_WORKER IS_ADAPTDL = bool(ENV.ADAPTDL.val) @@ -110,7 +108,7 @@ def _build_or_load_strategy(self, load=False): s.serialize() else: if IS_ADAPTDL and not load: - return + return None strategy_id = ENV.AUTODIST_STRATEGY_ID.val assert strategy_id s = base.Strategy.deserialize(strategy_id) diff --git a/autodist/cluster.py b/autodist/cluster.py index 831e2f3..f7538bf 100644 --- a/autodist/cluster.py +++ b/autodist/cluster.py @@ -40,15 +40,16 @@ from abc import ABCMeta, abstractmethod import paramiko - -import adaptdl.collective as collective -import socket from autodist.const import DEFAULT_PORT_RANGE, DEFAULT_WORKING_DIR, ENV from autodist.resource_spec import ResourceSpec from autodist.utils import logging warnings.filterwarnings(action='ignore', module=paramiko.__name__) IS_ADAPTDL = bool(ENV.ADAPTDL.val) +if IS_ADAPTDL: + import adaptdl.collective as collective + import socket + class Cluster(metaclass=ABCMeta): """Cluster manager for TensorFlow servers.""" @@ -154,7 +155,7 @@ def get_local_worker_task_index(self): if IS_ADAPTDL: logging.info(f"full address {self._full_addresses}") logging.info(f"local address {self.get_local_address()}") - return_ = [i for i, a in enumerate(self._full_addresses) if self.get_local_address() in a][0] + return_ = [i for i, a in enumerate(self._full_addresses) if self.get_local_address() in a][0] logging.info(f"returning {return_}") return return_ @@ -214,7 +215,7 @@ def start(self): full_address, job_name, task_index )) else: # remote - self.remote_pre_start_tf_server(address, tf_server_starter_filepath=module_file,chief=None) + self.remote_pre_start_tf_server(address, tf_server_starter_filepath=module_file, chief=False) file = os.path.join(DEFAULT_WORKING_DIR, os.path.basename(module_file)) bash = envs + envs_cuda + ['python', '-u', file] + args logging.info("Launching tf.server on %s" % address) @@ -225,9 +226,7 @@ def start(self): # pylint: disable=too-many-locals def start_chief(self): - """ - Start tf.servers on all nodes. AdaptDL version. Run on chief. - """ + """Start tf.servers on all nodes. AdaptDL version. Run on chief.""" # pylint: disable=import-outside-toplevel from autodist.utils import server_starter envs = {ENV.AUTODIST_MIN_LOG_LEVEL.name: 'ERROR'} @@ -253,14 +252,11 @@ def start_chief(self): logging.debug('$ local tf.server started at {}: job_name={} task_index={}'.format( full_address, job_name, task_index )) - self.remote_pre_start_tf_server(None, tf_server_starter_filepath=module_file,chief=True) - file = os.path.join(DEFAULT_WORKING_DIR, os.path.basename(module_file)) + self.remote_pre_start_tf_server(None, tf_server_starter_filepath=module_file, chief=True) # pylint: disable=too-many-locals def start_worker(self): - """ - Start tf.servers on all nodes. AdaptDL version. Run on non-chief. - """ + """Start tf.servers on all nodes. AdaptDL version. Run on non-chief.""" # pylint: disable=import-outside-toplevel from autodist.utils import server_starter envs = {ENV.AUTODIST_MIN_LOG_LEVEL.name: 'ERROR'} @@ -302,11 +298,14 @@ def remote_pre_start_tf_server(self, hostname, tf_server_starter_filepath, chief Args: hostname (str): host name or address tf_server_starter_filepath (str): local starter file path + chief (bool): indicator that this process is on chief or not. Only apply with adaptDL. working_dir (str): remote working directory """ logging.info("Copying necessary files to %s" % hostname) if IS_ADAPTDL: - self.remote_copy(local_path=tf_server_starter_filepath, remote_path=working_dir, hostname=hostname, chief=chief) + # pylint: disable=unexpected-keyword-arg + self.remote_copy(local_path=tf_server_starter_filepath, remote_path=working_dir, + hostname=hostname, chief=chief) self.remote_file_write( remote_path=os.path.join(working_dir, 'cluster_spec.json'), data=json.dumps(self.cluster_spec), @@ -314,13 +313,12 @@ def remote_pre_start_tf_server(self, hostname, tf_server_starter_filepath, chief chief=chief ) else: - assert chief is None - self.remote_copy(local_path=tf_server_starter_filepath, remote_path=working_dir, hostname=hostname, chief=None) + assert chief is False + self.remote_copy(local_path=tf_server_starter_filepath, remote_path=working_dir, hostname=hostname) self.remote_file_write( remote_path=os.path.join(working_dir, 'cluster_spec.json'), data=json.dumps(self.cluster_spec), - hostname=hostname, - chief=None + hostname=hostname ) @abstractmethod @@ -472,10 +470,41 @@ class ADAPTDLCluster(Cluster): def __init__(self, resource_spec): assert IS_ADAPTDL super().__init__(resource_spec) - + def remote_exec(self, args, hostname): - pass + """ + Execute a bash script remotely. disabled in AdaptDL. + + Args: + args (list): bash commands + hostname (str): host name or address + + Returns: + None + """ + return + + # pylint: disable=no-self-use + def local_exec(self, args, hostname): + """ + Execute a bash script locally. + + Args: + args (list): bash commands + hostname (str): host name or address + + Returns: + Process: process handle + """ + full_cmd = ' '.join(args) + logging.info(full_cmd) + if ENV.AUTODIST_DEBUG_REMOTE.val: + return None + # pylint: disable=subprocess-popen-preexec-fn + proc = subprocess.Popen(full_cmd, shell=True, preexec_fn=os.setsid) + return proc + # pylint: disable=arguments-differ def remote_file_write(self, remote_path, data, hostname, chief): """ Write a remote file. @@ -494,6 +523,7 @@ def remote_file_write(self, remote_path, data, hostname, chief): f.write(data_) f.close() + # pylint: disable=arguments-differ def remote_copy(self, local_path, remote_path, hostname, chief): """ Copy a file to a remote directory. @@ -515,7 +545,7 @@ def remote_copy(self, local_path, remote_path, hostname, chief): lines = collective.broadcast(None) if not os.path.isdir(remote_path): os.mkdir(remote_path) - f = open(os.path.join(remote_path, os.path.basename(local_path)),"w") - for l in lines: - f.write(l) + f = open(os.path.join(remote_path, os.path.basename(local_path)), "w") + for line in lines: + f.write(line) f.close() diff --git a/autodist/const.py b/autodist/const.py index 4643cb6..513ce56 100644 --- a/autodist/const.py +++ b/autodist/const.py @@ -23,8 +23,6 @@ from enum import Enum, auto import os -import socket -import adaptdl.env as env # Below consts can be modified if necessary. # Note that if one of these consts requires frequent modification, # it should probably be moved into `ENV`. @@ -86,13 +84,17 @@ class ENV(Enum): @property def val(self): """Return the output of the lambda on the system's value in the environment.""" - # pylint: disable=invalid-envvar-value, unpacking-non-sequence + # pylint: disable=invalid-envvar-value, unpacking-non-sequence, comparison-with-callable if self.name == "AUTODIST_WORKER" and self.ADAPTDL.val: - return self.val_autodist_worker + return self.val_autodist_worker() _, default_fn = self.value return default_fn(os.getenv(self.name)) + # pylint: disable=no-self-use, import-outside-toplevel def val_autodist_worker(self): + """Evaluate autodist_worker in AdaptDL.""" + import adaptdl.env as env + import socket f = open(os.path.join(env.share_path(), "resource_spec.yml")) lines = f.readlines() line_chief = lines[1] diff --git a/autodist/coordinator.py b/autodist/coordinator.py index baee63a..7840c66 100644 --- a/autodist/coordinator.py +++ b/autodist/coordinator.py @@ -19,15 +19,15 @@ import atexit import os -import socket from autodist.const import ENV, DEFAULT_SERIALIZATION_DIR from autodist.resource_spec import DeviceSpec from autodist.utils import logging -from autodist.const import ENV - -import adaptdl.collective as collective IS_ADAPTDL = bool(ENV.ADAPTDL.val) +if IS_ADAPTDL: + import socket + import adaptdl.collective as collective + class Coordinator: """ @@ -59,7 +59,6 @@ def launch_clients(self): """ assert not IS_ADAPTDL atexit.register(self.join) - replica_devices = [ DeviceSpec.from_string(device_string) for device_string in self._strategy.graph_config.replicas @@ -94,25 +93,18 @@ def launch_clients(self): ) proc = self.cluster.remote_exec(cmd, hostname=replica_host) self.threads.append(self._proc_wait_async(proc)) - def launch_clients_chief(self): - """ - Launch the user's code on each worker. ADAPTDL version, chief run. - """ - replica_devices = [ - DeviceSpec.from_string(device_string) - for device_string in self._strategy.graph_config.replicas - ] - replica_hosts = {d.host_address for d in replica_devices} + def launch_clients_chief(self): + """Launch the user's code on each worker. ADAPTDL version, chief run.""" env = { - ENV.AUTODIST_WORKER.name: None, - ENV.AUTODIST_STRATEGY_ID.name: self._strategy.id, - ENV.AUTODIST_MIN_LOG_LEVEL.name: ENV.AUTODIST_MIN_LOG_LEVEL.val, - ENV.AUTODIST_IS_TESTING.name: ENV.AUTODIST_IS_TESTING.val, - ENV.AUTODIST_PATCH_TF.name: ENV.AUTODIST_PATCH_TF.val, - ENV.AUTODIST_INTERNAL_TF.name: ENV.AUTODIST_INTERNAL_TF.val, - ENV.SYS_DATA_PATH.name: ENV.SYS_DATA_PATH.val, - ENV.SYS_RESOURCE_PATH.name: ENV.SYS_RESOURCE_PATH.val, + ENV.AUTODIST_WORKER.name: None, + ENV.AUTODIST_STRATEGY_ID.name: self._strategy.id, + ENV.AUTODIST_MIN_LOG_LEVEL.name: ENV.AUTODIST_MIN_LOG_LEVEL.val, + ENV.AUTODIST_IS_TESTING.name: ENV.AUTODIST_IS_TESTING.val, + ENV.AUTODIST_PATCH_TF.name: ENV.AUTODIST_PATCH_TF.val, + ENV.AUTODIST_INTERNAL_TF.name: ENV.AUTODIST_INTERNAL_TF.val, + ENV.SYS_DATA_PATH.name: ENV.SYS_DATA_PATH.val, + ENV.SYS_RESOURCE_PATH.name: ENV.SYS_RESOURCE_PATH.val, } collective.broadcast(env) @@ -123,14 +115,15 @@ def launch_clients_chief(self): hostname=None, chief=True ) - + def launch_clients_worker(self): + """Launch the user's code on each worker. ADAPTDL version, non-chief run.""" hostname = socket.gethostname() local_ip = socket.gethostbyname(hostname) env = collective.broadcast(None) env[ENV.AUTODIST_WORKER.name] = local_ip - for k,v in env.items(): + for k, v in env.items(): os.environ[k] = str(v) path = collective.broadcast(None) diff --git a/examples/benchmark/bert.py b/examples/benchmark/bert.py index 562d8af..8caa995 100644 --- a/examples/benchmark/bert.py +++ b/examples/benchmark/bert.py @@ -29,18 +29,13 @@ from absl import logging from utils.logs import logger -from utils.misc import keras_utils, gen_yml +from utils.misc import keras_utils from utils import bert_modeling as modeling from utils import bert_models from utils import common_flags from utils import input_pipeline from utils import bert_utils -import adaptdl.torch as adl -import adaptdl.env as env -import adaptdl.torch as adl - -gen_yml.generate() ######################################################################### # Import AutoDist and Strategy @@ -202,8 +197,8 @@ def main(_): else: os.environ['AUTODIST_PATCH_TF'] = 'False' resource_spec_file = os.path.join( - env.share_path(), - 'resource_spec.yml') + os.path.dirname(__file__), + '../resource_spec.yml') if FLAGS.autodist_strategy == 'PS': strategy = AutoDist( diff --git a/examples/benchmark/bert_with_adaptdl.py b/examples/benchmark/bert_with_adaptdl.py new file mode 100644 index 0000000..562d8af --- /dev/null +++ b/examples/benchmark/bert_with_adaptdl.py @@ -0,0 +1,265 @@ +# Copyright 2020 Petuum, Inc. All Rights Reserved. +# +# It includes the derived work based on: +# +# Copyright 2019 The TensorFlow Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ============================================================================== +from __future__ import absolute_import +from __future__ import division +from __future__ import print_function + +import yaml +import os +import sys +import tensorflow as tf +from absl import app +from absl import flags +from absl import logging + +from utils.logs import logger +from utils.misc import keras_utils, gen_yml + +from utils import bert_modeling as modeling +from utils import bert_models +from utils import common_flags +from utils import input_pipeline +from utils import bert_utils +import adaptdl.torch as adl +import adaptdl.env as env +import adaptdl.torch as adl + +gen_yml.generate() + +######################################################################### +# Import AutoDist and Strategy +from autodist import AutoDist +from autodist.strategy.all_reduce_strategy import AllReduce +from autodist.strategy.ps_strategy import PS +from autodist.strategy.ps_lb_strategy import PSLoadBalancing +from autodist.strategy.parallax_strategy import Parallax +from autodist.strategy.partitioned_ps_strategy import PartitionedPS +######################################################################### + +flags.DEFINE_string( + 'input_files', + None, + 'File path to retrieve training data for pre-training.') +flags.DEFINE_integer( + 'max_seq_length', 128, + 'The maximum total input sequence length after WordPiece tokenization. ' + 'Sequences longer than this will be truncated, and sequences shorter ' + 'than this will be padded.') +flags.DEFINE_integer('max_predictions_per_seq', 20, + 'Maximum predictions per sequence_output.') +flags.DEFINE_integer('train_batch_size', 8, 'Total batch size for training.') +flags.DEFINE_integer('chunk_size', 256, 'The chunk size for training.') +flags.DEFINE_integer('num_steps_per_epoch', 1000, + 'Total number of training steps to run per epoch.') +flags.DEFINE_string( + name='autodist_strategy', + default='PS', + help='the autodist strategy') +flags.DEFINE_boolean( + name='autodist_patch_tf', + default=True, + help='AUTODIST_PATCH_TF') + +flags.DEFINE_boolean(name='proxy', default=True, help='turn on off the proxy') + + +common_flags.define_common_bert_flags() + +FLAGS = flags.FLAGS + + +def get_pretrain_dataset_fn(input_file_pattern, seq_length, + max_predictions_per_seq, global_batch_size, + num_replicas_in_sync): + """Returns input dataset from input file string.""" + def _dataset_fn(ctx=None): + """Returns tf.data.Dataset for distributed BERT pretraining.""" + input_patterns = input_file_pattern.split(',') + batch_size = int(global_batch_size / num_replicas_in_sync) + train_dataset = input_pipeline.create_pretrain_dataset( + input_patterns, + seq_length, + max_predictions_per_seq, + batch_size, + is_training=True) + return train_dataset + + return _dataset_fn + + +def get_loss_fn(loss_factor=1.0): + """Returns loss function for BERT pretraining.""" + + def _bert_pretrain_loss_fn(unused_labels, losses, **unused_args): + return tf.keras.backend.mean(losses) * loss_factor + + return _bert_pretrain_loss_fn + + +def run_customized_training(strategy, + bert_config, + max_seq_length, + max_predictions_per_seq, + model_dir, + steps_per_epoch, + steps_per_loop, + epochs, + initial_lr, + input_files, + train_batch_size): + """Run BERT pretrain model training using low-level API.""" + if strategy is not None: + num_replicas_in_sync = strategy.num_replicas_in_sync + else: + num_replicas_in_sync = 1 + + train_input_fn = get_pretrain_dataset_fn(input_files, max_seq_length, + max_predictions_per_seq, + train_batch_size, + num_replicas_in_sync) + + def _get_pretrain_model(): + """Gets a pretraining model.""" + pretrain_model, core_model = bert_models.pretrain_model( + bert_config, max_seq_length, max_predictions_per_seq) + + pretrain_model.optimizer = tf.optimizers.Adam(lr=initial_lr) + if FLAGS.fp16_implementation == 'graph_rewrite': + pretrain_model.optimizer = tf.train.experimental.enable_mixed_precision_graph_rewrite( + pretrain_model.optimizer) + return pretrain_model, core_model + + time_callback = keras_utils.TimeHistory( + train_batch_size * steps_per_loop, 1) + + ########################################################################## + # Build with Graph mode and AutoDist scope in bert_utils + trained_model = bert_utils.run_customized_training_loop( + strategy=strategy, + model_fn=_get_pretrain_model, + loss_fn=get_loss_fn( + loss_factor=1.0 / + num_replicas_in_sync if FLAGS.scale_loss else 1.0), + model_dir=model_dir, + train_input_fn=train_input_fn, + steps_per_epoch=steps_per_epoch, + steps_per_loop=steps_per_loop, + epochs=epochs, + sub_model_export_name='pretrained/bert_model', + custom_callbacks=[time_callback]) + ########################################################################## + return trained_model + + +def run_bert_pretrain(strategy, gpu_num=1, node_num=1): + """Runs BERT pre-training.""" + + bert_config = modeling.BertConfig.from_json_file(FLAGS.bert_config_file) + logging.info( + 'Training using customized training loop TF 2.0 with distrubuted' + 'strategy.') + + return run_customized_training( + strategy, + bert_config, + FLAGS.max_seq_length, + FLAGS.max_predictions_per_seq, + FLAGS.model_dir, + FLAGS.num_steps_per_epoch, + FLAGS.steps_per_loop, + FLAGS.num_train_epochs, + FLAGS.learning_rate, + FLAGS.input_files, + FLAGS.train_batch_size * gpu_num * node_num) + + +def main(_): + assert tf.version.VERSION.startswith('2.') + + if not FLAGS.model_dir: + FLAGS.model_dir = '/tmp/bert/' + + ######################################################################### + # Construct AutoDist with ResourceSpec for Different Strategies + if FLAGS.autodist_patch_tf: + os.environ['AUTODIST_PATCH_TF'] = 'True' + else: + os.environ['AUTODIST_PATCH_TF'] = 'False' + resource_spec_file = os.path.join( + env.share_path(), + 'resource_spec.yml') + + if FLAGS.autodist_strategy == 'PS': + strategy = AutoDist( + resource_spec_file, PS( + local_proxy_variable=FLAGS.proxy)) + elif FLAGS.autodist_strategy == 'PSLoadBalancing': + strategy = AutoDist( + resource_spec_file, PSLoadBalancing( + local_proxy_variable=FLAGS.proxy)) + elif FLAGS.autodist_strategy == 'PartitionedPS': + strategy = AutoDist( + resource_spec_file, PartitionedPS( + local_proxy_variable=FLAGS.proxy)) + elif FLAGS.autodist_strategy == 'AllReduce': + strategy = AutoDist( + resource_spec_file, AllReduce( + chunk_size=FLAGS.chunk_size)) + elif FLAGS.autodist_strategy == 'Parallax': + strategy = AutoDist( + resource_spec_file, + Parallax( + chunk_size=FLAGS.chunk_size, + local_proxy_variable=FLAGS.proxy)) + else: + raise ValueError( + 'the strategy can be only from PS, PSLoadBalancing, PartitionedPS, AllReduce, Parallax') + + strategy.num_replicas_in_sync = strategy._resource_spec.num_gpus + + if strategy: + print('***** Number of cores used : ', strategy.num_replicas_in_sync) + + resource_info = yaml.safe_load(open(resource_spec_file, 'r')) + try: + node_num = len(resource_info['nodes']) + except ValueError: + print("nodes need to be set in specficiation file") + + try: + gpu_num = len(resource_info['nodes'][0]['gpus']) + except ValueError: + print("gpus need to be set in specficiation file") + ######################################################################### + + logdir = '/tmp/logs' + if not os.path.exists(logdir): + os.makedirs(logdir) + + logname = 'bert_strategy_{}_node_{}_gpu_{}_patch_{}_proxy_{}'.format( + FLAGS.autodist_strategy, node_num, gpu_num, FLAGS.autodist_patch_tf, FLAGS.proxy) + + logging.get_absl_handler().use_absl_log_file(logname, logdir) + # start running + run_bert_pretrain(strategy, gpu_num, node_num) + + +if __name__ == '__main__': + logging.set_verbosity(logging.INFO) + app.run(main) From 0fcb5bb57a1369c272dea58343b2cc882e7c6ed5 Mon Sep 17 00:00:00 2001 From: Dacheng Li Date: Thu, 31 Dec 2020 20:26:02 +0000 Subject: [PATCH 09/14] addressed comments --- autodist/autodist.py | 8 ++++++-- autodist/cluster.py | 14 ++++++-------- autodist/const.py | 5 +++-- autodist/resource_spec.py | 1 - 4 files changed, 15 insertions(+), 13 deletions(-) diff --git a/autodist/autodist.py b/autodist/autodist.py index dd3920f..962f1c6 100644 --- a/autodist/autodist.py +++ b/autodist/autodist.py @@ -107,6 +107,10 @@ def _build_or_load_strategy(self, load=False): s = self.build_strategy() s.serialize() else: + # At AdaptDL mode, when the worker pass through this before + # the chief has created the strategy, this should returns + # nothing. Later, when the chief has created the strategy, + # it can load it. if IS_ADAPTDL and not load: return None strategy_id = ENV.AUTODIST_STRATEGY_ID.val @@ -115,12 +119,12 @@ def _build_or_load_strategy(self, load=False): return s def _compile_strategy(self, strategy): - #logging.debug('Raw strategy: %s' % strategy) + logging.debug('Raw strategy: %s' % strategy) device_resolver = DeviceResolver(self._cluster) compiled_strategy = base.StrategyCompiler(self._original_graph_item) \ .set_device_resolver(device_resolver.resolve_to_device_str) \ .compile(strategy) - #logging.info('Compiled strategy: %s' % compiled_strategy) + logging.info('Compiled strategy: %s' % compiled_strategy) return compiled_strategy def _setup(self, strategy): diff --git a/autodist/cluster.py b/autodist/cluster.py index f7538bf..ecaf087 100644 --- a/autodist/cluster.py +++ b/autodist/cluster.py @@ -484,8 +484,8 @@ def remote_exec(self, args, hostname): """ return - # pylint: disable=no-self-use - def local_exec(self, args, hostname): + @staticmethod + def local_exec(args, hostname): """ Execute a bash script locally. @@ -504,8 +504,7 @@ def local_exec(self, args, hostname): proc = subprocess.Popen(full_cmd, shell=True, preexec_fn=os.setsid) return proc - # pylint: disable=arguments-differ - def remote_file_write(self, remote_path, data, hostname, chief): + def remote_file_write(self, remote_path, data, hostname, **kwargs): """ Write a remote file. @@ -515,7 +514,7 @@ def remote_file_write(self, remote_path, data, hostname, chief): hostname (str): host name or address chief (boolean): whether this is autodist chief """ - if chief: + if kwargs["chief"]: _ = collective.broadcast(data) else: data_ = collective.broadcast(None) @@ -523,8 +522,7 @@ def remote_file_write(self, remote_path, data, hostname, chief): f.write(data_) f.close() - # pylint: disable=arguments-differ - def remote_copy(self, local_path, remote_path, hostname, chief): + def remote_copy(self, local_path, remote_path, hostname, **kwargs): """ Copy a file to a remote directory. @@ -536,7 +534,7 @@ def remote_copy(self, local_path, remote_path, hostname, chief): """ # Make sure directory exists - if chief: + if kwargs["chief"]: f = open(local_path, "r") lines = f.readlines() _ = collective.broadcast(lines) diff --git a/autodist/const.py b/autodist/const.py index 513ce56..83e3d94 100644 --- a/autodist/const.py +++ b/autodist/const.py @@ -90,8 +90,9 @@ def val(self): _, default_fn = self.value return default_fn(os.getenv(self.name)) - # pylint: disable=no-self-use, import-outside-toplevel - def val_autodist_worker(self): + # pylint: disable=import-outside-toplevel + @staticmethod + def val_autodist_worker(): """Evaluate autodist_worker in AdaptDL.""" import adaptdl.env as env import socket diff --git a/autodist/resource_spec.py b/autodist/resource_spec.py index ef3ca36..5945aef 100644 --- a/autodist/resource_spec.py +++ b/autodist/resource_spec.py @@ -204,7 +204,6 @@ def _parse_node(self, node, num_nodes): gpu = DeviceSpec(host_address, host_cpu, DeviceType.GPU, gpu_index) self._add_device(gpu) self.__ssh_group[host_address] = node.get('ssh_config') - print(node.get('ssh_config')) if self.__ssh_group[host_address] is None and self.__chief_address != host_address: raise ValueError("Need to define SSH groups for all non-chief nodes.") # handle network bandwidth (optional) From 89ec0a8033a1b5cd378d6d9ad6cf16a0b1eaa6fd Mon Sep 17 00:00:00 2001 From: Dacheng Li Date: Fri, 1 Jan 2021 23:44:30 +0000 Subject: [PATCH 10/14] turn off proxy --- examples/benchmark/bert_with_adaptdl.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/benchmark/bert_with_adaptdl.py b/examples/benchmark/bert_with_adaptdl.py index 562d8af..1f293e8 100644 --- a/examples/benchmark/bert_with_adaptdl.py +++ b/examples/benchmark/bert_with_adaptdl.py @@ -76,7 +76,7 @@ default=True, help='AUTODIST_PATCH_TF') -flags.DEFINE_boolean(name='proxy', default=True, help='turn on off the proxy') +flags.DEFINE_boolean(name='proxy', default=False, help='turn on off the proxy') common_flags.define_common_bert_flags() From 6d7cf332a5abf4bcb6003ac93bb8f1b25aa09953 Mon Sep 17 00:00:00 2001 From: Dacheng Li Date: Mon, 11 Jan 2021 21:33:31 +0000 Subject: [PATCH 11/14] update hostip --- autodist/cluster.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/autodist/cluster.py b/autodist/cluster.py index ecaf087..a5920e8 100644 --- a/autodist/cluster.py +++ b/autodist/cluster.py @@ -471,6 +471,22 @@ def __init__(self, resource_spec): assert IS_ADAPTDL super().__init__(resource_spec) + def get_local_address(self): + """ + Get the local (ip) address. + + If labelled as AUTODIST_WORKER by the environment variable, + the value of it is the address of the local node; + otherwise the local node is chief. + + Returns: + str: Worker ip or chief address by default. + """ + hostname = socket.gethostname() + local_ip = socket.gethostbyname(hostname) + return local_ip + + return ENV.AUTODIST_WORKER.val or self._chief def remote_exec(self, args, hostname): """ Execute a bash script remotely. disabled in AdaptDL. From 0d0ce4158165224ae6b6ce5d38d4be492d78123f Mon Sep 17 00:00:00 2001 From: Dacheng Li Date: Mon, 11 Jan 2021 21:37:01 +0000 Subject: [PATCH 12/14] update staticmethod --- autodist/cluster.py | 17 ++++------------- 1 file changed, 4 insertions(+), 13 deletions(-) diff --git a/autodist/cluster.py b/autodist/cluster.py index a5920e8..02830d6 100644 --- a/autodist/cluster.py +++ b/autodist/cluster.py @@ -138,11 +138,6 @@ def get_local_address(self): Returns: str: Worker ip or chief address by default. """ - if IS_ADAPTDL: - hostname = socket.gethostname() - local_ip = socket.gethostbyname(hostname) - return local_ip - return ENV.AUTODIST_WORKER.val or self._chief def get_local_worker_task_index(self): @@ -471,22 +466,18 @@ def __init__(self, resource_spec): assert IS_ADAPTDL super().__init__(resource_spec) - def get_local_address(self): + @staticmethod + def get_local_address(): """ Get the local (ip) address. - If labelled as AUTODIST_WORKER by the environment variable, - the value of it is the address of the local node; - otherwise the local node is chief. - Returns: - str: Worker ip or chief address by default. + str: local ip """ hostname = socket.gethostname() local_ip = socket.gethostbyname(hostname) return local_ip - - return ENV.AUTODIST_WORKER.val or self._chief + def remote_exec(self, args, hostname): """ Execute a bash script remotely. disabled in AdaptDL. From a2305b187d7b051efd3c46675ec2e30a4e5da7e3 Mon Sep 17 00:00:00 2001 From: Dacheng Li Date: Mon, 11 Jan 2021 21:47:21 +0000 Subject: [PATCH 13/14] lint --- autodist/cluster.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/autodist/cluster.py b/autodist/cluster.py index 02830d6..9b252f4 100644 --- a/autodist/cluster.py +++ b/autodist/cluster.py @@ -477,7 +477,7 @@ def get_local_address(): hostname = socket.gethostname() local_ip = socket.gethostbyname(hostname) return local_ip - + def remote_exec(self, args, hostname): """ Execute a bash script remotely. disabled in AdaptDL. From 65f3649dfb22f4673b301354277ad5ad02936824 Mon Sep 17 00:00:00 2001 From: Dacheng Li Date: Mon, 11 Jan 2021 21:54:10 +0000 Subject: [PATCH 14/14] avoid chief build twice --- autodist/autodist.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/autodist/autodist.py b/autodist/autodist.py index 962f1c6..b6c82c4 100644 --- a/autodist/autodist.py +++ b/autodist/autodist.py @@ -159,7 +159,7 @@ def _initialize_graph(self): def _build(self): strategy = self._build_or_load_strategy() self._setup(strategy) # Put it before transforming to allow multiple works to transform concurrently - if IS_ADAPTDL: + if IS_ADAPTDL and not IS_AUTODIST_CHIEF: strategy = self._build_or_load_strategy(load=True) compiled_strategy = self._compile_strategy(strategy) graph_transformer = GraphTransformer(