Add option to skip herd avoidance check, but still trigger avoidance for other tasks

[thenewguy]

I have a situation where I would like to use thundering herd avoidance conditionally.

It would be easy to support this with a conditional check at https://github.com/PolicyStat/jobtastic/blob/master/jobtastic/task.py#L242 and pop the key out of the options that get passed to the superclass.

Something like the following should do the trick:

        # Check for an in-progress equivalent task to avoid duplicating work
        if not options.pop('disable_thundering_herd_avoidance', False):
            task_id = cache.get('herd:%s' % cache_key)
            if task_id:
                logging.info('Found existing in-progress task: %s', task_id)
                return self.AsyncResult(task_id)

This would work nicely with #56

[thenewguy]

Let me know if this would be acceptable and I can work it into the implementation of #56

[winhamwr]

To make sure I understand, you're talking about adding the ability to "turn off" thundering herd avoidance at task "call time", rather than at task definition time, right?

I'm a bit wary of adding any special kwargs that jobtastic hijacks, since it introduces the potential of existing jobtastic usages colliding with those special kwargs. Even if we use a long namespace like jobtastic_herd_avoidance_timeout, it violates the "ideally, there should be exactly one way to do things" idea.

I think there's a way to achieve what you're looking for right now via subclassing:

class FooTask(JobtasticTask):
    significant_kwargs = [
        ('numerators', str),
        ('denominators', str),
    ]
    herd_avoidance_timeout = 60
    ...

class FooSansHerdAvoidanceTask(FooTask):
    herd_avoidance_timeout = 0

If that pattern doesn't actually work because of a Jobtastic implementation detail, then I'm super interested in making it work. It might also be a good use case to document.

What do you think of the subclassing option to solve this?

[thenewguy]

@winhamwr I didn't like that idea at first, but from the perspective of a reusable app I think you are definitely correct. That is the cleaner approach for extending celery tasks in a reusable app.

[thenewguy]

@winhamwr So I've been trying to figure out why the subclass approach isn't giving the results I was looking for written as suggested.

I think the boolean mentioned in the original code snippet is still required. It would just be a class attr instead of a method kwarg.

Here is why:
Depending on the condition, I need to be able to ensure that my task runs and is computed. In this case, I want to avoid skipping the task due to herd avoidance. However, I want other tasks to be able to latch on and use the forced run for herd avoidance. In order for this to happen I need to be able to disable herd avoidance without setting the herd_avoidance_timeout to 0. Otherwise, the forced run does not leave its footprint in the cache so the avoidable tasks cannot latch to it.

[thenewguy]

@winhamwr Do you have a suggestion for how to test this? I've submitted a PR that implements/documents the class attr bypass_herd_avoidance

[thenewguy]

err I didn't submit a PR I created a branch on my fork and pushed the changes

[winhamwr]

@thenewguy I meant to take a look at this today. Sorry for the delay. I plan on taking a close look, tomorrow.

[thenewguy]

@winhamwr Here is a direct link to the commit: thenewguy@410d735 I've been using it for awhile now and haven't noticed any issues

[thenewguy]

@winhamwr not trying to pester you... but just checking back in on this

[winhamwr]

@thenewguy thanks for your persistance! I appreciate it.

I just re-reviewed #58 and it's ready to go with a very minor documentation tweak.

As far as thenewguy/jobtastic@410d735, I think I now understand your use case based on rereading your March 16 comment. I'm trying to think of what name for the attr best-communicates that idea.

bypass_thundering_herd_avoidance makes me think that it would be avoided altogether, when in fact we're only ignoring checking herd avoidance, not setting herd avoidance. I'm struggling with an alternative name that makes that distinction clear, though.

• always_start_new_herd?
• herd_leader_only?

Since I can't think of anything that concisely explains the behavior, I lean towards something weird enough that motivates reading of the documentation. Thoughts?

[winhamwr]

Also, for this feature, we definitely need an example in the README of how it could be used.

So you're setting cache_prefix to the same thing for two different task classes and having one of them be the "always run, no matter what" and the other be the "only run if I'm the only one running"? Am I understanding things correctly?

