schain Commit - r1256:7ed2e4983ab9 · Jicamarca Repository

Block publishing when input queues are full to avoid data loss

jespinoza -

r1256:7ed2e4983ab9

parent child

Context file:

r1256:7ed2e4983ab9

Collapse all files

schainpy/cli/cli.py +7 -8

             import click
             import schainpy
             import subprocess
             import os
             import sys
             import glob
             save_stdout = sys.stdout
             sys.stdout = open('/dev/null', 'w')
             from multiprocessing import cpu_count
             from schainpy.controller import Project
             from schainpy.model import Operation, ProcessingUnit
             from schainpy.utils import log
             from importlib import import_module
             from pydoc import locate
             from fuzzywuzzy import process
             from schainpy.cli import templates
             import inspect
             try:
                 from queue import Queue
             except:
                 from Queue import Queue
             sys.stdout = save_stdout
             def getProcs():
                 modules = dir(schainpy.model)
                 procs = check_module(modules, 'processing')
                 try:
                     procs.remove('ProcessingUnit')
                 except Exception as e:
                     pass
                 return procs
             def getOperations():
                 module = dir(schainpy.model)
                 noProcs = [x for x in module if not x.endswith('Proc')]
                 operations = check_module(noProcs, 'operation')
                 try:
                     operations.remove('Operation')
                     operations.remove('Figure')
                     operations.remove('Plot')
                 except Exception as e:
                     pass
                 return operations
             def getArgs(op):
                 module = locate('schainpy.model.{}'.format(op))
                 if hasattr(module, '__attrs__'):
                     args = module.__attrs__
                 else:
                     args = inspect.getargspec(module.run).args
                 try:
                     args.remove('self')
                 except Exception as e:
                     pass
                 try:
                     args.remove('dataOut')
                 except Exception as e:
                     pass
                 return args
             def getDoc(obj):
                 module = locate('schainpy.model.{}'.format(obj))
                 try:
-                    obj = module(1,2,3,Queue(),5)
+                    obj = module(1,2,3,Queue(),5,6)
                 except:
                     obj = module()
                 return obj.__doc__
             def getAll():
                 modules = getOperations()
                 modules.extend(getProcs())
                 return modules
             def print_version(ctx, param, value):
                 if not value or ctx.resilient_parsing:
                     return
                 click.echo(schainpy.__version__)
                 ctx.exit()
             PREFIX = 'experiment'
             @click.command()
             @click.option('--version', '-v', is_flag=True, callback=print_version, help='SChain version', type=str)
             @click.argument('command', default='run', required=True)
             @click.argument('nextcommand', default=None, required=False, type=str)
             def main(command, nextcommand, version):
                 """COMMAND LINE INTERFACE FOR SIGNAL CHAIN - JICAMARCA RADIO OBSERVATORY V3.0\n
                     Available commands.\n
                     xml: runs a schain XML generated file\n
                     run: runs any python script starting 'experiment_'\n
                     generate: generates a template schain script\n
                     list: return a list of available procs and operations\n
                     search: return avilable operations, procs or arguments of the given
                             operation/proc\n"""
                 if command == 'xml':
                     runFromXML(nextcommand)
                 elif command == 'generate':
                     generate()
                 elif command == 'test':
                     test()
                 elif command == 'run':
                     runschain(nextcommand)
                 elif command == 'search':
                     search(nextcommand)
                 elif command == 'list':
                     cmdlist(nextcommand)
                 else:
                     log.error('Command {} is not defined'.format(command))
             def check_module(possible, instance):
                 def check(x):
                     try:
                         instancia = locate('schainpy.model.{}'.format(x))
                         ret = instancia.proc_type == instance
                         return ret
                     except Exception as e:
                         return False
                 clean = clean_modules(possible)
                 return [x for x in clean if check(x)]
             def clean_modules(module):
                 noEndsUnder = [x for x in module if not x.endswith('__')]
                 noStartUnder = [x for x in noEndsUnder if not x.startswith('__')]
                 noFullUpper = [x for x in noStartUnder if not x.isupper()]
                 return noFullUpper
             def cmdlist(nextcommand):
                 if nextcommand is None:
                     log.error('Missing argument, available arguments: procs, operations', '')
                 elif nextcommand == 'procs':
                     procs = getProcs()
                     log.success(
                         'Current ProcessingUnits are:\n  {}'.format('\n  '.join(procs)), '')
                 elif nextcommand == 'operations':
                     operations = getOperations()
                     log.success('Current Operations are:\n  {}'.format(
                         '\n  '.join(operations)), '')
                 else:
                     log.error('Wrong argument', '')
             def search(nextcommand):
                 if nextcommand is None:
                     log.error('There is no Operation/ProcessingUnit to search', '')
                 else:
-                    #try:
+                    try:
-                    if True:
                         args = getArgs(nextcommand)
                         doc = getDoc(nextcommand)
                         if len(args) == 0:
                             log.success('\n{} has no arguments'.format(nextcommand), '')
                         else:
                             log.success('{}\n{}\n\narguments:\n  {}'.format(
                                 nextcommand, doc, ', '.join(args)), '')
-                    # except Exception as e:
+                    except Exception as e:
-                    #     log.error('Module `{}` does not exists'.format(nextcommand), '')
+                        log.error('Module `{}` does not exists'.format(nextcommand), '')
-                    #     allModules = getAll()
+                        allModules = getAll()
-                    #     similar = [t[0] for t in process.extract(nextcommand, allModules, limit=12) if t[1]>80]
+                        similar = [t[0] for t in process.extract(nextcommand, allModules, limit=12) if t[1]>80]
-                    #     log.success('Possible modules are: {}'.format(', '.join(similar)), '')
+                        log.success('Possible modules are: {}'.format(', '.join(similar)), '')
             def runschain(nextcommand):
                 if nextcommand is None:
                     currentfiles = glob.glob('./{}_*.py'.format(PREFIX))
                     numberfiles = len(currentfiles)
                     if numberfiles > 1:
                         log.error('There is more than one file to run')
                     elif numberfiles == 1:
                         subprocess.call(['python ' + currentfiles[0]], shell=True)
                     else:
                         log.error('There is no file to run')
                 else:
                     try:
                         subprocess.call(['python ' + nextcommand], shell=True)
                     except Exception as e:
                         log.error("I cannot run the file. Does it exists?")
             def basicInputs():
                 inputs = {}
                 inputs['name'] = click.prompt(
                     'Name of the project', default="project", type=str)
                 inputs['desc'] = click.prompt(
                     'Enter a description', default="A schain project", type=str)
                 inputs['multiprocess'] = click.prompt(
                     '''Select data type:
                 - Voltage (*.r):                [1]
                 - Spectra (*.pdata):            [2]
                 - Voltage and Spectra (*.r):    [3]
                 -->''', type=int)
                 inputs['path'] = click.prompt('Data path', default=os.getcwd(
                 ), type=click.Path(exists=True, resolve_path=True))
                 inputs['startDate'] = click.prompt(
                     'Start date', default='1970/01/01', type=str)
                 inputs['endDate'] = click.prompt(
                     'End date', default='2018/12/31', type=str)
                 inputs['startHour'] = click.prompt(
                     'Start hour', default='00:00:00', type=str)
                 inputs['endHour'] = click.prompt('End hour', default='23:59:59', type=str)
                 inputs['figpath'] = inputs['path'] + '/figs'
                 return inputs
             def generate():
                 inputs = basicInputs()
                 if inputs['multiprocess'] == 1:
                     current = templates.voltage.format(**inputs)
                 elif inputs['multiprocess'] == 2:
                     current = templates.spectra.format(**inputs)
                 elif inputs['multiprocess'] == 3:
                     current = templates.voltagespectra.format(**inputs)
                 scriptname = '{}_{}.py'.format(PREFIX, inputs['name'])
                 script = open(scriptname, 'w')
                 try:
                     script.write(current)
                     log.success('Script {} generated'.format(scriptname))
                 except Exception as e:
                     log.error('I cannot create the file. Do you have writing permissions?')
             def test():
                 log.warning('testing')
             def runFromXML(filename):
                 controller = Project()
                 if not controller.readXml(filename):
                     return
                 controller.start()
                 return

