[NRRPLT-5385] Adjust MPI launch API to simulation assembly interface

Change-Id: I7b142d863c95d9597b290f3ab27c85637e56af2d

hbp_nrp_music_interface/hbp_nrp_music_interface/cle/MUSICBrainLoader.py

@@ -24,7 +24,7 @@
 """`
 Load a brain network's MUSIC proxies
 """
-from hbp_nrp_cle.brainsim.pynn import simulator as sim
+import pyNN.nest as sim
 import hbp_nrp_cle.tf_framework.config as tf_config
 
 from hbp_nrp_music_xml.pynn.factory import PyNNProxyFactory

hbp_nrp_music_interface/hbp_nrp_music_interface/cle/MUSICPyNNCommunicationAdapter.py

@@ -27,7 +27,7 @@ instead of direct population access. Maxmimum code reuse and minimal duplication
 where possible.
 """
 from hbp_nrp_cle.brainsim.pynn_nest.PyNNNestCommunicationAdapter import PyNNNestCommunicationAdapter
-from hbp_nrp_cle.brainsim.pynn import simulator as sim
+import pyNN.nest as sim
 import hbp_nrp_cle.tf_framework.config as tf_config
 
 from hbp_nrp_music_interface.bibi.bibi_music_config import MUSICConfiguration

hbp_nrp_music_interface/hbp_nrp_music_interface/launch/MUSICBrainProcess.py

@@ -25,7 +25,7 @@
 A distributed brain process that can be launched standalone on remote hosts.
 """
 from hbp_nrp_cle.brainsim import config
-from hbp_nrp_cle.brainsim.pynn import simulator as sim
+import pyNN.nest as sim
 from hbp_nrp_cle.brainsim.pynn.PyNNControlAdapter import PyNNControlAdapter
 from hbp_nrp_cle.cle.ClosedLoopEngine import ClosedLoopEngine
 import hbp_nrp_cle.tf_framework.config as tf_config
@@ -80,7 +80,7 @@ class MUSICBrainProcess(object):
         config.rng_seed = int(rng_seed)
 
         # spawn CLE components that will handle loading the brain file and interfaces
-        self._brain_controller = PyNNControlAdapter()
+        self._brain_controller = PyNNControlAdapter(sim)
         self._brain_controller.initialize()
         self._brain_controller.load_brain(brain_file, **pop_dict)
 

hbp_nrp_music_interface/hbp_nrp_music_interface/launch/MUSICCLEProcess.py

