221. Using Python’s Trio Library

---

🔹 1. Basic Trio Async Function

import trio

async def say_hello():
    print("Hello, Trio!")

trio.run(say_hello)

✅ Fix: trio.run() starts an event loop and runs the coroutine.

---

🔹 2. Using trio.sleep() for Non-Blocking Delays

import trio

async def delayed_message():
    print("Waiting...")
    await trio.sleep(2)  # Non-blocking sleep
    print("Done waiting!")

trio.run(delayed_message)

✅ Fix: trio.sleep() lets other tasks run instead of blocking.

---

🔹 3. Running Multiple Tasks Concurrently

import trio

async def task1():
    print("Task 1 started")
    await trio.sleep(1)
    print("Task 1 finished")

async def task2():
    print("Task 2 started")
    await trio.sleep(2)
    print("Task 2 finished")

async def main():
    async with trio.open_nursery() as nursery:
        nursery.start_soon(task1)
        nursery.start_soon(task2)

trio.run(main)

✅ Fix: open_nursery() manages multiple tasks concurrently.

---

🔹 4. Using nursery.start() for Structured Concurrency

import trio

async def child_task():
    print("Child task running...")
    await trio.sleep(1)
    print("Child task completed.")

async def main():
    async with trio.open_nursery() as nursery:
        await nursery.start(child_task)  # Ensures task starts before moving forward

trio.run(main)

✅ Fix: nursery.start() guarantees task fully starts before continuing.

---

🔹 5. Handling Task Cancellation with trio.CancelScope

import trio

async def long_running_task():
    print("Task started...")
    try:
        await trio.sleep(5)
        print("Task completed!")
    except trio.Cancelled:
        print("Task was cancelled!")

async def main():
    async with trio.open_nursery() as nursery:
        with trio.CancelScope() as cancel_scope:
            nursery.start_soon(long_running_task)
            await trio.sleep(2)  # Wait for a while
            cancel_scope.cancel()  # Cancel the task

trio.run(main)

✅ Fix: trio.CancelScope() cancels a running task safely.

---

🔹 6. Using trio.to_thread.run_sync() for Blocking Calls

import trio
import time

def blocking_task():
    time.sleep(2)  # Simulates a slow blocking operation
    return "Blocking task done!"

async def main():
    result = await trio.to_thread.run_sync(blocking_task)
    print(result)

trio.run(main)

✅ Fix: trio.to_thread.run_sync() offloads blocking work to a separate thread.

---

🔹 7. Using Trio Channels for Inter-Task Communication

import trio

async def producer(send_channel):
    async with send_channel:
        for i in range(3):
            print(f"Sending: {i}")
            await send_channel.send(i)
            await trio.sleep(1)

async def consumer(receive_channel):
    async with receive_channel:
        async for value in receive_channel:
            print(f"Received: {value}")

async def main():
    send_channel, receive_channel = trio.open_memory_channel(3)
    async with trio.open_nursery() as nursery:
        nursery.start_soon(producer, send_channel)
        nursery.start_soon(consumer, receive_channel)

trio.run(main)

✅ Fix: trio.open_memory_channel() enables safe inter-task communication.

---

🔹 8. Trio’s Built-in Timeouts

import trio

async def long_task():
    await trio.sleep(5)  # Simulating a long task
    return "Completed"

async def main():
    with trio.move_on_after(2):  # Timeout after 2 seconds
        result = await long_task()
        print(result)  # Won't print because it times out

trio.run(main)

✅ Fix: trio.move_on_after() cancels tasks after a timeout.

---

🔹 9. Handling Exceptions Gracefully

import trio

async def faulty_task():
    raise ValueError("Something went wrong!")

async def main():
    async with trio.open_nursery() as nursery:
        try:
            nursery.start_soon(faulty_task)
        except Exception as e:
            print(f"Caught exception: {e}")

trio.run(main)

❌ Issue: Exception is NOT caught because it occurs in another task. ✅ Fix: Use exception groups or handle exceptions inside the task.

---

🔹 10. Trio’s Signal Handling (trio.lowlevel)

import trio
import signal

async def signal_handler():
    print("Signal received. Cleaning up...")
    await trio.sleep(1)
    print("Cleanup done!")

async def main():
    with trio.open_signal_receiver(signal.SIGINT) as signal_stream:
        async for _ in signal_stream:
            await signal_handler()

trio.run(main)

✅ Fix: trio.open_signal_receiver() handles Ctrl+C (SIGINT) safely.

---

🚀 Summary: Why Use Trio?

Feature	Trio
Structured Concurrency	✅ nursery ensures tasks are managed properly
Task Cancellation	✅ trio.CancelScope() allows safe cancellation
Thread Offloading	✅ trio.to_thread.run_sync() moves blocking calls to a thread
Inter-Task Communication	✅ trio.open_memory_channel() provides safe channels
Timeout Handling	✅ trio.move_on_after() automatically cancels slow tasks
Signal Handling	✅ trio.open_signal_receiver() handles OS signals like SIGINT

---

🚀 When to Use Trio?

Scenario	Use Trio?
Multiple async tasks (e.g., HTTP requests, I/O)	✅ Yes
Handling timeouts and cancellations properly	✅ Yes
Coordinating tasks with dependencies	✅ Yes
Replacing asyncio	✅ Yes (simpler & safer)
Blocking tasks with threads	✅ Yes (trio.to_thread.run_sync)
CPU-bound tasks (heavy computation)	❌ No (use multiprocessing)

---