trace.py 1.1 KB

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