6

A few days ago I has asked a question on SO about helping me design a paradigm for structuring multiple HTTP requests

Here's the scenario. I would like a have a multi-producer, multi-consumer system. My producers crawl and scrape a few sites and add the links that it finds into a queue. Since I'll be crawling multiple sites, I would like to have multiple producers/crawlers.

The consumers/workers feed off this queue, make TCP/UDP requests to these links and saves the results to my Django DB. I would also like to have multiple-workers as each queue item is totally independent of each other.

People suggested that use a coroutine library for this i.e. Gevent or Eventlet. Having never worked with coroutines, I read that even though the programming paradigm is similar to threaded paradigms, only one thread is actively executing but when blocking calls occur - such as I/O calls - the stacks are switched in-memory and the other green thread takes over until it encounters some sort of a blocking I/O call. Hopefully I got this right? Here's the code from one of my SO posts:

import gevent
from gevent.queue import *
import time
import random

q = JoinableQueue()
workers = []
producers = []


def do_work(wid, value):
    gevent.sleep(random.randint(0,2))
    print 'Task', value, 'done', wid


def worker(wid):
    while True:
        item = q.get()
        try:
            print "Got item %s" % item
            do_work(wid, item)
        finally:
            print "No more items"
            q.task_done()


def producer():
    while True:
        item = random.randint(1, 11)
        if item == 10:
            print "Signal Received"
            return
        else:
            print "Added item %s" % item
            q.put(item)


for i in range(4):
    workers.append(gevent.spawn(worker, random.randint(1, 100000)))

# This doesn't work.
for j in range(2):
    producers.append(gevent.spawn(producer))

# Uncommenting this makes this script work.
# producer()

q.join()

This works well because the sleep calls are blocking calls and when a sleep event occurs, another green thread takes over. This is a lot faster than sequential execution. As you can see, I don't have any code in my program that purposely yields the execution of one thread to another thread. I fail to see how this fits into scenario above as I would like to have all the threads executing simultaneously.

All works fine, but I feel the throughput that I've achieved using Gevent/Eventlets is higher than the original sequentially running program but drastically lower than what could be achieved using real-threading.

If I were to re-implement my program using threading mechanisms, each of my producers and consumers could simultaneously be working without the need to swap stacks in and out like coroutines.

Should this be re-implemented using threading? Is my design wrong? I've failed to see the real benefits of using coroutines.

Maybe my concepts are little muddy but this is what I've assimilated. Any help or clarification of my paradigm and concepts would be great.

Thanks

4
  • Why not use multiple processes? Feb 12, 2012 at 9:29
  • I don't know the pros and cons of multi-threading vs multi-processing so I don't know if its okay or not. Feb 12, 2012 at 9:59
  • 1
    there is no such thing as "real threading" ( only one actual OS thread executes at any given time ) in Python programs without resorting to C extensions ( or heavy weight OS Processes ) because of the Global Interpreter Lock.
    – user177800
    Feb 12, 2012 at 16:19
  • your producer doesn't yield control. There is no concurrency until producers finish.
    – jfs
    Feb 12, 2012 at 19:39

3 Answers 3

6

As you can see, I don't have any code in my program that purposely yields the execution of one thread to another thread. I fail to see how this fits into scenario above as I would like to have all the threads executing simultaneously.

There is a single OS thread but several greenlets. In your case gevent.sleep() allows workers to execute concurrently. Blocking IO calls such as urllib2.urlopen(url).read() do the same if you use urllib2 patched to work with gevent (by calling gevent.monkey.patch_*()).

See also A Curious Course on Coroutines and Concurrency to understand how a code can work concurrently in a single threaded environment.

To compare throughput differences between gevent, threading, multiprocessing you could write the code that compatible with all aproaches:

#!/usr/bin/env python
concurrency_impl = 'gevent' # single process, single thread
##concurrency_impl = 'threading' # single process, multiple threads
##concurrency_impl = 'multiprocessing' # multiple processes

if concurrency_impl == 'gevent':
    import gevent.monkey; gevent.monkey.patch_all()

import logging
import time
import random
from itertools import count, islice

info = logging.info

if concurrency_impl in ['gevent', 'threading']:
    from Queue import Queue as JoinableQueue
    from threading import Thread
if concurrency_impl == 'multiprocessing':
    from multiprocessing import Process as Thread, JoinableQueue

The rest of the script is the same for all concurrency implementations:

def do_work(wid, value):
    time.sleep(random.randint(0,2))
    info("%d Task %s done" % (wid, value))

def worker(wid, q):
    while True:
        item = q.get()
        try:
            info("%d Got item %s" % (wid, item))
            do_work(wid, item)
        finally:
            q.task_done()
            info("%d Done item %s" % (wid, item))

def producer(pid, q):
    for item in iter(lambda: random.randint(1, 11), 10):
        time.sleep(.1) # simulate a green blocking call that yields control
        info("%d Added item %s" % (pid, item))
        q.put(item)
    info("%d Signal Received" % (pid,))