+# ---LICENSE-BEGIN - DO NOT CHANGE OR MOVE THIS HEADER
+# This file is part of the Neurorobotics Platform software
+# Copyright (C) 2014,2015,2016,2017 Human Brain Project
+# https://www.humanbrainproject.eu
+#
+# The Human Brain Project is a European Commission funded project
+# in the frame of the Horizon2020 FET Flagship plan.
+# http://ec.europa.eu/programmes/horizon2020/en/h2020-section/fet-flagships
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either version 2
+# of the License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+# ---LICENSE-END
+"""
+This module contains the CLE process logic for the simulation assembly using MUSIC
+"""
+
+import os
+import argparse
+import logging
+from hbp_nrp_cleserver.server.CLEGazeboSimulationAssembly import CLEGazeboSimulationAssembly
+from hbp_nrp_music_interface.cle.MUSICPyNNCommunicationAdapter import MUSICPyNNCommunicationAdapter
+from hbp_nrp_music_interface.cle.MUSICPyNNControlAdapter import MUSICPyNNControlAdapter
+
+import pyNN.nest as nestsim
+import music
+
+logger = logging.getLogger(__name__)
+
+
+class MusicCLESimulationAssembly(CLEGazeboSimulationAssembly):
+    """
+    Defines the assembly of a simulation using MUSIC
+    """
+
+    def __init__(self, sim_id, exc, bibi_model, **par):
+        """
+        Creates a new simulation assembly to simulate an experiment using the CLE and Gazebo
+        :param sim_id: The simulation id
+        :param exc: The experiment configuration
+        :param bibi_model: The BIBI configuration
+        """
+        super(MusicCLESimulationAssembly, self).__init__(sim_id, exc, bibi_model, **par)
+
+    def _create_brain_adapters(self):
+        """
+        Creates the adapter components for the neural simulator
+
+        :return: A tuple of the communication and control adapter for the neural simulator
+        """
+        logger.info('Using MUSIC configuration and adapters for CLE')
+
+        # write pid to lock file so launcher can always terminate us
+        # (failsafe)
+        pid = os.getpid()
+        with open('{}.lock'.format(pid), 'w') as pf:
+            pf.write('{}'.format(pid))
+
+        # initialize music and set the CLE to use MUSIC adapters
+        music.Setup()
+
+        braincomm = MUSICPyNNCommunicationAdapter()
+        braincontrol = MUSICPyNNControlAdapter(nestsim)
+        return braincomm, braincontrol
+
+if __name__ == '__main__':  # pragma: no cover
+    # TODO: This should be separated into its own method such that we can unit
+    # test this code
+    simulation = None
+
+    # exit code, 0 for success and -1 for any failures
+    mpi_returncode = 0
+    try:
+        if os.environ["ROS_MASTER_URI"] == "":
+            raise Exception("You should run ROS first.")
+
+        parser = argparse.ArgumentParser()
+        parser.add_argument('--exdconf', dest='exd_file',
+                            help='specify the ExDConfiguration file', required=True)
+        parser.add_argument('--environment', dest='environment_file',
+                            help='specify the environment file', required=True)
+        parser.add_argument('--experiment-path', dest='path',
+                            help='specify the base experiment path', required=True)
+        parser.add_argument('--gzserver-host', dest='gzserver_host',
+                            help='the gzserver target host', required=True)
+        parser.add_argument('--reservation', dest='reservation', default=None,
+                            help='cluster resource reservation', required=False)
+        parser.add_argument('--sim-id', dest='sim_id', type=int,
+                            help='the simulation id to use', required=True)
+        parser.add_argument('--timeout', dest='timeout',
+                            help='the simulation default time allocated', required=True)
+        parser.add_argument('--rng-seed', dest='rng_seed',
+                            help='the global experiment RNG seed', required=True)
+        parser.add_argument('-v', '--verbose', action='store_true',
+                            help='increase output verbosity')
+
+        args = parser.parse_args()
+
+        # expand any parameters (e.g. NRP_EXPERIMENTS_DIRECTORY) in paths
+        args.exd_file = os.path.expandvars(args.exd_file)
+        args.environment_file = os.path.expandvars(args.environment_file)
+        args.path = os.path.expandvars(args.path)
+
+        # simplified launch process below from ROSCLESimulationFactory.py, avoid circular depdency
+        # by importing here
+        import rospy
+        from hbp_nrp_cleserver.server import ROS_CLE_NODE_NAME
+        from hbp_nrp_cleserver.server.ROSCLESimulationFactory import set_up_logger
+        from hbp_nrp_cleserver.server.ROSCLESimulationFactory import get_experiment_data
+
+        # reconfigure the logger to stdout as done in ROSCLESimulationFactory.py otherwise all
+        # output will be trapped by the ROS logger after the first ROS node is
+        # initialized
+        rospy.init_node(ROS_CLE_NODE_NAME, anonymous=True)
+        set_up_logger(None, args.verbose)
+
+        exd, bibi = get_experiment_data(args.exd_file)
+
+        # parse the timeout string command line argument into a valid datetime
+        import dateutil.parser as datetime_parser
+        timeout_parsed = datetime_parser.parse(args.timeout.replace('_', ' '))
+
+        # check the reservation argument, if empty default to None
+        if args.reservation == '':
+            args.reservation = None
+
+        # override the experiment RNG seed with the command line value
+        exd.rngSeed = int(args.rng_seed)
+
+        simulation = MusicCLESimulationAssembly(args.sim_id,
+                                                exd,
+                                                bibi,
+                                                gzserver_host=args.gzserver_host,
+                                                reservation=args.reservation,
+                                                timeout=timeout_parsed)
+        simulation.initialize(args.environment_file, None)
+        if simulation.cle_server is None:
+            raise Exception(
+                "Error in cle_function_init. Cannot start simulation.")
+
+        # FIXME: This should be done more cleanly within the adapter, see [NRRPLT-4858]
+        # the tag is a magic number to avoid circular build/release dependency for now but
+        # this will be removed when the referenced bug is fixed
+        # notify MPI/music processes that configuration is complete
+        from mpi4py import MPI
+        for rank in xrange(MPI.COMM_WORLD.Get_size()):
+            if rank != MPI.COMM_WORLD.Get_rank():
+                MPI.COMM_WORLD.send('ready', dest=rank, tag=100)
+        MPI.COMM_WORLD.Barrier()
+
+        logger.info('Starting CLE.')
+        simulation.run()  # This is a blocking call, not to be confused with
+        # threading.Thread.start
+
+    except Exception as e:  # pylint: disable=broad-except
+
+        # if running through MPI, catch Exception and terminate below to ensure brain processes
+        # are also killed
+        logger.error(
+            'CLE aborted with message {}, terminating.'.format(e.message))
+        # if no logger
+        print 'CLE aborted with message {}, terminating.'.format(e.message)
+        logger.exception(e)
+        mpi_returncode = -1
+
+    finally:
+
+        # always attempt to shutdown the CLE launcher and release resources
+        if simulation:
+            logger.info('Shutting down CLE.')
+            simulation.shutdown()
+            logger.info('Shutdown complete, terminating.')
+
+    # terminate the MUSIC spawned brain processes
+    # send a shutdown message in case the brain processes are in a recv loop at startup since they
+    # seem to block and ignore the Abort command until receiving a message
+    from mpi4py import MPI
+    for rank in xrange(MPI.COMM_WORLD.Get_size()):
+        MPI.COMM_WORLD.isend('shutdown', dest=rank, tag=100)
+    MPI.COMM_WORLD.Abort(mpi_returncode)

