Python: gevent源代码分析之用gevent threadpool实现多进程任务调度

import time
import gevent
from gevent.threadpool import ThreadPool
def sum(i):
    i = 0
    for i in xrange(100000000):
        i += i+1 * 20 / 10 * 10 /10
def test(n,m):
    m=m
    vals = []
    keys = []
    for i in xrange(m):
        vals.append(i)
        keys.append('a%s'%i)
    d = None
    for i in xrange(n):
        d = dict(zip(keys, vals))
    return d
pool = ThreadPool(20)
start = time.time()
for _ in xrange(10):
    pool.spawn(test, 1000000,100)
gevent.wait()
delay = time.time() - start
print 'Running "time.sleep(1)>" 4 times with 3 threads. Should take about 2 seconds: %.3fs' % delay 

def _set_size(self, size):
    if size < 0:
        raise ValueError('Size of the pool cannot be negative: %r' % (size, ))
    if size > self._maxsize:
        raise ValueError('Size of the pool cannot be bigger than maxsize: %r &gt; %r' % (size, self._maxsize))
    if self.manager:
        self.manager.kill()
    while self._size < size:
        self._add_thread()
    delay = 0.0001
    while self._size > size:
        while self._size - size &gt; self.task_queue.unfinished_tasks:
            self.task_queue.put(None)
        if getcurrent() is self.hub:
            break
        sleep(delay)
        delay = min(delay * 2, .05)
    if self._size:
        self.fork_watcher.start(self._on_fork)
    else:
        self.fork_watcher.stop()
size = property(_get_size, _set_size)
def _init(self, maxsize):
    self._size = 0
    self._semaphore = Semaphore(1)
    self._lock = Lock()
    self.task_queue = Queue()
    self._set_maxsize(maxsize)
def _on_fork(self):
    # fork() only leaves one thread; also screws up locks;
    # let's re-create locks and threads
    pid = os.getpid()
    if pid != self.pid:
        self.pid = pid
        # Do not mix fork() and threads; since fork() only copies one thread
        # all objects referenced by other threads has refcount that will never
        # go down to 0.
        self._init(self._maxsize)
def join(self):
    delay = 0.0005
    while self.task_queue.unfinished_tasks &gt; 0:
        sleep(delay)
        delay = min(delay * 2, .05)
def kill(self):
    self.size = 0
def _adjust_step(self):
    # if there is a possibility &amp; necessity for adding a thread, do it
    while self._size < self._maxsize and self.task_queue.unfinished_tasks > self._size:
        self._add_thread()
    # while the number of threads is more than maxsize, kill one
    # we do not check what's already in task_queue - it could be all Nones
    while self._size - self._maxsize &gt; self.task_queue.unfinished_tasks:
        self.task_queue.put(None)
    if self._size:
        self.fork_watcher.start(self._on_fork)
    else:
        self.fork_watcher.stop()
def _adjust_wait(self):
    delay = 0.0001
    while True:
        self._adjust_step()
        if self._size <= self._maxsize:
            return
        sleep(delay)
        delay = min(delay * 2, .05)
def adjust(self):
    self._adjust_step()
    if not self.manager and self._size > self._maxsize:
        # might need to feed more Nones into the pool
        self.manager = Greenlet.spawn(self._adjust_wait)
def _add_thread(self):
    with self._lock:
        self._size += 1
    try:
        start_new_thread(self._worker, ())
    except:
        with self._lock:
            self._size -= 1
        raise
def spawn(self, func, *args, **kwargs):
    while True:
        semaphore = self._semaphore
        semaphore.acquire()
        if semaphore is self._semaphore:
            break
    try:
        task_queue = self.task_queue
        result = AsyncResult()
        thread_result = ThreadResult(result, hub=self.hub)
        task_queue.put((func, args, kwargs, thread_result))
        self.adjust()
        # rawlink() must be the last call
        result.rawlink(lambda *args: self._semaphore.release())
        # XXX this _semaphore.release() is competing for order with get()
        # XXX this is not good, just make ThreadResult release the semaphore before doing anything else
    except:
        semaphore.release()
        raise
    return result