PHP I/O_URING扩展：利用Linux异步I/O接口绕过系统调用阻塞的底层实践

PHP I/O_URING扩展：利用Linux异步I/O接口绕过系统调用阻塞的底层实践

大家好，今天我们要探讨的是一个相当硬核的话题：如何利用Linux的io_uring接口，在PHP中实现真正的异步I/O，并绕过传统阻塞型系统调用带来的性能瓶颈。 这不仅仅是一个简单的扩展开发教程，更是一次深入理解操作系统底层机制和PHP扩展原理的机会。

I/O的演进与困境

在深入io_uring之前，我们先回顾一下I/O的发展历程和PHP在I/O处理上的困境。

• 同步阻塞I/O (Blocking I/O): 这是最传统的I/O模型。应用程序发起I/O请求后，必须等待I/O操作完成才能继续执行。CPU资源被白白浪费在等待上。

• 同步非阻塞I/O (Non-Blocking I/O): 应用程序发起I/O请求后，立即返回。如果数据未准备好，返回一个错误。应用程序需要不断轮询，检查I/O是否完成。虽然避免了阻塞，但轮询消耗大量CPU资源，效率低下。

• I/O多路复用 (I/O Multiplexing): select, poll, epoll等机制允许一个线程同时监听多个文件描述符。当其中一个描述符就绪时，应用程序再发起I/O操作。这是目前PHP中实现并发的主要手段，例如使用stream_select。 尽管改善了并发性能，但本质上仍然是同步I/O，每次I/O操作仍然需要一次系统调用，上下文切换的开销依然存在。

• 异步I/O (Asynchronous I/O): 应用程序发起I/O请求后，立即返回。I/O操作在后台进行，完成后通过某种方式通知应用程序。应用程序无需等待，可以继续执行其他任务。这是理想的I/O模型，可以最大程度地提高CPU利用率。

在PHP中，由于其自身的单线程特性，异步I/O的实现一直是个难题。虽然有一些基于pcntl扩展的多进程方案和基于libevent、libuv的扩展，但它们仍然有各自的局限性，例如多进程带来的内存消耗，或者依赖外部库带来的复杂性。

io_uring：Linux的异步I/O利器

io_uring是Linux内核提供的一种新的异步I/O接口，它相比传统的AIO（Asynchronous I/O）接口，具有更高的性能和更低的延迟。

io_uring的核心思想是使用两个环形缓冲区 (Ring Buffer)：

• Submission Queue (SQ): 应用程序将I/O请求提交到SQ。
• Completion Queue (CQ): 内核将I/O操作的完成事件放到CQ。

应用程序通过共享内存的方式与内核进行通信，避免了频繁的系统调用，从而提高了I/O效率。

io_uring的主要优势：

• 减少系统调用: 应用程序只需提交I/O请求到SQ，然后等待CQ中的完成事件，减少了系统调用的次数。
• 零拷贝: io_uring支持零拷贝操作，避免了数据在内核空间和用户空间之间的复制，进一步提高了性能。
• 支持多种I/O操作: io_uring支持多种I/O操作，包括文件I/O、网络I/O等。
• 高性能: 相比传统的AIO，io_uring具有更高的性能和更低的延迟。

PHP扩展开发基础

在编写PHP io_uring扩展之前，我们需要了解一些PHP扩展开发的基础知识。

• PHP扩展框架: PHP提供了一套扩展开发框架，允许开发者使用C/C++语言编写扩展。
• Zend API: Zend API是PHP的核心API，提供了访问PHP内部数据结构和功能的接口。
• PHP生命周期: 了解PHP的生命周期，包括模块初始化、请求初始化、请求处理、请求结束、模块关闭等阶段，对于编写正确的扩展至关重要。
• 内存管理: PHP使用引用计数进行内存管理。在扩展开发中，需要注意内存的分配和释放，避免内存泄漏。

PHP io_uring扩展的设计与实现

现在我们开始设计和实现一个简单的PHP io_uring扩展。这个扩展将提供以下功能：

