388

I am opening a file which has 100,000 URL's. I need to send an HTTP request to each URL and print the status code. I am using Python 2.6, and so far looked at the many confusing ways Python implements threading/concurrency. I have even looked at the python concurrence library, but cannot figure out how to write this program correctly. Has anyone come across a similar problem? I guess generally I need to know how to perform thousands of tasks in Python as fast as possible - I suppose that means 'concurrently'.

8
  • 62
    Make sure that you only do HEAD request (so that you don't download the whole document). See: stackoverflow.com/questions/107405/… Commented Apr 13, 2010 at 19:34
  • 9
    Excellent point, Kalmi. If all Igor wants is the status of the request, these 100K requests will go much, much, much quicker. Much quicker. Commented Apr 13, 2010 at 19:41
  • 3
    You don't need threads for this; the most efficient way is likely to use an asynchronous library like Twisted.
    – jemfinch
    Commented Apr 13, 2010 at 20:16
  • 6
    here are gevent, twisted, and asyncio -based code examples (tested on 1000000 requests)
    – jfs
    Commented Oct 7, 2014 at 13:21
  • 6
    @TarnayKálmán its possible for requests.get and requests.head (i.e. a page request vs a head request) to return different status codes, so this is not the best advice
    – Alex
    Commented Mar 10, 2017 at 0:47

21 Answers 21

232

Twistedless solution:

from urlparse import urlparse
from threading import Thread
import httplib, sys
from Queue import Queue

concurrent = 200

def doWork():
    while True:
        url = q.get()
        status, url = getStatus(url)
        doSomethingWithResult(status, url)
        q.task_done()

def getStatus(ourl):
    try:
        url = urlparse(ourl)
        conn = httplib.HTTPConnection(url.netloc)   
        conn.request("HEAD", url.path)
        res = conn.getresponse()
        return res.status, ourl
    except:
        return "error", ourl

def doSomethingWithResult(status, url):
    print status, url

q = Queue(concurrent * 2)
for i in range(concurrent):
    t = Thread(target=doWork)
    t.daemon = True
    t.start()
try:
    for url in open('urllist.txt'):
        q.put(url.strip())
    q.join()
except KeyboardInterrupt:
    sys.exit(1)

This one is slighty faster than the twisted solution and uses less CPU.

19
  • 17
    @Kalmi, why do you set Queue to concurrent*2 ? Commented Nov 6, 2013 at 19:45
  • 13
    Don't forget to close the connection conn.close(). Opening too many http connections may halt your script at some point and eats memory.
    – Aamir Rind
    Commented Jan 8, 2014 at 16:20
  • 8
    @hyh, the Queue module has been renamed to queue in Python 3. This is Python 2 code. Commented Dec 29, 2014 at 0:29
  • 5
    How much faster can you go if you want to talk with the SAME server each time, by persisting the connection? Can this even be done across threads, or with one persistent connection per thread?
    – mdurant
    Commented Jan 12, 2015 at 15:25
  • 3
    @mptevsion, if you are using CPython, you could (for example) just replace "print status, url" with "my_global_list.append((status, url))". (Most operations on) lists are implicitly thread-safe in CPython (and some other python implementations) due to the GIL, so this is safe to do. Commented Mar 5, 2016 at 15:41
99

Things have changed quite a bit since 2010 when this was posted and I haven't tried all the other answers but I have tried a few, and I found this to work the best for me using python3.6.

I was able to fetch about ~150 unique domains per second running on AWS.

import concurrent.futures
import requests
import time

out = []
CONNECTIONS = 100
TIMEOUT = 5

tlds = open('../data/sample_1k.txt').read().splitlines()
urls = ['http://{}'.format(x) for x in tlds[1:]]

def load_url(url, timeout):
    ans = requests.head(url, timeout=timeout)
    return ans.status_code

with concurrent.futures.ThreadPoolExecutor(max_workers=CONNECTIONS) as executor:
    future_to_url = (executor.submit(load_url, url, TIMEOUT) for url in urls)
    time1 = time.time()
    for future in concurrent.futures.as_completed(future_to_url):
        try:
            data = future.result()
        except Exception as exc:
            data = str(type(exc))
        finally:
            out.append(data)

            print(str(len(out)),end="\r")

    time2 = time.time()

print(f'Took {time2-time1:.2f} s')
7
  • 3
    I'm only asking because I don't know but could this futures stuff be replaced with async/await? Commented Oct 9, 2017 at 21:16
  • 3
    It could, but I have found the above to work better. you could use aiohttp but its not part of the standard lib and is changing quite a lot. It does work but I just havn't found it to work as well. I get higher error rates when I use it and for the life of me I can't get it to work as well as concurrent futures although in theory It seems that it should work better, see: stackoverflow.com/questions/45800857/… if you get it to work well please post your answer so I can test it. Commented Oct 9, 2017 at 21:52
  • 2
    This is a nitpick, but I think it's a lot cleaner to put time1 = time.time() at the top of the for loop and time2 = time.time() right after the for loop. Commented May 12, 2018 at 17:51
  • I tested your snippet, somehow it executes twice. Am I doing something wrong? Or is it meant to run twice? If its the latter case, can you also help me to understand how does it trigger twice?
    – Ronnie
    Commented Mar 17, 2020 at 3:17
  • 2
    It shouldn't run twice. Not sure why you are seeing that. Commented Mar 18, 2020 at 15:13
71

I know this is an old question, but in Python 3.7 you can do this using asyncio and aiohttp.

import asyncio
import aiohttp
from aiohttp import ClientSession, ClientConnectorError

async def fetch_html(url: str, session: ClientSession, **kwargs) -> tuple:
    try:
        resp = await session.request(method="GET", url=url, **kwargs)
    except ClientConnectorError:
        return (url, 404)
    return (url, resp.status)

async def make_requests(urls: set, **kwargs) -> None:
    async with ClientSession() as session:
        tasks = []
        for url in urls:
            tasks.append(
                fetch_html(url=url, session=session, **kwargs)
            )
        results = await asyncio.gather(*tasks)

    for result in results:
        print(f'{result[1]} - {str(result[0])}')

if __name__ == "__main__":
    import pathlib
    import sys

    assert sys.version_info >= (3, 7), "Script requires Python 3.7+."
    here = pathlib.Path(__file__).parent

    with open(here.joinpath("urls.txt")) as infile:
        urls = set(map(str.strip, infile))

    asyncio.run(make_requests(urls=urls))

You can read more about it and see an example here.

9
  • Is this similar to C# async/await and Kotlin Coroutines? Commented Aug 28, 2019 at 15:39
  • @IgorGanapolsky, yes, it's very similar to C# async/await. I'm not familiar with Kotlin Coroutines. Commented Aug 29, 2019 at 9:28
  • @sandyp, I am not sure if it works, but if you want to try you will have to use the UnixConnector for aiohttp. Read more here: docs.aiohttp.org/en/stable/client_reference.html#connectors. Commented Nov 14, 2019 at 7:56
  • Thanks @MariusStănescu. That is exactly what I used.
    – sandyp
    Commented Nov 15, 2019 at 22:48
  • +1 for showing asyncio.gather(*tasks). here is one such snippet that i used: urls= [fetch(construct_fetch_url(u),idx) for idx, u in enumerate(some_URI_list)] results = await asyncio.gather(*urls) Commented Nov 18, 2019 at 12:24
61

A solution using tornado asynchronous networking library

from tornado import ioloop, httpclient

i = 0

def handle_request(response):
    print(response.code)
    global i
    i -= 1
    if i == 0:
        ioloop.IOLoop.instance().stop()

http_client = httpclient.AsyncHTTPClient()
for url in open('urls.txt'):
    i += 1
    http_client.fetch(url.strip(), handle_request, method='HEAD')
ioloop.IOLoop.instance().start()

This code is using non-blocking network I/O and doesn't have any restriction. It can scale to tens of thousands of open connections. It will run in a single thread but will be a way faster then any threading solution. Checkout non-blocking I/O

9
  • 5
    It is a counter for determining when to exit the ``ioloop` -- so when you are done. Commented Jul 17, 2015 at 7:54
  • 3
    @mher I tested your code but I get only 599 response code. Don't you know where can be a problem? Thanks
    – Joozty
    Commented Nov 28, 2017 at 13:30
  • 3
    @mher - If I am not interested at all in the response, meaning only wish to send as many requests as possible as fast as possible towards the server, what (if any) should I modify in the example above ? Thanks !! Commented Dec 9, 2017 at 11:05
  • 24
    @Guy Avraham Good luck getting help on your ddos plan.
    – Walter
    Commented Jan 24, 2019 at 8:56
  • 4
    @Walter - you got me :) Actually I was trying to do some very naive "stress test" Commented Jan 24, 2019 at 14:20
47

Threads are absolutely not the answer here. They will provide both process and kernel bottlenecks, as well as throughput limits that are not acceptable if the overall goal is "the fastest way".

A little bit of twisted and its asynchronous HTTP client would give you much better results.

6
  • ironfroggy: I am leaning toward your sentiments. I tried implementing my solution with threads and queues (for automatic mutexes), but can you imagine how long it takes to populate a queue with 100,000 things?? I'm still playing around with different options and suggestions by everyone on this thread, and maybe Twisted will be a good solution. Commented Apr 13, 2010 at 20:30
  • 3
    You can avoid populating a queue with 100k things. Just process items one at a time from your input, then launch a thread to process the request corresponding to each item. (As I describe below, use a launcher thread to start the HTTP request threads when your thread count is below some threshold. Make the threads write the results out into a dict mapping URL to response, or append tuples to a list.) Commented Apr 13, 2010 at 20:59
  • ironfroggy: Also, I'm curious about what bottlenecks you've found using Python threads? And how do Python threads interact with the OS kernel? Commented Apr 13, 2010 at 21:01
  • Make sure you install the epoll reactor; otherwise you'll be using select/poll, and it will be very slow. Also, if you're going to actually try to have 100,000 connections open simultaneously (assuming your program is written that way, and the URLs are on different servers), you'll need to tune your OS so that you won't run out of file descriptors, ephemeral ports, etc. (it's probably easier to just make sure that you don't have more than, say, 10,000 outstanding connections at once). Commented Apr 15, 2010 at 2:59
  • erikg: you did recommend a great idea. However, the best result I was able to achieve with 200 threads was approx. 6 minutes. I'm sure there are ways to accomplish this in lesser time... Mark N: if Twisted is the way I decide to go, then epoll reactor is surely useful. However, if my script will be run from multiple machines, wouldn't that necessitate the installation of Twisted on EACH machine? I don't know if I can convince my boss to go that route... Commented Apr 27, 2010 at 18:52
27

(Note to self for next project)

Python 3 solution using only requests. It's the simplest and it's fast, no need for multiprocessing or complicated asynchronous libraries.

The most important aspect is to reuse connections, especially for HTTPS (TLS requires an extra round trip to open). Note that a connection is specific to a subdomain. If you scrape many pages on many domains, you can sort the list of URLs to maximize connection reuse (it effectively sorts by domain).

It will be as fast as any asynchronous code, when given enough threads. (requests releases the python GIL when waiting for the response).

[Production grade code with some logging and error handling]

import logging
import requests
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

# source: https://stackoverflow.com/a/68583332/5994461

THREAD_POOL = 16

# This is how to create a reusable connection pool with python requests.
session = requests.Session()
session.mount(
    'https://',
    requests.adapters.HTTPAdapter(pool_maxsize=THREAD_POOL,
                                  max_retries=3,
                                  pool_block=True)
)

def get(url):
    response = session.get(url)
    logging.info("request was completed in %s seconds [%s]", response.elapsed.total_seconds(), response.url)
    if response.status_code != 200:
        logging.error("request failed, error code %s [%s]", response.status_code, response.url)
    if 500 <= response.status_code < 600:
        # server is overloaded? give it a break
        time.sleep(5)
    return response

def download(urls):
    with ThreadPoolExecutor(max_workers=THREAD_POOL) as executor:
        # wrap in a list() to wait for all requests to complete
        for response in list(executor.map(get, urls)):
            if response.status_code == 200:
                print(response.content)

def main():
    logging.basicConfig(
        format='%(asctime)s.%(msecs)03d %(levelname)-8s %(message)s',
        level=logging.INFO,
        datefmt='%Y-%m-%d %H:%M:%S'
    )

    urls = [
        "https://httpstat.us/200",
        "https://httpstat.us/200",
        "https://httpstat.us/200",
        "https://httpstat.us/404",
        "https://httpstat.us/503"
    ]

    download(urls)

if __name__ == "__main__":
    main()
5
  • What do you mean by ~"sort URLs"? Commented Jul 30, 2021 at 13:44
  • Sort the list of URLS sorted(urls) Commented Jul 30, 2021 at 19:17
  • if you start running into 400 errors "request/cookie too large"- work a session.cookies.clear() into your code.
    – grantr
    Commented Apr 29, 2022 at 19:35
  • "requests releases the python GIL when waiting for the response". How do you know that? Thank you.
    – LucG
    Commented Dec 28, 2022 at 16:09
  • How do you do something useful with the request? like where can you cast the response to json and read out values? Commented May 2, 2023 at 20:29
26

Use grequests , it's a combination of requests + Gevent module .

GRequests allows you to use Requests with Gevent to make asyncronous HTTP Requests easily.

Usage is simple:

import grequests

urls = [
   'http://www.heroku.com',
   'http://tablib.org',
   'http://httpbin.org',
   'http://python-requests.org',
   'http://kennethreitz.com'
]

Create a set of unsent Requests:

>>> rs = (grequests.get(u) for u in urls)

Send them all at the same time:

>>> grequests.map(rs)
[<Response [200]>, <Response [200]>, <Response [200]>, <Response [200]>, <Response [200]>]
3
11

A good approach to solving this problem is to first write the code required to get one result, then incorporate threading code to parallelize the application.

In a perfect world this would simply mean simultaneously starting 100,000 threads which output their results into a dictionary or list for later processing, but in practice you are limited in how many parallel HTTP requests you can issue in this fashion. Locally, you have limits in how many sockets you can open concurrently, how many threads of execution your Python interpreter will allow. Remotely, you may be limited in the number of simultaneous connections if all the requests are against one server, or many. These limitations will probably necessitate that you write the script in such a way as to only poll a small fraction of the URLs at any one time (100, as another poster mentioned, is probably a decent thread pool size, although you may find that you can successfully deploy many more).

You can follow this design pattern to resolve the above issue:

  1. Start a thread which launches new request threads until the number of currently running threads (you can track them via threading.active_count() or by pushing the thread objects into a data structure) is >= your maximum number of simultaneous requests (say 100), then sleeps for a short timeout. This thread should terminate when there is are no more URLs to process. Thus, the thread will keep waking up, launching new threads, and sleeping until your are finished.
  2. Have the request threads store their results in some data structure for later retrieval and output. If the structure you are storing the results in is a list or dict in CPython, you can safely append or insert unique items from your threads without locks, but if you write to a file or require in more complex cross-thread data interaction you should use a mutual exclusion lock to protect this state from corruption.

I would suggest you use the threading module. You can use it to launch and track running threads. Python's threading support is bare, but the description of your problem suggests that it is completely sufficient for your needs.

Finally, if you'd like to see a pretty straightforward application of a parallel network application written in Python, check out ssh.py. It's a small library which uses Python threading to parallelize many SSH connections. The design is close enough to your requirements that you may find it to be a good resource.

7
  • 2
    erikg: would throwing in a queue into your equation be reasonable (for mutual-exclusion locking)? I suspect that Python's GIL isn't geared toward playing with thousands of threads. Commented Apr 13, 2010 at 20:33
  • Why do you need mutual-exclusion locking to prevent the generation of too many threads? I suspect I misunderstand the term. You can track running threads in a thread queue, removing them when they complete and adding more up to said thread limit. But in a simple case such as the one in question you can also just watch the number of active threads in the current Python process, wait until it falls below a threshold, and launch more threads up to the threshold as described. I guess you could consider this an implicit lock, but no explicit locks are required afaik. Commented Apr 13, 2010 at 20:54
  • erikg: don't multiple threads share state? On page 305 in O'Reilly's book "Python for Unix and Linux System Administration" it states: "... using threading without queues makes it more complex than many people can realistically handle. It is a much better idea to always use the queuing module if you find you need to use threads. Why? Because the queue module also alleviates the need to explicitly protect data with mutexes because the queue itself is already protected internally by a mutex." Again, I welcome your point of view on this. Commented Apr 13, 2010 at 22:22
  • Igor: You are absolutely right that you should use a lock. I've edited the post to reflect this. That said, practical experience with python suggests that you don't need to lock data structures which you modify atomically from your threads, such as by list.append or by the addition of a hash key. The reason, I believe, is the GIL, which provides operations such as list.append with a degree of atomicity. I am currently running a test to verify this (use 10k threads to append numbers 0-9999 to a list, check that all appends worked). After nearly 100 iterations the test has not failed. Commented Apr 29, 2010 at 17:29
  • Igor: I'm asked another question on this topic: stackoverflow.com/questions/2740435/… Commented Apr 29, 2010 at 20:48
9

If you're looking to get the best performance possible, you might want to consider using Asynchronous I/O rather than threads. The overhead associated with thousands of OS threads is non-trivial and the context switching within the Python interpreter adds even more on top of it. Threading will certainly get the job done but I suspect that an asynchronous route will provide better overall performance.

Specifically, I'd suggest the async web client in the Twisted library (http://www.twistedmatrix.com). It has an admittedly steep learning curve but it quite easy to use once you get a good handle on Twisted's style of asynchronous programming.

A HowTo on Twisted's asynchronous web client API is available at:

http://twistedmatrix.com/documents/current/web/howto/client.html

2
  • Rakis: I am currently looking into asynchronous and non-blocking I/O. I need to learn it better before I implement it. One comment I'd like to make on your post is that it is impossible (at least under my Linux distribution) to spawn "thousands of OS threads". There is a maximum number of threads that Python will allow you to spawn before the program breaks. And in my case (on CentOS 5) maximum number of threads is 303. Commented Apr 30, 2010 at 18:10
  • That's good to know. I've never tried spawning more than a handful in Python at once but I would have expected to be able to create more than that before it bombed.
    – Rakis
    Commented May 3, 2010 at 12:38
7

A solution:

from twisted.internet import reactor, threads
from urlparse import urlparse
import httplib
import itertools


concurrent = 200
finished=itertools.count(1)
reactor.suggestThreadPoolSize(concurrent)

def getStatus(ourl):
    url = urlparse(ourl)
    conn = httplib.HTTPConnection(url.netloc)   
    conn.request("HEAD", url.path)
    res = conn.getresponse()
    return res.status

def processResponse(response,url):
    print response, url
    processedOne()

def processError(error,url):
    print "error", url#, error
    processedOne()

def processedOne():
    if finished.next()==added:
        reactor.stop()

def addTask(url):
    req = threads.deferToThread(getStatus, url)
    req.addCallback(processResponse, url)
    req.addErrback(processError, url)   

added=0
for url in open('urllist.txt'):
    added+=1
    addTask(url.strip())

try:
    reactor.run()
except KeyboardInterrupt:
    reactor.stop()

Testtime:

[kalmi@ubi1:~] wc -l urllist.txt
10000 urllist.txt
[kalmi@ubi1:~] time python f.py > /dev/null 

real    1m10.682s
user    0m16.020s
sys 0m10.330s
[kalmi@ubi1:~] head -n 6 urllist.txt
http://www.google.com
http://www.bix.hu
http://www.godaddy.com
http://www.google.com
http://www.bix.hu
http://www.godaddy.com
[kalmi@ubi1:~] python f.py | head -n 6
200 http://www.bix.hu
200 http://www.bix.hu
200 http://www.bix.hu
200 http://www.bix.hu
200 http://www.bix.hu
200 http://www.bix.hu

Pingtime:

bix.hu is ~10 ms away from me
godaddy.com: ~170 ms
google.com: ~30 ms
1
  • 6
    Using Twisted as a threadpool is ignoring most of the benefits you can get from it. You should be using the async HTTP client instead. Commented Apr 14, 2010 at 12:39
4
pip install requests-threads

Example Usage using async/await — send 100 concurrent requests

from requests_threads import AsyncSession

session = AsyncSession(n=100)

async def _main():
    rs = []
    for _ in range(100):
        rs.append(await session.get('http://httpbin.org/get'))
    print(rs)

if __name__ == '__main__':
    session.run(_main)

This example works on Python 3 only. You can also provide your own asyncio event loop!

Example Usage using Twisted

from twisted.internet.defer import inlineCallbacks
from twisted.internet.task import react
from requests_threads import AsyncSession

session = AsyncSession(n=100)

@inlineCallbacks
def main(reactor):
    responses = []
    for i in range(100):
        responses.append(session.get('http://httpbin.org/get'))

    for response in responses:
        r = yield response
        print(r)

if __name__ == '__main__':
    react(main)

This example works on both Python 2 and Python 3.

Maybe it can be helpful my repo, one basic example, WRITING FAST ASYNC HTTP REQUESTS IN PYTHON

3

Here's an "async" solution that doesn't use asyncio, but the lower-level mechanism asyncio uses (on Linux): select(). (Or maybe asyncio uses poll, or epoll, but it's a similar principle.)

It's a slightly modified version of the example from PyCurl.

(For simplicity it requests the same URL multiple times, but you can easily modify it to retrieve a bunch of distinct URLs.)

(Another slight modification can make this retrieve the same URL over and over as an infinite loop. Hint: change while urls and handles to while handles, and change while nprocessed<nurls to while 1.)

import pycurl,io,gzip,signal, time, random
signal.signal(signal.SIGPIPE, signal.SIG_IGN)  # NOTE! We should ignore SIGPIPE when using pycurl.NOSIGNAL - see the libcurl tutorial for more info

NCONNS = 2  # Number of concurrent GET requests
url    = 'example.com'
urls   = [url for i in range(0x7*NCONNS)]  # Copy the same URL over and over

# Check args
nurls  = len(urls)
NCONNS = min(NCONNS, nurls)
print("\x1b[32m%s \x1b[0m(compiled against 0x%x)" % (pycurl.version, pycurl.COMPILE_LIBCURL_VERSION_NUM))
print(f'\x1b[37m{nurls} \x1b[91m@ \x1b[92m{NCONNS}\x1b[0m')

# Pre-allocate a list of curl objects
m         = pycurl.CurlMulti()
m.handles = []
for i in range(NCONNS):
  c = pycurl.Curl()
  c.setopt(pycurl.FOLLOWLOCATION,  1)
  c.setopt(pycurl.MAXREDIRS,       5)
  c.setopt(pycurl.CONNECTTIMEOUT,  30)
  c.setopt(pycurl.TIMEOUT,         300)
  c.setopt(pycurl.NOSIGNAL,        1)
  m.handles.append(c)

handles    = m.handles  # MUST make a copy?!
nprocessed = 0
while nprocessed<nurls:

  while urls and handles:  # If there is an url to process and a free curl object, add to multi stack
    url   = urls.pop(0)
    c     = handles.pop()
    c.buf = io.BytesIO()
    c.url = url  # store some info
    c.t0  = time.perf_counter()
    c.setopt(pycurl.URL,        c.url)
    c.setopt(pycurl.WRITEDATA,  c.buf)
    c.setopt(pycurl.HTTPHEADER, [f'user-agent: {random.randint(0,(1<<256)-1):x}', 'accept-encoding: gzip, deflate', 'connection: keep-alive', 'keep-alive: timeout=10, max=1000'])
    m.add_handle(c)

  while 1:  # Run the internal curl state machine for the multi stack
    ret, num_handles = m.perform()
    if ret!=pycurl.E_CALL_MULTI_PERFORM:  break

  while 1:  # Check for curl objects which have terminated, and add them to the handles
    nq, ok_list, ko_list = m.info_read()
    for c in ok_list:
      m.remove_handle(c)
      t1 = time.perf_counter()
      reply = gzip.decompress(c.buf.getvalue())
      print(f'\x1b[33mGET  \x1b[32m{t1-c.t0:.3f}  \x1b[37m{len(reply):9,}  \x1b[0m{reply[:32]}...')  # \x1b[35m{psutil.Process(os.getpid()).memory_info().rss:,} \x1b[0mbytes')
      handles.append(c)
    for c, errno, errmsg in ko_list:
      m.remove_handle(c)
      print('\x1b[31mFAIL {c.url} {errno} {errmsg}')
      handles.append(c)
    nprocessed = nprocessed + len(ok_list) + len(ko_list)
    if nq==0: break

  m.select(1.0)  # Currently no more I/O is pending, could do something in the meantime (display a progress bar, etc.). We just call select() to sleep until some more data is available.

for c in m.handles:
  c.close()
m.close()
2
  • Why is asyncio slower than select? Commented Dec 27, 2021 at 15:15
  • I don't think I said it's slower, but, now that you mention it, this is Python after all, so asyncio could have overhead when compared to raw calls to select (which is why I'm favorable to the idea of not using asyncio but using select directly if it's easy or if you have the time or if you're having fun). Commented Dec 28, 2021 at 7:46
2

Using a thread pool is a good option, and will make this fairly easy. Unfortunately, python doesn't have a standard library that makes thread pools ultra easy. But here is a decent library that should get you started: http://www.chrisarndt.de/projects/threadpool/

Code example from their site:

pool = ThreadPool(poolsize)
requests = makeRequests(some_callable, list_of_args, callback)
[pool.putRequest(req) for req in requests]
pool.wait()

Hope this helps.

3
  • I suggest that you specify q_size for ThreadPool like this: ThreadPool(poolsize, q_size=1000) So that you won't have 100000 WorkRequest objects in memory. "If q_size>0 the size of the work request queue is limited and the thread pool blocks when the queue is full and it tries to put more work requests in it (see putRequest method), unless you also use a positive timeout value for putRequest." Commented Apr 13, 2010 at 19:49
  • So far I'm trying to implement the threadpool solution - as suggested. However, I don't understand the parameter list in the makeRequests function. What is some_callable, list_of_args, callback? Perhaps if I saw a real code snippet that would help. I'm surprised that the author of that library didn't post ANY examples. Commented Apr 13, 2010 at 20:12
  • some_callable is your function that all your work is done in (connecting to the http server). list_of_args is arguments that will be passed into some_callabe. callback is a function that will be called when the worker thread is done. It takes two arguments, the worker object (don't need to concern your self with this really), and the results that the worker retrieved. Commented Apr 13, 2010 at 21:36
2

For your case, threading will probably do the trick as you'll probably be spending most time waiting for a response. There are helpful modules like Queue in the standard library that might help.

I did a similar thing with parallel downloading of files before and it was good enough for me, but it wasn't on the scale you are talking about.

If your task was more CPU-bound, you might want to look at the multiprocessing module, which will allow you to utilize more CPUs/cores/threads (more processes that won't block each other since the locking is per process)

1
  • The only thing I'd like to mention is that spawning multiple processes may be more expensive than spawning multiple threads. Also, there is no clear performance gain in sending out 100,000 HTTP requests with multiple processes vs. multiple threads. Commented Apr 30, 2010 at 18:16
2

This twisted async web client goes pretty fast.

#!/usr/bin/python2.7

from twisted.internet import reactor
from twisted.internet.defer import Deferred, DeferredList, DeferredLock
from twisted.internet.defer import inlineCallbacks
from twisted.web.client import Agent, HTTPConnectionPool
from twisted.web.http_headers import Headers
from pprint import pprint
from collections import defaultdict
from urlparse import urlparse
from random import randrange
import fileinput

pool = HTTPConnectionPool(reactor)
pool.maxPersistentPerHost = 16
agent = Agent(reactor, pool)
locks = defaultdict(DeferredLock)
codes = {}

def getLock(url, simultaneous = 1):
    return locks[urlparse(url).netloc, randrange(simultaneous)]

@inlineCallbacks
def getMapping(url):
    # Limit ourselves to 4 simultaneous connections per host
    # Tweak this number, but it should be no larger than pool.maxPersistentPerHost 
    lock = getLock(url,4)
    yield lock.acquire()
    try:
        resp = yield agent.request('HEAD', url)
        codes[url] = resp.code
    except Exception as e:
        codes[url] = str(e)
    finally:
        lock.release()


dl = DeferredList(getMapping(url.strip()) for url in fileinput.input())
dl.addCallback(lambda _: reactor.stop())

reactor.run()
pprint(codes)
2

Create epoll object,
open many client TCP sockets,
adjust their send buffers to be a bit more than request header,
send a request header — it should be immediate, just placing into a buffer, register socket in epoll object,
do .poll on epoll obect,
read first 3 bytes from each socket from .poll,
write them to sys.stdout followed by \n (don't flush), close the client socket.

Limit number of sockets opened simultaneously — handle errors when sockets are created. Create a new socket only if another is closed.
Adjust OS limits.
Try forking into a few (not many) processes: this may help to use CPU a bit more effectively.

1
  • @IgorGanapolsky Must be. I'd be surprised otherwise. But it certainly needs experimentation. Commented Jul 17, 2019 at 21:02
2

I found that using the tornado package to be the fastest and simplest way to achieve this:

from tornado import ioloop, httpclient, gen


def main(urls):
    """
    Asynchronously download the HTML contents of a list of URLs.
    :param urls: A list of URLs to download.
    :return: List of response objects, one for each URL.
    """

    @gen.coroutine
    def fetch_and_handle():
        httpclient.AsyncHTTPClient.configure(None, defaults=dict(user_agent='MyUserAgent'))
        http_client = httpclient.AsyncHTTPClient()
        waiter = gen.WaitIterator(*[http_client.fetch(url, raise_error=False, method='HEAD')
                                    for url in urls])
        results = []
        # Wait for the jobs to complete
        while not waiter.done():
            try:
                response = yield waiter.next()
            except httpclient.HTTPError as e:
                print(f'Non-200 HTTP response returned: {e}')
                continue
            except Exception as e:
                print(f'An unexpected error occurred querying: {e}')
                continue
            else:
                print(f'URL \'{response.request.url}\' has status code <{response.code}>')
                results.append(response)
        return results

    loop = ioloop.IOLoop.current()
    web_pages = loop.run_sync(fetch_and_handle)

    return web_pages

my_urls = ['url1.com', 'url2.com', 'url100000.com']
responses = main(my_urls)
print(responses[0])
1
  • 1
    My god finally thanks alot, I was stuck using shitty concurrent futures, and for some reason in my env it keeps getting stuck in a infinite loop on some url calls and I'm 100% sure I'm using it correctly. This solution with tornado is top notch
    – Mika C.
    Commented Dec 30, 2021 at 0:47
2

Scrapy framework will solve your problem fast and professionally. It will also cache all the requests, so that you can rerun the failed ones only later on.

Save this script as quotes_spider.py.

# quote_spiders.py
import json
import string
import scrapy
from scrapy.crawler import CrawlerProcess
from scrapy.item import Item, Field

class TextCleaningPipeline(object):
    def _clean_text(self, text):
        text = text.replace('“', '').replace('”', '')
        table = str.maketrans({key: None for key in string.punctuation})
        clean_text = text.translate(table)
        return clean_text.lower()

    def process_item(self, item, spider):
        item['text'] = self._clean_text(item['text'])
        return item

class JsonWriterPipeline(object):
    def open_spider(self, spider):
        self.file = open(spider.settings['JSON_FILE'], 'a')

    def close_spider(self, spider):
        self.file.close()

    def process_item(self, item, spider):
        line = json.dumps(dict(item)) + "\n"
        self.file.write(line)
        return item

class QuoteItem(Item):
    text = Field()
    author = Field()
    tags = Field()
    spider = Field()

class QuoteSpider(scrapy.Spider):
    name = "quotes"

    def start_requests(self):
        urls = [
            'http://quotes.toscrape.com/page/1/',
            'http://quotes.toscrape.com/page/2/',
            # ...
        ]
        for url in urls:
            yield scrapy.Request(url=url, callback=self.parse)

    def parse(self, response):
        for quote in response.css('div.quote'):
            item = QuoteItem()
            item['text'] = quote.css('span.text::text').get()
            item['author'] = quote.css('small.author::text').get()
            item['tags'] = quote.css('div.tags a.tag::text').getall()
            item['spider'] = self.name
            yield item

if __name__ == '__main__':
    settings = dict()
    settings['USER_AGENT'] = 'Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1)'
    settings['HTTPCACHE_ENABLED'] = True
    settings['CONCURRENT_REQUESTS'] = 20
    settings['CONCURRENT_REQUESTS_PER_DOMAIN'] = 20
    settings['JSON_FILE'] = 'items.jl'
    settings['ITEM_PIPELINES'] = dict()
    settings['ITEM_PIPELINES']['__main__.TextCleaningPipeline'] = 800
    settings['ITEM_PIPELINES']['__main__.JsonWriterPipeline'] = 801

    process = CrawlerProcess(settings=settings)
    process.crawl(QuoteSpider)
    process.start()

followed by

$ pip install Scrapy
$ python quote_spiders.py 

To fine tune the scraper adjust the CONCURRENT_REQUESTS and CONCURRENT_REQUESTS_PER_DOMAIN settings accordingly.

2
  • What makes Scrapy faster? Commented Jun 15, 2022 at 15:11
  • 1
    Scrapy is based on battle tested twisted framework, so can be as fast as asycio. Also has great docs and vibrant community around it.
    – Levon
    Commented Jun 16, 2022 at 11:05
0

Consider using Windmill , although Windmill probably cant do that many threads.

You could do it with a hand rolled Python script on 5 machines, each one connecting outbound using ports 40000-60000, opening 100,000 port connections.

Also, it might help to do a sample test with a nicely threaded QA app such as OpenSTA in order to get an idea of how much each server can handle.

Also, try looking into just using simple Perl with the LWP::ConnCache class. You'll probably get more performance (more connections) that way.

0

[Tool]

Apache Bench is all you need. - A command line computer program (CLI) for measuring the performance of HTTP web servers

A nice blog post for you: https://www.petefreitag.com/item/689.cfm (from Pete Freitag)

1
  • The OP was not about measuring one server. It was to send many requests concurrently to many servers, to collect the responses. Kind of like web crawling. Commented Feb 18, 2021 at 15:56
-1

The easiest way would be to use Python's built-in threading library. They're not "real" / kernel threads They have issues (like serialization), but are good enough. You'd want a queue & thread pool. One option is here, but it's trivial to write your own. You can't parallelize all 100,000 calls, but you can fire off 100 (or so) of them at the same time.

5
  • 7
    Python's threads are quite real, as opposed to Ruby's for instance. Under the hood they are implemented as native OS threads, at least on Unix/Linux and Windows. Maybe you're referring to the GIL, but it doesn't make the threads less real... Commented Apr 13, 2010 at 19:33
  • 2
    Eli is right about Python's threads, but Pestilence's point that you'd want to use a thread pool is correct, too. The last thing that you'd want to do in this case is try to start a separate thread for each of the 100K requests simultaneously. Commented Apr 13, 2010 at 19:37
  • 1
    Igor, you can't sensibly post code snippets in comments, but you can edit your question and add them there. Commented Apr 13, 2010 at 19:54
  • Pestilence: how many queues and threads-per-queue would you recommend for my solution? Commented Apr 14, 2010 at 17:42
  • plus this is an I/O bound task not CPU bound, the GIL largely affects CPU bound tasks
    – PirateApp
    Commented Apr 7, 2018 at 12:51

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