schainpy/controller.py +13 -7

             '''
             Updated on January , 2018, for multiprocessing purposes
             Author: Sergio Cortez
             Created on September , 2012
             '''
             from platform import python_version
             import sys
             import ast
             import datetime
             import traceback
             import math
             import time
             import zmq
-            from multiprocessing import Process, Queue, cpu_count
+            from multiprocessing import Process, Queue, Event, cpu_count
             from threading import Thread
             from xml.etree.ElementTree import ElementTree, Element, SubElement, tostring
             from xml.dom import minidom
             from schainpy.admin import Alarm, SchainWarning
             from schainpy.model import *
             from schainpy.utils import log
             DTYPES = {
                 'Voltage': '.r',
                 'Spectra': '.pdata'
             }
             def MPProject(project, n=cpu_count()):
                 '''
                 Project wrapper to run schain in n processes
                 '''
                 rconf = project.getReadUnitObj()
                 op = rconf.getOperationObj('run')
                 dt1 = op.getParameterValue('startDate')
                 dt2 = op.getParameterValue('endDate')
                 tm1 = op.getParameterValue('startTime')
                 tm2 = op.getParameterValue('endTime')
                 days = (dt2 - dt1).days
                 for day in range(days + 1):
                     skip = 0
                     cursor = 0
                     processes = []
                     dt = dt1 + datetime.timedelta(day)
                     dt_str = dt.strftime('%Y/%m/%d')
                     reader = JRODataReader()
                     paths, files = reader.searchFilesOffLine(path=rconf.path,
                                                              startDate=dt,
                                                              endDate=dt,
                                                              startTime=tm1,
                                                              endTime=tm2,
                                                              ext=DTYPES[rconf.datatype])
                     nFiles = len(files)
                     if nFiles == 0:
                         continue
                     skip = int(math.ceil(nFiles / n))
                     while nFiles > cursor * skip:
                         rconf.update(startDate=dt_str, endDate=dt_str, cursor=cursor,
                                      skip=skip)
                         p = project.clone()
                         p.start()
                         processes.append(p)
                         cursor += 1
                     def beforeExit(exctype, value, trace):
                         for process in processes:
                             process.terminate()
                             process.join()
                         print(traceback.print_tb(trace))
                     sys.excepthook = beforeExit
                     for process in processes:
                         process.join()
                         process.terminate()
                     time.sleep(3)
             def wait(context):
                 time.sleep(1)
                 c = zmq.Context()
                 receiver = c.socket(zmq.SUB)
                 receiver.connect('ipc:///tmp/schain_{}_pub'.format(self.id))
                 receiver.setsockopt(zmq.SUBSCRIBE, self.id.encode())
                 msg = receiver.recv_multipart()[1]
                 context.terminate()
             class ParameterConf():
                 id = None
                 name = None
                 value = None
                 format = None
                 __formated_value = None
                 ELEMENTNAME = 'Parameter'
                 def __init__(self):
                     self.format = 'str'
                 def getElementName(self):
                     return self.ELEMENTNAME
                 def getValue(self):
                     value = self.value
                     format = self.format
                     if self.__formated_value != None:
                         return self.__formated_value
                     if format == 'obj':
                         return value
                     if format == 'str':
                         self.__formated_value = str(value)
                         return self.__formated_value
                     if value == '':
                         raise ValueError('%s: This parameter value is empty' % self.name)
                     if format == 'list':
                         strList = [s.strip() for s in value.split(',')]
                         self.__formated_value = strList
                         return self.__formated_value
                     if format == 'intlist':
                         '''
                         Example:
                             value = (0,1,2)
                         '''
                         new_value = ast.literal_eval(value)
                         if type(new_value) not in (tuple, list):
                             new_value = [int(new_value)]
                         self.__formated_value = new_value
                         return self.__formated_value
                     if format == 'floatlist':
                         '''
                         Example:
                             value = (0.5, 1.4, 2.7)
                         '''
                         new_value = ast.literal_eval(value)
                         if type(new_value) not in (tuple, list):
                             new_value = [float(new_value)]
                         self.__formated_value = new_value
                         return self.__formated_value
                     if format == 'date':
                         strList = value.split('/')
                         intList = [int(x) for x in strList]
                         date = datetime.date(intList[0], intList[1], intList[2])
                         self.__formated_value = date
                         return self.__formated_value
                     if format == 'time':
                         strList = value.split(':')
                         intList = [int(x) for x in strList]
                         time = datetime.time(intList[0], intList[1], intList[2])
                         self.__formated_value = time
                         return self.__formated_value
                     if format == 'pairslist':
                         '''
                         Example:
                             value = (0,1),(1,2)
                         '''
                         new_value = ast.literal_eval(value)
                         if type(new_value) not in (tuple, list):
                             raise ValueError('%s has to be a tuple or list of pairs' % value)
                         if type(new_value[0]) not in (tuple, list):
                             if len(new_value) != 2:
                                 raise ValueError('%s has to be a tuple or list of pairs' % value)
                             new_value = [new_value]
                         for thisPair in new_value:
                             if len(thisPair) != 2:
                                 raise ValueError('%s has to be a tuple or list of pairs' % value)
                         self.__formated_value = new_value
                         return self.__formated_value
                     if format == 'multilist':
                         '''
                         Example:
                             value = (0,1,2),(3,4,5)
                         '''
                         multiList = ast.literal_eval(value)
                         if type(multiList[0]) == int:
                             multiList = ast.literal_eval('(' + value + ')')
                         self.__formated_value = multiList
                         return self.__formated_value
                     if format == 'bool':
                         value = int(value)
                     if format == 'int':
                         value = float(value)
                     format_func = eval(format)
                     self.__formated_value = format_func(value)
                     return self.__formated_value
                 def updateId(self, new_id):
                     self.id = str(new_id)
                 def setup(self, id, name, value, format='str'):
                     self.id = str(id)
                     self.name = name
                     if format == 'obj':
                         self.value = value
                     else:
                         self.value = str(value)
                     self.format = str.lower(format)
                     self.getValue()
                     return 1
                 def update(self, name, value, format='str'):
                     self.name = name
                     self.value = str(value)
                     self.format = format
                 def makeXml(self, opElement):
                     if self.name not in ('queue',):
                         parmElement = SubElement(opElement, self.ELEMENTNAME)
                         parmElement.set('id', str(self.id))
                         parmElement.set('name', self.name)
                         parmElement.set('value', self.value)
                         parmElement.set('format', self.format)
                 def readXml(self, parmElement):
                     self.id = parmElement.get('id')
                     self.name = parmElement.get('name')
                     self.value = parmElement.get('value')
                     self.format = str.lower(parmElement.get('format'))
                     # Compatible with old signal chain version
                     if self.format == 'int' and self.name == 'idfigure':
                         self.name = 'id'
                 def printattr(self):
                     print('Parameter[%s]: name = %s, value = %s, format = %s, project_id = %s' % (self.id, self.name, self.value, self.format, self.project_id))
             class OperationConf():
                 ELEMENTNAME = 'Operation'
                 def __init__(self):
                     self.id = '0'
                     self.name = None
                     self.priority = None
                     self.topic = None
                 def __getNewId(self):
                     return int(self.id) * 10 + len(self.parmConfObjList) + 1
                 def getId(self):
                     return self.id
                 def updateId(self, new_id):
                     self.id = str(new_id)
                     n = 1
                     for parmObj in self.parmConfObjList:
                         idParm = str(int(new_id) * 10 + n)
                         parmObj.updateId(idParm)
                         n += 1
                 def getElementName(self):
                     return self.ELEMENTNAME
                 def getParameterObjList(self):
                     return self.parmConfObjList
                 def getParameterObj(self, parameterName):
                     for parmConfObj in self.parmConfObjList:
                         if parmConfObj.name != parameterName:
                             continue
                         return parmConfObj
                     return None
                 def getParameterObjfromValue(self, parameterValue):
                     for parmConfObj in self.parmConfObjList:
                         if parmConfObj.getValue() != parameterValue:
                             continue
                         return parmConfObj.getValue()
                     return None
                 def getParameterValue(self, parameterName):
                     parameterObj = self.getParameterObj(parameterName)
                 #         if not parameterObj:
                 #             return None
                     value = parameterObj.getValue()
                     return value
                 def getKwargs(self):
                     kwargs = {}
                     for parmConfObj in self.parmConfObjList:
                         if self.name == 'run' and parmConfObj.name == 'datatype':
                             continue
                         kwargs[parmConfObj.name] = parmConfObj.getValue()
                     return kwargs
