Async

基于 PHP Fiber 的按入口隔离阻塞 IO 协程化扩展。

1. 设计理念

Async::register($scheduler)     // 注册调度回调（机制）
Async::await($callback, $hooks) // 启动协程，标记 hook 类型

  → 以 callback 为入口创建 Fiber
  → 调用链下标记的阻塞操作被 C 层 hook 拦截
  → C 层调用注册的调度回调，传入 handlerType + Fiber + data
  → 使用者负责把 Fiber 的恢复注册到自己的 event-loop
  → 未标记的阻塞操作不受影响

只提供机制，不提供策略。 C 层不绑定任何 event-loop，调度完全由使用者的 PHP 代码控制。

三层调度策略

优先级	调度类型	适用场景	data 类型
1	IO_READ / IO_WRITE	能拿到 stream resource 的 IO 操作	stream resource
2	DELAY / REPEAT	能拿到时间参数的操作	float 秒数
3	轮询 DELAY(0) 兜底	拿不到底层资源，反复让出+重试	0.0

2. 与 Swoole/Swow 的区别

	Swoole/Swow	Async
event-loop	内置绑定	使用者自选
影响范围	全局 hook	按入口隔离
可控性	低（全或无）	高（选择 hook 类型）
侵入性	需替换运行模式	register + await
调度器	框架控制	使用者控制
stream 实现	重写 ops 层（Swow 2400+ 行）	函数 handler 替换 + buffer 快速路径

3. 快速开始

3.1 安装

# 通过 PIE 安装（推荐）
pie install sharing/async

# 或手动编译
cd src
phpize
./configure --enable-async
make -j$(nproc)
make install

启用扩展：

echo "extension=async.so" >> /usr/local/etc/php/conf.d/async.ini

3.2 使用

use Revolt\EventLoop;

require __DIR__ . '/vendor/autoload.php';

// 注册调度回调
Async::register(function (int $handlerType, Fiber $fiber, mixed $data): void {
    match ($handlerType) {
        Async::HANDLER_TYPE_DELAY    => EventLoop::delay($data, fn() => $fiber->resume()),
        Async::HANDLER_TYPE_IO_READ  => EventLoop::onReadable($data, fn() => $fiber->resume()),
        Async::HANDLER_TYPE_IO_WRITE => EventLoop::onWritable($data, fn() => $fiber->resume()),
    };
});

// 心跳协程
Async::await(function () use ($conn) {
    while (true) {
        $conn->heartbeat();
        sleep(30); // 被 HOOK_SLEEP 协程化
    }
}, Async::HOOK_SLEEP);

// 消费协程
Async::await(function () use ($conn) {
    $conn->consume(fn($msg) => processMessage($msg));
}, Async::HOOK_SOCKET);

EventLoop::run();

4. 说明

4.1 Hooks

常量	值	说明	调度方式	覆盖的函数	备注
HOOK_SLEEP	1	睡眠	DELAY	sleep, usleep, time_nanosleep, time_sleep_until	精确延时调度
HOOK_SOCKET	2	网络 IO	IO_READ / IO_WRITE	fsockopen, stream_socket_client/accept, fread, fwrite, fgets, stream_select, stream_get_line, stream_get_contents	含 buffer 快速路径；stream resource 直传 event-loop
HOOK_FILE	4	文件 IO	IO_WRITE / IO_READ	flock	文件读取请用 fopen + fread，走 HOOK_SOCKET（IO_READ）；file_get_contents 内部走 C API 不被 hook
HOOK_PDO	8	PDO	轮询 DELAY(0) 兜底	PDO::query, PDO::exec, PDO::prepare, PDOStatement::execute	拿不到底层 fd；后续按驱动适配 IO_READ
HOOK_REDIS	16	Redis	轮询 DELAY(0) 兜底	-	未实现
HOOK_DNS	32	DNS	轮询 DELAY(0) 兜底	gethostbyname, gethostbyaddr, checkdnsrr, getmxrr	拿不到底层 fd；后续接入异步 DNS 库
HOOK_CURL	64	cURL	轮询 DELAY(0) 兜底	curl_exec	拿不到底层 fd；后续用 curl_multi 接口实现 IO_READ/IO_WRITE
HOOK_PROC	128	进程	轮询 DELAY(0) 兜底	exec, shell_exec, system, passthru, proc_get_status	子进程由 OS 管理，无法直接监听；轮询让出控制权避免阻塞其他协程
HOOK_ALL	4294967295	全部	-	以上所有	组合使用

