Merge branch 'master' of https://github.com/celery/celery into autoretry

celery/app/task.py

@@ -476,6 +476,7 @@ class Task(object):
             'chord': request.chord,
             'soft_time_limit': limit_soft,
             'time_limit': limit_hard,
+            'reply_to': request.reply_to,
         }
         options.update(
             {'queue': queue} if queue else (request.delivery_info or {})

celery/beat.py

@@ -209,10 +209,15 @@ class Scheduler(object):
         else:
             debug('%s sent. id->%s', entry.task, result.id)
 
+    def adjust(self, n, drift=-0.010):
+        if n and n > 0:
+            return n + drift
+        return n
+
     def is_due(self, entry):
         return entry.is_due()
 
-    def tick(self, drift=-0.010, event_t=event_t, min=min,
+    def tick(self, event_t=event_t, min=min,
              heappop=heapq.heappop, heappush=heapq.heappush,
              heapify=heapq.heapify):
         """Run a tick, that is one iteration of the scheduler.
@@ -220,10 +225,11 @@ class Scheduler(object):
         Executes all due tasks.
 
         """
+        adjust = self.adjust
         max_interval = self.max_interval
         H = self._heap
         if H is None:
-            H = self._heap = [event_t(e.is_due()[1] + drift or 0, 5, e)
+            H = self._heap = [event_t(adjust(e.is_due()[1]) or 0, 5, e)
                               for e in values(self.schedule)]
             heapify(H)
         event = H[0]
@@ -239,8 +245,7 @@ class Scheduler(object):
             else:
                 heappush(H, verify)
                 return min(verify[0], max_interval)
-        return min(next_time_to_run + drift if next_time_to_run
-                   else max_interval, max_interval)
+        return min(adjust(next_time_to_run) or max_interval, max_interval)
 
     def should_sync(self):
         return (

celery/tests/app/test_beat.py

@@ -287,7 +287,7 @@ class test_Scheduler(AppCase):
         scheduler = mScheduler(app=self.app)
         scheduler.add(name='test_pending_tick',
                       schedule=always_pending)
-        self.assertEqual(scheduler.tick(), 1)
+        self.assertEqual(scheduler.tick(), 1 - 0.010)
 
     def test_honors_max_interval(self):
         scheduler = mScheduler(app=self.app)
@@ -303,7 +303,7 @@ class test_Scheduler(AppCase):
                  {'schedule': mocked_schedule(False, j)})
                  for i, j in enumerate(nums))
         scheduler.update_from_dict(s)
-        self.assertEqual(scheduler.tick(), min(nums))
+        self.assertEqual(scheduler.tick(), min(nums) - 0.010)
 
     def test_schedule_no_remain(self):
         scheduler = mScheduler(app=self.app)

docs/configuration.rst

@@ -335,7 +335,7 @@ AMQP backend settings
 .. note::
 
     The AMQP backend requires RabbitMQ 1.1.0 or higher to automatically
-    expire results.  If you are running an older version of RabbitmQ
+    expire results.  If you are running an older version of RabbitMQ
     you should disable result expiration like this:
 
         CELERY_TASK_RESULT_EXPIRES = None
@@ -455,7 +455,7 @@ which is the same as::
 
     CELERY_RESULT_BACKEND = 'redis://'
 
-The fields of the URL is defined as folows:
+The fields of the URL are defined as follows:
 
 - *host*
 
@@ -657,7 +657,7 @@ which is the same as::
 
     CELERY_RESULT_BACKEND = "riak://"
 
-The fields of the URL is defined as folows:
+The fields of the URL are defined as follows:
 
 - *host*
 

docs/contributing.rst

@@ -639,7 +639,7 @@ Edit the file using your favorite editor:
 
     $ vim celery.worker.awesome.rst
 
-        # change every occurance of ``celery.schedules`` to
+        # change every occurrence of ``celery.schedules`` to
         # ``celery.worker.awesome``
 
 

docs/getting-started/introduction.rst

@@ -281,7 +281,7 @@ Quickjump
         - :ref:`see a list of running workers <monitoring-control>`
         - :ref:`purge all messages <monitoring-control>`
         - :ref:`inspect what the workers are doing <monitoring-control>`
-        - :ref:`see what tasks a worker has registerd <monitoring-control>`
+        - :ref:`see what tasks a worker has registered <monitoring-control>`
         - :ref:`migrate tasks to a new broker <monitoring-control>`
         - :ref:`see a list of event message types <event-reference>`
         - :ref:`contribute to Celery <contributing>`

docs/getting-started/next-steps.rst

@@ -92,7 +92,7 @@ When the worker starts you should see a banner and some messages::
 
      [2012-06-08 16:23:51,078: WARNING/MainProcess] celery@halcyon.local has started.
 
--- The *broker* is the URL you specifed in the broker argument in our ``celery``
+-- The *broker* is the URL you specified in the broker argument in our ``celery``
 module, you can also specify a different broker on the command-line by using
 the :option:`-b` option.
 

docs/glossary.rst

@@ -51,7 +51,7 @@ Glossary
         Idempotence is a mathematical property that describes a function that
         can be called multiple times without changing the result.
         Practically it means that a function can be repeated many times without
-        unintented effects, but not necessarily side-effect free in the pure
+        unintended effects, but not necessarily side-effect free in the pure
         sense (compare to :term:`nullipotent`).
 
     nullipotent

docs/userguide/application.rst

@@ -182,7 +182,7 @@ Example 2: Using a configuration module
 
 .. tip::
 
-    Using the name of a module is recomended
+    Using the name of a module is recommended
     as this means that the module doesn't need to be serialized
     when the prefork pool is used.  If you're
     experiencing configuration pickle errors then please try using

docs/userguide/calling.rst

@@ -47,7 +47,7 @@ The API defines a standard set of execution options, as well as three methods:
         executes 10 seconds from now.
 
     - ``T.apply_async(eta=now + timedelta(seconds=10))``
-        executes 10 seconds from now, specifed using ``eta``
+        executes 10 seconds from now, specified using ``eta``
 
     - ``T.apply_async(countdown=60, expires=120)``
         executes in one minute from now, but expires after 2 minutes.
@@ -447,7 +447,7 @@ Though this particular example is much better expressed as a group:
     >>> from celery import group
 
     >>> numbers = [(2, 2), (4, 4), (8, 8), (16, 16)]
-    >>> res = group(add.s(n) for i in numbers).apply_async()
+    >>> res = group(add.s(n) for n in numbers).apply_async()
 
     >>> res.get()
     [4, 8, 16, 32]

docs/userguide/canvas.rst

@@ -109,7 +109,7 @@ creates partials:
 - Any arguments added will be prepended to the args in the signature::
 
     >>> partial = add.s(2)          # incomplete signature
-    >>> partial.delay(4)            # 2 + 4
+    >>> partial.delay(4)            # 4 + 2
     >>> partial.apply_async((4, ))  # same
 
 - Any keyword arguments added will be merged with the kwargs in the signature,
@@ -125,7 +125,7 @@ creates partials:
     >>> s = add.signature((2, 2), countdown=10)
     >>> s.apply_async(countdown=1)  # countdown is now 1
 
-You can also clone signatures to create derivates:
+You can also clone signatures to create derivatives:
 
     >>> s = add.s(2)
     proj.tasks.add(2)

docs/userguide/extending.rst

@@ -768,7 +768,7 @@ will take some time so other transports still use a threading-based solution.
 
     Add callback to be called when ``fd`` is readable.
 
-    The callback will stay registered until explictly removed using
+    The callback will stay registered until explicitly removed using
     :meth:`hub.remove(fd) <hub.remove>`, or the fd is automatically discarded
     because it's no longer valid.
 

docs/userguide/routing.rst

@@ -402,7 +402,7 @@ using the ``basic.publish`` command::
 
 Now that the message is sent you can retrieve it again.  You can use the
 ``basic.get``` command here, which polls for new messages on the queue
-(which is alright for maintainence tasks, for services you'd want to use
+(which is alright for maintenance tasks, for services you'd want to use
 ``basic.consume`` instead)
 
 Pop a message off the queue::

docs/userguide/tasks.rst

@@ -19,7 +19,7 @@ many messages in advance and even if the worker is killed -- caused by power fai
 or otherwise -- the message will be redelivered to another worker.
 
 Ideally task functions should be :term:`idempotent`, which means that
-the function will not cause unintented effects even if called
+the function will not cause unintended effects even if called
 multiple times with the same arguments.
 Since the worker cannot detect if your tasks are idempotent, the default
 behavior is to acknowledge the message in advance, before it's executed,