-                def setup(self, id, name, priority, type, project_id, err_queue):
+                def setup(self, id, name, priority, type, project_id, err_queue, lock):
                     self.id = str(id)
                     self.project_id = project_id
                     self.name = name
                     self.type = type
                     self.priority = priority
                     self.err_queue = err_queue
+                    self.lock = lock
                     self.parmConfObjList = []
                 def removeParameters(self):
                     for obj in self.parmConfObjList:
                         del obj
                     self.parmConfObjList = []
                 def addParameter(self, name, value, format='str'):
                     if value is None:
                         return None
                     id = self.__getNewId()
                     parmConfObj = ParameterConf()
                     if not parmConfObj.setup(id, name, value, format):
                         return None
                     self.parmConfObjList.append(parmConfObj)
                     return parmConfObj
                 def changeParameter(self, name, value, format='str'):
                     parmConfObj = self.getParameterObj(name)
                     parmConfObj.update(name, value, format)
                     return parmConfObj
                 def makeXml(self, procUnitElement):
                     opElement = SubElement(procUnitElement, self.ELEMENTNAME)
                     opElement.set('id', str(self.id))
                     opElement.set('name', self.name)
                     opElement.set('type', self.type)
                     opElement.set('priority', str(self.priority))
                     for parmConfObj in self.parmConfObjList:
                         parmConfObj.makeXml(opElement)
                 def readXml(self, opElement, project_id):
                     self.id = opElement.get('id')
                     self.name = opElement.get('name')
                     self.type = opElement.get('type')
                     self.priority = opElement.get('priority')
                     self.project_id = str(project_id)
                     # Compatible with old signal chain version
                     # Use of 'run' method instead 'init'
                     if self.type == 'self' and self.name == 'init':
                         self.name = 'run'
                     self.parmConfObjList = []
                     parmElementList = opElement.iter(ParameterConf().getElementName())
                     for parmElement in parmElementList:
                         parmConfObj = ParameterConf()
                         parmConfObj.readXml(parmElement)
                         # Compatible with old signal chain version
                         # If an 'plot' OPERATION is found, changes name operation by the value of its type PARAMETER
                         if self.type != 'self' and self.name == 'Plot':
                             if parmConfObj.format == 'str' and parmConfObj.name == 'type':
                                 self.name = parmConfObj.value
                                 continue
                         self.parmConfObjList.append(parmConfObj)
                 def printattr(self):
                     print('%s[%s]: name = %s, type = %s, priority = %s, project_id = %s' % (self.ELEMENTNAME,
                                                                            self.id,
                                                                            self.name,
                                                                            self.type,
                                                                            self.priority,
                                                                            self.project_id))
                     for parmConfObj in self.parmConfObjList:
                         parmConfObj.printattr()
                 def createObject(self):
                     className = eval(self.name)
                     if self.type == 'other':
                         opObj = className()
                     elif self.type == 'external':
                         kwargs = self.getKwargs()
-                        opObj = className(self.id, self.id, self.project_id, self.err_queue, 'Operation', **kwargs)
+                        opObj = className(self.id, self.id, self.project_id, self.err_queue, self.lock, 'Operation', **kwargs)
                         opObj.start()
                         self.opObj = opObj
                     return opObj
             class ProcUnitConf():
                 ELEMENTNAME = 'ProcUnit'
                 def __init__(self):
                     self.id = None
                     self.datatype = None
                     self.name = None
                     self.inputId = None
                     self.opConfObjList = []
                     self.procUnitObj = None
                     self.opObjDict = {}