Don't execute code at a module level put it in main():

def main():
    logging.basicConfig(level=logging.INFO,
                        format="%(asctime)s %(process)d %(message)s")

    q = JoinableQueue()
    it = count(1)
    producers = [Thread(target=producer, args=(i, q)) for i in islice(it, 2)]
    workers = [Thread(target=worker, args=(i, q)) for i in islice(it, 4)]
    for t in producers+workers:
        t.daemon = True
        t.start()

    for t in producers: t.join() # put items in the queue
    q.join() # wait while it is empty
    # exit main thread (daemon workers die at this point)

if __name__=="__main__":    
   main()
7
  • Hi Sebastian, I've looked into my code and seen that my producers and consumers are working simultaneously. When a blocking operation occurs in one of my greenlets it yields control to the other greenlets. I've added the missing monkey_patch call so that the socket module is non blocking too but I can't get enough crunch out my processor. A regular PC has enough juice to have more simultaneous connections and more greenlets but I don't get enough speed. I'm very lost and confused as why it doesn't use more processor and work faster. Could you help me understand please? I'm very lost. Thanks. Feb 12, 2012 at 20:33
  • @Mridang Agarwalla: I've commented on the code that you posted in your question. producers do not work concurrently in it.
    – jfs
    Feb 12, 2012 at 20:54
  • 1
    @Mridang Agarwalla: if your problem is IO bound (disk, network) then it doesn't matter how fast your CPU is e.g., if you can write to disk only at 50MB/s then it doesn't matter that your CPU can process 1GB/s. Also your program can consume other finite resources such as number of open files. If you use gevent make sure that all blocking calls are "green" i.e., they don't block e.g., your database driver might not be compatible with gevent.
    – jfs
    Feb 12, 2012 at 21:02
  • 1
    @Mridang Agarwalla: 1. Connection pooling and a single db connection being green are separate issues but you can get both with a single tool if you need it (I don't know what is preferred for django+gevent+postgres). 2. If a request calls a non-green blocking call then it blocks the whole interpreter and no other requests can be processed until it returns (in other words your app will be slow if the call takes a considerable time).
    – jfs
    Feb 13, 2012 at 18:00
  • 1
    @vumaasha: Do you mean how would I answer the question in the title? (my answer is more about the specific code and misconceptions in the question body). If I would try to answer the title: The only benchmark that matters is your code on your hardware. There were answers in favor on both sides over the years. Though the more concurrent connections you need, the more likely OS threads are not the answer. To get the most throughput: find a bottleneck, eliminate it, repeat. Sometimes it is enough to get a better cable, sometimes you need to redesign the whole project. The question is too broad.
    – jfs
    Sep 13, 2016 at 8:01
1

gevent is great when you have very many (green) threads. I tested it with thousands and it worked very well. you have make sure all libraries you use both for scraping and for saving to the db get green. afaik if they use python's socket, gevent injection ought to work. extensions written in C (e.g. mysqldb) would block however and you'd need to use green equivalents instead.

if you use gevent you could mostly do away with queues, spawn new (green) thread for every task, code for the thread being as simple as db.save(web.get(address)). gevent will take care of preemption when some library in db or web blocks. it will work as long as your tasks fit in memory.

0

In this case, your problem is not with program speed (i.e choice of gevent or threading), but network IO throughput. That's (should be) the bottleneck that determines how fast the program runs.

Gevent is one nice way to make sure that is the bottleneck, and not your program's architecture.

This is the sort of process you'd want:

import gevent
from gevent.queue import Queue, JoinableQueue
from gevent.monkey import patch_all


patch_all()  # Patch urllib2, etc


def worker(work_queue, output_queue):
    for work_unit in work_queue:
        finished = do_work(work_unit)
        output_queue.put(finished)
        work_queue.task_done()


def producer(input_queue, work_queue):
    for url in input_queue:
        url_list = crawl(url)
        for work in url_list:
            work_queue.put(work)
        input_queue.task_done()


def do_work(work):
    gevent.sleep(0)  # Actually proces link here
    return work


def crawl(url):
    gevent.sleep(0)
    return list(url)  # Actually process url here

input = JoinableQueue()
work = JoinableQueue()
output = Queue()

workers = [gevent.spawn(worker, work, output) for i in range(0, 10)]
producers = [gevent.spawn(producer, input, work) for i in range(0, 10)]


list_of_urls = ['foo', 'bar']

for url in list_of_urls:
    input.put(url)

# Wait for input to finish processing
input.join()
print 'finished producing'
# Wait for workers to finish processing work
work.join()
print 'finished working'

# We now have output!
print 'output:'
for message in output:
    print message
# Or if you'd like, you could use the output as it comes!

You don't need to wait for input and work queues to finish, I've just demonstrated that here.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

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