hbp_nrp_music_interface/hbp_nrp_music_interface/launch/MUSICLauncher.py

@@ -35,23 +35,21 @@ import random
 import sys
 
 
+# This class intentionally does not inherit SimulationServer (even though it is an implementation of
+# it) in order to avoid duplicate notificators
 class MUSICLauncher(object):
     """
     Setup, build, and launch a distributed MUSIC instance that will spawn the CLE and
     requested brain processes.
     """
 
-    # pylint: disable=too-many-arguments
-    def __init__(self, exd_file, bibi_file, env_file, exp_path, server_host, reservation,
-                 sim_id, timeout):
+    def __init__(self, sim_id, exc, bibi, **par):
         """
         Store all experiment configuration parameters so that they can be propagated
         to the remote hosts.
 
-        :param exd_file Absolute path to the ExDConf file.
-        :param bibi_file Absolute path to the BIBI file.
-        :param env_file Absolute path to the simulation environment file.
-        :param exp_path Absolute path to the base dir of the experiment files.
+        :param exc: the experiment configuration
+        :param bibi: the BIBI configuration.
         :param server_host Target Gazebo/brain process host (e.g. local or lugano)
         :param reservation Reservation string for cluster backend (None is a valid option)
         :param sim_id The id of the simulation/experiment to be launched.
@@ -60,40 +58,44 @@ class MUSICLauncher(object):
 
         # we need to replace absolute paths with relative environment variable-based paths for the
         # remote hosts that may not share the same file structure
-        nrp_models_path = os.environ.get('NRP_MODELS_DIRECTORY').rstrip('/')
         nrp_experiments_path = os.environ.get('NRP_EXPERIMENTS_DIRECTORY').rstrip('/')
 
-        self._exd_file = exd_file.replace(nrp_experiments_path, '$NRP_EXPERIMENTS_DIRECTORY')
-        self._bibi_file = bibi_file.replace(nrp_experiments_path, '$NRP_EXPERIMENTS_DIRECTORY')
-        self._env_file = env_file.replace(nrp_models_path, '$NRP_MODELS_DIRECTORY')
-        self._exp_path = exp_path.replace(nrp_experiments_path, '$NRP_EXPERIMENTS_DIRECTORY')
+        self._exd_file = exc.path.replace(nrp_experiments_path, '$NRP_EXPERIMENTS_DIRECTORY')
+        self._bibi_file = bibi.path.replace(nrp_experiments_path, '$NRP_EXPERIMENTS_DIRECTORY')
+        self._exp_path = exc.dir.replace(nrp_experiments_path, '$NRP_EXPERIMENTS_DIRECTORY')
 
         # store the other launch parameters as provided
-        self._server_host = server_host
-        self._reservation = reservation
+        self._server_host = par['gzserver_host']
+        self._reservation = par['reservation']
         self._sim_id = sim_id
-        self._timeout = timeout
+        self._timeout = par['timeout']
+
+        self.__exc = exc
+        self.__bibi = bibi
+        self._env_file = None
 
         # host specific launch configuration/allocation
         self._launcher = None
-
         # the MPI process launcher for the CLE and brain processes
-        self.cle_server = None
+        self.mpilauncher = None
 
-    def init(self, bibi, exd):
+    # we should call the except_hook when something goes wrong in the simulation, but currently
+    # we don't
+    # pylint: disable=unused-argument
+    def initialize(self, environment_file, except_hook):
         """
         Construct the MUSIC launch configuration that will spawn CLE + brain processes
         on distributed hosts.