调度方式说明：

• IO_READ / IO_WRITE：能拿到 stream resource，直接传给 event-loop 的 onReadable / onWritable，零 CPU 空转
• DELAY(seconds)：精确延时调度，event-loop 的 delay 触发
• 轮询 DELAY(0) 兜底：拿不到底层资源时，反复 delay(0) 让出控制权让其他协程执行。不精确但至少不饿死其他协程

4.2 API

Async::register(Closure $register): void

注册调度回调。当 hook 拦截到阻塞操作时，C 层调用此回调。

回调签名：function (int $handlerType, Fiber $fiber, mixed $data)

handlerType	常量	含义	data 类型
1	HANDLER_TYPE_DELAY	延迟调度	float 秒数
2	HANDLER_TYPE_REPEAT	重复定时器	float 间隔秒数
3	HANDLER_TYPE_IO_READ	IO 可读	stream resource
4	HANDLER_TYPE_IO_WRITE	IO 可写	stream resource
5	HANDLER_TYPE_SIGNAL	信号	int signo

使用者在回调中负责将 Fiber 的恢复注册到自己的 event-loop，事件触发时调用 $fiber->resume()。

Async::await(Closure $callback, int $hooks = 0): int

启动协程。创建 Fiber 并设置 hook 上下文，执行到第一个挂起点返回。

$hooks 是 bitmask，指定哪些阻塞操作被协程化。未标记的阻塞操作保持正常行为。

4.3 环境要求

• PHP >= 8.1 (Fiber)
• NTS 或 ZTS 均可（单线程协程模型）
• C11 编译器
• revolt/event-loop（或其他 event-loop 实现）

5. 开发

5.1 项目结构

src/
├── php_async.h              # 公共头文件（常量、结构体、调度 API）
├── php_async.c              # 主入口（类注册、register/await 方法、模块生命周期）
├── async_fiber.c            # Fiber 管理 + 调度触发
├── async_hook.h             # hook 框架头文件（工厂化接口）
├── async_hook.c             # hook 框架实现（通用替换/恢复工具 + 模块注册表）
├── hooks/
│   ├── hook_sleep.c         # HOOK_SLEEP
│   ├── hook_socket.c        # HOOK_SOCKET（含 buffer 快速路径 + stream resource 传递）
│   ├── hook_file.c          # HOOK_FILE
│   ├── hook_dns.c           # HOOK_DNS
│   ├── hook_curl.c          # HOOK_CURL
│   ├── hook_proc.c          # HOOK_PROC
│   └── hook_pdo.c           # HOOK_PDO（类方法 hook）
├── async.stub.php           # stub（生成 arginfo 用）
├── async_arginfo.h          # 生成的 arginfo
├── config.m4                # Linux 构建配置
└── config.w32               # Windows 构建配置

5.2 添加新的 Hook

Hook 系统采用工厂化设计，每个 tag 的实现独立放在 hooks/hook_xxx.c 中，通过统一接口注册到全局表。

架构

async_hook.h  定义 async_hook_module_t 接口：
  - tag:       ASYNC_HOOK_XXX 常量
  - name:      模块名（调试用）
  - init:      MINIT 时调用，替换目标函数/类方法的 handler
  - shutdown:  MSHUTDOWN 时调用，恢复原始 handler

async_hook.c  提供通用工具 + 全局注册表：
  - async_hook_replace_function / restore_function  替换全局函数
  - async_hook_replace_method / restore_method      替换类方法
  - async_hook_modules[] 注册表 + init_all / shutdown_all

步骤

1. 在 php_async.h 中定义常量

#define ASYNC_HOOK_XXX  (1u << 8)

2. 创建 hooks/hook_xxx.c

#include "php_async.h"
#include "async_hook.h"