+                    self.mylock = Event()
                 def __getPriority(self):
                     return len(self.opConfObjList) + 1
                 def __getNewId(self):
                     return int(self.id) * 10 + len(self.opConfObjList) + 1
                 def getElementName(self):
                     return self.ELEMENTNAME
                 def getId(self):
                     return self.id
                 def updateId(self, new_id):
                     '''
                     new_id = int(parentId) * 10 + (int(self.id) % 10)
                     new_inputId = int(parentId) * 10 + (int(self.inputId) % 10)
                     # If this proc unit has not inputs
                     #if self.inputId == '0':
                         #new_inputId = 0
                     n = 1
                     for opConfObj in self.opConfObjList:
                         idOp = str(int(new_id) * 10 + n)
                         opConfObj.updateId(idOp)
                         n += 1
                     self.parentId = str(parentId)
                     self.id = str(new_id)
                     #self.inputId = str(new_inputId)
                     '''
                     n = 1
                 def getInputId(self):
                     return self.inputId
                 def getOperationObjList(self):
                     return self.opConfObjList
                 def getOperationObj(self, name=None):
                     for opConfObj in self.opConfObjList:
                         if opConfObj.name != name:
                             continue
                         return opConfObj
                     return None
                 def getOpObjfromParamValue(self, value=None):
                     for opConfObj in self.opConfObjList:
                         if opConfObj.getParameterObjfromValue(parameterValue=value) != value:
                             continue
                         return opConfObj
                     return None
                 def getProcUnitObj(self):
                     return self.procUnitObj
-                def setup(self, project_id, id, name, datatype, inputId, err_queue):
+                def setup(self, project_id, id, name, datatype, inputId, err_queue, lock):
                     '''
                         id sera el topico a publicar
                         inputId sera el topico a subscribirse
                     '''
                     # Compatible with old signal chain version
                     if datatype == None and name == None:
                         raise ValueError('datatype or name should be defined')
                     #Definir una condicion para inputId cuando sea 0
                     if name == None:
                         if 'Proc' in datatype:
                             name = datatype
                         else:
                             name = '%sProc' % (datatype)
                     if datatype == None:
                         datatype = name.replace('Proc', '')
                     self.id = str(id)
                     self.project_id = project_id
                     self.name = name
                     self.datatype = datatype
                     self.inputId = inputId
                     self.err_queue = err_queue
+                    self.lock = lock
                     self.opConfObjList = []
                     self.addOperation(name='run', optype='self')
                 def removeOperations(self):
                     for obj in self.opConfObjList:
                         del obj
                     self.opConfObjList = []
                     self.addOperation(name='run')
                 def addParameter(self, **kwargs):
                     '''
                     Add parameters to 'run' operation
                     '''
                     opObj = self.opConfObjList[0]
                     opObj.addParameter(**kwargs)
                     return opObj
                 def addOperation(self, name, optype='self'):
                     '''
                         Actualizacion - > proceso comunicacion
                         En el caso de optype='self', elminar. DEfinir comuncacion IPC -> Topic
                         definir el tipoc de socket o comunicacion ipc++
                     '''
                     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)
+                    opConfObj.setup(id, name=name, priority=priority, type=optype, project_id=self.project_id, err_queue=self.err_queue, lock=self.mylock)
                     self.opConfObjList.append(opConfObj)
                     return opConfObj
                 def makeXml(self, projectElement):
                     procUnitElement = SubElement(projectElement, self.ELEMENTNAME)
                     procUnitElement.set('id', str(self.id))
                     procUnitElement.set('name', self.name)
                     procUnitElement.set('datatype', self.datatype)
                     procUnitElement.set('inputId', str(self.inputId))
                     for opConfObj in self.opConfObjList:
                         opConfObj.makeXml(procUnitElement)
                 def readXml(self, upElement, project_id):
                     self.id = upElement.get('id')
                     self.name = upElement.get('name')
                     self.datatype = upElement.get('datatype')
                     self.inputId = upElement.get('inputId')
                     self.project_id = str(project_id)
                     if self.ELEMENTNAME == 'ReadUnit':
                         self.datatype = self.datatype.replace('Reader', '')
                     if self.ELEMENTNAME == 'ProcUnit':
                         self.datatype = self.datatype.replace('Proc', '')
                     if self.inputId == 'None':
                         self.inputId = '0'
                     self.opConfObjList = []
                     opElementList = upElement.iter(OperationConf().getElementName())
                     for opElement in opElementList:
                         opConfObj = OperationConf()
                         opConfObj.readXml(opElement, project_id)
                         self.opConfObjList.append(opConfObj)
                 def printattr(self):
                     print('%s[%s]: name = %s, datatype = %s, inputId = %s, project_id = %s' % (self.ELEMENTNAME,
                                                                               self.id,
                                                                               self.name,
                                                                               self.datatype,
                                                                               self.inputId,
                                                                               self.project_id))
                     for opConfObj in self.opConfObjList:
                         opConfObj.printattr()
                 def getKwargs(self):
                     opObj = self.opConfObjList[0]
                     kwargs = opObj.getKwargs()
                     return kwargs
                 def createObjects(self):
                     '''
                     Instancia de unidades de procesamiento.
                     '''
                     className = eval(self.name)
                     kwargs = self.getKwargs()
-                    procUnitObj = className(self.id, self.inputId, self.project_id, self.err_queue, 'ProcUnit', **kwargs)
+                    procUnitObj = className(self.id, self.inputId, self.project_id, self.err_queue, self.lock, 'ProcUnit', **kwargs)
                     log.success('creating process...', self.name)
                     for opConfObj in self.opConfObjList:
                         if opConfObj.type == 'self' and opConfObj.name == 'run':
                             continue
                         elif opConfObj.type == 'self':
                             opObj = getattr(procUnitObj, opConfObj.name)
                         else:
                             opObj = opConfObj.createObject()
                         log.success('adding operation: {}, type:{}'.format(
                             opConfObj.name,
                             opConfObj.type), self.name)
                         procUnitObj.addOperation(opConfObj, opObj)
                     procUnitObj.start()
                     self.procUnitObj = procUnitObj
                 def close(self):
                     for opConfObj in self.opConfObjList:
                         if opConfObj.type == 'self':
                             continue
                         opObj = self.procUnitObj.getOperationObj(opConfObj.id)
                         opObj.close()
                     self.procUnitObj.close()
                     return
             class ReadUnitConf(ProcUnitConf):
                 ELEMENTNAME = 'ReadUnit'
                 def __init__(self):
                     self.id = None
                     self.datatype = None
                     self.name = None
                     self.inputId = None
                     self.opConfObjList = []