-
-        :param bibi The BIBI configuration to launch brain processes with.
-        :param exd The parsed ExDConf configuration to launch brain processes with.
         """
 
+        nrp_models_path = os.environ.get('NRP_MODELS_DIRECTORY').rstrip('/')
+        self._env_file = environment_file.replace(nrp_models_path, '$NRP_MODELS_DIRECTORY')
+
         # create a host specific launcher
         if self._server_host == 'local':
             self._launcher = LocalLauncher()
         elif self._server_host == 'lugano':
-            self._launcher = LuganoLauncher(exd.bibiConf.processes + 1,
+            self._launcher = LuganoLauncher(self.__exc.bibiConf.processes + 1,
                                             self._timeout,
                                             self._reservation)
         else:
@@ -109,11 +111,12 @@ class MUSICLauncher(object):
         timeout_str = str(self._timeout).replace(' ', '_')
 
         # extract the multiprocess RNG seed to use or generate one if needed
-        rng_seed = exd.rngSeed if exd.rngSeed is not None else random.randint(1, sys.maxint)
+        rng_seed = self.__exc.rngSeed if self.__exc.rngSeed is not None else \
+            random.randint(1, sys.maxint)
 
         # build a MUSIC configuration script with correct brain ports, launchers and arugments
         # save it to the host launcher temp directory, this is the same for every host
-        music_conf = bibi_music_config.MUSICConfiguration(bibi)
+        music_conf = bibi_music_config.MUSICConfiguration(self.__bibi)
         music_conf.add_application('CLE',
                                    cle_launcher,
                                    ['--exdconf={}'.format(self._exd_file),
@@ -123,14 +126,13 @@ class MUSICLauncher(object):
                                     '--reservation={}'.format(reservation_str),
                                     '--sim-id={}'.format(self._sim_id),
                                     '--timeout={}'.format(timeout_str),
-                                    '--rng-seed={}'.format(rng_seed),
-                                    '--music'],
+                                    '--rng-seed={}'.format(rng_seed)],
                                    1)
         music_conf.add_application('BRAIN',
                                    brain_launcher,
                                    ['--bibi-file={}'.format(self._bibi_file),
                                     '--rng-seed={}'.format(rng_seed)],
-                                   exd.bibiConf.processes)
+                                   self.__exc.bibiConf.processes)
         music_conf.save(self._launcher.local_tmpdir)
 
         # deploy the generated configuration files / launch scripts to the target host
@@ -138,18 +140,24 @@ class MUSICLauncher(object):
 
         # build an mpi launch command for the requested host configuration, currently we launch
         # all processes on the same host
-        self.cle_server = MUSICMPILauncher()
-        self.cle_server.add_host(self._launcher.hostname,
-                                 self._launcher.host_tmpdir,
-                                 exd.bibiConf.processes + 1)
+        self.mpilauncher = MUSICMPILauncher()
+        self.mpilauncher.add_host(self._launcher.hostname,
+                                  self._launcher.host_tmpdir,
+                                  self.__exc.bibiConf.processes + 1)
 
         # construct the mpi command line with the above host/launch information
-        self.cle_server.build()
+        self.mpilauncher.build()
 
         # for error propagation reasons, we have to launch and init the MPI processes to emulate
         # the behavior of the single process launcher, if the mpirun command fails or the CLE/brain
         # processes fail then the error will be properly propagated
-        self.cle_server.launch()
+        self.mpilauncher.launch()
+
+    def run(self):
+        """
+        Runs the assembled simulation
+        """
+        self.mpilauncher.run()
 
     def shutdown(self):
         """
