Skip to content

Instantly share code, notes, and snippets.

@svieira
Forked from flyer103/test_pool.py
Created December 6, 2013 15:59
Show Gist options
  • Save svieira/7827139 to your computer and use it in GitHub Desktop.
Save svieira/7827139 to your computer and use it in GitHub Desktop.
#!/usr/bin/env python2.7
#coding: utf-8
import gevent
import gevent.pool
class TestPool(object):
def __init__(self, maxsize=10):
self.pool = gevent.pool.Pool(maxsize)
def run(self):
for i in xrange(8):
gworker = gevent.spawn(self._worker, i)
self.pool.add(gworker)
print('sizeof pool is %d' % (len(self.pool),))
print('worker %d in? %r' % (i, gworker in self.pool))
# The only thing I added
self.pool.join()
print('All tasks done.')
def _worker(self, pid):
print('My pid is %d' % (pid,))
gevent.sleep(0)
print('%d has done.' % (pid,))
if __name__ == '__main__':
test = TestPool()
test.run()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment