Adds new remote control command: time_limit(task_name, soft=None, hard=None)

celery/concurrency/processes/pool.py

@@ -273,7 +273,6 @@ class TimeoutHandler(PoolThread):
         self.cache = cache
         self.t_soft = t_soft
         self.t_hard = t_hard
-        self.putlock = putlock
         super(TimeoutHandler, self).__init__()
 
     def body(self):

celery/task/control.py

@@ -155,7 +155,7 @@ class Control(object):
     def rate_limit(self, task_name, rate_limit, destination=None, **kwargs):
         """Set rate limit for task by type.
 
-        :param task_name: Type of task to change rate limit for.
+        :param task_name: Name of task to change rate limit for.
         :param rate_limit: The rate limit as tasks per second, or a rate limit
             string (`"100/m"`, etc.
             see :attr:`celery.task.base.Task.rate_limit` for
@@ -176,6 +176,21 @@ class Control(object):
                                          "rate_limit": rate_limit},
                               **kwargs)
 
+    def time_limit(self, task_name, soft=None, hard=None, **kwargs):
+        """Set time limits for task by type.
+
+        :param task_name: Name of task to change time limits for.
+        :keyword soft: New soft time limit (in seconds).
+        :keyword hard: New hard time limit (in seconds).
+
+        Any additional keyword arguments are passed on to :meth:`broadcast`.
+
+        """
+        return self.broadcast("time_limit",
+                              arguments={"task_name": task_name,
+                                         "hard": hard, "soft": soft},
+                              **kwargs)
+
     def broadcast(self, command, arguments=None, destination=None,
             connection=None, connect_timeout=None, reply=False, timeout=1,
             limit=None, callback=None, channel=None):
@@ -214,6 +229,7 @@ class Control(object):
 _default_control = Control(app_or_default())
 broadcast = _default_control.broadcast
 rate_limit = _default_control.rate_limit
+time_limit = _default_control.time_limit
 ping = _default_control.ping
 revoke = _default_control.revoke
 discard_all = _default_control.discard_all

celery/worker/control/builtins.py

@@ -98,6 +98,24 @@ def rate_limit(panel, task_name, rate_limit, **kwargs):
     return {"ok": "new rate limit set successfully"}
 
 
+@Panel.register
+def time_limit(panel, task_name=None, hard=None, soft=None, **kwargs):
+    try:
+        task = tasks[task_name]
+    except KeyError:
+        panel.logger.error(
+            "Change time limit attempt for unknown task %s" % (task_name, ))
+        return {"error": "unknown task"}
+
+    task.soft_time_limit = soft
+    task.time_limit = hard
+
+    panel.logger.info(
+        "New time limits for tasks of type %s: soft=%s hard=%s" % (
+            task_name, soft, hard))
+    return {"ok": "time limits set successfully"}
+
+
 @Panel.register
 def dump_schedule(panel, safe=False, **kwargs):
     schedule = panel.consumer.eta_schedule.schedule

docs/userguide/workers.rst

@@ -138,7 +138,27 @@ Time limits can also be set using the :setting:`CELERYD_TASK_TIME_LIMIT` /
 
 .. note::
 
-    Time limits do not currently work on Windows.
+    Time limits do not currently work on Windows and other
+    platforms that do not support the ``SIGUSR1`` signal.
+
+
+Changing time limits at runtime
+-------------------------------
+.. versionadded:: 2.3
+
+You can change the soft and hard time limits for a task by using the
+``time_limit`` remote control command.
+
+Example changing the time limit for the ``tasks.crawl_the_web`` task
+to have a soft time limit of one minute, and a hard time limit of
+two minutes::
+
+    >>> from celery.task import control
+    >>> control.time_limit("tasks.crawl_the_web",
+                           soft=60, hard=120, reply=True)
+    [{'worker1.example.com': {'ok': 'time limits set successfully'}}]
+
+Only tasks that starts executing after the time limit change will be affected.
 
 .. _worker-maxtasksperchild: