jroproc_base.py
426 lines
| 12.3 KiB
| text/x-python
|
PythonLexer
Daniel Valdez
|
r487 | ''' | ||
George Yong
|
r1171 | 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. | ||||
| r1279 | 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'). | ||||
|
George Yong
|
r1171 | |||
| Based on: | ||||
| $Author: murco $ | ||||
| $Id: jroproc_base.py 1 2012-11-12 18:56:07Z murco $ | ||||
|
Daniel Valdez
|
r487 | ''' | ||
Juan C. Espinoza
|
r1187 | |||
|
Juan C. Espinoza
|
r1212 | import os | ||
| r1256 | import sys | |||
José Chávez
|
r929 | import inspect | ||
|
George Yong
|
r1171 | import zmq | ||
| import time | ||||
| import pickle | ||||
| r1241 | import traceback | |||
| r1252 | try: | |||
| from queue import Queue | ||||
| except: | ||||
| from Queue import Queue | ||||
Juan C. Espinoza
|
r1235 | from threading import Thread | ||
|
George Yong
|
r1171 | from multiprocessing import Process | ||
|
Juan C. Espinoza
|
r1187 | |||
|
George Yong
|
r1171 | from schainpy.utils import log | ||
|
José Chávez
|
r929 | |||
|
Daniel Valdez
|
r487 | |||
Miguel Valdez
|
r568 | class ProcessingUnit(object): | ||
|
Juan C. Valdez
|
r860 | |||
|
Daniel Valdez
|
r487 | """ | ||
|
George Yong
|
r1171 | Update - Jan 2018 - MULTIPROCESSING | ||
| r1279 | All the "call" methods present in the previous base were removed. | |||
|
George Yong
|
r1171 | The majority of operations are independant processes, thus | ||
| r1279 | the decorator is in charge of communicate the operation processes | |||
|
George Yong
|
r1171 | with the proccessing unit via IPC. | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | The constructor does not receive any argument. The remaining methods | ||
| are related with the operations to execute. | ||||
| r1279 | ||||
|
Juan C. Valdez
|
r860 | |||
|
Daniel Valdez
|
r487 | """ | ||
| r1254 | proc_type = 'processing' | |||
| __attrs__ = [] | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | def __init__(self): | ||
|
Juan C. Valdez
|
r860 | |||
|
Daniel Valdez
|
r487 | self.dataIn = None | ||
|
Miguel Valdez
|
r573 | self.dataOut = None | ||
|
Daniel Valdez
|
r487 | self.isConfig = False | ||
|
Juan C. Espinoza
|
r1177 | self.operations = [] | ||
|
Juan C. Espinoza
|
r1187 | self.plots = [] | ||
|
Juan C. Valdez
|
r860 | |||
|
José Chávez
|
r929 | def getAllowedArgs(self): | ||
|
Juan C. Espinoza
|
r1097 | if hasattr(self, '__attrs__'): | ||
| return self.__attrs__ | ||||
| else: | ||||
| return inspect.getargspec(self.run).args | ||||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1187 | def addOperation(self, conf, operation): | ||
|
Daniel Valdez
|
r487 | """ | ||
| r1279 | This method is used in the controller, and update the dictionary containing the operations to execute. The dict | |||
|
George Yong
|
r1171 | posses the id of the operation process (IPC purposes) | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | Agrega un objeto del tipo "Operation" (opObj) a la lista de objetos "self.objectList" y retorna el | ||
| identificador asociado a este objeto. | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | Input: | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | object : objeto de la clase "Operation" | ||
| Return: | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | objId : identificador del objeto, necesario para comunicar con master(procUnit) | ||
|
Daniel Valdez
|
r487 | """ | ||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1187 | self.operations.append( | ||
| (operation, conf.type, conf.id, conf.getKwargs())) | ||||
| r1279 | ||||
|
Juan C. Espinoza
|
r1187 | if 'plot' in self.name.lower(): | ||
| self.plots.append(operation.CODE) | ||||
|
George Yong
|
r1171 | |||
|
Miguel Valdez
|
r577 | def getOperationObj(self, objId): | ||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1177 | if objId not in list(self.operations.keys()): | ||
|
Miguel Valdez
|
r577 | return None | ||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1177 | return self.operations[objId] | ||
|
Juan C. Valdez
|
r860 | |||
|
Daniel Valdez
|
r487 | def operation(self, **kwargs): | ||
| """ | ||||
| Operacion directa sobre la data (dataOut.data). Es necesario actualizar los valores de los | ||||
| atributos del objeto dataOut | ||||
|
Juan C. Valdez
|
r860 | |||
|
Daniel Valdez
|
r487 | Input: | ||
|
Juan C. Valdez
|
r860 | |||
|
Daniel Valdez
|
r487 | **kwargs : Diccionario de argumentos de la funcion a ejecutar | ||
| """ | ||||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1187 | raise NotImplementedError | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | def setup(self): | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | raise NotImplementedError | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | def run(self): | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | raise NotImplementedError | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | def close(self): | ||
|
Juan C. Espinoza
|
r1187 | |||
|
George Yong
|
r1171 | return | ||
|
Juan C. Espinoza
|
r1187 | |||
|
George Yong
|
r1171 | class Operation(object): | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | """ | ||
| Update - Jan 2018 - MULTIPROCESSING | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | 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. | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | 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 | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | Ejemplo: Integraciones coherentes, necesita la informacion previa de los n perfiles anteriores (bufffer) | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | """ | ||
| r1254 | proc_type = 'operation' | |||
| __attrs__ = [] | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | def __init__(self): | ||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1187 | self.id = None | ||
|
George Yong
|
r1171 | self.isConfig = False | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | if not hasattr(self, 'name'): | ||
| self.name = self.__class__.__name__ | ||||
|
Juan C. Espinoza
|
r1187 | |||
|
George Yong
|
r1171 | def getAllowedArgs(self): | ||
| if hasattr(self, '__attrs__'): | ||||
| return self.__attrs__ | ||||
| else: | ||||
| return inspect.getargspec(self.run).args | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | def setup(self): | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | self.isConfig = True | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | raise NotImplementedError | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | def run(self, dataIn, **kwargs): | ||
|
Daniel Valdez
|
r487 | """ | ||
|
George Yong
|
r1171 | Realiza las operaciones necesarias sobre la dataIn.data y actualiza los | ||
| atributos del objeto dataIn. | ||||
|
Juan C. Valdez
|
r860 | |||
|
Miguel Valdez
|
r568 | Input: | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | dataIn : objeto del tipo JROData | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | Return: | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | None | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | Affected: | ||
| __buffer : buffer de recepcion de datos. | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | """ | ||
| if not self.isConfig: | ||||
| self.setup(**kwargs) | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | raise NotImplementedError | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | def close(self): | ||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1187 | return | ||
|
Juan C. Valdez
|
r860 | |||
| r1250 | class InputQueue(Thread): | |||
| r1279 | ||||
| r1256 | ''' | |||
| Class to hold input data for Proccessing Units and external Operations, | ||||
| ''' | ||||
| def __init__(self, project_id, inputId, lock=None): | ||||
| r1254 | ||||
| r1256 | Thread.__init__(self) | |||
| self.queue = Queue() | ||||
| self.project_id = project_id | ||||
| self.inputId = inputId | ||||
| self.lock = lock | ||||
| r1268 | self.islocked = False | |||
| r1256 | 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: | ||||
| obj = self.receiver.recv_multipart()[1] | ||||
| self.size += sys.getsizeof(obj) | ||||
| self.queue.put(obj) | ||||
| r1268 | def get(self): | |||
| if not self.islocked and self.size/1000000 > 512: | ||||
| r1279 | self.lock.n.value += 1 | |||
| r1268 | self.islocked = True | |||
| r1256 | self.lock.clear() | |||
| r1268 | elif self.islocked and self.size/1000000 <= 512: | |||
| self.islocked = False | ||||
| self.lock.n.value -= 1 | ||||
| if self.lock.n.value == 0: | ||||
| r1279 | self.lock.set() | |||
| r1256 | obj = self.queue.get() | |||
| r1268 | self.size -= sys.getsizeof(obj) | |||
| r1256 | return pickle.loads(obj) | |||
| r1254 | ||||
| r1279 | ||||
|
George Yong
|
r1171 | def MPDecorator(BaseClass): | ||
| """ | ||||
|
Juan C. Espinoza
|
r1177 | Multiprocessing class decorator | ||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1177 | This function add multiprocessing features to a BaseClass. Also, it handle | ||
| r1279 | the communication beetween processes (readers, procUnits and operations). | |||
|
George Yong
|
r1171 | """ | ||
|
Juan C. Espinoza
|
r1187 | |||
|
George Yong
|
r1171 | class MPClass(BaseClass, Process): | ||
|
Juan C. Espinoza
|
r1177 | |||
|
George Yong
|
r1171 | def __init__(self, *args, **kwargs): | ||
| super(MPClass, self).__init__() | ||||
| Process.__init__(self) | ||||
| self.operationKwargs = {} | ||||
| self.args = args | ||||
| self.kwargs = kwargs | ||||
|
Juan C. Espinoza
|
r1177 | self.sender = None | ||
| self.receiver = None | ||||
|
Juan C. Espinoza
|
r1235 | self.i = 0 | ||
| r1245 | self.t = time.time() | |||
|
Juan C. Espinoza
|
r1177 | self.name = BaseClass.__name__ | ||
| r1254 | self.__doc__ = BaseClass.__doc__ | |||
| r1279 | ||||
|
George Yong
|
r1201 | if 'plot' in self.name.lower() and not self.name.endswith('_'): | ||
| self.name = '{}{}'.format(self.CODE.upper(), 'Plot') | ||||
| r1279 | ||||
| self.start_time = time.time() | ||||
| r1241 | self.id = args[0] | |||
| self.inputId = args[1] | ||||
| self.project_id = args[2] | ||||
| self.err_queue = args[3] | ||||
| r1256 | self.lock = args[4] | |||
| self.typeProc = args[5] | ||||
| r1241 | self.err_queue.put('#_start_#') | |||
| r1268 | if self.inputId is not None: | |||
| self.queue = InputQueue(self.project_id, self.inputId, self.lock) | ||||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1177 | def subscribe(self): | ||
| ''' | ||||
| r1245 | Start the zmq socket receiver and subcribe to input ID. | |||
|
Juan C. Espinoza
|
r1177 | ''' | ||
| r1250 | ||||
| self.queue.start() | ||||
| r1279 | ||||
|
Juan C. Espinoza
|
r1177 | def listen(self): | ||
| ''' | ||||
|
Juan C. Espinoza
|
r1235 | This function waits for objects | ||
|
Juan C. Espinoza
|
r1177 | ''' | ||
| r1279 | ||||
| return self.queue.get() | ||||
|
George Yong
|
r1171 | |||
|
Juan C. Espinoza
|
r1177 | def set_publisher(self): | ||
| ''' | ||||
| r1279 | This function create a zmq socket for publishing objects. | |||
|
Juan C. Espinoza
|
r1177 | ''' | ||
|
George Yong
|
r1171 | |||
| r1245 | time.sleep(0.5) | |||
| r1279 | ||||
|
Juan C. Espinoza
|
r1177 | c = zmq.Context() | ||
|
Juan C. Espinoza
|
r1187 | self.sender = c.socket(zmq.PUB) | ||
| self.sender.connect( | ||||
| 'ipc:///tmp/schain/{}_sub'.format(self.project_id)) | ||||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1187 | def publish(self, data, id): | ||
|
Juan C. Espinoza
|
r1177 | ''' | ||
| r1245 | This function publish an object, to an specific topic. | |||
| r1256 | It blocks publishing when receiver queue is full to avoid data loss | |||
| r1279 | ''' | |||
Alexander Valdez
|
r1220 | if self.inputId is None: | ||
| r1256 | self.lock.wait() | |||
|
Juan C. Espinoza
|
r1177 | self.sender.send_multipart([str(id).encode(), pickle.dumps(data)]) | ||
|
George Yong
|
r1171 | def runReader(self): | ||
|
Juan C. Espinoza
|
r1177 | ''' | ||
| Run fuction for read units | ||||
| ''' | ||||
|
George Yong
|
r1171 | while True: | ||
|
Juan C. Valdez
|
r860 | |||
| r1241 | try: | |||
| BaseClass.run(self, **self.kwargs) | ||||
| except: | ||||
| r1279 | err = traceback.format_exc() | |||
| r1241 | if 'No more files' in err: | |||
| log.warning('No more files to read', self.name) | ||||
| else: | ||||
| self.err_queue.put('{}|{}'.format(self.name, err)) | ||||
| r1279 | self.dataOut.error = True | |||
|
Juan C. Espinoza
|
r1187 | for op, optype, opId, kwargs in self.operations: | ||
|
Juan C. Espinoza
|
r1198 | if optype == 'self' and not self.dataOut.flagNoData: | ||
|
Juan C. Espinoza
|
r1177 | op(**kwargs) | ||
|
Juan C. Espinoza
|
r1198 | elif optype == 'other' and not self.dataOut.flagNoData: | ||
|
Juan C. Espinoza
|
r1177 | self.dataOut = op.run(self.dataOut, **self.kwargs) | ||
|
Juan C. Espinoza
|
r1187 | elif optype == 'external': | ||
|
Juan C. Espinoza
|
r1177 | self.publish(self.dataOut, opId) | ||
|
Juan C. Espinoza
|
r1193 | if self.dataOut.flagNoData and not self.dataOut.error: | ||
|
George Yong
|
r1171 | continue | ||
|
Juan C. Espinoza
|
r1187 | |||
| self.publish(self.dataOut, self.id) | ||||
| r1279 | if self.dataOut.error: | |||
|
Juan C. Espinoza
|
r1187 | break | ||
| r1241 | time.sleep(0.5) | |||
|
Juan C. Espinoza
|
r1187 | |||
|
George Yong
|
r1171 | def runProc(self): | ||
|
Juan C. Espinoza
|
r1177 | ''' | ||
| Run function for proccessing units | ||||
| ''' | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | while True: | ||
| r1279 | self.dataIn = self.listen() | |||
|
George Yong
|
r1171 | |||
|
Juan C. Espinoza
|
r1187 | if self.dataIn.flagNoData and self.dataIn.error is None: | ||
|
Juan C. Espinoza
|
r1177 | continue | ||
| r1241 | 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: | ||||
|
Juan C. Espinoza
|
r1193 | self.dataOut.error = self.dataIn.error | ||
|
George Yong
|
r1206 | self.dataOut.flagNoData = True | ||
| r1279 | ||||
|
Juan C. Espinoza
|
r1177 | for op, optype, opId, kwargs in self.operations: | ||
|
Juan C. Espinoza
|
r1198 | if optype == 'self' and not self.dataOut.flagNoData: | ||
|
Juan C. Espinoza
|
r1177 | op(**kwargs) | ||
|
Juan C. Espinoza
|
r1198 | elif optype == 'other' and not self.dataOut.flagNoData: | ||
|
Juan C. Espinoza
|
r1177 | self.dataOut = op.run(self.dataOut, **kwargs) | ||
| r1279 | elif optype == 'external' and not self.dataOut.flagNoData: | |||
|
George Yong
|
r1206 | self.publish(self.dataOut, opId) | ||
| r1279 | ||||
| r1241 | self.publish(self.dataOut, self.id) | |||
| for op, optype, opId, kwargs in self.operations: | ||||
| r1279 | if optype == 'external' and self.dataOut.error: | |||
| r1241 | self.publish(self.dataOut, opId) | |||
| r1279 | ||||
| r1241 | if self.dataOut.error: | |||
|
Juan C. Espinoza
|
r1187 | break | ||
| r1279 | ||||
| r1241 | time.sleep(0.5) | |||
|
George Yong
|
r1171 | |||
| def runOp(self): | ||||
|
Juan C. Espinoza
|
r1177 | ''' | ||
|
Juan C. Espinoza
|
r1187 | Run function for external operations (this operations just receive data | ||
| ex: plots, writers, publishers) | ||||
|
Juan C. Espinoza
|
r1177 | ''' | ||
| r1279 | ||||
|
George Yong
|
r1171 | while True: | ||
|
Juan C. Espinoza
|
r1177 | dataOut = self.listen() | ||
|
George Yong
|
r1179 | |||
| r1241 | if not dataOut.error: | |||
| r1268 | try: | |||
| BaseClass.run(self, dataOut, **self.kwargs) | ||||
| except: | ||||
| self.err_queue.put('{}|{}'.format(self.name, traceback.format_exc())) | ||||
| dataOut.error = True | ||||
| r1241 | else: | |||
| r1279 | break | |||
|
Juan C. Espinoza
|
r1187 | |||
|
George Yong
|
r1171 | def run(self): | ||
|
Juan C. Espinoza
|
r1177 | if self.typeProc is "ProcUnit": | ||
|
Juan C. Espinoza
|
r1187 | |||
|
Juan C. Espinoza
|
r1177 | if self.inputId is not None: | ||
| self.subscribe() | ||||
| r1279 | ||||
|
Juan C. Espinoza
|
r1177 | self.set_publisher() | ||
|
George Yong
|
r1171 | |||
|
Juan C. Espinoza
|
r1177 | if 'Reader' not in BaseClass.__name__: | ||
| self.runProc() | ||||
|
George Yong
|
r1171 | else: | ||
| self.runReader() | ||||
| elif self.typeProc is "Operation": | ||||
|
Juan C. Espinoza
|
r1187 | |||
|
Juan C. Espinoza
|
r1177 | self.subscribe() | ||
|
George Yong
|
r1171 | self.runOp() | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | else: | ||
| raise ValueError("Unknown type") | ||||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1177 | self.close() | ||
| def close(self): | ||||
|
Juan C. Espinoza
|
r1187 | |||
| BaseClass.close(self) | ||||
| r1241 | self.err_queue.put('#_end_#') | |||
|
Juan C. Espinoza
|
r1187 | |||
|
Juan C. Espinoza
|
r1177 | if self.sender: | ||
| self.sender.close() | ||||
|
Juan C. Espinoza
|
r1187 | |||
|
Juan C. Espinoza
|
r1177 | if self.receiver: | ||
| self.receiver.close() | ||||
|
Juan C. Espinoza
|
r1187 | log.success('Done...(Time:{:4.2f} secs)'.format(time.time()-self.start_time), self.name) | ||
| return MPClass | ||||
