Re: [Kimchi-devel] [PATCH] [Wok 1/3] Issue #158: Move AsyncTask information to memory

On 08/26/2016 11:18 AM, pvital(a)linux.vnet.ibm.com wrote: > From: Paulo Vital <pvital(a)linux.vnet.ibm.com&gt; > > This patch moves all infrastructure of AsyncTask to store and read > the tasks' information (id, status, messages and target_uri) from > objectstore to memory, by the implementation of a dictionary of > AsyncTasks objects. > > It also removes the tasks cleanup process from objectstore. > > Signed-off-by: Paulo Vital <pvital(a)linux.vnet.ibm.com&gt; > --- > src/wok/asynctask.py | 58 +++++++++++++++++++++++++++++++++----------------- > src/wok/i18n.py | 2 -- > src/wok/objectstore.py | 6 ------ > src/wok/utils.py | 19 ----------------- > 4 files changed, 39 insertions(+), 46 deletions(-) > > diff --git a/src/wok/asynctask.py b/src/wok/asynctask.py > index fb614a2..3298787 100644 > --- a/src/wok/asynctask.py > +++ b/src/wok/asynctask.py > @@ -21,24 +21,55 @@ > > import cherrypy > import threading > +import time > import traceback > > > -from wok.exception import OperationFailed > +task_id = 0 > +tasks_queue = {} > > > -class AsyncTask(object): > - def __init__(self, id, target_uri, fn, objstore, opaque=None): > - if objstore is None: > - raise OperationFailed("WOKASYNC0001E") > +def get_next_task_id(): > + global task_id > + task_id += 1 > + return task_id > + > + I don't think we need it anymore. IMO it would be better to auto generate the task ID by using uuid or other tool. If you agree on that, you can send a separated patch to do that.

> +def add_task(target_uri, fn, opaque=None): > + # let's prevent memory leak > + clean_old_tasks() > + id = str(get_next_task_id()) > + tasks_queue[id] = AsyncTask(id, target_uri, fn, opaque) > + return id > + > + > +def remove_task(id): > + try: > + del tasks_queue[id] > + except KeyError: > + cherrypy.log.error_log.error("There's no task_id %s in tasks_queue." > + " Nothing changed." % id) > + > > +def clean_old_tasks(): > + """ > + Check for all tasks in tasks_queue and remove those if timestamp < than > + 12 hours ago and status equal do finished or failed. > + """ > + for id, task in tasks_queue.items(): > + if (task.timestamp < (time.time()-43200)): > + if (task.status is 'finished') or (task.status is 'failed'): > + remove_task(id) > + > + > +class AsyncTask(object): > + def __init__(self, id, target_uri, fn, opaque=None): > self.id = str(id) > self.target_uri = target_uri > self.fn = fn > - self.objstore = objstore > self.status = 'running' > - self.message = 'OK' > - self._save_helper() > + self.message = 'The request is being processing.' > + self.timestamp = time.time() > self._cp_request = cherrypy.serving.request > self.thread = threading.Thread(target=self._run_helper, > args=(opaque, self._status_cb)) > @@ -51,17 +82,6 @@ class AsyncTask(object): > > if message.strip(): > self.message = message > - self._save_helper() > - > - def _save_helper(self): > - obj = {} > - for attr in ('id', 'target_uri', 'message', 'status'): > - obj[attr] = getattr(self, attr) > - try: > - with self.objstore as session: > - session.store('task', self.id, obj) > - except Exception as e: > - raise OperationFailed('WOKASYNC0002E', {'err': e.message}) > > def _run_helper(self, opaque, cb): > cherrypy.serving.request = self._cp_request > diff --git a/src/wok/i18n.py b/src/wok/i18n.py > index 33107ee..1a0446b 100644 > --- a/src/wok/i18n.py > +++ b/src/wok/i18n.py > @@ -33,8 +33,6 @@ messages = { > "WOKAPI0008E": _("Parameters does not match requirement in schema: %(err)s"), > "WOKAPI0009E": _("You don't have permission to perform this operation."), > > - "WOKASYNC0001E": _("Datastore is not initiated in the model object."), > - "WOKASYNC0002E": _("Unable to start task due error: %(err)s"), > "WOKASYNC0003E": _("Timeout of %(seconds)s seconds expired while running task '%(task)s."), > > "WOKAUTH0001E": _("Authentication failed for user '%(username)s'. [Error code: %(code)s]"), > diff --git a/src/wok/objectstore.py b/src/wok/objectstore.py > index 59354f3..817f60c 100644 > --- a/src/wok/objectstore.py > +++ b/src/wok/objectstore.py > @@ -107,8 +107,6 @@ class ObjectStore(object): > c.execute('''SELECT * FROM sqlite_master WHERE type='table' AND > tbl_name='objects'; ''') > res = c.fetchall() > - # Because the tasks are regarded as temporary resource, the task states > - # are purged every time the daemon startup > if len(res) == 0: > c.execute('''CREATE TABLE objects > (id TEXT, type TEXT, json TEXT, version TEXT, > @@ -116,10 +114,6 @@ class ObjectStore(object): > conn.commit() > return > > - # Clear out expired objects from a previous session > - c.execute('''DELETE FROM objects WHERE type = 'task'; ''') > - conn.commit() > - > def _get_conn(self): > ident = threading.currentThread().ident > try: > diff --git a/src/wok/utils.py b/src/wok/utils.py > index 4c132a1..7599e85 100644 > --- a/src/wok/utils.py > +++ b/src/wok/utils.py > @@ -39,31 +39,12 @@ from datetime import datetime, timedelta > from multiprocessing import Process, Queue > from threading import Timer > > -from wok.asynctask import AsyncTask > from wok.config import paths, PluginPaths > from wok.exception import InvalidParameter, TimeoutExpired > from wok.stringutils import decode_value > > > wok_log = cherrypy.log.error_log > -task_id = 0 > - > - > -def get_next_task_id(): > - global task_id > - task_id += 1 > - return task_id > - > - > -def get_task_id(): > - global task_id > - return task_id > - > - > -def add_task(target_uri, fn, objstore, opaque=None): > - id = get_next_task_id() > - AsyncTask(id, target_uri, fn, objstore, opaque) > - return id > > > def is_digit(value):