dinislage/web2py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

scheduler.queue_task(...), thanks Niphlod

Browse Source
This commit is contained in:
mdipierro
2012-10-18 13:02:19 -05:00
parent 3124fd90b9
commit 8d5c82c531
2 changed files with 95 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
VERSION
+1 -1
View File
@@ -1 +1 @@
Version 2.1.1 (2012-10-17 17:00:46) dev Version 2.1.1 (2012-10-18 13:02:13) dev
gluon/scheduler.py
+94 -44
View File
@@ -87,6 +87,7 @@ try:
except: except:
from simplejson import loads, dumps from simplejson import loads, dumps
logger = logging.getLogger('web2py.scheduler')
from gluon import DAL, Field, IS_NOT_EMPTY, IS_IN_SET, IS_NOT_IN_DB, IS_INT_IN_RANGE, IS_DATETIME from gluon import DAL, Field, IS_NOT_EMPTY, IS_IN_SET, IS_NOT_IN_DB, IS_INT_IN_RANGE, IS_DATETIME
from gluon.utils import web2py_uuid from gluon.utils import web2py_uuid
@@ -115,7 +116,7 @@ CALLABLETYPES = (types.LambdaType, types.FunctionType,
class Task(object): class Task(object):
def __init__(self,app,function,timeout,args='[]',vars='{}',**kwargs): def __init__(self,app,function,timeout,args='[]',vars='{}',**kwargs):
logging.debug(' new task allocated: %s.%s' % (app,function)) logger.debug(' new task allocated: %s.%s' % (app,function))
self.app = app self.app = app
self.function = function self.function = function
self.timeout = timeout self.timeout = timeout
@@ -127,11 +128,11 @@ class Task(object):
class TaskReport(object): class TaskReport(object):
def __init__(self,status,result=None,output=None,tb=None): def __init__(self,status,result=None,output=None,tb=None):
logging.debug(' new task report: %s' % status) logger.debug(' new task report: %s' % status)
if tb: if tb:
logging.debug(' traceback: %s' % tb) logger.debug(' traceback: %s' % tb)
else: else:
logging.debug(' result: %s' % result) logger.debug(' result: %s' % result)
self.status = status self.status = status
self.result = result self.result = result
self.output = output self.output = output
@@ -173,7 +174,7 @@ def _decode_dict(dct):
def executor(queue,task, out): def executor(queue,task, out):
""" the background process """ """ the background process """
logging.debug(' task started') logger.debug(' task started')
class LogOutput(object): class LogOutput(object):
"""Facility to log output at intervals""" """Facility to log output at intervals"""
@@ -248,7 +249,7 @@ class MetaScheduler(threading.Thread):
queue = multiprocessing.Queue(maxsize=1) queue = multiprocessing.Queue(maxsize=1)
p = multiprocessing.Process(target=executor,args=(queue,task,out)) p = multiprocessing.Process(target=executor,args=(queue,task,out))
self.process = p self.process = p
logging.debug(' task starting') logger.debug(' task starting')
p.start() p.start()
task_output = "" task_output = ""
@@ -266,7 +267,7 @@ class MetaScheduler(threading.Thread):
not task.timeout or time.time()-start < task.timeout): not task.timeout or time.time()-start < task.timeout):
if tout: if tout:
try: try:
logging.debug(' partial output saved') logger.debug(' partial output saved')
db(sr.id==task.run_id).update(output = task_output) db(sr.id==task.run_id).update(output = task_output)
db.commit() db.commit()
except: except:
@@ -276,7 +277,7 @@ class MetaScheduler(threading.Thread):
while not out.empty(): while not out.empty():
tout += out.get() tout += out.get()
if tout: if tout:
logging.debug(' partial output: "%s"' % str(tout)) logger.debug(' partial output: "%s"' % str(tout))
if CLEAROUT in tout: if CLEAROUT in tout:
task_output = tout[tout.rfind(CLEAROUT)+len(CLEAROUT):] task_output = tout[tout.rfind(CLEAROUT)+len(CLEAROUT):]
else: else:
@@ -285,12 +286,12 @@ class MetaScheduler(threading.Thread):
p.terminate() p.terminate()
p.join() p.join()
self.have_heartbeat = False self.have_heartbeat = False
logging.debug(' task stopped by general exception') logger.debug(' task stopped by general exception')
tr = TaskReport(STOPPED) tr = TaskReport(STOPPED)
else: else:
if p.is_alive(): if p.is_alive():
p.terminate() p.terminate()
logging.debug(' task timeout') logger.debug(' task timeout')
try: try:
# we try to get a traceback here # we try to get a traceback here
tr = queue.get(timeout=2) tr = queue.get(timeout=2)
@@ -300,21 +301,21 @@ class MetaScheduler(threading.Thread):
tr = TaskReport(TIMEOUT) tr = TaskReport(TIMEOUT)
elif queue.empty(): elif queue.empty():
self.have_heartbeat = False self.have_heartbeat = False
logging.debug(' task stopped') logger.debug(' task stopped')
tr = TaskReport(STOPPED) tr = TaskReport(STOPPED)
else: else:
logging.debug(' task completed or failed') logger.debug(' task completed or failed')
tr = queue.get() tr = queue.get()
tr.output = task_output tr.output = task_output
return tr return tr
def die(self): def die(self):
logging.info('die!') logger.info('die!')
self.have_heartbeat = False self.have_heartbeat = False
self.terminate_process() self.terminate_process()
def give_up(self): def give_up(self):
logging.info('Giving up as soon as possible!') logger.info('Giving up as soon as possible!')
self.have_heartbeat = False self.have_heartbeat = False
def terminate_process(self): def terminate_process(self):
@@ -356,18 +357,18 @@ class MetaScheduler(threading.Thread):
try: try:
self.start_heartbeats() self.start_heartbeats()
while True and self.have_heartbeat: while True and self.have_heartbeat:
logging.debug('looping...') logger.debug('looping...')
task = self.pop_task() task = self.pop_task()
if task: if task:
self.empty_runs = 0 self.empty_runs = 0
self.report_task(task,self.async(task)) self.report_task(task,self.async(task))
else: else:
self.empty_runs += 1 self.empty_runs += 1
logging.debug('sleeping...') logger.debug('sleeping...')
if self.max_empty_runs != 0: if self.max_empty_runs != 0:
logging.debug('empty runs %s/%s', self.empty_runs, self.max_empty_runs) logger.debug('empty runs %s/%s', self.empty_runs, self.max_empty_runs)
if self.empty_runs >= self.max_empty_runs: if self.empty_runs >= self.max_empty_runs:
logging.info('empty runs limit reached, killing myself') logger.info('empty runs limit reached, killing myself')
self.die() self.die()
self.sleep() self.sleep()
except KeyboardInterrupt: except KeyboardInterrupt:
@@ -412,7 +413,7 @@ class Scheduler(MetaScheduler):
self.db = db self.db = db
self.db_thread = None self.db_thread = None
self.tasks = tasks self.tasks = tasks
self.group_names = group_names or ['main'] self.group_names = group_names
self.heartbeat = heartbeat self.heartbeat = heartbeat
self.worker_name = worker_name or socket.gethostname()+'#'+str(os.getpid()) self.worker_name = worker_name or socket.gethostname()+'#'+str(os.getpid())
self.worker_status = RUNNING, 1 #tuple containing status as recorded in self.worker_status = RUNNING, 1 #tuple containing status as recorded in
@@ -442,7 +443,7 @@ class Scheduler(MetaScheduler):
def define_tables(self,db,migrate): def define_tables(self,db,migrate):
from gluon.dal import DEFAULT from gluon.dal import DEFAULT
logging.debug('defining tables (migrate=%s)' % migrate) logger.debug('defining tables (migrate=%s)' % migrate)
now = self.now now = self.now
db.define_table( db.define_table(
'scheduler_task', 'scheduler_task',
@@ -513,25 +514,25 @@ class Scheduler(MetaScheduler):
self.start_heartbeats() self.start_heartbeats()
while True and self.have_heartbeat: while True and self.have_heartbeat:
if self.worker_status[0] == DISABLED: if self.worker_status[0] == DISABLED:
logging.debug('Someone stopped me, sleeping until better times come (%s)' % self.worker_status[1]) logger.debug('Someone stopped me, sleeping until better times come (%s)' % self.worker_status[1])
self.sleep() self.sleep()
continue continue
logging.debug('looping...') logger.debug('looping...')
task = self.pop_task() task = self.pop_task()
if task: if task:
self.empty_runs = 0 self.empty_runs = 0
self.report_task(task,self.async(task)) self.report_task(task,self.async(task))
else: else:
self.empty_runs += 1 self.empty_runs += 1
logging.debug('sleeping...') logger.debug('sleeping...')
if self.max_empty_runs != 0: if self.max_empty_runs != 0:
logging.debug('empty runs %s/%s', self.empty_runs, self.max_empty_runs) logger.debug('empty runs %s/%s', self.empty_runs, self.max_empty_runs)
if self.empty_runs >= self.max_empty_runs: if self.empty_runs >= self.max_empty_runs:
logging.info('empty runs limit reached, killing myself') logger.info('empty runs limit reached, killing myself')
self.die() self.die()
self.sleep() self.sleep()
except (KeyboardInterrupt, SystemExit): except (KeyboardInterrupt, SystemExit):
logging.info('catched') logger.info('catched')
self.die() self.die()
def pop_task(self): def pop_task(self):
@@ -548,7 +549,7 @@ class Scheduler(MetaScheduler):
break break
except: except:
db.rollback() db.rollback()
logging.error('TICKER: error assigning tasks') logger.error('TICKER: error assigning tasks')
return None return None
db.commit() db.commit()
grabbed = db(ts.assigned_worker_name==self.worker_name)\ grabbed = db(ts.assigned_worker_name==self.worker_name)\
@@ -559,9 +560,9 @@ class Scheduler(MetaScheduler):
task.update_record(status=RUNNING,last_run_time=now) task.update_record(status=RUNNING,last_run_time=now)
#noone will touch my task! #noone will touch my task!
db.commit() db.commit()
logging.debug(' work to do %s' % task.id) logger.debug(' work to do %s' % task.id)
else: else:
logging.debug('nothing to do') logger.debug('nothing to do')
return None return None
next_run_time = task.last_run_time + datetime.timedelta(seconds=task.period) next_run_time = task.last_run_time + datetime.timedelta(seconds=task.period)
times_run = task.times_run + 1 times_run = task.times_run + 1
@@ -571,7 +572,7 @@ class Scheduler(MetaScheduler):
run_again = False run_again = False
run_id = 0 run_id = 0
while True and not self.discard_results: while True and not self.discard_results:
logging.debug(' new scheduler_run record') logger.debug(' new scheduler_run record')
try: try:
run_id = db.scheduler_run.insert( run_id = db.scheduler_run.insert(
scheduler_task = task.id, scheduler_task = task.id,
@@ -582,7 +583,7 @@ class Scheduler(MetaScheduler):
break break
except: except:
db.rollback() db.rollback()
logging.info('new task %(id)s "%(task_name)s" %(application_name)s.%(function_name)s' % task) logger.info('new task %(id)s "%(task_name)s" %(application_name)s.%(function_name)s' % task)
return Task( return Task(
app = task.application_name, app = task.application_name,
function = task.function_name, function = task.function_name,
@@ -609,7 +610,7 @@ class Scheduler(MetaScheduler):
#result is 'null' as a string if task completed #result is 'null' as a string if task completed
#if it's stopped it's None as NoneType, so we record #if it's stopped it's None as NoneType, so we record
#the STOPPED "run" anyway #the STOPPED "run" anyway
logging.debug(' recording task report in db (%s)' % task_report.status) logger.debug(' recording task report in db (%s)' % task_report.status)
db(db.scheduler_run.id==task.run_id).update( db(db.scheduler_run.id==task.run_id).update(
status = task_report.status, status = task_report.status,
stop_time = now, stop_time = now,
@@ -617,7 +618,7 @@ class Scheduler(MetaScheduler):
output = task_report.output, output = task_report.output,
traceback = task_report.tb) traceback = task_report.tb)
else: else:
logging.debug(' deleting task report in db because of no result') logger.debug(' deleting task report in db because of no result')
db(db.scheduler_run.id==task.run_id).delete() db(db.scheduler_run.id==task.run_id).delete()
is_expired = (task.stop_time is_expired = (task.stop_time
and task.next_run_time > task.stop_time and task.next_run_time > task.stop_time
@@ -647,7 +648,7 @@ class Scheduler(MetaScheduler):
next_run_time = task.next_run_time, next_run_time = task.next_run_time,
status=status) status=status)
db.commit() db.commit()
logging.info('task completed (%s)' % task_report.status) logger.info('task completed (%s)' % task_report.status)
break break
except: except:
db.rollback() db.rollback()
@@ -659,7 +660,7 @@ class Scheduler(MetaScheduler):
def send_heartbeat(self,counter): def send_heartbeat(self,counter):
if not self.db_thread: if not self.db_thread:
logging.debug('thread building own DAL object') logger.debug('thread building own DAL object')
self.db_thread = DAL(self.db._uri,folder = self.db._adapter.folder) self.db_thread = DAL(self.db._uri,folder = self.db._adapter.folder)
self.define_tables(self.db_thread,migrate=False) self.define_tables(self.db_thread,migrate=False)
try: try:
@@ -679,13 +680,13 @@ class Scheduler(MetaScheduler):
if mybackedstatus.status == DISABLED: if mybackedstatus.status == DISABLED:
self.worker_status = DISABLED, self.worker_status[1]#keep sleeping self.worker_status = DISABLED, self.worker_status[1]#keep sleeping
if self.worker_status[1] == MAXHIBERNATION: if self.worker_status[1] == MAXHIBERNATION:
logging.debug('........recording heartbeat') logger.debug('........recording heartbeat')
db(sw.worker_name==self.worker_name).update( db(sw.worker_name==self.worker_name).update(
last_heartbeat = now) last_heartbeat = now)
elif mybackedstatus.status == TERMINATE: elif mybackedstatus.status == TERMINATE:
self.worker_status = TERMINATE, self.worker_status[1] self.worker_status = TERMINATE, self.worker_status[1]
logging.debug("Waiting to terminate the current task") logger.debug("Waiting to terminate the current task")
self.give_up() self.give_up()
return return
elif mybackedstatus.status == KILL: elif mybackedstatus.status == KILL:
@@ -693,7 +694,7 @@ class Scheduler(MetaScheduler):
self.die() self.die()
else: else:
logging.debug('........recording heartbeat') logger.debug('........recording heartbeat')
db(sw.worker_name==self.worker_name).update( db(sw.worker_name==self.worker_name).update(
last_heartbeat = now, status = ACTIVE) last_heartbeat = now, status = ACTIVE)
self.worker_status = ACTIVE, 1 #re-activating the process self.worker_status = ACTIVE, 1 #re-activating the process
@@ -702,7 +703,7 @@ class Scheduler(MetaScheduler):
if counter % 5 == 0: if counter % 5 == 0:
try: try:
# delete inactive workers # delete inactive workers
logging.debug(' freeing workers that have not sent heartbeat') logger.debug(' freeing workers that have not sent heartbeat')
inactive_workers = db( inactive_workers = db(
((sw.last_heartbeat<expiration) & (sw.status == ACTIVE)) | ((sw.last_heartbeat<expiration) & (sw.status == ACTIVE)) |
((sw.last_heartbeat<departure) & (sw.status != ACTIVE)) ((sw.last_heartbeat<departure) & (sw.status != ACTIVE))
@@ -730,11 +731,11 @@ class Scheduler(MetaScheduler):
if not ticker: if not ticker:
db(sw.worker_name == self.worker_name).update(is_ticker = True) db(sw.worker_name == self.worker_name).update(is_ticker = True)
db(sw.worker_name != self.worker_name).update(is_ticker = False) db(sw.worker_name != self.worker_name).update(is_ticker = False)
logging.info("TICKER: I'm a ticker (%s)" % self.worker_name) logger.info("TICKER: I'm a ticker (%s)" % self.worker_name)
db.commit() db.commit()
return True return True
else: else:
logging.info("%s is a ticker, I'm a poor worker" % ticker.worker_name) logger.info("%s is a ticker, I'm a poor worker" % ticker.worker_name)
return False return False
def assign_tasks(self): def assign_tasks(self):
@@ -798,12 +799,61 @@ class Scheduler(MetaScheduler):
#I didn't report tasks but I'm working nonetheless!!!! #I didn't report tasks but I'm working nonetheless!!!!
if len(tasks) > 0: if len(tasks) > 0:
self.empty_runs = 0 self.empty_runs = 0
logging.info('TICKER: workers are %s' % len(all_workers)) logger.info('TICKER: workers are %s' % len(all_workers))
logging.info('TICKER: tasks are %s' % len(tasks)) logger.info('TICKER: tasks are %s' % len(tasks))
def sleep(self): def sleep(self):
time.sleep(self.heartbeat*self.worker_status[1]) # should only sleep until next available task time.sleep(self.heartbeat*self.worker_status[1]) # should only sleep until next available task
def queue_task(self, function, args=[], vars={}, **kwargs):
if hasattr(function, '__name__'):
function = function.__name__
targs = 'args' in kwargs and kwargs.pop('args') or dumps(args)
tvars = 'vars' in kwargs and kwargs.pop('vars') or dumps(vars)
tuuid = 'uuid' in kwargs and kwargs.pop('uuid') or web2py_uuid()
tname = 'task_name' in kwargs and kwargs.pop('task_name') or function
rtn = self.db.scheduler_task.validate_and_insert(
function_name=function,
task_name=tname,
args=targs,
vars=tvars,
uuid=tuuid,
**kwargs)
if not rtn.errors:
rtn.uuid = tuuid
else:
rtn.uuid = None
return rtn
def task_status(self, ref, output=False):
from gluon.dal import Query
sr, st = self.db.scheduler_run, self.db.scheduler_task
if isinstance(ref, int):
q = st.id == ref
elif isinstance(ref, str):
q = st.uuid == ref
elif isinstance(ref, Query):
q = ref
else:
raise SyntaxError, "You can retrieve results only by id, uuid or Query"
fields = st.ALL
left = False
orderby = ~st.id
if output:
fields = st.ALL, sr.ALL
left = sr.on(sr.scheduler_task == st.id)
orderby = ~st.id|~sr.id
row = self.db(q).select(
*fields,
orderby=orderby,
left=left,
limitby=(0,1)
).first()
if output:
row.result = row.scheduler_run.result and \
loads(row.scheduler_run.result, object_hook=_decode_dict) or None
return row
def main(): def main():
""" """
allows to run worker without python web2py.py .... by simply python this.py allows to run worker without python web2py.py .... by simply python this.py
@@ -860,7 +910,7 @@ def main():
tasks = {} tasks = {}
group_names = [x.strip() for x in options.group_names.split(',')] group_names = [x.strip() for x in options.group_names.split(',')]
logging.getLogger().setLevel(options.logger_level) logger.getLogger().setLevel(options.logger_level)
print 'groups for this worker: '+', '.join(group_names) print 'groups for this worker: '+', '.join(group_names)
print 'connecting to database in folder: ' + options.db_folder or './' print 'connecting to database in folder: ' + options.db_folder or './'