trace.py 1.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647
  1. from celery import current_app, task, uuid
  2. from celery.five import range
  3. from celery.worker.consumer import Consumer
  4. from celery.worker.job import Request
  5. from celery.concurrency.solo import TaskPool
  6. from celery.app.amqp import TASK_BARE
  7. from time import time
  8. from Queue import Queue
  9. from librabbitmq import Message
  10. import socket
  11. import sys
  12. @task(accept_magic_kwargs=False)
  13. def T():
  14. pass
  15. tid = uuid()
  16. P = TaskPool()
  17. hostname = socket.gethostname()
  18. task = {'task': T.name, 'args': (), 'kwargs': {}, 'id': tid, 'flags': 0}
  19. app = current_app._get_current_object()
  20. ready_queue = Queue()
  21. def on_put(req):
  22. req.execute_using_pool(P)
  23. def on_ack(*a): pass
  24. m = Message(None, {}, {}, task)
  25. ready_queue.put = on_put
  26. x = Consumer(ready_queue, hostname=hostname, app=app)
  27. x.update_strategies()
  28. name = T.name
  29. ts = time()
  30. from celery.datastructures import AttributeDict
  31. from celery.task.trace import trace_task_ret
  32. request = AttributeDict(
  33. {'called_directly': False,
  34. 'callbacks': [],
  35. 'errbacks': [],
  36. 'chord': None}, **task)
  37. for i in range(100000):
  38. trace_task_ret(T, tid, (), {}, request)
  39. print(time() - ts)