+                    self.mylock = Event()
                 def getElementName(self):
                     return self.ELEMENTNAME
                 def setup(self, project_id, id, name, datatype, err_queue, path='', startDate='', endDate='',
                           startTime='', endTime='', server=None, **kwargs):
                     '''
                     *****el id del proceso sera el Topico
                     Adicion de {topic}, si no esta presente -> error
                     kwargs deben ser trasmitidos en la instanciacion
                     '''
                     # Compatible with old signal chain version
                     if datatype == None and name == None:
                         raise ValueError('datatype or name should be defined')
                     if name == None:
                         if 'Reader' in datatype:
                             name = datatype
                             datatype = name.replace('Reader','')
                         else:
                             name = '{}Reader'.format(datatype)
                     if datatype == None:
                         if 'Reader' in name:
                             datatype = name.replace('Reader','')
                         else:
                             datatype = name
                             name = '{}Reader'.format(name)
                     self.id = id
                     self.project_id = project_id
                     self.name = name
                     self.datatype = datatype
                     if path != '':
                         self.path = os.path.abspath(path)
                     self.startDate = startDate
                     self.endDate = endDate
                     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):
                     if 'datatype' in kwargs:
                         datatype = kwargs.pop('datatype')
                         if 'Reader' in datatype:
                             self.name = datatype
                         else:
                             self.name = '%sReader' % (datatype)
                         self.datatype = self.name.replace('Reader', '')
                     attrs = ('path', 'startDate', 'endDate',
                              'startTime', 'endTime')
                     for attr in attrs:
                         if attr in kwargs:
                             setattr(self, attr, kwargs.pop(attr))
                     self.updateRunOperation(**kwargs)
                 def removeOperations(self):
                     for obj in self.opConfObjList:
                         del obj
                     self.opConfObjList = []
                 def addRunOperation(self, **kwargs):
                     opObj = self.addOperation(name='run', optype='self')
                     if self.server is None:
                         opObj.addParameter(
                             name='datatype', value=self.datatype, format='str')
                         opObj.addParameter(name='path', value=self.path, format='str')
                         opObj.addParameter(
                             name='startDate', value=self.startDate, format='date')
                         opObj.addParameter(
                             name='endDate', value=self.endDate, format='date')
                         opObj.addParameter(
                             name='startTime', value=self.startTime, format='time')
                         opObj.addParameter(
                             name='endTime', value=self.endTime, format='time')
                         for key, value in list(kwargs.items()):
                             opObj.addParameter(name=key, value=value,
                                                format=type(value).__name__)
                     else:
                         opObj.addParameter(name='server', value=self.server, format='str')
                     return opObj
                 def updateRunOperation(self, **kwargs):
                     opObj = self.getOperationObj(name='run')
                     opObj.removeParameters()
                     opObj.addParameter(name='datatype', value=self.datatype, format='str')
                     opObj.addParameter(name='path', value=self.path, format='str')
                     opObj.addParameter(
                         name='startDate', value=self.startDate, format='date')
                     opObj.addParameter(name='endDate', value=self.endDate, format='date')
                     opObj.addParameter(
                         name='startTime', value=self.startTime, format='time')
                     opObj.addParameter(name='endTime', value=self.endTime, format='time')
                     for key, value in list(kwargs.items()):
                         opObj.addParameter(name=key, value=value,
                                            format=type(value).__name__)
                     return opObj
                 def readXml(self, upElement, project_id):
                     self.id = upElement.get('id')
                     self.name = upElement.get('name')
                     self.datatype = upElement.get('datatype')
                     self.project_id = str(project_id)  #yong
                     if self.ELEMENTNAME == 'ReadUnit':
                         self.datatype = self.datatype.replace('Reader', '')
                     self.opConfObjList = []
                     opElementList = upElement.iter(OperationConf().getElementName())
                     for opElement in opElementList:
                         opConfObj = OperationConf()
                         opConfObj.readXml(opElement, project_id)
                         self.opConfObjList.append(opConfObj)
                         if opConfObj.name == 'run':
                             self.path = opConfObj.getParameterValue('path')
                             self.startDate = opConfObj.getParameterValue('startDate')
                             self.endDate = opConfObj.getParameterValue('endDate')
                             self.startTime = opConfObj.getParameterValue('startTime')
                             self.endTime = opConfObj.getParameterValue('endTime')
             class Project(Process):
                 ELEMENTNAME = 'Project'
                 def __init__(self):
                     Process.__init__(self)
                     self.id = None
                     self.filename = None
                     self.description = None
                     self.email = None
                     self.alarm = None
                     self.procUnitConfObjDict = {}
                     self.err_queue = Queue()
                 def __getNewId(self):
                     idList = list(self.procUnitConfObjDict.keys())
                     id = int(self.id) * 10
                     while True:
                         id += 1
                         if str(id) in idList:
                             continue
                         break
                     return str(id)
                 def getElementName(self):
                     return self.ELEMENTNAME
                 def getId(self):
                     return self.id
                 def updateId(self, new_id):
                     self.id = str(new_id)
                     keyList = list(self.procUnitConfObjDict.keys())
                     keyList.sort()
                     n = 1
                     newProcUnitConfObjDict = {}
                     for procKey in keyList:
                         procUnitConfObj = self.procUnitConfObjDict[procKey]
                         idProcUnit = str(int(self.id) * 10 + n)
                         procUnitConfObj.updateId(idProcUnit)
                         newProcUnitConfObjDict[idProcUnit] = procUnitConfObj
                         n += 1
                     self.procUnitConfObjDict = newProcUnitConfObjDict
                 def setup(self, id=1, name='', description='', email=None, alarm=[]):
                     print(' ')
                     print('*' * 60)
                     print('*  Starting SIGNAL CHAIN PROCESSING (Multiprocessing) v%s *' % schainpy.__version__)
                     print('*' * 60)
                     print("*  Python " + python_version() + "  *")
                     print('*' * 19)
                     print(' ')
                     self.id = str(id)
                     self.description = description
                     self.email = email
                     self.alarm = alarm
                     if name:
                         self.name = '{} ({})'.format(Process.__name__, name)
                 def update(self, **kwargs):
                     for key, value in list(kwargs.items()):
                         setattr(self, key, value)
                 def clone(self):
                     p = Project()
                     p.procUnitConfObjDict = self.procUnitConfObjDict
                     return p
                 def addReadUnit(self, id=None, datatype=None, name=None, **kwargs):
                     '''
                     Actualizacion:
                         Se agrego un nuevo argumento: topic -relativo a la forma de comunicar los procesos simultaneos
                         * El id del proceso sera el topico al que se deben subscribir los procUnits para recibir la informacion(data)
                     '''
                     if id is None:
                         idReadUnit = self.__getNewId()
                     else:
                         idReadUnit = str(id)
                     readUnitConfObj = ReadUnitConf()
                     readUnitConfObj.setup(self.id, idReadUnit, name, datatype, self.err_queue, **kwargs)
                     self.procUnitConfObjDict[readUnitConfObj.getId()] = readUnitConfObj
                     return readUnitConfObj
                 def addProcUnit(self, inputId='0', datatype=None, name=None):
                     '''
                     Actualizacion:
                         Se agrego dos nuevos argumentos: topic_read (lee data de otro procUnit) y topic_write(escribe o envia data a otro procUnit)
                         Deberia reemplazar a "inputId"
                         ** A fin de mantener el inputID, este sera la representaacion del topicoal que deben subscribirse. El ID propio de la intancia
                         (proceso) sera el topico de la publicacion, todo sera asignado de manera dinamica.
                     '''
                     idProcUnit = self.__getNewId()
                     procUnitConfObj = ProcUnitConf()