@@ -157,9 +165,9 @@ class MUSICLauncher(object):
         """
 
         # terminate the mpirun command (if it is still running)
-        if self.cle_server:
-            self.cle_server.shutdown()
-        self.cle_server = None
+        if self.mpilauncher is not None:
+            self.mpilauncher.shutdown()
+        self.mpilauncher = None
 
         # perform any launcher host specific cleanup
         if self._launcher:

hbp_nrp_music_interface/hbp_nrp_music_interface/launch/host/LocalLauncher.py

@@ -67,7 +67,7 @@ class LocalLauncher(IHostLauncher):
         """
         # create .sh launcher scripts for the CLE and brain processes
         cle_launcher = self._create_launch_script('music_cle_launcher.sh',
-                                                  'hbp_nrp_cleserver.server.CLELauncher')
+                                                  'hbp_nrp_music_interface.launch.MUSICCLEProcess')
         brain_launcher = self._create_launch_script('music_brain_launcher.sh',
                                                     'hbp_nrp_music_interface.launch.' +
                                                     'MUSICBrainProcess')

hbp_nrp_music_interface/hbp_nrp_music_interface/tests/cle/test_brain_loader.py

@@ -25,7 +25,7 @@ import unittest
 
 from hbp_nrp_music_interface.cle.MUSICBrainLoader import load_proxies_from_xml
 import hbp_nrp_cle.tf_framework.config as tf_config
-from hbp_nrp_cle.brainsim.pynn import simulator as sim
+import pyNN.nest as sim
 
 from mock import Mock, patch
 import os
@@ -53,7 +53,7 @@ class TestBrainLoader(unittest.TestCase):
         self.assertRaises(Exception, load_proxies_from_xml, 'foo.xml', bar='foo')
 
     @patch('__builtin__.open', spec=open)
-    @patch('hbp_nrp_cle.brainsim.pynn.simulator.PopulationView', new=Mock(return_value='fake'))
+    @patch('pyNN.nest.PopulationView', new=Mock(return_value='fake'))
     @patch('hbp_nrp_music_xml.pynn.xml_factory.XmlFactory.create_proxies',
            new=Mock(return_value={'bar_to_brain': 'foo', 'bar_to_cle': 'bar'}))
     def test_proxy_dict_valid(self, mock_open):

hbp_nrp_music_interface/hbp_nrp_music_interface/tests/cle/test_communication_adapter.py

@@ -50,8 +50,8 @@ class MockPopulationView(object):
 @patch('hbp_nrp_cle.brainsim.pynn_nest.PyNNNestCommunicationAdapter.PyNNNestCommunicationAdapter.initialize')
 @patch('hbp_nrp_cle.brainsim.pynn_nest.PyNNNestCommunicationAdapter.PyNNNestCommunicationAdapter.register_spike_source')
 @patch('hbp_nrp_cle.brainsim.pynn_nest.PyNNNestCommunicationAdapter.PyNNNestCommunicationAdapter.register_spike_sink')
-@patch('hbp_nrp_cle.brainsim.pynn.simulator.PopulationView', MockPopulationView)
-@patch('hbp_nrp_cle.brainsim.pynn.simulator.Population', MockPopulation)
+@patch('pyNN.nest.PopulationView', MockPopulationView)
+@patch('pyNN.nest.Population', MockPopulation)
 class TestCommunicationAdapter(unittest.TestCase):
 
       def setUp(self):

hbp_nrp_music_interface/hbp_nrp_music_interface/tests/cle/test_control_adapter.py

