אחד התרגילים שאני אוהב לתת כשאני מלמד Multi Threaded Programming בשפת Python הוא פיתוח של Web Spider - או בעברית כלי שמקבל דף אינטרנט ומתחיל ממנו לחפש ולהוריד את כל דפי האתר. העכביש מתחיל בעמוד הראשון, מחפש שם לינקים לדפים נוספים, מוריד אותם ואז ממשיך לחפש בהם לינקים לדפים נוספים.

זה תרגיל טוב כי הוא ממחיש איך לסנכרן בין תהליכונים באמצעות תור: כל תהליכון שולף מהתור קישור, מוריד אותו ומחפש בתוצאה עוד קישורים אותם הוא מוסיף חזרה לתור. לאט לאט תהליכונים מוסיפים פריטים לתור ושולפים משימות והכל באותו הזמן עד שבסוף סיימנו למצוא ולהוריד את כל דפי האתר.

הנה דוגמא קצרה של פיתרון באמצעות Threads, ואחריה הסבר על הנקודות המרכזיות:

import threading
from threading import Thread, Lock
from queue import Queue, Empty
from urllib.parse import urlparse, urlunparse
from pathlib import Path
from bs4 import BeautifulSoup
import requests

base = 'https://www.python.org'
path = '/dev/peps/'
dest = Path('./pydoc')
found_links = set()
found_links_lock = Lock()
q = Queue()

tls = threading.local()

def run(q):
    tls.session = requests.Session()
    while True:
        href = q.get()
        if href is None:
            return

        with found_links_lock:
            if href in found_links:
                print(f"Found {href} that was already downloaded")
                q.task_done()
                continue

            found_links.update([href])

        discover(href)
        q.task_done()


def download(url, to):
    r = tls.session.get(url, allow_redirects=True)
    open(to, 'wb').write(r.content)


def parse(htmlfile):
    with open(htmlfile) as f:
        soup = BeautifulSoup(f.read(), 'lxml')
        new_links = set([
            a['href'] for a in soup.find_all('a')
            if a['href'].startswith('/dev/peps') and
               a['href'] not in found_links])

        for link in new_links:
            q.put(link)

        print(f"Parse done - added {len(new_links)} new tasks")


def discover(href):
    try:
        url = urlparse(href)
        srcpath = url.path

        if not srcpath.startswith(path):
            print("Path out of scope - ignore: ", href)
            return

        if srcpath.endswith('/'):
            destpath = Path(dest) / Path(srcpath[1:]) / 'index.html'
        else:
            destpath = Path(dest) / Path(srcpath[1:])

        destpath.parents[0].mkdir(parents=True, exist_ok=True)

        if url.netloc == '':
            href = base + href

        download(href, destpath)
        parse(destpath)

    except Exception as e:
        print("Failed to discover href ", href, e)
        raise e


def main():
    found_links.add('/dev/peps/')
    start = 'https://www.python.org/dev/peps/'
    q.put(start)
    workers = [Thread(target=run, args=(q,)) for _ in range(4)]
    for w in workers: w.start()
    q.join()
    for _ in range(4):
        q.put(None)

    for w in workers: w.join()


main()

מה הקוד עושה בגדול:

1. הפונקציה download מורידה קובץ מהרשת ושומרת אותו. בשביל שהפונקציה תעבוד יחסית מהר אני משתמש ב requests.Session. בגלל ש requests.Session לא עובד טוב בין תהליכונים לכל תהליכון יש Session משלו שנשמר באמצעות Thread Local Storage.

2. הפונקציה parse מקבלת קובץ html ומחפשת בו קישורים. כל פעם שמצאה קישור חדש היא מוסיפה אותו לתור.

3. הפונקציה המרכזית היא discover. פונקציה זו לוקחת קישור מהתור, מורידה אותו ומפעילה עליו את parse כדי לחפש בו קישורים.

4. הפונקציה run מנהלת את התהליכונים והסינכרון ביניהם. כל תהליכון יריץ את פונקציית run ויחכה לפריטים מהתור. הוא יעדכן את התור בחזרה כל פעם שסיים לטפל בפריט (באמצעות discover) וכך הקוד הראשי ידע מתי סיימנו לטפל בכל הפריטים.

5. בקוד הראשי main אני יוצר את התהליכונים ושולח אותם לדרכם. לאט לאט הם ימשכו קישורים מהתור ויוסיפו קישורים חדשים. הפקודה q.join מחכה עד שהתור יתרוקן וגם כל הקישורים שכעת בעבודה יסתיימו ורק אז חוזרת. שליחת None לכל התהליכונים מוציאה אותם מהלולאה כדי לסיים את התוכנית.