• io_uring_init(int $entries): resource: 初始化一个io_uring实例，$entries指定SQ和CQ的大小。
• io_uring_queue_read(resource $ring, int $fd, string $buffer, int $length, int $offset): int: 将一个读操作提交到SQ。
• io_uring_queue_write(resource $ring, int $fd, string $buffer, int $length, int $offset): int: 将一个写操作提交到SQ。
• io_uring_submit(resource $ring): int: 提交SQ中的所有I/O请求。
• io_uring_wait(resource $ring): array: 等待CQ中的完成事件。
• io_uring_close(resource $ring): void: 关闭io_uring实例。

1. 扩展框架搭建

首先，使用ext_skel工具创建一个扩展框架：

./ext_skel --extname=io_uring

这将生成一个名为io_uring的目录，包含扩展的基本文件，例如config.m4, php_io_uring.h, io_uring.c等。

2. 修改config.m4

修改config.m4文件，添加io_uring库的依赖：

PHP_ARG_WITH(io_uring, for io_uring support,
  [--with-io_uring[=DIR]  Include io_uring support])

if test "$PHP_IO_URING" != "no"; then
  AC_MSG_CHECKING([for io_uring library])
  # Check for io_uring header file
  AC_CHECK_HEADER([liburing.h],
    [AC_DEFINE([HAVE_IO_URING], 1, [Whether you have io_uring library])
     PHP_NEW_EXTENSION(io_uring, io_uring.c, $ext_shared)
    ],
    [AC_MSG_ERROR([io_uring library not found])]
  )
  AC_MSG_RESULT([found])
fi

3. 定义数据结构

在php_io_uring.h文件中，定义io_uring实例的数据结构：

#ifndef PHP_IO_URING_H
#define PHP_IO_URING_H

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#include "php.h"
#include "php_ini.h"
#include "ext/standard/info.h"
#include <liburing.h> // 引入liburing头文件

typedef struct {
  struct io_uring ring;
  int entries;
} php_io_uring_t;

extern zend_module_entry io_uring_module_entry;
#define phpext_io_uring_ptr &io_uring_module_entry

#endif  /* PHP_IO_URING_H */

4. 实现扩展函数

在io_uring.c文件中，实现扩展函数：

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#include "php.h"
#include "php_ini.h"
#include "ext/standard/info.h"
#include "php_io_uring.h"

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>
#include <string.h>
#include <errno.h>

/* If you declare any globals in php_io_uring.h uncomment this:
ZEND_DECLARE_MODULE_GLOBALS(io_uring)
*/

/* True global resources - no need for thread safety here */
static int le_io_uring; // Resource ID

/* {{{ PHP_INI
 */
/* Remove comments and fill if you need to have entries in php.ini
PHP_INI_BEGIN()
    STD_PHP_INI_ENTRY("io_uring.global_value",      "42", PHP_INI_ALL, OnUpdateLong,   global_value, zend_io_uring_globals, io_uring_globals)
    STD_PHP_INI_ENTRY("io_uring.global_string", "foobar", PHP_INI_ALL, OnUpdateString, global_string, zend_io_uring_globals, io_uring_globals)
PHP_INI_END()
*/
/* }}} */

/* {{{ proto resource io_uring_init(int $entries)
   Initializes an io_uring instance */
PHP_FUNCTION(io_uring_init)
{
    zend_long entries;
    php_io_uring_t *io_uring_instance;
    zval *resource;

    if (zend_parse_parameters(ZEND_NUM_ARGS(), "l", &entries) == FAILURE) {
        RETURN_FALSE;
    }

    if (entries <= 0 || entries > 4096) { // 限制 entries 的大小
        php_error_docref(NULL, E_WARNING, "entries must be between 1 and 4096");
        RETURN_FALSE;
    }

    io_uring_instance = emalloc(sizeof(php_io_uring_t));
    if (!io_uring_instance) {
        php_error_docref(NULL, E_ERROR, "Failed to allocate memory for io_uring instance");
        RETURN_FALSE;
    }

    memset(io_uring_instance, 0, sizeof(php_io_uring_t));
    io_uring_instance->entries = entries;

    if (io_uring_queue_init(entries, &io_uring_instance->ring, 0) < 0) {
        php_error_docref(NULL, E_ERROR, "Failed to initialize io_uring queue: %s", strerror(errno));
        efree(io_uring_instance);
        RETURN_FALSE;
    }

    resource = zend_resource_create(io_uring_instance, le_io_uring);
    RETURN_RES(resource);
}
/* }}} */

