I am building a commandline app to check proxies, which has a daemon flag (running forever or just once).There are producers, checkers and storers. I want them to run in parallel and be able to close if formers are closed:Make them run in parallel The Working Flow:When all producers are done, tell all checkers and allow them to exit, producers exitWhen all checkers are done, tell all stores and allow them to exit, checkers exitAll stores exitNow the approach is using a queue: Use queue join to block and loop queue to call each greenlet's close method. This just looks ugly to me.from __future__ import unicode_literals, absolute_importimport gevent.monkeygevent.monkey.patch_socket()import geventfrom gevent.queue import JoinableQueue, Queue, Emptyfrom gevent import Greenletimport requestsimport logginglogger = logging.getLogger('proxy_checker')LOOP_INTERVAL = 0.1class ProxyChecker(gevent.Greenlet):    def __init__(self, test_url, inqueue, outqueue):        self.test_url = test_url        self.inqueue = inqueue        self.outqueue = outqueue        Greenlet.__init__(self)    def _run(self):        self.running = True        while self.running:            try:                p = self.inqueue.get_nowait()                .....                check_proxy(self.test_url, p)                self.inqueue.task_done()                self.outqueue.put_nowait(p)            except Empty as e:                gevent.sleep(LOOP_INTERVAL)            except AssertionError as e:                logger.error(e)            except KeyboardInterrupt:                break        logger.info(ProxyChecker exiting...)    def stop(self):        self.running = Falseclass ProxyStorer(gevent.Greenlet):    def __init__(self, store, inqueue):        self.store = store        self.inqueue = inqueue        Greenlet.__init__(self)    def _run(self):        self.running = True        while self.running:            try:                p = self.inqueue.get_nowait()                self.store.save_or_update(**p)                self.inqueue.task_done()                # logger.debug('save item:%s' % pp)            except Empty as e:                gevent.sleep(LOOP_INTERVAL)            except KeyboardInterrupt:                break            except AssertionError as e:                logger.error(e)            except Exception as e:                logger.exception(e)        logger.info(ProxyStorer exiting...)    def stop(self):        self.running = Falsedef producer(proxy_queue, proxy_list):    for p in proxy_list:        proxy_queue.put(p)def run_check(test_url, proxy_list, store, deamon=False):    proxy_queue = JoinableQueue()    store_queue = JoinableQueue()    checker_num = 20    storer_num = 3    checker_threads = []    storeds_threads = []    producer_thread = gevent.spawn(producer, proxy_queue, proxy_list)    for i in range(checker_num):        checker = ProxyChecker(test_url, proxy_queue, store_queue)        checker.start()        checker_threads.append(checker)    for i in range(storer_num):        storer = ProxyStorer(store, store_queue)        storer.start()        storeds_threads.append(storer)    if not deamon:        producer_thread.join()        proxy_queue.join()        for c in checker_threads:            c.stop()        store_queue.join()        for s in storeds_threads:            s.stop()    gevent.joinall(checker_threads)    gevent.joinall(storeds_threads)    logger.info('Total %s proxies checked!', len(proxy_list))I think the flaws are:JoinableQueue, which forces me to write a queue.task_done every queue.getloop checker_threads to call the stop functionIs there any way to do some notification, such as flags likeproducer_done (set when the only one producer done)checker_done (set when producer_done==True and check_queue.qsize() == 0)such as add below check in checker's _run :while True:    try:    ....  code    except Empty as e:        # avoid to checking self.inqueue.qsize() == 0 in loop        if producer_done:            checker_done = True            break        else:            gevent.sleep(LOOP_INTERVAL)    ....  codelogger.info(ProxyChecker exiting...)Same as storer.
Make one greenlet tell another to shutdown
python;multithreading;asynchronous;event handling