import threading, time, random

def download_file(filename):
    """Simulates downloading a file with a random delay."""
    thread_name = threading.current_thread().name
    print(f"[{thread_name}] Starting download: {filename}")
    sleep_time = random.uniform(1, 3)
    time.sleep(sleep_time)
    print(f"[{thread_name}] Finished download: {filename}")

if __name__ == "__main__":
    filenames = ["doc1.pdf", "img2.jpg", "data3.csv"]
    threads = []
    
    for f in filenames:
        thread = threading.Thread(
            target=download_file, 
            args=(f,), 
            name=f"Downloader-{f[0]}"
        )
        threads.append(thread)
        thread.start()

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

    print("All downloads complete.")

import multiprocessing, time, os
def process_data(data_chunk):
    """Simulates a CPU-intensive task on a chunk of data."""
    pid = os.getpid()
    print(f"[PID:{pid}] Processing chunk {data_chunk['id']}...")
    # Simulate heavy computation
    result = sum(i * i for i in range(2**15))
    print(f"[PID:{pid}] Chunk {data_chunk['id']} done.")
    return result

if __name__ == "__main__":
    data_chunks = [{'id': i} for i in range(4)]
    # Create a pool of worker processes
    # On a 4-core machine, this can run 4 tasks in parallel
    with multiprocessing.Pool(processes=4) as pool:
        results = pool.map(process_data, data_chunks)
    print("All data processed.")

import asyncio
import random
import time

async def fetch_data(name, duration):
    """A coroutine that simulates fetching data."""
    print(f"[{time.time():.1f}s] Task {name}: Starting fetch...")
    # await asyncio.sleep() is the key part.
    # It pauses this coroutine and lets the event loop
    # run other tasks that are ready.
    await asyncio.sleep(duration)
    print(f"[{time.time():.1f}s] Task {name}: Finished fetch.")
    return f"Data from {name}"

async def main():
    start_time = time.time()
    # asyncio.create_task() schedules the coroutine to run
    # on the event loop as soon as possible. It doesn't block.
    task_a = asyncio.create_task(fetch_data("A", 2))
    task_b = asyncio.create_task(fetch_data("B", 3))

    print("Tasks scheduled. Waiting for completion...")
    # The 'await' keyword here pauses main() until task_a is done.
    result_a = await task_a
    # Then it pauses again until task_b is done.
    result_b = await task_b
    
    total_time = time.time() - start_time
    print(f"Results: {result_a}, {result_b}")
    print(f"Total time: {total_time:.2f}s (less than 2+3=5s)")

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

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

URLS = ['http://site1.com', 'http://error.com', 'http://site2.com']

def fetch_url(url):
    """Simulates fetching a URL, can succeed or fail."""
    print(f"Fetching {url}...")
    time.sleep(1) # Simulate network request
    if "error" in url:
        raise ConnectionError(f"Failed to connect to {url}")
    return f"Content from {url}"

# The 'with' statement ensures the pool is properly shut down.
with ThreadPoolExecutor(max_workers=3) as executor:
    # executor.submit() schedules a task and returns a Future object.
    # A Future is a placeholder for a result that will exist later.
    future_to_url = {executor.submit(fetch_url, url): url for url in URLS}
    
    print("Tasks submitted, waiting for results as they complete...")
    # as_completed() yields futures as they finish, in any order.
    for future in as_completed(future_to_url):
        url = future_to_url[future]
        try:
            # future.result() gets the return value of the function.
            # If the function raised an exception, result() re-raises it.
            result = future.result()
            print(f"Success: {result}")
        except Exception as e:
            print(f"Error fetching {url}: {e}")

from multiprocessing import Process, shared_memory
import numpy as np

