异步协程中控制流与资源锁的精细化管理

import asyncio
from dataclasses import dataclass
from typing import Awaitable, AsyncIterator, Dict, Tuple

# 假设的类和类型定义，简化以突出核心问题
class TextDetector:
    lock: asyncio.Lock = asyncio.Lock() # 每个检测器实例一个锁

    async def cooldown(self) -> bool:
        print(f"Detector {id(self)}: Cooldown started...")
        await asyncio.sleep(0.1) # 模拟冷却时间
        print(f"Detector {id(self)}: Cooldown finished.")
        return True

    async def detect(self, input_data: str) -> str:
        print(f"Detector {id(self)}: Detection started for '{input_data}'...")
        await asyncio.sleep(0.2) # 模拟检测时间
        print(f"Detector {id(self)}: Detection finished.")
        return f"Detected result for {input_data}"

@dataclass
class TextDetectorInput:
    language: str
    text: str

# 原始问题中的 cooldown_and_detect 尝试
async def original_cooldown_and_detect(detector: TextDetector, detector_input: TextDetectorInput):
    with detector.lock: # 问题所在：锁在这里被释放
        cooleddown = await detector.cooldown()
        # 这里返回 detector.detect()，它是一个 coroutine object，而不是已完成的 Future
        return detector.detect(detector_input.text)

# 模拟调用方
async def caller_example():
    detector1 = TextDetector()
    input_data = TextDetectorInput(language="en", text="Hello Async")

    print("--- Calling original_cooldown_and_detect ---")
    detection_coroutine = await original_cooldown_and_detect(detector1, input_data)
    # 此时，with 语句已经结束，锁已释放
    print("Caller: Received detection coroutine. Lock *might* be released already.")
    result = await detection_coroutine
    print(f"Caller: Final result: {result}")

# asyncio.run(caller_example())

import asyncio
from dataclasses import dataclass
from typing import Awaitable, AsyncIterator, Dict, Tuple

# 假设的类和类型定义，与上文一致
class TextDetector:
    lock: asyncio.Lock = asyncio.Lock() # 每个检测器实例一个锁

    async def cooldown(self) -> bool:
        print(f"Detector {id(self)}: Cooldown started...")
        await asyncio.sleep(0.1)
        print(f"Detector {id(self)}: Cooldown finished.")
        return True

    async def detect(self, input_data: str) -> str:
        print(f"Detector {id(self)}: Detection started for '{input_data}'...")
        await asyncio.sleep(0.2)
        print(f"Detector {id(self)}: Detection finished.")
        return f"Detected result for {input_data}"

@dataclass
class TextDetectorInput:
    language: str
    text: str

async def cooldown_and_detect(detector: TextDetector, detector_input: TextDetectorInput) -> Awaitable[str]:
    """
    执行冷却和检测过程，确保锁在整个检测任务完成后才释放。
    返回一个 asyncio.Task 对象，调用方可 await 此任务以获取最终结果。
    """
    # 1. 显式获取锁
    print(f"Detector {id(detector)}: Attempting to acquire lock...")
    await detector.lock.acquire()
    print(f"Detector {id(detector)}: Lock acquired.")
    try:
        # 2. 执行冷却操作
        cooleddown = await detector.cooldown()
        if not cooleddown:
            raise RuntimeError("Cooldown failed.")

        # 3. 创建异步任务来执行检测操作
        # 注意：这里创建的是一个 Task，而不是直接 await
        print(f"Detector {id(detector)}: Creating detection task...")
        detector_task = asyncio.create_task(detector.detect(detector_input.text))

        # 4. 注册完成回调，确保任务完成后释放锁
        # lambda task: detector.lock.release() 会在 detector_task 完成时被调用
        detector_task.add_done_callback(lambda task: detector.lock.release())
        print(f"Detector {id(detector)}: Detection task created with lock release callback.")

        # 5. 返回任务对象，控制流回到调用方
        return detector_task
    except Exception as error:
        # 如果冷却阶段发生异常，需要在此处释放锁
        print(f"Detector {id(detector)}: Error during cooldown or task creation: {error}. Releasing lock.")
        detector.lock.release()
        raise # 重新抛出异常

# 模拟调用方如何使用修正后的函数
async def caller_with_fixed_cooldown():
    detector1 = TextDetector()
    detector2 = TextDetector() # 另一个检测器实例，用于演示锁的竞争
    input_data1 = TextDetectorInput(language="en", text="Hello Async")
    input_data2 = TextDetectorInput(language="fr", text="Bonjour Async")

    print("\n--- Calling fixed cooldown_and_detect for detector1 ---")
    # cooldown_awaitable 现在是一个 asyncio.Task
    detection_task1 = await cooldown_and_detect(detector1, input_data1)
    print("Caller: Received detection task for detector1. It's now running in background.")

    # 尝试同时启动另一个检测器，看锁是否有效
    print("\n--- Attempting to call fixed cooldown_and_detect for detector2 ---")
    # 由于 detector1 还在持有锁，detector2 会等待
    detection_task2 = await cooldown_and_detect(detector2, input_data2)
    print("Caller: Received detection task for detector2. It's now running in background.")

    # 调用方可以等待这些任务完成
    print("\n--- Awaiting detection tasks ---")
    result1 = await detection_task1
    print(f"Caller: Detector1 final result: {result1}")

    result2 = await detection_task2
    print(f"Caller: Detector2 final result: {result2}")

    print("\n--- All tasks completed ---")

# 运行示例
if __name__ == "__main__":
    asyncio.run(caller_with_fixed_cooldown())