发表于
字数统计 2.4k 字 | 阅读时长 10 分钟

本篇文章，我们来实战分析一下Swoole PHP协程入口函数的实现原理以及细节。大家需要准备好一份Swoole源码好和我们一起动手操作，我们使用的PHP版本是7.3.12，Swoole的版本是v4.4.16。

首先，我们需要明白PHP协程的入口函数是swoole::PHPCoroutine::main_func，也就是说，我们创建的每一个PHP协程，都会从main_func这个函数开始。

我们现在来逐行分析main_func这个函数。首先，这个函数的参数是void *arg，是一个void类型的指针，从

1
php_coro_args *php_arg = (php_coro_args *) arg;

可以看出arg原来的类型是php_coro_args类型的指针，php_coro_args这个结构体对应的成员如下：

1
2
3
4
5
6
struct php_coro_args
{
    zend_fcall_info_cache *fci_cache;
    zval *argv;
    uint32_t argc;
};

可以发现，这三个成员是我们调用一个函数的基础，分别对应了函数本体、传递给函数的参数、传递给函数参数的个数。

在函数的开头，我们看到了BAILOUT这个东西，我们先来看看如果没有这个东西会有什么问题，我们把commit切一下：

1
2
3
git checkout ef1db99ecfa475ce34d4be744d1f811fadf566ac

git reset HEAD~

此时我们可以看到为支持BAILOUT功能而做出的文件改动。

（现在main_func这个函数名字是旧的名字create_func）

我们会发现，在create_func函数里面，增加了zend_first_try和zend_catch这个结构。我们很容易的想到这是用来捕获PHP异常的。

并且增加了swoole::Coroutine::bailout这个函数，这个函数会在zend_catch里面被调用：

1
2
3
zend_catch {
    Coroutine::bailout([](){ sw_zend_bailout(); });
} zend_end_try();

我们来看看bailout这个函数做了什么事情：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
void Coroutine::bailout(sw_coro_bailout_t func)
{
    if (!func)
    {
        swError("bailout without bailout function");
    }
    if (!current)
    {
        func();
    }
    else
    {
        Coroutine *co = current;
        while (co->origin)
        {
            co = co->origin;
        }
        // it will jump to main context directly (it also breaks contexts)
        on_bailout = func;
        co->yield();
    }
    // expect that never here
    exit(1);
}

其中：

1
2
3
4
if (!current)
{
    func();
}

表示如果不在协程环境，那么直接执行函数func。如果在协程环境，那么执行：

1
2
3
4
5
6
7
8
Coroutine *co = current;
while (co->origin)
{
    co = co->origin;
}
// it will jump to main context directly (it also breaks contexts)
on_bailout = func;
co->yield();

表示查找到主协程，然后再把函数on_bailout = func赋值给swoole::Coroutine::on_bailout这个函数指针。

然后调用co->yield()切换协程上下文。这个时候，就回到非协程环境了。

现在，我们来看两个问题，第一个问题是这里的函数func是什么，第二个问题是on_bailout在哪里被调用了。

首先来看第一个问题。函数func是：

1
sw_zend_bailout();

对应着：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
#define sw_zend_bailout() zend_bailout()

BEGIN_EXTERN_C()
ZEND_API ZEND_COLD void _zend_bailout(const char *filename, uint32_t lineno) /* {{{ */
{

    if (!EG(bailout)) {
        zend_output_debug_string(1, "%s(%d) : Bailed out without a bailout address!", filename, lineno);
        exit(-1);
    }
    gc_protect(1);
    CG(unclean_shutdown) = 1;
    CG(active_class_entry) = NULL;
    CG(in_compilation) = 0;
    EG(current_execute_data) = NULL;
    LONGJMP(*EG(bailout), FAILURE);
}
/* }}} */
END_EXTERN_C()

zend_bailout函数是用来结束程序运行的。我们重点关注LONGJMP(*EG(bailout), FAILURE)做了什么事情。我们知道在C语言里面，有一对函数setjmp和longjmp，分别用来保存当前程序运行的环境和恢复被保存的环境。

我们来看看宏LONGJMP展开来是什么东西：

1
# define LONGJMP(a,b) longjmp(a, b)

就是我们的longjmp函数。

我们可以看一下longjmp这个函数的描述：

1
2
3
Jump to the environment saved in ENV, making the setjmp call there return VAL, or 1 if VAL is 0.

翻译：跳转到在ENV中保存的环境，使setjmp调用在那里返回VAL，如果VAL为0则返回1。

所以，LONGJMP(*EG(bailout), FAILURE)的意图就比较明显了，程序会跳转到*EG(bailout)上下文中，然后使SETJMP调用在那里返回FAILURE，也就是-1。

我们再来看第二个问题。on_bailout是在函数swoole::Coroutine::check_end中被调用的：

1
2
3
4
5
6
7
8
9
10
11
inline void check_end()
{
    if (ctx.is_end())
    {
        close();
    }
    if (unlikely(on_bailout))
    {
        on_bailout();
    }
}

也就是说，当on_bailout这个函数指针不为空的时候，会去调用这个on_bailout，也就是函数sw_zend_bailout。并且，只有当程序逻辑进入了：

1
2
3
zend_catch {
    Coroutine::bailout([](){ sw_zend_bailout(); });
} zend_end_try();

的时候（即程序抛出了异常），这个on_bailout指针才不为空。

现在，我们执行如下命令：

1
git reset --hard a0384ea2981125fc9a7e1a68e489ffb5b40ad426

此时，我们的Swoole就处于没有bailout的样子了。我们重新编译、安装扩展后，编写如下测试脚本：

1
2
3
4
5
6
7
8
9
<?php

register_shutdown_function(function () {
    echo 'shutdown' . PHP_EOL;
});

echo 'execute script' . PHP_EOL;

throw new Exception();

我们执行脚本：

1
2
3
4
5
6
7
8
9
10
11
12
13
[root@592b0366acbf bailout]# php error.php
execute script
PHP Fatal error:  Uncaught Exception in /root/codeDir/phpCode/swoole/coroutine/bailout/error.php:9
Stack trace:
#0 {main}
  thrown in /root/codeDir/phpCode/swoole/coroutine/bailout/error.php on line 9

Fatal error: Uncaught Exception in /root/codeDir/phpCode/swoole/coroutine/bailout/error.php:9
Stack trace:
#0 {main}
  thrown in /root/codeDir/phpCode/swoole/coroutine/bailout/error.php on line 9
shutdown
[root@592b0366acbf bailout]#

可以发现，在脚本结束执行后，会调用我们通过register_shutdown_function函数注册的匿名函数，打印出字符串shutdown。

我们现在修改脚本：

1
2
3
4
5
6
7
8
9
10
11
<?php

register_shutdown_function(function () {
    echo 'shutdown' . PHP_EOL;
});

echo 'execute script' . PHP_EOL;

go(function () {
    throw new Exception;
});

然后执行脚本：

1
2
3
4
5
6
7
8
9
10
11
12
13
[root@592b0366acbf bailout]# php error.php
execute script
PHP Fatal error:  Uncaught Exception in /root/codeDir/phpCode/swoole/coroutine/bailout/error.php:10
Stack trace:
#0 {main}
  thrown in /root/codeDir/phpCode/swoole/coroutine/bailout/error.php on line 10

Fatal error: Uncaught Exception in /root/codeDir/phpCode/swoole/coroutine/bailout/error.php:10
Stack trace:
#0 {main}
  thrown in /root/codeDir/phpCode/swoole/coroutine/bailout/error.php on line 10
/root/php-7.3.12/main/main.c(1414) : Bailed out without a bailout address!
[root@592b0366acbf bailout]#

可以发现，此时我们通过register_shutdown_function函数注册的匿名函数无法被执行。第二个脚本和第一个脚本的区别就是第二个脚本在PHP协程里面抛出了异常。

为什么我们PHP协程里面跑出异常会出现这个问题呢？我们需要跟踪一下程序的执行流程。首先，当我们的程序跑出异常的时候，会使得协程入口函数create_func进入如下代码：

1
2
3
4
if (UNEXPECTED(EG(exception)))
{
    zend_exception_error(EG(exception), E_ERROR);
}

之后，程序就会调用php_error_cb，在这个函数里面打印出异常信息。

之后，程序就会执行如下代码：

1
2
3
4
5
6
7
8
if (type != E_PARSE) {
    /* restore memory limit */
    zend_set_memory_limit(PG(memory_limit));
    efree(buffer);
    zend_objects_store_mark_destructed(&EG(objects_store));
    zend_bailout();
    return;
}

我们发现，这里调用了zend_bailout函数，我们上面分析了这个函数，它会调用LONGJMP，然后把上下文切换到EG(bailout)。因为这里的EG(bailout)是空指针，所以zend_bailout这个函数自然就会报错了。

所以，Swoole的解决方案就是，在create_func这个协程入口函数体里面包了一层try ... catch。

我们切换一下commit：

1
2
3
git checkout ef1db99ecfa475ce34d4be744d1f811fadf566ac

git reset HEAD~

然后重新编译、安装扩展。

编写如下测试脚本：

1
2
3
4
5
6
7
8
9
10
11
<?php

register_shutdown_function(function () {
    echo 'shutdown' . PHP_EOL;
});

echo 'execute script' . PHP_EOL;

go(function () {
    throw new Exception;
});

执行脚本：

1
2
3
4
5
6
7
8
9
10
11
12
13
[root@592b0366acbf bailout]# php error.php
execute script
PHP Fatal error:  Uncaught Exception in /root/codeDir/phpCode/swoole/coroutine/bailout/error.php:10
Stack trace:
#0 {main}
  thrown in /root/codeDir/phpCode/swoole/coroutine/bailout/error.php on line 10

Fatal error: Uncaught Exception in /root/codeDir/phpCode/swoole/coroutine/bailout/error.php:10
Stack trace:
#0 {main}
  thrown in /root/codeDir/phpCode/swoole/coroutine/bailout/error.php on line 10
shutdown
[root@592b0366acbf bailout]#