def worker_process(shm_name, shape, dtype):
    """A worker that connects to shared memory and modifies it."""
    # Connect to the existing shared memory block
    existing_shm = shared_memory.SharedMemory(name=shm_name)
    # Create a NumPy array backed by the shared memory buffer
    shared_array = np.ndarray(shape, dtype=dtype, buffer=existing_shm.buf)
    
    print(f"Worker sees initial data: {shared_array}")
    shared_array[:] = np.flip(shared_array) # Modify the data in place
    print(f"Worker modified data to: {shared_array}")
    
    # Close the shared memory block (doesn't destroy it)
    existing_shm.close()

if __name__ == "__main__":
    # Create a NumPy array in the main process
    original_array = np.array([1, 2, 3, 4, 5])
    
    # Create a new shared memory block
    shm = shared_memory.SharedMemory(create=True, size=original_array.nbytes)
    # Create a NumPy array that uses the shared memory
    shared_array = np.ndarray(original_array.shape, dtype=original_array.dtype, buffer=shm.buf)
    shared_array[:] = original_array[:] # Copy data into shared memory
    
    print(f"Main process created shared data: {shared_array}")

    # Create and start the worker process
    p = Process(target=worker_process, args=(shm.name, original_array.shape, original_array.dtype))
    p.start()
    p.join() # Wait for the worker to finish

    print(f"Main process sees modified data: {shared_array}")

    # Clean up the shared memory block
    shm.close()
    shm.unlink()

import asyncio
import time

async def api_call(session_id, sem):
    """Simulates an API call that needs to be rate-limited."""
    # async with sem: will wait here if the semaphore counter is zero.
    # It decrements the counter on entry and increments on exit.
    async with sem:
        print(f"Session {session_id}: Acquired semaphore, making API call...")
        # Simulate the time taken for the API call
        await asyncio.sleep(1)
        print(f"Session {session_id}: Finished API call, releasing semaphore.")
    return f"Result from {session_id}"

async def main():
    # Create a semaphore that allows only 2 concurrent tasks
    api_semaphore = asyncio.Semaphore(2)
    
    # Create more tasks than the semaphore limit
    tasks = [api_call(i, api_semaphore) for i in range(5)]
    
    print("Dispatching 5 API calls with a limit of 2 concurrent calls...")
    # asyncio.gather will run all tasks, but the semaphore
    # will control the actual execution flow.
    results = await asyncio.gather(*tasks)
    print(f"\\nAll results: {results}")

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

from joblib import Parallel, delayed
import time
import os

def process_input(i):
    """
    A simple function that simulates work and
    returns which process handled it.
    """
    pid = os.getpid()
    print(f"Processing item {i} on process {pid}")
    time.sleep(0.5)
    return (i, pid)

# n_jobs=-1 means use all available CPU cores.
# The Parallel object creates a pool of worker processes.
# 'delayed' is a wrapper that makes the function call lazy,
# so it can be sent to the worker processes.
print("Dispatching 10 jobs to worker processes...")
results = Parallel(n_jobs=-1)(
    delayed(process_input)(i) for i in range(10)
)

print("\\n--- Results ---")
for i, pid in results:
    print(f"Input {i} was processed by {pid}")

# This code must be at the top of the script
from gevent import monkey
monkey.patch_all()

import gevent
import time

def task(pid):
    """
    A task that would normally be blocking,
    but gevent makes time.sleep non-blocking.
    """
    start_time = time.time()
    print(f"Task {pid}: Starting at {time.strftime('%X')}")
    # Because of monkey-patching, this sleep is cooperative.
    # It yields control to the gevent hub, allowing other
    # greenlets to run.
    time.sleep(1)
    end_time = time.time()
    print(f"Task {pid}: Finished in {end_time - start_time:.2f}s")

def asynchronous():
    # gevent.spawn creates a greenlet and schedules it to run.
    threads = [gevent.spawn(task, i) for i in range(3)]
    # gevent.joinall waits for all the greenlets in the list to complete.
    gevent.joinall(threads)

print("Running with gevent (simulation):")
# Without gevent, this would take ~3 seconds.
# With gevent, it takes ~1 second.
asynchronous()