Swoole Server中master进程投递数据到worker进程的性能优化

Comments

在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减少了一次内存拷贝。

codinghuangPHP Developer

PHP开发者~