workers.rst 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600
  1. .. _guide-worker:
  2. ===============
  3. Workers Guide
  4. ===============
  5. .. contents::
  6. :local:
  7. .. _worker-starting:
  8. Starting the worker
  9. ===================
  10. You can start celeryd to run in the foreground by executing the command::
  11. $ celeryd --loglevel=INFO
  12. You probably want to use a daemonization tool to start
  13. `celeryd` in the background. See :ref:`daemonizing` for help
  14. using `celeryd` with popular daemonization tools.
  15. For a full list of available command line options see
  16. :mod:`~celery.bin.celeryd`, or simply do::
  17. $ celeryd --help
  18. You can also start multiple workers on the same machine. If you do so
  19. be sure to give a unique name to each individual worker by specifying a
  20. host name with the :option:`--hostname|-n` argument::
  21. $ celeryd --loglevel=INFO --concurrency=10 -n worker1.example.com
  22. $ celeryd --loglevel=INFO --concurrency=10 -n worker2.example.com
  23. $ celeryd --loglevel=INFO --concurrency=10 -n worker3.example.com
  24. .. _worker-stopping:
  25. Stopping the worker
  26. ===================
  27. Shutdown should be accomplished using the :sig:`TERM` signal.
  28. When shutdown is initiated the worker will finish all currently executing
  29. tasks before it actually terminates, so if these tasks are important you should
  30. wait for it to finish before doing anything drastic (like sending the :sig:`KILL`
  31. signal).
  32. If the worker won't shutdown after considerate time, for example because
  33. of tasks stuck in an infinite-loop, you can use the :sig:`KILL` signal to
  34. force terminate the worker, but be aware that currently executing tasks will
  35. be lost (unless the tasks have the :attr:`~@Task.acks_late`
  36. option set).
  37. Also as processes can't override the :sig:`KILL` signal, the worker will
  38. not be able to reap its children, so make sure to do so manually. This
  39. command usually does the trick::
  40. $ ps auxww | grep celeryd | awk '{print $2}' | xargs kill -9
  41. .. _worker-restarting:
  42. Restarting the worker
  43. =====================
  44. Other than stopping then starting the worker to restart, you can also
  45. restart the worker using the :sig:`HUP` signal::
  46. $ kill -HUP $pid
  47. The worker will then replace itself with a new instance using the same
  48. arguments as it was started with.
  49. .. _worker-concurrency:
  50. Concurrency
  51. ===========
  52. By default multiprocessing is used to perform concurrent execution of tasks,
  53. but you can also use :ref:`Eventlet <concurrency-eventlet>`. The number
  54. of worker processes/threads can be changed using the :option:`--concurrency`
  55. argument and defaults to the number of CPUs available on the machine.
  56. .. admonition:: Number of processes (multiprocessing)
  57. More worker processes are usually better, but there's a cut-off point where
  58. adding more processes affects performance in negative ways.
  59. There is even some evidence to support that having multiple celeryd's running,
  60. may perform better than having a single worker. For example 3 celeryd's with
  61. 10 worker processes each. You need to experiment to find the numbers that
  62. works best for you, as this varies based on application, work load, task
  63. run times and other factors.
  64. .. _worker-persistent-revokes:
  65. Persistent revokes
  66. ==================
  67. Revoking tasks works by sending a broadcast message to all the workers,
  68. the workers then keep a list of revoked tasks in memory.
  69. If you want tasks to remain revoked after worker restart you need to
  70. specify a file for these to be stored in, either by using the `--statedb`
  71. argument to :mod:`~celery.bin.celeryd` or the :setting:`CELERYD_STATE_DB`
  72. setting. See :setting:`CELERYD_STATE_DB` for more information.
  73. Note that remote control commands must be working for revokes to work.
  74. Remote control commands are only supported by the amqp, redis and mongodb
  75. transports at this point.
  76. .. _worker-time-limits:
  77. Time limits
  78. ===========
  79. .. versionadded:: 2.0
  80. :supported pools: processes
  81. A single task can potentially run forever, if you have lots of tasks
  82. waiting for some event that will never happen you will block the worker
  83. from processing new tasks indefinitely. The best way to defend against
  84. this scenario happening is enabling time limits.
  85. The time limit (`--time-limit`) is the maximum number of seconds a task
  86. may run before the process executing it is terminated and replaced by a
  87. new process. You can also enable a soft time limit (`--soft-time-limit`),
  88. this raises an exception the task can catch to clean up before the hard
  89. time limit kills it:
  90. .. code-block:: python
  91. from myapp import celery
  92. from celery.exceptions import SoftTimeLimitExceeded
  93. @celery.task
  94. def mytask():
  95. try:
  96. do_work()
  97. except SoftTimeLimitExceeded:
  98. clean_up_in_a_hurry()
  99. Time limits can also be set using the :setting:`CELERYD_TASK_TIME_LIMIT` /
  100. :setting:`CELERYD_SOFT_TASK_TIME_LIMIT` settings.
  101. .. note::
  102. Time limits do not currently work on Windows and other
  103. platforms that do not support the ``SIGUSR1`` signal.
  104. Changing time limits at runtime
  105. -------------------------------
  106. .. versionadded:: 2.3
  107. You can change the soft and hard time limits for a task by using the
  108. ``time_limit`` remote control command.
  109. Example changing the time limit for the ``tasks.crawl_the_web`` task
  110. to have a soft time limit of one minute, and a hard time limit of
  111. two minutes::
  112. >>> celery.control.time_limit("tasks.crawl_the_web",
  113. soft=60, hard=120, reply=True)
  114. [{'worker1.example.com': {'ok': 'time limits set successfully'}}]
  115. Only tasks that starts executing after the time limit change will be affected.
  116. .. _worker-maxtasksperchild:
  117. Max tasks per child setting
  118. ===========================
  119. .. versionadded:: 2.0
  120. :supported pools: processes
  121. With this option you can configure the maximum number of tasks
  122. a worker can execute before it's replaced by a new process.
  123. This is useful if you have memory leaks you have no control over
  124. for example from closed source C extensions.
  125. The option can be set using the `--maxtasksperchild` argument
  126. to `celeryd` or using the :setting:`CELERYD_MAX_TASKS_PER_CHILD` setting.
  127. .. _worker-autoreload:
  128. Autoreloading
  129. =============
  130. .. versionadded:: 2.5
  131. :supported pools: processes, eventlet, gevent, threads, solo
  132. Starting :program:`celeryd` with the :option:`--autoreload` option will
  133. enable the worker to watch for file system changes to all imported task
  134. modules imported (and also any non-task modules added to the
  135. :setting:`CELERY_IMPORTS` setting or the :option:`-I|--include` option).
  136. This is an experimental feature intended for use in development only,
  137. using auto-reload in production is discouraged as the behavior of reloading
  138. a module in Python is undefined, and may cause hard to diagnose bugs and
  139. crashes. Celery uses the same approach as the auto-reloader found in e.g.
  140. the Django ``runserver`` command.
  141. When auto-reload is enabled the worker starts an additional thread
  142. that watches for changes in the file system. New modules are imported,
  143. and already imported modules are reloaded whenever a change is detected,
  144. and if the processes pool is used the child processes will finish the work
  145. they are doing and exit, so that they can be replaced by fresh processes
  146. effectively reloading the code.
  147. File system notification backends are pluggable, and it comes with three
  148. implementations:
  149. * inotify (Linux)
  150. Used if the :mod:`pyinotify` library is installed.
  151. If you are running on Linux this is the recommended implementation,
  152. to install the :mod:`pyinotify` library you have to run the following
  153. command::
  154. $ pip install pyinotify
  155. * kqueue (OS X/BSD)
  156. * stat
  157. The fallback implementation simply polls the files using ``stat`` and is very
  158. expensive.
  159. You can force an implementation by setting the :envvar:`CELERYD_FSNOTIFY`
  160. environment variable::
  161. $ env CELERYD_FSNOTIFY=stat celeryd -l info --autoreload
  162. .. _worker-remote-control:
  163. Remote control
  164. ==============
  165. .. versionadded:: 2.0
  166. :supported pools: processes, eventlet, gevent, blocking:threads/solo (see note)
  167. :supported transports: amqp, redis, mongodb
  168. Workers have the ability to be remote controlled using a high-priority
  169. broadcast message queue. The commands can be directed to all, or a specific
  170. list of workers.
  171. Commands can also have replies. The client can then wait for and collect
  172. those replies. Since there's no central authority to know how many
  173. workers are available in the cluster, there is also no way to estimate
  174. how many workers may send a reply, so the client has a configurable
  175. timeout — the deadline in seconds for replies to arrive in. This timeout
  176. defaults to one second. If the worker doesn't reply within the deadline
  177. it doesn't necessarily mean the worker didn't reply, or worse is dead, but
  178. may simply be caused by network latency or the worker being slow at processing
  179. commands, so adjust the timeout accordingly.
  180. In addition to timeouts, the client can specify the maximum number
  181. of replies to wait for. If a destination is specified, this limit is set
  182. to the number of destination hosts.
  183. .. seealso::
  184. The :program:`celeryctl` program is used to execute remote control
  185. commands from the command line. It supports all of the commands
  186. listed below. See :ref:`monitoring-celeryctl` for more information.
  187. .. note::
  188. The solo and threads pool supports remote control commands,
  189. but any task executing will block any waiting control command,
  190. so it is of limited use if the worker is very busy. In that
  191. case you must increase the timeout waitin for replies in the client.
  192. .. _worker-broadcast-fun:
  193. The :meth:`~@control.broadcast` function.
  194. ----------------------------------------------------
  195. This is the client function used to send commands to the workers.
  196. Some remote control commands also have higher-level interfaces using
  197. :meth:`~@control.broadcast` in the background, like
  198. :meth:`~@control.rate_limit` and :meth:`~@control.ping`.
  199. Sending the :control:`rate_limit` command and keyword arguments::
  200. >>> from celery.task.control import broadcast
  201. >>> celery.control.broadcast("rate_limit",
  202. ... arguments={"task_name": "myapp.mytask",
  203. ... "rate_limit": "200/m"})
  204. This will send the command asynchronously, without waiting for a reply.
  205. To request a reply you have to use the `reply` argument::
  206. >>> celery.control.broadcast("rate_limit", {
  207. ... "task_name": "myapp.mytask", "rate_limit": "200/m"}, reply=True)
  208. [{'worker1.example.com': 'New rate limit set successfully'},
  209. {'worker2.example.com': 'New rate limit set successfully'},
  210. {'worker3.example.com': 'New rate limit set successfully'}]
  211. Using the `destination` argument you can specify a list of workers
  212. to receive the command::
  213. >>> celery.control.broadcast("rate_limit", {
  214. ... "task_name": "myapp.mytask",
  215. ... "rate_limit": "200/m"}, reply=True,
  216. ... destination=["worker1.example.com"])
  217. [{'worker1.example.com': 'New rate limit set successfully'}]
  218. Of course, using the higher-level interface to set rate limits is much
  219. more convenient, but there are commands that can only be requested
  220. using :meth:`~@control.broadcast`.
  221. .. _worker-rate-limits:
  222. .. control:: rate_limit
  223. Rate limits
  224. -----------
  225. Example changing the rate limit for the `myapp.mytask` task to accept
  226. 200 tasks a minute on all servers::
  227. >>> celery.control.rate_limit("myapp.mytask", "200/m")
  228. Example changing the rate limit on a single host by specifying the
  229. destination host name::
  230. >>> celery.control.rate_limit("myapp.mytask", "200/m",
  231. ... destination=["worker1.example.com"])
  232. .. warning::
  233. This won't affect workers with the
  234. :setting:`CELERY_DISABLE_RATE_LIMITS` setting on. To re-enable rate limits
  235. then you have to restart the worker.
  236. .. control:: revoke
  237. Revoking tasks
  238. --------------
  239. All worker nodes keeps a memory of revoked task ids, either in-memory or
  240. persistent on disk (see :ref:`worker-persistent-revokes`).
  241. When a worker receives a revoke request it will skip executing
  242. the task, but it won't terminate an already executing task unless
  243. the `terminate` option is set.
  244. If `terminate` is set the worker child process processing the task
  245. will be terminated. The default signal sent is `TERM`, but you can
  246. specify this using the `signal` argument. Signal can be the uppercase name
  247. of any signal defined in the :mod:`signal` module in the Python Standard
  248. Library.
  249. Terminating a task also revokes it.
  250. **Example**
  251. ::
  252. >>> celery.control.revoke("d9078da5-9915-40a0-bfa1-392c7bde42ed")
  253. >>> celery.control.revoke("d9078da5-9915-40a0-bfa1-392c7bde42ed",
  254. ... terminate=True)
  255. >>> celery.control.revoke("d9078da5-9915-40a0-bfa1-392c7bde42ed",
  256. ... terminate=True, signal="SIGKILL")
  257. .. control:: shutdown
  258. Remote shutdown
  259. ---------------
  260. This command will gracefully shut down the worker remotely::
  261. >>> celery.control.broadcast("shutdown") # shutdown all workers
  262. >>> celery.control.broadcast("shutdown, destination="worker1.example.com")
  263. .. control:: ping
  264. Ping
  265. ----
  266. This command requests a ping from alive workers.
  267. The workers reply with the string 'pong', and that's just about it.
  268. It will use the default one second timeout for replies unless you specify
  269. a custom timeout::
  270. >>> celery.control.ping(timeout=0.5)
  271. [{'worker1.example.com': 'pong'},
  272. {'worker2.example.com': 'pong'},
  273. {'worker3.example.com': 'pong'}]
  274. :meth:`~@control.ping` also supports the `destination` argument,
  275. so you can specify which workers to ping::
  276. >>> ping(['worker2.example.com', 'worker3.example.com'])
  277. [{'worker2.example.com': 'pong'},
  278. {'worker3.example.com': 'pong'}]
  279. .. _worker-enable-events:
  280. .. control:: enable_events
  281. .. control:: disable_events
  282. Enable/disable events
  283. ---------------------
  284. You can enable/disable events by using the `enable_events`,
  285. `disable_events` commands. This is useful to temporarily monitor
  286. a worker using :program:`celeryev`/:program:`celerymon`.
  287. .. code-block:: python
  288. >>> celery.control.broadcast("enable_events")
  289. >>> celery.control.broadcast("disable_events")
  290. Adding/Reloading modules
  291. ------------------------
  292. .. versionadded:: 2.5
  293. The remote control command ``pool_restart`` sends restart requests to
  294. the workers child processes. It is particularly useful for forcing
  295. the worker to import new modules, or for reloading already imported
  296. modules. This command does not interrupt executing tasks.
  297. Example
  298. ~~~~~~~
  299. Running the following command will result in the `foo` and `bar` modules
  300. being imported by the worker processes:
  301. .. code-block:: python
  302. >>> from celery.task.control import broadcast
  303. >>> celery.control.broadcast("pool_restart",
  304. ... arguments={"modules": ["foo", "bar"]})
  305. Use the ``reload`` argument to reload modules it has already imported:
  306. .. code-block:: python
  307. >>> celery.control.broadcast("pool_restart",
  308. ... arguments={"modules": ["foo"],
  309. ... "reload": True})
  310. If you don't specify any modules then all known tasks modules will
  311. be imported/reloaded:
  312. .. code-block:: python
  313. >>> celery.control.broadcast("pool_restart", arguments={"reload": True})
  314. The ``modules`` argument is a list of modules to modify. ``reload``
  315. specifies whether to reload modules if they have previously been imported.
  316. By default ``reload`` is disabled. The `pool_restart` command uses the
  317. Python :func:`reload` function to reload modules, or you can provide
  318. your own custom reloader by passing the ``reloader`` argument.
  319. .. note::
  320. Module reloading comes with caveats that are documented in :func:`reload`.
  321. Please read this documentation and make sure your modules are suitable
  322. for reloading.
  323. .. seealso::
  324. - http://pyunit.sourceforge.net/notes/reloading.html
  325. - http://www.indelible.org/ink/python-reloading/
  326. - http://docs.python.org/library/functions.html#reload
  327. .. _worker-custom-control-commands:
  328. Writing your own remote control commands
  329. ----------------------------------------
  330. Remote control commands are registered in the control panel and
  331. they take a single argument: the current
  332. :class:`~celery.worker.control.ControlDispatch` instance.
  333. From there you have access to the active
  334. :class:`~celery.worker.consumer.Consumer` if needed.
  335. Here's an example control command that restarts the broker connection:
  336. .. code-block:: python
  337. from celery.worker.control import Panel
  338. @Panel.register
  339. def reset_connection(panel):
  340. panel.logger.critical("Connection reset by remote control.")
  341. panel.consumer.reset_connection()
  342. return {"ok": "connection reset"}
  343. These can be added to task modules, or you can keep them in their own module
  344. then import them using the :setting:`CELERY_IMPORTS` setting::
  345. CELERY_IMPORTS = ("myapp.worker.control", )
  346. .. _worker-inspect:
  347. Inspecting workers
  348. ==================
  349. :class:`@control.inspect` lets you inspect running workers. It
  350. uses remote control commands under the hood.
  351. .. code-block:: python
  352. # Inspect all nodes.
  353. >>> i = celery.control.inspect()
  354. # Specify multiple nodes to inspect.
  355. >>> i = celery.control.inspect(["worker1.example.com",
  356. "worker2.example.com"])
  357. # Specify a single node to inspect.
  358. >>> i = celery.control.inspect("worker1.example.com")
  359. .. _worker-inspect-registered-tasks:
  360. Dump of registered tasks
  361. ------------------------
  362. You can get a list of tasks registered in the worker using the
  363. :meth:`~@control.inspect.registered`::
  364. >>> i.registered()
  365. [{'worker1.example.com': ['celery.delete_expired_task_meta',
  366. 'celery.execute_remote',
  367. 'celery.map_async',
  368. 'celery.ping',
  369. 'celery.task.http.HttpDispatchTask',
  370. 'tasks.add',
  371. 'tasks.sleeptask']}]
  372. .. _worker-inspect-active-tasks:
  373. Dump of currently executing tasks
  374. ---------------------------------
  375. You can get a list of active tasks using
  376. :meth:`~@control.inspect.active`::
  377. >>> i.active()
  378. [{'worker1.example.com':
  379. [{"name": "tasks.sleeptask",
  380. "id": "32666e9b-809c-41fa-8e93-5ae0c80afbbf",
  381. "args": "(8,)",
  382. "kwargs": "{}"}]}]
  383. .. _worker-inspect-eta-schedule:
  384. Dump of scheduled (ETA) tasks
  385. -----------------------------
  386. You can get a list of tasks waiting to be scheduled by using
  387. :meth:`~@control.inspect.scheduled`::
  388. >>> i.scheduled()
  389. [{'worker1.example.com':
  390. [{"eta": "2010-06-07 09:07:52", "priority": 0,
  391. "request": {
  392. "name": "tasks.sleeptask",
  393. "id": "1a7980ea-8b19-413e-91d2-0b74f3844c4d",
  394. "args": "[1]",
  395. "kwargs": "{}"}},
  396. {"eta": "2010-06-07 09:07:53", "priority": 0,
  397. "request": {
  398. "name": "tasks.sleeptask",
  399. "id": "49661b9a-aa22-4120-94b7-9ee8031d219d",
  400. "args": "[2]",
  401. "kwargs": "{}"}}]}]
  402. Note that these are tasks with an eta/countdown argument, not periodic tasks.
  403. .. _worker-inspect-reserved:
  404. Dump of reserved tasks
  405. ----------------------
  406. Reserved tasks are tasks that has been received, but is still waiting to be
  407. executed.
  408. You can get a list of these using
  409. :meth:`~@control.inspect.reserved`::
  410. >>> i.reserved()
  411. [{'worker1.example.com':
  412. [{"name": "tasks.sleeptask",
  413. "id": "32666e9b-809c-41fa-8e93-5ae0c80afbbf",
  414. "args": "(8,)",
  415. "kwargs": "{}"}]}]