-                    procUnitConfObj.setup(self.id, idProcUnit, name, datatype, inputId, self.err_queue)
+                    input_proc = self.procUnitConfObjDict[inputId]
+                    procUnitConfObj.setup(self.id, idProcUnit, name, datatype, inputId, self.err_queue, input_proc.mylock)
                     self.procUnitConfObjDict[procUnitConfObj.getId()] = procUnitConfObj
                     return procUnitConfObj
                 def removeProcUnit(self, id):
                     if id in list(self.procUnitConfObjDict.keys()):
                         self.procUnitConfObjDict.pop(id)
                 def getReadUnitId(self):
                     readUnitConfObj = self.getReadUnitObj()
                     return readUnitConfObj.id
                 def getReadUnitObj(self):
                     for obj in list(self.procUnitConfObjDict.values()):
                         if obj.getElementName() == 'ReadUnit':
                             return obj
                     return None
                 def getProcUnitObj(self, id=None, name=None):
                     if id != None:
                         return self.procUnitConfObjDict[id]
                     if name != None:
                         return self.getProcUnitObjByName(name)
                     return None
                 def getProcUnitObjByName(self, name):
                     for obj in list(self.procUnitConfObjDict.values()):
                         if obj.name == name:
                             return obj
                     return None
                 def procUnitItems(self):
                     return list(self.procUnitConfObjDict.items())
                 def makeXml(self):
                     projectElement = Element('Project')
                     projectElement.set('id', str(self.id))
                     projectElement.set('name', self.name)
                     projectElement.set('description', self.description)
                     for procUnitConfObj in list(self.procUnitConfObjDict.values()):
                         procUnitConfObj.makeXml(projectElement)
                     self.projectElement = projectElement
                 def writeXml(self, filename=None):
                     if filename == None:
                         if self.filename:
                             filename = self.filename
                         else:
                             filename = 'schain.xml'
                     if not filename:
                         print('filename has not been defined. Use setFilename(filename) for do it.')
                         return 0
                     abs_file = os.path.abspath(filename)
                     if not os.access(os.path.dirname(abs_file), os.W_OK):
                         print('No write permission on %s' % os.path.dirname(abs_file))
                         return 0
                     if os.path.isfile(abs_file) and not(os.access(abs_file, os.W_OK)):
                         print('File %s already exists and it could not be overwriten' % abs_file)
                         return 0
                     self.makeXml()
                     ElementTree(self.projectElement).write(abs_file, method='xml')
                     self.filename = abs_file
                     return 1
                 def readXml(self, filename=None):
                     if not filename:
                         print('filename is not defined')
                         return 0
                     abs_file = os.path.abspath(filename)
                     if not os.path.isfile(abs_file):
                         print('%s file does not exist' % abs_file)
                         return 0
                     self.projectElement = None
                     self.procUnitConfObjDict = {}
                     try:
                         self.projectElement = ElementTree().parse(abs_file)
                     except:
                         print('Error reading %s, verify file format' % filename)
                         return 0
                     self.project = self.projectElement.tag
                     self.id = self.projectElement.get('id')
                     self.name = self.projectElement.get('name')
                     self.description = self.projectElement.get('description')
                     readUnitElementList = self.projectElement.iter(
                         ReadUnitConf().getElementName())
                     for readUnitElement in readUnitElementList:
                         readUnitConfObj = ReadUnitConf()
                         readUnitConfObj.readXml(readUnitElement, self.id)
                         self.procUnitConfObjDict[readUnitConfObj.getId()] = readUnitConfObj
                     procUnitElementList = self.projectElement.iter(
                         ProcUnitConf().getElementName())
                     for procUnitElement in procUnitElementList:
                         procUnitConfObj = ProcUnitConf()
                         procUnitConfObj.readXml(procUnitElement, self.id)
                         self.procUnitConfObjDict[procUnitConfObj.getId()] = procUnitConfObj
                     self.filename = abs_file
                     return 1
                 def __str__(self):
                     print('Project: name = %s, description = %s, id = %s' % (
                                                                         self.name,
                                                                         self.description,
                                                                         self.id))
                     for procUnitConfObj in self.procUnitConfObjDict.values():
                         print(procUnitConfObj)
                 def createObjects(self):
                     keys = list(self.procUnitConfObjDict.keys())
                     keys.sort()
                     for key in keys:
                         self.procUnitConfObjDict[key].createObjects()
                 def monitor(self):
                     t = Thread(target=self.__monitor, args=(self.err_queue, self.ctx))
                     t.start()
                 def __monitor(self, queue, ctx):
                     import socket
                     procs = 0
                     err_msg = ''
                     while True:
                         msg = queue.get()
                         if '#_start_#' in msg:
                             procs += 1
                         elif '#_end_#' in msg:
                             procs -=1
                         else:
                             err_msg = msg
                         if procs == 0 or 'Traceback' in err_msg:
                             break
                         time.sleep(0.1)
                     if '|' in err_msg:
                         name, err = err_msg.split('|')
                         if 'SchainWarning' in err:
                             log.warning(err.split('SchainWarning:')[-1].split('\n')[0].strip(), name)
                         elif 'SchainError' in err:
                             log.error(err.split('SchainError:')[-1].split('\n')[0].strip(), name)
                         else:
                             log.error(err, name)
                     else:
                         name, err = self.name, err_msg
                     time.sleep(2)
                     for conf in self.procUnitConfObjDict.values():
                         for confop in conf.opConfObjList:
                             if confop.type == 'external':
                                 confop.opObj.terminate()
                         conf.procUnitObj.terminate()
                     ctx.term()
                     message = ''.join(err)
                     if err_msg:
                         subject = 'SChain v%s: Error running %s\n' % (
                             schainpy.__version__, self.name)
                         subtitle = 'Hostname: %s\n' % socket.gethostbyname(
                             socket.gethostname())
                         subtitle += 'Working directory: %s\n' % os.path.abspath('./')
                         subtitle += 'Configuration file: %s\n' % self.filename
                         subtitle += 'Time: %s\n' % str(datetime.datetime.now())
                         readUnitConfObj = self.getReadUnitObj()
                         if readUnitConfObj:
                             subtitle += '\nInput parameters:\n'
                             subtitle += '[Data path = %s]\n' % readUnitConfObj.path
                             subtitle += '[Data type = %s]\n' % readUnitConfObj.datatype
                             subtitle += '[Start date = %s]\n' % readUnitConfObj.startDate
                             subtitle += '[End date = %s]\n' % readUnitConfObj.endDate
                             subtitle += '[Start time = %s]\n' % readUnitConfObj.startTime
                             subtitle += '[End time = %s]\n' % readUnitConfObj.endTime
                         a = Alarm(
                             modes=self.alarm,
                             email=self.email,
                             message=message,
                             subject=subject,
                             subtitle=subtitle,
                             filename=self.filename
                         )
                         a.start()
                 def isPaused(self):
                     return 0
                 def isStopped(self):
                     return 0
                 def runController(self):
                     '''
                     returns 0 when this process has been stopped, 1 otherwise
                     '''
                     if self.isPaused():
                         print('Process suspended')
                         while True:
                             time.sleep(0.1)
                             if not self.isPaused():
                                 break
                             if self.isStopped():
                                 break
                         print('Process reinitialized')
                     if self.isStopped():
                         print('Process stopped')
                         return 0
                     return 1
                 def setFilename(self, filename):
                     self.filename = filename
                 def setProxy(self):
                     if not os.path.exists('/tmp/schain'):
                         os.mkdir('/tmp/schain')
                     self.ctx = zmq.Context()
                     xpub = self.ctx.socket(zmq.XPUB)
                     xpub.bind('ipc:///tmp/schain/{}_pub'.format(self.id))
                     xsub = self.ctx.socket(zmq.XSUB)
                     xsub.bind('ipc:///tmp/schain/{}_sub'.format(self.id))
                     self.monitor()
                     try:
                         zmq.proxy(xpub, xsub)
                     except zmq.ContextTerminated:
                         xpub.close()
                         xsub.close()
                 def run(self):
                     log.success('Starting {}: {}'.format(self.name, self.id), tag='')
                     self.start_time = time.time()
                     self.createObjects()
                     self.setProxy()
                     log.success('{} Done (Time: {}s)'.format(
                         self.name,
                         time.time()-self.start_time), '')

