tasks.rst 32 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095
  1. .. _guide-tasks:
  2. =======
  3. Tasks
  4. =======
  5. .. contents::
  6. :local:
  7. This guide gives an overview of how tasks are defined. For a complete
  8. listing of task attributes and methods, please see the
  9. :class:`API reference <celery.task.base.BaseTask>`.
  10. .. _task-basics:
  11. Basics
  12. ======
  13. A task is a class that encapsulates a function and its execution options.
  14. Given a function create_user`, that takes two arguments: `username` and
  15. `password`, you can create a task like this:
  16. .. code-block:: python
  17. from django.contrib.auth import User
  18. @celery.task()
  19. def create_user(username, password):
  20. User.objects.create(username=username, password=password)
  21. Task options are added as arguments to `task`:
  22. .. code-block:: python
  23. @celery.task(serializer="json")
  24. def create_user(username, password):
  25. User.objects.create(username=username, password=password)
  26. .. _task-request-info:
  27. Context
  28. =======
  29. `task.request` contains information and state related
  30. the currently executing task, and must always contain the following
  31. attributes:
  32. :id: The unique id of the executing task.
  33. :taskset: The unique id of the taskset this task is a member of (if any).
  34. :args: Positional arguments.
  35. :kwargs: Keyword arguments.
  36. :retries: How many times the current task has been retried.
  37. An integer starting at `0`.
  38. :is_eager: Set to :const:`True` if the task is executed locally in
  39. the client, and not by a worker.
  40. :logfile: The file the worker logs to. See `Logging`_.
  41. :loglevel: The current log level used.
  42. :delivery_info: Additional message delivery information. This is a mapping
  43. containing the exchange and routing key used to deliver this
  44. task. Used by e.g. :meth:`~celery.task.base.BaseTask.retry`
  45. to resend the task to the same destination queue.
  46. **NOTE** As some messaging backends doesn't have advanced routing
  47. capabilities, you can't trust the availability of keys in this mapping.
  48. Example Usage
  49. -------------
  50. ::
  51. @celery.task
  52. def add(x, y):
  53. print("Executing task id %r, args: %r kwargs: %r" % (
  54. add.request.id, add.request.args, add.request.kwargs))
  55. .. _task-logging:
  56. Logging
  57. =======
  58. You can use the workers logger to add diagnostic output to
  59. the worker log:
  60. .. code-block:: python
  61. @celery.task()
  62. def add(x, y):
  63. logger = add.get_logger()
  64. logger.info("Adding %s + %s" % (x, y))
  65. return x + y
  66. There are several logging levels available, and the workers `loglevel`
  67. setting decides whether or not they will be written to the log file.
  68. Of course, you can also simply use `print` as anything written to standard
  69. out/-err will be written to the log file as well.
  70. .. _task-retry:
  71. Retrying a task if something fails
  72. ==================================
  73. Simply use :meth:`~celery.task.base.BaseTask.retry` to re-send the task.
  74. It will do the right thing, and respect the
  75. :attr:`~celery.task.base.BaseTask.max_retries` attribute:
  76. .. code-block:: python
  77. @celery.task()
  78. def send_twitter_status(oauth, tweet):
  79. try:
  80. twitter = Twitter(oauth)
  81. twitter.update_status(tweet)
  82. except (Twitter.FailWhaleError, Twitter.LoginError), exc:
  83. send_twitter_status.retry(exc=exc)
  84. Here we used the `exc` argument to pass the current exception to
  85. :meth:`~celery.task.base.BaseTask.retry`. At each step of the retry this exception
  86. is available as the tombstone (result) of the task. When
  87. :attr:`~celery.task.base.BaseTask.max_retries` has been exceeded this is the
  88. exception raised. However, if an `exc` argument is not provided the
  89. :exc:`~celery.exceptions.RetryTaskError` exception is raised instead.
  90. .. _task-retry-custom-delay:
  91. Using a custom retry delay
  92. --------------------------
  93. When a task is to be retried, it will wait for a given amount of time
  94. before doing so. The default delay is in the
  95. :attr:`~celery.task.base.BaseTask.default_retry_delay`
  96. attribute on the task. By default this is set to 3 minutes. Note that the
  97. unit for setting the delay is in seconds (int or float).
  98. You can also provide the `countdown` argument to
  99. :meth:`~celery.task.base.BaseTask.retry` to override this default.
  100. .. code-block:: python
  101. @celery.task(default_retry_delay=30 * 60) # retry in 30 minutes.
  102. def add(x, y):
  103. try:
  104. ...
  105. except Exception, exc:
  106. self.retry(exc=exc, countdown=60) # override the default and
  107. # retry in 1 minute
  108. .. _task-options:
  109. Task options
  110. ============
  111. General
  112. -------
  113. .. _task-general-options:
  114. .. attribute:: Task.name
  115. The name the task is registered as.
  116. You can set this name manually, or just use the default which is
  117. automatically generated using the module and class name. See
  118. :ref:`task-names`.
  119. .. attribute Task.request
  120. If the task is being executed this will contain information
  121. about the current request. Thread local storage is used.
  122. See :ref:`task-request-info`.
  123. .. attribute:: Task.abstract
  124. Abstract classes are not registered, but are used as the
  125. base class for new task types.
  126. .. attribute:: Task.max_retries
  127. The maximum number of attempted retries before giving up.
  128. If this exceeds the :exc:`~celery.exceptions.MaxRetriesExceeded`
  129. an exception will be raised. *NOTE:* You have to :meth:`retry`
  130. manually, it's not something that happens automatically.
  131. .. attribute:: Task.default_retry_delay
  132. Default time in seconds before a retry of the task
  133. should be executed. Can be either :class:`int` or :class:`float`.
  134. Default is a 3 minute delay.
  135. .. attribute:: Task.rate_limit
  136. Set the rate limit for this task type, i.e. how many times in
  137. a given period of time is the task allowed to run.
  138. If this is :const:`None` no rate limit is in effect.
  139. If it is an integer, it is interpreted as "tasks per second".
  140. The rate limits can be specified in seconds, minutes or hours
  141. by appending `"/s"`, `"/m"` or `"/h"` to the value.
  142. Example: `"100/m"` (hundred tasks a minute). Default is the
  143. :setting:`CELERY_DEFAULT_RATE_LIMIT` setting, which if not specified means
  144. rate limiting for tasks is disabled by default.
  145. .. attribute:: Task.ignore_result
  146. Don't store task state. Note that this means you can't use
  147. :class:`~celery.result.AsyncResult` to check if the task is ready,
  148. or get its return value.
  149. .. attribute:: Task.store_errors_even_if_ignored
  150. If :const:`True`, errors will be stored even if the task is configured
  151. to ignore results.
  152. .. attribute:: Task.send_error_emails
  153. Send an email whenever a task of this type fails.
  154. Defaults to the :setting:`CELERY_SEND_TASK_ERROR_EMAILS` setting.
  155. See :ref:`conf-error-mails` for more information.
  156. .. attribute:: Task.error_whitelist
  157. If the sending of error emails is enabled for this task, then
  158. this is a white list of exceptions to actually send emails about.
  159. .. attribute:: Task.serializer
  160. A string identifying the default serialization
  161. method to use. Defaults to the :setting:`CELERY_TASK_SERIALIZER`
  162. setting. Can be `pickle` `json`, `yaml`, or any custom
  163. serialization methods that have been registered with
  164. :mod:`kombu.serialization.registry`.
  165. Please see :ref:`executing-serializers` for more information.
  166. .. attribute:: Task.backend
  167. The result store backend to use for this task. Defaults to the
  168. :setting:`CELERY_RESULT_BACKEND` setting.
  169. .. attribute:: Task.acks_late
  170. If set to :const:`True` messages for this task will be acknowledged
  171. **after** the task has been executed, not *just before*, which is
  172. the default behavior.
  173. Note that this means the task may be executed twice if the worker
  174. crashes in the middle of execution, which may be acceptable for some
  175. applications.
  176. The global default can be overridden by the :setting:`CELERY_ACKS_LATE`
  177. setting.
  178. .. attribute:: Task.track_started
  179. If :const:`True` the task will report its status as "started"
  180. when the task is executed by a worker.
  181. The default value is :const:`False` as the normal behaviour is to not
  182. report that level of granularity. Tasks are either pending, finished,
  183. or waiting to be retried. Having a "started" status can be useful for
  184. when there are long running tasks and there is a need to report which
  185. task is currently running.
  186. The host name and process id of the worker executing the task
  187. will be available in the state metadata (e.g. `result.info["pid"]`)
  188. The global default can be overridden by the
  189. :setting:`CELERY_TRACK_STARTED` setting.
  190. .. seealso::
  191. The API reference for :class:`~celery.task.base.BaseTask`.
  192. .. _task-message-options:
  193. Message and routing options
  194. ---------------------------
  195. .. attribute:: Task.queue
  196. Use the routing settings from a queue defined in :setting:`CELERY_QUEUES`.
  197. If defined the :attr:`exchange` and :attr:`routing_key` options will be
  198. ignored.
  199. .. attribute:: Task.exchange
  200. Override the global default `exchange` for this task.
  201. .. attribute:: Task.routing_key
  202. Override the global default `routing_key` for this task.
  203. .. attribute:: Task.mandatory
  204. If set, the task message has mandatory routing. By default the task
  205. is silently dropped by the broker if it can't be routed to a queue.
  206. However -- If the task is mandatory, an exception will be raised
  207. instead.
  208. Not supported by amqplib.
  209. .. attribute:: Task.immediate
  210. Request immediate delivery. If the task cannot be routed to a
  211. task worker immediately, an exception will be raised. This is
  212. instead of the default behavior, where the broker will accept and
  213. queue the task, but with no guarantee that the task will ever
  214. be executed.
  215. Not supported by amqplib.
  216. .. attribute:: Task.priority
  217. The message priority. A number from 0 to 9, where 0 is the
  218. highest priority.
  219. Not supported by RabbitMQ.
  220. .. seealso::
  221. :ref:`executing-routing` for more information about message options,
  222. and :ref:`guide-routing`.
  223. .. _task-names:
  224. Task names
  225. ==========
  226. The task type is identified by the *task name*.
  227. If not provided a name will be automatically generated using the module
  228. and class name.
  229. For example:
  230. .. code-block:: python
  231. >>> @celery.task(name="sum-of-two-numbers")
  232. >>> def add(x, y):
  233. ... return x + y
  234. >>> add.name
  235. 'sum-of-two-numbers'
  236. The best practice is to use the module name as a prefix to classify the
  237. tasks using namespaces. This way the name won't collide with the name from
  238. another module:
  239. .. code-block:: python
  240. >>> @celery.task(name="tasks.add")
  241. >>> def add(x, y):
  242. ... return x + y
  243. >>> add.name
  244. 'tasks.add'
  245. Which is exactly the name that is automatically generated for this
  246. task if the module name is "tasks.py":
  247. .. code-block:: python
  248. >>> @celery.task()
  249. >>> def add(x, y):
  250. ... return x + y
  251. >>> add.name
  252. 'tasks.add'
  253. .. _task-naming-relative-imports:
  254. Automatic naming and relative imports
  255. -------------------------------------
  256. Relative imports and automatic name generation does not go well together,
  257. so if you're using relative imports you should set the name explicitly.
  258. For example if the client imports the module "myapp.tasks" as ".tasks", and
  259. the worker imports the module as "myapp.tasks", the generated names won't match
  260. and an :exc:`~celery.exceptions.NotRegistered` error will be raised by the worker.
  261. This is also the case if using Django and using `project.myapp`::
  262. INSTALLED_APPS = ("project.myapp", )
  263. The worker will have the tasks registered as "project.myapp.tasks.*",
  264. while this is what happens in the client if the module is imported as
  265. "myapp.tasks":
  266. .. code-block:: python
  267. >>> from myapp.tasks import add
  268. >>> add.name
  269. 'myapp.tasks.add'
  270. For this reason you should never use "project.app", but rather
  271. add the project directory to the Python path::
  272. import os
  273. import sys
  274. sys.path.append(os.getcwd())
  275. INSTALLED_APPS = ("myapp", )
  276. This makes more sense from the reusable app perspective anyway.
  277. .. _tasks-decorating:
  278. Decorating tasks
  279. ================
  280. When using other decorators you must make sure that the `task`
  281. decorator is applied last:
  282. .. code-block:: python
  283. @task
  284. @decorator2
  285. @decorator1
  286. def add(x, y):
  287. return x + y
  288. Which means the `@task` decorator must be the top statement.
  289. .. _task-states:
  290. Task States
  291. ===========
  292. Celery can keep track of the tasks current state. The state also contains the
  293. result of a successful task, or the exception and traceback information of a
  294. failed task.
  295. There are several *result backends* to choose from, and they all have
  296. different strenghts and weaknesses (see :ref:`task-result-backends`).
  297. During its lifetime a task will transition through several possible states,
  298. and each state may have arbitrary metadata attached to it. When a task
  299. moves into a new state the previous state is
  300. forgotten about, but some transitions can be deducted, (e.g. a task now
  301. in the :state:`FAILED` state, is implied to have been in the
  302. :state:`STARTED` state at some point).
  303. There are also sets of states, like the set of
  304. :state:`failure states <FAILURE_STATES>`, and the set of
  305. :state:`ready states <READY_STATES>`.
  306. The client uses the membership of these sets to decide whether
  307. the exception should be re-raised (:state:`PROPAGATE_STATES`), or whether
  308. the result can be cached (it can if the task is ready).
  309. You can also define :ref:`custom-states`.
  310. .. _task-result-backends:
  311. Result Backends
  312. ---------------
  313. Celery needs to store or send the states somewhere. There are several
  314. built-in backends to choose from: SQLAlchemy/Django ORM, Memcached, Redis,
  315. AMQP, MongoDB, Tokyo Tyrant and Redis -- or you can define your own.
  316. No backend works well for every use case.
  317. You should read about the strenghts and weaknesses of each backend, and choose
  318. the most appropriate for your needs.
  319. .. seealso::
  320. :ref:`conf-result-backend`
  321. AMQP Result Backend
  322. ~~~~~~~~~~~~~~~~~~~
  323. The AMQP result backend is special as it does not actually *store* the states,
  324. but rather sends them as messages. This is an important difference as it
  325. means that a result *can only be retrieved once*; If you have two processes
  326. waiting for the same result, one of the processes will never receive the
  327. result!
  328. Even with that limitation, it is an excellent choice if you need to receive
  329. state changes in real-time. Using messaging means the client does not have to
  330. poll for new states.
  331. There are several other pitfalls you should be aware of when using the AMQP
  332. backend:
  333. * Every new task creates a new queue on the server, with thousands of tasks
  334. the broker may be overloaded with queues and this will affect performance in
  335. negative ways. If you're using RabbitMQ then each queue will be a separate
  336. Erlang process, so if you're planning to keep many results simultaneously you
  337. may have to increase the Erlang process limit, and the maximum number of file
  338. descriptors your OS allows.
  339. * Old results will not be cleaned automatically, so you must make sure to
  340. consume the results or else the number of queues will eventually go out of
  341. control. If you're running RabbitMQ 2.1.1 or higher you can take advantage
  342. of the ``x-expires`` argument to queues, which will expire queues after a
  343. certain time limit after they are unused. The queue expiry can be set (in
  344. seconds) by the :setting:`CELERY_AMQP_TASK_RESULT_EXPIRES` setting (not
  345. enabled by default).
  346. For a list of options supported by the AMQP result backend, please see
  347. :ref:`conf-amqp-result-backend`.
  348. Database Result Backend
  349. ~~~~~~~~~~~~~~~~~~~~~~~
  350. Keeping state in the database can be convenient for many, especially for
  351. web applications with a database already in place, but it also comes with
  352. limitations.
  353. * Polling the database for new states is expensive, and so you should
  354. increase the polling intervals of operations such as `result.wait()`, and
  355. `tasksetresult.join()`
  356. * Some databases uses a default transaction isolation level that
  357. is not suitable for polling tables for changes.
  358. In MySQL the default transaction isolation level is `REPEATABLE-READ`, which
  359. means the transaction will not see changes by other transactions until the
  360. transaction is commited. It is recommended that you change to the
  361. `READ-COMMITTED` isolation level.
  362. .. _task-builtin-states:
  363. Built-in States
  364. ---------------
  365. .. state:: PENDING
  366. PENDING
  367. ~~~~~~~
  368. Task is waiting for execution or unknown.
  369. Any task id that is not know is implied to be in the pending state.
  370. .. state:: STARTED
  371. STARTED
  372. ~~~~~~~
  373. Task has been started.
  374. Not reported by default, to enable please see :attr`Task.track_started`.
  375. :metadata: `pid` and `hostname` of the worker process executing
  376. the task.
  377. .. state:: SUCCESS
  378. SUCCESS
  379. ~~~~~~~
  380. Task has been successfully executed.
  381. :metadata: `result` contains the return value of the task.
  382. :propagates: Yes
  383. :ready: Yes
  384. .. state:: FAILURE
  385. FAILURE
  386. ~~~~~~~
  387. Task execution resulted in failure.
  388. :metadata: `result` contains the exception occurred, and `traceback`
  389. contains the backtrace of the stack at the point when the
  390. exception was raised.
  391. :propagates: Yes
  392. .. state:: RETRY
  393. RETRY
  394. ~~~~~
  395. Task is being retried.
  396. :metadata: `result` contains the exception that caused the retry,
  397. and `traceback` contains the backtrace of the stack at the point
  398. when the exceptions was raised.
  399. :propagates: No
  400. .. state:: REVOKED
  401. REVOKED
  402. ~~~~~~~
  403. Task has been revoked.
  404. :propagates: Yes
  405. .. _custom-states:
  406. Custom states
  407. -------------
  408. You can easily define your own states, all you need is a unique name.
  409. The name of the state is usually an uppercase string. As an example
  410. you could have a look at :mod:`abortable tasks <~celery.contrib.abortable>`
  411. which defines its own custom :state:`ABORTED` state.
  412. Use :meth:`Task.update_state <celery.task.base.BaseTask.update_state>` to
  413. update a tasks state::
  414. @celery.task
  415. def upload_files(filenames):
  416. for i, file in enumerate(filenames):
  417. upload_files.update_state(state="PROGRESS",
  418. meta={"current": i, "total": len(filenames)})
  419. Here we created the state `"PROGRESS"`, which tells any application
  420. aware of this state that the task is currently in progress, and also where
  421. it is in the process by having `current` and `total` counts as part of the
  422. state metadata. This can then be used to create e.g. progress bars.
  423. .. _task-how-they-work:
  424. How it works
  425. ============
  426. Here comes the technical details, this part isn't something you need to know,
  427. but you may be interested.
  428. All defined tasks are listed in a registry. The registry contains
  429. a list of task names and their task classes. You can investigate this registry
  430. yourself:
  431. .. code-block:: python
  432. >>> from celery import registry
  433. >>> from celery import task
  434. >>> registry.tasks
  435. {'celery.delete_expired_task_meta':
  436. <PeriodicTask: celery.delete_expired_task_meta (periodic)>,
  437. 'celery.task.http.HttpDispatchTask':
  438. <Task: celery.task.http.HttpDispatchTask (regular)>,
  439. 'celery.execute_remote':
  440. <Task: celery.execute_remote (regular)>,
  441. 'celery.map_async':
  442. <Task: celery.map_async (regular)>,
  443. 'celery.ping':
  444. <Task: celery.ping (regular)>}
  445. This is the list of tasks built-in to celery. Note that we had to import
  446. `celery.task` first for these to show up. This is because the tasks will
  447. only be registered when the module they are defined in is imported.
  448. The default loader imports any modules listed in the
  449. :setting:`CELERY_IMPORTS` setting.
  450. The entity responsible for registering your task in the registry is a
  451. meta class, :class:`~celery.task.base.TaskType`. This is the default
  452. meta class for :class:`~celery.task.base.BaseTask`.
  453. If you want to register your task manually you can mark the
  454. task as :attr:`~celery.task.base.BaseTask.abstract`:
  455. .. code-block:: python
  456. class MyTask(Task):
  457. abstract = True
  458. This way the task won't be registered, but any task inheriting from
  459. it will be.
  460. When tasks are sent, we don't send any actual function code, just the name
  461. of the task to execute. When the worker then receives the message it can look
  462. up the name in its task registry to find the execution code.
  463. This means that your workers should always be updated with the same software
  464. as the client. This is a drawback, but the alternative is a technical
  465. challenge that has yet to be solved.
  466. .. _task-best-practices:
  467. Tips and Best Practices
  468. =======================
  469. .. _task-ignore_results:
  470. Ignore results you don't want
  471. -----------------------------
  472. If you don't care about the results of a task, be sure to set the
  473. :attr:`~celery.task.base.BaseTask.ignore_result` option, as storing results
  474. wastes time and resources.
  475. .. code-block:: python
  476. @celery.task(ignore_result=True)
  477. def mytask(...)
  478. something()
  479. Results can even be disabled globally using the :setting:`CELERY_IGNORE_RESULT`
  480. setting.
  481. .. _task-disable-rate-limits:
  482. Disable rate limits if they're not used
  483. ---------------------------------------
  484. Disabling rate limits altogether is recommended if you don't have
  485. any tasks using them. This is because the rate limit subsystem introduces
  486. quite a lot of complexity.
  487. Set the :setting:`CELERY_DISABLE_RATE_LIMITS` setting to globally disable
  488. rate limits:
  489. .. code-block:: python
  490. CELERY_DISABLE_RATE_LIMITS = True
  491. .. _task-synchronous-subtasks:
  492. Avoid launching synchronous subtasks
  493. ------------------------------------
  494. Having a task wait for the result of another task is really inefficient,
  495. and may even cause a deadlock if the worker pool is exhausted.
  496. Make your design asynchronous instead, for example by using *callbacks*.
  497. **Bad**:
  498. .. code-block:: python
  499. @celery.task()
  500. def update_page_info(url):
  501. page = fetch_page.delay(url).get()
  502. info = parse_page.delay(url, page).get()
  503. store_page_info.delay(url, info)
  504. @celery.task()
  505. def fetch_page(url):
  506. return myhttplib.get(url)
  507. @celery.task()
  508. def parse_page(url, page):
  509. return myparser.parse_document(page)
  510. @celery.task()
  511. def store_page_info(url, info):
  512. return PageInfo.objects.create(url, info)
  513. **Good**:
  514. .. code-block:: python
  515. @celery.task(ignore_result=True)
  516. def update_page_info(url):
  517. # fetch_page -> parse_page -> store_page
  518. fetch_page.delay(url, callback=subtask(parse_page,
  519. callback=subtask(store_page_info)))
  520. @celery.task(ignore_result=True)
  521. def fetch_page(url, callback=None):
  522. page = myhttplib.get(url)
  523. if callback:
  524. # The callback may have been serialized with JSON,
  525. # so best practice is to convert the subtask dict back
  526. # into a subtask object.
  527. subtask(callback).delay(url, page)
  528. @celery.task(ignore_result=True)
  529. def parse_page(url, page, callback=None):
  530. info = myparser.parse_document(page)
  531. if callback:
  532. subtask(callback).delay(url, info)
  533. @celery.task(ignore_result=True)
  534. def store_page_info(url, info):
  535. PageInfo.objects.create(url, info)
  536. We use :class:`~celery.task.sets.subtask` here to safely pass
  537. around the callback task. :class:`~celery.task.sets.subtask` is a
  538. subclass of dict used to wrap the arguments and execution options
  539. for a single task invocation.
  540. .. seealso::
  541. :ref:`sets-subtasks` for more information about subtasks.
  542. .. _task-performance-and-strategies:
  543. Performance and Strategies
  544. ==========================
  545. .. _task-granularity:
  546. Granularity
  547. -----------
  548. The task granularity is the amount of computation needed by each subtask.
  549. In general it is better to split the problem up into many small tasks, than
  550. have a few long running tasks.
  551. With smaller tasks you can process more tasks in parallel and the tasks
  552. won't run long enough to block the worker from processing other waiting tasks.
  553. However, executing a task does have overhead. A message needs to be sent, data
  554. may not be local, etc. So if the tasks are too fine-grained the additional
  555. overhead may not be worth it in the end.
  556. .. seealso::
  557. The book `Art of Concurrency`_ has a whole section dedicated to the topic
  558. of task granularity.
  559. .. _`Art of Concurrency`: http://oreilly.com/catalog/9780596521547
  560. .. _task-data-locality:
  561. Data locality
  562. -------------
  563. The worker processing the task should be as close to the data as
  564. possible. The best would be to have a copy in memory, the worst would be a
  565. full transfer from another continent.
  566. If the data is far away, you could try to run another worker at location, or
  567. if that's not possible - cache often used data, or preload data you know
  568. is going to be used.
  569. The easiest way to share data between workers is to use a distributed cache
  570. system, like `memcached`_.
  571. .. seealso::
  572. The paper `Distributed Computing Economics`_ by Jim Gray is an excellent
  573. introduction to the topic of data locality.
  574. .. _`Distributed Computing Economics`:
  575. http://research.microsoft.com/pubs/70001/tr-2003-24.pdf
  576. .. _`memcached`: http://memcached.org/
  577. .. _task-state:
  578. State
  579. -----
  580. Since celery is a distributed system, you can't know in which process, or
  581. on what machine the task will be executed. You can't even know if the task will
  582. run in a timely manner.
  583. The ancient async sayings tells us that “asserting the world is the
  584. responsibility of the task”. What this means is that the world view may
  585. have changed since the task was requested, so the task is responsible for
  586. making sure the world is how it should be; If you have a task
  587. that re-indexes a search engine, and the search engine should only be
  588. re-indexed at maximum every 5 minutes, then it must be the tasks
  589. responsibility to assert that, not the callers.
  590. Another gotcha is Django model objects. They shouldn't be passed on as
  591. arguments to tasks. It's almost always better to re-fetch the object from
  592. the database when the task is running instead, as using old data may lead
  593. to race conditions.
  594. Imagine the following scenario where you have an article and a task
  595. that automatically expands some abbreviations in it:
  596. .. code-block:: python
  597. class Article(models.Model):
  598. title = models.CharField()
  599. body = models.TextField()
  600. @celery.task
  601. def expand_abbreviations(article):
  602. article.body.replace("MyCorp", "My Corporation")
  603. article.save()
  604. First, an author creates an article and saves it, then the author
  605. clicks on a button that initiates the abbreviation task.
  606. >>> article = Article.objects.get(id=102)
  607. >>> expand_abbreviations.delay(model_object)
  608. Now, the queue is very busy, so the task won't be run for another 2 minutes.
  609. In the meantime another author makes changes to the article, so
  610. when the task is finally run, the body of the article is reverted to the old
  611. version because the task had the old body in its argument.
  612. Fixing the race condition is easy, just use the article id instead, and
  613. re-fetch the article in the task body:
  614. .. code-block:: python
  615. @celery.task
  616. def expand_abbreviations(article_id):
  617. article = Article.objects.get(id=article_id)
  618. article.body.replace("MyCorp", "My Corporation")
  619. article.save()
  620. >>> expand_abbreviations(article_id)
  621. There might even be performance benefits to this approach, as sending large
  622. messages may be expensive.
  623. .. _task-database-transactions:
  624. Database transactions
  625. ---------------------
  626. Let's have a look at another example:
  627. .. code-block:: python
  628. from django.db import transaction
  629. @transaction.commit_on_success
  630. def create_article(request):
  631. article = Article.objects.create(....)
  632. expand_abbreviations.delay(article.pk)
  633. This is a Django view creating an article object in the database,
  634. then passing the primary key to a task. It uses the `commit_on_success`
  635. decorator, which will commit the transaction when the view returns, or
  636. roll back if the view raises an exception.
  637. There is a race condition if the task starts executing
  638. before the transaction has been committed; The database object does not exist
  639. yet!
  640. The solution is to *always commit transactions before sending tasks
  641. depending on state from the current transaction*:
  642. .. code-block:: python
  643. @transaction.commit_manually
  644. def create_article(request):
  645. try:
  646. article = Article.objects.create(...)
  647. except:
  648. transaction.rollback()
  649. raise
  650. else:
  651. transaction.commit()
  652. expand_abbreviations.delay(article.pk)
  653. .. _task-example:
  654. Example
  655. =======
  656. Let's take a real wold example; A blog where comments posted needs to be
  657. filtered for spam. When the comment is created, the spam filter runs in the
  658. background, so the user doesn't have to wait for it to finish.
  659. We have a Django blog application allowing comments
  660. on blog posts. We'll describe parts of the models/views and tasks for this
  661. application.
  662. blog/models.py
  663. --------------
  664. The comment model looks like this:
  665. .. code-block:: python
  666. from django.db import models
  667. from django.utils.translation import ugettext_lazy as _
  668. class Comment(models.Model):
  669. name = models.CharField(_("name"), max_length=64)
  670. email_address = models.EmailField(_("email address"))
  671. homepage = models.URLField(_("home page"),
  672. blank=True, verify_exists=False)
  673. comment = models.TextField(_("comment"))
  674. pub_date = models.DateTimeField(_("Published date"),
  675. editable=False, auto_add_now=True)
  676. is_spam = models.BooleanField(_("spam?"),
  677. default=False, editable=False)
  678. class Meta:
  679. verbose_name = _("comment")
  680. verbose_name_plural = _("comments")
  681. In the view where the comment is posted, we first write the comment
  682. to the database, then we launch the spam filter task in the background.
  683. .. _task-example-blog-views:
  684. blog/views.py
  685. -------------
  686. .. code-block:: python
  687. from django import forms
  688. from django.http import HttpResponseRedirect
  689. from django.template.context import RequestContext
  690. from django.shortcuts import get_object_or_404, render_to_response
  691. from blog import tasks
  692. from blog.models import Comment
  693. class CommentForm(forms.ModelForm):
  694. class Meta:
  695. model = Comment
  696. def add_comment(request, slug, template_name="comments/create.html"):
  697. post = get_object_or_404(Entry, slug=slug)
  698. remote_addr = request.META.get("REMOTE_ADDR")
  699. if request.method == "post":
  700. form = CommentForm(request.POST, request.FILES)
  701. if form.is_valid():
  702. comment = form.save()
  703. # Check spam asynchronously.
  704. tasks.spam_filter.delay(comment_id=comment.id,
  705. remote_addr=remote_addr)
  706. return HttpResponseRedirect(post.get_absolute_url())
  707. else:
  708. form = CommentForm()
  709. context = RequestContext(request, {"form": form})
  710. return render_to_response(template_name, context_instance=context)
  711. To filter spam in comments we use `Akismet`_, the service
  712. used to filter spam in comments posted to the free weblog platform
  713. `Wordpress`. `Akismet`_ is free for personal use, but for commercial use you
  714. need to pay. You have to sign up to their service to get an API key.
  715. To make API calls to `Akismet`_ we use the `akismet.py`_ library written by
  716. `Michael Foord`_.
  717. .. _task-example-blog-tasks:
  718. blog/tasks.py
  719. -------------
  720. .. code-block:: python
  721. from akismet import Akismet
  722. from celery.task import task
  723. from django.core.exceptions import ImproperlyConfigured
  724. from django.contrib.sites.models import Site
  725. from blog.models import Comment
  726. @task
  727. def spam_filter(comment_id, remote_addr=None):
  728. logger = spam_filter.get_logger()
  729. logger.info("Running spam filter for comment %s" % comment_id)
  730. comment = Comment.objects.get(pk=comment_id)
  731. current_domain = Site.objects.get_current().domain
  732. akismet = Akismet(settings.AKISMET_KEY, "http://%s" % domain)
  733. if not akismet.verify_key():
  734. raise ImproperlyConfigured("Invalid AKISMET_KEY")
  735. is_spam = akismet.comment_check(user_ip=remote_addr,
  736. comment_content=comment.comment,
  737. comment_author=comment.name,
  738. comment_author_email=comment.email_address)
  739. if is_spam:
  740. comment.is_spam = True
  741. comment.save()
  742. return is_spam
  743. .. _`Akismet`: http://akismet.com/faq/
  744. .. _`akismet.py`: http://www.voidspace.org.uk/downloads/akismet.py
  745. .. _`Michael Foord`: http://www.voidspace.org.uk/