Python Asyncio源码解析：为何_get_running_loop执行C实现而非Python实现？

你观察到的现象是CPython内置模块常用的优化手段：在模块初始化阶段，用C实现的函数替换Python层面定义的同名函数，既保证性能，又保留Python代码作为降级实现和逻辑参考。

具体原理拆解 #

1. 模块初始化时的函数覆盖 #

asyncio的核心部分通过C扩展实现（对应_asynciomodule.c），当Python解释器加载asyncio模块时，C扩展的初始化逻辑会将C实现的_get_running_loop函数挂载到模块的命名空间中，直接覆盖你在events.py里看到的Python版本定义。

2. C函数与Python名称的绑定流程 #

CPython通过一套固定的机制完成C函数到Python函数名的绑定：

• 第一步：定义C实现函数
  就是你贴出的_asyncio_get_running_loop_impl，它直接操作线程状态（_PyThreadState_GET()）获取当前运行的事件循环，比Python层通过_running_loop.loop_pid判断的效率更高。
• 第二步：关联函数名与C实现
  在C扩展中，会通过PyMethodDef结构体将C函数和Python层面的函数名绑定：

  static PyMethodDef _asyncio_methods[] = {
      {"_get_running_loop", _asyncio_get_running_loop_impl, METH_NOARGS, 
       PyDoc_STR("Return the running event loop or None.")},
      // 其他asyncio内置方法的绑定...
      {NULL, NULL}  // 数组结束标记
  };
  

• 第三步：初始化模块时完成替换
  在C扩展的初始化函数PyInit__asyncio中，会调用PyModule_AddFunctions或直接修改模块的字典，将上述绑定的函数注册到asyncio模块中，此时Python层原本定义的_get_running_loop就会被C版本覆盖。

3. 保留Python实现的意义 #

• 降级兼容：如果C扩展因特殊原因无法加载（比如某些定制编译的Python环境），Python层的实现可以作为 fallback 保证模块正常运行。
• 逻辑参考：Python代码比C代码更易读，开发者可以通过Python版本快速理解函数的设计意图，而不需要直接阅读复杂的C源码。

你提供的代码参考 #

Python层定义（events.py） #

def get_running_loop():
    """Return the running event loop. Raise a RuntimeError if there is none.
    This function is thread-specific.
    """
    # NOTE: this function is implemented in C (see _asynciomodule.c)
    loop = _get_running_loop()
    if loop is None:
        raise RuntimeError('no running event loop')
    return loop

def _get_running_loop():
    """Return the running event loop or None.
    This is a low-level function intended to be used by event loops.
    This function is thread-specific.
    """
    # NOTE: this function is implemented in C (see _asynciomodule.c)
    running_loop, pid = _running_loop.loop_pid
    if running_loop is not None and pid == os.getpid():
        return running_loop

C层实现（_asynciomodule.c） #

static PyObject *
_asyncio_get_running_loop_impl(PyObject *module)
/*[clinic end generated code: output=c247b5f9e529530e input=2a3bf02ba39f173d]*/
{
    PyObject *loop;
    _PyThreadStateImpl *ts = (_PyThreadStateImpl *)_PyThreadState_GET();
    loop = Py_XNewRef(ts->asyncio_running_loop);
    if (loop == NULL) {
        /* There's no currently running event loop */
        PyErr_SetString(
            PyExc_RuntimeError, "no running event loop");
        return NULL;
    }
    return loop;
}

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