Tornado: getting many HTTP 599 errors for valid urls

[YS-]

I'm using tornado AsyncHTTPClient with the following code, I basically call the scrape function with a url generator list that contains 10K urls.
I expect to have maximum 50 concurrent requests at any time, which doesn't seem to work as the entire process ends in about 2 minutes.

I got ~200 valid responses and ~9000 HTTP 599 error. I checked many urls that threw this error and they do load in less than 10 sec', I'm able to reach most urls using urllib2/requests with a smaller timeout (5 seconds).

All requests sent to different servers, running from ubuntu with python 2.7.3 & tornado version = "4.1".

I suspect that something is wrong as I can fetch most urls using other (blocking) libraries.

import tornado.ioloop
import tornado.httpclient

class Fetcher(object):
    def __init__(self, ioloop):
        self.ioloop = ioloop
        self.client = tornado.httpclient.AsyncHTTPClient(io_loop=ioloop, max_clients=50)
        self.client.configure(None, defaults=dict(user_agent="Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/41.0.2272.101 Safari/537.36",
                                                  connect_timeout=20,request_timeout=20, validate_cert=False))

    def fetch(self, url):
        self.client.fetch(url, self.handle_response)

    @property
    def active(self):
        """True if there are active fetching happening"""
        return len(self.client.active) != 0

    def handle_response(self, response):
        if response.error:
            print "Error: %s, time: %s, url: %s" % (response.error, response.time_info, response.effective_url)
        else:
           # print "clients %s" % self.client.active
            print "Got %d bytes" % (len(response.body))

        if not self.active:
            self.ioloop.stop()

def scrape(urls):
    ioloop = tornado.ioloop.IOLoop.instance()
    ioloop.add_callback(scrapeEverything, *urls)
    ioloop.start()

def scrapeEverything(*urls):
    fetcher = Fetcher(tornado.ioloop.IOLoop.instance())

    for url in urls:
        fetcher.fetch(url)

if __name__ == '__main__':
scrape()

[bdarnell]

You're starting all the fetches at once but telling AsyncHTTPClient to give up and return a 599 Timeout if it can't complete the request in 20 seconds (the request_timeout option). You need to either increase request_timeout to the amount of time you're willing to wait for the response (including time spent waiting in the queue), or maintain your own queue and feed urls into AsyncHTTPClient gradually (the queue and semaphore classes that are being introduced in the upcoming Tornado 4.2 can help here; until then you can use Toro: http://toro.readthedocs.org/en/stable/examples/web_spider_example.html)

[YS-]

the timeout setting is per request, so I assume it doesn't matter how many fetches I make, you basically say that the timeout is for all outgoing requests? how should one open many parallel connections without managing a queue inside a process that handles each response, if a timeout occurs it doesn't impact all other requests, I understand from the docs that it's per request.

[bdarnell]

The timeout is per request, but the clock starts as soon as you call fetch. When you do for url in urls: fetcher.fetch(url), that starts the clock for all of them at about the same time, even though many of them won't be able to start immediately. As I said before, I recommend maintaining your own queue. AsyncHTTPClient's queueing and timeout behavior is designed for interactive use as a part of a web server; for batch crawls like this you probably want different queuing behavior (and perhaps better prioritization, etc). If you don't want to manage your own queue, a quick fix is to increase request_timeout, but this sacrifices the ability to detect individual requests that are unresponsive once connected.

[akellehe]

Thanks, guys, this is a helpful discussion. I didn't realize time on the client's queue counted against the request_timeout. With that in mind I created a separate queue to manage a backlog of requests as you guys mentioned and problem solved :rocket: :rocket:

[akellehe]

I realize this might be useful for others experiencing the same problem. This solution works by only realizing those 599s when there is a timeout on the server/network not, for example, when the client becomes CPU bound. Here's an example:

import collections
import tornado.httpclient
​
​
class BacklogClient(object):
    MAX_CONCURRENT_REQUESTS = 20
