בפייתון הפונקציה asyncio.to_thread משתמשת ב Thread Pool הדיפולטי של המערכת כדי להפעיל קוד סינכרוני בצורה שנראית אסינכרונית. היא עשויה להיראות כמו פתרון קסם כשיש לנו קוד אסינכרוני שצריך להפעיל פונקציה סינכרונית ארוכה ואנחנו לא רוצים לתקוע את ה Main Loop.

לדוגמה נניח שכתבנו פונקציה ping_ip עם הקוד הבא:

def ping_ip(ip: str, count: int = 4) -> tuple[str, float]:
    """Ping an IP address `count` times and return (ip, success_percent)."""
    if sys.platform == "win32":
        cmd = ["ping", "-n", str(count), ip]
    else:
        cmd = ["ping", "-c", str(count), ip]

    result = subprocess.run(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        errors="replace",
    )
    output = result.stdout + result.stderr

    # Parse packet loss percentage from output like "0.0% packet loss"
    match = re.search(r"(\d+(?:\.\d+)?)%\s+packet\s+loss", output, re.IGNORECASE)
    if match:
        loss_percent = float(match.group(1))
        success_percent = 100.0 - loss_percent
    else:
        # Could not parse output; assume total failure
        success_percent = 0.0

    print(f"IP = {ip}; success = {success_percent}")
    return ip, success_percent

ונניח שאנחנו רוצים להשתמש בה מתוכנית אסינכרונית באופן הבא:

    ips = [...]
    tasks = [ping_ip(ip) for ip in ips]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    for result in results:
        if isinstance(result, Exception):
            print(f"Error: {result}")
        else:
            ip, success = result
            print(f"{ip}: {success:.1f}% successful pings") 

זה כמובן לא יעבוד כי ping_ip היא לא פונקציה אסינכרונית. היא לא מחזירה את השליטה ל Main Loop בזמן ביצוע ה ping והקוד פשוט ירוץ בצורה סדרתית בדיקה אחרי בדיקה.

במצב כזה מישהו יכול להציע להשתמש ב asynio.to_thread ואז לכתוב:

    tasks = [asyncio.to_thread(ping_ip, ip) for ip in ips]    
    results = await asyncio.gather(*tasks, return_exceptions=True)

    for result in results:
        if isinstance(result, Exception):
            print(f"Error: {result}")
        else:
            ip, success = result
            print(f"{ip}: {success:.1f}% successful pings")

עובד? בטח. אבל.

ההבדל בין Thread Pool ל Asyncio הוא בדיוק הסיבה בגללה רצינו להשתמש ב asyncio. הבעיה עם Threads היא שיש מספר מוגבל מהם כי מערכת ההפעלה לא אוהבת כשיוצרים יותר מדי. ה Default Thread Pool מכיל או 32 threads או מספר הליבות + 4, הקטן מביניהם. לעומת זאת ב asyncio אין לנו הגבלה על כמה משימות רצות במקביל.

איך זה מתורגם לביצועים? נשווה בין שתי הגרסאות.

בתוכנית אחת אני משתמש ב to_thread כדי להפעיל את פונקציית ה ping שכבר יש לי. זמן הריצה 20 שניות:

uv run ping_ips_to_thread.py  0.12s user 0.22s system 1% cpu 20.283 total

בתוכנית שניה כתבתי מחדש את פונקציית ה ping בגישה אסינכרונית עם asyncio.create_subprocess_shell. הפעם זמן הריצה ירד ל 14 שניות לאותה רשימת כתובות:

uv run ping_ips.py  0.13s user 0.21s system 2% cpu 14.203 total

ככל שרשימת הכתובות תתארך כך גם ההפרש בין התוכניות יגדל.