From 79deac5f82129a279f8095e61f6f7ab10e8eb654 Mon Sep 17 00:00:00 2001
From: rafael <rafael.sarmiento@cscs.ch>
Date: Thu, 4 Jun 2020 09:59:04 +0200
Subject: [PATCH 1/4] add ipcmagic check

---
 cscs-checks/apps/jupyter/check_ipcmagic.py    | 62 +++++++++++++++++++
 .../jupyter/src/tf-hvd-sgd-ipc-tf-1.14.py     | 42 +++++++++++++
 2 files changed, 104 insertions(+)
 create mode 100644 cscs-checks/apps/jupyter/check_ipcmagic.py
 create mode 100644 cscs-checks/apps/jupyter/src/tf-hvd-sgd-ipc-tf-1.14.py

diff --git a/cscs-checks/apps/jupyter/check_ipcmagic.py b/cscs-checks/apps/jupyter/check_ipcmagic.py
new file mode 100644
index 0000000000..118ec92c0e
--- /dev/null
+++ b/cscs-checks/apps/jupyter/check_ipcmagic.py
@@ -0,0 +1,62 @@
+import reframe as rfm
+import reframe.utility.sanity as sn
+from reframe.core.backends import getlauncher
+
+
+@rfm.simple_test
+class IPCMagicCheck(rfm.RunOnlyRegressionTest):
+    def __init__(self):
+        self.descr = 'Distributed training with TensorFlow using ipyparallel'
+        self.valid_systems = ['daint:gpu', 'dom:gpu']
+        self.valid_prog_environs = ['PrgEnv-gnu']
+        self.modules = ['ipcmagic/0.1-CrayGNU-19.10']
+        self.pre_run = [
+            'module unload dask',
+            'module load Horovod/0.16.4-CrayGNU-19.10-tf-1.14.0']
+        self.num_tasks = 2
+        self.num_tasks_per_node = 1
+        self.executable = 'ipython'
+        self.executable_opts = ['tf-hvd-sgd-ipc-tf-1.14.py']
+        nids = sn.extractall(r'nid(?P<nid>\d+)',
+                             self.stdout, 'nid', str)
+        self.sanity_patterns = sn.all([
+            sn.assert_ne(nids, []),
+            sn.assert_ne(nids[0], nids[1])
+        ])
+        self.reference = {
+            'daint:gpu': {
+                'slope': (2.0, -0.1, 0.1, ''),
+                'offset': (0.0, -0.1, 0.1, ''),
+                'retries': (0, None, None, ''),
+                'time': (10, None, None, 'seconds'),
+            },
+            'dom:gpu': {
+                'slope': (2.0, -0.1, 0.1, ''),
+                'offset': (0.0, -0.1, 0.1, ''),
+                'retries': (0, None, None, ''),
+                'time': (10, None, None, 'seconds'),
+            }
+        }
+        self.perf_patterns = {
+            'slope': sn.extractsingle(r'slope=(?P<slope>\S+)',
+                                      self.stdout, 'slope', float),
+            'offset': sn.extractsingle(r'offset=(?P<offset>\S+)',
+                                       self.stdout, 'offset', float),
+            'retries': self.retries(),
+            'time': sn.extractsingle(r'IPCluster is ready\!\s+'
+                                     r'\((?P<time>\d+) seconds\)',
+                                     self.stdout, 'time', float)
+        }
+        self.maintainers = ['RS', 'TR']
+        self.tags = {'production'}
+
+    @rfm.run_before('run')
+    def prepare_run(self):
+        # Change the job launcher since `ipython`
+        # needs to be emitted without `srun`.
+        self.job.launcher = getlauncher('local')()
+
+    @sn.sanity_function
+    def retries(self):
+        return 4 - sn.count(sn.findall(r'IPCluster is already running.',
+                                       self.stdout))
diff --git a/cscs-checks/apps/jupyter/src/tf-hvd-sgd-ipc-tf-1.14.py b/cscs-checks/apps/jupyter/src/tf-hvd-sgd-ipc-tf-1.14.py
new file mode 100644
index 0000000000..a1fc30f477
--- /dev/null
+++ b/cscs-checks/apps/jupyter/src/tf-hvd-sgd-ipc-tf-1.14.py
@@ -0,0 +1,42 @@
+import ipcmagic
+import ipyparallel as ipp
+
+
+get_ipython().run_line_magic('ipcluster', '--version')
+
+
+get_ipython().run_line_magic('ipcluster', 'start -n 2 --mpi')
+# Repeat a few of times in case of `TimeoutError`.
+# After the cluser starts, the following calls won't do nothing
+# but printing "IPCluster is already running".
+# This mimics what the user would do in such case.
+get_ipython().run_line_magic('ipcluster', 'start -n 2 --mpi')
+get_ipython().run_line_magic('ipcluster', 'start -n 2 --mpi')
+get_ipython().run_line_magic('ipcluster', 'start -n 2 --mpi')
+get_ipython().run_line_magic('ipcluster', 'start -n 2 --mpi')
+
+c = ipp.Client()
+
+print('cluster ids:', c.ids)
+
+get_ipython().run_cell_magic('px', '', 'import os\nprint(os.popen("ps -u $USER | grep ip").read())')
+
+get_ipython().run_cell_magic('px', '', 'import socket\nsocket.gethostname()')
+
+get_ipython().run_cell_magic('px', '', 'import numpy as np\nimport tensorflow as tf\nimport horovod.tensorflow as hvd')
+
+get_ipython().run_cell_magic('px', '', 'hvd.init()')
+
+get_ipython().run_cell_magic('px', '', '# Note that the generated rando data is different from one node to the other\nnsamples = 1000\nref_slope = 2.0\nref_offset = 0.0\nnoise = np.random.random((nsamples, 1)) - 0.5\nx_train = np.random.random((nsamples, 1)) - 0.5\ny_train = ref_slope * x_train + ref_offset + noise')
+
+get_ipython().run_cell_magic('px', '', '#input pipeline\ndataset = tf.data.Dataset.from_tensor_slices((x_train.astype(np.float32),\n                                              y_train.astype(np.float32)))\ndataset = dataset.shard(hvd.size(), hvd.rank())\ndataset = dataset.batch(500)\ndataset = dataset.repeat(500)\niterator = dataset.make_one_shot_iterator()\nnext_item = iterator.get_next()')
+
+get_ipython().run_cell_magic('px', '', '# Define the model\nslope = tf.Variable(np.random.randn())\noffset = tf.Variable(np.random.randn())\n\nx, y = next_item  # The model is the continuation of the pipeline\n\ny_hat = slope * x + offset\n\nloss = tf.losses.mean_squared_error(y_hat, y)\n\nopt = tf.train.GradientDescentOptimizer(.5)\ntrain = hvd.DistributedOptimizer(opt).minimize(loss)')
+
+get_ipython().run_cell_magic('px', '', 'hooks = [hvd.BroadcastGlobalVariablesHook(0)]')
+
+get_ipython().run_cell_magic('px', '', "history = []\n\nwith tf.train.MonitoredTrainingSession(hooks=hooks) as sess:\n    # Initialization of the variables `slope` and `offset`\n    # is done automatically by tf.train.MonitoredTrainingSession\n    print('rank', hvd.rank(),\n          'inital slope   = %12.6f\\n       initial offset = %12.6f' %\n          sess.run((slope, offset)))\n    while not sess.should_stop():\n        _, loss_val, m, n = sess.run((train, loss, slope, offset))\n        history.append([sess.run(slope), sess.run(offset), loss_val])")
+
+get_ipython().run_cell_magic('px', '', "print('slope=%f   offset=%f   loss=%f' % tuple(history[-1]))")
+
+get_ipython().run_line_magic('ipcluster', 'stop')