schainpy/model/proc/jroproc_base.py +19 -18

             '''
             Updated for multiprocessing
             Author : Sergio Cortez
             Jan 2018
             Abstract:
                 Base class for processing units and operations. A decorator provides multiprocessing features and interconnect the processes created.
                 The argument (kwargs) sent from the controller is parsed and filtered via the decorator for each processing unit or operation instantiated.
                 The decorator handle also the methods inside the processing unit to be called from the main script (not as operations) (OPERATION -> type ='self').
             Based on:
                 $Author: murco $
                 $Id: jroproc_base.py 1 2012-11-12 18:56:07Z murco $
             '''
             import os
+            import sys
             import inspect
             import zmq
             import time
             import pickle
             import traceback
             try:
                 from queue import Queue
             except:
                 from Queue import Queue
             from threading import Thread
             from multiprocessing import Process
             from schainpy.utils import log
             class ProcessingUnit(object):
                 """
                 Update - Jan 2018 - MULTIPROCESSING
                 All the "call" methods present in the previous base were removed.
                 The majority of operations are independant processes, thus
                 the decorator is in charge of communicate the operation processes
                 with the proccessing unit via IPC.
                 The constructor does not receive any argument. The remaining methods
                 are related with the operations to execute.
                 """
                 proc_type = 'processing'
                 __attrs__ = []
                 def __init__(self):
                     self.dataIn = None
                     self.dataOut = None
                     self.isConfig = False
                     self.operations = []
                     self.plots = []
                 def getAllowedArgs(self):
                     if hasattr(self, '__attrs__'):
                         return self.__attrs__
                     else:
                         return inspect.getargspec(self.run).args
                 def addOperation(self, conf, operation):
                     """
                     This method is used in the controller, and update the dictionary containing the operations to execute. The dict
                     posses the id of the operation process (IPC purposes)
                         Agrega un objeto del tipo "Operation" (opObj) a la lista de objetos "self.objectList" y retorna el
                         identificador asociado a este objeto.
                         Input:
                             object    :    objeto de la clase "Operation"
                         Return:
                             objId    :    identificador del objeto, necesario para comunicar con master(procUnit)
                     """
                     self.operations.append(
                         (operation, conf.type, conf.id, conf.getKwargs()))
                     if 'plot' in self.name.lower():
                         self.plots.append(operation.CODE)
                 def getOperationObj(self, objId):
                     if objId not in list(self.operations.keys()):
                         return None
                     return self.operations[objId]
                 def operation(self, **kwargs):
                     """
                     Operacion directa sobre la data (dataOut.data). Es necesario actualizar los valores de los
                     atributos del objeto dataOut
                     Input:
                         **kwargs    :    Diccionario de argumentos de la funcion a ejecutar
                     """
                     raise NotImplementedError
                 def setup(self):
                     raise NotImplementedError
                 def run(self):
                     raise NotImplementedError
                 def close(self):
                     return
             class Operation(object):
                 """
                 Update - Jan 2018 - MULTIPROCESSING
                 Most of the methods remained the same. The decorator parse the arguments and executed the run() method for each process.
                 The constructor doe snot receive any argument, neither the baseclass.
                     Clase base para definir las operaciones adicionales que se pueden agregar a la clase ProcessingUnit
                     y necesiten acumular informacion previa de los datos a procesar. De preferencia usar un buffer de
                     acumulacion dentro de esta clase
                     Ejemplo: Integraciones coherentes, necesita la informacion previa de los n perfiles anteriores (bufffer)
                 """
                 proc_type = 'operation'
                 __attrs__ = []
                 def __init__(self):
                     self.id = None
                     self.isConfig = False
                     if not hasattr(self, 'name'):
                         self.name = self.__class__.__name__
                 def getAllowedArgs(self):
                     if hasattr(self, '__attrs__'):
                         return self.__attrs__
                     else:
                         return inspect.getargspec(self.run).args
                 def setup(self):
                     self.isConfig = True
                     raise NotImplementedError
                 def run(self, dataIn, **kwargs):
                     """
                     Realiza las operaciones necesarias sobre la dataIn.data y actualiza los
                     atributos del objeto dataIn.
                     Input:
                         dataIn    :    objeto del tipo JROData
                     Return:
                         None
                     Affected:
                         __buffer    :    buffer de recepcion de datos.
                     """
                     if not self.isConfig:
                         self.setup(**kwargs)
                     raise NotImplementedError
                 def close(self):
                     return
             class InputQueue(Thread):
                 '''
                 Class to hold input data for Proccessing Units and external Operations,
                 '''
-            	    def __init__(self, project_id, inputId):
+                def __init__(self, project_id, inputId, lock=None):
                     Thread.__init__(self)
                     self.queue = Queue()
                     self.project_id = project_id
                     self.inputId = inputId
