Merge pull request #1315 from niphlod/enhancement/scheduler

can now process tasks with huge_results

gluon/scheduler.py

@@ -9,6 +9,26 @@ Background processes made simple
 ---------------------------------
 """
 
+
+import os
+import time
+import multiprocessing
+import sys
+import threading
+import traceback
+import signal
+import socket
+import datetime
+import logging
+import optparse
+import tempfile
+import types
+import Queue
+from gluon import DAL, Field, IS_NOT_EMPTY, IS_IN_SET, IS_NOT_IN_DB
+from gluon import IS_INT_IN_RANGE, IS_DATETIME, IS_IN_DB
+from gluon.utils import web2py_uuid
+from gluon.storage import Storage
+
 USAGE = """
 ## Example
 
@@ -67,20 +87,6 @@ sudo restart web2py-scheduler
 sudo status web2py-scheduler
 """
 
-import os
-import time
-import multiprocessing
-import sys
-import threading
-import traceback
-import signal
-import socket
-import datetime
-import logging
-import optparse
-import types
-import Queue
-
 path = os.getcwd()
 
 if 'WEB2PY_PATH' not in os.environ:
@@ -101,12 +107,6 @@ IDENTIFIER = "%s#%s" % (socket.gethostname(), os.getpid())
 
 logger = logging.getLogger('web2py.scheduler.%s' % IDENTIFIER)
 
-from gluon import DAL, Field, IS_NOT_EMPTY, IS_IN_SET, IS_NOT_IN_DB
-from gluon import IS_INT_IN_RANGE, IS_DATETIME, IS_IN_DB
-from gluon.utils import web2py_uuid
-from gluon.storage import Storage
-
-
 QUEUED = 'QUEUED'
 ASSIGNED = 'ASSIGNED'
 RUNNING = 'RUNNING'
@@ -168,24 +168,25 @@ class TaskReport(object):
 
 
 class JobGraph(object):
-    """Experimental: with JobGraph you can specify
+    """Experimental: dependencies amongs tasks"""
-    dependencies amongs tasks"""
 
     def __init__(self, db, job_name):
         self.job_name = job_name or 'job_0'
         self.db = db
 
     def add_deps(self, task_parent, task_child):
-        """Creates a dependency between task_parent and task_child"""
+        """Create a dependency between task_parent and task_child."""
         self.db.scheduler_task_deps.insert(task_parent=task_parent,
                                            task_child=task_child,
                                            job_name=self.job_name)
 
     def validate(self, job_name=None):
-        """Validates if all tasks job_name can be completed, i.e. there
+        """Validate if all tasks job_name can be completed.
-        are no mutual dependencies among tasks.
+
+        Checks if there are no mutual dependencies among tasks.
         Commits at the end if successfull, or it rollbacks the entire
-        transaction. Handle with care!"""
+        transaction. Handle with care!
+        """
         db = self.db
         sd = db.scheduler_task_deps
         if job_name:
@@ -223,14 +224,6 @@ class JobGraph(object):
             db.rollback()
             return None
 
-
-def demo_function(*argv, **kwargs):
-    """ test function """
-    for i in range(argv[0]):
-        print 'click', i
-        time.sleep(1)
-    return 'done'
-
 # the two functions below deal with simplejson decoding as unicode, esp for the dict decode
 # and subsequent usage as function Keyword arguments unicode variable names won't work!
 # borrowed from http://stackoverflow.com/questions/956867/how-to-get-string-objects-instead-unicode-ones-from-json-in-python
@@ -261,11 +254,12 @@ def _decode_dict(dct):
 
 
 def executor(queue, task, out):
-    """The function used to execute tasks in the background process"""
+    """The function used to execute tasks in the background process."""
     logger.debug('    task started')
 
     class LogOutput(object):
-        """Facility to log output at intervals"""
+        """Facility to log output at intervals."""
+
         def __init__(self, out_queue):
             self.out_queue = out_queue
             self.stdout = sys.stdout
@@ -280,7 +274,11 @@ def executor(queue, task, out):
         def write(self, data):
             self.out_queue.put(data)
 
-    W2P_TASK = Storage({'id': task.task_id, 'uuid': task.uuid})
+    W2P_TASK = Storage({
+                       'id': task.task_id,
+                       'uuid': task.uuid,
+                       'run_id': task.run_id
+                       })
     stdout = LogOutput(out)
     try:
         if task.app:
@@ -318,6 +316,11 @@ def executor(queue, task, out):
             result = eval(task.function)(
                 *loads(task.args, object_hook=_decode_dict),
                 **loads(task.vars, object_hook=_decode_dict))
+        if len(result) >= 1024:
+            fd, temp_path = tempfile.mkstemp(suffix='.w2p_sched')
+            with os.fdopen(fd, 'w') as f:
+                f.write(result)
+            result = 'w2p_special:%s' % temp_path
         queue.put(TaskReport('COMPLETED', result=result))
     except BaseException, e:
         tb = traceback.format_exc()
@@ -335,7 +338,7 @@ class MetaScheduler(threading.Thread):
         self.empty_runs = 0
 
     def async(self, task):
-        """Starts the background process
+        """Start the background process.
 
         Args:
             task : a `Task` object
@@ -410,6 +413,12 @@ class MetaScheduler(threading.Thread):
             else:
                 logger.debug('  task completed or failed')
                 tr = queue.get()
+        result = tr.result
+        if result and result.startswith('w2p_special'):
+            temp_path = result.replace('w2p_special:', '', 1)
+            with open(temp_path) as f:
+                tr.result = f.read()
+            os.unlink(temp_path)
         tr.output = task_output
         return tr
 
@@ -444,50 +453,23 @@ class MetaScheduler(threading.Thread):
         self.start()
 
     def send_heartbeat(self, counter):
-        print 'thum'
+        raise NotImplementedError
-        time.sleep(1)
 
     def pop_task(self):
         """Fetches a task ready to be executed"""
-        return Task(
+        raise NotImplementedError
-            app=None,
-            function='demo_function',
-            timeout=7,
-            args='[2]',
-            vars='{}')
 
     def report_task(self, task, task_report):
         """Creates a task report"""
-        print 'reporting task'
+        raise NotImplementedError
-        pass
 
     def sleep(self):
-        pass
+        raise NotImplementedError
 
     def loop(self):
         """Main loop, fetching tasks and starting executor's background
         processes"""
-        try:
+        raise NotImplementedError
-            self.start_heartbeats()
-            while True and self.have_heartbeat:
-                logger.debug('looping...')
-                task = self.pop_task()
-                if task:
-                    self.empty_runs = 0
-                    self.report_task(task, self.async(task))
-                else:
-                    self.empty_runs += 1
-                    logger.debug('sleeping...')
-                    if self.max_empty_runs != 0:
-                        logger.debug('empty runs %s/%s',
-                                     self.empty_runs, self.max_empty_runs)
-                        if self.empty_runs >= self.max_empty_runs:
-                            logger.info(
-                                'empty runs limit reached, killing myself')
-                            self.die()
-                    self.sleep()
-        except KeyboardInterrupt:
-            self.die()
 
 
 TASK_STATUS = (QUEUED, RUNNING, COMPLETED, FAILED, TIMEOUT, STOPPED, EXPIRED)
@@ -594,11 +576,11 @@ class Scheduler(MetaScheduler):
         return True
 
     def now(self):
-        """Shortcut that fetches current time based on UTC preferences"""
+        """Shortcut that fetches current time based on UTC preferences."""
         return self.utc_time and datetime.datetime.utcnow() or datetime.datetime.now()
 
     def set_requirements(self, scheduler_task):
-        """Called to set defaults for lazy_tables connections"""
+        """Called to set defaults for lazy_tables connections."""
         from gluon import current
         if hasattr(current, 'request'):
             scheduler_task.application_name.default = '%s/%s' % (
@@ -606,7 +588,7 @@ class Scheduler(MetaScheduler):
             )
 
     def define_tables(self, db, migrate):
-        """Defines Scheduler tables structure"""
+        """Define Scheduler tables structure."""
         from pydal.base import DEFAULT
         logger.debug('defining tables (migrate=%s)', migrate)
         now = self.now
@@ -693,14 +675,14 @@ class Scheduler(MetaScheduler):
 
     @staticmethod
     def total_seconds(td):
-        # backport for py2.6
+        """Backport for py2.6."""
         if hasattr(td, 'total_seconds'):
             return td.total_seconds()
         else:
             return (td.microseconds + (td.seconds + td.days * 24 * 3600) * 10 ** 6) / 10.0 ** 6
 
     def loop(self, worker_name=None):
-        """Main loop
+        """Main loop.
 
         This works basically as a neverending loop that:
 
@@ -752,7 +734,8 @@ class Scheduler(MetaScheduler):
             self.die()
 
     def wrapped_assign_tasks(self, db):
-        """Commodity function to call `assign_tasks` and trap exceptions
+        """Commodity function to call `assign_tasks` and trap exceptions.
+
         If an exception is raised, assume it happened because of database
         contention and retries `assign_task` after 0.5 seconds
         """
