发表于
字数统计 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源码的调试了。

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

长连接最大的问题就是连接失效。

redis/mysql等服务器为什么会限制最大连接数

内核角度，过多的TCP连接非常的不环保

首先，过多的内存占用，尤其是阻塞的情况下。

例如如下代码。

服务器：

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

$serv = new Swoole\Server("127.0.0.1", 6666, SWOOLE_BASE);
$serv->set(array(
    'worker_num' => 1,
));

$serv->on('connect', function ($serv, $fd) {
    var_dump("Client: Connect $fd");
});

$serv->on('receive', function ($serv, $fd, $reactor_id, $data) {
    var_dump(strlen($data));
    sleep(100000);
});

$serv->on('close', function ($serv, $fd) {
});

$serv->start();

代码很简单，服务器在第一次收到数据的时候，就调用sleep函数阻塞起来。（因为只有一个worker进程，所以后面不会处理客户端发来的其他数据）

客户端代码如下：

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

use Swoole\Coroutine\Socket;

$socket = new Socket(AF_INET, SOCK_STREAM, 0);

go(function () use ($socket) {
    $socket->connect('127.0.0.1', 6666);

    $i = 0;
    while ($i++ < 100000000) {
        var_dump($socket->send(str_repeat("1", 1024)));
    }
    echo "done\n";
});

代码很简单，就是创建一个协程，然后创建一个客户端，往服务器发送数据。一共发送了100000000 * 1024字节的数据。

我们先启动服务器：

1
[root@aaeb2c267d0f test]# php server.php

然后再启动客户端：

1
2
3
4
5
6
7
8
[root@aaeb2c267d0f test]# php client.php
# 省略其他的输出
int(1024)
int(1024)
int(1024)
int(1024)
int(1024)
int(617)

我们发现，客户端程序会阻塞住。

我们再来看看服务器端这边的输出：

1
2
3
[root@aaeb2c267d0f test]# php server.php
string(17) "Client: Connect 1"
int(6144)

我们发现，服务器读取到了6144字节的数据之后，就没有继续再读取了。（我们发现，服务器端读取到的字节大小不一定是1024，这也说明了TCP是面向字节流的，每次调用send函数，数据不一定就会立马发送过去。同理，服务器端也不会收到数据后就立马读取）

此时，我们关闭客户端这边的进程：

1
2
^C
[root@aaeb2c267d0f test]#

然后查看socket内核缓存区的状态：

1
2
3
4
5
6
7
8
[root@aaeb2c267d0f test]# cat /proc/net/sockstat
sockets: used 177
TCP: inuse 7 orphan 1 tw 0 alloc 15 mem 1211
UDP: inuse 1 mem 2
UDPLITE: inuse 0
RAW: inuse 0
FRAG: inuse 0 memory 0
[root@aaeb2c267d0f test]#

我们发现内核缓冲区是有数据没有被清空的。

所以我们发现，当服务器进程因为一个socket而被阻塞起来了，那么后续的连接都无法被处理，导致内存占用严重。

其次，过多的端口占用、fd占用。我们来看一个TCP自连接的问题。

代码如下：

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

use Swoole\Coroutine\Socket;

$socket = new Socket(AF_INET, SOCK_STREAM, 0);

go(function () use ($socket) {
    for ($i = 1; $i < 65536; $i++) {
        if ($socket->connect('127.0.0.1', $i)) {
            echo "connected\n";
            break;
        }
    }

    var_dump($i);
});

执行结果如下：

1
2
3
4
[root@aaeb2c267d0f test]# php client.php
connected
int(43991)
[root@aaeb2c267d0f test]#

我们发现，如果我们不限制连接的个数，那么客户端很可能就会耗尽可用的端口。可用的端口范围可以通过如下命令查看：

1
2
[root@aaeb2c267d0f test]# sysctl -a | grep ip_local_port_range
net.ipv4.ip_local_port_range = 32768    60999

传输效率问题

因为长连接可以减少TCP三次握手的时间，所以长连接的效率会高于短连接（短连接每次发送数据需要进行三次握手）。

综上，我们要控制长连接的数量，也就是说在限制了最大连接数的情况下，连接弥足珍贵。那我们确实有那么多进程怎么办，只能尽可能提高利用率，共享连接。

共享连接池

多协程共享连接

当协程需要使用一个连接的时候，可以从Channel里面pop出来一个连接。

多进程共享连接

跨机器共享连接

发表于
字数统计 2.5k 字 | 阅读时长 12 分钟

这篇文章我们开始来分析一下php stream server的源码，为后面我们hook php stream server做准备。PHP版本是7.1.0。

我们的调试代码如下：

1
2
3
4
5
6
<?php

$socket = stream_socket_server(
    'tcp://0.0.0.0:6666',
    $errno, $errstr, STREAM_SERVER_BIND | STREAM_SERVER_LISTEN
);

开始调试：

1
2
sh-4.2# cgdb php
(gdb)

我们先在php_init_stream_wrappers处打一个断点：

1
2
3
(gdb) b php_init_stream_wrappers
Breakpoint 1 at 0x7df8aa: file /root/php-7.1.0/main/streams/streams.c, line 1651.
(gdb)

然后运行代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
(gdb) r test.php
Starting program: /home/codes/php/php-7.1.0/output/bin/php test.php
[Thread debugging using libthread_db enabled]
Using host libthread_db library "/lib64/libthread_db.so.1".

Breakpoint 1, php_init_stream_wrappers (module_number=0) at /root/php-7.1.0/main/streams/streams.c:1651
Missing separate debuginfos, use: debuginfo-install glibc-2.17-260.el7_6.4.x86_64 libxml2-2.9.1-6.el7_2.3.x86_64 nss-softokn-freebl-3.36.0-5.el7_5.x86_64 xz-libs-5.2.2-1.el7.x86_64 zlib-1.2.7-18.el7.x86_6
4
(gdb)

