/usr/lib/python2.7/dist-packages/celery/concurrency/threads.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 | # -*- coding: utf-8 -*-
"""
celery.concurrency.threads
~~~~~~~~~~~~~~~~~~~~~~~~~~
Pool implementation using threads.
"""
from __future__ import absolute_import
from celery.five import UserDict
from .base import apply_target, BasePool
__all__ = ['TaskPool']
class NullDict(UserDict):
def __setitem__(self, key, value):
pass
class TaskPool(BasePool):
def __init__(self, *args, **kwargs):
try:
import threadpool
except ImportError:
raise ImportError(
'The threaded pool requires the threadpool module.')
self.WorkRequest = threadpool.WorkRequest
self.ThreadPool = threadpool.ThreadPool
super(TaskPool, self).__init__(*args, **kwargs)
def on_start(self):
self._pool = self.ThreadPool(self.limit)
# threadpool stores all work requests until they are processed
# we don't need this dict, and it occupies way too much memory.
self._pool.workRequests = NullDict()
self._quick_put = self._pool.putRequest
self._quick_clear = self._pool._results_queue.queue.clear
def on_stop(self):
self._pool.dismissWorkers(self.limit, do_join=True)
def on_apply(self, target, args=None, kwargs=None, callback=None,
accept_callback=None, **_):
req = self.WorkRequest(apply_target, (target, args, kwargs, callback,
accept_callback))
self._quick_put(req)
# threadpool also has callback support,
# but for some reason the callback is not triggered
# before you've collected the results.
# Clear the results (if any), so it doesn't grow too large.
self._quick_clear()
return req
|