@@ -39,7 +39,7 @@ class MockedMUSICSetup(object):
 class TestControlAdapter(unittest.TestCase):
 
       def test_missing_xml(self):
-          mc = MUSICPyNNControlAdapter()
+          mc = MUSICPyNNControlAdapter(Mock())
           self.assertRaises(Exception, mc.load_brain, 'foo.xml')
 
       @patch('music.Setup', name=MockedMUSICSetup())
@@ -51,7 +51,7 @@ class TestControlAdapter(unittest.TestCase):
           m.return_value.read.return_value = 'mock brain source'
 
           with patch('__builtin__.open', m, create=True):
-              mc = MUSICPyNNControlAdapter()
+              mc = MUSICPyNNControlAdapter(Mock())
               mc.load_brain('foo.xml')
 
           mocked_environ.assert_called_once_with('NRP_MUSIC_DIRECTORY')

hbp_nrp_music_interface/hbp_nrp_music_interface/tests/launch/test_music_launcher.py

@@ -51,29 +51,34 @@ class TestMUSICLauncher(unittest.TestCase):
 
       @patch('os.environ.get', return_value='')
       def setUp(self, mock_environ_get):
-          self.__launcher = MUSICLauncher('exd_conf', 'bibi_file', 'env_file', '/exp_path',
-                                          'local', None, 1, '1234 T 567')
+          exc_mock = Mock()
+          exc_mock.path = 'exd_conf'
+          exc_mock.dir = '/exp_path'
+          exc_mock.rngSeed = 123456
+          exc_mock.bibiConf.processes = 10
+          bibi_mock = Mock()
+          bibi_mock.path = 'bibi_file'
+          self.__launcher = MUSICLauncher(1, exc_mock, bibi_mock,
+                                          gzserver_host='local',
+                                          reservation=None,
+                                          timeout=None)
 
       @patch('hbp_nrp_music_interface.launch.MUSICLauncher.LocalLauncher', MockLocalLauncher)
       @patch('hbp_nrp_music_interface.launch.MUSICLauncher.bibi_music_config.MUSICConfiguration')
       @patch('hbp_nrp_music_interface.launch.MUSICLauncher.MUSICMPILauncher')
       def test_init(self, mock_mpi, mock_conf):
-
-          # mock the ExDConf and parameters
-          exd = Mock()
-          exd.rngSeed = 123456
-          exd.bibiConf.processes = 10
-
           # mock all of the local launcher functionality
-          self.__launcher.init('bibi', exd)
+          with patch('hbp_nrp_music_interface.launch.MUSICLauncher.os.environ.get',
+                     return_value='/somewhere/over/the/rainbow'):
+            self.__launcher.initialize('env_file', None)
 
           # call to generate launch scripts
           self.__launcher._launcher.deploy.assert_called_once()
 
           # assert that all of the MPI configuration/launch commands were called
-          self.__launcher.cle_server.add_host.assert_called_once_with('localhost', '/mock_host_tmpdir', 11)
-          self.__launcher.cle_server.build.assert_called_once()
-          self.__launcher.cle_server.launch.assert_called_once()
+          self.__launcher.mpilauncher.add_host.assert_called_once_with('localhost', '/mock_host_tmpdir', 11)
+          self.__launcher.mpilauncher.build.assert_called_once()
+          self.__launcher.mpilauncher.launch.assert_called_once()
 
       @patch('os.system')
       def test_shutdown(self, system_mock):
@@ -85,7 +90,7 @@ class TestMUSICLauncher(unittest.TestCase):
           mock_launcher.shutdown = Mock()
 
           # set values like in init() and shutdown
-          self.__launcher.cle_server = mock_cle_server
+          self.__launcher.mpilauncher = mock_cle_server
           self.__launcher._launcher = mock_launcher
           self.__launcher.shutdown()
 
@@ -94,7 +99,7 @@ class TestMUSICLauncher(unittest.TestCase):
           mock_launcher.shutdown.assert_called_once()
 
           # verify the local variables are unset
-          self.assertEqual(self.__launcher.cle_server, None)
+          self.assertEqual(self.__launcher.mpilauncher, None)
           self.assertEqual(self.__launcher._launcher, None)
 
           # verify the ros cleanup command has been called