​
    def __init__(self, ioloop):
        self.ioloop = ioloop
        self.client = tornado.httpclient.AsyncHTTPClient(max_clients=self.MAX_CONCURRENT_REQUESTS)
        self.client.configure(None, defaults=dict(connect_timeout=20, request_timeout=30))
        self.backlog = collections.deque()
        self.concurrent_requests = 0
​
    def __get_callback(self, function):
        def wrapped(*args, **kwargs):
            self.concurrent_requests -= 1
            return function(*args, **kwargs)
        return wrapped
​
    def fetch(self, request, callback=None):
        wrapped = self.__get_callback(callback)
​
        self.backlog.append((request, wrapped))
​
        while self.backlog and self.concurrent_requests < self.MAX_CONCURRENT_REQUESTS:
            request, callback = self.backlog.popleft()
            self.client.fetch(request, callback=callback)
            self.concurrent_requests += 1

[fidiego]

Experimented with and used the BacklogClient with success.
:+1:

[dovy]

@akellehe share to provide a working sample of using BacklogClient?

[akellehe]

@dovy i think this guy works, but it was untested when i posted it... you may need to implement a method to flush the final bit since i think the flush gets triggered when a message is added.

https://github.com/tornadoweb/tornado/issues/1400#issuecomment-204438178

[dovy]

@akellehe on Python slack? :P

[dalloriam]

@akellehe I just stumbled on your BacklogClient implementation (great idea by the way) and for the sake of completeness, I suggest daisy-chaining requests when the callback is executed instead of implementing a flush() method, as this guarantees the queue will be emptied. Something like this:

class BacklogClient(object):
  MAX_CONCURRENT_REQUESTS = 20

  def __init__(self, ioloop):
    self.ioloop = ioloop
    self.client = httpclient.AsyncHTTPClient(max_clients=self.MAX_CONCURRENT_REQUESTS)
    self.client.configure(None, defaults=dict(connect_timeout=20, request_timeout=30))
    self.backlog = collections.deque()
    self.concurrent_requests = 0

  def __get_callback(self, function):
    def wrapped(*args, **kwargs):
      self.concurrent_requests -= 1
      if self.backlog:
        request, callback = self.backlog.popleft()
        self.client.fetch(request, callback=callback)
        self.concurrent_requests += 1
      return function(*args, **kwargs)
    return wrapped

  def fetch(self, request, callback=None):
    wrapped = self.__get_callback(callback)

    self.backlog.append((request, wrapped))

    while self.backlog and self.concurrent_requests < self.MAX_CONCURRENT_REQUESTS:
      request, callback = self.backlog.popleft()
      self.client.fetch(request, callback=callback)
      self.concurrent_requests += 1

You could, of course, add an additional run_request() method to get rid of the duplication between the fetch() and __get_callback() methods.

[dovy]

@Dalloriam like the idea, though I implemented the flush version. Care to finish yours off to get the run_request() in there to remove duplication?

For those interested in the flush alternative, here's a full working example:

import collections
import tornado.httpclient

class BacklogClient(object):
    MAX_CONCURRENT_REQUESTS = 20

    def __init__(self, ioloop):
        self.ioloop = ioloop
        self.client = tornado.httpclient.AsyncHTTPClient(max_clients=self.MAX_CONCURRENT_REQUESTS)
        self.client.configure(None, defaults=dict(connect_timeout=20, request_timeout=30))
        self.backlog = collections.deque()
        self.concurrent_requests = 0
        self.callback = {}

    def __get_callback(self, function):
        def wrapped(*args, **kwargs):
            self.concurrent_requests -= 1
            return function(*args, **kwargs)

        return wrapped

    def flush(self):
        while self.backlog and self.concurrent_requests < self.MAX_CONCURRENT_REQUESTS:
            request, callback = self.backlog.popleft()
            self.client.fetch(request, callback=callback)
            self.concurrent_requests += 1

    def fetch(self, request, callback=None):
        wrapped = self.__get_callback(callback)
        self.backlog.append((request, wrapped))
        self.flush()