@@ -773,7 +756,8 @@ class Scheduler(MetaScheduler):
                 time.sleep(0.5)
 
     def wrapped_pop_task(self):
-        """Commodity function to call `pop_task` and trap exceptions
+        """Commodity function to call `pop_task` and trap exceptions.
+
         If an exception is raised, assume it happened because of database
         contention and retries `pop_task` after 0.5 seconds
         """
@@ -793,7 +777,7 @@ class Scheduler(MetaScheduler):
                 time.sleep(0.5)
 
     def pop_task(self, db):
-        """Grabs a task ready to be executed from the queue"""
+        """Grab a task ready to be executed from the queue."""
         now = self.now()
         st = self.db.scheduler_task
         if self.is_a_ticker and self.do_assign_tasks:
@@ -874,7 +858,8 @@ class Scheduler(MetaScheduler):
             uuid=task.uuid)
 
     def wrapped_report_task(self, task, task_report):
-        """Commodity function to call `report_task` and trap exceptions
+        """Commodity function to call `report_task` and trap exceptions.
+
         If an exception is raised, assume it happened because of database
         contention and retries `pop_task` after 0.5 seconds
         """
@@ -891,8 +876,10 @@ class Scheduler(MetaScheduler):
                 time.sleep(0.5)
 
     def report_task(self, task, task_report):
