Scheduler task runs before it is supposed to run

[Bernardo Leon]

Hi, I am trying to run tasks in the future based on a time the user specifies but as soon as the task is queued it is executed, what am I doing wrong?

I am using web2py Version 2.14.6-stable+timestamp.2016.05.10.00.21.47

This is my model.db (part of it)

db = DAL('sqlite://recomendadorexpertos.db')

db.define_table('topicos',

               Field('curso', 'reference cursos'),

               Field('nombre', 'string', required=True, length=512),

               Field('palabras_clave', 'text', readable=False, required=True, length=2048),

               Field('fecha_inicio_identificacion', 'date', required=True, comment=T('Indica la fecha desde la cual se empezará a buscar expertos.')),

               Field('fecha_fin_identificacion', 'date', required=True, comment=T('Indica la fehca en la cual se publicarán las recomendaciónes de expertos.')),

               Field('fecha_fin_publicacion', 'date', comment=T('Indica la fecha hasta cuando se publicaran las recomendaciones de expertos (opcional)')),

               Field('timestamp_get_tweets_etiquetado', 'datetime', readable=False, writable=False), #Para mostrar al usuario desde cuando se estan descargando tweets para etiquetado.

               Field('link_descarga_tweets_etiquetado', 'string', readable=False, writable=False, length=2048),

               Field('upload_archivo_etiquetado', 'upload', length=524288, uploadfolder='/var/www/web2py/applications/RecomendadorExpertos/archivos_etiquetados/', autodelete=True, comment=T('Primero inicie el proceso de descarga de tweets para este topico, descargue el archivo para etiquetado, etiquetelo y subalo aqui.'), label=T('Subir archivo etiquetado'), readable=False), #Cambiar el nombre del archivo que suba a [nombre_topico].csv

               Field('path_modelo', 'string', readable=False, writable=False, length=2048), #Ruta del modelo entrenado

               Field('topico_configurado_completamente', 'integer', readable=True, writable=False, default=1, represent=representTopicoConfigurado), #Sirve para mostrar un check alado del archivo subido una vez terminado de generarse el modelo, para que el profesor sepa que ya esta todo bien.

               Field('id_scheduler_task_descarga_tweets', 'integer', readable=False, writable=False),

               Field('id_scheduler_task_identificacion_expertos', 'integer', readable=False, writable=False),

               Field('id_scheduler_task_publicacion_expertos', 'integer', readable=False, writable=False))

This is my Scheduler.py

def task_iniciar_descarga_tweets(row):

    return 'descarga de tweets terminada'

def task_iniciar_identificacion_expertos(nombre_topico):

    return 'identificacion de expertos terminada'

def task_publicar_expertos(nombre_topico):

    return 'publicacion de expertos terminada'

dbScheduler = DAL('sqlite://scheduler.db')

scheduler = Scheduler(dbScheduler, tasks={'task_iniciar_descarga_tweets':task_iniciar_descarga_tweets,

                                          'task_iniciar_identificacion_expertos':task_iniciar_identificacion_expertos,

                                          'task_publicar_expertos':task_publicar_expertos})

And this is my default.py controller (part of it)

@auth.requires_login()

def onCreateTopico(form):

    row_task_iniciar_identificacion = scheduler.queue_task('task_iniciar_identificacion_expertos', pvars={'nombre_topico':form.vars.nombre}, start_time=datetime.datetime.combine(form.vars.fecha_inicio_identificacion, datetime.time.min), stop_time=datetime.datetime.combine(form.vars.fecha_fin_identificacion, datetime.time.min), retry_failed=3)

    row_task_iniciar_publicacion = scheduler.queue_task('task_publicar_expertos', pvars={'nombre_topico':form.vars.nombre}, start_time=datetime.datetime.combine(form.vars.fecha_fin_identificacion, datetime.time.min), retry_failed=3)

    

    db(db.topicos.id==form.vars.id).update(id_scheduler_task_identificacion_expertos=row_task_iniciar_identificacion.id, id_scheduler_task_publicacion_expertos=row_task_iniciar_publicacion.id)

    db.commit()

For instance, if I queue the task "task_iniciar_identificacion_expertos" with a future start_time the worker executes this task inmediately, how can I set the worker to execute this task in the future?

Thank you!

[Niphlod]

sorry, a bug slipped in. In the wait for the fix, please use next_run_time in addition to start_time.

[Bernardo Leon]

Thank you for your reply. Can you reply this thread when the fix is published?

I don't know if this is related to the same bug but in my worker process terminal I have this message:

ERROR:web2py.scheduler.cedia-Satellite-L55-C#18606:    error popping tasks

why do I have this error? Thanks.

[Niphlod]

that error is due - mostly - when the backend is too much pressured (i.e. you're using sqlite, or too many workers, or a too low heatbeat)

this is the fix  https://github.com/web2py/web2py/pull/1453 . you can subscribe to it to know when it'll be merged.

[Bernardo Leon]

Niphlod I will use postgresql in production and your solution as well. Thank you for helping me! ;)

[Niphlod]

BTW: it has been merged . You can safely take scheduler.py from master and overwrite your own if you want to avid specifying next_run_time