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

             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__
                 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:
                         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:

schainpy/controller.py +13 -7

             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
                     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.type = type
                     self.priority = priority
                     self.err_queue = err_queue
+                    self.lock = lock
                     self.parmConfObjList = []
                 def removeParameters(self):
                         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
                     self.opConfObjList = []
                     self.procUnitObj = None
                     self.opObjDict = {}
+                    self.mylock = Event()
                 def __getPriority(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
                     self.datatype = datatype
                     self.inputId = inputId
                     self.err_queue = err_queue
+                    self.lock = lock
                     self.opConfObjList = []
                     self.addOperation(name='run', optype='self')
                     id = self.__getNewId()
                     priority = self.__getPriority() # Sin mucho sentido, pero puede usarse
                     opConfObj = OperationConf()
-                    opConfObj.setup(id, name=name, priority=priority, type=optype, project_id=self.project_id, err_queue=self.err_queue)
+                    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
                     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:
                     self.name = None
                     self.inputId = None
                     self.opConfObjList = []
+                    self.mylock = Event()
                 def getElementName(self):
                     self.endTime = endTime
                     self.server = server
                     self.err_queue = err_queue
+                    self.lock = self.mylock
                     self.addRunOperation(**kwargs)
                 def update(self, **kwargs):
                     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

schainpy/model/proc/jroproc_base.py +39 -38

@@ -13,6 +13,7 Based on:
13	'''	13	'''
14		14
15	import os	15	import os
		16	import sys
16	import inspect	17	import inspect
17	import zmq	18	import zmq
18	import time	19	import time
@@ -181,31 +182,40 class Operation(object):
181		182
182	class InputQueue(Thread):	183	class InputQueue(Thread):
183		184
184	'''	185	'''
185	Class to hold input data for Proccessing Units and external Operations,	186	Class to hold input data for Proccessing Units and external Operations,
186	'''	187	'''
187		188
188	def __init__(self, project_id, inputId):	189	def __init__(self, project_id, inputId, lock=None):
189
190	Thread.__init__(self)
191	self.queue = Queue()
192	self.project_id = project_id
193	self.inputId = inputId
194
195	def run(self):
196
197	c = zmq.Context()
198	self.receiver = c.socket(zmq.SUB)
199	self.receiver.connect(
200	'ipc:///tmp/schain/{}_pub'.format(self.project_id))
201	self.receiver.setsockopt(zmq.SUBSCRIBE, self.inputId.encode())
202
203	while True:
204	self.queue.put(self.receiver.recv_multipart()[1])
205
206	def get(self):
207		190
208	return pickle.loads(self.queue.get())	191	Thread.__init__(self)
		192	self.queue = Queue()
		193	self.project_id = project_id
		194	self.inputId = inputId
		195	self.lock = lock
		196	self.size = 0
		197
		198	def run(self):
		199
		200	c = zmq.Context()
		201	self.receiver = c.socket(zmq.SUB)
		202	self.receiver.connect(
		203	'ipc:///tmp/schain/{}_pub'.format(self.project_id))
		204	self.receiver.setsockopt(zmq.SUBSCRIBE, self.inputId.encode())
		205
		206	while True:
		207	obj = self.receiver.recv_multipart()[1]
		208	self.size += sys.getsizeof(obj)
		209	self.queue.put(obj)
		210
		211	def get(self):
		212	if self.size/1000000 > 2048:
		213	self.lock.clear()
		214	else:
		215	self.lock.set()
		216	obj = self.queue.get()
		217	self.size -= sys.getsizeof(obj)
		218	return pickle.loads(obj)
209		219
210		220
211	def MPDecorator(BaseClass):	221	def MPDecorator(BaseClass):
@@ -239,9 +249,10 def MPDecorator(BaseClass):
239	self.inputId = args[1]	249	self.inputId = args[1]
240	self.project_id = args[2]	250	self.project_id = args[2]
241	self.err_queue = args[3]	251	self.err_queue = args[3]
242	self.~~typeProc~~ = args[4]	252	self.lock = args[4]
		253	self.typeProc = args[5]
243	self.err_queue.put('#_start_#')	254	self.err_queue.put('#_start_#')
244	self.queue = InputQueue(self.project_id, self.inputId)	255	self.queue = InputQueue(self.project_id, self.inputId, self.lock)
245		256
246	def subscribe(self):	257	def subscribe(self):
247	'''	258	'''
@@ -272,21 +283,11 def MPDecorator(BaseClass):
272	def publish(self, data, id):	283	def publish(self, data, id):
273	'''	284	'''
274	This function publish an object, to an specific topic.	285	This function publish an object, to an specific topic.
275	For Read Units (inputId == None) adds a little delay	286	It blocks publishing when receiver queue is full to avoid data loss
276	to avoid data loss
277	'''	287	'''
278		288
279	if self.inputId is None:	289	if self.inputId is None:
280	self.i += 1	290	self.lock.wait()
281	if self.i % 40 == 0 and time.time()-self.t > 0.1:
282	self.i = 0
283	self.t = time.time()
284	time.sleep(0.05)
285	elif self.i % 40 == 0:
286	self.i = 0
287	self.t = time.time()
288	time.sleep(0.01)
289
290	self.sender.send_multipart([str(id).encode(), pickle.dumps(data)])	291	self.sender.send_multipart([str(id).encode(), pickle.dumps(data)])
291		292
292	def runReader(self):	293	def runReader(self):

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