# `Queue` example - a concurrent web spider

Tornado’s `tornado.queues` module (and the very similar `Queue` classes in `asyncio`) implements an asynchronous producer / consumer pattern for coroutines, analogous to the pattern implemented for threads by the Python standard library’s `queue` module.

A coroutine that yields `Queue.get` pauses until there is an item in the queue. If the queue has a maximum size set, a coroutine that yields `Queue.put` pauses until there is room for another item.

A `Queue` maintains a count of unfinished tasks, which begins at zero. `put` increments the count; `task_done` decrements it.

In the web-spider example here, the queue begins containing only base_url. When a worker fetches a page it parses the links and puts new ones in the queue, then calls `task_done` to decrement the counter once. Eventually, a worker fetches a page whose URLs have all been seen before, and there is also no work left in the queue. Thus that worker’s call to `task_done` decrements the counter to zero. The main coroutine, which is waiting for `join`, is unpaused and finishes.

    #!/usr/bin/env python3

    import asyncio
    import time
    from datetime import timedelta

    from html.parser import HTMLParser
    from urllib.parse import urljoin, urldefrag

    from tornado import gen, httpclient, queues

    base_url = "http://www.tornadoweb.org/en/stable/"
    concurrency = 10

    async def get_links_from_url(url):
        """Download the page at `url` and parse it for links.

        Returned links have had the fragment after `#` removed, and have been made
        absolute so, e.g. the URL 'gen.html#tornado.gen.coroutine' becomes
        'http://www.tornadoweb.org/en/stable/gen.html'.
        """
        response = await httpclient.AsyncHTTPClient().fetch(url)
        print("fetched %s" % url)

        html = response.body.decode(errors="ignore")
        return [urljoin(url, remove_fragment(new_url)) for new_url in get_links(html)]

    def remove_fragment(url):
        pure_url, frag = urldefrag(url)
        return pure_url

    def get_links(html):
        class URLSeeker(HTMLParser):
            def __init__(self):
                HTMLParser.__init__(self)
                self.urls = []

            def handle_starttag(self, tag, attrs):
                href = dict(attrs).get("href")
                if href and tag == "a":
                    self.urls.append(href)

        url_seeker = URLSeeker()
        url_seeker.feed(html)
        return url_seeker.urls

    async def main():
        q = queues.Queue()
        start = time.time()
        fetching, fetched, dead = set(), set(), set()

        async def fetch_url(current_url):
            if current_url in fetching:
                return

            print("fetching %s" % current_url)
            fetching.add(current_url)
            urls = await get_links_from_url(current_url)
            fetched.add(current_url)

            for new_url in urls:
                # Only follow links beneath the base URL
                if new_url.startswith(base_url):
                    await q.put(new_url)

        async def worker():
            async for url in q:
                if url is None:
                    return
                try:
                    await fetch_url(url)
                except Exception as e:
                    print(f"Exception: {e} {url}")
                    dead.add(url)
                finally:
                    q.task_done()

        await q.put(base_url)

        # Start workers, then wait for the work queue to be empty.
        workers = gen.multi([worker() for _ in range(concurrency)])
        await q.join(timeout=timedelta(seconds=300))
        assert fetching == (fetched | dead)
        print("Done in %d seconds, fetched %s URLs." % (time.time() - start, len(fetched)))
        print("Unable to fetch %s URLs." % len(dead))

        # Signal all the workers to exit.
        for _ in range(concurrency):
            await q.put(None)
        await workers

    if __name__ == "__main__":
        asyncio.run(main())