Python：如何让wait函数触发父生成器yield？保持API兼容

我有一个基于Python的音乐库，内置时钟系统。目前采用多线程实现，通过Events、Locks等机制保持同步，但我希望将其改为高效的单线程实现。

当前时钟支持使用fork函数启动音乐进程，用法如下：

def some_musical_process():
    play_a_note()
    wait(2)
    play_another_note()
    wait(1.5)
    # ... 其他操作

fork(some_musical_process)

调用fork时，目标函数会在独立线程中运行。wait函数会暂停当前进程，等待指定拍数后再唤醒。但多线程实现的问题在于，很难精准地启停这些音乐进程。

我认为这个场景更适合用生成器实现，示例如下：

import random
import time

def yielding_routine():
    while True:
        print("hello")
        yield random.choice([0.5, 1.0])

# 调度器逻辑
yr = yielding_routine()
delay = next(yr)
while True:
    time.sleep(delay)
    delay = next(yr)

但我不想修改现有API，希望用户依然可以调用wait而非直接使用yield——一方面是为了向后兼容，另一方面是因为库中还有wait_for_children_to_finish这类特殊的wait函数。我期望实现的效果如下：

def yielding_routine():
    while True:
        print("hello")
        wait(random.choice([0.5, 1.0]))

def wait(dur):
    # 某种方式让父生成器yield dur

也就是让wait函数触发它所在的父生成器yield出指定时长。Python支持这种实现吗？哪怕是用eval这类hack手段？本质上我需要类似宏的功能，不确定Python有没有合适的实现方式，或者有没有其他我没考虑到的替代方案？

1. 生成器装饰器+全局上下文（无依赖hack实现） #

用装饰器把普通函数转换成生成器，同时通过全局变量跟踪当前执行的生成器，让wait函数直接触发父生成器的yield操作，完全保留原有API。

示例代码：

import time
from functools import wraps

# 全局存储当前执行的生成器上下文
_current_generator = None

def wrap_as_generator(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        global _current_generator
        # 保存之前的上下文，处理嵌套调用
        prev_gen = _current_generator
        try:
            # 初始化生成器并标记为当前上下文
            gen = func(*args, **kwargs)
            _current_generator = gen
            # 启动生成器，获取第一个等待时长
            delay = next(gen)
            while True:
                time.sleep(delay)
                # 继续执行生成器，获取下一个等待时长
                delay = gen.send(None)
        except StopIteration:
            pass
        finally:
            # 恢复之前的上下文
            _current_generator = prev_gen
    return wrapper

def wait(dur):
    if _current_generator is not None:
        # 触发父生成器yield，返回等待时长
        yield dur

# 修改fork函数，自动给目标函数套装饰器
def fork(func):
    decorated_func = wrap_as_generator(func)
    decorated_func()

# 测试用例
def play_a_note():
    print("Playing note...")

def play_another_note():
    print("Playing another note...")

@wrap_as_generator
def some_musical_process():
    play_a_note()
    wait(2)
    play_another_note()
    wait(1.5)
    play_a_note()

fork(some_musical_process)

2. Greenlet协程切换（更优雅的非hack方案） #

使用greenlet轻量级协程库，让wait函数直接切换回调度器并传递等待时长，无需全局变量，逻辑更清晰稳定。

示例代码：

from greenlet import greenlet
import time

# 调度器协程，负责处理等待和协程切换
def scheduler():
    while True:
        # 等待子协程切换回来，获取等待时长
        wait_duration = greenlet.getcurrent().parent.switch()
        time.sleep(wait_duration)
        # 切换回子协程继续执行
        greenlet.getcurrent().switch()

# 初始化调度器
sched = greenlet(scheduler)
sched.switch()

def wait(dur):
    # 切换回调度器，传递等待时长
    greenlet.getcurrent().parent.switch(dur)

def fork(func):
    # 创建子协程并启动
    child_gr = greenlet(func)
    child_gr.switch()

# 测试用例
def play_a_note():
    print("Playing note...")

def play_another_note():
    print("Playing another note...")

def some_musical_process():
    play_a_note()
    wait(2)
    play_another_note()
    wait(1.5)
    play_a_note()

fork(some_musical_process)

3. 备选方案：asyncio异步实现（需轻微API调整） #

如果可以接受给wait添加await关键字，Python原生的asyncio是更标准的单线程异步方案，调度逻辑更成熟，但会破坏原有API的完全兼容性。

内容的提问来源于stack exchange，提问作者MarcTheSpark