Dynamically-created ScheduledTask ignored about django-carrot HOT 4 CLOSED

There's a bug in carrot.management.commands.carrot here.

in carrot.scheduler.ScheduledTaskThread, there is an infinite loop that polls your app DB to make sure that the ScheduledTask object that the thread is running still exists. If it doesn't, then the thread stops itself. This produces the Current task has been removed from the queryset... message you are seeing, and this behaviour is expected.

However, the problem is in the .handle() method of the main carrot command. As with ScheduledTaskThread, this process starts a thread that regularly polls the app DB, in this case to check for newly added ScheduledTask objects, and creates a new ScheduledTaskThread if it finds anything new.

Unfortunately, there's a pretty big problem with the logic in the .handle() method, which is preventing your new ScheduledTask from being picked up.

I'm going to try and get a fix out for this today. Until then, you should be able to work around this by adding a delay between deleting one ScheduledTask and creating a new one. Something like this should do the trick:

    ScheduledTask.objects.filter(task_name__contains=self.pk).delete()
    time.sleep(3)
    if self.status == 1:

from django-carrot.

OK, I have a fix in place in V1.3.0a3. Here's how I tested it:

    first_task = create_scheduled_task('myapp.tasks.hello_world', {'seconds': 3})
    time.sleep(5)
    first_task.delete()
    second_task = create_scheduled_task('myapp.tasks.hello_world', {'seconds': 3})

If you fire up carrot v1.3.0a2 then run the above script, the first scheduled task will get created and triggered once, then get deleted. However, although the second scheduled task then gets created, the scheduler doesn't notice it, and it never actually gets triggered. This seems to be the same problem as you are reporting above

Trying the same thing in V1.3.0a3 results in the correct behaviour - the first scheduled task gets created, triggered once, deleted, then the second scheduled task gets created, picked up by the scheduler, and triggered every 3 seconds as expected as long as carrot is running.

Going to run 1.3.0a3 for a few days in our production environment here before I switch the alpha tags off.

from django-carrot.

Thanks very much for the fixes & attention!

Trying out V1.3.0a3 with the following tweaked model method, to address the case where an active object needs to change its schedule by calling its own save() method:

def save(self, *args, **kwargs):
    super().save(*args, **kwargs)
    self._setup_recurrence()
…
def _setup_recurrence(self):
    if self.status == 0 or self.reset_recur:
       ScheduledTask.objects.filter(task_name__contains=self.pk).delete()
       sleep(5)
    if self.status == 1:
       create_scheduled_task(
            recur_wrapper,
            {'seconds': self.recur_period},
            task_name=self.task_name,
            instance_pk=self.pk,
            model_name=self._meta.model_name,
            app_label=self._meta.app_label,
        )

Gives the following log errors. Step-by-step:

1. Object starts in inactive state. The carrot process is started:

$ sudo python manage.py carrot --logfile ../log/carrot.log 
Password:
WARNING: Carrot service is already running with the following PID. Running more than one instance of carrot may lead to a memory leak:
67006
found 0 scheduled tasks to run
Successfully started scheduler
Successfully started 2 consumers for queue default
All queues consumer sets started successfully. Full logs are at ../log/carrot.log.
default-consumer-1 2018-09-05 23:26:15,923 INFO:: Connecting to amqp://guest:guest@localhost:5672/%2f
default-consumer-2 2018-09-05 23:26:15,924 INFO:: Connecting to amqp://guest:guest@localhost:5672/%2f
default-consumer-2 2018-09-05 23:26:15,950 INFO:: Connection opened
default-consumer-1 2018-09-05 23:26:15,951 INFO:: Connection opened
default-consumer-2 2018-09-05 23:26:15,952 INFO:: Channel opened
default-consumer-1 2018-09-05 23:26:15,952 INFO:: Channel opened
default-consumer-2 2018-09-05 23:26:15,953 INFO:: Exchange declared
default-consumer-1 2018-09-05 23:26:15,954 INFO:: Exchange declared
default-consumer-2 2018-09-05 23:26:15,959 INFO:: Queue bound
default-consumer-2 2018-09-05 23:26:15,960 INFO:: Starting consumer default-consumer-2
default-consumer-1 2018-09-05 23:26:15,960 INFO:: Queue bound
default-consumer-1 2018-09-05 23:26:15,960 INFO:: Starting consumer default-consumer-1

2. Object is activated: status = 1, and save() is called:

New active scheduled tasks have been added to the queryset
adding new task utils.models.recur_wrapper
Publishing message utils.models.recur_wrapper
default-consumer-1 2018-09-05 23:27:58,437 INFO:: Consuming task utils.models.recur_wrapper, ID=4dea3321-d45d-42c8-81d3-b8df8ad76afa
default-consumer-1 2018-09-05 23:27:58,437 INFO:: default-consumer-1 2018-09-05 23:27:58,437 INFO:: Starting task utils.models.recur_wrapper
MyModel| Calling `recur_wrapper`...
default-consumer-1 2018-09-05 23:27:58,449 ERROR:: Task utils.models.recur_wrapper failed due to the following exception: A ScheduledTask with this task_name already exists. Please specific a unique name using the task_name parameter

3. It ran for a few iterations (60s interval), with the same errors. At this point, I checked if there really were multiple ScheduledTasks, but:

In [1]: ScheduledTask.objects.all()
Out[1]: <QuerySet [<ScheduledTask: utils.models.recur_wrapper>]>

In [2]: ScheduledTask.objects.all()[0].task_name
Out[2]: 'JqmJctdbmZa9P4AkuHjahf'

In [3]: MyModel.objects.filter(pk='ScheduledTask.objects.all()[0].task_name).count()
Out[3]: 1

4. Now the object is deactivated by setting status = 0 and calling save(), giving the following log messages:

New active scheduled tasks have been added to the queryset
adding new task 
Current task has been removed from the queryset. Stopping the thread
Current task has been removed from the queryset. Stopping the thread

PS: re: the PID error:

$ ps aux | grep 67006
tehfink         70971   0.0  0.0  4267768    524 s002  R+    7:46PM   0:00.00 grep 67006
root             67006   0.0  0.6  4415212 108292 s006  S+    7:26PM   0:04.96 python manage.py carrot --logfile ../log/carrot.log

from django-carrot.

Sorry for leaving this for a while; I had found the error a while ago, but have been traveling.

re: # 2, My _setup_recurrence code was missing an important piece. This line:
if self.status == 1:
should've been:
if self.status == 1 and not ScheduledTask.objects.filter(task_name__contains=self.pk).exists():
which resolves the IntegrityError: ERROR:: Task utils.models.recur_wrapper failed due to the following exception: A ScheduledTask with this task_name already exists. Please specific a unique name using the task_name parameter

Re: # 4, it seems that this section of code from carrot.py is activated even if a Task is deleted:
if new_qs.count() > len(self.pks) or newly_added:
because of this line that gives newly_added a value regardless:
newly_added = set(self.pks) - active_pks

For example, on loop 1 with an active task:
new_qs.count: 1, self.pks: [11], len(self.pks): 1, active_pks: {11}, newly_added: set()
Now the task is deleted on loop 2, resulting in:
new_qs.count: 0, self.pks: [11], len(self.pks): 1, active_pks: set(), newly_added: {11}

PS: I've since simplified the _setup_recurrence code on the model by using a OneToOneField:
task = models.OneToOneField('carrot.ScheduledTask', on_delete=models.CASCADE, editable=False, null=True, blank=True)
Are there any drawbacks or hidden pitfalls to using this method?

from django-carrot.