[thenewguy]

Still thinking about your attr naming comment but wanted to comment on the other...

Correct.

Forgive me if I am slightly rusty on the specifics of this now, but from memory and reading back through my code comments I encountered a race condition that made this distinction important. I can become more familiar if necessary but I don't think it will be at this point.

Consider the use case of computing a hash based etag on a very large remotely stored file (aka hash generation is time consuming) that is used in the response url for long term caching. When the content that the hash is generated from changes, we know the etag needs to be recomputed. In this case we want to ensure the task is queued to prevent a stale etag. However, when generating redirect links based on the etag we do not actually want to compute a new etag. But, in this case, we do want to wait for the etag to update if it hasn't already to prevent redirecting to the stale content, so we latch onto the currently running job before returning

[thenewguy]

from __future__ import absolute_import

from django.apps import apps
from task_mixins.tasks import LockingTask


class ComputeEtagTaskWithHerdAvoidance(LockingTask):
    #
    # CELERY CONFIG
    #
    acks_late = True

    #
    # JOBTASTIC CONFIG
    #
    herd_avoidance_timeout = 7200
    significant_kwargs = [
        ("app_label", str),
        ("model_name", str),
        ("pk", str),
    ]

    def calculate_result_with_lock(self, app_label, model_name, pk):
        model_cls = apps.get_model(app_label, model_name)
        try:
            instance = model_cls.objects.get(pk=pk)
        except model_cls.DoesNotExist as exc:
            raise self.retry(exc=exc, countdown=1)
        etag = instance.compute_etag()
        if etag != instance.etag:
            instance.etag = etag
            instance.etag_clear_on_save = False
            instance.save(update_fields=['etag'])
        return etag


class ComputeEtagTask(ComputeEtagTaskWithHerdAvoidance):
    bypass_herd_avoidance = True

    @classmethod
    def _get_cache_key(cls, **kwargs):
        return ComputeEtagTaskWithHerdAvoidance._get_cache_key(**kwargs)

[winhamwr]

Consider the use case of computing a hash based etag on a very large remotely stored file

That makes perfect sense. Thanks for the explanation and for sharing the code sample.

def _get_cache_key(cls, **kwargs):

Overriding _get_cache_key totally works. You should be able to accomplish the same thing by setting a cache_prefix on ComputeEtagTaskWithHerdAvoidance. That setting was intended so you could have tasks that share herd/cache behavior without requiring subclasses.

Re-reading the docs on that setting, I did a really bad job of explaining that. I'm glad you were able to get the desired behavior, anyway.

[thenewguy]

@winhamwr I think part of this is similar to #68

[thenewguy]

This is my LockingTask mixin that covers some of your caveats:

class LockingTask(ExtendedJobtasticTask):
    abstract = True

    max_retry_countdown = 60
    lock_max_age = 60

    # disable herd avoidance and caching because it
    # is important that the task runs every time
    # the values change for an object by default
    herd_avoidance_timeout = 0
    cache_duration = -1

    def calculate_result(self, *args, **kwargs):
        lock_name = self._get_cache_key(**kwargs)
        max_age = self.lock_max_age
        try:
            with NonBlockingLock.objects.acquire_lock(lock_name=lock_name, max_age=max_age):
                return self.calculate_result_with_lock(*args, **kwargs)
        except AlreadyLocked as exc:
            retry_limit = self.request.retries + 1
            retry_countdown = min(2 * retry_limit, self.max_retry_countdown)
            raise self.retry(exc=exc, countdown=retry_countdown,
                             max_retries=retry_limit)

    def calculate_result_with_lock(self, *args, **kwargs):
        raise NotImplementedError(
            "Subclass must define `calculate_result_with_lock` method.")

ExtendedJobtasticTask just implements my changes for #56 and this one

[thenewguy]

I'm not sure how I lost track of this, but I've submitted a PR with the approved name

#81

[thenewguy]

@winhamwr I've submitted a second PR so that you can pick one if you like and merge. PR #82 additionally issues a task revoke if the cached task isn't completed

[thenewguy]

@winhamwr any chance on merging? this one is killing me slowly <3