1649│ int php_init_stream_wrappers(int module_number)
1650│ {
1651├───────> le_stream = zend_register_list_destructors_ex(stream_resource_regular_dtor, NULL, "stream", module_number);
1652│         le_pstream = zend_register_list_destructors_ex(NULL, stream_resource_persistent_dtor, "persistent stream", module_number);

此时断点触发。

然后，我们看一下函数调用栈：

1
2
3
4
5
6
(gdb) bt
#0  php_init_stream_wrappers (module_number=0) at /root/php-7.1.0/main/streams/streams.c:1651
#1  0x00000000007bfe8b in php_module_startup (sf=0x10b8c20 <cli_sapi_module>, additional_modules=0x0, num_additional_modules=0) at /root/php-7.1.0/main/main.c:2227
#2  0x000000000092e814 in php_cli_startup (sapi_module=0x10b8c20 <cli_sapi_module>) at /root/php-7.1.0/sapi/cli/php_cli.c:424
#3  0x0000000000930767 in main (argc=2, argv=0x10dc280) at /root/php-7.1.0/sapi/cli/php_cli.c:1345
(gdb)

可以看到，在PHP启动的时候，会去调用php_init_stream_wrappers这个函数。我们继续看看php_init_stream_wrappers这个函数做了什么事情。运行到1661：

1
2
3
4
5
6
7
8
9
10
11
12
(gdb) u 1661

1661├───────> return (php_stream_xport_register("tcp", php_stream_generic_socket_factory) == SUCCESS
1662│                         &&
1663│                         php_stream_xport_register("udp", php_stream_generic_socket_factory) == SUCCESS
1664│ #if defined(AF_UNIX) && !(defined(PHP_WIN32) || defined(__riscos__) || defined(NETWARE))
1665│                         &&
1666│                         php_stream_xport_register("unix", php_stream_generic_socket_factory) == SUCCESS
1667│                         &&
1668│                         php_stream_xport_register("udg", php_stream_generic_socket_factory) == SUCCESS
1669│ #endif
1670│                 ) ? SUCCESS : FAILURE;

停在了执行php_stream_xport_register这个位置。根据名字我们大概可以猜出来，这个函数应该是去注册某个东西。我们进入这个函数：

1
2
3
4
5
6
7
8
(gdb) s
php_stream_xport_register (protocol=0xd770bf "tcp", factory=0x7f006f <php_stream_generic_socket_factory>) at /root/php-7.1.0/main/streams/transports.c:34
(gdb)

 32│ PHPAPI int php_stream_xport_register(const char *protocol, php_stream_transport_factory factory)
 33│ {
 34├───────> return zend_hash_str_update_ptr(&xport_hash, protocol, strlen(protocol), factory) ? SUCCESS : FAILURE;
 35│ }

可以发现，这个函数实际上就是更新一下xport_hash这个hash表，哈希表里面存的东西是factory。factory是php_stream_transport_factory类型。我们看一看php_stream_transport_factory具体是什么样子的。在文件php_stream_transport.h里面：

1
2
3
4
5
6
7
typedef php_stream *(php_stream_transport_factory_func)(const char *proto, size_t protolen,
        const char *resourcename, size_t resourcenamelen,
        const char *persistent_id, int options, int flags,
        struct timeval *timeout,
        php_stream_context *context STREAMS_DC);
typedef php_stream_transport_factory_func *php_stream_transport_factory;
PHPAPI php_stream_transport_factory_func php_stream_generic_socket_factory;

可以看的，php_stream_generic_socket_factory是php_stream_transport_factory_func类型的变量。而php_stream_transport_factory_func是一个函数指针的类型。返回值是php_stream *，参数是：

1
2
3
4
5
const char *proto, size_t protolen,
const char *resourcename, size_t resourcenamelen,
const char *persistent_id, int options, int flags,
struct timeval *timeout,
php_stream_context *context STREAMS_DC

所以，我们可以得出一个结论：php_stream_xport_register注册了一个函数指针，这个函数指针是php_stream_transport_factory_func类型的函数指针。

OK，我们继续：

1
2
3
4
5
6
7
8
(gdb) finish
Run till exit from #0  php_stream_xport_register (protocol=0xd770bf "tcp", factory=0x7f006f <php_stream_generic_socket_factory>) at /root/php-7.1.0/main/streams/transports.c:34
php_init_stream_wrappers (module_number=0) at /root/php-7.1.0/main/streams/streams.c:1670
Value returned is $1 = 0
(gdb)

1670├───────────────> ) ? SUCCESS : FAILURE;
1671│ }

所以，php_init_stream_wrappers的作用就是去注册PHP默认的php_stream函数。然后，我们在函数zif_stream_socket_server处打一个断点：

1
2
3
(gdb) b zif_stream_socket_server
Breakpoint 3 at 0x7a3c9a: file /root/php-7.1.0/ext/standard/streamsfuncs.c, line 179.
(gdb)

然后继续运行：

1
2
3
4
5
6
7
8
9
10
11
(gdb) c
Continuing.