static zif_handler orig_func = NULL;

PHP_FUNCTION(async_hook_func)
{
    if (async_should_hook(ASYNC_HOOK_XXX)) {
        async_schedule_delay(0.0); // 或 async_schedule_io_read / io_write
    }
    orig_func(INTERNAL_FUNCTION_PARAM_PASSTHRU);
}

static void async_hook_xxx_init(void)
{
    async_hook_replace_function(ZEND_STRL("func"),
        ZEND_FN(async_hook_func), &orig_func);
}

static void async_hook_xxx_shutdown(void)
{
    async_hook_restore_function(ZEND_STRL("func"), orig_func);
}

const async_hook_module_t async_hook_module_xxx = {
    .tag = ASYNC_HOOK_XXX,
    .name = "xxx",
    .init = async_hook_xxx_init,
    .shutdown = async_hook_xxx_shutdown,
};

3. 在 async_hook.h 中添加 extern 声明

extern const async_hook_module_t async_hook_module_xxx;

4. 在 async_hook.c 的 async_hook_init_all() 中注册

async_hook_register_module(&async_hook_module_xxx);

5. 在 config.m4 和 config.w32 中添加源文件

6. 在 php_async.c 的 MINIT 中注册 PHP 常量

zend_declare_class_constant_long(async_ce, ZEND_STRL("HOOK_XXX"), ASYNC_HOOK_XXX);

7. 更新 async.stub.php

Hook 类方法的示例

替换类方法使用 async_hook_replace_method / async_hook_restore_method（参考 hook_pdo.c）：

static zend_class_entry *redis_ce = NULL;

static void async_hook_redis_init(void)
{
    redis_ce = zend_hash_str_find_ptr(CG(class_table), ZEND_STRL("redis"));
    if (redis_ce) {
        async_hook_replace_method(redis_ce, ZEND_STRL("get"),
            ZEND_FN(async_hook_redis_get), &orig_redis_get);
    }
}

5.3 测试

cd /var/www/sharing
ASYNC_EXT=src/modules/async.so php tests/run_all.php

5.4 编译

cd src
phpize
./configure --enable-async
make -j$(nproc)
cp modules/async.so /usr/local/lib/php/extensions/no-debug-non-zts-20240924/

6. TODO 清单

已实现

• 基础架构（工厂化 hook 框架）
• register / await API
• Fiber 创建/启动/挂起
• 三层调度策略（IO_READ/IO_WRITE → DELAY → DELAY(0) 兜底）
• HOOK_SLEEP：sleep, usleep, time_nanosleep, time_sleep_until（DELAY 精确调度）
• HOOK_SOCKET：fread, fwrite, fgets, stream_select 等（IO_READ/IO_WRITE，含 buffer 快速路径，stream resource 直传）
• HOOK_FILE：flock（IO_WRITE）
• HOOK_DNS：gethostbyname 等（DELAY(0) 兜底）
• HOOK_CURL：curl_exec（DELAY(0) 兜底）
• HOOK_PROC：exec, shell_exec, system, passthru, proc_get_status（轮询 DELAY(0) 兜底）
• HOOK_PDO：PDO::query/exec/prepare, PDOStatement::execute（DELAY(0) 兜底，类方法 hook）
• PIE 支持（composer.json type: php-ext）

待实现

• Fiber 完整生命周期管理（resume 值传递）
• HOOK_SOCKET 完整实现（非阻塞 connect + IO 事件，fsockopen/stream_socket_client）
• HOOK_REDIS：Redis 类方法 hook
• HOOK_PDO 精确调度：pgsql 用 PQsocket 获取 fd 改 IO_READ，mysql 用 mysqlnd 内部 stream
• HOOK_CURL 精确调度：curl_multi 接口实现 IO_READ/IO_WRITE
• HOOK_DNS 精确调度：接入异步 DNS 库（c-ares/libcat）
• HOOK_FILE 精确调度：本地文件用 AIO / 线程池
• stream ops 层替换：覆盖 file_get_contents 等走 C API 的函数
• SIGNAL handler 支持：HOOK_PROC 子进程退出用 SIGCHLD 精确监听，替代轮询