我们发现，此时register_shutdown_function注册的匿名函数被执行了。也就是说，程序没有因为我们抛出异常使得php cli提早终止执行了。

我们现在来跟踪一下程序的执行流程。

首先，程序会执行zend_first_try，而这个宏会去设置EG(bailout)的地址，这个地址就在create_func里面，并且会保存原来的EG(bailout)地址。

然后，我们的PHP脚本退出时候，程序会执行到以下代码：

1
2
3
4
if (UNEXPECTED(EG(exception)))
{
    zend_exception_error(EG(exception), E_ERROR);
}

然后程序依然会执行函数php_error_cb来打印出异常信息。

然后程序会执行函数zend_bailout：

1
2
3
4
5
6
7
8
9
10
11
12
13
ZEND_API ZEND_COLD void _zend_bailout(const char *filename, uint32_t lineno) /* {{{ */
{
    if (!EG(bailout)) {
        zend_output_debug_string(1, "%s(%d) : Bailed out without a bailout address!", filename, lineno);
        exit(-1);
    }
    gc_protect(1);
    CG(unclean_shutdown) = 1;
    CG(active_class_entry) = NULL;
    CG(in_compilation) = 0;
    EG(current_execute_data) = NULL;
    LONGJMP(*EG(bailout), FAILURE);
    }

因为EG(bailout)不在为空了，所以程序会执行到代码：

1
LONGJMP(*EG(bailout), FAILURE)

此时，程序会回到Swoole内核create_func的zend_first_try宏里面。然后进入zend_catch：

1
2
3
zend_catch {
    Coroutine::bailout([](){ sw_zend_bailout(); });
} zend_end_try();

然后执行函数swoole::Coroutine::bailout的如下代码：

1
2
3
4
5
6
7
8
9
10
11
else
{
    Coroutine *co = current;
    while (co->origin)
    {
        co = co->origin;
    }
    // it will jump to main context directly (it also breaks contexts)
    on_bailout = func;
    co->yield();
}

因为，我们现在就处于第一个协程里面，所以co->origin为nullptr。接着，程序执行代码：

1
on_bailout = func;

实际上就是：

1
on_bailout = sw_zend_bailout;

接着，程序执行co->yield()，切换上下文，此时，就会到非协程环境。然后程序就执行到了函数swoole::Coroutine::run里面的check_end()这个位置：

1
2
3
4
5
6
7
8
9
inline long run()
{
    long cid = this->cid;
    origin = current;
    current = this;
    ctx.swap_in();
    check_end();
    return cid;
}

我们前面分析过，check_end这个函数会去调用on_bailout回调函数，也就是sw_zend_bailout。而sw_zend_bailout对应着zend_bailout：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
ZEND_API ZEND_COLD void _zend_bailout(const char *filename, uint32_t lineno) /* {{{ */
{

    if (!EG(bailout)) {
        zend_output_debug_string(1, "%s(%d) : Bailed out without a bailout address!", filename, lineno);
        exit(-1);
    }
    gc_protect(1);
    CG(unclean_shutdown) = 1;
    CG(active_class_entry) = NULL;
    CG(in_compilation) = 0;
    EG(current_execute_data) = NULL;
    LONGJMP(*EG(bailout), FAILURE);
}

此时的EG(bailout)地址已经不再是create_func函数的zend_first_try里面了，而是zend_first_try中的__orig_bailout。

所以，当程序执行完LONGJMP(*EG(bailout), FAILURE)之后，就会回到php cli的php_execute_script函数的zend_try里面：

