canvas.rst 26 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926
  1. .. _guide-canvas:
  2. =============================
  3. Canvas: Designing Workflows
  4. =============================
  5. .. contents::
  6. :local:
  7. :depth: 2
  8. .. _canvas-subtasks:
  9. .. _canvas-signatures:
  10. Signatures
  11. ==========
  12. .. versionadded:: 2.0
  13. You just learned how to call a task using the tasks ``delay`` method
  14. in the :ref:`calling <guide-calling>` guide, and this is often all you need,
  15. but sometimes you may want to pass the signature of a task invocation to
  16. another process or as an argument to another function.
  17. A :func:`~celery.signature` wraps the arguments, keyword arguments, and execution options
  18. of a single task invocation in a way such that it can be passed to functions
  19. or even serialized and sent across the wire.
  20. - You can create a signature for the ``add`` task using its name like this::
  21. >>> from celery import signature
  22. >>> signature('tasks.add', args=(2, 2), countdown=10)
  23. tasks.add(2, 2)
  24. This task has a signature of arity 2 (two arguments): ``(2, 2)``,
  25. and sets the countdown execution option to 10.
  26. - or you can create one using the task's ``signature`` method::
  27. >>> add.signature((2, 2), countdown=10)
  28. tasks.add(2, 2)
  29. - There is also a shortcut using star arguments::
  30. >>> add.s(2, 2)
  31. tasks.add(2, 2)
  32. - Keyword arguments are also supported::
  33. >>> add.s(2, 2, debug=True)
  34. tasks.add(2, 2, debug=True)
  35. - From any signature instance you can inspect the different fields::
  36. >>> s = add.signature((2, 2), {'debug': True}, countdown=10)
  37. >>> s.args
  38. (2, 2)
  39. >>> s.kwargs
  40. {'debug': True}
  41. >>> s.options
  42. {'countdown': 10}
  43. - It supports the "Calling API" which means it supports ``delay`` and
  44. ``apply_async`` or being called directly.
  45. Calling the signature will execute the task inline in the current process::
  46. >>> add(2, 2)
  47. 4
  48. >>> add.s(2, 2)()
  49. 4
  50. ``delay`` is our beloved shortcut to ``apply_async`` taking star-arguments::
  51. >>> result = add.delay(2, 2)
  52. >>> result.get()
  53. 4
  54. ``apply_async`` takes the same arguments as the :meth:`Task.apply_async <@Task.apply_async>` method::
  55. >>> add.apply_async(args, kwargs, **options)
  56. >>> add.signature(args, kwargs, **options).apply_async()
  57. >>> add.apply_async((2, 2), countdown=1)
  58. >>> add.signature((2, 2), countdown=1).apply_async()
  59. - You can't define options with :meth:`~@Task.s`, but a chaining
  60. ``set`` call takes care of that::
  61. >>> add.s(2, 2).set(countdown=1)
  62. proj.tasks.add(2, 2)
  63. Partials
  64. --------
  65. With a signature, you can execute the task in a worker::
  66. >>> add.s(2, 2).delay()
  67. >>> add.s(2, 2).apply_async(countdown=1)
  68. Or you can call it directly in the current process::
  69. >>> add.s(2, 2)()
  70. 4
  71. Specifying additional args, kwargs or options to ``apply_async``/``delay``
  72. creates partials:
  73. - Any arguments added will be prepended to the args in the signature::
  74. >>> partial = add.s(2) # incomplete signature
  75. >>> partial.delay(4) # 2 + 4
  76. >>> partial.apply_async((4, )) # same
  77. - Any keyword arguments added will be merged with the kwargs in the signature,
  78. with the new keyword arguments taking precedence::
  79. >>> s = add.s(2, 2)
  80. >>> s.delay(debug=True) # -> add(2, 2, debug=True)
  81. >>> s.apply_async(kwargs={'debug': True}) # same
  82. - Any options added will be merged with the options in the signature,
  83. with the new options taking precedence::
  84. >>> s = add.signature((2, 2), countdown=10)
  85. >>> s.apply_async(countdown=1) # countdown is now 1
  86. You can also clone signatures to create derivates:
  87. >>> s = add.s(2)
  88. proj.tasks.add(2)
  89. >>> s.clone(args=(4, ), kwargs={'debug': True})
  90. proj.tasks.add(2, 4, debug=True)
  91. Immutability
  92. ------------
  93. .. versionadded:: 3.0
  94. Partials are meant to be used with callbacks, any tasks linked or chord
  95. callbacks will be applied with the result of the parent task.
  96. Sometimes you want to specify a callback that does not take
  97. additional arguments, and in that case you can set the signature
  98. to be immutable::
  99. >>> add.apply_async((2, 2), link=reset_buffers.signature(immutable=True))
  100. The ``.si()`` shortcut can also be used to create immutable signatures::
  101. >>> add.apply_async((2, 2), link=reset_buffers.si())
  102. Only the execution options can be set when a signature is immutable,
  103. so it's not possible to call the signature with partial args/kwargs.
  104. .. note::
  105. In this tutorial I sometimes use the prefix operator `~` to signatures.
  106. You probably shouldn't use it in your production code, but it's a handy shortcut
  107. when experimenting in the Python shell::
  108. >>> ~sig
  109. >>> # is the same as
  110. >>> sig.delay().get()
  111. .. _canvas-callbacks:
  112. Callbacks
  113. ---------
  114. .. versionadded:: 3.0
  115. Callbacks can be added to any task using the ``link`` argument
  116. to ``apply_async``::
  117. add.apply_async((2, 2), link=other_task.s())
  118. The callback will only be applied if the task exited successfully,
  119. and it will be applied with the return value of the parent task as argument.
  120. As I mentioned earlier, any arguments you add to a signature,
  121. will be prepended to the arguments specified by the signature itself!
  122. If you have the signature::
  123. >>> sig = add.s(10)
  124. then `sig.delay(result)` becomes::
  125. >>> add.apply_async(args=(result, 10))
  126. ...
  127. Now let's call our ``add`` task with a callback using partial
  128. arguments::
  129. >>> add.apply_async((2, 2), link=add.s(8))
  130. As expected this will first launch one task calculating :math:`2 + 2`, then
  131. another task calculating :math:`4 + 8`.
  132. The Primitives
  133. ==============
  134. .. versionadded:: 3.0
  135. .. topic:: Overview
  136. - ``group``
  137. The group primitive is a signature that takes a list of tasks that should
  138. be applied in parallel.
  139. - ``chain``
  140. The chain primitive lets us link together signatures so that one is called
  141. after the other, essentially forming a *chain* of callbacks.
  142. - ``chord``
  143. A chord is just like a group but with a callback. A chord consists
  144. of a header group and a body, where the body is a task that should execute
  145. after all of the tasks in the header are complete.
  146. - ``map``
  147. The map primitive works like the built-in ``map`` function, but creates
  148. a temporary task where a list of arguments is applied to the task.
  149. E.g. ``task.map([1, 2])`` results in a single task
  150. being called, applying the arguments in order to the task function so
  151. that the result is::
  152. res = [task(1), task(2)]
  153. - ``starmap``
  154. Works exactly like map except the arguments are applied as ``*args``.
  155. For example ``add.starmap([(2, 2), (4, 4)])`` results in a single
  156. task calling::
  157. res = [add(2, 2), add(4, 4)]
  158. - ``chunks``
  159. Chunking splits a long list of arguments into parts, e.g the operation::
  160. >>> items = zip(xrange(1000), xrange(1000)) # 1000 items
  161. >>> add.chunks(items, 10)
  162. will split the list of items into chunks of 10, resulting in 100
  163. tasks (each processing 10 items in sequence).
  164. The primitives are also signature objects themselves, so that they can be combined
  165. in any number of ways to compose complex workflows.
  166. Here's some examples:
  167. - Simple chain
  168. Here's a simple chain, the first task executes passing its return value
  169. to the next task in the chain, and so on.
  170. .. code-block:: python
  171. >>> from celery import chain
  172. # 2 + 2 + 4 + 8
  173. >>> res = chain(add.s(2, 2), add.s(4), add.s(8))()
  174. >>> res.get()
  175. 16
  176. This can also be written using pipes::
  177. >>> (add.s(2, 2) | add.s(4) | add.s(8))().get()
  178. 16
  179. - Immutable signatures
  180. Signatures can be partial so arguments can be
  181. added to the existing arguments, but you may not always want that,
  182. for example if you don't want the result of the previous task in a chain.
  183. In that case you can mark the signature as immutable, so that the arguments
  184. cannot be changed::
  185. >>> add.signature((2, 2), immutable=True)
  186. There's also an ``.si`` shortcut for this::
  187. >>> add.si(2, 2)
  188. Now you can create a chain of independent tasks instead::
  189. >>> res = (add.si(2, 2) | add.si(4, 4) | add.s(8, 8))()
  190. >>> res.get()
  191. 16
  192. >>> res.parent.get()
  193. 8
  194. >>> res.parent.parent.get()
  195. 4
  196. - Simple group
  197. You can easily create a group of tasks to execute in parallel::
  198. >>> from celery import group
  199. >>> res = group(add.s(i, i) for i in xrange(10))()
  200. >>> res.get(timeout=1)
  201. [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]
  202. - Simple chord
  203. The chord primitive enables us to add callback to be called when
  204. all of the tasks in a group have finished executing, which is often
  205. required for algorithms that aren't embarrassingly parallel::
  206. >>> from celery import chord
  207. >>> res = chord((add.s(i, i) for i in xrange(10)), xsum.s())()
  208. >>> res.get()
  209. 90
  210. The above example creates 10 task that all start in parallel,
  211. and when all of them are complete the return values are combined
  212. into a list and sent to the ``xsum`` task.
  213. The body of a chord can also be immutable, so that the return value
  214. of the group is not passed on to the callback::
  215. >>> chord((import_contact.s(c) for c in contacts),
  216. ... notify_complete.si(import_id)).apply_async()
  217. Note the use of ``.si`` above which creates an immutable signature.
  218. - Blow your mind by combining
  219. Chains can be partial too::
  220. >>> c1 = (add.s(4) | mul.s(8))
  221. # (16 + 4) * 8
  222. >>> res = c1(16)
  223. >>> res.get()
  224. 160
  225. Which means that you can combine chains::
  226. # ((4 + 16) * 2 + 4) * 8
  227. >>> c2 = (add.s(4, 16) | mul.s(2) | (add.s(4) | mul.s(8)))
  228. >>> res = c2()
  229. >>> res.get()
  230. 352
  231. Chaining a group together with another task will automatically
  232. upgrade it to be a chord::
  233. >>> c3 = (group(add.s(i, i) for i in xrange(10)) | xsum.s())
  234. >>> res = c3()
  235. >>> res.get()
  236. 90
  237. Groups and chords accepts partial arguments too, so in a chain
  238. the return value of the previous task is forwarded to all tasks in the group::
  239. >>> new_user_workflow = (create_user.s() | group(
  240. ... import_contacts.s(),
  241. ... send_welcome_email.s()))
  242. ... new_user_workflow.delay(username='artv',
  243. ... first='Art',
  244. ... last='Vandelay',
  245. ... email='art@vandelay.com')
  246. If you don't want to forward arguments to the group then
  247. you can make the signatures in the group immutable::
  248. >>> res = (add.s(4, 4) | group(add.si(i, i) for i in xrange(10)))()
  249. >>> res.get()
  250. <GroupResult: de44df8c-821d-4c84-9a6a-44769c738f98 [
  251. bc01831b-9486-4e51-b046-480d7c9b78de,
  252. 2650a1b8-32bf-4771-a645-b0a35dcc791b,
  253. dcbee2a5-e92d-4b03-b6eb-7aec60fd30cf,
  254. 59f92e0a-23ea-41ce-9fad-8645a0e7759c,
  255. 26e1e707-eccf-4bf4-bbd8-1e1729c3cce3,
  256. 2d10a5f4-37f0-41b2-96ac-a973b1df024d,
  257. e13d3bdb-7ae3-4101-81a4-6f17ee21df2d,
  258. 104b2be0-7b75-44eb-ac8e-f9220bdfa140,
  259. c5c551a5-0386-4973-aa37-b65cbeb2624b,
  260. 83f72d71-4b71-428e-b604-6f16599a9f37]>
  261. >>> res.parent.get()
  262. 8
  263. .. _canvas-chain:
  264. Chains
  265. ------
  266. .. versionadded:: 3.0
  267. Tasks can be linked together, which in practice means adding
  268. a callback task::
  269. >>> res = add.apply_async((2, 2), link=mul.s(16))
  270. >>> res.get()
  271. 4
  272. The linked task will be applied with the result of its parent
  273. task as the first argument, which in the above case will result
  274. in ``mul(4, 16)`` since the result is 4.
  275. The results will keep track of any subtasks called by the original task,
  276. and this can be accessed from the result instance::
  277. >>> res.children
  278. [<AsyncResult: 8c350acf-519d-4553-8a53-4ad3a5c5aeb4>]
  279. >>> res.children[0].get()
  280. 64
  281. The result instance also has a :meth:`~@AsyncResult.collect` method
  282. that treats the result as a graph, enabling you to iterate over
  283. the results::
  284. >>> list(res.collect())
  285. [(<AsyncResult: 7b720856-dc5f-4415-9134-5c89def5664e>, 4),
  286. (<AsyncResult: 8c350acf-519d-4553-8a53-4ad3a5c5aeb4>, 64)]
  287. By default :meth:`~@AsyncResult.collect` will raise an
  288. :exc:`~@IncompleteStream` exception if the graph is not fully
  289. formed (one of the tasks has not completed yet),
  290. but you can get an intermediate representation of the graph
  291. too::
  292. >>> for result, value in res.collect(intermediate=True)):
  293. ....
  294. You can link together as many tasks as you like,
  295. and signatures can be linked too::
  296. >>> s = add.s(2, 2)
  297. >>> s.link(mul.s(4))
  298. >>> s.link(log_result.s())
  299. You can also add *error callbacks* using the ``link_error`` argument::
  300. >>> add.apply_async((2, 2), link_error=log_error.s())
  301. >>> add.signature((2, 2), link_error=log_error.s())
  302. Since exceptions can only be serialized when pickle is used
  303. the error callbacks take the id of the parent task as argument instead:
  304. .. code-block:: python
  305. from __future__ import print_function
  306. import os
  307. from proj.celery import app
  308. @app.task
  309. def log_error(task_id):
  310. result = app.AsyncResult(task_id)
  311. result.get(propagate=False) # make sure result written.
  312. with open(os.path.join('/var/errors', task_id), 'a') as fh:
  313. print('--\n\n{0} {1} {2}'.format(
  314. task_id, result.result, result.traceback), file=fh)
  315. To make it even easier to link tasks together there is
  316. a special signature called :class:`~celery.chain` that lets
  317. you chain tasks together:
  318. .. code-block:: python
  319. >>> from celery import chain
  320. >>> from proj.tasks import add, mul
  321. # (4 + 4) * 8 * 10
  322. >>> res = chain(add.s(4, 4), mul.s(8), mul.s(10))
  323. proj.tasks.add(4, 4) | proj.tasks.mul(8) | proj.tasks.mul(10)
  324. Calling the chain will call the tasks in the current process
  325. and return the result of the last task in the chain::
  326. >>> res = chain(add.s(4, 4), mul.s(8), mul.s(10))()
  327. >>> res.get()
  328. 640
  329. It also sets ``parent`` attributes so that you can
  330. work your way up the chain to get intermediate results::
  331. >>> res.parent.get()
  332. 64
  333. >>> res.parent.parent.get()
  334. 8
  335. >>> res.parent.parent
  336. <AsyncResult: eeaad925-6778-4ad1-88c8-b2a63d017933>
  337. Chains can also be made using the ``|`` (pipe) operator::
  338. >>> (add.s(2, 2) | mul.s(8) | mul.s(10)).apply_async()
  339. Graphs
  340. ~~~~~~
  341. In addition you can work with the result graph as a
  342. :class:`~celery.datastructures.DependencyGraph`:
  343. .. code-block:: python
  344. >>> res = chain(add.s(4, 4), mul.s(8), mul.s(10))()
  345. >>> res.parent.parent.graph
  346. 285fa253-fcf8-42ef-8b95-0078897e83e6(1)
  347. 463afec2-5ed4-4036-b22d-ba067ec64f52(0)
  348. 872c3995-6fa0-46ca-98c2-5a19155afcf0(2)
  349. 285fa253-fcf8-42ef-8b95-0078897e83e6(1)
  350. 463afec2-5ed4-4036-b22d-ba067ec64f52(0)
  351. You can even convert these graphs to *dot* format::
  352. >>> with open('graph.dot', 'w') as fh:
  353. ... res.parent.parent.graph.to_dot(fh)
  354. and create images:
  355. .. code-block:: bash
  356. $ dot -Tpng graph.dot -o graph.png
  357. .. image:: ../images/result_graph.png
  358. .. _canvas-group:
  359. Groups
  360. ------
  361. .. versionadded:: 3.0
  362. A group can be used to execute several tasks in parallel.
  363. The :class:`~celery.group` function takes a list of signatures::
  364. >>> from celery import group
  365. >>> from proj.tasks import add
  366. >>> group(add.s(2, 2), add.s(4, 4))
  367. (proj.tasks.add(2, 2), proj.tasks.add(4, 4))
  368. If you **call** the group, the tasks will be applied
  369. one after one in the current process, and a :class:`~celery.result.GroupResult`
  370. instance is returned which can be used to keep track of the results,
  371. or tell how many tasks are ready and so on::
  372. >>> g = group(add.s(2, 2), add.s(4, 4))
  373. >>> res = g()
  374. >>> res.get()
  375. [4, 8]
  376. Group also supports iterators::
  377. >>> group(add.s(i, i) for i in xrange(100))()
  378. A group is a signature object, so it can be used in combination
  379. with other signatures.
  380. Group Results
  381. ~~~~~~~~~~~~~
  382. The group task returns a special result too,
  383. this result works just like normal task results, except
  384. that it works on the group as a whole::
  385. >>> from celery import group
  386. >>> from tasks import add
  387. >>> job = group([
  388. ... add.s(2, 2),
  389. ... add.s(4, 4),
  390. ... add.s(8, 8),
  391. ... add.s(16, 16),
  392. ... add.s(32, 32),
  393. ... ])
  394. >>> result = job.apply_async()
  395. >>> result.ready() # have all subtasks completed?
  396. True
  397. >>> result.successful() # were all subtasks successful?
  398. True
  399. >>> result.get()
  400. [4, 8, 16, 32, 64]
  401. The :class:`~celery.result.GroupResult` takes a list of
  402. :class:`~celery.result.AsyncResult` instances and operates on them as
  403. if it was a single task.
  404. It supports the following operations:
  405. * :meth:`~celery.result.GroupResult.successful`
  406. Return :const:`True` if all of the subtasks finished
  407. successfully (e.g. did not raise an exception).
  408. * :meth:`~celery.result.GroupResult.failed`
  409. Return :const:`True` if any of the subtasks failed.
  410. * :meth:`~celery.result.GroupResult.waiting`
  411. Return :const:`True` if any of the subtasks
  412. is not ready yet.
  413. * :meth:`~celery.result.GroupResult.ready`
  414. Return :const:`True` if all of the subtasks
  415. are ready.
  416. * :meth:`~celery.result.GroupResult.completed_count`
  417. Return the number of completed subtasks.
  418. * :meth:`~celery.result.GroupResult.revoke`
  419. Revoke all of the subtasks.
  420. * :meth:`~celery.result.GroupResult.join`
  421. Gather the results for all of the subtasks
  422. and return a list with them ordered by the order of which they
  423. were called.
  424. .. _canvas-chord:
  425. Chords
  426. ------
  427. .. versionadded:: 2.3
  428. .. note::
  429. Tasks used within a chord must *not* ignore their results. If the result
  430. backend is disabled for *any* task (header or body) in your chord you
  431. should read ":ref:`chord-important-notes`".
  432. A chord is a task that only executes after all of the tasks in a group have
  433. finished executing.
  434. Let's calculate the sum of the expression
  435. :math:`1 + 1 + 2 + 2 + 3 + 3 ... n + n` up to a hundred digits.
  436. First you need two tasks, :func:`add` and :func:`tsum` (:func:`sum` is
  437. already a standard function):
  438. .. code-block:: python
  439. @app.task
  440. def add(x, y):
  441. return x + y
  442. @app.task
  443. def tsum(numbers):
  444. return sum(numbers)
  445. Now you can use a chord to calculate each addition step in parallel, and then
  446. get the sum of the resulting numbers::
  447. >>> from celery import chord
  448. >>> from tasks import add, tsum
  449. >>> chord(add.s(i, i)
  450. ... for i in xrange(100))(tsum.s()).get()
  451. 9900
  452. This is obviously a very contrived example, the overhead of messaging and
  453. synchronization makes this a lot slower than its Python counterpart::
  454. sum(i + i for i in xrange(100))
  455. The synchronization step is costly, so you should avoid using chords as much
  456. as possible. Still, the chord is a powerful primitive to have in your toolbox
  457. as synchronization is a required step for many parallel algorithms.
  458. Let's break the chord expression down:
  459. .. code-block:: python
  460. >>> callback = tsum.s()
  461. >>> header = [add.s(i, i) for i in range(100)]
  462. >>> result = chord(header)(callback)
  463. >>> result.get()
  464. 9900
  465. Remember, the callback can only be executed after all of the tasks in the
  466. header have returned. Each step in the header is executed as a task, in
  467. parallel, possibly on different nodes. The callback is then applied with
  468. the return value of each task in the header. The task id returned by
  469. :meth:`chord` is the id of the callback, so you can wait for it to complete
  470. and get the final return value (but remember to :ref:`never have a task wait
  471. for other tasks <task-synchronous-subtasks>`)
  472. .. _chord-errors:
  473. Error handling
  474. ~~~~~~~~~~~~~~
  475. So what happens if one of the tasks raises an exception?
  476. Errors will propagate to the callback, so the callback will not be executed
  477. instead the callback changes to failure state, and the error is set
  478. to the :exc:`~@ChordError` exception:
  479. .. code-block:: python
  480. >>> c = chord([add.s(4, 4), raising_task.s(), add.s(8, 8)])
  481. >>> result = c()
  482. >>> result.get()
  483. Traceback (most recent call last):
  484. File "<stdin>", line 1, in <module>
  485. File "*/celery/result.py", line 120, in get
  486. interval=interval)
  487. File "*/celery/backends/amqp.py", line 150, in wait_for
  488. raise self.exception_to_python(meta['result'])
  489. celery.exceptions.ChordError: Dependency 97de6f3f-ea67-4517-a21c-d867c61fcb47
  490. raised ValueError('something something',)
  491. While the traceback may be different depending on which result backend is
  492. being used, you can see the error description includes the id of the task that failed
  493. and a string representation of the original exception. You can also
  494. find the original traceback in ``result.traceback``.
  495. Note that the rest of the tasks will still execute, so the third task
  496. (``add.s(8, 8)``) is still executed even though the middle task failed.
  497. Also the :exc:`~@ChordError` only shows the task that failed
  498. first (in time): it does not respect the ordering of the header group.
  499. .. _chord-important-notes:
  500. Important Notes
  501. ~~~~~~~~~~~~~~~
  502. Tasks used within a chord must *not* ignore their results. In practice this
  503. means that you must enable a :const:`CELERY_RESULT_BACKEND` in order to use
  504. chords. Additionally, if :const:`CELERY_IGNORE_RESULT` is set to :const:`True`
  505. in your configuration, be sure that the individual tasks to be used within
  506. the chord are defined with :const:`ignore_result=False`. This applies to both
  507. Task subclasses and decorated tasks.
  508. Example Task subclass:
  509. .. code-block:: python
  510. class MyTask(Task):
  511. abstract = True
  512. ignore_result = False
  513. Example decorated task:
  514. .. code-block:: python
  515. @app.task(ignore_result=False)
  516. def another_task(project):
  517. do_something()
  518. By default the synchronization step is implemented by having a recurring task
  519. poll the completion of the group every second, calling the signature when
  520. ready.
  521. Example implementation:
  522. .. code-block:: python
  523. from celery import maybe_signature
  524. @app.task(bind=True)
  525. def unlock_chord(self, group, callback, interval=1, max_retries=None):
  526. if group.ready():
  527. return maybe_signature(callback).delay(group.join())
  528. raise self.retry(countdown=interval, max_retries=max_retries)
  529. This is used by all result backends except Redis and Memcached, which
  530. increment a counter after each task in the header, then applying the callback
  531. when the counter exceeds the number of tasks in the set. *Note:* chords do not
  532. properly work with Redis before version 2.2; you will need to upgrade to at
  533. least 2.2 to use them.
  534. The Redis and Memcached approach is a much better solution, but not easily
  535. implemented in other backends (suggestions welcome!).
  536. .. note::
  537. If you are using chords with the Redis result backend and also overriding
  538. the :meth:`Task.after_return` method, you need to make sure to call the
  539. super method or else the chord callback will not be applied.
  540. .. code-block:: python
  541. def after_return(self, *args, **kwargs):
  542. do_something()
  543. super(MyTask, self).after_return(*args, **kwargs)
  544. .. _canvas-map:
  545. Map & Starmap
  546. -------------
  547. :class:`~celery.map` and :class:`~celery.starmap` are built-in tasks
  548. that calls the task for every element in a sequence.
  549. They differ from group in that
  550. - only one task message is sent
  551. - the operation is sequential.
  552. For example using ``map``:
  553. .. code-block:: python
  554. >>> from proj.tasks import add
  555. >>> ~xsum.map([range(10), range(100)])
  556. [45, 4950]
  557. is the same as having a task doing:
  558. .. code-block:: python
  559. @app.task
  560. def temp():
  561. return [xsum(range(10)), xsum(range(100))]
  562. and using ``starmap``::
  563. >>> ~add.starmap(zip(range(10), range(10)))
  564. [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]
  565. is the same as having a task doing:
  566. .. code-block:: python
  567. @app.task
  568. def temp():
  569. return [add(i, i) for i in range(10)]
  570. Both ``map`` and ``starmap`` are signature objects, so they can be used as
  571. other signatures and combined in groups etc., for example
  572. to call the starmap after 10 seconds::
  573. >>> add.starmap(zip(range(10), range(10))).apply_async(countdown=10)
  574. .. _canvas-chunks:
  575. Chunks
  576. ------
  577. Chunking lets you divide an iterable of work into pieces, so that if
  578. you have one million objects, you can create 10 tasks with hundred
  579. thousand objects each.
  580. Some may worry that chunking your tasks results in a degradation
  581. of parallelism, but this is rarely true for a busy cluster
  582. and in practice since you are avoiding the overhead of messaging
  583. it may considerably increase performance.
  584. To create a chunks signature you can use :meth:`@Task.chunks`:
  585. .. code-block:: python
  586. >>> add.chunks(zip(range(100), range(100)), 10)
  587. As with :class:`~celery.group` the act of sending the messages for
  588. the chunks will happen in the current process when called:
  589. .. code-block:: python
  590. >>> from proj.tasks import add
  591. >>> res = add.chunks(zip(range(100), range(100)), 10)()
  592. >>> res.get()
  593. [[0, 2, 4, 6, 8, 10, 12, 14, 16, 18],
  594. [20, 22, 24, 26, 28, 30, 32, 34, 36, 38],
  595. [40, 42, 44, 46, 48, 50, 52, 54, 56, 58],
  596. [60, 62, 64, 66, 68, 70, 72, 74, 76, 78],
  597. [80, 82, 84, 86, 88, 90, 92, 94, 96, 98],
  598. [100, 102, 104, 106, 108, 110, 112, 114, 116, 118],
  599. [120, 122, 124, 126, 128, 130, 132, 134, 136, 138],
  600. [140, 142, 144, 146, 148, 150, 152, 154, 156, 158],
  601. [160, 162, 164, 166, 168, 170, 172, 174, 176, 178],
  602. [180, 182, 184, 186, 188, 190, 192, 194, 196, 198]]
  603. while calling ``.apply_async`` will create a dedicated
  604. task so that the individual tasks are applied in a worker
  605. instead::
  606. >>> add.chunks(zip(range(100), range(100), 10)).apply_async()
  607. You can also convert chunks to a group::
  608. >>> group = add.chunks(zip(range(100), range(100), 10)).group()
  609. and with the group skew the countdown of each task by increments
  610. of one::
  611. >>> group.skew(start=1, stop=10)()
  612. which means that the first task will have a countdown of 1, the second
  613. a countdown of 2 and so on.