Breakpoint 2, zif_stream_socket_server (execute_data=0x7ffff5e140d0, return_value=0x7ffff5e140b0) at /root/php-7.1.0/ext/standard/streamsfuncs.c:179
(gdb)

 175│ PHP_FUNCTION(stream_socket_server)
 176│ {
 177│         char *host;
 178│         size_t host_len;
 179├───────> zval *zerrno = NULL, *zerrstr = NULL, *zcontext = NULL;

此时，已经进入了我们PHP脚本写的stream_socket_server里面了。我们继续来看看这个函数实现了什么。我们运行到207行之前：

1
2
3
4
5
6
7
(gdb) u 207
zif_stream_socket_server (execute_data=0x7ffff5e140d0, return_value=0x7ffff5e140b0) at /root/php-7.1.0/ext/standard/streamsfuncs.c:207
(gdb)

 207├───────> stream = php_stream_xport_create(host, host_len, REPORT_ERRORS,
 208│                         STREAM_XPORT_SERVER | (int)flags,
 209│                         NULL, NULL, context, &errstr, &err);

可以看到，php_stream_xport_create函数会依据我们传入的host、flag等信息创建出一个stream。我们很容易的可以猜到，这个应该就是php_stream *类型的指针。我们进入php_stream_xport_create函数里面：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
(gdb) s
_php_stream_xport_create (name=0x7ffff5e027e8 "tcp://0.0.0.0:6666", namelen=18, options=8, flags=13, persistent_id=0x0, timeout=0x0, context=0x7ffff5e01640, error_string=0x7fffffffb308, error_code=0x7ffff
fffb31c, __php_stream_call_depth=0, __zend_filename=0xd6ffb0 "/root/php-7.1.0/ext/standard/streamsfuncs.c", __zend_lineno=209, __zend_orig_filename=0x0, __zend_orig_lineno=0) at /root/php-7.1.0/main/strea
ms/transports.c:60
(gdb)

 52│ PHPAPI php_stream *_php_stream_xport_create(const char *name, size_t namelen, int options,
 53│                 int flags, const char *persistent_id,
 54│                 struct timeval *timeout,
 55│                 php_stream_context *context,
 56│                 zend_string **error_string,
 57│                 int *error_code
 58│                 STREAMS_DC)
 59│ {
 60├───────> php_stream *stream = NULL;
 61│         php_stream_transport_factory factory = NULL;
 62│         const char *p, *protocol = NULL;

可以看的，php_stream_xport_create确实会创建一个php_stream类型的指针。我们继续运行到109行之前：

1
2
3
4
5
6
7
8
9
10
(gdb) u 109
_php_stream_xport_create (name=0x7ffff5e027ee "0.0.0.0:6666", namelen=12, options=8, flags=13, persistent_id=0x0, timeout=0x7fffffffb230, context=0x7ffff5e01640, error_string=0x7fffffffb308, error_code=0x
7fffffffb31c, __php_stream_call_depth=0, __zend_filename=0xd6ffb0 "/root/php-7.1.0/ext/standard/streamsfuncs.c", __zend_lineno=209, __zend_orig_filename=0x0, __zend_orig_lineno=0) at /root/php-7.1.0/main/
streams/transports.c:109
(gdb)

109├───────> if (protocol) {
110│                 char *tmp = estrndup(protocol, n);
111│                 if (NULL == (factory = zend_hash_str_find_ptr(&xport_hash, tmp, n))) {
112│                         char wrapper_name[32];

我们发现，在111行会根据protocol的内容（在这个测试脚本中protocol是tcp）去xport_hash这个hash表里面查找factory，而这个factory应该就是我们之前注册的那个函数指针。我们继续执行到126行：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
(gdb) u 126
_php_stream_xport_create (name=0x7ffff5e027ee "0.0.0.0:6666", namelen=12, options=8, flags=13, persistent_id=0x0, timeout=0x7fffffffb230, context=0x7ffff5e01640, error_string=0x7fffffffb308, error_code=0x
7fffffffb31c, __php_stream_call_depth=0, __zend_filename=0xd6ffb0 "/root/php-7.1.0/ext/standard/streamsfuncs.c", __zend_lineno=209, __zend_orig_filename=0x0, __zend_orig_lineno=0) at /root/php-7.1.0/main/
streams/transports.c:127
(gdb)

127├───────> if (factory == NULL) {
128│                 /* should never happen */
129│                 php_error_docref(NULL, E_WARNING, "Could not find a factory !?");
130│                 return NULL;
131│         }
132│
133│         stream = (factory)(protocol, n,
134│                         (char*)name, namelen, persistent_id, options, flags, timeout,
135│                         context STREAMS_REL_CC);

我们发现，在133行这个位置会去调用factory这个函数，并且创建出一个stream。我们继续运行，进入factory对应的函数里面：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
(gdb) n
(gdb) s
php_stream_generic_socket_factory (proto=0x7ffff5e027e8 "tcp://0.0.0.0:6666", protolen=3, resourcename=0x7ffff5e027ee "0.0.0.0:6666", resourcenamelen=12, persistent_id=0x0, options=8, flags=13, timeout=0x
7fffffffb230, context=0x7ffff5e01640, __php_stream_call_depth=1, __zend_filename=0xd788d0 "/root/php-7.1.0/main/streams/transports.c", __zend_lineno=135, __zend_orig_filename=0xd6ffb0 "/root/php-7.1.0/ext
/standard/streamsfuncs.c", __zend_orig_lineno=209) at /root/php-7.1.0/main/streams/xp_socket.c:884
(gdb)

878│ PHPAPI php_stream *php_stream_generic_socket_factory(const char *proto, size_t protolen,
879│                 const char *resourcename, size_t resourcenamelen,
880│                 const char *persistent_id, int options, int flags,
881│                 struct timeval *timeout,
882│                 php_stream_context *context STREAMS_DC)
883│ {
884├───────> php_stream *stream = NULL;

我们发现，我们进入了在php_init_stream_wrappers函数中注册的php_stream_generic_socket_factory函数里面。

我们知道，是需要创建一个socket的，这样客户端才可以和服务器进行通信。那么，创建完php_stream之后，在哪个地方创建的socket呢？我们继续调试，退出php_stream_generic_socket_factory函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
(gdb) finish
Run till exit from #0  php_stream_generic_socket_factory (proto=0x7ffff5e027e8 "tcp://0.0.0.0:6666", protolen=3, resourcename=0x7ffff5e027ee "0.0.0.0:6666", resourcenamelen=12, persistent_id=0x0, options=
8, flags=13, timeout=0x7fffffffb230, context=0x7ffff5e01640, __php_stream_call_depth=1, __zend_filename=0xd788d0 "/root/php-7.1.0/main/streams/transports.c", __zend_lineno=135, __zend_orig_filename=0xd6ff
b0 "/root/php-7.1.0/ext/standard/streamsfuncs.c", __zend_orig_lineno=209) at /root/php-7.1.0/main/streams/xp_socket.c:884
0x00000000007ed2f0 in _php_stream_xport_create (name=0x7ffff5e027ee "0.0.0.0:6666", namelen=12, options=8, flags=13, persistent_id=0x0, timeout=0x7fffffffb230, context=0x7ffff5e01640, error_string=0x7ffff
fffb308, error_code=0x7fffffffb31c, __php_stream_call_depth=0, __zend_filename=0xd6ffb0 "/root/php-7.1.0/ext/standard/streamsfuncs.c", __zend_lineno=209, __zend_orig_filename=0x0, __zend_orig_lineno=0) at
 /root/php-7.1.0/main/streams/transports.c:133
Value returned is $4 = (php_stream *) 0x7ffff5e5fa00
(gdb)

133├───────> stream = (factory)(protocol, n,
134│                         (char*)name, namelen, persistent_id, options, flags, timeout,
135│                         context STREAMS_REL_CC);

在php_stream_xport_bind处打一个断点：

1
2
3
(gdb) b php_stream_xport_bind
Breakpoint 5 at 0x7ed66c: file /root/php-7.1.0/main/streams/transports.c, line 205.
(gdb)

然后继续运行：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
(gdb) c
Continuing.

Breakpoint 5, php_stream_xport_bind (stream=0x7ffff5e5fa00, name=0x7ffff5e027ee "0.0.0.0:6666", namelen=12, error_text=0x7fffffffb240) at /root/php-7.1.0/main/streams/transports.c:205
(gdb)

196│ /* Bind the stream to a local address */
197│ PHPAPI int php_stream_xport_bind(php_stream *stream,
198│                 const char *name, size_t namelen,
199│                 zend_string **error_text
200│                 )
201│ {
202│         php_stream_xport_param param;
203│         int ret;
204│
205├───────> memset(&param, 0, sizeof(param));
206│         param.op = STREAM_XPORT_OP_BIND;
207│         param.inputs.name = (char*)name;
208│         param.inputs.namelen = namelen;
209│         param.want_errortext = error_text ? 1 : 0;
210│
211│         ret = php_stream_set_option(stream, PHP_STREAM_OPTION_XPORT_API, 0, &param);

根据php_stream_xport_bind的注释，我们很容易知道，这个函数的作用肯定是会去调用bind函数来绑定ip和端口。并且我们发现，php_stream_xport_bind的核心函数就是php_stream_set_option。因为php_stream_set_option是一个宏，并且宏展开之后是_php_stream_set_option，所以我们在_php_stream_set_option处打一个断点：

1
2
3
(gdb) b _php_stream_set_option
Breakpoint 7 at 0x7dee15: file /root/php-7.1.0/main/streams/streams.c, line 1347.
(gdb)

继续运行：

1
2
3
4
5
6
7
8
9
10
11
12
13
(gdb) c
Continuing.

Breakpoint 7, _php_stream_set_option (stream=0x7ffff5e5fa00, option=7, value=0, ptrparam=0x7fffffffb110) at /root/php-7.1.0/main/streams/streams.c:1347
(gdb)

1345│ PHPAPI int _php_stream_set_option(php_stream *stream, int option, int value, void *ptrparam)
1346│ {
1347├───────> int ret = PHP_STREAM_OPTION_RETURN_NOTIMPL;
1348│
1349│         if (stream->ops->set_option) {
1350│                 ret = stream->ops->set_option(stream, option, value, ptrparam);
1351│         }

我们发现_php_stream_set_option函数的核心就是stream->ops->set_option，这个函数里面应该有我们希望看到的代码，我们进入这个函数：

1
2
3
4
5
6
7
8
(gdb) n
(gdb) n
(gdb) s

846│ static int php_tcp_sockop_set_option(php_stream *stream, int option, int value, void *ptrparam)
847│ {
848├───────> php_netstream_data_t *sock = (php_netstream_data_t*)stream->abstract;
849│         php_stream_xport_param *xparam;

我们发现进入了php_tcp_sockop_set_option函数里面，这个函数里面有一段核心的代码，通过switch语句根据xparam->op来选择执行：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
switch(option) {
    case PHP_STREAM_OPTION_XPORT_API:
        xparam = (php_stream_xport_param *)ptrparam;

        switch(xparam->op) {
            case STREAM_XPORT_OP_CONNECT:
            case STREAM_XPORT_OP_CONNECT_ASYNC:
                xparam->outputs.returncode = php_tcp_sockop_connect(stream, sock, xparam);
                return PHP_STREAM_OPTION_RETURN_OK;

            case STREAM_XPORT_OP_BIND:
                xparam->outputs.returncode = php_tcp_sockop_bind(stream, sock, xparam);
                return PHP_STREAM_OPTION_RETURN_OK;


            case STREAM_XPORT_OP_ACCEPT:
                xparam->outputs.returncode = php_tcp_sockop_accept(stream, sock, xparam STREAMS_CC);
                return PHP_STREAM_OPTION_RETURN_OK;
            default:
                /* fall through */
                ;
        }
}

在上面的调试过程中，我们知道，此时的option的值是PHP_STREAM_OPTION_XPORT_API。然后我们继续执行，看看会进入后面的那个case分支里面：

1
2
3
4
5
6
7
8
(gdb) n
(gdb) n
(gdb) n
(gdb) n
(gdb)
861│                                 case STREAM_XPORT_OP_BIND:
862├───────────────────────────────────────> xparam->outputs.returncode = php_tcp_sockop_bind(stream, sock, xparam);
863│                                         return PHP_STREAM_OPTION_RETURN_OK;

我们发现，进入了STREAM_XPORT_OP_BIND这个分支里面。我们进入函数php_tcp_sockop_bind里面：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
(gdb) s
php_tcp_sockop_bind (stream=0x7ffff5e5fa00, sock=0x7ffff5e02870, xparam=0x7fffffffb110) at /root/php-7.1.0/main/streams/xp_socket.c:615
(gdb)

612│ static inline int php_tcp_sockop_bind(php_stream *stream, php_netstream_data_t *sock,
613│                 php_stream_xport_param *xparam)
614│ {
615├───────> char *host = NULL;
616│         int portno, err;
617│         long sockopts = STREAM_SOCKOP_NONE;
618│         zval *tmpzval = NULL;
619│
620│ #ifdef AF_UNIX
621│         if (stream->ops == &php_stream_unix_socket_ops || stream->ops == &php_stream_unixdg_socket_ops) {
622│                 struct sockaddr_un unix_addr;
623│
624│                 sock->socket = socket(PF_UNIX, stream->ops == &php_stream_unix_socket_ops ? SOCK_STREAM : SOCK_DGRAM, 0);

我们发现，函数php_tcp_sockop_bind里面有创建socket的代码。说明创建socket的代码被封装在了php_tcp_sockop_bind里面。并且在这个函数的后面，我们也看到了bind这个函数。

OK，我们现在分析完了stream_socket_server这个PHP函数的工作流程。我们在想，如果我们调用php_stream_xport_register去替换掉xport_hash里面保存的php_stream_generic_socket_factory函数指针，是不是就可以协程化了呢？

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

大部分的传统FPM项目性能瓶颈在于每次请求重新创建ZendVM的开销、IO阻塞导致的上下文频繁切换。Swoole解决的就是这类问题。

这篇文章教大家如何让Swoole的HTTP服务器性能达到最大。

压测脚本如下，机器的配置是单核、2G内存、50G硬盘：

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
<?php

use Swoole\Http\Request;
use Swoole\Http\Response;

$process = new Swoole\Process(function (Swoole\Process $process) {
    $server = new Swoole\Http\Server('127.0.0.1', 9501, SWOOLE_BASE);
    $server->set([
        'log_file' => '/dev/null',
        'log_level' => SWOOLE_LOG_INFO,
        'worker_num' => swoole_cpu_num() * 2,
        // 'hook_flags' => SWOOLE_HOOK_ALL,
    ]);
    $server->on('workerStart', function () use ($process, $server) {
        $process->write('1');
    });
    $server->on('request', function (Request $request, Response $response) use ($server) {
        try {
            $redis = new Redis;
            $redis->connect('127.0.0.1', 6379);
            $greeter = $redis->get('greeter');
            if (!$greeter) {
                throw new RedisException('get data failed');
            }
            $response->end("<h1>{$greeter}</h1>");
        } catch (\Throwable $th) {
            $response->status(500);
            $response->end();
        }
    });
    $server->start();
});

if ($process->start()) {
    register_shutdown_function(function () use ($process) {
        $process::kill($process->pid);
        $process::wait();
    });
    $process->read(1);
    System('ab -c 256 -n 10000 -k http://127.0.0.1:9501/ 2>&1');
}

首先，我们创建了一个Swoole\Process对象，这个对象会开启一个子进程，在子进程中，我创建了一个HTTP Server，这个服务器是BASE模式的。除了BASE模式之外，还有一种PROCESS模式。在PROCESS模式下，套接字连接是在Master进程维持的，Master进程和Worker进程会多一层IPC通信的开销，但是，当Worker进程奔溃的时候，因为连接是在Master进程维持的，所以连接不会被断开。所以，Process模式适用于维护大量长连接的场景。BASE模式是在每个工作进程维持自己的连接，所以性能会比Master更好。并且，在HTTP Server下，BASE模式会更加的适用。

这里，我们将worker_num，也就是进程的数量设置为当前机器CPU核数的两倍。但是，在实际的项目中，我们需要不断的压测，来调整这个参数。

在workerStart的时候，也就是工作进程启动的时候，我们让子进程向管道中随意写入一个数据给父进程，父进程此时会读到一点数据，读到数据后，父进程才开始压测。

此时，压测的请求会进入onRequest回调。在这个回调中，我们创建了一个Redis客户端，这个客户端会连接Redis服务器，并请求一条数据。得到数据后，我们调用end方法来响应压测的请求。当错误时，我们返回一个错误码为500的响应。

在开始压测前，我们需要安装Redis扩展：

1
pecl install redis

然后php.ini配置中开启redis扩展即可。

我们还需要在Redis服务器里面插入一条数据：

1
2
3
4
5
127.0.0.1:6379> SET greeter swoole
OK
127.0.0.1:6379> GET greeter
"swoole"
127.0.0.1:6379>

OK，我们现在进行压测：

1
2
3
4
5
6
7
8
9
10
11
12
~/codeDir/phpCode/swoole/server # php server.php
Concurrency Level:      256
Time taken for tests:   2.293 seconds
Complete requests:      10000
Failed requests:        0
Keep-Alive requests:    10000
Total transferred:      1680000 bytes
HTML transferred:       150000 bytes
Requests per second:    4361.00 [#/sec] (mean)
Time per request:       58.702 [ms] (mean)
Time per request:       0.229 [ms] (mean, across all concurrent requests)
Transfer rate:          715.48 [Kbytes/sec] received

我们发现，现在的QPS比较低，只有4361.00。

因为，我们目前使用的Redis扩展是PHP官方的同步阻塞客户端，没有利用到协程（或者说异步的特性）。当进程去连接Redis服务器的时候，可能会阻塞整个进程，导致进程无法处理其他的连接，这样，这个HTTP Server处理请求的速度就不可能快。但是，这个压测结果会比FPM下好，因为Swoole是常驻进程的。

现在，我们来开启Swoole提供的RuntimeHook机制，也就是在运行时动态的将PHP同步阻塞的方法全部替换为异步非阻塞的协程调度的方法。我们只需要在server->set配置中加入一行即可：

1
'hook_flags' => SWOOLE_HOOK_ALL

此时，我们再来运行这个脚本：

1
2
3
4
5
6
7
8
9
10
11
Concurrency Level:      256
Time taken for tests:   1.643 seconds
Complete requests:      10000
Failed requests:        0
Keep-Alive requests:    10000
Total transferred:      1680000 bytes
HTML transferred:       150000 bytes
Requests per second:    6086.22 [#/sec] (mean)
Time per request:       42.062 [ms] (mean)
Time per request:       0.164 [ms] (mean, across all concurrent requests)
Transfer rate:          998.52 [Kbytes/sec] received

我们发现，此时的QPS还是有一定的提升的。（这里，视频中压测的时候，会夯住请求，导致QPS非常的低，但是我实际测试的时候没有发生这个情况，估计是和Redis服务器本身对连接个数的的配置有关）

但是，为了避免请求数量过多，导致创建连接个数过多的问题，我们可以使用一个Redis连接池来解决。（同步阻塞是没有Redis连接过多的问题的，因为一旦worker进程阻塞住了，那么后面的请求就不会继续执行了，也就不会创建新的Redis连接了。因此，在同步阻塞的模式下，Redis的连接数量最大是worker进程的个数）

现在，我们来实现一下Redis连接池：

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
class RedisQueue
{
    protected $pool;

    public function __construct()
    {
        $this->pool = new SplQueue;
    }

    public function get(): Redis
    {
        if ($this->pool->isEmpty()) {
            $redis = new \Redis();
            $redis->connect('127.0.0.1', 6379);
            return $redis;
        }
        return $this->pool->dequeue();
    }

    public function put(Redis $redis)
    {
        $this->pool->enqueue($redis);
    }

    public function close()
    {
        $this->pool = null;
    }
}

这里通过spl的队列实现的连接池。如果连接池中没有连接的时候，我们就新建一个连接，并且把创建的这个连接返回；如果连接池里面有连接，那么我们获取队列中前面的一个连接。当我们用完连接的时候，，就可以调用put方法归还连接。这样，我们就可以在一定程度上复用Redis的连接，缓解Redis服务器的压力，以及频繁创建Redis连接的开销也会降低。

我们现在使用这个连接池队列：

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
74
<?php

use Swoole\Http\Request;
use Swoole\Http\Response;

$process = new Swoole\Process(function (Swoole\Process $process) {
    $server = new Swoole\Http\Server('127.0.0.1', 9501, SWOOLE_BASE);
    $server->set([
        'log_file' => '/dev/null',
        'log_level' => SWOOLE_LOG_INFO,
        'worker_num' => swoole_cpu_num() * 2,
        'hook_flags' => SWOOLE_HOOK_ALL,
    ]);
    $server->on('workerStart', function () use ($process, $server) {
        $server->pool = new RedisQueue;
        $process->write('1');
    });
    $server->on('request', function (Request $request, Response $response) use ($server) {
        try {
            $redis = $server->pool->get();
            // $redis = new Redis;
            // $redis->connect('127.0.0.1', 6379);
            $greeter = $redis->get('greeter');
            if (!$greeter) {
                throw new RedisException('get data failed');
            }
            $server->pool->put($redis);
            $response->end("<h1>{$greeter}</h1>");
        } catch (\Throwable $th) {
            $response->status(500);
            $response->end();
        }
    });
    $server->start();
});

if ($process->start()) {
    register_shutdown_function(function () use ($process) {
        $process::kill($process->pid);
        $process::wait();
    });
    $process->read(1);
    System('ab -c 256 -n 10000 -k http://127.0.0.1:9501/ 2>&1');
}

class RedisQueue
{
    protected $pool;

    public function __construct()
    {
        $this->pool = new SplQueue;
    }

    public function get(): Redis
    {
        if ($this->pool->isEmpty()) {
            $redis = new \Redis();
            $redis->connect('127.0.0.1', 6379);
            return $redis;
        }
        return $this->pool->dequeue();
    }

    public function put(Redis $redis)
    {
        $this->pool->enqueue($redis);
    }

    public function close()
    {
        $this->pool = null;
    }
}

我们在worker进程初始化的时候，创建了这个RedisQueue。然后在onRequest的阶段，从这个RedisQueue里面获取一个Redis连接。

现在，我们来进行压测：

1
2
3
4
5
6
7
8
9
10
11
Concurrency Level:      256
Time taken for tests:   1.188 seconds
Complete requests:      10000
Failed requests:        0
Keep-Alive requests:    10000
Total transferred:      1680000 bytes
HTML transferred:       150000 bytes
Requests per second:    8416.18 [#/sec] (mean)
Time per request:       30.418 [ms] (mean)
Time per request:       0.119 [ms] (mean, across all concurrent requests)
Transfer rate:          1380.78 [Kbytes/sec] received

QPS提升到了8416.18。

但是，通过splqueue实现的连接池是有缺陷的，因为这个队列是可以无限长的。这样，当并发量特别大的时候，还是会有可能创建非常多的连接，因为连接池里面可能始终都是空的。

这个时候，我们可以使用Channel来实现连接池。代码如下：

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
class RedisPool
{
    protected $pool;

    public function __construct(int $size = 100)
    {
        $this->pool = new \Swoole\Coroutine\Channel($size);
        for ($i = 0; $i < $size; $i++)
        {
            while (true) {
                try {
                    $redis = new \Redis();
                    $redis->connect('127.0.0.1', 6379);
                    $this->put($redis);
                    break;
                } catch (\Throwable $th) {
                    usleep(1 * 1000);
                    continue;
                }
            }
        }
    }

    public function get(): \Redis
    {
        return $this->pool->pop();
    }

    public function put(\Redis $redis)
    {
        $this->pool->push($redis);
    }

    public function close()
    {
        $this->pool->close();
        $this->pool = null;
    }
}

可以看到，我们在这个构造方法中，将这个Channel的size设置为这个传入的参数。并且，创建size个连接。这些连接会在初始化连接池的时候就被创建，处于就就绪状态。这个有好处也有坏处，坏处就是在每个进程初始化的时候，就会占用一些连接，但是此时的进程并不会接收连接。好处就是提前创建好了Redis连接，这样服务器响应的延迟就会降低。

虽然，其他地方的代码其实和RedisQueue的实现一样。但是，底层是和RedisQueue大有不同的。因为当Channel里面没有Redis连接的时候，会让当前的协程挂起，让其他的协程继续被执行。等有协程把Redis连接还回到连接池里面的时候，这个被挂起的协程才会继续执行。这就是协程协作的原理。

现在，我们修改服务器的代码：

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
74
75
76
77
78
79
80
81
82
83
84
<?php

use Swoole\Http\Request;
use Swoole\Http\Response;

$process = new Swoole\Process(function (Swoole\Process $process) {
    $server = new Swoole\Http\Server('127.0.0.1', 9501, SWOOLE_BASE);
    $server->set([
        'log_file' => '/dev/null',
        'log_level' => SWOOLE_LOG_INFO,
        'worker_num' => swoole_cpu_num() * 2,
        'hook_flags' => SWOOLE_HOOK_ALL,
    ]);
    $server->on('workerStart', function () use ($process, $server) {
        $server->pool = new RedisPool(64);
        $process->write('1');
    });
    $server->on('request', function (Request $request, Response $response) use ($server) {
        try {
            $redis = $server->pool->get();
            // $redis = new Redis;
            // $redis->connect('127.0.0.1', 6379);
            $greeter = $redis->get('greeter');
            if (!$greeter) {
                throw new RedisException('get data failed');
            }
            $server->pool->put($redis);
            $response->end("<h1>{$greeter}</h1>");
        } catch (\Throwable $th) {
            $response->status(500);
            $response->end();
        }
    });
    $server->start();
});

if ($process->start()) {
    register_shutdown_function(function () use ($process) {
        $process::kill($process->pid);
        $process::wait();
    });
    $process->read(1);
    System('ab -c 256 -n 10000 -k http://127.0.0.1:9501/ 2>&1');
}

class RedisPool
{
    protected $pool;

    public function __construct(int $size = 100)
    {
        $this->pool = new \Swoole\Coroutine\Channel($size);
        for ($i = 0; $i < $size; $i++)
        {
            while (true) {
                try {
                    $redis = new \Redis();
                    $redis->connect('127.0.0.1', 6379);
                    $this->put($redis);
                    break;
                } catch (\Throwable $th) {
                    usleep(1 * 1000);
                    continue;
                }
            }
        }
    }

    public function get(): \Redis
    {
        return $this->pool->pop();
    }

    public function put(\Redis $redis)
    {
        $this->pool->push($redis);
    }

    public function close()
    {
        $this->pool->close();
        $this->pool = null;
    }
}

只需要修改workerStart里面的部分即可，其他地方不需要做修改。这样，每个进程最多只能创建64个Redis连接。

我们继续压测：

1
2
3
4
5
6
7
8
9
10
11
Concurrency Level:      256
Time taken for tests:   0.817 seconds
Complete requests:      10000
Failed requests:        0
Keep-Alive requests:    10000
Total transferred:      1680000 bytes
HTML transferred:       150000 bytes
Requests per second:    12234.30 [#/sec] (mean)
Time per request:       20.925 [ms] (mean)
Time per request:       0.082 [ms] (mean, across all concurrent requests)
Transfer rate:          2007.19 [Kbytes/sec] received

我们发现QPS还是有所提升的，为什么我的QPS没有视频里面的提升明显呢？这个和测试环境有关。我自己的机器已经无法再提升了。就好像学霸最高只可以考100分一样的道理。（实际上，经过我的测试，如果我调整连接池的最大连接数，QPS会有所提升）

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

感谢twosee大佬的点拨。

这篇文章，我们来讲一下PHP扩展开发的常见问题，引用计数管理。因为现有的书籍有些老旧，跟不上PHP的发展，所以很有必要学习一下。

PHP版本是7.3.5。

我们来写一个测试扩展函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
ZEND_BEGIN_ARG_INFO_EX(arginfo_study_test, 0, 0, 1)
    ZEND_ARG_ARRAY_INFO(0, arr, 0)
ZEND_END_ARG_INFO()

PHP_FUNCTION(test)
{
    zval *arr;

    if(zend_parse_parameters(ZEND_NUM_ARGS(), "a", &arr) == FAILURE){
        RETURN_FALSE;
    }
    RETURN_ARR(Z_ARR_P(arr));
}

这个函数做的事情很简单，接收一个数组，然后直接返回回去。

(小伙伴们自己记得注册一下创建的这个测试函数)

OK。我们编译、安装扩展：

1
2
3
4
5
6
7
8
9
~/codeDir/cppCode/study # make clean ; make ; make install
----------------------------------------------------------------------

Build complete.
Don't forget to run 'make test'.

Installing shared extensions:     /usr/local/lib/php/extensions/no-debug-non-zts-20180731/
Installing header files:          /usr/local/include/php/
~/codeDir/cppCode/study #

然后编写测试脚本：

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

$a = array("time" => time());
xdebug_debug_zval("a");
$b = test($a);
xdebug_debug_zval("a");
xdebug_debug_zval("b");

unset($a);

var_dump($b);

执行：

1
2
3
4
5
6
7
~/codeDir/cppCode/study # php test.php
a: (refcount=1, is_ref=0)=array ('time' => (refcount=0, is_ref=0)=1569290383)
a: (refcount=1, is_ref=0)=array ('time' => (refcount=0, is_ref=0)=1569290383)
b: (refcount=1, is_ref=0)=array ('time' => (refcount=0, is_ref=0)=1569290383)
/root/codeDir/cppCode/study/test.php:11:
&array
~/codeDir/cppCode/study #

我们发现，没有打印出$b这个数组。出现了bug，我们分析一下。

因为PHP对于复杂类型，例如字符串、数组、对象都是通过引用计数来管理的，不会拷贝出一个副本。所以，当我们在PHP脚本传递一个数组进入PHP扩展的时候，实际上只是增加了这个数组的引用计数。我们可以用gdb来调试一下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
 34│ PHP_FUNCTION(test)
 35│ {
 36│     zval *arr;
 37│
 38│     if(zend_parse_parameters(ZEND_NUM_ARGS(), "a", &arr) == FAILURE){
 39│         RETURN_FALSE;
 40│     }
 41├───> RETURN_ARR(Z_ARR_P(arr));
 42│ }

(gdb) p *arr.value.arr
$3 = {gc = {refcount = 2, u = {type_info = 23}}, u = {v = {flags = 24 '\030', _unused = 0 '\000', nIteratorsCount = 0 '\000', _unused2 = 0 '\000'}, flags = 24}, nTableMask = 4294967280, arData = 0x7ffff76
66680, nNumUsed = 1, nNumOfElements = 1, nTableSize = 8, nInternalPointer = 0, nNextFreeElement = 0, pDestructor = 0x555555a46860 <zval_ptr_dtor>}
(gdb)

我们发现，在调用了zend_parse_parameters解析出PHP脚本传递过来的数组之后，这个数组的引用计数变成了2。这一步增加引用计数的操作是PHP底层自动帮我们做的。

然后，通过xdebug打印发现，从扩展里面返回数组到PHP脚本后，它的引用计数又变回了1。这一步也是PHP帮我们做的。

接下来，我们unset了$a，使得这个数组的引用计数变成了0。又因为$b和$a指向的是一个数组，所以，此时我们再使用$b，就会报错了。

以上bug会在动态生成数组时出现。如果我们初始的数组是一个不可变数组，那么，同样的代码是不会出现bug的。因为不可变数组的初始引用计数是2，而不是1。我们可以测试一下：

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

$a = array("time" => 1111);
xdebug_debug_zval("a");
$b = test($a);
xdebug_debug_zval("a");
xdebug_debug_zval("b");

unset($a);

var_dump($b);

我们把time对应的value写死为1111。然后执行脚本：

1
2
3
4
5
6
7
8
9
10
~/codeDir/cppCode/study # php test.php
a: (refcount=2, is_ref=0)=array ('time' => (refcount=0, is_ref=0)=1111)
a: (refcount=2, is_ref=0)=array ('time' => (refcount=0, is_ref=0)=1111)
b: (refcount=2, is_ref=0)=array ('time' => (refcount=0, is_ref=0)=1111)
/root/codeDir/cppCode/study/test.php:11:
array(1) {
  'time' =>
  int(1111)
}
~/codeDir/cppCode/study #

我们发现，因为不可变数组初始的引用计数是2，当我们对unset($a)之后，它的引用计数变成了1。此时，这个数组还是可以用的，所以后面使用$b不会报错。

正是因为从PHP脚本传入到扩展，以及从扩展传出到PHP脚本的那个数组是同一个，所以，$a和$b都是指向同一个数组的，因此，我们需要在扩展层面手动为这个数组的引用计数+1。所以，扩展代码应该改为：

1
2
3
4
5
6
7
8
9
10
PHP_FUNCTION(test)
{
    zval *arr;

    if(zend_parse_parameters(ZEND_NUM_ARGS(), "a", &arr) == FAILURE){
        RETURN_FALSE;
    }
    Z_TRY_ADDREF_P(arr);
    RETURN_ARR(Z_ARR_P(arr));
}

重新编译、安装扩展：

1
2
3
4
5
6
7
8
9
~/codeDir/cppCode/study # make clean ; make ; make install
----------------------------------------------------------------------

Build complete.
Don't forget to run 'make test'.

Installing shared extensions:     /usr/local/lib/php/extensions/no-debug-non-zts-20180731/
Installing header files:          /usr/local/include/php/
~/codeDir/cppCode/study #

测试脚本为：

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

$a = array("time" => time());
xdebug_debug_zval("a");
$b = test($a);
xdebug_debug_zval("a");
xdebug_debug_zval("b");

unset($a);

var_dump($b);

指向脚本：

1
2
3
4
5
6
7
8
9
10
~/codeDir/cppCode/study # php test.php
a: (refcount=1, is_ref=0)=array ('time' => (refcount=0, is_ref=0)=1569291353)
a: (refcount=2, is_ref=0)=array ('time' => (refcount=0, is_ref=0)=1569291353)
b: (refcount=2, is_ref=0)=array ('time' => (refcount=0, is_ref=0)=1569291353)
/root/codeDir/cppCode/study/test.php:11:
array(1) {
  'time' =>
  int(1569291353)
}
~/codeDir/cppCode/study #

我们发现，从扩展传回到PHP的时候，这个数组的引用计数变成了2。因此，当我们unset($a)之后，这个数组的引用计数变成了1。此时，我们再次使用$b就不会出错了。

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

这篇文章，我们来实战操作一下扩展的内存管理，感受一下内存泄漏，加深对PHP内存管理的理解。

首先，创建扩展目录：

1
~/codeDir/cCode/php-7.1.0/ext # ./ext_skel --extname=memory

然后进入目录：

1
~/codeDir/cCode/php-7.1.0/ext # cd memory/

替换文件config.m4为如下内容：

1
2
3
4
5
6
7
PHP_ARG_ENABLE(memory, whether to enable memory support,
Make sure that the comment is aligned:
[  --enable-memory           Enable memory support])

if test "$PHP_MEMORY" != "no"; then
  PHP_NEW_EXTENSION(memory, memory.c, $ext_shared,, -DZEND_ENABLE_STATIC_TSRMLS_CACHE=1)
fi

然后编辑文件memory.c里面的PHP_FUNCTION(confirm_memory_compiled)方法：

1
2
3
4
PHP_FUNCTION(confirm_memory_compiled)
{
  void *foo = malloc(2 * 1024 * 1024);
}

这里，我们定义了一个PHP函数confirm_memory_compiled。它做的事情很简单，就是从堆中申请一块2M的内存，并且没有主动释放。

接着，编译、安装扩展：

1
2
~/codeDir/cCode/php-7.1.0/ext/memory # phpize ; ./configure
~/codeDir/cCode/php-7.1.0/ext/memory # make ; make install

然后把扩展加入配置文件里面：

1
extension=memory.so

然后确认是否安装扩展成功：

1
2
3
4
5
6
~/codeDir/cCode/php-7.1.0/ext/memory # php --ri memory

memory

memory support => enabled
~/codeDir/cCode/php-7.1.0/ext/memory #

然后编写测试脚本：

1
2
<?php
confirm_memory_compiled();

执行脚本：

1
2
~/codeDir/cCode/php-7.1.0/ext/memory # php memory.php
~/codeDir/cCode/php-7.1.0/ext/memory #

没有报错，说明我们的脚本正常执行了。

接着，我们启动一个PHP自带的服务器：

1
2
3
4
5
~/codeDir/cCode/php-7.1.0/ext/memory # php -S 127.0.0.1:80 -t ./
PHP 7.3.5 Development Server started at Mon Sep 23 13:08:13 2019
Listening on http://127.0.0.1:80
Document root is /root/codeDir/cCode/php-7.1.0/ext/memory
Press Ctrl-C to quit.

然后，另起一个终端，执行top命令，用来观察PHP进程的内存使用情况：

1
35247 root      20   0   27.8m  14.7m   0.0   0.7   0:00.02 S  `- php -S 127.0.0.1:80 -t ./

可以看到，在启动服务器的时候，PHP占用了27.8M的内存。

我们请求一次我们的服务器：

1
~/codeDir/cppCode/study # curl 127.0.0.1/memory.php

然后查看PHP占的内存：

1
13713 root      20   0   29.8m  14.9m   0.0   0.7   0:00.01 S  `- php -S 127.0.0.1:80 -t ./

我们发现PHP多占了2M的内存。

我们再请求一次：

1
~/codeDir/cppCode/study # curl 127.0.0.1/memory.php

然后再次查看PHP占的内存：

1
13713 root      20   0   31.8m  15.0m   0.0   0.8   0:00.03 S  `- php -S 127.0.0.1:80 -t ./

我们发现PHP又多占了2M的内存。

所以说，如果我们在一次请求的生命周期通过malloc分配了内存，但是没有释放，那么就会造成PHP整个生命周期的内存泄漏。

我们修改扩展函数：

1
2
3
4
PHP_FUNCTION(confirm_memory_compiled)
{
  void *foo = emalloc(2 * 1024 * 1024);
}

然后，重新编译、安装扩展：

1
~/codeDir/cCode/php-7.1.0/ext/memory # make clean ; make ; make install

重新启动服务器：

1
2
3
4
5
~/codeDir/cCode/php-7.1.0/ext/memory # php -S 127.0.0.1:80 -t ./
PHP 7.3.5 Development Server started at Mon Sep 23 14:27:22 2019
Listening on http://127.0.0.1:80
Document root is /root/codeDir/cCode/php-7.1.0/ext/memory
Press Ctrl-C to quit.

此时，PHP进程占用的内存：

1
14470 root      20   0   27.8m  14.6m   0.0   0.7   0:00.01 S  `- php -S 127.0.0.1:80 -t ./

然后，我们请求一次服务器：

1
~/codeDir/cppCode/study # curl 127.0.0.1/memory.php

查看PHP内存占用情况：

1
14470 root      20   0   27.8m  15.2m   0.0   0.8   0:00.02 S  `- php -S 127.0.0.1:80 -t ./

发现，没有增长。

再次请求服务器：

1
~/codeDir/cppCode/study # curl 127.0.0.1/memory.php

再次查看PHP内存占用情况：

1
14470 root      20   0   27.8m  15.2m   0.7   0.8   0:00.03 S  `- php -S 127.0.0.1:80 -t ./

发现还是没有增长。

所以说，如果我们在一次请求的生命周期中通过emalloc分配了内存，但是没有释放，那么在PHP整个生命周期是不会造成内存泄漏的。因为在请求结束的时候，PHP会自动帮我们释放掉这些内存。但是，在一次请求中，如果一直不自己释放内存，那么这次请求很可能会内存不够，导致PHP进程挂掉。

以上对malloc和emalloc的分析适用于FPM模式。但是对于Swoole这类扩展，接管了PHP的请求生命周期，所有对Swoole的请求都是在同一个请求生命周期里面，并且，这个请求生命周期一直不会结束。所以，就算我们使用了emalloc这类内存管理器，如果没有主动释放，也是会造成内存泄漏的，因为此时PHP的请求生命周期不会结束，因此PHP不会自己帮我们去释放这些内存。

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

为了给我们的扩展实现Channel做准备，这里，很有必要简单分析一下Swoole协程的Channel实现原理。核心点如下：

1
2
3
4
5
6
1、什么情况下可以pop
2、当channel不可以pop的时候，应该如何处理消费者协程
3、什么情况下可以push
4、当channel不可以push的时候，应该如何处理生产者协程
5、当channel可以pop的时候，应该如何通知消费者协程
6、当channel可以push的时候，应该如何通知生产者协程

如果解决了这些问题，就可以去实现Channel了。

什么情况下可以pop

这个问题很简单，当channel里面有数据的时候。我们来看看Swoole对应的源码，在方法Channel::pop(double timeout)里面：

1
2
void *data = data_queue.front();
data_queue.pop();

消费者协程和生产者协程

我们定义一个协程是消费者协程还是生产者协程，取决于这个协程正在执行哪种操作。如果这个协程此时正在执行pop操作，那么这个协程此时就是消费者协程；如果这个协程此时正在执行push操作，那么这个协程此时就是生产者协程。也就是说，协程是生产者还是消费者不是死的，是会随着协程的操作动态变化的。

当channel不可以pop的时候，应该如何处理消费者协程

当channel不可以pop的时候，我们应该挂起这个消费者协程。我们来看看Swoole对应的代码：

1
2
3
4
5
6
if (is_empty() || !consumer_queue.empty())
{
    // 省略其他代码
    yield(CONSUMER);
    // 省略其他代码
}

我们可以看到，当Channel为空的时候，消费者协程是不可以进行pop操作的，此时被yield出去了。

什么情况下可以push

这个问题很简单，当channel容器没有满的时候。我们来看看Swoole对应的源码，在方法Channel::push(void *data, double timeout)里面：

1
2
data_queue.push(data);
swTraceLog(SW_TRACE_CHANNEL, "push data to channel, count=%ld", length());

当channel不可以push的时候，应该如何处理生产者协程

当channel不可以push的时候，我们应该挂起这个生产者协程。我们来看看Swoole对应的代码：

1
2
3
4
5
6
if (is_full() || !producer_queue.empty())
{
    // 省略其他的代码
    yield(PRODUCER);
    // 省略其他的代码
}

我们可以看到，当Channel满了的时候，生产者协程是不可以进行push操作的，此时被yield出去了。

当channel可以pop的时候，应该如何通知消费者协程

首先，我们这里需要明白，是谁在通知消费者协程。是Swoole的那套事件循环吗？不是的。

通知消费这协程的是生产者协程。我们来看看pop的逆操作push的代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
/**
 * push data
 */
data_queue.push(data);
swTraceLog(SW_TRACE_CHANNEL, "push data to channel, count=%ld", length());
/**
 * notify consumer
 */
if (!consumer_queue.empty())
{
    Coroutine *co = pop_coroutine(CONSUMER);
    co->resume();
}

可以看到，当生产者协程push完数据的时候，此时channel必定是有数据的。然后，如果有消费者协程在等待channel的话，那么就唤醒第一个等待的那个消费者协程。（所以规则是：谁先等待channel，谁就先被唤醒）

当channel可以push的时候，应该如何通知生产者协程

这个的原理和上一个问题的答案类似，不重复说了。一句话：消费者通知的生产者协程。

发表于
字数统计 1.4k 字 | 阅读时长 6 分钟

我们先来看一段代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
#include <stdio.h>
#include <sys/types.h>
#include <unistd.h>
#include <sys/wait.h>

int main()
{
    pid_t id;

    id = fork();
    if (id == 0)
    {
        printf("Hello from Child!, pid: %d\n", getpid());
    }
    else
    {
        printf("Hello from Parent!, pid: %d\n", getpid());
    }

    return 0;
}

执行结果如下：

1
2
3
4
~/codeDir/cCode/test # ./a.out
Hello from Parent!, pid: 2290
Hello from Child!, pid: 2291
~/codeDir/cCode/test #

我们看看进程状态：

1
2
3
4
~/codeDir # ps -a
  PID TTY          TIME CMD
 2292 pts/2    00:00:00 ps
~/codeDir #

我们发现，进程直接退出了。

然后，我们再来看一段代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
#include <stdio.h>
#include <sys/types.h>
#include <unistd.h>
#include <sys/wait.h>

int main()
{
    pid_t id;

    id = fork();
    if (id == 0)
    {
        printf("Hello from Child!, pid: %d\n", getpid());
        sleep(-1);
    }
    else
    {
        printf("Hello from Parent!, pid: %d\n", getpid());
    }

    return 0;
}

我们让子进程无限阻塞住，然后父进程先退出。执行结果如下：

1
2
3
4
~/codeDir/cCode/test # ./a.out
Hello from Parent!, pid: 2314
Hello from Child!, pid: 2315
~/codeDir/cCode/test #

我们发现，终端不会卡住，而是直接退出了。说明，终端是否会卡住，是由父进程是否退出决定的。然后，我们看看进程状态：

1
2
3
4
5
~/codeDir # ps -a
  PID TTY          TIME CMD
 2315 pts/4    00:00:00 a.out
 2316 pts/2    00:00:00 ps
~/codeDir #

我们发现，此时子进程（PID是2315）还存在。因为进程2315的父进程2314不在了，所以进程2315会变成孤儿进程，此时，它的父进程会变成init进程。

此时，我们可以通过wait函数来等待子进程的结束：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
#include <stdio.h>
#include <sys/types.h>
#include <unistd.h>
#include <sys/wait.h>

int main()
{
    pid_t id;

    id = fork();
    if (id == 0)
    {
        printf("Hello from Child!, pid: %d\n", getpid());
        sleep(-1);
    }
    else
    {
        wait(NULL);
        printf("Hello from Parent!, pid: %d\n", getpid());
    }

    return 0;
}

其中，wait可以等待子进程结束。代码执行结果如下：

1
2
~/codeDir/cCode/test # ./a.out
Hello from Child!, pid: 2327

我们发现，此时的终端没有直接退出。我们查看一下进程状态：

1
2
3
4
5
6
~/codeDir # ps -a
  PID TTY          TIME CMD
 2326 pts/4    00:00:00 a.out
 2327 pts/4    00:00:00 a.out
 2329 pts/2    00:00:00 ps
~/codeDir #

我们发现，父进程和子进程都没挂掉。正是因为我们子进程一直没有退出，所以父进程阻塞在了wait这个地方。

我们再来看一段代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
#include <stdio.h>
#include <sys/types.h>
#include <unistd.h>
#include <sys/wait.h>

int main()
{
    pid_t id;

    id = fork();
    if (id == 0)
    {
        printf("Hello from Child!, pid: %d\n", getpid());
    }
    else
    {
        printf("Hello from Parent!, pid: %d\n", getpid());
        sleep(-1);
    }

    return 0;
}

目的是，子进程退出后，父进程一直阻塞住不退出。

执行结果如下：

1
2
3
~/codeDir/cCode/test # ./a.out
Hello from Parent!, pid: 2337
Hello from Child!, pid: 2338

因为父进程没有退出，所以终端会被卡住。此时，我们看看进程状态。

1
2
3
4
5
6
~/codeDir # ps -a
  PID TTY          TIME CMD
 2337 pts/4    00:00:00 a.out
 2338 pts/4    00:00:00 a.out <defunct>
 2339 pts/2    00:00:00 ps
~/codeDir #

我们发现，这里依然会看到子进程2338的信息，为什么呢？因为这个子进程目前是一个僵尸进程。我们发现这个进程旁边有一个标识defunct，代表这个进程执行完了它的任务，已经挂了，但是，这个进程的一些基本信息依然保存着。这个僵尸进程的信息依旧被内核保存在进程表里面。

我们现在用kill命令给僵尸进程发送信号看看：

1
2
3
4
5
6
7
~/codeDir # kill 2338
~/codeDir # ps -a
  PID TTY          TIME CMD
 2337 pts/4    00:00:00 a.out
 2338 pts/4    00:00:00 a.out <defunct>
 2342 pts/2    00:00:00 ps
~/codeDir #

我们发现，这个僵尸进程还是会被查找到？为什么呢？因为这个僵尸进程本来就是死的，当然kill会失效。那么，我们如何去避免这个问题呢？

可以用wait函数。wait函数除了可以等待子进程的结束，其实还可以获取到子进程的信息。这也是为什么一个子进程挂了之后，它的基本信息不会立马被内核清除的原因，因为，操作系统认为父进程会在某个时候要用到子进程的一些信息，例如子进程的退出状态。只有当父进程获取完子进程的信息之后，操作系统才会清理子进程残留的信息。而操作系统如何知道父进程读取过子进程的信息呢？那就是wait函数。实际上，wait函数可以传入一个获取子进程信息的结构体，只不过我们的代码没有去获取而已，说明此时我们是不关心子进程的状态的。

分析完之后，我们来看下一段代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
#include <stdio.h>
#include <sys/types.h>
#include <unistd.h>
#include <sys/wait.h>

int main()
{
    pid_t id;

    id = fork();
    if (id == 0)
    {
        printf("Hello from Child!, pid: %d\n", getpid());
        sleep(1);
    }
    else
    {
        wait(NULL);
        printf("Hello from Parent!, pid: %d\n", getpid());
        sleep(-1);
    }

    return 0;
}

这里，我们的父进程会去调用wait函数，等待子进程结束。等子进程执行完毕之后，父进程才回去执行。一旦父进程调用了wait函数，操作系统就知道父进程获取过子进程的信息，因此，操作系统就可以放心的去清理僵尸进程的残留信息了。

我们执行代码：

1
2
3
~/codeDir/cCode/test # ./a.out
Hello from Child!, pid: 2350
Hello from Parent!, pid: 2349

我们看看进程的状态：

1
2
3
4
5
~/codeDir # ps -a
  PID TTY          TIME CMD
 2349 pts/4    00:00:00 a.out
 2351 pts/2    00:00:00 ps
~/codeDir #

我们发现，子进程2350不再出现在进程状态列表里面了，说明子进程已经被清理完毕了。

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

有一个需求，是把表里面所有的数据都查询出来，并且生成json文件。因为一张表里面的数据很多，所以不可能一次性全部查询出来，所以需要用到chunk。之前用的gorm，但是发现gorm没有chunk方式的查询。如果要自己去实现这种操作，就需要去管理偏移量，而且还容易出现bug，所以就找了一个库，叫做gorose。用起来挺舒服的。

代码如下：

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
package main

import (
    "fmt"

    "github.com/gohouse/gorose"
)

// User struct
// type User struct {
//  ID       int
//  UserName string
// }

const (
    dbHost     = "tcp(host.docker.internal:3306)"
    dbName     = "test"
    dbUser     = "root"
    dbPassword = "123456"
)

func main() {
    dsn := dbUser + ":" + dbPassword + "@" + dbHost + "/" + dbName + "?charset=utf8"
    var dbConfig = gorose.DbConfigSingle{
        Driver: "mysql",
        Dsn:    dsn,
    }

    connection, err := gorose.Open(&dbConfig)
    if err != nil {
        fmt.Println(err)
        return
    }

    session := connection.NewSession()

    user := session.Table("users")
    user.Fields("id", "username", "number").Chunk(2, func(data []map[string]interface{}) {
    fmt.Println(data)
    })
}

执行结果如下：

1
2
3
4
5
~/codeDir/golangCode/test # go run main.go
[map[id:1 username:a number:1] map[id:2 username:b number:2]]
[map[number:3 id:3 username:c] map[id:4 username:d number:4]]
[map[id:5 username:e number:5]]
~/codeDir/golangCode/test #

可以看出，每次都会查询出2条记录。

这个框架一个缺点就是文档不是很清楚，报错也有点不习惯。但是先用这个库解决一下chunk查询的问题吧。