tasks.rst 31 KB

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