base.py 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430
  1. # -*- coding: utf-8 -*-
  2. """
  3. celery.backends.base
  4. ~~~~~~~~~~~~~~~~~~~~
  5. Result backend base classes.
  6. - :class:`BaseBackend` defines the interface.
  7. - :class:`KeyValueStoreBackend` is a common base class
  8. using K/V semantics like _get and _put.
  9. """
  10. from __future__ import absolute_import
  11. import time
  12. import sys
  13. from datetime import timedelta
  14. from future_builtins import map
  15. from kombu import serialization
  16. from kombu.utils.encoding import bytes_to_str, ensure_bytes, from_utf8
  17. from celery import states
  18. from celery.app import current_task
  19. from celery.datastructures import LRUCache
  20. from celery.exceptions import TimeoutError, TaskRevokedError
  21. from celery.result import from_serializable
  22. from celery.utils import timeutils
  23. from celery.utils.serialization import (
  24. get_pickled_exception,
  25. get_pickleable_exception,
  26. create_exception_cls,
  27. )
  28. EXCEPTION_ABLE_CODECS = frozenset(['pickle', 'yaml'])
  29. is_py3k = sys.version_info >= (3, 0)
  30. def unpickle_backend(cls, args, kwargs):
  31. """Returns an unpickled backend."""
  32. return cls(*args, **kwargs)
  33. class BaseBackend(object):
  34. READY_STATES = states.READY_STATES
  35. UNREADY_STATES = states.UNREADY_STATES
  36. EXCEPTION_STATES = states.EXCEPTION_STATES
  37. TimeoutError = TimeoutError
  38. #: Time to sleep between polling each individual item
  39. #: in `ResultSet.iterate`. as opposed to the `interval`
  40. #: argument which is for each pass.
  41. subpolling_interval = None
  42. #: If true the backend must implement :meth:`get_many`.
  43. supports_native_join = False
  44. def __init__(self, app=None, serializer=None, max_cached_results=None,
  45. **kwargs):
  46. from celery.app import app_or_default
  47. self.app = app_or_default(app)
  48. self.serializer = serializer or self.app.conf.CELERY_RESULT_SERIALIZER
  49. (self.content_type,
  50. self.content_encoding,
  51. self.encoder) = serialization.registry._encoders[self.serializer]
  52. self._cache = LRUCache(limit=max_cached_results or
  53. self.app.conf.CELERY_MAX_CACHED_RESULTS)
  54. def mark_as_started(self, task_id, **meta):
  55. """Mark a task as started"""
  56. return self.store_result(task_id, meta, status=states.STARTED)
  57. def mark_as_done(self, task_id, result):
  58. """Mark task as successfully executed."""
  59. return self.store_result(task_id, result, status=states.SUCCESS)
  60. def mark_as_failure(self, task_id, exc, traceback=None):
  61. """Mark task as executed with failure. Stores the execption."""
  62. return self.store_result(task_id, exc, status=states.FAILURE,
  63. traceback=traceback)
  64. def mark_as_retry(self, task_id, exc, traceback=None):
  65. """Mark task as being retries. Stores the current
  66. exception (if any)."""
  67. return self.store_result(task_id, exc, status=states.RETRY,
  68. traceback=traceback)
  69. def mark_as_revoked(self, task_id, reason=''):
  70. return self.store_result(task_id, TaskRevokedError(reason),
  71. status=states.REVOKED, traceback=None)
  72. def prepare_exception(self, exc):
  73. """Prepare exception for serialization."""
  74. if self.serializer in EXCEPTION_ABLE_CODECS:
  75. return get_pickleable_exception(exc)
  76. return {'exc_type': type(exc).__name__, 'exc_message': str(exc)}
  77. def exception_to_python(self, exc):
  78. """Convert serialized exception to Python exception."""
  79. if self.serializer in EXCEPTION_ABLE_CODECS:
  80. return get_pickled_exception(exc)
  81. return create_exception_cls(from_utf8(exc['exc_type']),
  82. sys.modules[__name__])(exc['exc_message'])
  83. def prepare_value(self, result):
  84. """Prepare value for storage."""
  85. return result
  86. def encode(self, data):
  87. _, _, payload = serialization.encode(data, serializer=self.serializer)
  88. return payload
  89. def decode(self, payload):
  90. payload = is_py3k and payload or str(payload)
  91. return serialization.decode(payload,
  92. content_type=self.content_type,
  93. content_encoding=self.content_encoding)
  94. def wait_for(self, task_id, timeout=None, propagate=True, interval=0.5):
  95. """Wait for task and return its result.
  96. If the task raises an exception, this exception
  97. will be re-raised by :func:`wait_for`.
  98. If `timeout` is not :const:`None`, this raises the
  99. :class:`celery.exceptions.TimeoutError` exception if the operation
  100. takes longer than `timeout` seconds.
  101. """
  102. time_elapsed = 0.0
  103. while 1:
  104. status = self.get_status(task_id)
  105. if status == states.SUCCESS:
  106. return self.get_result(task_id)
  107. elif status in states.PROPAGATE_STATES:
  108. result = self.get_result(task_id)
  109. if propagate:
  110. raise result
  111. return result
  112. # avoid hammering the CPU checking status.
  113. time.sleep(interval)
  114. time_elapsed += interval
  115. if timeout and time_elapsed >= timeout:
  116. raise TimeoutError('The operation timed out.')
  117. def prepare_expires(self, value, type=None):
  118. if value is None:
  119. value = self.app.conf.CELERY_TASK_RESULT_EXPIRES
  120. if isinstance(value, timedelta):
  121. value = timeutils.timedelta_seconds(value)
  122. if value is not None and type:
  123. return type(value)
  124. return value
  125. def encode_result(self, result, status):
  126. if status in self.EXCEPTION_STATES and isinstance(result, Exception):
  127. return self.prepare_exception(result)
  128. else:
  129. return self.prepare_value(result)
  130. def store_result(self, task_id, result, status, traceback=None, **kwargs):
  131. """Update task state and result."""
  132. result = self.encode_result(result, status)
  133. return self._store_result(task_id, result, status, traceback, **kwargs)
  134. def forget(self, task_id):
  135. self._cache.pop(task_id, None)
  136. self._forget(task_id)
  137. def _forget(self, task_id):
  138. raise NotImplementedError('backend does not implement forget.')
  139. def get_status(self, task_id):
  140. """Get the status of a task."""
  141. return self.get_task_meta(task_id)['status']
  142. def get_traceback(self, task_id):
  143. """Get the traceback for a failed task."""
  144. return self.get_task_meta(task_id).get('traceback')
  145. def get_result(self, task_id):
  146. """Get the result of a task."""
  147. meta = self.get_task_meta(task_id)
  148. if meta['status'] in self.EXCEPTION_STATES:
  149. return self.exception_to_python(meta['result'])
  150. else:
  151. return meta['result']
  152. def get_children(self, task_id):
  153. """Get the list of subtasks sent by a task."""
  154. try:
  155. return self.get_task_meta(task_id)['children']
  156. except KeyError:
  157. pass
  158. def get_task_meta(self, task_id, cache=True):
  159. if cache:
  160. try:
  161. return self._cache[task_id]
  162. except KeyError:
  163. pass
  164. meta = self._get_task_meta_for(task_id)
  165. if cache and meta.get('status') == states.SUCCESS:
  166. self._cache[task_id] = meta
  167. return meta
  168. def reload_task_result(self, task_id):
  169. """Reload task result, even if it has been previously fetched."""
  170. self._cache[task_id] = self.get_task_meta(task_id, cache=False)
  171. def reload_group_result(self, group_id):
  172. """Reload group result, even if it has been previously fetched."""
  173. self._cache[group_id] = self.get_group_meta(group_id, cache=False)
  174. def get_group_meta(self, group_id, cache=True):
  175. if cache:
  176. try:
  177. return self._cache[group_id]
  178. except KeyError:
  179. pass
  180. meta = self._restore_group(group_id)
  181. if cache and meta is not None:
  182. self._cache[group_id] = meta
  183. return meta
  184. def restore_group(self, group_id, cache=True):
  185. """Get the result for a group."""
  186. meta = self.get_group_meta(group_id, cache=cache)
  187. if meta:
  188. return meta['result']
  189. def save_group(self, group_id, result):
  190. """Store the result of an executed group."""
  191. return self._save_group(group_id, result)
  192. def delete_group(self, group_id):
  193. self._cache.pop(group_id, None)
  194. return self._delete_group(group_id)
  195. def cleanup(self):
  196. """Backend cleanup. Is run by
  197. :class:`celery.task.DeleteExpiredTaskMetaTask`."""
  198. pass
  199. def process_cleanup(self):
  200. """Cleanup actions to do at the end of a task worker process."""
  201. pass
  202. def on_chord_part_return(self, task, propagate=False):
  203. pass
  204. def fallback_chord_unlock(self, group_id, body, result=None, **kwargs):
  205. kwargs['result'] = [r.id for r in result]
  206. self.app.tasks['celery.chord_unlock'].apply_async((group_id, body, ),
  207. kwargs, countdown=1)
  208. on_chord_apply = fallback_chord_unlock
  209. def current_task_children(self):
  210. current = current_task()
  211. if current:
  212. return [r.serializable() for r in current.request.children]
  213. def __reduce__(self, args=(), kwargs={}):
  214. return (unpickle_backend, (self.__class__, args, kwargs))
  215. BaseDictBackend = BaseBackend # XXX compat
  216. class KeyValueStoreBackend(BaseBackend):
  217. task_keyprefix = ensure_bytes('celery-task-meta-')
  218. group_keyprefix = ensure_bytes('celery-taskset-meta-')
  219. chord_keyprefix = ensure_bytes('chord-unlock-')
  220. implements_incr = False
  221. def get(self, key):
  222. raise NotImplementedError('Must implement the get method.')
  223. def mget(self, keys):
  224. raise NotImplementedError('Does not support get_many')
  225. def set(self, key, value):
  226. raise NotImplementedError('Must implement the set method.')
  227. def delete(self, key):
  228. raise NotImplementedError('Must implement the delete method')
  229. def incr(self, key):
  230. raise NotImplementedError('Does not implement incr')
  231. def expire(self, key, value):
  232. pass
  233. def get_key_for_task(self, task_id):
  234. """Get the cache key for a task by id."""
  235. return self.task_keyprefix + ensure_bytes(task_id)
  236. def get_key_for_group(self, group_id):
  237. """Get the cache key for a group by id."""
  238. return self.group_keyprefix + ensure_bytes(group_id)
  239. def get_key_for_chord(self, group_id):
  240. """Get the cache key for the chord waiting on group with given id."""
  241. return self.chord_keyprefix + ensure_bytes(group_id)
  242. def _strip_prefix(self, key):
  243. """Takes bytes, emits string."""
  244. key = ensure_bytes(key)
  245. for prefix in self.task_keyprefix, self.group_keyprefix:
  246. if key.startswith(prefix):
  247. return bytes_to_str(key[len(prefix):])
  248. return bytes_to_str(key)
  249. def _mget_to_results(self, values, keys):
  250. if hasattr(values, 'items'):
  251. # client returns dict so mapping preserved.
  252. return dict((self._strip_prefix(k), self.decode(v))
  253. for k, v in values.iteritems()
  254. if v is not None)
  255. else:
  256. # client returns list so need to recreate mapping.
  257. return dict((bytes_to_str(keys[i]), self.decode(value))
  258. for i, value in enumerate(values)
  259. if value is not None)
  260. def get_many(self, task_ids, timeout=None, interval=0.5):
  261. ids = set(task_ids)
  262. cached_ids = set()
  263. for task_id in ids:
  264. try:
  265. cached = self._cache[task_id]
  266. except KeyError:
  267. pass
  268. else:
  269. if cached['status'] in states.READY_STATES:
  270. yield bytes_to_str(task_id), cached
  271. cached_ids.add(task_id)
  272. ids ^= cached_ids
  273. iterations = 0
  274. while ids:
  275. keys = list(ids)
  276. r = self._mget_to_results(self.mget([self.get_key_for_task(k)
  277. for k in keys]), keys)
  278. self._cache.update(r)
  279. ids ^= set(map(bytes_to_str, r))
  280. for key, value in r.iteritems():
  281. yield bytes_to_str(key), value
  282. if timeout and iterations * interval >= timeout:
  283. raise TimeoutError('Operation timed out ({0})'.format(timeout))
  284. time.sleep(interval) # don't busy loop.
  285. iterations += 0
  286. def _forget(self, task_id):
  287. self.delete(self.get_key_for_task(task_id))
  288. def _store_result(self, task_id, result, status, traceback=None):
  289. meta = {'status': status, 'result': result, 'traceback': traceback,
  290. 'children': self.current_task_children()}
  291. self.set(self.get_key_for_task(task_id), self.encode(meta))
  292. return result
  293. def _save_group(self, group_id, result):
  294. self.set(self.get_key_for_group(group_id),
  295. self.encode({'result': result.serializable()}))
  296. return result
  297. def _delete_group(self, group_id):
  298. self.delete(self.get_key_for_group(group_id))
  299. def _get_task_meta_for(self, task_id):
  300. """Get task metadata for a task by id."""
  301. meta = self.get(self.get_key_for_task(task_id))
  302. if not meta:
  303. return {'status': states.PENDING, 'result': None}
  304. return self.decode(meta)
  305. def _restore_group(self, group_id):
  306. """Get task metadata for a task by id."""
  307. meta = self.get(self.get_key_for_group(group_id))
  308. # previously this was always pickled, but later this
  309. # was extended to support other serializers, so the
  310. # structure is kind of weird.
  311. if meta:
  312. meta = self.decode(meta)
  313. result = meta['result']
  314. if isinstance(result, (list, tuple)):
  315. return {'result': from_serializable(result)}
  316. return meta
  317. def on_chord_apply(self, group_id, body, result=None, **kwargs):
  318. if self.implements_incr:
  319. self.app.GroupResult(group_id, result).save()
  320. else:
  321. self.fallback_chord_unlock(group_id, body, result, **kwargs)
  322. def on_chord_part_return(self, task, propagate=False):
  323. if not self.implements_incr:
  324. return
  325. from celery import subtask
  326. from celery.result import GroupResult
  327. gid = task.request.group
  328. if not gid:
  329. return
  330. key = self.get_key_for_chord(gid)
  331. deps = GroupResult.restore(gid, backend=task.backend)
  332. val = self.incr(key)
  333. if val >= len(deps):
  334. subtask(task.request.chord).delay(deps.join(propagate=propagate))
  335. deps.delete()
  336. self.client.delete(key)
  337. else:
  338. self.expire(key, 86400)
  339. class DisabledBackend(BaseBackend):
  340. _cache = {} # need this attribute to reset cache in tests.
  341. def store_result(self, *args, **kwargs):
  342. pass
  343. def _is_disabled(self, *args, **kwargs):
  344. raise NotImplementedError('No result backend configured. '
  345. 'Please see the documentation for more information.')
  346. wait_for = get_status = get_result = get_traceback = _is_disabled