import asyncio

async def quick_task():
    # 這個任務瞬間完成
    return "done"

async def wrapper():
    # 設置30秒超時，但實際任務1秒就完成
    return await asyncio.wait_for(quick_task(), timeout=30)

async def main():
    task = asyncio.create_task(wrapper())
    await asyncio.sleep(0)  # 讓任務開始執(zhí)行
    task.cancel()           # 手動取消
    
    try:
        await task
    except asyncio.CancelledError:
        print("任務被取消了")
    else:
        print("任務沒有被取消！")

asyncio.run(main())

import asyncio

async def cancellable_wait_for(coro, timeout):
    """
    一個更可靠的wait_for版本，確保取消信號不會被吞掉
    """
    task = asyncio.create_task(coro)
    
    try:
        # 等待任務完成，或者超時
        return await asyncio.wait_for(task, timeout=timeout)
    except asyncio.TimeoutError:
        # 超時了，取消任務
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            # 確保取消信號被傳播出去
            pass
        raise  # 重新拋出TimeoutError
    except asyncio.CancelledError:
        # 外部取消了，把取消信號傳遞給內部任務
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass
        raise  # 重新拋出CancelledError

async def my_task():
    try:
        await asyncio.sleep(10)
        return "完成"
    except asyncio.CancelledError:
        print("內部任務被取消了")
        raise

async def main():
    task = asyncio.create_task(cancellable_wait_for(my_task(), timeout=5))
    await asyncio.sleep(2)
    task.cancel()
    
    try:
        await task
    except asyncio.CancelledError:
        print("外部：確實被取消了")

asyncio.run(main())

# pip install quattro
import asyncio
from quattro import move_on_after

async def main():
    with move_on_after(5) as scope:  # 5秒超時
        result = await some_slow_operation()
        print(result)
    
    if scope.cancelled_caught:
        print("超時被取消了，但程序繼續(xù)運行")

asyncio.run(main())

import asyncio

async def cancellable_operation(timeout):
    """
    在操作內部主動檢查超時和取消信號
    """
    start = asyncio.get_event_loop().time()
    
    # 假設這是一系列的小步驟
    for step in range(10):
        # 檢查是否超時
        if asyncio.get_event_loop().time() - start > timeout:
            raise asyncio.TimeoutError()
        
        # 檢查是否被取消（通過捕獲取消信號）
        try:
            await asyncio.sleep(0.5)  # 模擬一個小步驟
        except asyncio.CancelledError:
            # 做清理工作
            print("收到取消信號，正在清理...")
            raise  # 重新拋出，讓上層知道被取消了
        
        print(f"完成步驟 {step}")
    
    return "全部完成"

async def main():
    task = asyncio.create_task(cancellable_operation(timeout=3))
    await asyncio.sleep(2)
    task.cancel()
    
    try:
        result = await task
        print(result)
    except asyncio.TimeoutError:
        print("超時了")
    except asyncio.CancelledError:
        print("被取消了")

asyncio.run(main())

import asyncio
import threading
import time

def blocking_task(stop_event):
    """
    這是一個會阻塞的同步函數
    它定期檢查stop_event，如果被設置了就主動退出
    """
    print("同步任務開始")
    for i in range(30):
        if stop_event.is_set():
            print("收到停止信號，主動退出")
            return "被停止了"
        print(f"工作中... {i}")
        time.sleep(1)  # 模擬耗時操作
    return "正常完成"

async def main():
    stop_event = threading.Event()
    loop = asyncio.get_running_loop()
    
    # 把同步任務扔到線程池里執(zhí)行
    task = loop.run_in_executor(None, blocking_task, stop_event)
    
    # 5秒后發(fā)送停止信號
    await asyncio.sleep(5)
    stop_event.set()
    
    # 等待任務結束
    result = await task
    print(f"結果: {result}")

asyncio.run(main())

import asyncio
from concurrent.futures import ProcessPoolExecutor
import time

def cpu_intensive_task():
    """模擬一個費CPU的大活兒"""
    total = 0
    for i in range(100000000):
        total += i
        # 每1000萬次檢查一次，但進程間通信復雜，先忽略細節(jié)
        if i % 10000000 == 0:
            print(f"計算到 {i}")
    return total

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor(max_workers=1) as pool:
        # 在子進程中執(zhí)行
        task = loop.run_in_executor(pool, cpu_intensive_task)
        
        # 等待3秒
        await asyncio.sleep(3)
        print("3秒到了，但子進程不會自己停...")
        # 注意：ProcessPoolExecutor沒有優(yōu)雅的取消方法
        # 只能shutdown(wait=False)但會有warning
        # 或者terminate，但不推薦