1 # -*- coding: iso-8859-1 -*-
2 # Copyright (C) 2007-2023 CEA, EDF, OPEN CASCADE
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.
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.
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
18 # See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
21 # File : SALOME_PyNode.py
22 # Author : Christian CAREMOLI, EDF
35 from SALOME_ContainerHelper import ScriptExecInfo
37 MY_CONTAINER_ENTRY_IN_GLBS = "my_container"
39 class Generic(SALOME__POA.GenericObj):
40 """A Python implementation of the GenericObj CORBA IDL"""
41 def __init__(self,poa):
46 #print("Register called : %d"%self.cnt)
50 #print("UnRegister called : %d"%self.cnt)
53 oid=self.poa.servant_to_id(self)
54 self.poa.deactivate_object(oid)
57 print("WARNING SALOME::GenericObj::Destroy() function is obsolete! Use UnRegister() instead.")
61 #print("Destuctor called")
64 class PyNode_i (Engines__POA.PyNode,Generic):
65 """The implementation of the PyNode CORBA IDL"""
66 def __init__(self, nodeName,code,poa,my_container):
67 """Initialize the node : compilation in the local context"""
68 Generic.__init__(self,poa)
69 self.nodeName=nodeName
71 self.my_container=my_container._container
72 linecache.cache[nodeName]=0,None,code.split('\n'),nodeName
73 ccode=compile(code,nodeName,'exec')
75 self.context[MY_CONTAINER_ENTRY_IN_GLBS] = self.my_container
76 exec(ccode, self.context)
78 def getContainer(self):
79 return self.my_container
87 def defineNewCustomVar(self,varName,valueOfVar):
88 self.context[varName] = pickle.loads(valueOfVar)
91 def executeAnotherPieceOfCode(self,code):
92 """Called for initialization of container lodging self."""
94 ccode=compile(code,self.nodeName,'exec')
95 exec(ccode, self.context)
97 raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"","PyScriptNode (%s) : code to be executed \"%s\"" %(self.nodeName,code),0))
99 def execute(self,funcName,argsin):
100 """Execute the function funcName found in local context with pickled args (argsin)"""
102 argsin,kws=pickle.loads(argsin)
103 func=self.context[funcName]
104 argsout=func(*argsin,**kws)
105 argsout=pickle.dumps(argsout,-1)
108 exc_typ,exc_val,exc_fr=sys.exc_info()
109 l=traceback.format_exception(exc_typ,exc_val,exc_fr)
110 raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"".join(l),"PyNode: %s, function: %s" % (self.nodeName,funcName),0))
112 class SenderByte_i(SALOME__POA.SenderByte,Generic):
113 def __init__(self,poa,bytesToSend):
114 Generic.__init__(self,poa)
115 self.bytesToSend = bytesToSend
118 return len(self.bytesToSend)
120 def sendPart(self,n1,n2):
121 return self.bytesToSend[n1:n2]
123 SALOME_FILE_BIG_OBJ_DIR = "SALOME_FILE_BIG_OBJ_DIR"
125 SALOME_BIG_OBJ_ON_DISK_THRES_VAR = "SALOME_BIG_OBJ_ON_DISK_THRES"
128 SALOME_BIG_OBJ_ON_DISK_THRES_DFT = 50000000
130 from ctypes import c_int
134 return len( bytes(TypeCounter(0) ) )
136 def GetSizeOfBufferedReader(f):
138 This method returns in bytes size of a file openned.
142 f (io.IOBase): buffered reader returned by open
150 f.seek(0,io.SEEK_END)
152 f.seek(pos,io.SEEK_SET)
155 def GetObjectFromFile(fname, visitor = None):
156 with open(fname,"rb") as f:
157 cntb = f.read( GetSizeOfTCnt() )
158 cnt = TypeCounter.from_buffer_copy( cntb ).value
160 visitor.setHDDMem( GetSizeOfBufferedReader(f) )
161 visitor.setFileName( fname )
165 def DumpInFile(obj,fname):
166 with open(fname,"wb") as f:
167 f.write( bytes( TypeCounter(1) ) )
170 def IncrRefInFile(fname):
171 with open(fname,"rb") as f:
172 cntb = f.read( GetSizeOfTCnt() )
173 cnt = TypeCounter.from_buffer_copy( cntb ).value
174 with open(fname,"rb+") as f:
175 #import KernelServices ; KernelServices.EntryForDebuggerBreakPoint()
176 f.write( bytes( TypeCounter(cnt+1) ) )
178 def DecrRefInFile(fname):
180 with open(fname,"rb") as f:
181 cntb = f.read( GetSizeOfTCnt() )
182 cnt = TypeCounter.from_buffer_copy( cntb ).value
184 #import KernelServices ; KernelServices.EntryForDebuggerBreakPoint()
188 with open(fname,"rb+") as f:
189 f.write( bytes( TypeCounter(cnt-1) ) )
191 def GetBigObjectOnDiskThreshold():
193 if SALOME_BIG_OBJ_ON_DISK_THRES_VAR in os.environ:
194 return int( os.environ[SALOME_BIG_OBJ_ON_DISK_THRES_VAR] )
196 return SALOME_BIG_OBJ_ON_DISK_THRES_DFT
198 def ActivateProxyMecanismOrNot( sizeInByte ):
199 thres = GetBigObjectOnDiskThreshold()
203 return sizeInByte > thres
205 def GetBigObjectDirectory():
207 if SALOME_FILE_BIG_OBJ_DIR not in os.environ:
208 raise RuntimeError("An object of size higher than limit detected and no directory specified to dump it in file !")
209 return os.path.expanduser( os.path.expandvars( os.environ[SALOME_FILE_BIG_OBJ_DIR] ) )
211 def GetBigObjectFileName():
213 Return a filename in the most secure manner (see tempfile documentation)
216 with tempfile.NamedTemporaryFile(dir=GetBigObjectDirectory(),prefix="mem_",suffix=".pckl") as f:
220 class BigObjectOnDiskBase:
221 def __init__(self, fileName, objSerialized):
223 :param fileName: the file used to dump into.
224 :param objSerialized: the object in pickeled form
225 :type objSerialized: bytes
227 self._filename = fileName
228 # attribute _destroy is here to tell client side or server side
229 # only client side can be with _destroy set to True. server side due to risk of concurrency
230 # so pickled form of self must be done with this attribute set to False.
231 self._destroy = False
232 self.__dumpIntoFile(objSerialized)
234 def getDestroyStatus(self):
239 IncrRefInFile( self._filename )
241 # should never happen !
242 RuntimeError("Invalid call to incrRef !")
246 DecrRefInFile( self._filename )
248 # should never happen !
249 RuntimeError("Invalid call to decrRef !")
251 def unlinkOnDestructor(self):
254 def doNotTouchFile(self):
256 Method called slave side. The life cycle management of file is client side not slave side.
258 self._destroy = False
262 DecrRefInFile( self._filename )
264 def getFileName(self):
265 return self._filename
267 def __dumpIntoFile(self, objSerialized):
268 DumpInFile( objSerialized, self._filename )
270 def get(self, visitor = None):
271 obj, _ = GetObjectFromFile( self._filename, visitor )
275 return float( self.get() )
278 return int( self.get() )
282 if isinstance(obj,str):
285 raise RuntimeError("Not a string")
287 class BigObjectOnDisk(BigObjectOnDiskBase):
288 def __init__(self, fileName, objSerialized):
289 BigObjectOnDiskBase.__init__(self, fileName, objSerialized)
291 class BigObjectOnDiskListElement(BigObjectOnDiskBase):
292 def __init__(self, pos, length, fileName):
293 self._filename = fileName
294 self._destroy = False
296 self._length = length
298 def get(self, visitor = None):
299 fullObj = BigObjectOnDiskBase.get(self, visitor)
300 return fullObj[ self._pos ]
302 def __getitem__(self, i):
306 return len(self.get())
308 class BigObjectOnDiskSequence(BigObjectOnDiskBase):
309 def __init__(self, length, fileName, objSerialized):
310 BigObjectOnDiskBase.__init__(self, fileName, objSerialized)
311 self._length = length
313 def __getitem__(self, i):
314 return BigObjectOnDiskListElement(i, self._length, self.getFileName())
319 class BigObjectOnDiskList(BigObjectOnDiskSequence):
320 def __init__(self, length, fileName, objSerialized):
321 BigObjectOnDiskSequence.__init__(self, length, fileName, objSerialized)
323 class BigObjectOnDiskTuple(BigObjectOnDiskSequence):
324 def __init__(self, length, fileName, objSerialized):
325 BigObjectOnDiskSequence.__init__(self, length, fileName, objSerialized)
327 def ProxyfyPickeled( obj, pickleObjInit = None, visitor = None ):
329 This method return a proxy instance of pickled form of object given in input.
333 obj (pickelable type) : object to be proxified
334 pickleObjInit (bytes) : Optionnal. Original pickeled form of object to be proxyfied if already computed. If not this method generate it
338 BigObjectOnDiskBase: proxy instance
340 pickleObj = pickleObjInit
341 if pickleObj is None:
342 pickleObj = pickle.dumps( obj , pickle.HIGHEST_PROTOCOL )
343 fileName = GetBigObjectFileName()
345 visitor.setHDDMem( len(pickleObj) )
346 visitor.setFileName(fileName)
347 if isinstance( obj, list):
348 proxyObj = BigObjectOnDiskList( len(obj), fileName, pickleObj )
349 elif isinstance( obj, tuple):
350 proxyObj = BigObjectOnDiskTuple( len(obj), fileName , pickleObj )
352 proxyObj = BigObjectOnDisk( fileName , pickleObj )
355 def SpoolPickleObject( obj, visitor = None ):
357 with InOutputObjVisitorCM(visitor) as v:
358 pickleObjInit = pickle.dumps( obj , pickle.HIGHEST_PROTOCOL )
359 if not ActivateProxyMecanismOrNot( len(pickleObjInit) ):
362 proxyObj = ProxyfyPickeled( obj, pickleObjInit, v.visitor() )
363 pickleProxy = pickle.dumps( proxyObj , pickle.HIGHEST_PROTOCOL )
366 from SALOME_ContainerHelper import InOutputObjVisitorCM, InOutputObjVisitor
368 def UnProxyObjectSimple( obj, visitor = None ):
370 Method to be called in Remote mode. Alterate the obj _status attribute.
371 Because the slave process does not participate in the reference counting
375 visitor (InOutputObjVisitor): A visitor to keep track of amount of memory on chip and those on HDD
378 with InOutputObjVisitorCM(visitor) as v:
379 logging.debug( "UnProxyObjectSimple {}".format(type(obj)) )
380 if isinstance(obj,BigObjectOnDiskBase):
383 elif isinstance( obj, list):
386 retObj.append( UnProxyObjectSimple(elt,v.visitor()) )
391 def UnProxyObjectSimpleLocal( obj ):
393 Method to be called in Local mode. Do not alterate the PyObj counter
395 if isinstance(obj,BigObjectOnDiskBase):
397 elif isinstance( obj, list):
400 retObj.append( UnProxyObjectSimpleLocal(elt) )
406 def __init__(self, fileName):
407 self._filename = fileName
410 return self._filename
413 if os.path.exists( self._filename ):
414 os.unlink( self._filename )
416 def LaunchMonitoring( intervalInMs ):
418 Launch a subprocess monitoring self process.
419 This monitoring subprocess is a python process lauching every intervalInMs ms evaluation of
420 CPU usage and RSS memory.
421 Communication between subprocess and self is done by file.
424 def BuildPythonFileForCPUPercent( intervalInMs ):
427 with tempfile.NamedTemporaryFile(prefix="htop_",suffix=".py") as f:
429 tempOutFile = "{}.txt".format( os.path.splitext( tempPyFile )[0] )
431 with open(tempPyFile,"w") as f:
432 f.write("""import psutil
434 process = psutil.Process( pid )
436 with open("{}","a") as f:
438 f.write( "{{}}\\n".format( str( process.cpu_percent() ) ) )
439 f.write( "{{}}\\n".format( str( process.memory_info().rss ) ) )
441 time.sleep( {} / 1000.0 )
442 """.format(pid, tempOutFile, intervalInMs))
443 return FileDeleter(tempPyFile), FileDeleter(tempOutFile)
444 pyFileName, outFileName = BuildPythonFileForCPUPercent( intervalInMs )
445 KernelBasis.LaunchMonitoring(pyFileName.filename,outFileName.filename)
446 return pyFileName, outFileName
448 def StopMonitoring( ):
450 Retrieve data of monitoring and kill monitoring subprocess.
454 list<float,str> : list of pairs. First param of pair is CPU usage. Second param of pair is rss memory usage
457 ret = KernelBasis.StopMonitoring()
459 mem_rss = [ int(elt) for elt in ret[1::2]]
460 return [(a,b) for a,b in zip(cpu,mem_rss)]
462 class SeqByteReceiver:
463 # 2GB limit to trigger split into chunks
464 CHUNK_SIZE = 2000000000
465 def __init__(self,sender):
468 self._obj.UnRegister()
471 size = self._obj.getSize()
472 if size <= SeqByteReceiver.CHUNK_SIZE:
473 return self.fetchOneShot( size )
475 return self.fetchByChunks( size )
476 def fetchOneShot(self,size):
477 return self._obj.sendPart(0,size)
478 def fetchByChunks(self,size):
480 To avoid memory peak parts over 2GB are sent using EFF_CHUNK_SIZE size.
482 data_for_split_case = bytes(0)
483 EFF_CHUNK_SIZE = SeqByteReceiver.CHUNK_SIZE // 8
484 iStart = 0 ; iEnd = EFF_CHUNK_SIZE
485 while iStart!=iEnd and iEnd <= size:
486 part = self._obj.sendPart(iStart,iEnd)
487 data_for_split_case = bytes(0).join( [data_for_split_case,part] )
488 iStart = iEnd; iEnd = min(iStart + EFF_CHUNK_SIZE,size)
489 return data_for_split_case
491 class LogOfCurrentExecutionSession:
492 def __init__(self, handleToCentralizedInst):
493 self._remote_handle = handleToCentralizedInst
494 self._current_instance = ScriptExecInfo()
496 def addInfoOnLevel2(self, key, value):
497 setattr(self._current_instance,key,value)
499 def finalizeAndPushToMaster(self):
500 self._remote_handle.assign( pickle.dumps( self._current_instance ) )
502 class PyScriptNode_i (Engines__POA.PyScriptNode,Generic):
503 """The implementation of the PyScriptNode CORBA IDL that executes a script"""
504 def __init__(self, nodeName,code,poa,my_container,logscript):
505 """Initialize the node : compilation in the local context"""
506 Generic.__init__(self,poa)
507 self.nodeName=nodeName
509 self.my_container_py = my_container
510 self.my_container=my_container._container
511 linecache.cache[nodeName]=0,None,code.split('\n'),nodeName
512 self.ccode=compile(code,nodeName,'exec')
514 self.context[MY_CONTAINER_ENTRY_IN_GLBS] = self.my_container
515 self._log_script = logscript
516 self._current_execution_session = None
517 sys.stdout.flush() ; sys.stderr.flush() # flush to correctly capture log per execution session
520 # force removal of self.context. Don t know why it s not done by default
521 self.removeAllVarsInContext()
524 def getContainer(self):
525 return self.my_container
533 def defineNewCustomVar(self,varName,valueOfVar):
534 self.context[varName] = pickle.loads(valueOfVar)
537 def executeAnotherPieceOfCode(self,code):
538 """Called for initialization of container lodging self."""
540 ccode=compile(code,self.nodeName,'exec')
541 exec(ccode, self.context)
543 raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"","PyScriptNode (%s) : code to be executed \"%s\"" %(self.nodeName,code),0))
545 def assignNewCompiledCode(self,codeStr):
548 self.ccode=compile(codeStr,self.nodeName,'exec')
550 raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"","PyScriptNode.assignNewCompiledCode (%s) : code to be executed \"%s\"" %(self.nodeName,codeStr),0))
552 def executeSimple(self, key, val):
554 Same as execute method except that no pickelization mecanism is implied here. No output is expected
557 self.context.update({ "env" : [(k,v) for k,v in zip(key,val)]})
558 exec(self.ccode,self.context)
560 exc_typ,exc_val,exc_fr=sys.exc_info()
561 l=traceback.format_exception(exc_typ,exc_val,exc_fr)
562 print("".join(l)) ; sys.stdout.flush() # print error also in logs of remote container
563 raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"".join(l),"PyScriptNode: %s" % (self.nodeName),0))
565 def execute(self,outargsname,argsin):
566 """Execute the script stored in attribute ccode with pickled args (argsin)"""
568 argsname,kws=pickle.loads(argsin)
569 self.context.update(kws)
570 exec(self.ccode, self.context)
572 for arg in outargsname:
573 if arg not in self.context:
574 raise KeyError("There is no variable %s in context" % arg)
575 argsout.append(self.context[arg])
576 argsout=pickle.dumps(tuple(argsout),-1)
579 exc_typ,exc_val,exc_fr=sys.exc_info()
580 l=traceback.format_exception(exc_typ,exc_val,exc_fr)
581 print("".join(l)) ; sys.stdout.flush() # print error also in logs of remote container
582 raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"".join(l),"PyScriptNode: %s, outargsname: %s" % (self.nodeName,outargsname),0))
584 def executeFirst(self,argsin):
585 """ Same than first part of self.execute to reduce memory peak."""
587 self.beginOfCurrentExecutionSession()
589 self.addTimeInfoOnLevel2("startInputTime")
590 if True: # to force call of SeqByteReceiver's destructor
591 argsInPy = SeqByteReceiver( argsin )
592 data = argsInPy.data()
593 self.addInfoOnLevel2("inputMem",len(data))
594 _,kws=pickle.loads(data)
595 vis = InOutputObjVisitor()
597 # fetch real data if necessary
598 kws[elt] = UnProxyObjectSimple( kws[elt],vis)
599 self.addInfoOnLevel2("inputHDDMem",vis)
600 self.context.update(kws)
601 self.addTimeInfoOnLevel2("endInputTime")
603 exc_typ,exc_val,exc_fr=sys.exc_info()
604 l=traceback.format_exception(exc_typ,exc_val,exc_fr)
605 print("".join(l)) ; sys.stdout.flush() # print error also in logs of remote container
606 raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"".join(l),"PyScriptNode:First %s" % (self.nodeName),0))
608 def executeSecond(self,outargsname):
609 """ Same than second part of self.execute to reduce memory peak."""
612 self.addTimeInfoOnLevel2("startExecTime")
614 self.addInfoOnLevel2("measureTimeResolution",self.my_container_py.monitoringtimeresms())
615 monitoringParams = LaunchMonitoring( self.my_container_py.monitoringtimeresms() )
616 exec(self.ccode, self.context)
617 cpumeminfo = StopMonitoring( )
619 self.addInfoOnLevel2("CPUMemDuringExec",cpumeminfo)
621 self.addTimeInfoOnLevel2("endExecTime")
622 self.addTimeInfoOnLevel2("startOutputTime")
624 for arg in outargsname:
625 if arg not in self.context:
626 raise KeyError("There is no variable %s in context" % arg)
627 argsout.append(self.context[arg])
630 vis = InOutputObjVisitor()
632 # the proxy mecanism is catched here
633 argPickle = SpoolPickleObject( arg, vis )
634 retArg = SenderByte_i( self.poa,argPickle )
635 id_o = self.poa.activate_object(retArg)
636 retObj = self.poa.id_to_reference(id_o)
637 ret.append( retObj._narrow( SALOME.SenderByte ) )
638 outputMem += len(argPickle)
639 self.addInfoOnLevel2("outputMem",outputMem)
640 self.addInfoOnLevel2("outputHDDMem",vis)
641 self.addTimeInfoOnLevel2("endOutputTime")
642 self.endOfCurrentExecutionSession()
645 exc_typ,exc_val,exc_fr=sys.exc_info()
646 l=traceback.format_exception(exc_typ,exc_val,exc_fr)
647 print("".join(l)) ; sys.stdout.flush() # print error also in logs of remote container
648 raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"".join(l),"PyScriptNode:Second %s, outargsname: %s" % (self.nodeName,outargsname),0))
650 def listAllVarsInContext(self):
652 pat = re.compile("^__([a-z]+)__$")
653 return [elt for elt in self.context if not pat.match(elt) and elt != MY_CONTAINER_ENTRY_IN_GLBS]
655 def removeAllVarsInContext(self):
656 for elt in self.listAllVarsInContext():
657 del self.context[elt]
659 def getValueOfVarInContext(self,varName):
661 return pickle.dumps(self.context[varName],-1)
663 exc_typ,exc_val,exc_fr=sys.exc_info()
664 l=traceback.format_exception(exc_typ,exc_val,exc_fr)
665 raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"".join(l),"PyScriptNode: %s" %self.nodeName,0))
668 def assignVarInContext(self, varName, value):
670 self.context[varName][0] = pickle.loads(value)
672 exc_typ,exc_val,exc_fr=sys.exc_info()
673 l=traceback.format_exception(exc_typ,exc_val,exc_fr)
674 raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"".join(l),"PyScriptNode: %s" %self.nodeName,0))
677 def callMethodOnVarInContext(self, varName, methodName, args):
679 return pickle.dumps( getattr(self.context[varName][0],methodName)(*pickle.loads(args)),-1 )
681 exc_typ,exc_val,exc_fr=sys.exc_info()
682 l=traceback.format_exception(exc_typ,exc_val,exc_fr)
683 raise SALOME.SALOME_Exception(SALOME.ExceptionStruct(SALOME.BAD_PARAM,"".join(l),"PyScriptNode: %s" %self.nodeName,0))
686 def beginOfCurrentExecutionSession(self):
687 self._current_execution_session = LogOfCurrentExecutionSession( self._log_script.addExecutionSession() )
689 def endOfCurrentExecutionSession(self):
690 self._current_execution_session.finalizeAndPushToMaster()
691 self._current_execution_session = None
693 def addInfoOnLevel2(self, key, value):
694 self._current_execution_session.addInfoOnLevel2(key, value)
696 def addTimeInfoOnLevel2(self, key):
697 from datetime import datetime
698 self._current_execution_session.addInfoOnLevel2(key,datetime.now())