src/Container/SALOME_PyNode.py

   1 #  -*- coding: iso-8859-1 -*-
   2 # Copyright (C) 2007-2023  CEA, EDF, OPEN CASCADE
   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 #  File   : SALOME_PyNode.py
  22 #  Author : Christian CAREMOLI, EDF
  23 #  Module : SALOME
  24 #  $Header$
  25 #
  26 import sys,traceback
  27 import linecache
  28 import pickle
  29 import Engines__POA
  30 import SALOME__POA
  31 import SALOME
  32 import logging
  33 import os
  34 import sys
  35 from SALOME_ContainerHelper import ScriptExecInfo
  36
  37 MY_CONTAINER_ENTRY_IN_GLBS = "my_container"
  38
  39 MY_PERFORMANCE_LOG_ENTRY_IN_GLBS = "my_log_4_this_session"
  40
  41 class Generic(SALOME__POA.GenericObj):
  42   """A Python implementation of the GenericObj CORBA IDL"""
  43   def __init__(self,poa):
  44     self.poa=poa
  45     self.cnt=1
  46
  47   def Register(self):
  48     #print("Register called : %d"%self.cnt)
  49     self.cnt+=1
  50
  51   def UnRegister(self):
  52     #print("UnRegister called : %d"%self.cnt)
  53     self.cnt-=1
  54     if self.cnt <= 0:
  55       oid=self.poa.servant_to_id(self)
  56       self.poa.deactivate_object(oid)
  57
  58   def Destroy(self):
  59     print("WARNING SALOME::GenericObj::Destroy() function is obsolete! Use UnRegister() instead.")
  60     self.UnRegister()
  61
  62   def __del__(self):
  63     #print("Destuctor called")
  64     pass
  65
  66 class PyNode_i (Engines__POA.PyNode,Generic):
  67   """The implementation of the PyNode CORBA IDL"""
  68   def __init__(self, nodeName,code,poa,my_container):
  69     """Initialize the node : compilation in the local context"""
  70     Generic.__init__(self,poa)
  71     self.nodeName=nodeName
  72     self.code=code
  73     self.my_container=my_container._container
  74     linecache.cache[nodeName]=0,None,code.split('\n'),nodeName
  75     ccode=compile(code,nodeName,'exec')
  76     self.context={}
  77     self.context[MY_CONTAINER_ENTRY_IN_GLBS] = self.my_container
  78     exec(ccode, self.context)
  79
  80   def getContainer(self):
  81     return self.my_container
  82
  83   def getCode(self):
  84     return self.code
  85
  86   def getName(self):
  87     return self.nodeName
  88
  89   def defineNewCustomVar(self,varName,valueOfVar):
  90     self.context[varName] = pickle.loads(valueOfVar)
  91     pass
  92
  93   def executeAnotherPieceOfCode(self,code):
  94     """Called for initialization of container lodging self."""
  95     try:
  96       ccode=compile(code,self.nodeName,'exec')
  97       exec(ccode, self.context)
  98     except Exception:
  99       raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"","PyScriptNode (%s) : code to be executed \"%s\"" %(self.nodeName,code),0))
 100
 101   def execute(self,funcName,argsin):
 102     """Execute the function funcName found in local context with pickled args (argsin)"""
 103     try:
 104       argsin,kws=pickle.loads(argsin)
 105       func=self.context[funcName]
 106       argsout=func(*argsin,**kws)
 107       argsout=pickle.dumps(argsout,-1)
 108       return argsout
 109     except Exception:
 110       exc_typ,exc_val,exc_fr=sys.exc_info()
 111       l=traceback.format_exception(exc_typ,exc_val,exc_fr)
 112       raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"".join(l),"PyNode: %s, function: %s" % (self.nodeName,funcName),0))
 113
 114 class SenderByte_i(SALOME__POA.SenderByte,Generic):
 115   def __init__(self,poa,bytesToSend):
 116     Generic.__init__(self,poa)
 117     self.bytesToSend = bytesToSend
 118
 119   def getSize(self):
 120     return len(self.bytesToSend)
 121
 122   def sendPart(self,n1,n2):
 123     return self.bytesToSend[n1:n2]
 124
 125 SALOME_FILE_BIG_OBJ_DIR = "SALOME_FILE_BIG_OBJ_DIR"
 126
 127 SALOME_BIG_OBJ_ON_DISK_THRES_VAR = "SALOME_BIG_OBJ_ON_DISK_THRES"
 128
 129 # default is 50 MB
 130 SALOME_BIG_OBJ_ON_DISK_THRES_DFT = 50000000
 131
 132 DicoForProxyFile = { }
 133
 134 def GetSizeOfBufferedReader(f):
 135   """
 136   This method returns in bytes size of a file openned.
 137
 138   Args:
 139   ----
 140       f (io.IOBase): buffered reader returned by open
 141
 142   Returns
 143   -------
 144       int: number of bytes
 145   """
 146   import io
 147   pos = f.tell()
 148   f.seek(0,io.SEEK_END)
 149   pos2 = f.tell()
 150   f.seek(pos,io.SEEK_SET)
 151   return pos2-pos
 152
 153 def GetObjectFromFile(fname, visitor = None):
 154   with open(fname,"rb") as f:
 155     if visitor:
 156       visitor.setHDDMem( GetSizeOfBufferedReader(f) )
 157       visitor.setFileName( fname )
 158     obj = pickle.load(f)
 159   return obj
 160
 161 def DumpInFile(obj,fname):
 162   with open(fname,"wb") as f:
 163     f.write( obj )
 164
 165 def IncrRefInFile(fname):
 166   if fname in DicoForProxyFile:
 167     DicoForProxyFile[fname] += 1
 168   else:
 169     DicoForProxyFile[fname] = 2
 170   pass
 171
 172 def DecrRefInFile(fname):
 173   if fname not in DicoForProxyFile:
 174     cnt = 1
 175   else:
 176     cnt = DicoForProxyFile[fname]
 177     DicoForProxyFile[fname] -= 1
 178     if cnt == 1:
 179       del DicoForProxyFile[fname]
 180   if cnt == 1:
 181     if os.path.exists(fname):
 182       os.unlink( fname )
 183   pass
 184
 185 def GetBigObjectOnDiskThreshold():
 186   import os
 187   if SALOME_BIG_OBJ_ON_DISK_THRES_VAR in os.environ:
 188     return int( os.environ[SALOME_BIG_OBJ_ON_DISK_THRES_VAR] )
 189   else:
 190     return SALOME_BIG_OBJ_ON_DISK_THRES_DFT
 191
 192 def ActivateProxyMecanismOrNot( sizeInByte ):
 193   thres = GetBigObjectOnDiskThreshold()
 194   if thres == -1:
 195     return False
 196   else:
 197     return sizeInByte > thres
 198
 199 def GetBigObjectDirectory():
 200   import os
 201   if SALOME_FILE_BIG_OBJ_DIR not in os.environ:
 202     raise RuntimeError("An object of size higher than limit detected and no directory specified to dump it in file !")
 203   return os.path.expanduser( os.path.expandvars( os.environ[SALOME_FILE_BIG_OBJ_DIR] ) )
 204
 205 def GetBigObjectFileName():
 206   """
 207   Return a filename in the most secure manner (see tempfile documentation)
 208   """
 209   import tempfile
 210   with tempfile.NamedTemporaryFile(dir=GetBigObjectDirectory(),prefix="mem_",suffix=".pckl") as f:
 211     ret = f.name
 212   return ret
 213
 214 class BigObjectOnDiskBase:
 215   def __init__(self, fileName, objSerialized):
 216     """
 217     :param fileName: the file used to dump into.
 218     :param objSerialized: the object in pickeled form
 219     :type objSerialized: bytes
 220     """
 221     self._filename = fileName
 222     # attribute _destroy is here to tell client side or server side
 223     # only client side can be with _destroy set to True. server side due to risk of concurrency
 224     # so pickled form of self must be done with this attribute set to False.
 225     self._destroy = False
 226     self.__dumpIntoFile(objSerialized)
 227
 228   def getDestroyStatus(self):
 229     return self._destroy
 230
 231   def incrRef(self):
 232     if self._destroy:
 233       IncrRefInFile( self._filename )
 234     else:
 235       # should never happen !
 236       RuntimeError("Invalid call to incrRef !")
 237
 238   def decrRef(self):
 239     if self._destroy:
 240       DecrRefInFile( self._filename )
 241     else:
 242       # should never happen !
 243       RuntimeError("Invalid call to decrRef !")
 244
 245   def unlinkOnDestructor(self):
 246     self._destroy = True
 247
 248   def doNotTouchFile(self):
 249     """
 250     Method called slave side. The life cycle management of file is client side not slave side.
 251     """
 252     self._destroy = False
 253
 254   def __del__(self):
 255     if self._destroy:
 256       DecrRefInFile( self._filename )
 257
 258   def getFileName(self):
 259     return self._filename
 260
 261   def __dumpIntoFile(self, objSerialized):
 262     DumpInFile( objSerialized, self._filename )
 263
 264   def get(self, visitor = None):
 265     obj = GetObjectFromFile( self._filename, visitor )
 266     return obj
 267
 268   def __float__(self):
 269     return float( self.get() )
 270
 271   def __int__(self):
 272     return int( self.get() )
 273
 274   def __str__(self):
 275     obj = self.get()
 276     if isinstance(obj,str):
 277         return obj
 278     else:
 279         raise RuntimeError("Not a string")
 280
 281 class BigObjectOnDisk(BigObjectOnDiskBase):
 282   def __init__(self, fileName, objSerialized):
 283     BigObjectOnDiskBase.__init__(self, fileName, objSerialized)
 284
 285 class BigObjectOnDiskListElement(BigObjectOnDiskBase):
 286   def __init__(self, pos, length, fileName):
 287     self._filename = fileName
 288     self._destroy = False
 289     self._pos = pos
 290     self._length = length
 291
 292   def get(self, visitor = None):
 293     fullObj = BigObjectOnDiskBase.get(self, visitor)
 294     return fullObj[ self._pos ]
 295
 296   def __getitem__(self, i):
 297     return self.get()[i]
 298
 299   def __len__(self):
 300     return len(self.get())
 301
 302 class BigObjectOnDiskSequence(BigObjectOnDiskBase):
 303   def __init__(self, length, fileName, objSerialized):
 304     BigObjectOnDiskBase.__init__(self, fileName, objSerialized)
 305     self._length = length
 306
 307   def __getitem__(self, i):
 308     return BigObjectOnDiskListElement(i, self._length, self.getFileName())
 309
 310   def __len__(self):
 311     return self._length
 312
 313 class BigObjectOnDiskList(BigObjectOnDiskSequence):
 314   def __init__(self, length, fileName, objSerialized):
 315     BigObjectOnDiskSequence.__init__(self, length, fileName, objSerialized)
 316
 317 class BigObjectOnDiskTuple(BigObjectOnDiskSequence):
 318   def __init__(self, length, fileName, objSerialized):
 319     BigObjectOnDiskSequence.__init__(self, length, fileName, objSerialized)
 320
 321 def ProxyfyPickeled( obj, pickleObjInit = None, visitor = None ):
 322   """
 323   This method return a proxy instance of pickled form of object given in input.
 324
 325   Args:
 326   ----
 327       obj (pickelable type) : object to be proxified
 328       pickleObjInit (bytes) : Optionnal. Original pickeled form of object to be proxyfied if already computed. If not this method generate it
 329
 330   Returns
 331   -------
 332       BigObjectOnDiskBase: proxy instance
 333   """
 334   pickleObj = pickleObjInit
 335   if pickleObj is None:
 336     pickleObj = pickle.dumps( obj , pickle.HIGHEST_PROTOCOL )
 337   fileName = GetBigObjectFileName()
 338   if visitor:
 339     visitor.setHDDMem( len(pickleObj) )
 340     visitor.setFileName(fileName)
 341   if isinstance( obj, list):
 342     proxyObj = BigObjectOnDiskList( len(obj), fileName, pickleObj )
 343   elif isinstance( obj, tuple):
 344     proxyObj = BigObjectOnDiskTuple( len(obj), fileName , pickleObj )
 345   else:
 346     proxyObj = BigObjectOnDisk( fileName , pickleObj )
 347   return proxyObj
 348
 349 def SpoolPickleObject( obj, visitor = None ):
 350   import pickle
 351   with InOutputObjVisitorCM(visitor) as v:
 352     pickleObjInit = pickle.dumps( obj , pickle.HIGHEST_PROTOCOL )
 353     if not ActivateProxyMecanismOrNot( len(pickleObjInit) ):
 354       return pickleObjInit
 355     else:
 356       proxyObj = ProxyfyPickeled( obj, pickleObjInit, v.visitor() )
 357       pickleProxy = pickle.dumps( proxyObj , pickle.HIGHEST_PROTOCOL )
 358       return pickleProxy
 359
 360 from SALOME_ContainerHelper import InOutputObjVisitorCM, InOutputObjVisitor
 361
 362 def UnProxyObjectSimple( obj, visitor = None ):
 363   """
 364   Method to be called in Remote mode. Alterate the obj _status attribute.
 365   Because the slave process does not participate in the reference counting
 366
 367   Args:
 368   ----
 369       visitor (InOutputObjVisitor): A visitor to keep track of amount of memory on chip and those on HDD
 370
 371   """
 372   with InOutputObjVisitorCM(visitor) as v:
 373     logging.debug( "UnProxyObjectSimple {}".format(type(obj)) )
 374     if isinstance(obj,BigObjectOnDiskBase):
 375       obj.doNotTouchFile()
 376       return obj.get( v )
 377     elif isinstance( obj, list):
 378       retObj = []
 379       for elt in obj:
 380         retObj.append( UnProxyObjectSimple(elt,v.visitor()) )
 381       return retObj
 382     else:
 383       return obj
 384
 385 def UnProxyObjectSimpleLocal( obj ):
 386   """
 387   Method to be called in Local mode. Do not alterate the PyObj counter
 388   """
 389   if isinstance(obj,BigObjectOnDiskBase):
 390     return obj.get()
 391   elif isinstance( obj, list):
 392     retObj = []
 393     for elt in obj:
 394       retObj.append( UnProxyObjectSimpleLocal(elt) )
 395     return retObj
 396   else:
 397     return obj
 398
 399 class FileHolder:
 400   def __init__(self, fileName):
 401     self._filename = fileName
 402   @property
 403   def filename(self):
 404     return self._filename
 405
 406 class FileDeleter(FileHolder):
 407   def __init__(self, fileName):
 408     super().__init__( fileName )
 409   def __del__(self):
 410     import os
 411     if os.path.exists( self._filename ):
 412       os.unlink( self._filename )
 413
 414 class MonitoringInfo:
 415   def __init__(self, pyFileName, outFileName, pid):
 416     self._py_file_name = pyFileName
 417     self._out_file_name = outFileName
 418     self._pid = pid
 419
 420   @property
 421   def pyFileName(self):
 422     return self._py_file_name
 423
 424   @property
 425   def pid(self):
 426     return self._pid
 427
 428   @pid.setter
 429   def pid(self, value):
 430     self._pid = value
 431
 432   @property
 433   def outFileName(self):
 434     return self._out_file_name
 435
 436 def FileSystemMonitoring(intervalInMs, dirNameToInspect, outFileName = None):
 437     """
 438     This method loops indefinitely every intervalInMs milliseconds to scan
 439     number of inodes and size of content recursively included into the in input directory.
 440
 441     Args:
 442     ----
 443
 444     outFileName (str) : name of file inside the results will be written. If None a new file is generated
 445
 446     See also CPUMemoryMonitoring
 447     """
 448     global orb
 449     import os
 450     dirNameToInspect2 = os.path.abspath( os.path.expanduser(dirNameToInspect) )
 451     import tempfile
 452     import logging
 453     import KernelBasis
 454     # outFileNameSave stores the content of outFileName during phase of dumping
 455     with tempfile.NamedTemporaryFile(prefix="fs_monitor_",suffix=".txt") as f:
 456       outFileNameSave = f.name
 457     with tempfile.NamedTemporaryFile(prefix="fs_monitor_",suffix=".py") as f:
 458       tempPyFile = f.name
 459     tempOutFile = outFileName
 460     if tempOutFile is None:
 461       tempOutFile = "{}.txt".format( os.path.splitext( tempPyFile )[0] )
 462     with open(tempPyFile,"w") as f:
 463         f.write("""
 464 import subprocess as sp
 465 import re
 466 import os
 467 import time
 468 import datetime
 469 with open("{tempOutFile}","a") as f:
 470   f.write( "{{}}\\n".format( "{dirNameToInspect2}" ) )
 471   while(True):
 472     nbinodes = sp.check_output("{{}} | wc -l".format( " ".join(["find","{dirNameToInspect2}"]),  ), shell = True).decode().strip()
 473     szOfDirStr = re.split("[\s]+",sp.check_output(["du","-sh","{dirNameToInspect2}"]).decode())[0]
 474     f.write( "{{}}\\n".format( str( datetime.datetime.now().timestamp() ) ) )
 475     f.write( "{{}}\\n".format( str( nbinodes  ) ) )
 476     f.write( "{{}}\\n".format( str( szOfDirStr ) ) )
 477     f.flush()
 478     time.sleep( {intervalInMs} / 1000.0 )
 479 """.format( **locals()))
 480     logging.debug( "File for FS monitoring dump file : {}".format(tempPyFile) )
 481     pyFileName = FileDeleter( tempPyFile )
 482     if outFileName is None:
 483       outFileName = FileDeleter( tempOutFile )
 484     else:
 485       outFileName = FileHolder(outFileName)
 486     return MonitoringInfo(pyFileName,outFileName,None)
 487
 488 def CPUMemoryMonitoring( intervalInMs, outFileName = None ):
 489   """
 490   Launch a subprocess monitoring self process.
 491   This monitoring subprocess is a python process lauching every intervalInMs ms evaluation of
 492   CPU usage and RSS memory of the calling process.
 493   Communication between subprocess and self is done by file.
 494
 495   Args:
 496   ----
 497     outFileName (str) : name of file inside the results will be written. If None a new file is generated
 498
 499   See also FileSystemMonitoring
 500   """
 501   import KernelBasis
 502   def BuildPythonFileForCPUPercent( intervalInMs, outFileName):
 503     import os
 504     import tempfile
 505     with tempfile.NamedTemporaryFile(prefix="cpu_mem_monitor_",suffix=".py") as f:
 506       tempPyFile = f.name
 507     tempOutFile = outFileName
 508     if tempOutFile is None:
 509       tempOutFile = "{}.txt".format( os.path.splitext( tempPyFile )[0] )
 510     pid = os.getpid()
 511     with open(tempPyFile,"w") as f:
 512       f.write("""import psutil
 513 pid = {}
 514 process = psutil.Process( pid )
 515 import time
 516 with open("{}","a") as f:
 517   while True:
 518     f.write( "{{}}\\n".format( str( process.cpu_percent() ) ) )
 519     f.write( "{{}}\\n".format( str( process.memory_info().rss  ) ) )
 520     f.flush()
 521     time.sleep( {} / 1000.0 )
 522 """.format(pid, tempOutFile, intervalInMs))
 523     if outFileName is None:
 524       autoOutFile = FileDeleter(tempOutFile)
 525     else:
 526       autoOutFile = FileHolder(tempOutFile)
 527     return FileDeleter(tempPyFile),autoOutFile
 528   pyFileName, outFileName = BuildPythonFileForCPUPercent( intervalInMs, outFileName )
 529   return MonitoringInfo(pyFileName, outFileName, None)
 530
 531 class GenericPythonMonitoringLauncherCtxMgr:
 532     def __init__(self, monitoringParams):
 533         """
 534         Args:
 535         ----
 536             monitoringParams (MonitoringInfo)
 537         """
 538         self._monitoring_params = monitoringParams
 539     def __enter__(self):
 540         import KernelBasis
 541         pid = KernelBasis.LaunchMonitoring(self._monitoring_params.pyFileName.filename)
 542         self._monitoring_params.pid = pid
 543         return self._monitoring_params
 544
 545     def __exit__(self,exctype, exc, tb):
 546         StopMonitoring( self._monitoring_params )
 547
 548 def StopMonitoring( monitoringInfo ):
 549   """
 550   Kill monitoring subprocess.
 551
 552   Args:
 553   ----
 554       monitoringInfo (MonitoringInfo): info returned by LaunchMonitoring
 555   """
 556   import KernelBasis
 557   KernelBasis.StopMonitoring(monitoringInfo.pid)
 558
 559 def ReadCPUMemInfoInternal( fileName ):
 560   import KernelBasis
 561   ret = KernelBasis.ReadFloatsInFile( fileName )
 562   cpu = ret[::2]
 563   mem_rss = [ int(elt) for elt in ret[1::2]]
 564   return [(a,b) for a,b in zip(cpu,mem_rss)]
 565
 566 def ReadCPUMemInfo( monitoringInfo ):
 567   """
 568   Retrieve CPU/Mem data of monitoring.
 569
 570   Args:
 571   ----
 572       monitoringInfo (MonitoringInfo): info returned by LaunchMonitoring
 573
 574   Returns
 575   -------
 576     list<float,str> : list of pairs. First param of pair is CPU usage. Second param of pair is rss memory usage
 577   """
 578   return ReadCPUMemInfoInternal( monitoringInfo.outFileName.filename )
 579
 580 class InodeSizeInfo:
 581   def __init__(self, dirNameMonitored, timeStamps, nbInodes, volumeOfDir):
 582     """
 583     Args:
 584     ----
 585     timeStamps (list<datetimestruct>)
 586     nbInodes (list<int>)
 587     volumeOfDir (list<str>)
 588     """
 589     self._dir_name_monitored = dirNameMonitored
 590     self._data = [(t,a,b) for t,a,b in zip(timeStamps,nbInodes,volumeOfDir)]
 591   def __str__(self):
 592     st = """Filename monitored : {self.dirNameMonitored}
 593 Data : ${self.data}
 594 """.format( **locals() )
 595     return st
 596   @property
 597   def dirNameMonitored(self):
 598     return self._dir_name_monitored
 599   @property
 600   def data(self):
 601     """
 602     list of triplets. First param of triplet is datetimestruct
 603                                       Second param of triplet is #inodes.
 604                                       Thirst param of triplet is size.
 605     """
 606     return self._data
 607
 608 def ReadInodeSizeInfoInternal( fileName ):
 609   import datetime
 610   import os
 611   with open(fileName, "r") as f:
 612     coarseData = [ elt.strip() for elt in f.readlines() ]
 613   dirNameMonitored = coarseData[0] ; coarseData = coarseData[1:]
 614   tss = [ datetime.datetime.fromtimestamp( float(elt) ) for elt in coarseData[::3] ]
 615   nbInodes = [int(elt) for elt in coarseData[1::3]]
 616   volumeOfDir = coarseData[2::3]
 617   return InodeSizeInfo(dirNameMonitored,tss,nbInodes,volumeOfDir)
 618
 619 def ReadInodeSizeInfo( monitoringInfo ):
 620   """
 621   Retrieve nb of inodes and size of monitoring
 622
 623   Args:
 624   ----
 625       monitoringInfo (MonitoringInfo): info returned by LaunchMonitoring
 626
 627   Returns
 628   -------
 629     InodeSizeInfo
 630   """
 631   return ReadInodeSizeInfoInternal( monitoringInfo.outFileName.filename )
 632
 633 class SeqByteReceiver:
 634   # 2GB limit to trigger split into chunks
 635   CHUNK_SIZE = 2000000000
 636   def __init__(self,sender):
 637     self._obj = sender
 638   def __del__(self):
 639     self._obj.UnRegister()
 640     pass
 641   def data(self):
 642     size = self._obj.getSize()
 643     if size <= SeqByteReceiver.CHUNK_SIZE:
 644       return self.fetchOneShot( size )
 645     else:
 646       return self.fetchByChunks( size )
 647   def fetchOneShot(self,size):
 648     return self._obj.sendPart(0,size)
 649   def fetchByChunks(self,size):
 650       """
 651       To avoid memory peak parts over 2GB are sent using EFF_CHUNK_SIZE size.
 652       """
 653       data_for_split_case = bytes(0)
 654       EFF_CHUNK_SIZE = SeqByteReceiver.CHUNK_SIZE // 8
 655       iStart = 0 ; iEnd = EFF_CHUNK_SIZE
 656       while iStart!=iEnd and iEnd <= size:
 657         part = self._obj.sendPart(iStart,iEnd)
 658         data_for_split_case = bytes(0).join( [data_for_split_case,part] )
 659         iStart = iEnd; iEnd = min(iStart + EFF_CHUNK_SIZE,size)
 660       return data_for_split_case
 661
 662 class LogOfCurrentExecutionSession:
 663   def __init__(self, handleToCentralizedInst):
 664     self._remote_handle = handleToCentralizedInst
 665     self._current_instance = ScriptExecInfo()
 666
 667   def addFreestyleAndFlush(self, value):
 668     self._current_instance.freestyle = value
 669     self.finalizeAndPushToMaster()
 670
 671   def addInfoOnLevel2(self, key, value):
 672     setattr(self._current_instance,key,value)
 673
 674   def finalizeAndPushToMaster(self):
 675     self._remote_handle.assign( pickle.dumps( self._current_instance ) )
 676
 677 class PyScriptNode_i (Engines__POA.PyScriptNode,Generic):
 678   """The implementation of the PyScriptNode CORBA IDL that executes a script"""
 679   def __init__(self, nodeName,code,poa,my_container,logscript):
 680     """Initialize the node : compilation in the local context"""
 681     Generic.__init__(self,poa)
 682     self.nodeName=nodeName
 683     self.code=code
 684     self.my_container_py = my_container
 685     self.my_container=my_container._container
 686     linecache.cache[nodeName]=0,None,code.split('\n'),nodeName
 687     self.ccode=compile(code,nodeName,'exec')
 688     self.context={}
 689     self.context[MY_CONTAINER_ENTRY_IN_GLBS] = self.my_container
 690     self._log_script = logscript
 691     self._current_execution_session = None
 692     sys.stdout.flush() ; sys.stderr.flush() # flush to correctly capture log per execution session
 693
 694   def __del__(self):
 695     # force removal of self.context. Don t know why it s not done by default
 696     self.removeAllVarsInContext()
 697     pass
 698
 699   def getContainer(self):
 700     return self.my_container
 701
 702   def getCode(self):
 703     return self.code
 704
 705   def getName(self):
 706     return self.nodeName
 707
 708   def defineNewCustomVar(self,varName,valueOfVar):
 709     self.context[varName] = pickle.loads(valueOfVar)
 710     pass
 711
 712   def executeAnotherPieceOfCode(self,code):
 713     """Called for initialization of container lodging self."""
 714     try:
 715       ccode=compile(code,self.nodeName,'exec')
 716       exec(ccode, self.context)
 717     except Exception:
 718       raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"","PyScriptNode (%s) : code to be executed \"%s\"" %(self.nodeName,code),0))
 719
 720   def assignNewCompiledCode(self,codeStr):
 721     try:
 722       self.code=codeStr
 723       self.ccode=compile(codeStr,self.nodeName,'exec')
 724     except Exception:
 725       raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"","PyScriptNode.assignNewCompiledCode (%s) : code to be executed \"%s\"" %(self.nodeName,codeStr),0))
 726
 727   def executeSimple(self, key, val):
 728     """
 729     Same as execute method except that no pickelization mecanism is implied here. No output is expected
 730     """
 731     try:
 732       self.context.update({ "env" : [(k,v) for k,v in zip(key,val)]})
 733       exec(self.ccode,self.context)
 734     except Exception:
 735       exc_typ,exc_val,exc_fr=sys.exc_info()
 736       l=traceback.format_exception(exc_typ,exc_val,exc_fr)
 737       print("".join(l)) ; sys.stdout.flush() # print error also in logs of remote container
 738       raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"".join(l),"PyScriptNode: %s" % (self.nodeName),0))
 739
 740   def execute(self,outargsname,argsin):
 741     """Execute the script stored in attribute ccode with pickled args (argsin)"""
 742     try:
 743       argsname,kws=pickle.loads(argsin)
 744       self.context.update(kws)
 745       exec(self.ccode, self.context)
 746       argsout=[]
 747       for arg in outargsname:
 748         if arg not in self.context:
 749           raise KeyError("There is no variable %s in context" % arg)
 750         argsout.append(self.context[arg])
 751       argsout=pickle.dumps(tuple(argsout),-1)
 752       return argsout
 753     except Exception:
 754       exc_typ,exc_val,exc_fr=sys.exc_info()
 755       l=traceback.format_exception(exc_typ,exc_val,exc_fr)
 756       print("".join(l)) ; sys.stdout.flush() # print error also in logs of remote container
 757       raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"".join(l),"PyScriptNode: %s, outargsname: %s" % (self.nodeName,outargsname),0))
 758
 759   def executeFirst(self,argsin):
 760     """ Same than first part of self.execute to reduce memory peak."""
 761     def ArgInMananger(self,argsin):
 762       argsInPy = SeqByteReceiver( argsin )
 763       data = argsInPy.data()
 764       self.addInfoOnLevel2("inputMem",len(data))
 765       _,kws=pickle.loads(data)
 766       return kws
 767     try:
 768       self.beginOfCurrentExecutionSession()
 769       self.addTimeInfoOnLevel2("startInputTime")
 770       # to force call of SeqByteReceiver's destructor
 771       kws = ArgInMananger(self,argsin)
 772       vis = InOutputObjVisitor()
 773       for elt in kws:
 774         # fetch real data if necessary
 775         kws[elt] = UnProxyObjectSimple( kws[elt],vis)
 776       self.addInfoOnLevel2("inputHDDMem",vis)
 777       self.context.update(kws)
 778       self.addTimeInfoOnLevel2("endInputTime")
 779     except Exception:
 780       exc_typ,exc_val,exc_fr=sys.exc_info()
 781       l=traceback.format_exception(exc_typ,exc_val,exc_fr)
 782       print("".join(l)) ; sys.stdout.flush() # print error also in logs of remote container
 783       raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"".join(l),"PyScriptNode:First %s" % (self.nodeName),0))
 784
 785   def executeSecond(self,outargsname):
 786     """ Same than second part of self.execute to reduce memory peak."""
 787     import sys
 788     try:
 789       self.addTimeInfoOnLevel2("startExecTime")
 790       ##
 791       self.addInfoOnLevel2("measureTimeResolution",self.my_container_py.monitoringtimeresms())
 792       with GenericPythonMonitoringLauncherCtxMgr( CPUMemoryMonitoring( self.my_container_py.monitoringtimeresms() ) ) as monitoringParams:
 793         exec(self.ccode, self.context)
 794         cpumeminfo = ReadCPUMemInfo( monitoringParams )
 795       ##
 796       self.addInfoOnLevel2("CPUMemDuringExec",cpumeminfo)
 797       del monitoringParams
 798       self.addTimeInfoOnLevel2("endExecTime")
 799       self.addTimeInfoOnLevel2("startOutputTime")
 800       argsout=[]
 801       for arg in outargsname:
 802         if arg not in self.context:
 803           raise KeyError("There is no variable %s in context" % arg)
 804         argsout.append(self.context[arg])
 805       ret = [ ]
 806       outputMem = 0
 807       vis = InOutputObjVisitor()
 808       for arg in argsout:
 809         # the proxy mecanism is catched here
 810         argPickle = SpoolPickleObject( arg, vis )
 811         retArg = SenderByte_i( self.poa,argPickle )
 812         id_o = self.poa.activate_object(retArg)
 813         retObj = self.poa.id_to_reference(id_o)
 814         ret.append( retObj._narrow( SALOME.SenderByte ) )
 815         outputMem += len(argPickle)
 816       self.addInfoOnLevel2("outputMem",outputMem)
 817       self.addInfoOnLevel2("outputHDDMem",vis)
 818       self.addTimeInfoOnLevel2("endOutputTime")
 819       self.endOfCurrentExecutionSession()
 820       return ret
 821     except Exception:
 822       exc_typ,exc_val,exc_fr=sys.exc_info()
 823       l=traceback.format_exception(exc_typ,exc_val,exc_fr)
 824       print("".join(l)) ; sys.stdout.flush() # print error also in logs of remote container
 825       raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"".join(l),"PyScriptNode:Second %s, outargsname: %s" % (self.nodeName,outargsname),0))
 826
 827   def listAllVarsInContext(self):
 828       import re
 829       pat = re.compile("^__([a-z]+)__$")
 830       return [elt for elt in self.context if not pat.match(elt) and elt != MY_CONTAINER_ENTRY_IN_GLBS]
 831
 832   def removeAllVarsInContext(self):
 833       for elt in self.listAllVarsInContext():
 834         del self.context[elt]
 835
 836   def getValueOfVarInContext(self,varName):
 837     try:
 838       return pickle.dumps(self.context[varName],-1)
 839     except Exception:
 840       exc_typ,exc_val,exc_fr=sys.exc_info()
 841       l=traceback.format_exception(exc_typ,exc_val,exc_fr)
 842       raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"".join(l),"PyScriptNode: %s" %self.nodeName,0))
 843     pass
 844
 845   def assignVarInContext(self, varName, value):
 846     try:
 847       self.context[varName][0] = pickle.loads(value)
 848     except Exception:
 849       exc_typ,exc_val,exc_fr=sys.exc_info()
 850       l=traceback.format_exception(exc_typ,exc_val,exc_fr)
 851       raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"".join(l),"PyScriptNode: %s" %self.nodeName,0))
 852     pass
 853
 854   def callMethodOnVarInContext(self, varName, methodName, args):
 855     try:
 856       return pickle.dumps( getattr(self.context[varName][0],methodName)(*pickle.loads(args)),-1 )
 857     except Exception:
 858       exc_typ,exc_val,exc_fr=sys.exc_info()
 859       l=traceback.format_exception(exc_typ,exc_val,exc_fr)
 860       raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"".join(l),"PyScriptNode: %s" %self.nodeName,0))
 861     pass
 862
 863   def beginOfCurrentExecutionSession(self):
 864     self._current_execution_session = LogOfCurrentExecutionSession( self._log_script.addExecutionSession() )
 865     self.context[MY_PERFORMANCE_LOG_ENTRY_IN_GLBS] = self._current_execution_session
 866
 867   def endOfCurrentExecutionSession(self):
 868     self._current_execution_session.finalizeAndPushToMaster()
 869     self._current_execution_session = None
 870
 871   def addInfoOnLevel2(self, key, value):
 872     self._current_execution_session.addInfoOnLevel2(key, value)
 873
 874   def addTimeInfoOnLevel2(self, key):
 875     from datetime import datetime
 876     self._current_execution_session.addInfoOnLevel2(key,datetime.now())