Python Green Threads vs OS Threads

import threading
import time

def cpu_intensive_task(n):
    """Simulate CPU-intensive work"""
    total = 0
    for i in range(n * 1000000):
        total += i
    return total

# Create OS threads in Python
threads = []
for i in range(4):
    thread = threading.Thread(target=cpu_intensive_task, args=(10,))
    threads.append(thread)
    thread.start()

# Wait for all threads to complete
for thread in threads:
    thread.join()

import asyncio

async def io_task(name, duration):
    """Simulate I/O-bound work"""
    print(f"Task {name} starting")
    await asyncio.sleep(duration)  # Cooperative yield point
    print(f"Task {name} completed")
    return f"Result from {name}"

async def main():
    # Create multiple coroutines (green threads)
    tasks = [
        io_task("A", 2),
        io_task("B", 1),
        io_task("C", 3)
    ]

    # Run concurrently on a single OS thread
    results = await asyncio.gather(*tasks)
    print(f"Results: {results}")

# Event loop manages green thread scheduling
asyncio.run(main())

import gevent
from gevent import monkey
monkey.patch_all()  # Patch standard library for green thread support

def fetch_url(url):
    """Simulate network request"""
    print(f"Fetching {url}")
    gevent.sleep(1)  # Yields control to other green threads
    return f"Content from {url}"

# Create green threads
greenlets = [
    gevent.spawn(fetch_url, f"http://example.com/{i}")
    for i in range(1000)  # Can create thousands easily!
]

# Wait for all to complete
gevent.joinall(greenlets)

import asyncio
import aiohttp

async def fetch_data(session, url):
    try:
        async with asyncio.timeout(10):
            async with session.get(url) as response:
                response.raise_for_status()
                return await response.text()
    except (asyncio.TimeoutError, aiohttp.ClientError) as e:
        return None  # or log and retry

async def main():
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(*[
            fetch_data(session, f"http://api.example.com/{i}")
            for i in range(100)
        ])

# BAD: blocks the entire event loop for 5 seconds
async def bad_handler():
    time.sleep(5)           # WRONG — use await asyncio.sleep(5)
    requests.get(url)       # WRONG — use aiohttp
    data = db.query(sql)    # WRONG — use asyncpg or aiomysql

# GOOD: run blocking code in a thread pool
async def good_handler():
    loop = asyncio.get_event_loop()
    # Offload blocking call to thread pool
    result = await loop.run_in_executor(None, requests.get, url)

import threading
import time

# Even with multiple OS threads, the GIL prevents true parallelism
def count(n):
    while n > 0:
        n -= 1

# These threads won't run in parallel due to GIL
t1 = threading.Thread(target=count, args=(100000000,))
t2 = threading.Thread(target=count, args=(100000000,))

start = time.time()
t1.start()
t2.start()
t1.join()
t2.join()
print(f"Time with threads: {time.time() - start}")

# Often slower than sequential due to GIL contention!

import sys
print(sys._is_gil_enabled())  # False on python3.13t

# CPU-bound threads now actually run in parallel
# 4 threads on 4 cores → ~4x speedup (finally!)

# Install free-threaded Python
# On macOS: brew install python@3.13 --with-freethreading
# On Ubuntu: sudo apt install python3.13-nogil

# Run with free threading
python3.13t your_script.py

// Go example - goroutines are green threads mapped to OS threads
func main() {
    // Create thousands of goroutines (green threads)
    for i := 0; i < 10000; i++ {
        go func(id int) {
            // Go runtime maps these to a pool of OS threads
            fmt.Printf("Goroutine %d\n", id)
        }(i)
    }
}

import multiprocessing
import asyncio

async def async_worker(data):
    """Green thread worker for I/O"""
    await asyncio.sleep(0.1)
    return data * 2

def process_worker(chunk):
    """OS process for CPU work"""
    # Run event loop in each process
    async def process_chunk():
        tasks = [async_worker(item) for item in chunk]
        return await asyncio.gather(*tasks)

    return asyncio.run(process_chunk())

# Combine multiprocessing (true parallelism) with asyncio (green threads)
if __name__ == "__main__":
    data = range(1000)
    chunks = [data[i:i+100] for i in range(0, len(data), 100)]

    with multiprocessing.Pool() as pool:
        results = pool.map(process_worker, chunks)

import os

# For I/O-bound thread pools
max_workers = min(32, (os.cpu_count() or 1) + 4)

# For CPU-bound (use multiprocessing instead, but if you must)
max_workers = os.cpu_count() or 1

import asyncio

# Limit concurrent requests to prevent overwhelming the server
sem = asyncio.Semaphore(50)

async def limited_fetch(session, url):
    async with sem:
        async with session.get(url) as resp:
            return await resp.text()

# Even with 10,000 URLs, only 50 run concurrently
tasks = [limited_fetch(session, url) for url in urls]
results = await asyncio.gather(*tasks)

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(fetch_users())
        task2 = tg.create_task(fetch_orders())
        task3 = tg.create_task(fetch_inventory())
    # All tasks guaranteed complete (or all cancelled on error)
    users, orders, inventory = task1.result(), task2.result(), task3.result()

import signal

async def shutdown(loop, signal=None):
    tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
    for task in tasks:
        task.cancel()
    await asyncio.gather(*tasks, return_exceptions=True)
    loop.stop()

loop = asyncio.get_event_loop()
for sig in (signal.SIGTERM, signal.SIGINT):
    loop.add_signal_handler(sig, lambda s=sig: asyncio.create_task(shutdown(loop, s)))

# Measuring context switch time
import threading
import asyncio
import time

# OS Thread context switch
def thread_switch_test():
    event1 = threading.Event()
    event2 = threading.Event()
    switches = 100000

    def thread1():
        for _ in range(switches):
            event2.set()
            event1.wait()
            event1.clear()

    def thread2():
        for _ in range(switches):
            event2.wait()
            event2.clear()
            event1.set()

    t1 = threading.Thread(target=thread1)
    t2 = threading.Thread(target=thread2)

    start = time.time()
    t1.start()
    t2.start()
    t1.join()
    t2.join()

    total_time = time.time() - start
    return total_time / (switches * 2)

# Green thread (coroutine) context switch
async def coro_switch_test():
    switches = 100000
    counter = 0

    async def coro1():
        nonlocal counter
        for _ in range(switches):
            counter += 1
            await asyncio.sleep(0)

    async def coro2():
        nonlocal counter
        for _ in range(switches):
            counter += 1
            await asyncio.sleep(0)

    start = time.time()
    await asyncio.gather(coro1(), coro2())
    total_time = time.time() - start

    return total_time / (switches * 2)

# Results typically show:
# OS Thread switch: ~5-10 microseconds
# Green thread switch: ~0.1-0.5 microseconds

from concurrent.futures import ThreadPoolExecutor
import psycopg2

def query_database(query):
    conn = psycopg2.connect("postgresql://...")
    cursor = conn.cursor()
    cursor.execute(query)
    result = cursor.fetchall()
    conn.close()
    return result

# Limited by thread overhead
with ThreadPoolExecutor(max_workers=100) as executor:
    futures = [executor.submit(query_database, f"SELECT * FROM table_{i}")
               for i in range(100)]

import asyncio
import asyncpg

async def query_database(pool, query):
    async with pool.acquire() as conn:
        return await conn.fetch(query)

async def main():
    # Can handle thousands of concurrent queries
    pool = await asyncpg.create_pool("postgresql://...")

    tasks = [query_database(pool, f"SELECT * FROM table_{i}")
             for i in range(10000)]

    results = await asyncio.gather(*tasks)
    await pool.close()
    return results