From bf0b81d3e5a310ef387655c763ba7843845588c5 Mon Sep 17 00:00:00 2001
From: rafael <rafael.sarmiento@cscs.ch>
Date: Thu, 4 Jun 2020 10:26:56 +0200
Subject: [PATCH 2/4] use default module version

---
 cscs-checks/apps/jupyter/check_ipcmagic.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/cscs-checks/apps/jupyter/check_ipcmagic.py b/cscs-checks/apps/jupyter/check_ipcmagic.py
index 118ec92c0e..c1259f7cf7 100644
--- a/cscs-checks/apps/jupyter/check_ipcmagic.py
+++ b/cscs-checks/apps/jupyter/check_ipcmagic.py
@@ -9,7 +9,7 @@ def __init__(self):
         self.descr = 'Distributed training with TensorFlow using ipyparallel'
         self.valid_systems = ['daint:gpu', 'dom:gpu']
         self.valid_prog_environs = ['PrgEnv-gnu']
-        self.modules = ['ipcmagic/0.1-CrayGNU-19.10']
+        self.modules = ['ipcmagic']
         self.pre_run = [
             'module unload dask',
             'module load Horovod/0.16.4-CrayGNU-19.10-tf-1.14.0']

From f66118e9229efcb258e04d82e6bb01e8f30f315f Mon Sep 17 00:00:00 2001
From: rafael <rafael.sarmiento@cscs.ch>
Date: Wed, 10 Jun 2020 08:42:58 +0200
Subject: [PATCH 3/4] fix comments

---
 cscs-checks/apps/jupyter/check_ipcmagic.py | 29 ++++++++++------------
 1 file changed, 13 insertions(+), 16 deletions(-)