/* {{{ proto int io_uring_queue_read(resource $ring, int $fd, string $buffer, int $length, int $offset)
   Queues a read operation */
PHP_FUNCTION(io_uring_queue_read)
{
    zval *ring_resource;
    zend_long fd, length, offset;
    char *buffer;
    size_t buffer_len;
    php_io_uring_t *io_uring_instance;
    struct io_uring_sqe *sqe;

    if (zend_parse_parameters(ZEND_NUM_ARGS(), "rlsll", &ring_resource, &fd, &buffer, &buffer_len, &length, &offset) == FAILURE) {
        RETURN_FALSE;
    }

    io_uring_instance = (php_io_uring_t *)zend_fetch_resource(Z_RES_P(ring_resource), "io_uring", le_io_uring);
    if (!io_uring_instance) {
        RETURN_FALSE;
    }

    if (length <= 0 || length > buffer_len) {
      php_error_docref(NULL, E_WARNING, "Invalid length: %ld", length);
      RETURN_FALSE;
    }

    sqe = io_uring_get_sqe(&io_uring_instance->ring);
    if (!sqe) {
        php_error_docref(NULL, E_WARNING, "Submission queue is full");
        RETURN_FALSE;
    }

    io_uring_prep_read(sqe, fd, buffer, length, offset);
    io_uring_sqe_set_data(sqe, buffer); // 存储buffer指针，方便在完成事件中释放内存

    RETURN_LONG(0); // 成功加入队列，返回0
}
/* }}} */

/* {{{ proto int io_uring_queue_write(resource $ring, int $fd, string $buffer, int $length, int $offset)
   Queues a write operation */
PHP_FUNCTION(io_uring_queue_write)
{
    zval *ring_resource;
    zend_long fd, length, offset;
    char *buffer;
    size_t buffer_len;
    php_io_uring_t *io_uring_instance;
    struct io_uring_sqe *sqe;

    if (zend_parse_parameters(ZEND_NUM_ARGS(), "rlsll", &ring_resource, &fd, &buffer, &buffer_len, &length, &offset) == FAILURE) {
        RETURN_FALSE;
    }

    io_uring_instance = (php_io_uring_t *)zend_fetch_resource(Z_RES_P(ring_resource), "io_uring", le_io_uring);
    if (!io_uring_instance) {
        RETURN_FALSE;
    }

    if (length <= 0 || length > buffer_len) {
      php_error_docref(NULL, E_WARNING, "Invalid length: %ld", length);
      RETURN_FALSE;
    }

    sqe = io_uring_get_sqe(&io_uring_instance->ring);
    if (!sqe) {
        php_error_docref(NULL, E_WARNING, "Submission queue is full");
        RETURN_FALSE;
    }

    io_uring_prep_write(sqe, fd, buffer, length, offset);
    io_uring_sqe_set_data(sqe, buffer); // 存储buffer指针，方便在完成事件中释放内存

    RETURN_LONG(0); // 成功加入队列，返回0
}
/* }}} */

/* {{{ proto int io_uring_submit(resource $ring)
   Submits queued operations */
PHP_FUNCTION(io_uring_submit)
{
    zval *ring_resource;
    php_io_uring_t *io_uring_instance;
    int ret;

    if (zend_parse_parameters(ZEND_NUM_ARGS(), "r", &ring_resource) == FAILURE) {
        RETURN_FALSE;
    }

    io_uring_instance = (php_io_uring_t *)zend_fetch_resource(Z_RES_P(ring_resource), "io_uring", le_io_uring);
    if (!io_uring_instance) {
        RETURN_FALSE;
    }

    ret = io_uring_submit(&io_uring_instance->ring);
    if (ret < 0) {
        php_error_docref(NULL, E_WARNING, "io_uring_submit failed: %s", strerror(-ret));
        RETURN_FALSE;
    }

    RETURN_LONG(ret); // 返回提交的请求数量
}
/* }}} */

