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.err_queue = err_queue
                      self.addRunOperation(**kwargs)
                  def update(self, **kwargs):
                      self.opConfObjList = []
-                 def addRunOperation(self, **kwargs):
+                 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:
+                 def get(self):
+                     if not self.islocked and self.size/1000000 > 512:
+                         self.lock.n.value += 1
+                         self.islocked = True
                          self.lock.clear()
-                     else:
-                         self.lock.set()
+                     elif self.islocked and self.size/1000000 <= 512:
+                         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)
+                     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