+                    self.lock = lock
+                    self.size = 0
                 def run(self):
                     c = zmq.Context()
                     self.receiver = c.socket(zmq.SUB)
                     self.receiver.connect(
                         'ipc:///tmp/schain/{}_pub'.format(self.project_id))
                     self.receiver.setsockopt(zmq.SUBSCRIBE, self.inputId.encode())
                     while True:
-            	            self.queue.put(self.receiver.recv_multipart()[1])
+                        obj = self.receiver.recv_multipart()[1]
+                        self.size += sys.getsizeof(obj)
+                        self.queue.put(obj)
                 def get(self):
+                    if self.size/1000000 > 2048:
-            	        return pickle.loads(self.queue.get())
+                        self.lock.clear()
+                    else:
+                        self.lock.set()
+                    obj = self.queue.get()
+                    self.size -= sys.getsizeof(obj)
+                    return pickle.loads(obj)
             def MPDecorator(BaseClass):
                 """
                 Multiprocessing class decorator
                 This function add multiprocessing features to a BaseClass. Also, it handle
                 the communication beetween processes (readers, procUnits and operations).
                 """
                 class MPClass(BaseClass, Process):
                     def __init__(self, *args, **kwargs):
                         super(MPClass, self).__init__()
                         Process.__init__(self)
                         self.operationKwargs = {}
                         self.args = args
                         self.kwargs = kwargs
                         self.sender = None
                         self.receiver = None
                         self.i = 0
                         self.t = time.time()
                         self.name = BaseClass.__name__
                         self.__doc__ = BaseClass.__doc__
                         if 'plot' in self.name.lower() and not self.name.endswith('_'):
                             self.name = '{}{}'.format(self.CODE.upper(), 'Plot')
                         self.start_time = time.time()
                         self.id = args[0]
                         self.inputId = args[1]
                         self.project_id = args[2]
                         self.err_queue = args[3]
-                        self.typeProc = args[4]
+                        self.lock = args[4]
+                        self.typeProc = args[5]
                         self.err_queue.put('#_start_#')
-                        self.queue = InputQueue(self.project_id, self.inputId)
+                        self.queue = InputQueue(self.project_id, self.inputId, self.lock)
                     def subscribe(self):
                         '''
                         Start the zmq socket receiver and subcribe to input ID.
                         '''
                         self.queue.start()
                     def listen(self):
                         '''
                         This function waits for objects
                         '''
                         return self.queue.get()
                     def set_publisher(self):
                         '''
                         This function create a zmq socket for publishing objects.
                         '''
                         time.sleep(0.5)
                         c = zmq.Context()
                         self.sender = c.socket(zmq.PUB)
                         self.sender.connect(
                             'ipc:///tmp/schain/{}_sub'.format(self.project_id))
                     def publish(self, data, id):
                         '''
                         This function publish an object, to an specific topic.
-                        For Read Units (inputId == None) adds a little delay
+                        It blocks publishing when receiver queue is full to avoid data loss
-                        to avoid data loss
                         '''
                         if self.inputId is None:
-                            self.i += 1
+                            self.lock.wait()
-                            if self.i % 40 == 0 and time.time()-self.t > 0.1:
-                                self.i = 0
-                                self.t = time.time()
-                                time.sleep(0.05)
-                            elif self.i % 40 == 0:
-                                self.i = 0
-                                self.t = time.time()
-                                time.sleep(0.01)
                         self.sender.send_multipart([str(id).encode(), pickle.dumps(data)])
                     def runReader(self):
                         '''
                         Run fuction for read units
                         '''
                         while True:
                             try:
                                 BaseClass.run(self, **self.kwargs)
                             except:
                                 err = traceback.format_exc()
                                 if 'No more files' in err:
                                     log.warning('No more files to read', self.name)
                                 else:
                                     self.err_queue.put('{}|{}'.format(self.name, err))
                                 self.dataOut.error = True
                             for op, optype, opId, kwargs in self.operations:
                                 if optype == 'self' and not self.dataOut.flagNoData:
                                     op(**kwargs)
                                 elif optype == 'other' and not self.dataOut.flagNoData:
                                     self.dataOut = op.run(self.dataOut, **self.kwargs)
                                 elif optype == 'external':
                                     self.publish(self.dataOut, opId)
                             if self.dataOut.flagNoData and not self.dataOut.error:
                                 continue
                             self.publish(self.dataOut, self.id)
                             if self.dataOut.error:
                                 break
                         time.sleep(0.5)
                     def runProc(self):
                         '''
                         Run function for proccessing units
                         '''
                         while True:
                             self.dataIn = self.listen()
                             if self.dataIn.flagNoData and self.dataIn.error is None:
                                 continue
                             elif not self.dataIn.error:
                                 try:
                                     BaseClass.run(self, **self.kwargs)
                                 except:
                                     self.err_queue.put('{}|{}'.format(self.name, traceback.format_exc()))
                                     self.dataOut.error = True
                             elif self.dataIn.error:
                                 self.dataOut.error = self.dataIn.error
                                 self.dataOut.flagNoData = True
                             for op, optype, opId, kwargs in self.operations:
                                 if optype == 'self' and not self.dataOut.flagNoData:
                                     op(**kwargs)
                                 elif optype == 'other' and not self.dataOut.flagNoData:
                                     self.dataOut = op.run(self.dataOut, **kwargs)
                                 elif optype == 'external' and not self.dataOut.flagNoData:
                                     self.publish(self.dataOut, opId)
                             self.publish(self.dataOut, self.id)
                             for op, optype, opId, kwargs in self.operations:
                                 if optype == 'external' and self.dataOut.error:
                                     self.publish(self.dataOut, opId)
                             if self.dataOut.error:
                                 break
                         time.sleep(0.5)
                     def runOp(self):
                         '''
                         Run function for external operations (this operations just receive data
                         ex: plots, writers, publishers)
                         '''
                         while True:
                             dataOut = self.listen()
                             if not dataOut.error:
                                 BaseClass.run(self, dataOut, **self.kwargs)
                             else:
                                 break
                     def run(self):
                         if self.typeProc is "ProcUnit":
                             if self.inputId is not None:
                                 self.subscribe()
                             self.set_publisher()
                             if 'Reader' not in BaseClass.__name__:
                                 self.runProc()
                             else:
                                 self.runReader()
                         elif self.typeProc is "Operation":
                             self.subscribe()
                             self.runOp()
                         else:
                             raise ValueError("Unknown type")
                         self.close()
                     def close(self):
                         BaseClass.close(self)
                         self.err_queue.put('#_end_#')
                         if self.sender:
                             self.sender.close()
                         if self.receiver:
                             self.receiver.close()
                         log.success('Done...(Time:{:4.2f} secs)'.format(time.time()-self.start_time), self.name)
                 return MPClass

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