/* {{{ proto array io_uring_wait(resource $ring)
   Waits for completion events */
PHP_FUNCTION(io_uring_wait)
{
    zval *ring_resource;
    php_io_uring_t *io_uring_instance;
    struct io_uring_cqe *cqe;
    int ret;

    if (zend_parse_parameters(ZEND_NUM_ARGS(), "r", &ring_resource) == FAILURE) {
        RETURN_FALSE;
    }

    io_uring_instance = (php_io_uring_t *)zend_fetch_resource(Z_RES_P(ring_resource), "io_uring", le_io_uring);
    if (!io_uring_instance) {
        RETURN_FALSE;
    }

    // io_uring_wait_cqe(&io_uring_instance->ring, &cqe); // 阻塞等待
    ret = io_uring_peek_cqe(&io_uring_instance->ring, &cqe); // 非阻塞检查

    if (ret < 0) {
        if (ret == -EAGAIN) {
            RETURN_EMPTY_ARRAY(); // 没有完成事件，返回空数组
        } else {
            php_error_docref(NULL, E_WARNING, "io_uring_wait_cqe failed: %s", strerror(-ret));
            RETURN_FALSE;
        }

    }

    array_init(return_value);

    if (cqe) {
        add_assoc_long(return_value, "res", cqe->res); // 返回结果
        add_assoc_long(return_value, "flags", cqe->flags); // 返回标志
        char *buf = (char *)io_uring_cqe_get_data(cqe);
        add_assoc_string(return_value, "buffer", buf);
        io_uring_cqe_seen(&io_uring_instance->ring, cqe); // 标记已处理
    }

    // io_uring_cq_advance(&io_uring_instance->ring, 1); // 推进 CQ 索引 (如果使用 io_uring_get_cqe)
    return;
}
/* }}} */

/* {{{ proto void io_uring_close(resource $ring)
   Closes an io_uring instance */
PHP_FUNCTION(io_uring_close)
{
    zval *ring_resource;
    php_io_uring_t *io_uring_instance;

    if (zend_parse_parameters(ZEND_NUM_ARGS(), "r", &ring_resource) == FAILURE) {
        RETURN_FALSE;
    }

    io_uring_instance = (php_io_uring_t *)zend_fetch_resource(Z_RES_P(ring_resource), "io_uring", le_io_uring);
    if (!io_uring_instance) {
        RETURN_FALSE;
    }

    io_uring_queue_exit(&io_uring_instance->ring);
    efree(io_uring_instance);
}
/* }}} */

/* {{{ PHP_MINIT_FUNCTION
 */
PHP_MINIT_FUNCTION(io_uring)
{
    /* If you have INI entries, uncomment these lines
    REGISTER_INI_ENTRIES();
    */

    le_io_uring = zend_register_resource_ex( "io_uring", NULL, ZEND_REGISTERED_RESOURCE_PERSISTENT );
    return SUCCESS;
}
/* }}} */

/* {{{ PHP_MSHUTDOWN_FUNCTION
 */
PHP_MSHUTDOWN_FUNCTION(io_uring)
{
    /* uncomment this line if you have INI entries
    UNREGISTER_INI_ENTRIES();
    */
    return SUCCESS;
}
/* }}} */

/* {{{ PHP_RINIT_FUNCTION
 */
PHP_RINIT_FUNCTION(io_uring)
{
#if defined(COMPILE_DL_IO_URING) && defined(ZTS)
    ZEND_TSRMLS_CACHE_UPDATE();
#endif
    return SUCCESS;
}
/* }}} */

/* {{{ PHP_RSHUTDOWN_FUNCTION
 */
PHP_RSHUTDOWN_FUNCTION(io_uring)
{
    return SUCCESS;
}
/* }}} */

/* {{{ PHP_MINFO_FUNCTION
 */
