Add Intel(R) MPI support for horovodrun (horovod#2374)

Signed-off-by: Yana Shchyokotova <[email protected]>

Author: shirosankaku

.buildkite/gen-pipeline.sh

@@ -169,25 +169,25 @@ run_mpi_integration() {
       ":muscle: MPI Stall (${test})" \
       "bash -c \"${oneccl_env} \\\$(cat /mpirun_command) python /horovod/test/integration/test_stall.py\""
 
-    if [[ ${test} == *"openmpi"* ]]; then
+    if [[ ${test} == *"openmpi"* ]] || [[ ${test} == *"oneccl"* ]]; then
       run_test "${test}" "${queue}" \
         ":terminal: MPI Horovodrun (${test})" \
-        "horovodrun -np 2 -H localhost:2 python /horovod/examples/tensorflow/tensorflow_mnist.py"
+        "bash -c \"${oneccl_env} horovodrun -np 2 -H localhost:2 python /horovod/examples/tensorflow/tensorflow_mnist.py\""
       run_test "${test}" "${queue}" \
         ":terminal: MPI Horovodrun (${test})" \
-        "bash -c \"echo 'localhost slots=2' > hostfile && horovodrun -np 2 -hostfile hostfile python /horovod/examples/mxnet/mxnet_mnist.py\""
+        "bash -c \"${oneccl_env} echo 'localhost slots=2' > hostfile && horovodrun -np 2 -hostfile hostfile python /horovod/examples/mxnet/mxnet_mnist.py\""
     fi
   fi
 
   # TensorFlow 2.0 tests
   if [[ ${test} == *"tf2_"* ]] || [[ ${test} == *"tfhead"* ]]; then
     run_test "${test}" "${queue}" \
       ":tensorflow: MPI TensorFlow 2.0 MNIST (${test})" \
-      "bash -c \"\\\$(cat /mpirun_command) python /horovod/examples/tensorflow2/tensorflow2_mnist.py\""
+      "bash -c \"${oneccl_env} \\\$(cat /mpirun_command) python /horovod/examples/tensorflow2/tensorflow2_mnist.py\""
 
     run_test "${test}" "${queue}" \
       ":tensorflow: MPI TensorFlow 2.0 Keras MNIST (${test})" \
-      "bash -c \"\\\$(cat /mpirun_command) python /horovod/examples/tensorflow2/tensorflow2_keras_mnist.py\""
+      "bash -c \"${oneccl_env} \\\$(cat /mpirun_command) python /horovod/examples/tensorflow2/tensorflow2_keras_mnist.py\""
   fi
 }
 

CHANGELOG.md

@@ -8,7 +8,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
 ### Added
 
-### Changed
+- Added support for Intel(R) MPI in horovodrun. ([#2374](https://github.com/horovod/horovod/pull/2374))
+
+### Changed 
 
 ### Deprecated
 

Dockerfile.test.cpu

@@ -106,9 +106,9 @@ RUN if [[ ${MPI_KIND} == "OpenMPI" ]]; then \
             cp /mpirun_command_ofi /mpirun_command_mpi && \
             sed -i 's/export CCL_ATL_TRANSPORT=ofi;/export CCL_ATL_TRANSPORT=mpi;/g' /mpirun_command_mpi && \
             sed -i 's/export HOROVOD_CCL_CACHE=1;/export HOROVOD_CCL_CACHE=0;/g' /mpirun_command_mpi && \
+            echo "/mpirun_command_mpi" > /mpirun_command && \
             echo "-L/usr/local/oneccl/lib -lmpi -I/usr/local/oneccl/include" > /mpicc_oneccl && \
-            chmod +x /mpicc_oneccl && \
-            echo "/mpirun_command_mpi" > /mpirun_command; \
+            chmod +x /mpicc_oneccl; \            
     elif [[ ${MPI_KIND} == "MPICH" ]]; then \
         apt-get install -y mpich && \
             echo "mpirun -np 2" > /mpirun_command; \

docs/oneccl.rst

@@ -14,23 +14,13 @@ Source ``setvars.sh`` to start using oneCCL.
 
     source <install_dir>/env/setvars.sh
 
-2. Install the `Intel(R) MPI Library <https://software.intel.com/en-us/mpi-library>`_.
-
-To install the Intel MPI Library, follow `these instructions <https://software.intel.com/en-us/mpi-library/documentation/get-started>`_.
-
-Source ``mpivars.sh`` script to establish the proper environment settings.
-
-.. code-block:: bash
-        
-    source <installdir_MPI>/intel64/bin/mpivars.sh release_mt
-
-3. Set ``HOROVOD_CPU_OPERATIONS`` variable
+2. Set ``HOROVOD_CPU_OPERATIONS`` variable
 
 .. code-block:: bash
 
     export HOROVOD_CPU_OPERATIONS=CCL
 
-4. Install Horovod from source code
+3. Install Horovod from source code
 
 .. code-block:: bash
 

docs/running.rst

@@ -57,6 +57,7 @@ Usage of ``horovodrun`` requires one of the following:
 * MPICH
 * OpenRTE
 * Gloo
+* Intel(R) MPI
 
 If you do not have MPI installed, you can run ``horovodrun`` using Gloo.  Gloo dependencies come with Horovod
 automatically, and only require CMake to be available on your system at the time you install Horovod.
@@ -97,4 +98,10 @@ In some advanced cases you might want fine-grained control over options passed t
 To learn how to run Horovod training directly using Open MPI,
 read `Run Horovod with Open MPI <mpirun.rst>`_.
 
+Run Horovod with Intel(R) MPI
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+``horovodrun`` automatically converts some parameters to the format supported by Intel(R) MPI ``mpirun``. The set of allowed options includes ``-np``, ``-H`` and
+ssh arguments (-p, -i). Intel(R) MPI ``mpirun`` does not support MCA parameters, but you can set some of the options via `environment variables <https://software.intel.com/content/www/us/en/develop/documentation/mpi-developer-reference-linux/environment-variable-reference.html>`__. 
+For additional information refer to `Intel(R) MPI official documentation <https://software.intel.com/content/www/us/en/develop/documentation/mpi-developer-reference-linux/top/command-reference/mpiexec-hydra/global-options.html>`__.
+
 .. inclusion-marker-end-do-not-remove

horovod/runner/mpi_run.py

@@ -25,6 +25,7 @@
 _OMPI_IMPL = 'OpenMPI'
 _SMPI_IMPL = 'SpectrumMPI'
 _MPICH_IMPL = 'MPICH'
+_IMPI_IMPL = "IntelMPI"
 _UNKNOWN_IMPL = 'Unknown'
 _MISSING_IMPL = 'Missing'
 
@@ -35,6 +36,8 @@
 _SMPI_FLAGS_TCP = ['-tcp']
 # MPICH Flags
 _MPICH_FLAGS = []
+# Intel MPI Flags
+_IMPI_FLAGS = []
 
 # Threshold for large cluster MPI issues:
 _LARGE_CLUSTER_THRESHOLD = 64
@@ -70,6 +73,10 @@ def is_mpich(env=None):
     return _get_mpi_implementation(env) == _MPICH_IMPL
 
 
+def is_intel_mpi(env=None):
+    return _get_mpi_implementation(env) == _IMPI_IMPL
+
+
 def _get_mpi_implementation(env=None):
     """
     Detects the available MPI implementation by invoking `mpirun --version`.
@@ -78,7 +85,7 @@ def _get_mpi_implementation(env=None):
     represents the stdout and stderr as a string.
 
     Returns one of:
-    - _OMPI_IMPL, _SMPI_IMPL or _MPICH_IMPL for known implementations
+    - _OMPI_IMPL, _SMPI_IMPL, _MPICH_IMPL or _IMPI_IMPL for known implementations
     - _UNKNOWN_IMPL for any unknown implementation
     - _MISSING_IMPL if `mpirun --version` could not be executed.
 
@@ -98,6 +105,8 @@ def _get_mpi_implementation(env=None):
             return _SMPI_IMPL
         elif 'MPICH' in output:
             return _MPICH_IMPL
+        elif 'Intel(R) MPI' in output:
+            return _IMPI_IMPL
 
         print('Unknown MPI implementation given in output of mpirun --version:', file=sys.stderr)
         print(output, file=sys.stderr)
@@ -110,13 +119,15 @@ def _get_mpi_implementation(env=None):
 
 def _get_mpi_implementation_flags(tcp_flag, env=None):
     if is_open_mpi(env):
-        return list(_OMPI_FLAGS), list(_NO_BINDING_ARGS)
+        return list(_OMPI_FLAGS), list(_NO_BINDING_ARGS), _OMPI_IMPL
     elif is_spectrum_mpi(env):
-        return list(_SMPI_FLAGS) if not tcp_flag else list(_SMPI_FLAGS_TCP), list(_SOCKET_BINDING_ARGS)
+        return (list(_SMPI_FLAGS_TCP) if tcp_flag else list(_SMPI_FLAGS)), list(_SOCKET_BINDING_ARGS), _SMPI_IMPL
     elif is_mpich(env):
-        return list(_MPICH_FLAGS), list(_NO_BINDING_ARGS)
+        return list(_MPICH_FLAGS), list(_NO_BINDING_ARGS), _MPICH_IMPL
+    elif is_intel_mpi(env):
+        return list(_IMPI_FLAGS), [], _IMPI_IMPL
     else:
-        return None, None
+        return None, None, None
 
 
 def mpi_run(settings, nics, env, command, stdout=None, stderr=None):
@@ -138,10 +149,12 @@ def mpi_run(settings, nics, env, command, stdout=None, stderr=None):
         raise Exception('env argument must be a dict, not {type}: {env}'
                         .format(type=type(env), env=env))
 
-    mpi_impl_flags, impl_binding_args = _get_mpi_implementation_flags(settings.tcp_flag, env=env)
+    mpi_impl_flags, impl_binding_args, mpi = _get_mpi_implementation_flags(settings.tcp_flag, env=env)
     if mpi_impl_flags is None:
         raise Exception(_MPI_NOT_FOUND_ERROR_MSG)
 
+    impi = _IMPI_IMPL == mpi
+
     ssh_args = []
     if settings.ssh_port:
         ssh_args += [f'-p {settings.ssh_port}']
@@ -151,51 +164,71 @@ def mpi_run(settings, nics, env, command, stdout=None, stderr=None):
     mpi_ssh_args = ''
     if ssh_args:
         joined_ssh_args = ' '.join(ssh_args)
-        mpi_ssh_args = f'-mca plm_rsh_args \"{joined_ssh_args}\"'
-
-    # if user does not specify any hosts, mpirun by default uses local host.
-    # There is no need to specify localhost.
-    hosts_arg = '-H {hosts}'.format(hosts=settings.hosts)
+        mpi_ssh_args = f'-bootstrap=ssh -bootstrap-exec-args \"{joined_ssh_args}\"' if impi else f'-mca plm_rsh_args \"{joined_ssh_args}\"'
 
     tcp_intf_arg = '-mca btl_tcp_if_include {nics}'.format(
-        nics=','.join(nics)) if nics else ''
-    nccl_socket_intf_arg = '-x NCCL_SOCKET_IFNAME={nics}'.format(
+        nics=','.join(nics)) if nics and not impi else ''
+    nccl_socket_intf_arg = '-{opt} NCCL_SOCKET_IFNAME={nics}'.format(
+        opt='genv' if impi else 'x',
         nics=','.join(nics)) if nics else ''
 
     # On large cluster runs (e.g. Summit), we need extra settings to work around OpenMPI issues
-    host_names, _ = hosts.parse_hosts_and_slots(settings.hosts)
-    if host_names and len(host_names) >= _LARGE_CLUSTER_THRESHOLD:
+    host_names, host_to_slots = hosts.parse_hosts_and_slots(settings.hosts)
+    if not impi and host_names and len(host_names) >= _LARGE_CLUSTER_THRESHOLD:
         mpi_impl_flags.append('-mca plm_rsh_no_tree_spawn true')
         mpi_impl_flags.append('-mca plm_rsh_num_concurrent {}'.format(len(host_names)))
 
-    if settings.prefix_output_with_timestamp:
+    # if user does not specify any hosts, mpirun by default uses local host.
+    # There is no need to specify localhost.
+    hosts_arg = '-{opt} {hosts}'.format(opt='hosts' if impi else 'H',
+                hosts=','.join(host_names) if host_names and impi else settings.hosts)
+
+    ppn_arg = ' '
+    if host_to_slots and impi:
+        ppn = host_to_slots[host_names[0]]
+        for h_name in host_names[1:]:
+            if ppn != host_to_slots[h_name]:
+                raise Exception('''Different slots in -hosts parameter are not supported in Intel(R) MPI.
+                                 Use -machinefile <machine_file> for this purpose.''')
+        ppn_arg = ' -ppn {} '.format(ppn)
+
+    if settings.prefix_output_with_timestamp and not impi:
         mpi_impl_flags.append('--timestamp-output')
 
-    binding_args = settings.binding_args if settings.binding_args else ' '.join(impl_binding_args)
+    binding_args = settings.binding_args if settings.binding_args and not impi else ' '.join(impl_binding_args)
+
+    basic_args = '-l' if impi else '--allow-run-as-root --tag-output'
+
+    output = []
+    if settings.output_filename:
+        output.append('-outfile-pattern' if impi else '--output-filename')
+        output.append(settings.output_filename)
+
+    env_list = '' if impi else ' '.join(
+                    '-x %s' % key for key in sorted(env.keys()) if env_util.is_exportable(key))
 
     # Pass all the env variables to the mpirun command.
     mpirun_command = (
-        'mpirun --allow-run-as-root --tag-output '
-        '-np {num_proc} {hosts_arg} '
+        'mpirun {basic_args} '
+        '-np {num_proc}{ppn_arg}{hosts_arg} '
         '{binding_args} '
         '{mpi_args} '
         '{mpi_ssh_args} '
         '{tcp_intf_arg} '
         '{nccl_socket_intf_arg} '
         '{output_filename_arg} '
         '{env} {extra_mpi_args} {command}'  # expect a lot of environment variables
-        .format(num_proc=settings.num_proc,
+        .format(basic_args=basic_args,
+                num_proc=settings.num_proc,
+                ppn_arg=ppn_arg,
                 hosts_arg=hosts_arg,
                 binding_args=binding_args,
                 mpi_args=' '.join(mpi_impl_flags),
                 tcp_intf_arg=tcp_intf_arg,
                 nccl_socket_intf_arg=nccl_socket_intf_arg,
                 mpi_ssh_args=mpi_ssh_args,
-                output_filename_arg='--output-filename ' + settings.output_filename
-                                    if settings.output_filename else '',
-                env=' '.join('-x %s' % key for key in sorted(env.keys())
-                             if env_util.is_exportable(key)),
-
+                output_filename_arg=' '.join(output),
+                env=env_list,
                 extra_mpi_args=settings.extra_mpi_args if settings.extra_mpi_args else '',
                 command=' '.join(quote(par) for par in command))
     )

test/integration/test_spark.py

@@ -581,7 +581,7 @@ def fn():
             return 0
 
         def mpi_impl_flags(tcp, env=None):
-            return ["--mock-mpi-impl-flags"], ["--mock-mpi-binding-args"]
+            return ["--mock-mpi-impl-flags"], ["--mock-mpi-binding-args"], None
 
         def gloo_exec_command_fn(driver_addresses, key, settings, env):
             def _exec_command(command, alloc_info, event):
@@ -625,7 +625,7 @@ def fn():
             return 1
 
         def mpi_impl_flags(tcp, env=None):
-            return ["--mock-mpi-impl-flags"], ["--mock-mpi-binding-args"]
+            return ["--mock-mpi-impl-flags"], ["--mock-mpi-binding-args"], None
 
         def exception(*args, **argv):
             raise Exception('Test Exception')
@@ -647,7 +647,7 @@ def exception(*args, **argv):
                 self.assertEqual(str(e.value), 'Test Exception')
 
                 # call the mocked _get_mpi_implementation_flags method
-                mpi_flags, binding_args = horovod.runner.mpi_run._get_mpi_implementation_flags(False)
+                mpi_flags, binding_args, _ = horovod.runner.mpi_run._get_mpi_implementation_flags(False)
                 self.assertIsNotNone(mpi_flags)
                 expected_command = ('mpirun '
                                     '--allow-run-as-root --tag-output '

test/integration/test_static_run.py

@@ -29,7 +29,7 @@
 from horovod.runner.common.util import safe_shell_exec
 from horovod.runner import _HorovodArgs
 from horovod.runner.launch import _check_all_hosts_ssh_successful, _run
-from horovod.runner.mpi_run import mpi_available, is_mpich
+from horovod.runner.mpi_run import mpi_available, is_mpich, is_intel_mpi
 
 sys.path.append(os.path.join(os.path.dirname(__file__), os.pardir, 'utils'))
 
@@ -143,6 +143,8 @@ def test_run_success(self, controller, mode, run):
                 self.skipTest("MPI is not available")
             if is_mpich():
                 self.skipTest("MPICH is not testable")
+            if is_intel_mpi():
+                self.skipTest("Intel(R) MPI is not testable because it is based on MPICH")
 
         self.do_test_run_with_controller_success(controller, mode, run)
 
@@ -156,6 +158,8 @@ def test_run_failure(self, controller, mode, run):
                 self.skipTest("MPI is not available")
             if is_mpich():
                 self.skipTest("MPICH is not testable")
+            if is_intel_mpi():
+                self.skipTest("Intel(R) MPI is not testable because it is based on MPICH")
 
         self.do_test_run_with_controller_failure(controller, mode, run)