schain Commit - r1268:b2726fff6520 · Jicamarca Repository

Fix excessive memory RAM consumption

jespinoza -

r1268:b2726fff6520

parent child

Context file:

r1268:b2726fff6520

Collapse all files

schainpy/controller.py +8 -8

             import math
             import time
             import zmq
-            from multiprocessing import Process, Queue, Event, cpu_count
+            from multiprocessing import Process, Queue, Event, Value, cpu_count
             from threading import Thread
             from xml.etree.ElementTree import ElementTree, Element, SubElement, tostring
             from xml.dom import minidom
                     self.opConfObjList = []
                     self.procUnitObj = None
                     self.opObjDict = {}
-                    self.mylock = Event()
                 def __getPriority(self):
                     id = self.__getNewId()
                     priority = self.__getPriority() # Sin mucho sentido, pero puede usarse
                     opConfObj = OperationConf()
-                    opConfObj.setup(id, name=name, priority=priority, type=optype, project_id=self.project_id, err_queue=self.err_queue, lock=self.mylock)
+                    opConfObj.setup(id, name=name, priority=priority, type=optype, project_id=self.project_id, err_queue=self.err_queue, lock=self.lock)
                     self.opConfObjList.append(opConfObj)
                     return opConfObj
                     self.name = None
                     self.inputId = None
                     self.opConfObjList = []
-                    self.mylock = Event()
+                    self.lock = Event()
+                    self.lock.set()
+                    self.lock.n = Value('d', 0)
                 def getElementName(self):
                     self.startTime = startTime
                     self.endTime = endTime
                     self.server = server
                     self.err_queue = err_queue
-                    self.lock = self.mylock
                     self.addRunOperation(**kwargs)
                 def update(self, **kwargs):
                     self.opConfObjList = []
                 def addRunOperation(self, **kwargs):
                     opObj = self.addOperation(name='run', optype='self')
                     idProcUnit = self.__getNewId()
                     procUnitConfObj = ProcUnitConf()
                     input_proc = self.procUnitConfObjDict[inputId]
-                    procUnitConfObj.setup(self.id, idProcUnit, name, datatype, inputId, self.err_queue, input_proc.mylock)
+                    procUnitConfObj.setup(self.id, idProcUnit, name, datatype, inputId, self.err_queue, input_proc.lock)
                     self.procUnitConfObjDict[procUnitConfObj.getId()] = procUnitConfObj
                     return procUnitConfObj

schainpy/model/proc/jroproc_base.py +23 -10

                     self.project_id = project_id
                     self.inputId = inputId
                     self.lock = lock
+                    self.islocked = False
                     self.size = 0
                 def run(self):
                         self.size += sys.getsizeof(obj)
                         self.queue.put(obj)
                 def get(self):
-                    if self.size/1000000 > 2048:
+                    if not self.islocked and self.size/1000000 > 512:
+                        self.lock.n.value += 1
+                        self.islocked = True
                         self.lock.clear()
-                    else:
+                    elif self.islocked and self.size/1000000 <= 512:
-                        self.lock.set()
+                        self.islocked = False
+                        self.lock.n.value -= 1
+                        if self.lock.n.value == 0:
+                            self.lock.set()
                     obj = self.queue.get()
                     self.size -= sys.getsizeof(obj)
                     return pickle.loads(obj)
             def MPDecorator(BaseClass):
                 """
                 Multiprocessing class decorator
                         self.lock = args[4]
                         self.typeProc = args[5]
                         self.err_queue.put('#_start_#')
-                        self.queue = InputQueue(self.project_id, self.inputId, self.lock)
+                        if self.inputId is not None:
+                            self.queue = InputQueue(self.project_id, self.inputId, self.lock)
                     def subscribe(self):
                         '''
                         '''
                         This function publish an object, to an specific topic.
                         It blocks publishing when receiver queue is full to avoid data loss
                         '''
                         if self.inputId is None:
                             self.lock.wait()
                         self.sender.send_multipart([str(id).encode(), pickle.dumps(data)])
                             dataOut = self.listen()
                             if not dataOut.error:
-                                BaseClass.run(self, dataOut, **self.kwargs)
+                                try:
+                                    BaseClass.run(self, dataOut, **self.kwargs)
+                                except:
+                                    self.err_queue.put('{}|{}'.format(self.name, traceback.format_exc()))
+                                    dataOut.error = True
                             else:
                                 break

General Comments 0

Write
Preview

You need to be logged in to leave comments. Login now

No TODOs yet

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages