mirror of
https://github.com/ChronosX88/medved.git
synced 2024-10-18 08:00:59 +00:00
71 lines
2.6 KiB
Python
71 lines
2.6 KiB
Python
from lib import Service, Loader, Loadable
|
|
|
|
from time import sleep
|
|
|
|
from rq import Queue
|
|
from redis import Redis
|
|
|
|
|
|
class Executor(Service):
|
|
"""Base class for executors"""
|
|
def __init__(self, thread, id, root):
|
|
super().__init__(thread, id, root)
|
|
self._logger.add_field('service', 'Executor')
|
|
self._logger.add_field('vname', self.__class__.__name__)
|
|
|
|
|
|
class RQExecutor(Executor):
|
|
"""rq (redis queue) executor"""
|
|
def __init__(self, id, root):
|
|
super().__init__(self.__run, id, root)
|
|
|
|
def __run(self):
|
|
redis_conn = Redis(host=self.lcnf.get('redis').get('host'))
|
|
jobs = []
|
|
known_sources = {}
|
|
|
|
while self._running:
|
|
sleep(self.lcnf.get('delay', 0.5))
|
|
try:
|
|
for pn, pipeline in self.cnf.get("pipelines").items():
|
|
if pn not in self.cnf.get('core').get('pipelines'):
|
|
continue
|
|
for step in pipeline['steps']:
|
|
q = Queue(step.get('priority', 'normal'), connection=redis_conn)
|
|
for job_id in jobs:
|
|
job = q.fetch_job(job_id)
|
|
if job:
|
|
if job.result is not None:
|
|
self._logger.debug("%s|%s", job_id, job._status)
|
|
self._data.update(job.result)
|
|
job.cleanup()
|
|
jobs.remove(job_id)
|
|
if len(jobs) + 1 > self.lcnf.get('qsize', 200):
|
|
continue
|
|
filter = {"steps.%s" % step['task']: {'$exists': False}}
|
|
filter.update({key: value for key, value in step.get("if", {}).items()})
|
|
count = step.get('parallel', 1)
|
|
# get as much as possible from own pool
|
|
items = self._data.get(block=False, count=count, filter=filter)
|
|
# obtain everything else from source
|
|
if len(items) < count:
|
|
source = None
|
|
source_id = pipeline.get('source')
|
|
if source_id in known_sources:
|
|
source = known_sources[source_id]
|
|
else:
|
|
source = Loader.by_id('storage', source_id)
|
|
known_sources[source_id] = source
|
|
new_items = source.get(block=False, count=(count - len(items)), filter=filter)
|
|
items.extend(new_items)
|
|
source.remove(new_items)
|
|
|
|
if items:
|
|
for i in items:
|
|
i['steps'][step['task']] = None
|
|
self._data.update(items)
|
|
job = q.enqueue("lib.exec.Task.run", step['task'], items)
|
|
self._logger.info("%s|%s|%s|%s", job.id, step.get('priority', 'normal'), step['task'], len(items))
|
|
jobs.append(job.id)
|
|
except Exception as e:
|
|
self._logger.error("Error in executor main thread: %s", e) |