diff --git a/cscs-checks/apps/jupyter/check_ipcmagic.py b/cscs-checks/apps/jupyter/check_ipcmagic.py
index c1259f7cf7..f70dfe75c5 100644
--- a/cscs-checks/apps/jupyter/check_ipcmagic.py
+++ b/cscs-checks/apps/jupyter/check_ipcmagic.py
@@ -12,7 +12,8 @@ def __init__(self):
         self.modules = ['ipcmagic']
         self.pre_run = [
             'module unload dask',
-            'module load Horovod/0.16.4-CrayGNU-19.10-tf-1.14.0']
+            'module load Horovod/0.16.4-CrayGNU-19.10-tf-1.14.0'
+        ]
         self.num_tasks = 2
         self.num_tasks_per_node = 1
         self.executable = 'ipython'
@@ -25,16 +26,16 @@ def __init__(self):
         ])
         self.reference = {
             'daint:gpu': {
-                'slope': (2.0, -0.1, 0.1, ''),
-                'offset': (0.0, -0.1, 0.1, ''),
-                'retries': (0, None, None, ''),
-                'time': (10, None, None, 'seconds'),
+                'slope': (2.0, -0.1, 0.1, None),
+                'offset': (0.0, -0.1, 0.1, None),
+                'retries': (0, None, None, None),
+                'time': (10, None, None, 's'),
             },
             'dom:gpu': {
-                'slope': (2.0, -0.1, 0.1, ''),
-                'offset': (0.0, -0.1, 0.1, ''),
-                'retries': (0, None, None, ''),
-                'time': (10, None, None, 'seconds'),
+                'slope': (2.0, -0.1, 0.1, None),
+                'offset': (0.0, -0.1, 0.1, None),
+                'retries': (0, None, None, None),
+                'time': (10, None, None, 's'),
             }
         }
         self.perf_patterns = {
@@ -42,7 +43,8 @@ def __init__(self):
                                       self.stdout, 'slope', float),
             'offset': sn.extractsingle(r'offset=(?P<offset>\S+)',
                                        self.stdout, 'offset', float),
-            'retries': self.retries(),
+            'retries': 4 - sn.count(sn.findall(r'IPCluster is already running',
+                                               self.stdout)),
             'time': sn.extractsingle(r'IPCluster is ready\!\s+'
                                      r'\((?P<time>\d+) seconds\)',
                                      self.stdout, 'time', float)
@@ -53,10 +55,5 @@ def __init__(self):
     @rfm.run_before('run')
     def prepare_run(self):
         # Change the job launcher since `ipython`
-        # needs to be emitted without `srun`.
+        # needs to be launched without `srun`.
         self.job.launcher = getlauncher('local')()
-
-    @sn.sanity_function
-    def retries(self):
-        return 4 - sn.count(sn.findall(r'IPCluster is already running.',
-                                       self.stdout))

From ac53d19d23678627c61d0364bf394213868bce5c Mon Sep 17 00:00:00 2001
From: rafael <rafael.sarmiento@cscs.ch>
Date: Wed, 10 Jun 2020 13:36:15 +0200
Subject: [PATCH 4/4] add license

---
 cscs-checks/apps/jupyter/check_ipcmagic.py             | 5 +++++
 cscs-checks/apps/jupyter/src/tf-hvd-sgd-ipc-tf-1.14.py | 5 +++++
 2 files changed, 10 insertions(+)

diff --git a/cscs-checks/apps/jupyter/check_ipcmagic.py b/cscs-checks/apps/jupyter/check_ipcmagic.py
index f70dfe75c5..822aab1d5c 100644
--- a/cscs-checks/apps/jupyter/check_ipcmagic.py
+++ b/cscs-checks/apps/jupyter/check_ipcmagic.py
@@ -1,3 +1,8 @@
+# Copyright 2016-2020 Swiss National Supercomputing Centre (CSCS/ETH Zurich)
+# ReFrame Project Developers. See the top-level LICENSE file for details.
+#
+# SPDX-License-Identifier: BSD-3-Clause
+
 import reframe as rfm
 import reframe.utility.sanity as sn
 from reframe.core.backends import getlauncher
diff --git a/cscs-checks/apps/jupyter/src/tf-hvd-sgd-ipc-tf-1.14.py b/cscs-checks/apps/jupyter/src/tf-hvd-sgd-ipc-tf-1.14.py
index a1fc30f477..96ead3178b 100644
--- a/cscs-checks/apps/jupyter/src/tf-hvd-sgd-ipc-tf-1.14.py
+++ b/cscs-checks/apps/jupyter/src/tf-hvd-sgd-ipc-tf-1.14.py
@@ -1,3 +1,8 @@
+# Copyright 2016-2020 Swiss National Supercomputing Centre (CSCS/ETH Zurich)
+# ReFrame Project Developers. See the top-level LICENSE file for details.
+#
+# SPDX-License-Identifier: BSD-3-Clause
+
 import ipcmagic
 import ipyparallel as ipp