如何用 C 实现一个 Python Awaitable 函数

Mar 23, 2020

2 minute read

这是一个取巧的方式，用 C 实现一个普通的 Python 函数，但返回一个 asyncio.Future 对象。用副线程去执行，等到执行有结果后，再调用 asyncio.Future.set_result 方法去设置函数的结果。先看一下直接用 Python 的话，是怎么实现这个的。

import asyncio
import os
import threading

def call_system(cmd: str, fut: asyncio.Future):
    exit_code = os.system(cmd)
    loop = fut.get_loop()
    loop.call_soon_threadsafe(fut.set_result, exit_code)

def system(cmd):
    fut = asyncio.Future()
    threading.Thread(target=call_system, args=(cmd, fut)).start()
    return fut

可以注意到副线程并没有直接执行 Future.set_result 。因为从副线程回调的话，必须调用 loop.call_soon_threadsafe 去让 loop 去调度回调。避免了执行 await spam.system('ls') 卡住。

现在需要做的事，就是把以上的代码翻译成 C 代码

#define PY_SSIZE_T_CLEAN
#include <Python.h>
#include <pthread.h>
#include <stdio.h>

struct call_system_ctx {
  char *command;
  PyObject *fut;
};

static void *call_system(void *args) {
  printf("=== pthread launch\n");
  struct call_system_ctx *ctx = (struct call_system_ctx *)args;
  printf("=== call system(%s)\n", ctx->command);
  int exit_code = system(ctx->command);
  printf("=== call system(%s) done\n", ctx->command);

  printf("=== acquire GIL\n");
  PyGILState_STATE gstate = PyGILState_Ensure();
  PyObject *loop = PyObject_CallMethod(ctx->fut, "get_loop", NULL);
  PyObject *set_result = PyObject_GetAttrString(ctx->fut, "set_result");
  printf("=== schedule callback\n");
  PyObject *handler = PyObject_CallMethod(loop, "call_soon_threadsafe", "(O,i)",
                                          set_result, exit_code);
  Py_DECREF(handler);
  Py_DECREF(set_result);
  Py_DECREF(loop);
  Py_DECREF(ctx->fut);
  printf("=== release GIL\n");
  PyGILState_Release(gstate);
  printf("=== pthread exit\n");
  free(ctx);
  pthread_exit(NULL);
  return 0;
}

static PyObject *spam_system(PyObject *self, PyObject *args) {
  char *command;

  if (!PyArg_ParseTuple(args, "s", &command))
    return NULL;

  // create a asyncio.Future object
  PyObject *asyncio_module = PyImport_ImportModule("asyncio");
  PyObject *fut = PyObject_CallMethod(asyncio_module, "Future", NULL);
  Py_DECREF(asyncio_module);
  // pass command and fut parameters
  struct call_system_ctx *ctx = malloc(sizeof(struct call_system_ctx));
  ctx->command = command;
  ctx->fut = fut;
  // increase the reference count of the asyncio.Future object, in case the
  // caller does not keep it
  Py_INCREF(fut);
  // create a thread to call system
  pthread_t thread_id;
  int rc = pthread_create(&thread_id, NULL, call_system, (void *)ctx);
  if (rc) {
    printf("=== Fail to create thread\n");
  }
  // return the unfinished asyncio.Future object
  return fut;
}

static PyMethodDef SpamMethods[] = {
    {"system", spam_system, METH_VARARGS, "Execute a shell command."},
    {NULL, NULL, 0, NULL}};

static struct PyModuleDef spammodule = {PyModuleDef_HEAD_INIT, "spam", NULL, -1,
                                        SpamMethods};

PyMODINIT_FUNC PyInit_spam(void) { return PyModule_Create(&spammodule); }

再补上 setup.py，执行 python setup.py install 就可以编译安装成一个 Python 的第三方函数库了。

from distutils.core import setup, Extension

setup(name="spam", version="1.0", ext_modules=[Extension("spam", ["spammodule.c"])])

总结

这个方法实现起来十分简单。当初在 Stackoverflow 看到一个回答，感觉很不错。链接给在下方了。大触只讲述了如何实现，但没有给具体的实现。所以本文算是对这个问题的一个补充。该回答提到的其他方式，有空我再补上其 C 代码的实现。展开的话，还可以分享一下 async for 和 async with 的 C 代码实现。

linw1995

总结

参考