1
2
3
4
5
6
7
8
zend_try {
    char realfile[MAXPATHLEN];

#ifdef PHP_WIN32
    if(primary_file->filename) {
        UpdateIniFromRegistry((char*)primary_file->filename);
    }
#endif

接着，php cli程序可以顺利的回到do_cli函数的：

1
2
3
4
out:
if (request_started) {
    php_request_shutdown((void *) 0);
}

这个位置。

执行完函数php_request_shutdown之后，我们在PHP脚本里面通过register_shutdown_function函数注册的匿名函数就会被执行。最终，php cli程序正常退出。

总结

Swoole内核通过zend_first_try保存当前地址到EG(bailout)，使得程序在抛出异常的时候，程序先回到Swoole内核的zend_first_try里面。然后，恢复EG(bailout)为__orig_bailout。最终，使得php cli程序正常退出。

发表于
字数统计 265 字 | 阅读时长 1 分钟

我们本章的目标是通过Protobuf让客户端和服务器进行通信，需要用到Swoole扩展，版本是v4.4.16。

我们安装Swoole官方的IDE helper：

1
[root@592b0366acbf grpc-demo]# composer require swoole/ide-helper --dev

我们现在创建目录RPC：

1
[root@592b0366acbf grpc-demo]# mkdir RPC

然后再创建文件RPC/Server.php：

1
[root@592b0366acbf grpc-demo]# touch RPC/Server.php

内容如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
<?php

namespace RPC;

use Message\User;
use Swoole\Coroutine\Server as CoroutineServer;
use Swoole\Coroutine\Server\Connection;

class Server extends CoroutineServer
{
    public function __construct($host, $port)
    {
        parent::__construct($host, $port);
    }

    public function handler(Connection $conn)
    {
        while(true) {
            $data = $conn->recv();
            if ($data === '') {
                echo 'the client is disconnected' . PHP_EOL;
                break;
            } else if ($data === false) {
                echo socket_strerror($this->errCode);
            }
            $user = new User();
            $user->mergeFromString($data);
            echo $user->getId() . PHP_EOL;
            echo $user->getAge() . PHP_EOL;
            echo $user->getUsername() . PHP_EOL;
        }
    }

    public function start(): bool
    {
        $this->handle([$this, 'handler']);
        parent::start();
        return true;
    }
}

然后我们为composer.json增加对应的autoload：

1
2
3
4
5
6
7
"autoload": {
    "psr-4": {
        "Message\\": "Protobuf/Message",
        "GPBMetadata\\": "Protobuf/GPBMetadata",
        "RPC\\": "RPC"
    }
}

我们为RPC命名空间增加了一项。

接着执行命令：

1
[root@592b0366acbf grpc-demo]# composer dump-autoload

然后创建文件

发表于
字数统计 449 字 | 阅读时长 2 分钟

需要提前准备好protoc这个工具，我们可以在github上面下载源码然后进行编译安装：

1
2
3
4
5
git clone https://github.com/grpc/grpc.git

cd grpc

make grpc_php_plugin

因为编译安装的过程会因为每个人的编译工具的不同而不同，且这个过程可能会遇到很多的问题，所以安装工具的过程就不多展开来说了。（编译的过程中如果遇到问题，可以在评论区进行提问）

OK，我们现在开始使用Protobuf。

首先，创建目录grpc-demo并且进入目录grpc-demo：

1
2
3
mkdir grpc-demo

cd grpc-demo/

然后我们通过composer来初始化我们的项目：

1
[root@592b0366acbf grpc-demo]# composer init

（出现了交互的信息直接回车就行）

然后安装google/protobuf依赖：

1
[root@592b0366acbf grpc-demo]# composer require google/protobuf

然后，我们编写我们的协议（文件名字是user.proto）：

1
2
3
4
5
6
7
8
9
syntax = "proto3";

package message;

message User {
    uint32 id = 1;
    string username = 2;
    uint32 age = 3;
}

OK，现在我们就可以通过protoc工具来生成对应的协议解析代码了：

1
2
3
[root@592b0366acbf grpc-demo]# mkdir Protobuf

[root@592b0366acbf grpc-demo]# protoc --php_out=Protobuf/ user.proto

执行完这条命令之后，Protobuf目录下面就会生成协议解析的代码了。

然后我们为composer.json增加autoload：

1
2
3
4
5
6
"autoload": {
    "psr-4": {
        "Message\\": "Protobuf/Message",
        "GPBMetadata\\": "Protobuf/GPBMetadata"
    }
}

然后执行命令：

1
[root@592b0366acbf grpc-demo]# composer dump-autoload

接着，我们创建测试脚本index.php：

1
[root@592b0366acbf grpc-demo]# touch index.php

然后文件内容为：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
<?php

use Message\User;

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

$user1 = new User();
$user2 = new User();

$user1->setId(10086);
$user1->setUsername('codinghuang');
$user1->setAge(22);
$data = $user1->serializeToString();
var_dump($data);

$user2->mergeFromString($data);
var_dump($user2->getId());
var_dump($user2->getUsername());
var_dump($user2->getAge());

执行结果如下：

1
2
3
4
5
6
[root@592b0366acbf grpc-demo]# php index.php
string(18) �N
             codinghuang"
int(10086)
string(11) "codinghuang"
int(22)

发表于
字数统计 4.5k 字 | 阅读时长 21 分钟

前言

在Swoole4.5版本中（目前还未发布），我们的Server有一个性能需要优化的地方，就是worker进程在收到master进程发来的包的时候，需要进行两次的拷贝，才可以把数据从PHP扩展层传递到PHP上层（也就是我们事件回调函数需要拿到的data）。

优化前

我们先来分析一下为什么会有性能的问题。首先，我们需要一份会有性能问题的代码。我们git clone下swoole-src代码，然后git checkout到8235c82fea2130534a16fd20771dcab3408a763e这个commit位置：

1
git checkout 8235c82fea2130534a16fd20771dcab3408a763e

我们来分析一下代码，首先看master进程是如何封装数据然后发送给worker进程的。在函数process_send_packet里面，我们看核心的地方：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
static int process_send_packet(swServer *serv, swPipeBuffer *buf, swSendData *resp, send_func_t _send, void* private_data)
{
    const char* data = resp->data;
    uint32_t send_n = resp->info.len;
    off_t offset = 0;

    uint32_t max_length = serv->ipc_max_size - sizeof(buf->info);

    if (send_n <= max_length)
    {
        buf->info.flags = 0;
        buf->info.len = send_n;
        memcpy(buf->data, data, send_n);

        int retval = _send(serv, buf, sizeof(buf->info) + send_n, private_data);
        return retval;
    }

    buf->info.flags = SW_EVENT_DATA_CHUNK;

    while (send_n > 0)
    {
        if (send_n > max_length)
        {
            buf->info.len = max_length;
        }
        else
        {
            buf->info.flags |= SW_EVENT_DATA_END;
            buf->info.len = send_n;
        }

        memcpy(buf->data, data + offset, buf->info.len);

        if (_send(serv, buf, sizeof(buf->info) + buf->info.len, private_data) < 0)
        {
            return SW_ERR;
        }

        send_n -= buf->info.len;
        offset += buf->info.len;
    }

    return SW_OK;
}

首先，我们来说一下process_send_packet这个函数的参数：

其中，

swServer *serv就是我们创建的那个Server。

swPipeBuffer *buf指向的内存里面的数据需要发送给worker进程。

swSendData *resp里面存放了master进程收到的客户端数据以及一个swDataHead info头部。

_send是一个回调函数，这里面的逻辑就是master进程把swPipeBuffer *buf里面的数据发送给worker进程。

void* private_data这里是一个swWorker *worker类型的指针转换过来的。指定了master进程需要发送的那个worker进程。

说明一点，这里我们是以Server设置了eof选项为例子讲解的（假设设置了"\r\n"）。因为TCP是面向字节流的，即使客户端发送了一个很大的包过来，服务器一次read出来的数据也不见得非常大。如果不设置eof的话，是不会导致我们这篇文章所说的性能问题。

介绍完了process_send_packet函数的参数之后，我们来看看代码是如何实现的：

1
const char* data = resp->data;

首先，让data指向resp->data，也就是客户端发来的实际数据。例如，客户端发来了字符串hello world\r\n，那么data里面存放的就是hello world\r\n。

1
uint32_t send_n = resp->info.len;

标志着resp->data数据的长度。例如，客户端往服务器发送了1M的数据，那么resp->info.len就是1048576。

1
off_t offset = 0;

用来标志哪些数据master进程已经发送给了worker进程。

1
uint32_t max_length = serv->ipc_max_size - sizeof(buf->info);

max_length表示master进程一次往worker进程发送的包最大长度。

注意：master进程和worker进程是通过udg方式进行通信的。所以，master进程发送多少，worker进程就直接收多少

1
2
3
4
5
6
7
8
9
if (send_n <= max_length)
{
    buf->info.flags = 0;
    buf->info.len = send_n;
    memcpy(buf->data, data, send_n);

    int retval = _send(serv, buf, sizeof(buf->info) + send_n, private_data);
    return retval;
}

如果master进程要发给worker进程的数据小于max_length，那么就直接调用_send函数，直接把数据发给worker进程。

1
buf->info.flags = SW_EVENT_DATA_CHUNK;

当send_n大于max_length的时候，设置buf->info.flags为CHUNK，也就意味着需要把客户端发来的数据先拆分成一小段一小段的数据，然后再发送给worker进程。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
while (send_n > 0)
{
    if (send_n > max_length)
    {
        buf->info.len = max_length;
    }
    else
    {
        buf->info.flags |= SW_EVENT_DATA_END;
        buf->info.len = send_n;
    }

    memcpy(buf->data, data + offset, buf->info.len);

    if (_send(serv, buf, sizeof(buf->info) + buf->info.len, private_data) < 0)
    {
        return SW_ERR;
    }

    send_n -= buf->info.len;
    offset += buf->info.len;
}

逻辑比较简单，就是一个分段发送的过程。这里需要注意的两点：

1
2
1、buf->info.len的长度需要更新为小段的chunk的长度，而不是大数据包的长度
2、最后一个chunk的info.flags需要变成SW_EVENT_DATA_END，意味着一个完整的包已经发完了

OK，分析完了master进程发包的过程，我们来分析一下worker进程收包的过程。

我们先看一下函数swWorker_onPipeReceive：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
static int swWorker_onPipeReceive(swReactor *reactor, swEvent *event)
{
    swServer *serv = (swServer *) reactor->ptr;
    swFactory *factory = &serv->factory;
    swPipeBuffer *buffer = serv->pipe_buffers[0];
    int ret;

    _read_from_pipe:

    if (read(event->fd, buffer, serv->ipc_max_size) > 0)
    {
        ret = swWorker_onTask(factory, (swEventData *) buffer);
        if (buffer->info.flags & SW_EVENT_DATA_CHUNK)
        {
            //no data
            if (ret < 0 && errno == EAGAIN)
            {
                return SW_OK;
            }
            else if (ret > 0)
            {
                goto _read_from_pipe;
            }
        }
        return ret;
    }

    return SW_ERR;
}

这个就是worker进程接收master进程发来的数据的代码。

我们看的，worker进程会直接把数据先读取到buffer内存里面，然后调用swWorker_onTask。我们再来看看swWorker_onTask函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
int swWorker_onTask(swFactory *factory, swEventData *task)
{
    swServer *serv = (swServer *) factory->ptr;
    swWorker *worker = SwooleWG.worker;

    //worker busy
    worker->status = SW_WORKER_BUSY;
    //packet chunk
    if (task->info.flags & SW_EVENT_DATA_CHUNK)
    {
        if (serv->merge_chunk(serv, task->info.reactor_id, task->data, task->info.len) < 0)
        {
            swoole_error_log(SW_LOG_WARNING, SW_ERROR_SESSION_DISCARD_DATA,
                    "cannot merge chunk to worker buffer, data[fd=%d, size=%d] lost", task->info.fd, task->info.len);
            return SW_OK;
        }
        //wait more data
        if (!(task->info.flags & SW_EVENT_DATA_END))
        {
            return SW_OK;
        }
    }

    switch (task->info.type)
    {
    case SW_SERVER_EVENT_SEND_DATA:
        //discard data
        if (swWorker_discard_data(serv, task) == SW_TRUE)
        {
            break;
        }
        swWorker_do_task(serv, worker, task, serv->onReceive);
        break;
    // 省略其他的case
    default:
        swWarn("[Worker] error event[type=%d]", (int )task->info.type);
        break;
    }

    //worker idle
    worker->status = SW_WORKER_IDLE;

    //maximum number of requests, process will exit.
    if (!SwooleWG.run_always && worker->request_count >= SwooleWG.max_request)
    {
        swWorker_stop(worker);
    }
    return SW_OK;
}

我们重点看看性能问题代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
if (task->info.flags & SW_EVENT_DATA_CHUNK)
{
    if (serv->merge_chunk(serv, task->info.reactor_id, task->data, task->info.len) < 0)
    {
        swoole_error_log(SW_LOG_WARNING, SW_ERROR_SESSION_DISCARD_DATA,
                "cannot merge chunk to worker buffer, data[fd=%d, size=%d] lost", task->info.fd, task->info.len);
        return SW_OK;
    }
    //wait more data
    if (!(task->info.flags & SW_EVENT_DATA_END))
    {
        return SW_OK;
    }
}

这里，worker进程会先判断master发来的数据是否是CHUNK数据，如果是，那么会进行merge_chunk的操作。我们看看merge_chunk对应的函数：

1
2
3
4
5
6
static int swServer_worker_merge_chunk(swServer *serv, int key, const char *data, size_t len)
{
    swString *package = swServer_worker_get_input_buffer(serv, key);
    //merge data to package buffer
    return swString_append_ptr(package, data, len);
}

我们会先根据key的值（实际上是reactor线程的id），获取一块全局的内存，然后把接收到的chunk数据，追加到这个全局的内存上面，而swString_append_ptr执行的就是memcpy的操作。

所以，这就是一个性能问题了。worker进程接收到的所有数据都会被完整的拷贝一遍。如果客户端发来的数据很大，这个拷贝的开销也是很大的。

我们再看看Swoole内核是如何把data提供给PHP应用层的，主要函数是：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
void php_swoole_get_recv_data(swServer *serv, zval *zdata, swEventData *req, char *header, uint32_t header_length)
{
    char *data = NULL;

    size_t length = serv->get_packet(serv, req, &data);
    if (header_length >= length)
    {
        ZVAL_EMPTY_STRING(zdata);
    }
    else
    {
        ZVAL_STRINGL(zdata, data + header_length, length - header_length);
    }
    if (header_length > 0)
    {
        memcpy(header, data, header_length);
    }
}

程序会进入以下代码：

1
ZVAL_STRINGL(zdata, data + header_length, length - header_length);

这个地方是通过ZVAL_STRINGL来创建zend_string的，也就意味着PHP底层会把Swoole内核中存储的data完整的拷贝一份到zend_string里面。然后再让zdata（也就是PHP应用层会用到的data）的zend_value指针指向这个zend_string。这里，又多了一次完整的内存拷贝。

上述过程我们可以通过下面这幅图总结：

优化前

因此，我们对这部分合并的代码进行了一个优化。我们让worker进程在接收master进程的数据之前，就准备好一块足够大的内存，然后直接用这块内存把master进程发来的数据接收即可。

优化一

我们先更新一下swoole-src的源码：

1
git checkout 529ad44d578930b3607abedcfc278364df34bc73

我们依旧先看看process_send_packet函数的代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
static int process_send_packet(swServer *serv, swPipeBuffer *buf, swSendData *resp, send_func_t _send, void* private_data)
{
    const char* data = resp->data;
    uint32_t send_n = resp->info.len;
    off_t offset = 0;
    uint32_t copy_n;

    uint32_t max_length = serv->ipc_max_size - sizeof(buf->info);

    if (send_n <= max_length)
    {
        buf->info.flags = 0;
        buf->info.len = send_n;
        memcpy(buf->data, data, send_n);

        int retval = _send(serv, buf, sizeof(buf->info) + send_n, private_data);
        return retval;
    }

    buf->info.flags = SW_EVENT_DATA_CHUNK;
    buf->info.len = send_n;

    while (send_n > 0)
    {
        if (send_n > max_length)
        {
            copy_n = max_length;
        }
        else
        {
            buf->info.flags |= SW_EVENT_DATA_END;
            copy_n = send_n;
        }

        memcpy(buf->data, data + offset, copy_n);

        swTrace("finish, type=%d|len=%d", buf->info.type, copy_n);

        if (_send(serv, buf, sizeof(buf->info) + copy_n, private_data) < 0)
        {
            return SW_ERR;
        }

        send_n -= copy_n;
        offset += copy_n;
    }

    return SW_OK;
}

我们聚焦修改的地方，主要是对CHUNK的处理：

1
2
buf->info.flags = SW_EVENT_DATA_CHUNK;
buf->info.len = send_n;

我们发现，buf->info.len的长度不是每个小段chunk的长度了，而是整个大包的长度了。为什么可以这样做呢？因为master进程与worker进程是通过udg进行通信的，所以，worker进程在调用recv的时候，返回值实际上就是chunk的长度了，所以buf->info.len里面存储chunk的长度没有必要。

其他地方的逻辑和之前的代码没有区别。

我们再来看看worker进程是如何接收master进程发来的数据的。在函数swWorker_onPipeReceive里面：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
static int swWorker_onPipeReceive(swReactor *reactor, swEvent *event)
{
    int ret;
    ssize_t recv_n = 0;
    swServer *serv = (swServer *) reactor->ptr;
    swFactory *factory = &serv->factory;
    swPipeBuffer *pipe_buffer = serv->pipe_buffers[0];
    void *buffer;
    struct iovec buffers[2];

    // peek
    recv_n = recv(event->fd, &pipe_buffer->info, sizeof(pipe_buffer->info), MSG_PEEK);
    if (recv_n < 0 && errno == EAGAIN)
    {
        return SW_OK;
    }
    else if (recv_n < 0)
    {
        return SW_ERR;
    }

    if (pipe_buffer->info.flags & SW_EVENT_DATA_CHUNK)
    {
        buffer = serv->get_buffer(serv, &pipe_buffer->info);
        _read_from_pipe:

        buffers[0].iov_base = &pipe_buffer->info;
        buffers[0].iov_len = sizeof(pipe_buffer->info);
        buffers[1].iov_base = buffer;
        buffers[1].iov_len = serv->ipc_max_size - sizeof(pipe_buffer->info);

        recv_n = readv(event->fd, buffers, 2);
        if (recv_n < 0 && errno == EAGAIN)
        {
            return SW_OK;
        }
        if (recv_n > 0)
        {
            serv->add_buffer_len(serv, &pipe_buffer->info, recv_n - sizeof(pipe_buffer->info));
        }

        if (pipe_buffer->info.flags & SW_EVENT_DATA_CHUNK)
        {
            //wait more chunk data
            if (!(pipe_buffer->info.flags & SW_EVENT_DATA_END))
            {
                goto _read_from_pipe;
            }
            else
            {
                pipe_buffer->info.flags |= SW_EVENT_DATA_OBJ_PTR;
                /**
                 * Because we don't want to split the swEventData parameters into swDataHead and data,
                 * we store the value of the worker_buffer pointer in swEventData.data.
                 * The value of this pointer will be fetched in the swServer_worker_get_packet function.
                 */
                serv->copy_buffer_addr(serv, pipe_buffer);
            }
        }
    }
    else
    {
        recv_n = read(event->fd, pipe_buffer, serv->ipc_max_size);
    }

    if (recv_n > 0)
    {
        ret = swWorker_onTask(factory, (swEventData *) pipe_buffer, recv_n - sizeof(pipe_buffer->info));
        return ret;
    }

    return SW_ERR;
}

其中，

1
2
3
4
5
6
7
8
9
recv_n = recv(event->fd, &pipe_buffer->info, sizeof(pipe_buffer->info), MSG_PEEK);
if (recv_n < 0 && errno == EAGAIN)
{
    return SW_OK;
}
else if (recv_n < 0)
{
    return SW_ERR;
}

我们先对内核缓冲区里面的数据进行一次peek操作，来获取到head部分。这样我们就知道数据是否是以CHUNK方式发来的了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
if (pipe_buffer->info.flags & SW_EVENT_DATA_CHUNK)
{
    buffer = serv->get_buffer(serv, &pipe_buffer->info);
    _read_from_pipe:

    buffers[0].iov_base = &pipe_buffer->info;
    buffers[0].iov_len = sizeof(pipe_buffer->info);
    buffers[1].iov_base = buffer;
    buffers[1].iov_len = serv->ipc_max_size - sizeof(pipe_buffer->info);

    recv_n = readv(event->fd, buffers, 2);
    if (recv_n < 0 && errno == EAGAIN)
    {
        return SW_OK;
    }
    if (recv_n > 0)
    {
        serv->add_buffer_len(serv, &pipe_buffer->info, recv_n - sizeof(pipe_buffer->info));
    }

    if (pipe_buffer->info.flags & SW_EVENT_DATA_CHUNK)
    {
        //wait more chunk data
        if (!(pipe_buffer->info.flags & SW_EVENT_DATA_END))
        {
            goto _read_from_pipe;
        }
        else
        {
            pipe_buffer->info.flags |= SW_EVENT_DATA_OBJ_PTR;
            /**
                * Because we don't want to split the swEventData parameters into swDataHead and data,
                * we store the value of the worker_buffer pointer in swEventData.data.
                * The value of this pointer will be fetched in the swServer_worker_get_packet function.
                */
            serv->copy_buffer_addr(serv, pipe_buffer);
        }
    }
}

如果是CHUNK方式发来的数据，那么我们执行如下的操作：

1
buffer = serv->get_buffer(serv, &pipe_buffer->info);

get_buffer是一个回调函数，对应：

1
2
3
4
5
6
7
8
9
10
11
static void* swServer_worker_get_buffer(swServer *serv, swDataHead *info)
{
    swString *worker_buffer = swServer_worker_get_input_buffer(serv, info->reactor_id);

    if (worker_buffer->size < info->len)
    {
        swString_extend(worker_buffer, info->len);
    }

    return worker_buffer->str + worker_buffer->length;
}

这里，我们会先判断这块全局的buffer是否足够的大，可以接收完整个大包，如果不够大，我们扩容到足够的大。

1
2
3
4
5
6
7
8
_read_from_pipe:

buffers[0].iov_base = &pipe_buffer->info;
buffers[0].iov_len = sizeof(pipe_buffer->info);
buffers[1].iov_base = buffer;
buffers[1].iov_len = serv->ipc_max_size - sizeof(pipe_buffer->info);

recv_n = readv(event->fd, buffers, 2);

然后，我们调用readv，把head和实际的数据分别存在了两个地方。这么做是避免为了把head和实际的数据做拆分而导致的内存拷贝。

通过以上方式，Swoole Server减少了一次内存拷贝。

上述过程我们可以通过下面这幅图总结：

第一次优化

从图中我们可以看出，步骤2到步骤3这里还是会有一次完整的拷贝，我们也把它给优化掉了。我们来看优化后的代码。

优化二

我们先更新一下swoole-src的源码：

1
git checkout 5278bb30c9b6b84753fa1950cef3226f1cfb515c

master进程发送数据到worker进程的代码没有变化，主要是worker进程这边对接收buffer处理的变化。我们会发现函数swWorker_onPipeReceive没有任何改动，改动的是处理buffer的几个回调函数。我们一一来看下。

首先是函数指针swServer::get_buffer对应了函数php_swoole_server_worker_get_buffer：

1
2
3
4
5
6
7
8
9
10
11
12
13
static void* php_swoole_server_worker_get_buffer(swServer *serv, swDataHead *info)
{
    zend_string *worker_buffer = php_swoole_server_worker_get_input_buffer(serv, info->reactor_id);

    if (worker_buffer == NULL)
    {
        worker_buffer = zend_string_alloc(info->len, 0);
        worker_buffer->len = 0;
        php_swoole_server_worker_set_buffer(serv, info, worker_buffer);
    }

    return worker_buffer->val + worker_buffer->len;
}

这里先调用函数php_swoole_server_worker_get_input_buffer来获取接收master进程发来数据的buffer。如果说没有获取到，那么说明我们之前的worker_buffer没有创建或者接收完所有的数据之后被销毁了，此时我们需要通过函数zend_string_alloc分配一块内存。这里需要注意的一个地方就是，info->len是master进程发送给worker进程的总长度，也就意味着我们需要把zend_string的len手动初始化为0。毕竟zend_string没有offest这个成员，所以这里我们只能够把len当作offset来用了。

获取到zend_string这块worker buffer之后，我们就可以通过readv来读取master进程发送给worker进程的数据了。

获取完数据之后，我们调用swServer::add_buffer_len函数指针对应的php_swoole_server_worker_add_buffer_len这个函数来增加偏移量：

1
2
3
4
5
static void php_swoole_server_worker_add_buffer_len(swServer *serv, swDataHead *info, size_t len)
{
    zend_string *worker_buffer = php_swoole_server_worker_get_input_buffer(serv, info->reactor_id);
    worker_buffer->len += len;
}

当我们接收完master进程发送过来的所有数据之后，我们调用swServer::copy_buffer_addr函数指针对应的php_swoole_server_worker_copy_buffer_addr来把zend_string的地址拷贝到swPipeBuffer::data里面。

这样，我们通过函数指针的简单替换，实现了C层面的buffer到PHP层面的buffer切换。

最后，我们需要把接收到的数据，也就是zend_string里面的数据提供给PHP应用层。我们来看看swoole_websocket_onMessage这个函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
int swoole_websocket_onMessage(swServer *serv, swEventData *req)
{
    int fd = req->info.fd;
    uchar flags = 0;
    zend_long opcode = 0;

    zval zdata;
    char frame_header[2];
    memcpy(frame_header, &req->info.ext_flags, sizeof(frame_header));

    php_swoole_get_recv_data(serv, &zdata, req);

    // frame info has already decoded in swWebSocket_dispatch_frame
    flags  = frame_header[0];
    opcode = frame_header[1];

    if (opcode == WEBSOCKET_OPCODE_CLOSE && !serv->listen_list->open_websocket_close_frame)
    {
        zval_ptr_dtor(&zdata);
        return SW_OK;
    }

#ifdef SW_HAVE_ZLIB
    /**
     * RFC 7692
     */
    if (serv->websocket_compression && (flags & SW_WEBSOCKET_FLAG_RSV1))
    {
        swString_clear(swoole_zlib_buffer);
        if (!websocket_message_uncompress(swoole_zlib_buffer, Z_STRVAL(zdata), Z_STRLEN(zdata)))
        {
            zval_ptr_dtor(&zdata);
            return SW_OK;
        }
        zval_ptr_dtor(&zdata);
        ZVAL_STRINGL(&zdata, swoole_zlib_buffer->str, swoole_zlib_buffer->length);
        flags ^= (SW_WEBSOCKET_FLAG_RSV1 | SW_WEBSOCKET_FLAG_COMPRESS);
    }
#endif

    zend_fcall_info_cache *fci_cache = php_swoole_server_get_fci_cache(serv, req->info.server_fd, SW_SERVER_CB_onMessage);
    zval args[2];

    args[0] = *(zval *) serv->ptr2;
    php_swoole_websocket_construct_frame(&args[1], opcode, Z_STRVAL(zdata), Z_STRLEN(zdata), flags);
    zend_update_property_long(swoole_websocket_frame_ce, &args[1], ZEND_STRL("fd"), fd);

    if (UNEXPECTED(!zend::function::call(fci_cache, 2, args, NULL, SwooleG.enable_coroutine)))
    {
        php_swoole_error(E_WARNING, "%s->onMessage handler error", ZSTR_VAL(swoole_websocket_server_ce->name));
        serv->close(serv, fd, 0);
    }

    zval_ptr_dtor(&zdata);
    zval_ptr_dtor(&args[1]);

    return SW_OK;
}

其中，php_swoole_get_recv_data这个函数就是用来获取zend_string数据的，我们分析下代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
void php_swoole_get_recv_data(swServer *serv, zval *zdata, swEventData *req)
{
    char *data = NULL;
    zend_string *worker_buffer;

    size_t length = serv->get_packet(serv, req, &data);
    if (length == 0)
    {
        ZVAL_EMPTY_STRING(zdata);
    }
    else
    {
        if (req->info.flags & SW_EVENT_DATA_OBJ_PTR)
        {
            worker_buffer = (zend_string *) (data - XtOffsetOf(zend_string, val));
            ZVAL_STR(zdata, worker_buffer);
        }
        else
        {
            ZVAL_STRINGL(zdata, data, length);
        }
    }
}

因为在swWorker_onPipeReceive函数里面把req->info.flags设置为了SW_EVENT_DATA_OBJ_PTR，所以函数会执行以下代码：

1
2
3
4
5
if (req->info.flags & SW_EVENT_DATA_OBJ_PTR)
{
    worker_buffer = (zend_string *) (data - XtOffsetOf(zend_string, val));
    ZVAL_STR(zdata, worker_buffer);
}

其中，zdata就是PHP应用层会使用到的data，而ZVAL_STR是让zdata里面的zend_value直接指向了worker_buffer，没有任何的内存拷贝。

通过以上方式，Swoole Server再次减少了一次内存拷贝。

上述过程我们可以通过下面这幅图总结：

第二次优化

最终，我们把4次内存拷贝降低到了1次，因此onMessage回调函数性能提升了4倍。

性能对比

我们的压测程序如下，Server代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
<?php

use Swoole\WebSocket\Server;

$start = microtime(true);

$server = new Server("0.0.0.0", 9501);

$server->set([
    "worker_num" => 1,
    'package_max_length' => 1024 * 1024 * 4,
]);

$server->on('open', function (Server $server, $request) {
});

$server->on('message', function (Server $server, $frame) use ($start) {
    if ($frame->data === "end") {
        $end = microtime(true);
        var_dump(($end - $start) * 1000);
    }
});

$server->on('close', function ($ser, $fd) {
    echo "client {$fd} closed\n";
});

$server->start();

Client代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
<?php

use Swoole\Coroutine;
use Swoole\Coroutine\Http\Client;

use function Co\run;

run(function () {
    $cli = new Client("127.0.0.1", 9501);
    $ret = $cli->upgrade("/websocket");

    if (!$ret) {
        echo "ERROR\n";
        return;
    }

    for ($i = 0; $i < 2000; $i++) {
        $cli->push(str_repeat('a', 0.5 * 1024 * 1024));
    }

    $cli->push("end");
    sleep(100000);
});

内存拷贝：

内存拷贝

CPU使用率：

CPU使用率

总结

  1. 本次优化主要是通过减少内存拷贝来提升Server的性能，这也是服务器优化的重点。

  2. 本次优化我们通过设计Buffer的接口，来达到切换Swoole内核层Buffer和PHP层Buffer的目的。首先，通过调用get_buffer来获取到接收数据的zend_string（接收第一个Chunk前，都会分配一个新的zend_string，因为PHP底层会通过内存池来管理，所以分配zend_string的性能开销不用太担心）；当接收完数据的时候，调用add_buffer_len来更新zend_string的len，也就是我们的offset；当我们接收完所有的Chunk之后，调用copy_buffer_addr来保存zend_string的地址。最后，我们通过ZVAL_STR来设置PHP应用层要用的data。

  3. 本次优化我们使用了readv代替read。这么做的目的是，Swoole进程间通信的时候，使用了一个header来保存data的信息，例如数据是哪个reactor线程发来的，发送的数据大小是多少。但是，这个header对于PHP应用层来说是无需关心的，所以，我们就需要把本来连续的header:data内存单独分开了，把所有的data拼接起来。但是，这样就避免不了对data的拷贝了。所以我们需要通过readv来在接收数据的时候就分离header和data。但是，因为readv是需要指定每一段buffer的长度的，所以我们在接收第一个Chunk之前，需要调用read的peek方法来获取到header的长度，但是这个系统调用的开销是非常小的。

发表于
字数统计 712 字 | 阅读时长 3 分钟

这篇文章，我们来介绍下如何通过PHP扩展在PHP脚本将要退出前执行代码。我们可以看一段Swoole的协程代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
<?php

use Swoole\Coroutine;

go(function () {
    go(function () {
        Coroutine::sleep(2);
        var_dump("1");
    });
    go(function () {
        Coroutine::sleep(2);
        var_dump("2");
    });
    var_dump("3");
});
var_dump("4");

执行结果如下：

1
2
3
4
string(1) "3"
string(1) "4"
string(1) "1"
string(1) "2"

我们发现，在打印4之后，PHP脚本算是执行完了，这个时候，PHP进程也快要退出了。那为什么还会打印出1和2呢？

因为PHP是一门解释性语言，虽然PHP脚本的代码跑完了，但是PHP命令解释器还没有跑完，自然可以让PHP代码继续跑。所以，实际上，这个代码会这样：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
<?php

use Swoole\Coroutine;
use Swoole\Event;

go(function () {
    go(function () {
        Coroutine::sleep(2);
        var_dump("1");
    });
    go(function () {
        Coroutine::sleep(2);
        var_dump("2");
    });
    var_dump("3");
});
var_dump("4");

Event::wait();

在脚本的最后会调用Event::wait()来等待事件的结束。（而这里的事件就是定时器的事件）

好的，我们现在通过一个小demo来实现这个功能。

首先，我们需要创建PHP扩展的基础骨架。通过ext_skel工具生成：

1
[root@64fa874bf7d4 ext]# php ext_skel.php --ext register

然后进入register目录：

1
[root@64fa874bf7d4 ext]# cd register/

替换config.m4为如下内容：

1
2
3
4
5
6
7
PHP_ARG_ENABLE(register, whether to enable register support,
Make sure that the comment is aligned:
[  --enable-register           Enable register support])
​
if test "$PHP_REGISTER" != "no"; then
  PHP_NEW_EXTENSION(register, register.c, $ext_shared,, -DZEND_ENABLE_STATIC_TSRMLS_CACHE=1)
fi

然后，我们创建和编写测试脚本：

1
[root@64fa874bf7d4 register]# touch test.php
1
2
3
4
5
<?php

function test() {
    var_dump("codinghuang");
}

这段脚本只定义了一个test函数，并没有调用。现在我们的任务就是去调用它。

我们开始编写PHP扩展。在文件register.c的PHP_RINIT_FUNCTION里面注册test函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
#include "ext/standard/basic_functions.h"
#include "zend_API.h"

PHP_RINIT_FUNCTION(register)
{
#if defined(ZTS) && defined(COMPILE_DL_REGISTER)
	ZEND_TSRMLS_CACHE_UPDATE();
#endif

	php_shutdown_function_entry shutdown_function_entry;
    shutdown_function_entry.arg_count = 1;
    shutdown_function_entry.arguments = (zval *) safe_emalloc(sizeof(zval), 1, 0);
    ZVAL_STRING(&shutdown_function_entry.arguments[0], "test");
    register_user_shutdown_function("test", ZSTR_LEN(Z_STR(shutdown_function_entry.arguments[0])), &shutdown_function_entry);

	return SUCCESS;
}

这里，我们首先对php_shutdown_function_entry结构进行初始化，php_shutdown_function_entry.arguments的第一个位置填函数的名字。shutdown_function_entry.arg_count填写1，因为函数名字也算做是arguments。初始化完php_shutdown_function_entry之后，我们调用register_user_shutdown_function函数即可注册test函数了。这样，就会在php请求shutdown阶段调用我们注册的函数了。

然后编译、安装扩展：

1
[root@64fa874bf7d4 register]# phpize && ./configure && make && make install

然后把扩展在php.ini文件里面进行开启：

1
2
; Enable zlib extension module
extension=register.so

然后执行脚本：

1
2
3
[root@64fa874bf7d4 register]# php test.php
string(11) "codinghuang"
[root@64fa874bf7d4 register]#

我们发现，成功的调用了test函数。

这让我想起了我之前面试腾讯的时候，有一道题目，说是如何在每一个PHP函数调用之前，都执行一段代码。这个问题以后补上。

发表于
字数统计 1.2k 字 | 阅读时长 5 分钟

Swoole在v4版本后内置了Library模块，使用PHP代码编写内核功能，使得底层设施更加稳定可靠。

这篇文章我们就来分析一下如何使用以及编写Library。

首先，在Swoole v4的早期版本，swoole-src下面是有一个library目录的，里面就存放了很多的PHP代码，也就是Library库的代码。在后续版本，这个library就独立出了一个仓库。

简单介绍了背景之后，我们就来分析下这个Library如何工作以及如何使用的。

如何把Library作为Swoole扩展的内置库

eval

首先，我们要明白Library可以作为Swoole内置库的工作原理。我们先来看一段代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
<?php

echo "start\n";
$code = '
    class Library
    {
        public function func1() {
            echo "codinghuang\n";
        }
    }
';
eval($code);
(new Library)->func1();
echo "end\n";

执行结果如下：

1
2
3
start
codinghuang
end

我们发现，通过eval可以把$code对应的字符串作为代码来执行。Swoole也是使用了eval的能力，把PHP代码作为Swoole扩展的内置库。在Swoole扩展层面，就是调用了zend_eval_stringl来执行PHP代码的。

既然如此，Swoole肯定就需要读取写好的PHP Library，然后在扩展加载的时候，执行zend_eval_stringl。然后，这些类、函数、变量等等就生成了。

Swoole如何读取到写好的Library代码

其实，Library代码都在文件swoole-src/php_swoole_library.h里面。我们其中一个例子：

1
2
3
4
5
6
7
8
9
10
11
12
13
static const char* swoole_library_source_constants =
    "\n"
    "/**\n"
    " * This file is part of Swoole.\n"
    " *\n"
    " * @link     https://www.swoole.com\n"
    " * @contact  team@swoole.com\n"
    " * @license  https://github.com/swoole/library/blob/master/LICENSE\n"
    " */\n"
    "\n"
    "declare(strict_types=1);\n"
    "\n"
    "define('SWOOLE_LIBRARY', true);\n";

可以发现，这段PHP代码作为C++的字符串存在。然后，在函数php_swoole_load_library里面就调用了：

1
zend::eval(swoole_library_source_constants, "@swoole-src/library/constants.php");

来执行这段PHP代码。这样，常量SWOOLE_LIBRARY就被定义了。我们在来看看其他的Library代码（因为太长，我省略了部分）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
static const char* swoole_library_source_core_constant =
    "\n"
    "/**\n"
    " * This file is part of Swoole.\n"
    " *\n"
    " * @link     https://www.swoole.com\n"
    " * @contact  team@swoole.com\n"
    " * @license  https://github.com/swoole/library/blob/master/LICENSE\n"
    " */\n"
    "\n"
    "declare(strict_types=1);\n"
    "\n"
    "namespace Swoole;\n"
    "\n"
    "class Constant\n"
    "{\n"
    "    public const OPTION_BUFFER_INPUT_SIZE = 'buffer_input_size';\n"
    "\n"
    "    public const OPTION_BUFFER_OUTPUT_SIZE = 'buffer_output_size';\n"
    "\n"
    "    public const OPTION_MESSAGE_QUEUE_KEY = 'message_queue_key';\n"
    "\n"
    "    public const OPTION_BACKLOG = 'backlog';\n"
    "\n"
    "    public const OPTION_KERNEL_SOCKET_RECV_BUFFER_SIZE = 'kernel_socket_recv_buffer_size';\n"
    "\n"
    "    public const OPTION_KERNEL_SOCKET_SEND_BUFFER_SIZE = 'kernel_socket_send_buffer_size';\n"
    "\n"
    "    public const OPTION_TCP_DEFER_ACCEPT = 'tcp_defer_accept';\n"
    "\n"
    "    public const OPTION_OPEN_TCP_KEEPALIVE = 'open_tcp_keepalive';\n"
    "\n"
    "    public const OPTION_OPEN_HTTP_PROTOCOL = 'open_http_protocol';\n"
    "\n"
    "    public const OPTION_OPEN_WEBSOCKET_PROTOCOL = 'open_websocket_protocol';\n"
    "\n"
    "    public const OPTION_WEBSOCKET_SUBPROTOCOL = 'websocket_subprotocol';\n"
    "\n"
    "    public const OPTION_OPEN_WEBSOCKET_CLOSE_FRAME = 'open_websocket_close_frame';\n"
    "\n"
    "    public const OPTION_OPEN_HTTP2_PROTOCOL = 'open_http2_protocol';\n"
    "\n"
    "    public const OPTION_OPEN_REDIS_PROTOCOL = 'open_redis_protocol';\n"
    "\n"
    "    public const OPTION_TCP_KEEPIDLE = 'tcp_keepidle';\n"
    "\n"
    "    public const OPTION_TCP_KEEPINTERVAL = 'tcp_keepinterval';\n"
    "\n"
    "    public const OPTION_TCP_KEEPCOUNT = 'tcp_keepcount';\n"
    "\n"
    "    public const OPTION_TCP_FASTOPEN = 'tcp_fastopen';\n"
    "\n"
    "    public const OPTION_PACKAGE_BODY_START = 'package_body_start';\n"
    "\n"
    "    public const OPTION_SSL_CLIENT_CERT_FILE = 'ssl_client_cert_file';\n"
    "\n"
    "    public const OPTION_SSL_PREFER_SERVER_CIPHERS = 'ssl_prefer_server_ciphers';\n"
    "\n"
    "    public const OPTION_SSL_CIPHERS = 'ssl_ciphers';\n"
    "\n"
    "    public const OPTION_SSL_ECDH_CURVE = 'ssl_ecdh_curve';\n"
    "\n"
    "    public const OPTION_SSL_DHPARAM = 'ssl_dhparam';\n"
    "\n"
    "    public const OPTION_OPEN_SSL = 'open_ssl';\n"
    "\n"
    "    /* }}} OPTION */\n"
    "}\n";

我们发现，这里有太多的PHP常量了，并且常量对应的值很多都是Swoole的配置项这种，例如buffer_input_size。所以，这些php_swoole_library.h文件里面的Library肯定不是手写的。

实际上，php_swoole_library.h里面的Library代码是通过swoole-src/remake_library.sh工具自动生成的。

但是，如果你在比较新的swoole-src代码里面直接跑remake_library.sh脚本，是会报错的：

1
2
3
4
5
6
7
[root@64fa874bf7d4 swoole-src]# ./remake_library.sh
rm swoole.lo
rm php_swoole_library.h
sh: line 0: cd: /root/codeDir/cppCode/swoole-src/library: No such file or directory
sh: line 0: cd: /root/codeDir/cppCode/swoole-src/library: No such file or directory
❌ Unable to get commit id of library in [/root/codeDir/cppCode/swoole-src/library]
[root@64fa874bf7d4 swoole-src]#

因为library代码已经不在swoole-src下面了。

如何生成Library代码

所以，我们需要先准备好library代码，我们可以直接在swoole-src下面git clone下library代码：

1
git clone git@github.com:swoole/library.git

然后，在目录swoole-src下执行remake_library.sh脚本即可把library的代码生成在php_swoole_library.h文件里面：

1
2
3
4
5
[root@64fa874bf7d4 swoole-src]# ./remake_library.sh
🚀🚀🚀Generated swoole php library successfully!🚀🚀🚀
remake...
done
[root@64fa874bf7d4 swoole-src]#

这样，就可以把library仓库最新的代码生成到php_swoole_library.h文件里面了。

Swoole的那些常量如何编写的

我们发现，在library里面有一个文件library/src/core/Constant.php，这里面包含了很多Swoole内核的常量字符串，比如说：

1
2
3
4
5
6
public const EVENT_RECEIVE = 'receive';
public const EVENT_CONNECT = 'connect';
public const OPTION_SSL_CERT_FILE = 'ssl_cert_file';
public const OPTION_SSL_KEY_FILE = 'ssl_key_file';
public const OPTION_BUFFER_INPUT_SIZE = 'buffer_input_size';
public const OPTION_BUFFER_OUTPUT_SIZE = 'buffer_output_size';

等等，这些要写起来是非常的繁琐，而且很容易漏了。并且，如果你在Swoole内核中修改了配置项，那么我们就需要在Constant.php里面去更新对应的常量，非常的麻烦。所以，官方提供了一个工具swoole-src/tools/constant-generator.php。只要我们跑这个PHP脚本，它就会通过正则匹配，把Swoole内核里面的那些常量字符串找出来，然后生成到文件library/src/core/Constant.php里面，大大的提高了工作效率。我们来演示一下：

1
2
3
[root@64fa874bf7d4 tools]# php constant-generator.php
🚀🚀🚀Constant generator successfully done!🚀🚀🚀
[root@64fa874bf7d4 tools]#

从这里我们发现，library的代码除了手动编写的之外，还有部分是工具生成的。

（未完）

发表于
字数统计 280 字 | 阅读时长 1 分钟

想给自己的2020规划下。
主线：

  1. 精通Swoole，熟悉PHP常见的内核代码
  2. 学习编译原理，参考PHP语法写一个tinyphp，后续开一个新的专栏，整理成类似于PHP扩展开发教程的文章

其他：

  1. 学习C++
  2. 学习设计模式

如果可以完成第一点，可以给自己打60分，说明正常完成了日常工作。
如果可以完成第二点，可以给自己打80分，说明把大学期间欠下的编译原理还完了。（大四花了一年去还操作系统和计算机网络的债，然而，自己还没有写过一个比较完整的操作系统，这个估计要2021年去还债了）
如果可以完成第三点，可以给自己打90分，说明除了PHP和C语言，又多了一门可以吃饭的语言了。
如果可以完成第四点，可以给自己打95分，说明可以有能力收拾自己的臭代码了。
如果能学会炒几盆拿手的菜，就可以给自己打100分了。

发表于
字数统计 2.8k 字 | 阅读时长 13 分钟

在Swoole4.5版本中（目前还未发布），我们的Server有一个性能需要优化的地方，就是worker进程在收到master进程发来的包的时候，需要进行两次的拷贝，才可以把数据从PHP扩展层传递到PHP上层（也就是我们事件回调函数的data参数）。

我们先来分析一下为什么会有性能的问题。首先，我们需要一份会有性能问题的代码。我们git clone下swoole-src代码，然后git checkout到8235c82fea2130534a16fd20771dcab3408a763e这个commit位置：

1
git checkout 8235c82fea2130534a16fd20771dcab3408a763e

我们来分析一下代码，首先看master进程是如何封装数据然后发送给worker进程的。在函数process_send_packet里面，我们看核心的地方：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
static int process_send_packet(swServer *serv, swPipeBuffer *buf, swSendData *resp, send_func_t _send, void* private_data)
{
    const char* data = resp->data;
    uint32_t send_n = resp->info.len;
    off_t offset = 0;

    uint32_t max_length = serv->ipc_max_size - sizeof(buf->info);

    if (send_n <= max_length)
    {
        buf->info.flags = 0;
        buf->info.len = send_n;
        memcpy(buf->data, data, send_n);

        int retval = _send(serv, buf, sizeof(buf->info) + send_n, private_data);
        return retval;
    }

    buf->info.flags = SW_EVENT_DATA_CHUNK;

    while (send_n > 0)
    {
        if (send_n > max_length)
        {
            buf->info.len = max_length;
        }
        else
        {
            buf->info.flags |= SW_EVENT_DATA_END;
            buf->info.len = send_n;
        }

        memcpy(buf->data, data + offset, buf->info.len);

        if (_send(serv, buf, sizeof(buf->info) + buf->info.len, private_data) < 0)
        {
            return SW_ERR;
        }

        send_n -= buf->info.len;
        offset += buf->info.len;
    }

    return SW_OK;
}

首先，我们来说一下process_send_packet这个函数的参数：

其中，

swServer *serv就是我们创建的那个Server。

swPipeBuffer *buf指向的内存里面的数据需要发送给worker进程。

swSendData *resp里面存放了master进程收到的客户端数据以及一个swDataHead info头部。

_send是一个回调函数，这里面的逻辑就是master进程把swPipeBuffer *buf里面的数据发送给worker进程。

void* private_data这里是一个swWorker *worker类型的指针转换过来的。指定了master进程需要发送的那个worker进程。

说明一点，这里我们是以Server设置了eof选项为例子讲解的（假设设置了"\r\n"）。因为TCP是面向字节流的，即使客户端发送了一个很大的包过来，服务器一次read出来的数据也不见得非常大。如果不设置eof的话，是不会导致我们这篇文章所说的性能问题。

介绍完了process_send_packet函数的参数之后，我们来看看代码是如何实现的：

1
const char* data = resp->data;

首先，让data指向resp->data，也就是客户端发来的实际数据。例如，客户端发来了字符串hello world\r\n，那么data里面存放的就是hello world\r\n。

1
uint32_t send_n = resp->info.len;

标志着resp->data数据的长度。例如，客户端往服务器发送了1M的数据，那么resp->info.len就是1048576。

1
off_t offset = 0;

用来标志哪些数据master进程已经发送给了worker进程。

1
uint32_t max_length = serv->ipc_max_size - sizeof(buf->info);

max_length表示master进程一次往worker进程发送的包最大长度。

注意：master进程和worker进程是通过udg方式进行通信的。所以，master进程发送多少，worker进程就直接收多少

1
2
3
4
5
6
7
8
9
if (send_n <= max_length)
{
    buf->info.flags = 0;
    buf->info.len = send_n;
    memcpy(buf->data, data, send_n);

    int retval = _send(serv, buf, sizeof(buf->info) + send_n, private_data);
    return retval;
}

如果master进程要发给worker进程的数据小于max_length，那么就直接调用_send函数，直接把数据发给worker进程。

1
buf->info.flags = SW_EVENT_DATA_CHUNK;

当send_n大于max_length的时候，设置buf->info.flags为CHUNK，也就意味着需要把客户端发来的数据先拆分成一小段一小段的数据，然后再发送给worker进程。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
while (send_n > 0)
{
    if (send_n > max_length)
    {
        buf->info.len = max_length;
    }
    else
    {
        buf->info.flags |= SW_EVENT_DATA_END;
        buf->info.len = send_n;
    }

    memcpy(buf->data, data + offset, buf->info.len);

    if (_send(serv, buf, sizeof(buf->info) + buf->info.len, private_data) < 0)
    {
        return SW_ERR;
    }

    send_n -= buf->info.len;
    offset += buf->info.len;
}

逻辑比较简单，就是一个分段发送的过程。这里需要注意的两点：

1
2
1、buf->info.len的长度需要更新为小段的chunk的长度，而不是大数据包的长度
2、最后一个chunk的info.flags需要变成SW_EVENT_DATA_END，意味着一个完整的包已经发完了

OK，分析完了master进程发包的过程，我们来分析一下worker进程收包的过程。

我们先看一下函数swWorker_onPipeReceive：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
static int swWorker_onPipeReceive(swReactor *reactor, swEvent *event)
{
    swServer *serv = (swServer *) reactor->ptr;
    swFactory *factory = &serv->factory;
    swPipeBuffer *buffer = serv->pipe_buffers[0];
    int ret;

    _read_from_pipe:

    if (read(event->fd, buffer, serv->ipc_max_size) > 0)
    {
        ret = swWorker_onTask(factory, (swEventData *) buffer);
        if (buffer->info.flags & SW_EVENT_DATA_CHUNK)
        {
            //no data
            if (ret < 0 && errno == EAGAIN)
            {
                return SW_OK;
            }
            else if (ret > 0)
            {
                goto _read_from_pipe;
            }
        }
        return ret;
    }

    return SW_ERR;
}

这个就是worker进程接收master进程发来的数据的代码。

我们看的，worker进程会直接把数据先读取到buffer内存里面，然后调用swWorker_onTask。我们再来看看swWorker_onTask函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
int swWorker_onTask(swFactory *factory, swEventData *task)
{
    swServer *serv = (swServer *) factory->ptr;
    swWorker *worker = SwooleWG.worker;

    //worker busy
    worker->status = SW_WORKER_BUSY;
    //packet chunk
    if (task->info.flags & SW_EVENT_DATA_CHUNK)
    {
        if (serv->merge_chunk(serv, task->info.reactor_id, task->data, task->info.len) < 0)
        {
            swoole_error_log(SW_LOG_WARNING, SW_ERROR_SESSION_DISCARD_DATA,
                    "cannot merge chunk to worker buffer, data[fd=%d, size=%d] lost", task->info.fd, task->info.len);
            return SW_OK;
        }
        //wait more data
        if (!(task->info.flags & SW_EVENT_DATA_END))
        {
            return SW_OK;
        }
    }

    switch (task->info.type)
    {
    case SW_SERVER_EVENT_SEND_DATA:
        //discard data
        if (swWorker_discard_data(serv, task) == SW_TRUE)
        {
            break;
        }
        swWorker_do_task(serv, worker, task, serv->onReceive);
        break;
    // 省略其他的case
    default:
        swWarn("[Worker] error event[type=%d]", (int )task->info.type);
        break;
    }

    //worker idle
    worker->status = SW_WORKER_IDLE;

    //maximum number of requests, process will exit.
    if (!SwooleWG.run_always && worker->request_count >= SwooleWG.max_request)
    {
        swWorker_stop(worker);
    }
    return SW_OK;
}

我们重点看看性能问题代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
if (task->info.flags & SW_EVENT_DATA_CHUNK)
{
    if (serv->merge_chunk(serv, task->info.reactor_id, task->data, task->info.len) < 0)
    {
        swoole_error_log(SW_LOG_WARNING, SW_ERROR_SESSION_DISCARD_DATA,
                "cannot merge chunk to worker buffer, data[fd=%d, size=%d] lost", task->info.fd, task->info.len);
        return SW_OK;
    }
    //wait more data
    if (!(task->info.flags & SW_EVENT_DATA_END))
    {
        return SW_OK;
    }
}

这里，worker进程会先判断master发来的数据是否是CHUNK数据，如果是，那么会进行merge_chunk的操作。我们看看merge_chunk对应的函数：

1
2
3
4
5
6
static int swServer_worker_merge_chunk(swServer *serv, int key, const char *data, size_t len)
{
    swString *package = swServer_worker_get_input_buffer(serv, key);
    //merge data to package buffer
    return swString_append_ptr(package, data, len);
}

我们会先根据key的值（实际上是reactor线程的id），获取一块全局的内存，然后把接收到的chunk数据，追加到这个全局的内存上面，而swString_append_ptr执行的就是memcpy的操作。

所以，这就是一个性能问题了。worker进程接收到的所有数据都会被完整的拷贝一遍。如果客户端发来的数据很大，这个拷贝的开销也是很大声的。

因此，我们对这部分合并的代码进行了一个优化。我们让worker进程在接收master进程的数据之前，就准备好一块足够大的内存，然后直接把master进程发来的数据下来即可。

我们先更新一下swoole-src的源码：

1
git checkout 529ad44d578930b3607abedcfc278364df34bc73

我们依旧先看看process_send_packet函数的代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
static int process_send_packet(swServer *serv, swPipeBuffer *buf, swSendData *resp, send_func_t _send, void* private_data)
{
    const char* data = resp->data;
    uint32_t send_n = resp->info.len;
    off_t offset = 0;
    uint32_t copy_n;

    uint32_t max_length = serv->ipc_max_size - sizeof(buf->info);

    if (send_n <= max_length)
    {
        buf->info.flags = 0;
        buf->info.len = send_n;
        memcpy(buf->data, data, send_n);

        int retval = _send(serv, buf, sizeof(buf->info) + send_n, private_data);
        return retval;
    }

    buf->info.flags = SW_EVENT_DATA_CHUNK;
    buf->info.len = send_n;

    while (send_n > 0)
    {
        if (send_n > max_length)
        {
            copy_n = max_length;
        }
        else
        {
            buf->info.flags |= SW_EVENT_DATA_END;
            copy_n = send_n;
        }

        memcpy(buf->data, data + offset, copy_n);

        swTrace("finish, type=%d|len=%d", buf->info.type, copy_n);

        if (_send(serv, buf, sizeof(buf->info) + copy_n, private_data) < 0)
        {
            return SW_ERR;
        }

        send_n -= copy_n;
        offset += copy_n;
    }

    return SW_OK;
}

我们聚焦修改的地方，主要是对CHUNK的处理：

1
2
buf->info.flags = SW_EVENT_DATA_CHUNK;
buf->info.len = send_n;

我们发现，buf->info.len的长度不是每个小段chunk的长度了，而是整个大包的长度了。为什么可以这样做呢？因为master进程与worker进程是通过udg进行通信的，所以，worker进程在调用recv的时候，返回值实际上就是chunk的长度了，所以buf->info.len里面存储chunk的长度没有必要。

其他地方的逻辑和之前的代码没有区别。

我们再来看看worker进程是如何接收master进程发来的数据的。在函数swWorker_onPipeReceive里面：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
static int swWorker_onPipeReceive(swReactor *reactor, swEvent *event)
{
    int ret;
    ssize_t recv_n = 0;
    swServer *serv = (swServer *) reactor->ptr;
    swFactory *factory = &serv->factory;
    swPipeBuffer *pipe_buffer = serv->pipe_buffers[0];
    void *buffer;
    struct iovec buffers[2];

    // peek
    recv_n = recv(event->fd, &pipe_buffer->info, sizeof(pipe_buffer->info), MSG_PEEK);
    if (recv_n < 0 && errno == EAGAIN)
    {
        return SW_OK;
    }
    else if (recv_n < 0)
    {
        return SW_ERR;
    }

    if (pipe_buffer->info.flags & SW_EVENT_DATA_CHUNK)
    {
        buffer = serv->get_buffer(serv, &pipe_buffer->info);
        _read_from_pipe:

        buffers[0].iov_base = &pipe_buffer->info;
        buffers[0].iov_len = sizeof(pipe_buffer->info);
        buffers[1].iov_base = buffer;
        buffers[1].iov_len = serv->ipc_max_size - sizeof(pipe_buffer->info);

        recv_n = readv(event->fd, buffers, 2);
        if (recv_n < 0 && errno == EAGAIN)
        {
            return SW_OK;
        }
        if (recv_n > 0)
        {
            serv->add_buffer_len(serv, &pipe_buffer->info, recv_n - sizeof(pipe_buffer->info));
        }

        if (pipe_buffer->info.flags & SW_EVENT_DATA_CHUNK)
        {
            //wait more chunk data
            if (!(pipe_buffer->info.flags & SW_EVENT_DATA_END))
            {
                goto _read_from_pipe;
            }
            else
            {
                pipe_buffer->info.flags |= SW_EVENT_DATA_OBJ_PTR;
                /**
                 * Because we don't want to split the swEventData parameters into swDataHead and data,
                 * we store the value of the worker_buffer pointer in swEventData.data.
                 * The value of this pointer will be fetched in the swServer_worker_get_packet function.
                 */
                serv->copy_buffer_addr(serv, pipe_buffer);
            }
        }
    }
    else
    {
        recv_n = read(event->fd, pipe_buffer, serv->ipc_max_size);
    }

    if (recv_n > 0)
    {
        ret = swWorker_onTask(factory, (swEventData *) pipe_buffer, recv_n - sizeof(pipe_buffer->info));
        return ret;
    }

    return SW_ERR;
}

其中，

1
2
3
4
5
6
7
8
9
recv_n = recv(event->fd, &pipe_buffer->info, sizeof(pipe_buffer->info), MSG_PEEK);
if (recv_n < 0 && errno == EAGAIN)
{
    return SW_OK;
}
else if (recv_n < 0)
{
    return SW_ERR;
}

我们先对内核缓冲区里面的数据进行一次peek操作，来获取到head部分。这样我们就知道数据是否是以CHUNK方式发来的了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
if (pipe_buffer->info.flags & SW_EVENT_DATA_CHUNK)
{
    buffer = serv->get_buffer(serv, &pipe_buffer->info);
    _read_from_pipe:

    buffers[0].iov_base = &pipe_buffer->info;
    buffers[0].iov_len = sizeof(pipe_buffer->info);
    buffers[1].iov_base = buffer;
    buffers[1].iov_len = serv->ipc_max_size - sizeof(pipe_buffer->info);

    recv_n = readv(event->fd, buffers, 2);
    if (recv_n < 0 && errno == EAGAIN)
    {
        return SW_OK;
    }
    if (recv_n > 0)
    {
        serv->add_buffer_len(serv, &pipe_buffer->info, recv_n - sizeof(pipe_buffer->info));
    }

    if (pipe_buffer->info.flags & SW_EVENT_DATA_CHUNK)
    {
        //wait more chunk data
        if (!(pipe_buffer->info.flags & SW_EVENT_DATA_END))
        {
            goto _read_from_pipe;
        }
        else
        {
            pipe_buffer->info.flags |= SW_EVENT_DATA_OBJ_PTR;
            /**
                * Because we don't want to split the swEventData parameters into swDataHead and data,
                * we store the value of the worker_buffer pointer in swEventData.data.
                * The value of this pointer will be fetched in the swServer_worker_get_packet function.
                */
            serv->copy_buffer_addr(serv, pipe_buffer);
        }
    }
}

如果是CHUNK方式发来的数据，那么我们执行如下的操作：

1
buffer = serv->get_buffer(serv, &pipe_buffer->info);

get_buffer是一个回调函数，对应：

1
2
3
4
5
6
7
8
9
10
11
static void* swServer_worker_get_buffer(swServer *serv, swDataHead *info)
{
    swString *worker_buffer = swServer_worker_get_input_buffer(serv, info->reactor_id);

    if (worker_buffer->size < info->len)
    {
        swString_extend(worker_buffer, info->len);
    }

    return worker_buffer->str + worker_buffer->length;
}

这里，我们会先判断这块全局的buffer是否足够的大，可以接收完整个大包，如果不够大，我们扩容到足够的大。

1
2
3
4
5
6
7
8
_read_from_pipe:

buffers[0].iov_base = &pipe_buffer->info;
buffers[0].iov_len = sizeof(pipe_buffer->info);
buffers[1].iov_base = buffer;
buffers[1].iov_len = serv->ipc_max_size - sizeof(pipe_buffer->info);

recv_n = readv(event->fd, buffers, 2);

然后，我们调用readv，把head和实际的数据分别存在了两个地方。这么做是避免为了把head和实际的数据做拆分而导致的内存拷贝。

通过以上方式，Swoole Server减少了一次内存拷贝。

发表于
字数统计 562 字 | 阅读时长 2 分钟

环境：Mac OS X

搭建环境工具：Docker

思路是通过Docker起一个Linux容器，然后通过vscode的Remote-SSH插件登陆进这个Linux容器。这样就可以在Mac OS X或者Windows下调试和Linux系统相关的代码了。

首先，git clone下我准备好的Dockerfile：

1
git clone git@github.com:dockero/php_centos.git

php_centos的目录结构如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
tree
.
├── Dockerfile
├── README.md
├── docker-compose.yml
└── etc
    └── php.d
        ├── curl.ini
        ├── openssl.ini
        ├── swoole.ini
        ├── zip.ini
        └── zlib.ini

2 directories, 8 files

其中，

Dockerfile用来构建镜像的，已经预先安装了php、swoole以及其他基础的扩展。

docker-compose.yml主要是在创建容器的时候设置一些环境变量。

etc目录则是存放php扩展的.ini文件。这里，我们一个扩展对应一个.ini文件。

除此之外，你在目录php_centos下需要自己创建一个.env文件，用来控制swoole的版本，以及设置公钥等。

这里是我的一份模板：

1
2
3
4
5
6
7
HTTP_PROXY=http://127.0.0.1:8080
HTTPS_PROXY=http://127.0.0.1:8080
CODEDIR_VOLUME=~/codeDir:/root/codeDir
HOST_SSH_PORT=9522
PASSWORD=123456
SWOOLE_VERSION=4.4.12
SSH_PUB_KEY=填写你的公钥

每个参数在README里面都有说明。

OK，我们编译镜像：

1
docker-compose build

然后启动容器：

1
docker-compose up -d

之后，我们执行命令：

1
ssh php

即可登陆容器了。（如果在登陆容器的时候，需要输入密码，你可以执行命令ssh-add 私钥来免密登陆）

然后，我们需要在vscode的同一个工作区里面存放PHP源码、Swoole源码、PHP测试脚本、.vscode目录。所有需要的东西如下：

1
2
3
4
5
6
tree -L 1
.
├── test.php
├── php-7.3.12
├── swoole-src
└── .vscode

我们在.vscode目录里面创建文件launch.json：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
{
    // 使用 IntelliSense 了解相关属性。 
    // 悬停以查看现有属性的描述。
    // 欲了解更多信息，请访问: https://go.microsoft.com/fwlink/?linkid=830387
    "version": "0.2.0",
    "configurations": [
        {
            "name": "debug swoole",
            "type": "cppdbg",
            "request": "launch",
            "program": "/usr/bin/php",
            "args": ["${file}"],
            "stopAtEntry": false,
            "cwd": "${workspaceFolder}",
            "environment": [],
            "externalConsole": false,
            "MIMode": "gdb",
            "setupCommands": [
                {
                    "description": "为 gdb 启用整齐打印",
                    "text": "-enable-pretty-printing",
                    "ignoreFailures": true
                }
            ]
        }
    ]
}

接下来，我们用vscode打开测试脚本test.php，并且停留在这个窗口（注意，必须在停留test.php窗口的时候点击调试才行），然后点击调试按钮即可进行Swoole源码的调试了。