3
\$\begingroup\$

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:
    1. When all producers are done, tell all checkers and allow them to exit, producers exit
    2. When all checkers are done, tell all stores and allow them to exit, checkers exit
    3. All stores exit

Now 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_import

import gevent.monkey

gevent.monkey.patch_socket()

import gevent
from gevent.queue import JoinableQueue, Queue, Empty
from gevent import Greenlet
import requests

import logging

logger = logging.getLogger('proxy_checker')


LOOP_INTERVAL = 0.1

class 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 = False


class 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 = False

def 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:

  1. JoinableQueue, which forces me to write a queue.task_done every queue.get

  2. loop checker_threads to call the stop function

Is there any way to do some notification, such as flags like

  • producer_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)

    ....  code

logger.info("ProxyChecker exiting...")

Same as storer.

\$\endgroup\$

1 Answer 1

1
\$\begingroup\$

Sure, you can send 'done' messages through the queue. So:

DONE = Object() # unique object

while self.running:
    try:
        p = self.inqueue.get_nowait()
        if p is DONE:
            self.inqueue.task_done()
            self.outqueue.put_nowait(DONE)
            self.outqueue.close()
            break

is checks for object identity, so there's no way to get real data that will cause the is DONE clause to match.

And then you can get rid of the loops that call .stop() and just do:

if not deamon:
    producer_thread.join()
    proxy_queue.join()
    store_queue.join()

...and the threads will close the queues and exit.

Also, more minorly,

checker_threads = []
...
for i in range(checker_num):
    checker = ProxyChecker(test_url, proxy_queue, store_queue)
    checker.start()
    checker_threads.append(checker)

looks like a place for a list comprehension. Consider:

checker_threads = [ ProxyChecker(test_url, proxy_queue, store_queue) for i in range(checker_num) ]

for thread in checker_threads: thread.start()

or if you want to make the list comprehension line a bit shorter:

PC = lambda : ProxyChecker(test_url, proxy_queue, store_queue)
checker_threads = [ PC() for i in range(checker_num) ]
for thread in checker_threads: thread.start()

It's a little less efficient (you loop twice instead of once) but it saves a line or two.

You can, of course, do something similar with storer_threads.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.