src/yacsloader_swig/Test/testWorkloadManager.py

   1 #!/usr/bin/env python3
   2 # Copyright (C) 2006-2022  CEA/DEN, EDF R&D
   3 #
   4 # This library is free software; you can redistribute it and/or
   5 # modify it under the terms of the GNU Lesser General Public
   6 # License as published by the Free Software Foundation; either
   7 # version 2.1 of the License, or (at your option) any later version.
   8 #
   9 # This library is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12 # Lesser General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Lesser General Public
  15 # License along with this library; if not, write to the Free Software
  16 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
  17 #
  18 # See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
  19 #
  20
  21 import sys
  22 import pilot
  23 import SALOMERuntime
  24 import loader
  25 import unittest
  26 import tempfile
  27 import os
  28 import salome
  29
  30 NB_NODE=15
  31 class TestEdit(unittest.TestCase):
  32
  33     def setUp(self):
  34         SALOMERuntime.RuntimeSALOME_setRuntime()
  35         self.r = pilot.getRuntime()
  36         self.l = loader.YACSLoader()
  37         self.e = pilot.ExecutorSwig()
  38         # We need a catalog which contains only one resource named "localhost"
  39         # with NB_NODE cores. The modifications made here are not saved to the
  40         # catalog file.
  41         salome.salome_init()
  42         resourceManager = salome.lcc.getResourcesManager()
  43         resource_definition = resourceManager.GetResourceDefinition("localhost")
  44         resource_definition.nb_node = NB_NODE
  45         resource_definition.nb_proc_per_node = 1
  46         resourceManager.AddResource(resource_definition, False, "")
  47         resource_required = salome.ResourceParameters()
  48         resource_required.can_run_containers = True
  49         res_list = resourceManager.GetFittingResources(resource_required)
  50         for r in res_list:
  51           if r != "localhost":
  52             resourceManager.RemoveResource(r, False, "")
  53         resource_definition = resourceManager.GetResourceDefinition("localhost")
  54         self.assertEqual(resource_definition.nb_node, NB_NODE)
  55         self.assertEqual(resource_definition.nb_proc_per_node, 1)
  56
  57     def tearDown(self):
  58         cm = salome.lcc.getContainerManager()
  59         cm.ShutdownContainers()
  60
  61     def test1(self):
  62         """ Two parallel foreach-s with different containers
  63         """
  64         proc = self.l.load("samples/wlm_2foreach.xml")
  65         self.e.RunW(proc,0)
  66         self.assertEqual(proc.getState(),pilot.DONE)
  67         res_port = proc.getChildByName("End").getOutputPort("r")
  68         # theoretical time should be 15s
  69         execution_time = res_port.getPyObj()
  70         # lower time means some resources are overloaded
  71         msg = "Execution time is too short : {}s".format(execution_time)
  72         self.assertTrue(execution_time > 13, msg)
  73         # The containers may need some time to be launched.
  74         # We need some delay to add to the 15s.
  75         msg = "Execution time is too long : {}s".format(execution_time)
  76         self.assertTrue(execution_time < 20, msg)
  77
  78     def test2(self):
  79         """ Two parallel foreach-s with different containers and python nodes
  80             using cache.
  81         """
  82         proc = self.l.load("samples/wlm_2foreach_with_cache.xml")
  83         self.e.RunW(proc,0)
  84         self.assertEqual(proc.getState(),pilot.DONE)
  85         ok = proc.getChildByName("End").getOutputPort("ok")
  86         self.assertTrue(ok)
  87         total_time = proc.getChildByName("End").getOutputPort("total_time")
  88         # theoretical time should be 16s
  89         execution_time = total_time.getPyObj()
  90         # lower time means some resources are overloaded
  91         msg = "Execution time is too short : {}s".format(execution_time)
  92         self.assertTrue(execution_time > 14, msg)
  93         # The containers may need some time to be launched.
  94         # We need some delay to add to the 16s.
  95         msg = "Execution time is too long : {}s".format(execution_time)
  96         self.assertTrue(execution_time < 20, msg)
  97         coeff_cont = proc.getChildByName("End").getOutputPort("coeff_cont").getPyObj()
  98         msg = "coeff_cont too low:"+str(coeff_cont)
  99         self.assertTrue(coeff_cont >= NB_NODE, msg)
 100         msg = "coeff_cont too high:"+str(coeff_cont)
 101         self.assertTrue(coeff_cont <= 2*NB_NODE, msg)
 102
 103     def test3(self):
 104         """ Launch 8 independent nodes in parallel.
 105         """
 106         proc = self.l.load("samples/wlm_8nodes.xml")
 107         self.e.RunW(proc,0)
 108         self.assertEqual(proc.getState(),pilot.DONE)
 109         ok = proc.getChildByName("End").getOutputPort("ok")
 110         if not ok :
 111           err_message = proc.getChildByName("End").getOutputPort("err_message").getPyObj()
 112           self.fail(err_message)
 113
 114     def test4(self):
 115         """ Verify the execution is stoped if no resource can run a task.
 116         """
 117         proc = self.l.load("samples/wlm_error.xml")
 118         self.e.RunW(proc,0)
 119         self.assertEqual(proc.getState(),pilot.FAILED)
 120         self.assertEqual(proc.getChildByName("ErrorNode").getState(),pilot.ERROR)
 121
 122     def test5(self):
 123         """ Foreach with 1000 points and several nodes in the block.
 124         """
 125         proc = self.l.load("samples/wlm_complex_foreach.xml")
 126         self.e.RunW(proc,0)
 127         self.assertEqual(proc.getState(),pilot.DONE)
 128
 129 if __name__ == '__main__':
 130   dir_test = tempfile.mkdtemp(suffix=".yacstest")
 131   file_test = os.path.join(dir_test,"UnitTestsResult")
 132   with open(file_test, 'a') as f:
 133       f.write("  --- TEST src/yacsloader: testWorkloadManager.py\n")
 134       suite = unittest.makeSuite(TestEdit)
 135       result=unittest.TextTestRunner(f, descriptions=1, verbosity=1).run(suite)
 136   sys.exit(not result.wasSuccessful())