import time
from tornado import ioloop, httpclient


class TornadoBacklog:

    def __init__(self):

        self.queue = 0
        self.debug = 1
        self.toProcess = [
            'http://google.com',
            'http://yahoo.com'
        ]

    def handle_request(self, response):

        # DO YOUR STUFF HERE

        if self.backlog.backlog and self.backlog.concurrent_requests > 0:
            self.backlog.flush()
        else:
            ioloop.IOLoop.instance().stop()

            # END STUFF, SUM UP EVERYTHING, FINAL THINGS FOR YOU

    def launch(self):

        self.ioloop = ioloop.IOLoop.current()
        self.backlog = BacklogClient(self.ioloop)

        for item in self.toProcess:
            self.backlog.fetch(
                httpclient.HTTPRequest(
                    item,
                    method='GET',
                    headers=None,
                ),
                self.handle_request
            )

        self.ioloop.start()


def main():
    start_time = time.time()

    scraper = TornadoBacklog()
    scraper.launch()

    elapsed_time = time.time() - start_time
    print('Process took %f seconds processed %d items.' % (elapsed_time, len(scraper.toProcess)))


if __name__ == "__main__":
    main()

As @akellehe pointed out to me, using the flush method allows for async callbacks making sure you're never bound by a single response, but always filling the queue to process at the maximum capacity. Though I am very interested in the solution @Dalloriam has.

[dalloriam]

@dovy Here is my implementation without flush():

class BacklogClient(object):
  MAX_CONCURRENT_REQUESTS = 20

  def __init__(self, ioloop):
    self.ioloop = ioloop
    self.client = httpclient.AsyncHTTPClient(max_clients=self.MAX_CONCURRENT_REQUESTS)
    self.client.configure(None, defaults=dict(connect_timeout=20, request_timeout=30))
    self.backlog = collections.deque()
    self.concurrent_requests = 0

  def __get_callback(self, function):
    def wrapped(*args, **kwargs):
      self.concurrent_requests -= 1
      self.try_run_request()
      return function(*args, **kwargs)
    return wrapped

  def try_run_request(self):
    while self.backlog and self.concurrent_requests < self.MAX_CONCURRENT_REQUESTS:
      request, callback = self.backlog.popleft()
      self.client.fetch(request, callback=callback)
      self.concurrent_requests += 1

  def fetch(self, request, callback=None):
    wrapped = self.__get_callback(callback)

    self.backlog.append((request, wrapped))
    self.try_run_request()

One could adapt your example by replacing

if self.backlog.backlog:
  self.backlog.flush()
else:
  ioloop.IOLoop.instance().stop()

by

if not self.backlog.backlog:
  ioloop.IOLoop.instance().stop()

(altough I haven't tested it).
Cheers!

[dovy]

I just tested it and it works great. Here's @Dalloriam example, within my full working example for posterity. I've moved over to this variation because I like self-contained classes. ;)

import collections
import tornado.httpclient


class BacklogClient(object):
    MAX_CONCURRENT_REQUESTS = 20

    def __init__(self, ioloop):
        self.ioloop = ioloop
        self.client = tornado.httpclient.AsyncHTTPClient(max_clients=self.MAX_CONCURRENT_REQUESTS)
        self.client.configure(None, defaults=dict(connect_timeout=20, request_timeout=30))
        self.backlog = collections.deque()
        self.concurrent_requests = 0

    def __get_callback(self, function):
        def wrapped(*args, **kwargs):
            self.concurrent_requests -= 1
            self.try_run_request()
            return function(*args, **kwargs)

        return wrapped

    def try_run_request(self):
        while self.backlog and self.concurrent_requests < self.MAX_CONCURRENT_REQUESTS:
            request, callback = self.backlog.popleft()
            self.client.fetch(request, callback=callback)
            self.concurrent_requests += 1

    def fetch(self, request, callback=None):
        wrapped = self.__get_callback(callback)

        self.backlog.append((request, wrapped))
        self.try_run_request()