-        """Takes care of storing the result according to preferences
+        """Take care of storing the result according to preferences.
-        and deals with logic for repeating tasks"""
+
+        Deals with logic for repeating tasks.
+        """
         db = self.db
         now = self.now()
         st = db.scheduler_task
@@ -914,12 +901,12 @@ class Scheduler(MetaScheduler):
                 logger.debug(' deleting task report in db because of no result')
                 db(sr.id == task.run_id).delete()
         # if there is a stop_time and the following run would exceed it
-        is_expired = (task.stop_time
+        is_expired = (task.stop_time and
-                      and task.next_run_time > task.stop_time
+                      task.next_run_time > task.stop_time and
-                      and True or False)
+                      True or False)
-        status = (task.run_again and is_expired and EXPIRED
+        status = (task.run_again and is_expired and EXPIRED or
-                  or task.run_again and not is_expired
+                  task.run_again and not is_expired and
-                  and QUEUED or COMPLETED)
+                  QUEUED or COMPLETED)
         if task_report.status == COMPLETED:
             d = dict(status=status,
                      next_run_time=task.next_run_time,
@@ -945,27 +932,26 @@ class Scheduler(MetaScheduler):
         logger.info('task completed (%s)', task_report.status)
 
     def update_dependencies(self, db, task_id):
+        """Unblock execution paths for Jobs."""
         db(db.scheduler_task_deps.task_child == task_id).update(can_visit=True)
 
     def adj_hibernation(self):
-        """Used to increase the "sleep" interval for DISABLED workers"""
+        """Used to increase the "sleep" interval for DISABLED workers."""
         if self.w_stats.status == DISABLED:
             wk_st = self.w_stats.sleep
             hibernation = wk_st + HEARTBEAT if wk_st < MAXHIBERNATION else MAXHIBERNATION
             self.w_stats.sleep = hibernation
 
     def send_heartbeat(self, counter):
-        """This function is vital for proper coordination among available
+        """Coordination among available workers.
-        workers.
-        It:
 
+        It:
         - sends the heartbeat
         - elects a ticker among available workers (the only process that
             effectively dispatch tasks to workers)
         - deals with worker's statuses
         - does "housecleaning" for dead workers
         - triggers tasks assignment to workers
-
         """
         if not self.db_thread:
             logger.debug('thread building own DAL object')
@@ -1053,7 +1039,8 @@ class Scheduler(MetaScheduler):
         self.sleep()
 
     def being_a_ticker(self):
-        """Elects a TICKER process that assigns tasks to available workers.
+        """Elect a TICKER process that assigns tasks to available workers.
+
         Does its best to elect a worker that is not busy processing other tasks
         to allow a proper distribution of tasks among all active workers ASAP
         """
@@ -1087,7 +1074,7 @@ class Scheduler(MetaScheduler):
             return False
 
     def assign_tasks(self, db):
-        """Assigns task to workers, that can then pop them from the queue
+        """Assign task to workers, that can then pop them from the queue.
 
         Deals with group_name(s) logic, in order to assign linearly tasks
         to available workers for those groups
@@ -1137,9 +1124,6 @@ class Scheduler(MetaScheduler):
 
         all_available = db(
             (st.status.belongs((QUEUED, ASSIGNED))) &
-            ((st.times_run < st.repeats) | (st.repeats == 0)) &
-            (st.start_time <= now) &
-            ((st.stop_time == None) | (st.stop_time > now)) &
             (st.next_run_time <= now) &
             (st.enabled == True) &
             (st.id.belongs(no_deps))
@@ -1151,8 +1135,8 @@ class Scheduler(MetaScheduler):
         # intelligence (like esteeming how many tasks will a worker complete
         # before the ticker reassign them around, but the gain is quite small
         # 50 is a sweet spot also for fast tasks, with sane heartbeat values
-        # NB: ticker reassign tasks every 5 cycles, so if a worker completes its
+        # NB: ticker reassign tasks every 5 cycles, so if a worker completes
-        # 50 tasks in less than heartbeat*5 seconds,
+        # its 50 tasks in less than heartbeat*5 seconds,
         # it won't pick new tasks until heartbeat*5 seconds pass.
 
         # If a worker is currently elaborating a long task, its tasks needs to
@@ -1165,7 +1149,7 @@ class Scheduler(MetaScheduler):
         x = 0
         for group in wkgroups.keys():
             tasks = all_available(st.group_name == group).select(
-                limitby=(0, limit), orderby = st.next_run_time)
+                limitby=(0, limit), orderby=st.next_run_time)
             # let's break up the queue evenly among workers
             for task in tasks:
                 x += 1
@@ -1183,8 +1167,6 @@ class Scheduler(MetaScheduler):
                         status=ASSIGNED,
                         assigned_worker_name=assigned_wn
                     )
