title: Celery Beat: Limit to single task instance at a time
body: <p>I have celery beat and celery (four workers) to do some processing steps in bulk. One of those tasks is roughly along the lines of, "for each X that hasn't had a Y created, create a Y."</p>
  
  <p>The task is run periodically at a semi-rapid rate (10sec). The task completes very quickly. There are other tasks going on as well.</p>
  
  <p>I've run into the issue multiple times in which the beat tasks apparently become backlogged, and so the same task (from different beat times) are executed simultaneously, causing incorrectly duplicated work. It also appears that the tasks are executed out-of-order.</p>
  
  <ol>
  <li><p>Is it possible to limit celery beat to ensure only one outstanding instance of a task at a time? Is setting something like <code>rate_limit=5</code> on the task the "correct" way to of doing this?</p></li>
  <li><p>Is it possible to ensure that beat tasks are executed in-order, e.g. instead of dispatching a task, beat adds it to a task chain?</p></li>
  <li><p>What's the best way of handling this, short of making those tasks themselves execute atomically and are safe to be executed concurrently? That was not a restriction I would have expected of beat tasks…</p></li>
  </ol>
  
  <p>The task itself is defined naïvely:
  </p>
  
  <pre><code>@periodic_task(run_every=timedelta(seconds=10))
  def add_y_to_xs():
      # Do things in a database
      return
  </code></pre>
  
  <p>Here's an actual (cleaned) log:</p>
  
  <ul>
  <li><code>[00:00.000]</code> foocorp.tasks.add_y_to_xs sent. id->#1</li>
  <li><code>[00:00.001]</code> Received task: foocorp.tasks.add_y_to_xs[#1]</li>
  <li><code>[00:10.009]</code> foocorp.tasks.add_y_to_xs sent. id->#2</li>
  <li><code>[00:20.024]</code> foocorp.tasks.add_y_to_xs sent. id->#3</li>
  <li><code>[00:26.747]</code> Received task: foocorp.tasks.add_y_to_xs[#2]</li>
  <li><code>[00:26.748]</code> TaskPool: Apply #2</li>
  <li><code>[00:26.752]</code> Received task: foocorp.tasks.add_y_to_xs[#3]</li>
  <li><code>[00:26.769]</code> Task accepted: foocorp.tasks.add_y_to_xs[#2] pid:26528</li>
  <li><code>[00:26.775]</code> Task foocorp.tasks.add_y_to_xs[#2] succeeded in 0.0197986490093s: None</li>
  <li><code>[00:26.806]</code> TaskPool: Apply #1</li>
  <li><code>[00:26.836]</code> TaskPool: Apply #3</li>
  <li><code>[01:30.020]</code> Task accepted: foocorp.tasks.add_y_to_xs[#1] pid:26526</li>
  <li><code>[01:30.053]</code> Task accepted: foocorp.tasks.add_y_to_xs[#3] pid:26529</li>
  <li><code>[01:30.055]</code> foocorp.tasks.add_y_to_xs[#1]: Adding Y for X id #9725</li>
  <li><code>[01:30.070]</code> foocorp.tasks.add_y_to_xs[#3]: Adding Y for X id #9725</li>
  <li><code>[01:30.074]</code> Task foocorp.tasks.add_y_to_xs[#1] succeeded in 0.0594762689434s: None</li>
  <li><code>[01:30.087]</code> Task foocorp.tasks.add_y_to_xs[#3] succeeded in 0.0352867960464s: None</li>
  </ul>
  
  <p>We're currently using Celery 3.1.4 with RabbitMQ as the transport.</p>
  
  <p><strong>EDIT</strong> Dan, here's what I came up with:</p>
  
  <p>Dan, here's what I ended up using:</p>
  
  <pre class="lang-py prettyprint-override"><code>from sqlalchemy import func
  from sqlalchemy.exc import DBAPIError
  from contextlib import contextmanager
  
  
  def _psql_advisory_lock_blocking(conn, lock_id, shared, timeout):
      lock_fn = (func.pg_advisory_xact_lock_shared
                 if shared else
                 func.pg_advisory_xact_lock)
      if timeout:
          conn.execute(text('SET statement_timeout TO :timeout'),
                       timeout=timeout)
      try:
          conn.execute(select([lock_fn(lock_id)]))
      except DBAPIError:
          return False
      return True
  
  
  def _psql_advisory_lock_nonblocking(conn, lock_id, shared):
      lock_fn = (func.pg_try_advisory_xact_lock_shared
                 if shared else
                 func.pg_try_advisory_xact_lock)
      return conn.execute(select([lock_fn(lock_id)])).scalar()
  
  
  class DatabaseLockFailed(Exception):
      pass
  
  
  @contextmanager
  def db_lock(engine, name, shared=False, block=True, timeout=None):
      """
      Context manager which acquires a PSQL advisory transaction lock with a
      specified name.
      """
      lock_id = hash(name)
  
      with engine.begin() as conn, conn.begin():
          if block:
              locked = _psql_advisory_lock_blocking(conn, lock_id, shared,
                                                    timeout)
          else:
              locked = _psql_advisory_lock_nonblocking(conn, lock_id, shared)
          if not locked:
              raise DatabaseLockFailed()
          yield
  </code></pre>
  
  <p>And the celery task decorator (used only for periodic tasks):
  </p>
  
  <pre><code>from functools import wraps
  from preo.extensions import db
  
  
  def locked(name=None, block=True, timeout='1s'):
      """
      Using a PostgreSQL advisory transaction lock, only runs this task if the
      lock is available. Otherwise logs a message and returns `None`.
      """
      def with_task(fn):
          lock_id = name or 'celery:{}.{}'.format(fn.__module__, fn.__name__)
  
          @wraps(fn)
          def f(*args, **kwargs):
              try:
                  with db_lock(db.engine, name=lock_id, block=block,
                               timeout=timeout):
                      return fn(*args, **kwargs)
              except DatabaseLockFailed:
                  logger.error('Failed to get lock.')
                  return None
          return f
      return with_task
  </code></pre>