src/yacsloader_swig/Test/testYacsPerfTest0.py

   1 #!/usr/bin/env python3
   2 # Copyright (C) 2023-2024  CEA, EDF
   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 unittest
  22 import tempfile
  23 import os
  24
  25 import pilot
  26 import SALOMERuntime
  27 import loader
  28 import salome
  29
  30 class TestYacsPerf0(unittest.TestCase):
  31     def test0(self):
  32         """
  33         [EDF28531] : Perf to check when number of threads in charge scheduler side is lower than number of // tasks
  34         """
  35         NB_OF_PARALLEL_NODES = 100
  36         NB_OF_PARALLEL_THREADS = 10
  37         salome.salome_init()
  38         SALOMERuntime.RuntimeSALOME.setRuntime()
  39         r=SALOMERuntime.getSALOMERuntime()
  40         p=r.createProc("PerfTest0")
  41         p.setProperty("executor","workloadmanager") # important line here to avoid that gg container treat several tasks in //.
  42         ti=p.createType("int","int")
  43         td=p.createType("double","double")
  44         tdd=p.createSequenceTc("seqdouble","seqdouble",td)
  45         tddd=p.createSequenceTc("seqseqdouble","seqseqdouble",tdd)
  46         tdddd=p.createSequenceTc("seqseqseqdouble","seqseqseqdouble",tddd)
  47         pyobj=p.createInterfaceTc("python:obj:1.0","pyobj",[])
  48         seqpyobj=p.createSequenceTc("list[pyobj]","list[pyobj]",pyobj)
  49         cont=p.createContainer("gg","Salome")
  50         cont.setProperty("nb_parallel_procs","1")
  51         cont.setAttachOnCloningStatus(True)
  52         cont.setProperty("attached_on_cloning","1")
  53         cont.setProperty("type","multi")
  54         cont.setProperty("container_name","gg")
  55         ######## Level 0
  56         startNode = r.createScriptNode("Salome","start")
  57         startNode.setExecutionMode("local")
  58         startNode.setScript("""o2 = list(range({}))""".format(NB_OF_PARALLEL_NODES))
  59         po2 = startNode.edAddOutputPort("o2",seqpyobj)
  60         p.edAddChild(startNode)
  61         #
  62         fe = r.createForEachLoopDyn("fe",pyobj)
  63         p.edAddChild(fe)
  64         p.edAddCFLink(startNode,fe)
  65         p.edAddLink(po2,fe.edGetSeqOfSamplesPort())
  66         internalNode = r.createScriptNode("Salome","internalNode")
  67         internalNode.setExecutionMode("remote")
  68         internalNode.setContainer(cont)
  69         internalNode.setScript("""
  70 ret = 3*ppp
  71 """)
  72         fe.edSetNode(internalNode)
  73         ix = internalNode.edAddInputPort("ppp",pyobj)
  74         oret = internalNode.edAddOutputPort("ret",pyobj)
  75         p.edAddLink( fe.edGetSamplePort(), ix )
  76         #
  77         endNode = r.createScriptNode("Salome","end")
  78         endNode.setExecutionMode("local")
  79         endNode.setContainer(None)
  80         ozeret = endNode.edAddOutputPort("ozeret",seqpyobj)
  81         izeret = endNode.edAddInputPort("izeret",seqpyobj)
  82         endNode.setScript("""ozeret = izeret""")
  83         p.edAddChild(endNode)
  84         p.edAddCFLink(fe,endNode)
  85         p.edAddLink( oret, izeret )
  86         if False:
  87             fname = "PerfTest0.xml"
  88             p.saveSchema(fname)
  89
  90             import loader
  91             l=loader.YACSLoader()
  92             p=l.load(fname)
  93         print("Start computation")
  94         import datetime
  95         st = datetime.datetime.now()
  96         ex=pilot.ExecutorSwig()
  97         ex.setMaxNbOfThreads(NB_OF_PARALLEL_THREADS)
  98         ex.RunW(p,0)
  99         self.assertEqual(p.getState(),pilot.DONE)
 100         salome.cm.ShutdownContainers()
 101         print("End of computation {}".format( str(datetime.datetime.now()-st) ) )
 102         if p.getChildByName("end").getOutputPort("ozeret").getPyObj() != [3*i for i in range(NB_OF_PARALLEL_NODES)]:
 103             raise RuntimeError("Ooops")
 104
 105     def test1(self):
 106         """
 107         [EDF28562] : test of exclusion output port of squeezeMemory mecanism
 108         """
 109         salome.salome_init()
 110         import KernelBasis
 111         KernelBasis.SetVerbosityActivated(False)
 112         SALOMERuntime.RuntimeSALOME.setRuntime()
 113         r=SALOMERuntime.getSALOMERuntime()
 114         p=r.createProc("Squeeze")
 115         pyobj=p.createInterfaceTc("python:obj:1.0","pyobj",[])
 116         cont=p.createContainer("gg","Salome")
 117         cont.setProperty("nb_parallel_procs","1")
 118         cont.setAttachOnCloningStatus(True)
 119         cont.setProperty("attached_on_cloning","1")
 120         cont.setProperty("type","multi")
 121         cont.setProperty("container_name","gilles")
 122         startNode = r.createScriptNode("Salome","startNode")
 123         startNode.setExecutionMode("remote")
 124         startNode.setContainer(cont)
 125         startNode.setSqueezeStatus(True)
 126         startNode.setScript("""
 127 o1,o2,o3,o4 = 31,32,33,34
 128 """)
 129         o1 = startNode.edAddOutputPort("o1",pyobj)
 130         o2 = startNode.edAddOutputPort("o2",pyobj)
 131         o3 = startNode.edAddOutputPort("o3",pyobj)
 132         o4 = startNode.edAddOutputPort("o4",pyobj)
 133         p.edAddChild(startNode)
 134         endNode = r.createScriptNode("Salome","endNode")
 135         endNode.setExecutionMode("remote")
 136         endNode.setContainer(cont)
 137         endNode.setSqueezeStatus(True)
 138         endNode.setScript("""
 139 o5,o6 = 45,46
 140 """)
 141         o5 = endNode.edAddOutputPort("o5",pyobj)
 142         o6 = endNode.edAddOutputPort("o6",pyobj)
 143         p.edAddChild(endNode)
 144         p.edAddCFLink(startNode,endNode)
 145         # disable proxy
 146         salome.cm.SetOverrideEnvForContainersSimple([("SALOME_BIG_OBJ_ON_DISK_THRES","-1")])
 147         # First part squeeze.
 148         ex=pilot.ExecutorSwig()
 149         ex.RunW(p,0)
 150         self.assertEqual(p.getState(),pilot.DONE)
 151         for outp in ["o1","o2","o3","o4"]:
 152             self.assertTrue( p.getChildByName("startNode").getOutputPort(outp).getPyObj() is None )
 153         for outp in ["o5","o6"]:
 154             self.assertTrue( p.getChildByName("endNode").getOutputPort(outp).getPyObj() is None )
 155         #### Now it s time
 156         #KernelBasis.SetVerbosityActivated(True)
 157         startNode.setSqueezeStatusWithExceptions(True,["o1","o3"])#<- The key point is here
 158         endNode.setSqueezeStatusWithExceptions(True,["o5"])#<- The key point is here
 159         ex=pilot.ExecutorSwig()
 160         ex.RunW(p,0)
 161         self.assertEqual(p.getState(),pilot.DONE)
 162         salome.cm.ShutdownContainers()
 163         for outp in ["o2","o4"]:
 164             self.assertTrue( p.getChildByName("startNode").getOutputPort(outp).getPyObj() is None )
 165         for outp in ["o6"]:
 166             self.assertTrue( p.getChildByName("endNode").getOutputPort(outp).getPyObj() is None )
 167         #
 168         for outp,valExp in [("o1",31),("o3",33)]:
 169             self.assertEqual( p.getChildByName("startNode").getOutputPort(outp).getPyObj(), valExp )
 170         for outp,valExp in [("o5",45)]:
 171             self.assertEqual( p.getChildByName("endNode").getOutputPort(outp).getPyObj(), valExp )
 172
 173
 174 if __name__ == '__main__':
 175   with tempfile.TemporaryDirectory() as dir_test:
 176     file_test = os.path.join(dir_test,"UnitTestsResult")
 177     with open(file_test, 'a') as f:
 178         f.write("  --- TEST src/yacsloader: testYacsPerfTest0.py\n")
 179         suite = unittest.makeSuite(TestYacsPerf0)
 180         result=unittest.TextTestRunner(f, descriptions=1, verbosity=1).run(suite)
 181         if not result.wasSuccessful():
 182            raise RuntimeError("Test failed !")