-                    if not task.task_name:
-                        d['task_name'] = task.function_name
                     db(
                         (st.id == task.id) &
                         (st.status.belongs((QUEUED, ASSIGNED)))
@@ -1204,14 +1186,13 @@ class Scheduler(MetaScheduler):
         logger.info('TICKER: tasks are %s', x)
 
     def sleep(self):
-        """Calculates the number of seconds to sleep according to worker's
+        """Calculate the number of seconds to sleep."""
-        status and `heartbeat` parameter"""
         time.sleep(self.w_stats.sleep)
         # should only sleep until next available task
 
     def set_worker_status(self, group_names=None, action=ACTIVE,
                           exclude=None, limit=None, worker_name=None):
-        """Internal function to set worker's status"""
+        """Internal function to set worker's status."""
         ws = self.db.scheduler_worker
         if not group_names:
             group_names = self.group_names
@@ -1235,10 +1216,12 @@ class Scheduler(MetaScheduler):
                 self.db(ws.id.belongs(workers)).update(status=action)
 
     def disable(self, group_names=None, limit=None, worker_name=None):
-        """Sets DISABLED on the workers processing `group_names` tasks.
+        """Set DISABLED on the workers processing `group_names` tasks.
+
         A DISABLED worker will be kept alive but it won't be able to process
         any waiting tasks, essentially putting it to sleep.
-        By default, all group_names of Scheduler's instantation are selected"""
+        By default, all group_names of Scheduler's instantation are selected
+        """
         self.set_worker_status(
             group_names=group_names,
             action=DISABLED,
@@ -1283,8 +1266,9 @@ class Scheduler(MetaScheduler):
             pvars: "raw" kwargs to be passed to the function. Automatically
                 jsonified
             kwargs: all the parameters available (basically, every
-                `scheduler_task` column). If args and vars are here, they should
+                `scheduler_task` column). If args and vars are here, they
-                be jsonified already, and they will override pargs and pvars
+                should be jsonified already, and they will override pargs
+                and pvars
 
         Returns:
             a dict just as a normal validate_and_insert(), plus a uuid key

gluon/tests/test_scheduler.py

@@ -9,6 +9,7 @@ import unittest
 import glob
 import datetime
 import sys
+
 from fix_path import fix_sys_path
 
 
@@ -255,6 +256,14 @@ class testForSchedulerRunnerBase(BaseTestScheduler):
         from gluon import current
         fdest = os.path.join(current.request.folder, 'models', 'scheduler.py')
         os.unlink(fdest)
+        additional_files = [
+            os.path.join(current.request.folder, 'private', 'demo8.pholder')
+        ]
+        for f in additional_files:
+            try:
+                os.unlink(f)
+            except:
+                pass
 
     def writefunction(self, content, initlines=None):
         from gluon import current
@@ -266,7 +275,7 @@ import time
 from gluon.scheduler import Scheduler
 db_dal = os.path.abspath(os.path.join(request.folder, '..', '..', 'dummy2.db'))
 sched_dal = DAL('sqlite://%s' % db_dal, folder=os.path.dirname(db_dal))
-sched = Scheduler(sched_dal, max_empty_runs=20, migrate=False, heartbeat=1)
+sched = Scheduler(sched_dal, max_empty_runs=15, migrate=False, heartbeat=1)
             """
         with open(fdest, 'w') as q:
             q.write(initlines)
@@ -274,34 +283,179 @@ sched = Scheduler(sched_dal, max_empty_runs=20, migrate=False, heartbeat=1)
 
     def exec_sched(self):
         import subprocess
-        call_args = [sys.executable, 'web2py.py', '-K', 'welcome']
+        call_args = [sys.executable, 'web2py.py', '--no-banner', '-D', '20','-K', 'welcome']
         ret = subprocess.call(call_args, env=dict(os.environ))
         return ret
 
+    def fetch_results(self, sched, task):
+        info = sched.task_status(task.id)
+        task_runs = self.db(self.db.scheduler_run.task_id == task.id).select()
+        return info, task_runs
+
+    def exec_asserts(self, stmts, tag):
+        for stmt in stmts:
+            self.assertEqual(stmt[1], True, msg="%s - %s" % (tag, stmt[0]))
+
 
 class TestsForSchedulerRunner(testForSchedulerRunnerBase):
 
-    def testBasic(self):
+    def testRepeats_and_Expired_and_Prio(self):
         s = Scheduler(self.db)
-        foo = s.queue_task('foo')
+        repeats = s.queue_task('demo1', ['a', 'b'], dict(c=1, d=2), repeats=2, period=5)
+        a_while_ago = datetime.datetime.now() - datetime.timedelta(seconds=60)
+        expired = s.queue_task('demo4', stop_time=a_while_ago)
+        prio1 = s.queue_task('demo1', ['scheduled_first'])
+        prio2 = s.queue_task('demo1', ['scheduled_second'], next_run_time=a_while_ago)
         self.db.commit()
         self.writefunction(r"""
-def foo():
+def demo1(*args,**vars):
-    return 'a'
+    print 'you passed args=%s and vars=%s' % (args, vars)
+    return args[0]
+
+def demo4():
+    time.sleep(15)
+    print "I'm printing something"
+    return dict(a=1, b=2)
 """)
         ret = self.exec_sched()
-        # process finished just fine
         self.assertEqual(ret, 0)
-        info = s.task_status(foo.id, output=True)
+        # repeats check
-        self.assertEqual(info.result, 'a')
+        task, task_run = self.fetch_results(s, repeats)
+        res = [
+            ("task status completed", task.status == 'COMPLETED'),
+            ("task times_run is 2", task.times_run == 2),
+            ("task ran 2 times only", len(task_run) == 2),
+            ("scheduler_run records are COMPLETED ", (task_run[0].status == task_run[1].status == 'COMPLETED')),
+            ("period is respected", (task_run[1].start_time > task_run[0].start_time + datetime.timedelta(seconds=task.period)))
+        ]
+        self.exec_asserts(res, 'REPEATS')
+
+        # expired check
+        task, task_run = self.fetch_results(s, expired)
+        res = [
+            ("task status expired", task.status == 'EXPIRED'),
+            ("task times_run is 0", task.times_run == 0),
+            ("task didn't run at all", len(task_run) == 0)
+        ]
+        self.exec_asserts(res, 'EXPIRATION')
+
+        # prio check
+        task1 = s.task_status(prio1.id, output=True)
+        task2 = s.task_status(prio2.id, output=True)
+        res = [
+            ("tasks status completed", task1.scheduler_task.status == task2.scheduler_task.status == 'COMPLETED'),
+            ("priority2 was executed before priority1" , task1.scheduler_run.id > task2.scheduler_run.id)
+        ]
+        self.exec_asserts(res, 'PRIORITY')
+
+    def testNoReturn_and_Timeout_and_Progress(self):
+        s = Scheduler(self.db)
+        noret1 = s.queue_task('demo5')
+        noret2 = s.queue_task('demo3')
+        timeout1 = s.queue_task('demo4', timeout=5)
+        timeout2 = s.queue_task('demo4')
+        progress = s.queue_task('demo6', sync_output=2)
+        self.db.commit()
+        self.writefunction(r"""
+def demo3():
+    time.sleep(15)
+    print 1/0
+    return None
+
+def demo4():
+    time.sleep(15)
+    print "I'm printing something"
+    return dict(a=1, b=2)
+
+def demo5():
+    time.sleep(15)
+    print "I'm printing something"
+    rtn = dict(a=1, b=2)
+
+def demo6():
+    time.sleep(5)
+    print '50%'
+    time.sleep(5)
+    print '!clear!100%'
+    return 1
+""")
+        ret = self.exec_sched()
+        self.assertEqual(ret, 0)
+        # noreturn check
+        task1, task_run1 = self.fetch_results(s, noret1)
+        task2, task_run2 = self.fetch_results(s, noret2)
+        res = [
+            ("tasks no_returns1 completed", task1.status == 'COMPLETED'),
+            ("tasks no_returns2 failed", task2.status == 'FAILED'),
+            ("no_returns1 doesn't have a scheduler_run record", len(task_run1) == 0),
+            ("no_returns2 has a scheduler_run record FAILED", (len(task_run2) == 1 and task_run2[0].status == 'FAILED')),
+        ]
+        self.exec_asserts(res, 'NO_RETURN')
+
+        # timeout check
+        task1 = s.task_status(timeout1.id, output=True)
+        task2 = s.task_status(timeout2.id, output=True)
+        res = [
+            ("tasks timeouts1 timeoutted", task1.scheduler_task.status == 'TIMEOUT'),
+            ("tasks timeouts2 completed", task2.scheduler_task.status == 'COMPLETED')
+        ]
+        self.exec_asserts(res, 'TIMEOUT')
+
+        # progress check
+        task1 = s.task_status(progress.id, output=True)
+        res = [
+            ("tasks percentages completed", task1.scheduler_task.status == 'COMPLETED'),
+            ("output contains only 100%", task1.scheduler_run.run_output.strip() == "100%")
+        ]
+        self.exec_asserts(res, 'PROGRESS')
+
+    def testDrift_and_env_and_immediate(self):
+        s = Scheduler(self.db)
+        immediate = s.queue_task('demo1', ['a', 'b'], dict(c=1, d=2), immediate=True)
+        env = s.queue_task('demo7')
+        drift = s.queue_task('demo1', ['a', 'b'], dict(c=1, d=2), period=93, prevent_drift=True)
+        self.db.commit()
+        self.writefunction(r"""
+def demo1(*args,**vars):
+    print 'you passed args=%s and vars=%s' % (args, vars)
+    return args[0]
+import random
+def demo7():
+    time.sleep(random.randint(1,5))
+    print W2P_TASK, request.now
+    return W2P_TASK.id, W2P_TASK.uuid, W2P_TASK.run_id
+""")
+        ret = self.exec_sched()
+        self.assertEqual(ret, 0)
+        # immediate check, can only check that nothing breaks
+        task1 = s.task_status(immediate.id)
+        res = [
+            ("tasks status completed", task1.status == 'COMPLETED'),
+        ]
+        self.exec_asserts(res, 'IMMEDIATE')
+
+        # drift check
+        task, task_run = self.fetch_results(s, drift)
+        res = [
+            ("task status completed", task.status == 'COMPLETED'),
+            ("next_run_time is exactly start_time + period", (task.next_run_time == task.start_time + datetime.timedelta(seconds=task.period)))
+        ]
+        self.exec_asserts(res, 'DRIFT')
+
+        # env check
+        task1 = s.task_status(env.id, output=True)
+        res = [
+            ("task %s returned W2P_TASK correctly" % (task1.scheduler_task.id),  task1.result == [task1.scheduler_task.id, task1.scheduler_task.uuid, task1.scheduler_run.id]),
+        ]
+        self.exec_asserts(res, 'ENV')
+
 
     def testRetryFailed(self):
         s = Scheduler(self.db)
-        failed = s.queue_task('demo2', retry_failed=1, period=5)
+        failed = s.queue_task('demo2', retry_failed=1, period=1)
-        failed_consecutive = s.queue_task('demo8', retry_failed=2, repeats=2, period=5)
+        failed_consecutive = s.queue_task('demo8', retry_failed=2, repeats=2, period=1)
         self.db.commit()
         self.writefunction(r"""
-
 def demo2():
     1/0
 
@@ -323,32 +477,50 @@ def demo8():
         # process finished just fine
         self.assertEqual(ret, 0)
         # failed - checks
-        info = s.task_status(failed.id)
+        task, task_run = self.fetch_results(s, failed)
-        task_runs = self.db(self.db.scheduler_run.task_id == info.id).select()
         res = [
-            ("task status failed", info.status == 'FAILED'),
+            ("task status failed", task.status == 'FAILED'),
-            ("task times_run is 0", info.times_run == 0),
+            ("task times_run is 0", task.times_run == 0),
-            ("task times_failed is 2", info.times_failed == 2),
+            ("task times_failed is 2", task.times_failed == 2),
-            ("task ran 2 times only", len(task_runs) == 2),
+            ("task ran 2 times only", len(task_run) == 2),
-            ("scheduler_run records are FAILED", (task_runs[0].status == task_runs[1].status == 'FAILED')),
+            ("scheduler_run records are FAILED", (task_run[0].status == task_run[1].status == 'FAILED')),
-            ("period is respected", (task_runs[1].start_time > task_runs[0].start_time + datetime.timedelta(seconds=info.period)))
+            ("period is respected", (task_run[1].start_time > task_run[0].start_time + datetime.timedelta(seconds=task.period)))
         ]
-        for a in res:
+        self.exec_asserts(res, 'FAILED')
-            self.assertEqual(a[1], True, msg=a[0])
 
         # failed consecutive - checks
-        info = s.task_status(failed_consecutive.id)
+        task, task_run = self.fetch_results(s, failed_consecutive)
-        task_runs = self.db(self.db.scheduler_run.task_id == info.id).select()
         res = [
-            ("task status completed", info.status == 'COMPLETED'),
+            ("task status completed", task.status == 'COMPLETED'),
-            ("task times_run is 2", info.times_run == 2),
+            ("task times_run is 2", task.times_run == 2),
-            ("task times_failed is 0", info.times_failed == 0),
+            ("task times_failed is 0", task.times_failed == 0),
-            ("task ran 6 times", len(task_runs) == 6),
+            ("task ran 6 times", len(task_run) == 6),
-            ("scheduler_run records for COMPLETED is 2", len([run.status for run in task_runs if run.status == 'COMPLETED']) == 2),
+            ("scheduler_run records for COMPLETED is 2", len([run.status for run in task_run if run.status == 'COMPLETED']) == 2),
-            ("scheduler_run records for FAILED is 4", len([run.status for run in task_runs if run.status == 'FAILED']) == 4),
+            ("scheduler_run records for FAILED is 4", len([run.status for run in task_run if run.status == 'FAILED']) == 4),
         ]
-        for a in res:
+        self.exec_asserts(res, 'FAILED_CONSECUTIVE')
-            self.assertEqual(a[1], True, msg=a[0])
+
+    def testHugeResult(self):
+        s = Scheduler(self.db)
+        huge_result = s.queue_task('demo10', retry_failed=1, period=1)
+        self.db.commit()
+        self.writefunction(r"""
+def demo10():
+    res = 'a' * 99999
+    return dict(res=res)
+""")
+        ret = self.exec_sched()
+        # process finished just fine
+        self.assertEqual(ret, 0)
+        # huge_result - checks
+        task = s.task_status(huge_result.id, output=True)
+        res = [
+            ("task status completed", task.scheduler_task.status == 'COMPLETED'),
+            ("task times_run is 1", task.scheduler_task.times_run == 1),
+            ("result is the correct one", task.result == dict(res='a' * 99999))
+        ]
+        self.exec_asserts(res, 'HUGE_RESULT')
+
 
 if __name__ == '__main__':
     unittest.main()