PHP_MINFO_FUNCTION(io_uring)
{
    php_info_print_table_start();
    php_info_print_table_header(2, "io_uring support", "enabled");
    php_info_print_table_end();

    /* Remove comments if you have entries in php.ini
    DISPLAY_INI_ENTRIES();
    */
}
/* }}} */

/* {{{ io_uring_functions[]
 *
 * Every user visible function must have an entry in io_uring_functions[].
 */
const zend_function_entry io_uring_functions[] = {
    PHP_FE(io_uring_init,      arginfo_io_uring_init)
    PHP_FE(io_uring_queue_read, arginfo_io_uring_queue_read)
    PHP_FE(io_uring_queue_write, arginfo_io_uring_queue_write)
    PHP_FE(io_uring_submit, arginfo_io_uring_submit)
    PHP_FE(io_uring_wait, arginfo_io_uring_wait)
    PHP_FE(io_uring_close, arginfo_io_uring_close)
    PHP_FE_END  /* Must be the last line in io_uring_functions[] */
};
/* }}} */

/* {{{ io_uring_module_entry
 */
zend_module_entry io_uring_module_entry = {
    STANDARD_MODULE_HEADER,
    "io_uring",
    io_uring_functions,
    PHP_MINIT(io_uring),
    PHP_MSHUTDOWN(io_uring),
    PHP_RINIT(io_uring),       /* Replace with NULL if there's nothing to do at request start */
    PHP_RSHUTDOWN(io_uring),     /* Replace with NULL if there's nothing to do at request end */
    PHP_MINFO(io_uring),
    PHP_IO_URING_VERSION,
    STANDARD_MODULE_PROPERTIES
};
/* }}} */

#ifdef COMPILE_DL_IO_URING
#ifdef ZTS
ZEND_TSRMLS_CACHE_DEFINE()
#endif
ZEND_GET_MODULE(io_uring)
#endif

/*
 * Local variables:
 * tab-width: 4
 * c-basic-offset: 4
 * End:
 * vim600: noet sw=4 ts=4 fdm=marker
 * vim<600: noet sw=4 ts=4
 */

5. 添加参数信息

在 io_uring.c 中, 添加以下 arginfo 结构体来定义函数的参数类型和信息：

ZEND_BEGIN_ARG_INFO_EX(arginfo_io_uring_init, 0, 0, 1)
    ZEND_ARG_INFO(0, entries)
ZEND_END_ARG_INFO()

ZEND_BEGIN_ARG_INFO_EX(arginfo_io_uring_queue_read, 0, 0, 5)
    ZEND_ARG_INFO(0, ring)
    ZEND_ARG_INFO(0, fd)
    ZEND_ARG_INFO(0, buffer)
    ZEND_ARG_INFO(0, length)
    ZEND_ARG_INFO(0, offset)
ZEND_END_ARG_INFO()

ZEND_BEGIN_ARG_INFO_EX(arginfo_io_uring_queue_write, 0, 0, 5)
    ZEND_ARG_INFO(0, ring)
    ZEND_ARG_INFO(0, fd)
    ZEND_ARG_INFO(0, buffer)
    ZEND_ARG_INFO(0, length)
    ZEND_ARG_INFO(0, offset)
ZEND_END_ARG_INFO()

ZEND_BEGIN_ARG_INFO_EX(arginfo_io_uring_submit, 0, 0, 1)
    ZEND_ARG_INFO(0, ring)
ZEND_END_ARG_INFO()

ZEND_BEGIN_ARG_INFO_EX(arginfo_io_uring_wait, 0, 0, 1)
    ZEND_ARG_INFO(0, ring)
ZEND_END_ARG_INFO()

ZEND_BEGIN_ARG_INFO_EX(arginfo_io_uring_close, 0, 0, 1)
    ZEND_ARG_INFO(0, ring)
ZEND_END_ARG_INFO()

6. 编译和安装扩展

phpize
./configure --with-io_uring
make
sudo make install

在php.ini文件中启用扩展：

extension=io_uring.so

7. 示例代码

<?php

// 创建一个文件
$filename = "test.txt";
$fd = fopen($filename, "w+");
if (!$fd) {
    die("Failed to open file");
}
$fd_num = intval($fd); // 获取文件描述符

