/usr/lib/python2.7/dist-packages/celery/worker/strategy.py is in python-celery 3.1.20-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 | # -*- coding: utf-8 -*-
"""
celery.worker.strategy
~~~~~~~~~~~~~~~~~~~~~~
Task execution strategy (optimization).
"""
from __future__ import absolute_import
import logging
from kombu.async.timer import to_timestamp
from kombu.utils.encoding import safe_repr
from celery.utils.log import get_logger
from celery.utils.timeutils import timezone
from .job import Request
from .state import task_reserved
__all__ = ['default']
logger = get_logger(__name__)
def default(task, app, consumer,
info=logger.info, error=logger.error, task_reserved=task_reserved,
to_system_tz=timezone.to_system):
hostname = consumer.hostname
eventer = consumer.event_dispatcher
Req = Request
connection_errors = consumer.connection_errors
_does_info = logger.isEnabledFor(logging.INFO)
events = eventer and eventer.enabled
send_event = eventer.send
call_at = consumer.timer.call_at
apply_eta_task = consumer.apply_eta_task
rate_limits_enabled = not consumer.disable_rate_limits
get_bucket = consumer.task_buckets.__getitem__
handle = consumer.on_task_request
limit_task = consumer._limit_task
def task_message_handler(message, body, ack, reject, callbacks,
to_timestamp=to_timestamp):
req = Req(body, on_ack=ack, on_reject=reject,
app=app, hostname=hostname,
eventer=eventer, task=task,
connection_errors=connection_errors,
message=message)
if req.revoked():
return
if _does_info:
info('Received task: %s', req)
if events:
send_event(
'task-received',
uuid=req.id, name=req.name,
args=safe_repr(req.args), kwargs=safe_repr(req.kwargs),
retries=req.request_dict.get('retries', 0),
eta=req.eta and req.eta.isoformat(),
expires=req.expires and req.expires.isoformat(),
)
if req.eta:
try:
if req.utc:
eta = to_timestamp(to_system_tz(req.eta))
else:
eta = to_timestamp(req.eta, timezone.local)
except OverflowError as exc:
error("Couldn't convert eta %s to timestamp: %r. Task: %r",
req.eta, exc, req.info(safe=True), exc_info=True)
req.acknowledge()
else:
consumer.qos.increment_eventually()
call_at(eta, apply_eta_task, (req, ), priority=6)
else:
if rate_limits_enabled:
bucket = get_bucket(task.name)
if bucket:
return limit_task(req, bucket, 1)
task_reserved(req)
if callbacks:
[callback() for callback in callbacks]
handle(req)
return task_message_handler
|