grequests pool with multiple request.session?

Question

I want to make a lot of url requets to a REST webserivce. Typically between 75-90k. However, I need to throttle the number of concurrent connections to the webservice.

I started playing around with grequests in the following manner, but quickly started chewing up opened sockets.

concurrent_limit = 30
urllist = buildUrls()
hdrs = {'Host' : 'hostserver'}
g_requests = (grequests.get(url, headers=hdrs) for url in urls)
g_responses = grequests.map(g_requests, size=concurrent_limit)

As this runs for a minute or so, I get hit with 'maximum number of sockets reached' errors. As far as I can tell, each one of the requests.get calls in grequests uses it's own session which means a new socket is opened for each request.

I found a note on github referring how to make grequests use a single session. But this seems to effectively bottleneck all requests into a single shared pool. That seems to defeat the purpose of asynchronous http requests.

s = requests.session()
rs = [grequests.get(url, session=s) for url in urls]
grequests.map(rs)

Is is possible to use grequests or gevent.Pool in a way that creates a number of sessions?

Put another way: How can I make many concurrent http requests using either through queuing or connection pooling?

score 7 · Accepted Answer · answered Dec 12 '13 at 21:43

I ended up not using grequests to solve my problem. I'm still hopeful it might be possible.

I used threading:

class MyAwesomeThread(Thread):
    """
    Threading wrapper to handle counting and processing of tasks
    """
    def __init__(self, session, q):
        self.q = q
        self.count = 0
        self.session = session
        self.response = None
        Thread.__init__(self)

    def run(self): 
        """TASK RUN BY THREADING"""
        while True:
            url, host = self.q.get()
            httpHeaders = {'Host' : host}
            self.response = session.get(url, headers=httpHeaders)
            # handle response here
            self.count+= 1
            self.q.task_done()
        return

q=Queue()
threads = []
for i in range(CONCURRENT):
    session = requests.session()
    t=MyAwesomeThread(session,q)
    t.daemon=True # allows us to send an interrupt 
    threads.append(t)


## build urls and add them to the Queue
for url in buildurls():
    q.put_nowait((url,host))

## start the threads
for t in threads:
    t.start()

I think I have a similar problem: http://stackoverflow.com/questions/34593643/gevent-grequests-wont-spawn-more-greenlets how did the OS like 70 threads? — domoarigato, Jan 04 '16 at 20:28

walker · Answer 2 · 2013-11-28T06:52:58.763

2

rs is a AsyncRequest list。each AsyncRequest have it's own session.

rs = [grequests.get(url) for url in urls]
grequests.map(rs)
for ar in rs:
    print(ar.session.cookies)

edited Nov 28 '13 at 06:52

answered Nov 28 '13 at 06:34

walker

21
3

3

Yeah, the problem I found was that EVERY request had it's own session. This meant 70k sessions. The OS doesn't like that very much. I need to limit the number of sessions. – Marcel Wilson Dec 12 '13 at 21:26

score 2 · Answer 3 · edited Oct 28 '16 at 16:22

2

Something like this:

NUM_SESSIONS = 50
sessions = [requests.Session() for i in range(NUM_SESSIONS)]
reqs = []
i = 0
for url in urls:
    reqs.append(grequests.get(url, session=sessions[i % NUM_SESSIONS]
    i+=1
responses = grequests.map(reqs, size=NUM_SESSIONS*5)

That should spread the requests over 50 different sessions.

edited Oct 28 '16 at 16:22

shaktimaan

11,962
2
29
33

answered Sep 29 '15 at 09:04

Erik Forsberg

4,819
3
27
31

Why is the `.map` throttle size set to 5 times the number of sessions? Won't the number of concurrent connections essentially be bottlenecked by the number of sessions anyway? – Michael May 25 '17 at 08:12

grequests pool with multiple request.session?

3 Answers3