// 初始化 io_uring
$ring = io_uring_init(1024);

// 准备读取缓冲区
$buffer = str_repeat(' ', 1024); // 预分配 1024 字节的缓冲区

// 提交读取请求
io_uring_queue_read($ring, $fd_num, $buffer, 1024, 0);

// 提交请求
io_uring_submit($ring);

// 等待完成事件
$result = io_uring_wait($ring);

if ($result) {
    echo "Read data: " . $result["buffer"] . "n";
    echo "Result code: " . $result["res"] . "n";
} else {
    echo "No completion eventn";
}

// 关闭 io_uring
io_uring_close($ring);

// 关闭文件
fclose($fd);
unlink($filename); // 删除创建的文件
?>

代码解释：

1. 初始化 io_uring: io_uring_init(1024) 初始化了一个包含 1024 个条目的环形缓冲区。

2. 准备读取缓冲区: $buffer = str_repeat(' ', 1024) 创建了一个 1024 字节的字符串，用于存储从文件中读取的数据。

3. 提交读取请求: io_uring_queue_read($ring, $fd_num, $buffer, 1024, 0) 将一个读取请求提交到 io_uring 的提交队列（SQ）。参数包括 io_uring 资源、文件描述符、缓冲区、读取长度和偏移量。

4. 提交请求: io_uring_submit($ring) 将提交队列（SQ）中的所有请求提交到内核。

5. 等待完成事件: io_uring_wait($ring) 等待 io_uring 的完成队列（CQ）中的事件。如果成功，返回一个包含结果信息的数组；否则，返回 false。

6. 处理结果: 如果读取操作成功，则输出读取的数据和结果代码。

7. 关闭 io_uring 和文件: io_uring_close($ring) 关闭 io_uring 实例，fclose($fd) 关闭文件。

遇到的问题与解决方案

在开发和使用io_uring扩展的过程中，可能会遇到一些问题：

• liburing版本问题: 确保安装的liburing版本足够新，能够支持所需的特性。
• 内存管理: 由于PHP的内存管理机制，需要在扩展中谨慎处理内存的分配和释放，避免内存泄漏。可以使用emalloc和efree等Zend API进行内存管理。 尤其是在 io_uring_sqe_set_data 中存储的指针，需要在 io_uring_wait 完成事件后进行释放。
• 错误处理: 在扩展中需要对io_uring的各种操作进行错误处理，例如io_uring_queue_init, io_uring_submit等，并向PHP报告错误信息。
• 并发问题: 由于PHP是单线程的，因此需要避免在扩展中使用阻塞操作。可以使用非阻塞的io_uring_peek_cqe来检查完成事件，或者使用stream_select等机制来实现并发。需要注意的是，即使是单线程，多个请求同时操作同一个io_uring实例也可能导致问题，需要进行适当的同步。

更进一步：高级特性与优化

除了上面介绍的基本功能，io_uring还提供了一些高级特性，可以进一步提高I/O性能：

• 零拷贝I/O: io_uring支持零拷贝I/O，可以避免数据在内核空间和用户空间之间的复制。
• Polled I/O: io_uring支持轮询模式，可以进一步减少系统调用的次数，提高性能。
• Linked Operations: 可以将多个I/O操作链接在一起，形成一个链式操作，减少延迟。
• Registered Buffers: 可以预先注册一些缓冲区，避免每次I/O操作都需要分配新的缓冲区。

在PHP扩展中，可以利用这些高级特性，进一步优化I/O性能。

总结

通过今天的内容，我们了解了io_uring的原理和优势，以及如何在PHP中利用io_uring扩展实现异步I/O。 这种方法可以显著提高PHP应用程序的I/O性能，特别是在处理大量并发I/O请求的场景下。

展望

io_uring作为Linux内核的一项重要技术，在未来将会在更多领域得到应用。 随着PHP的不断发展，相信会有更多基于io_uring的扩展出现，为PHP开发者提供更高效、更灵活的I/O解决方案。 这是一个值得关注和探索的方向。