Add horovod.run.run to make horovod notebook friendly (new impl)

.buildkite/gen-pipeline.sh

@@ -28,7 +28,6 @@ tests=( \
        test-cpu-openmpi-py3_6-tf2_0_0-keras2_2_4-torch1_2_0-mxnet1_5_0-pyspark2_4_0 \
        test-cpu-openmpi-py2_7-tfhead-kerashead-torchhead-mxnethead-pyspark2_4_0 \
        test-cpu-openmpi-py3_6-tfhead-kerashead-torchhead-mxnethead-pyspark2_4_0 \
-       test-cpu-mpich-py3_6-tf1_14_0-keras2_2_4-torch1_1_0-mxnet1_5_0-pyspark2_4_0 \
        test-cpu-mlsl-py3_6-tf1_14_0-keras2_2_4-torch1_1_0-mxnet1_5_0-pyspark2_4_0 \
        test-gpu-openmpi-py2_7-tf1_6_0-keras2_1_2-torch0_4_1-mxnet1_4_1-pyspark2_3_2 \
        test-gpu-openmpi-py3_5-tf1_6_0-keras2_1_2-torch0_4_1-mxnet1_4_1-pyspark2_3_2 \
@@ -42,7 +41,6 @@ tests=( \
        test-gpu-openmpi-py3_6-tf2_0_0-keras2_2_4-torch1_2_0-mxnet1_5_0-pyspark2_4_0 \
        test-gpu-openmpi-py2_7-tfhead-kerashead-torchhead-mxnethead-pyspark2_4_0 \
        test-gpu-openmpi-py3_6-tfhead-kerashead-torchhead-mxnethead-pyspark2_4_0 \
-       test-gpu-mpich-py3_6-tf1_14_0-keras2_2_4-torch1_1_0-mxnet1_5_0-pyspark2_4_0 \
        test-mixed-openmpi-py3_6-tf1_14_0-keras2_2_4-torch1_1_0-mxnet1_5_0-pyspark2_4_0 \
 )
 
@@ -116,9 +114,16 @@ run_all() {
     exclude_keras_if_needed="| sed 's/[a-z_]*keras[a-z_.]*//g'"
   fi
 
+  local exclude_interactiverun="| sed 's/test_interactiverun.py//g'"
+
   run_test "${test}" "${queue}" \
     ":pytest: Run PyTests (${test})" \
-    "bash -c \"cd /horovod/test && (echo test_*.py ${exclude_keras_if_needed} | xargs -n 1 \\\$(cat /mpirun_command) pytest -v --capture=no)\""
+    "bash -c \"cd /horovod/test && (echo test_*.py ${exclude_keras_if_needed} ${exclude_interactiverun} | xargs -n 1 \\\$(cat /mpirun_command) pytest -v --capture=no)\""
+
+  # Run test_interactiverun.py
+  run_test "${test}" "${queue}" \
+    ":pytest: Run PyTests test_interactiverun (${test})" \
+    "bash -c \"cd /horovod/test && pytest -v --capture=no test_interactiverun.py\""
 
   # Legacy TensorFlow tests
   if [[ ${test} != *"tf2_"* ]]; then
@@ -185,9 +190,11 @@ run_gloo() {
     exclude_spark_if_needed="| sed 's/[a-z_]*spark[a-z_.]*//g'"
   fi
 
+  local exclude_interactiverun="| sed 's/test_interactiverun.py//g'"
+
   run_test "${test}" "${queue}" \
     ":pytest: Run PyTests (${test})" \
-    "bash -c \"cd /horovod/test && (echo test_*.py ${exclude_spark_if_needed} | xargs -n 1 horovodrun -np 2 -H localhost:2 --gloo pytest -v --capture=no)\""
+    "bash -c \"cd /horovod/test && (echo test_*.py ${exclude_spark_if_needed} ${exclude_interactiverun} | xargs -n 1 horovodrun -np 2 -H localhost:2 --gloo pytest -v --capture=no)\""
 
   run_test "${test}" "${queue}" \
     ":muscle: Test Keras MNIST (${test})" \

bin/horovodrun

@@ -15,7 +15,7 @@
 # limitations under the License.
 # ==============================================================================
 
-from horovod.run import run
+from horovod.run.run import run_commandline
 
 if __name__ == '__main__':
-    run.run()
+    run_commandline()

docs/requirements.txt

@@ -2,3 +2,4 @@ git+https://github.com/sphinx-doc/sphinx@2.0
 sphinxcontrib-napoleon
 alabaster
 nbsphinx
+pyyaml

horovod/run/__init__.py

@@ -0,0 +1,16 @@
+# Copyright 2019 Uber Technologies, 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.
+# =============================================================================
+
+from .run import run

horovod/run/common/util/settings.py

@@ -17,7 +17,7 @@
 class Settings(object):
 
     def __init__(self, verbose=0, ssh_port=None, key=None, timeout=None,
-                 num_hosts=None, num_proc=None, hosts=None, command=None):
+                 num_hosts=None, num_proc=None, hosts=None):
         """
         :param verbose: level of verbosity
         :type verbose: int
@@ -44,4 +44,9 @@ def __init__(self, verbose=0, ssh_port=None, key=None, timeout=None,
         self.num_hosts = num_hosts
         self.num_proc = num_proc
         self.hosts = hosts
+        self.command = None
+        self.run_func_mode = None
+
+    def set_command(self, command, run_func_mode):
         self.command = command
+        self.run_func_mode = run_func_mode

horovod/run/gloo_run.py

@@ -18,7 +18,7 @@
 import threading
 import signal
 
-from horovod.run.rendezvous.http_server import RendezvousServer
+from horovod.run.http.http_server import RendezvousServer
 from horovod.run.common.util import env as env_util, safe_shell_exec
 from horovod.run.util import threads
 from psutil import net_if_addrs
@@ -154,7 +154,7 @@ def set_event_on_sigterm(signum, frame):
 
     args_list = []
     for alloc_info in host_alloc_plan:
-        # generate env for rendezvous
+        # generate env for http
         horovod_rendez_env = 'HOROVOD_RANK={rank} HOROVOD_SIZE={size} ' \
                              'HOROVOD_LOCAL_RANK={local_rank} HOROVOD_LOCAL_SIZE={local_size} ' \
                              'HOROVOD_CROSS_RANK={cross_rank} HOROVOD_CROSS_SIZE={cross_size} ' \
@@ -195,25 +195,16 @@ def set_event_on_sigterm(signum, frame):
                                            block_until_all_done=True)
 
 
-def gloo_run(settings, remote_host_names, common_intfs, env):
+def gloo_run(settings, remote_host_names, common_intfs, env, server_ip):
     # allocate processes into slots
     host_alloc_plan = _allocate(settings.hosts, settings.num_proc)
 
-    # create global rendezvous server
+    # create global http server
     global_rendezv = RendezvousServer(settings.verbose)
-    # Start rendezvous server and get port that it is listening
+    # Start http server and get port that it is listening
     global_rendezv_port = global_rendezv.start_server(host_alloc_plan)
 
-    # get the server IPv4 address
     iface = list(common_intfs)[0]
-    server_ip = None
-    for addr in net_if_addrs()[iface]:
-        if addr.family == AF_INET:
-            server_ip = addr.address
-
-    if not server_ip:
-        raise RuntimeError(
-            'Cannot find an IPv4 address of the common interface.')
 
     run_command = (
         'HOROVOD_GLOO_RENDEZVOUS_ADDR={addr} '

horovod/run/http/http_client.py

@@ -0,0 +1,48 @@
+# Copyright 2019 Uber Technologies, 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 sys
+import base64
+if sys.version < '3':
+    from urllib2 import urlopen
+    from urllib2 import Request
+    from urllib2 import HTTPError, URLError
+else:
+    from urllib.request import urlopen
+    from urllib.request import Request
+    from urllib.error import HTTPError, URLError
+
+
+def read_data_from_kvstore(addr, port, scope, key):
+    try:
+        url = "http://{addr}:{port}/{scope}/{key}".format(
+            addr=addr, port=str(port), scope=scope, key=key
+        )
+        req = Request(url)
+        resp = urlopen(req)
+        return base64.b64decode(resp.read())
+    except (HTTPError, URLError) as e:
+                raise RuntimeError("Read data from KVStore server failed.", e)
+
+
+def put_data_into_kvstore(addr, port, scope, key, value):
+    try:
+        url = "http://{addr}:{port}/{scope}/{key}".format(
+            addr=addr, port=str(port), scope=scope, key=key
+        )
+        req = Request(url, data=base64.b64encode(value))
+        req.get_method = lambda: "PUT"  # for urllib2 compatibility
+        urlopen(req)
+    except (HTTPError, URLError) as e:
+        raise RuntimeError("Put data input KVStore server failed.", e)

horovod/run/rendezvous/http_server.py → horovod/run/http/http_server.py

@@ -30,7 +30,7 @@
 OK = 200
 
 
-class RendezvousHandler(SimpleHTTPServer.SimpleHTTPRequestHandler):
+class KVStoreHandler(SimpleHTTPServer.SimpleHTTPRequestHandler):
     # Set timeout
     timeout = SINGLE_REQUEST_TIMEOUT
 
@@ -39,7 +39,7 @@ def do_GET(self):
         paths = self.path.split('/')
         if len(paths) < 3:
             print(
-                'Rendezvous ERROR: Invalid request path: {path}.'.format(
+                'KVStore ERROR: Invalid request path: {path}.'.format(
                     path=self.path))
             self.send_status_code(BAD_REQUEST)
             return
@@ -61,7 +61,7 @@ def do_PUT(self):
         paths = self.path.split('/')
         if len(paths) < 3:
             print(
-                'Rendezvous ERROR: Invalid request path: {path}.'.format(
+                'KVStore ERROR: Invalid request path: {path}.'.format(
                     path=self.path))
             self.send_status_code(BAD_REQUEST)
             return
@@ -75,7 +75,7 @@ def do_PUT(self):
         except socket.timeout:
             if self.server.verbose:
                 print(
-                    'Rendezvous ERROR: Timeout when receiving {content_bytes} '
+                    'KVStore ERROR: Timeout when receiving {content_bytes} '
                     'bytes, aborting this incomplete request.' .format(
                         content_bytes=content_length))
 
@@ -91,6 +91,18 @@ def do_PUT(self):
 
         self.send_status_code(OK)
 
+    def send_status_code(self, status_code):
+        self.send_response(status_code)
+        self.send_header("Content-Length", 0)
+        self.end_headers()
+
+    # Override this function to prevent SimpleHTTPServer printing every
+    # request out.
+    def log_message(self, format, *args):
+        pass
+
+
+class RendezvousHandler(KVStoreHandler):
     # Override DELETE handler
     def do_DELETE(self):
         paths = self.path.split('/')
@@ -108,24 +120,14 @@ def do_DELETE(self):
 
         self.send_status_code(OK)
 
-    def send_status_code(self, status_code):
-        self.send_response(status_code)
-        self.send_header("Content-Length", 0)
-        self.end_headers()
-
-    # Override this function to prevent SimpleHTTPServer printing every
-    # request out.
-    def log_message(self, format, *args):
-        pass
-
 
 class RendezvousHTTPServer(BaseHTTPServer.HTTPServer, object):
     def __init__(self, addr, handler, verbose):
         # This class has to inherit from object since HTTPServer is an old-style
         # class that does not inherit from object.
         super(RendezvousHTTPServer, self).__init__(addr, handler)
 
-        # Lists for finished rendezvous workers
+        # Lists for finished http workers
         self.finished_list_lock = threading.Lock()
         self.finished_list = collections.defaultdict(list)
 
@@ -146,7 +148,7 @@ def extract_scope_size(self, host_alloc_plan):
             local_rank = slot_info.local_rank
             self.scope_size['cross_' + str(local_rank)] = slot_info.cross_size
 
-    # Decide whether all ranks have confirmed rendezvous completion.
+    # Decide whether all ranks have confirmed http completion.
     def should_continue(self):
         should_continue = False
         with self.finished_list_lock:
@@ -185,7 +187,7 @@ def start_server(self, host_alloc_plan):
                 addr, RendezvousHandler, self.verbose))
         self.httpd.extract_scope_size(host_alloc_plan)
         if self.verbose:
-            print('Rendezvous INFO: HTTP rendezvous server started.')
+            print('Rendezvous INFO: HTTP http server started.')
 
         # start the listening loop
         self.listen_thread = threading.Thread(target=self.listen_loop)
@@ -199,6 +201,53 @@ def listen_loop(self):
         while self.httpd.should_continue():
             self.httpd.handle_request()
 
+        self.httpd.server_close()
+
         if self.verbose:
             print('Rendezvous INFO: Rendezvous finishes.')
         # Because this thread is daemonized, no need to join.
+
+
+class KVStoreHTTPServer(BaseHTTPServer.HTTPServer, object):
+    def __init__(self, addr, handler, verbose):
+        super(KVStoreHTTPServer, self).__init__(addr, handler)
+
+        # Cache that provides the store
+        self.cache_lock = threading.Lock()
+        self.cache = {}
+
+        self.verbose = verbose
+
+
+class KVStoreServer:
+
+    def __init__(self, verbose):
+        self.httpd = None
+        self.listen_thread = None
+        self.verbose = verbose
+
+    # KVStore server finds a available port, create http socket,
+    # and start listening loop to handle request
+    def start_server(self):
+        self.httpd, port = find_port(
+            lambda addr: KVStoreHTTPServer(
+                addr, KVStoreHandler, self.verbose))
+
+        self.listen_thread = threading.Thread(
+            target=lambda: self.httpd.serve_forever())
+        self.listen_thread.daemon = True
+        self.listen_thread.start()
+
+        if self.verbose:
+            print('KVStoreServer INFO: KVStore server started. Listen on port ' + str(port))
+
+        return port
+
+    def shutdown_server(self):
+        self.httpd.shutdown()
+
+        self.httpd.server_close()
+
+        if self.verbose:
+            print('KVStoreServer INFO: KVStore server finishes.')
+        # Because this thread is daemonized, no need to join.

horovod/run/mpi_run.py

@@ -98,5 +98,12 @@ def mpi_run(settings, common_intfs, env):
 
     if settings.verbose >= 2:
         print(mpirun_command)
+
     # Execute the mpirun command.
-    os.execve('/bin/sh', ['/bin/sh', '-c', mpirun_command], env)
+    if settings.run_func_mode:
+        exit_code = safe_shell_exec.execute(mpirun_command, env=env)
+        if exit_code != 0:
+            raise RuntimeError("mpirun failed with exit code {exit_code}".format(exit_code=exit_code))
+    else:
+        os.execve('/bin/sh', ['/bin/sh', '-c', mpirun_command], env)
+