import time
from tornado import ioloop, httpclient


class TornadoBacklog:
    def __init__(self):

        self.queue = 0
        self.debug = 1
        self.toProcess = [
            'http://google.com',
            'http://yahoo.com'
        ]

    def handle_request(self, response):

        # DO YOUR STUFF HERE

        if not self.backlog.backlog and self.backlog.concurrent_requests == 0:
            ioloop.IOLoop.instance().stop()

            # END STUFF, SUM UP EVERYTHING, FINAL THINGS FOR YOU

    def launch(self):

        self.ioloop = ioloop.IOLoop.current()
        self.backlog = BacklogClient(self.ioloop)

        for item in self.toProcess:
            self.backlog.fetch(
                httpclient.HTTPRequest(
                    item,
                    method='GET',
                    headers=None,
                ),
                self.handle_request
            )

        self.ioloop.start()


def main():
    start_time = time.time()

    scraper = TornadoBacklog()
    scraper.launch()

    elapsed_time = time.time() - start_time
    print('Process took %f seconds processed %d items.' % (elapsed_time, len(scraper.toProcess)))


if __name__ == "__main__":
    main()

[dovy]

Updated examples with if not self.backlog.backlog and self.backlog.concurrent_requests == 0: to ensure the last thread has completed before stopping the ioloop.

[zt706]

I have the same error and I had solved the error. I used the BacklogClient and set the
AsyncHTTPClient.configure("tornado.curl_httpclient.CurlAsyncHTTPClient") when the ioloop start (http://www.tornadoweb.org/en/stable/httpclient.html)

[yuriiz]

HTTP 599 still happens even for single request to valid URL. Example:

#!/usr/bin/env python
from tornado import httpclient, ioloop


def chunk(chunk):
    print('got', len(chunk), 'bytes')


def done(response):
    print(response)

client = httpclient.AsyncHTTPClient()
client.fetch(httpclient.HTTPRequest(
    'http://ftp.freebsd.org/pub/FreeBSD/snapshots/amd64/amd64/ISO-IMAGES/11.0/FreeBSD-11.0-STABLE-amd64-20161117-r308738-disc1.iso.xz',  # nopep8
    streaming_callback=chunk
), done)

ioloop.IOLoop.current().start()

Output:

$ bin/python test.py 
HTTPResponse(_body=None,buffer=None,code=599,effective_url='http://ftp.freebsd.org/pub/FreeBSD/snapshots/amd64/amd64/ISO-IMAGES/11.0/FreeBSD-11.0-STABLE-amd64-20161117-r308738-disc1.iso.xz',error=HTTP 599: Connection closed,headers=<tornado.httputil.HTTPHeaders object at 0x7fcdf099e650>,reason='Unknown',request=<tornado.httpclient.HTTPRequest object at 0x7fcdf099e0d0>,request_time=0.7075650691986084,time_info={})

(tornado is latest from pypi)

[misssprite]

But there may be timeout exceptions though, even with max_concurrent under control, due to network failure, or so.
Where can I catch the exception and add the request back to the queue?

[kolombo]

There is some bug in simple_http_client that rejects connection on big file download.
Try to switch it to curl_http_client, it worked well for me.

httpclient.AsyncHTTPClient.configure("tornado.curl_httpclient.CurlAsyncHTTPClient")

[bdarnell]

@kolombo You probably need to increase max_body_size and use a streaming_callback to download large files with simple_httpclient.

[kolombo]

I used streaming_callback but it was not clear to me that I still need to change max_body_size as well.
I also expected that with using streaming_callback full response body is not saved in memory but delegated by read chunks so setting max_body_size is not necessary.

But according to the code comparing max_body_size to Content-length is done in HTTP1Connection before delegation to streaming_callback.

    if content_length > self._max_body_size:
        raise httputil.HTTPInputError("Content-Length too long")