swoole的整体数据流程图可以看官方文档swoole运行流程
在主进程中有两个关键线程 主线程和reactor线程， 主线程用来处理 accept() 事件，创建新的socket fd， 接收到新连接后会把新的socket 连接放到reactor线程的的事件监听循环中， reactor线程负责接收从client端发过来的数据，按协议解析后，通过pipe管道传输给worker进程处理，worker进程处理完后，把结果通过pipe回传给reactor线程，reactor线程再按协议把结果通过socket发送给client。 可以看到reactor线程负责数据的IO和传输，在linux环境下，这些IO事件通过epoll机制来处理。
下一步我们看下主线程和 reactor线程数据交换的过程的源码
第一步要找到 swServer_start(swServer *serv) 函数， 这个是swoole启动server的入口，在这个函数中很容易就能读到这么一句
int swServer_start(swServer *serv) {
......
ret = swServer_start_proxy(serv); //启动主线程和reactor线程，并注册各种事件处理函数，和协议解析
......
}
下面着重讲 swServer_start_proxy
//src/network/Server.c
static int swServer_start_proxy(swServer *serv)
{
int ret;
swReactor *main_reactor = SwooleG.memory_pool->alloc(SwooleG.memory_pool, sizeof(swReactor));
ret = swReactor_create(main_reactor, SW_REACTOR_MAXEVENTS); // 创建主线程reactor对象事件处理，只是创建对象，主线程不用pthread_create() 真正去创建线程
//...... // 省略区
ret = swReactorThread_start(serv, main_reactor); //把server监听端口的listen事件加入到主线程事件监听，创建reactor线程
//...... // 省略区
main_reactor->setHandle(main_reactor, SW_FD_LISTEN, swServer_master_onAccept); //主线程reactor listen事件处理handle
//...... // 省略区
return main_reactor->wait(main_reactor, NULL); //主线程循环
}
int swReactorThread_start(swServer *serv, swReactor *main_reactor_ptr)
{
//..... 省略区
swListenPort *ls;
LL_FOREACH(serv->listen_list, ls)
{
if (ls->type == SW_SOCK_UDP || ls->type == SW_SOCK_UDP6 || ls->type == SW_SOCK_UNIX_DGRAM)
{
continue;
}
main_reactor_ptr->add(main_reactor_ptr, ls->sock, SW_FD_LISTEN); //把监听的端口fd加入到 主线程事件监听循环中
}
//..... 省略区
return SW_OK;
}
int swServer_master_onAccept(swReactor *reactor, swEvent *event)
{
//..... 省略区
//SW_ACCEPT_AGAIN
for (i = 0; i < SW_ACCEPT_MAX_COUNT; i++)
{
new_fd = accept(event->fd, (struct sockaddr *) &client_addr, &client_addrlen); // accept新socket
//..... 省略区
if (sub_reactor->add(sub_reactor, new_fd, SW_FD_TCP | SW_EVENT_WRITE) < 0) //新socket加入到 reactor线程事件监听中
{
bzero(conn, sizeof(swConnection));
close(new_fd);
return SW_OK;
}
}
return SW_OK;
}
- swReactor_create() ， 创建主线程reactor对象，只是创建对象，主线程不用pthread_create() 真正去创建线程，主线程就在主进程中
- swReactorThread_start() , 这里面主要分成两件事， 一个是把server监听端口的listen事件加入到主线程事件监听中去，这样一旦有连接达到监听端口就会调用主线程的 swServer_master_onAccept()来处理新连接。 另外一件事就是创建reactor线程并注册各种事件监听和协议处理
3.main_reactor->setHandle() , 这个注册主线程reactor listen事件处理handle ， 也就是注册swServer_master_onAccept()
4.main_reactor->wait(main_reactor, NULL); 这个是主线程循环，主线程一直在这个函数中循环，等待新连接，然后把新连接通过swServer_master_onAccept() ，拿到新的socket， 再把新的socket ， 通过 sub_reactor->add(sub_reactor, new_fd, SW_FD_TCP | SW_EVENT_WRITE) 加入到reactor线程的事件处理循环中
同过上面的几部可以清楚的看到主线程是如何监听listen事件，又是如何accept()新连接， 并把新连接加入到reactor线程的监听中的
下面继续说reactor线程
reactor线程通过 swReactorThread_start() ， 调用创建
//src/network/ReactorThread.c 函数中间有省略，只粘贴了关键部分
int swReactorThread_start(swServer *serv, swReactor *main_reactor_ptr)
{
//..... 省略区
//create reactor thread
for (i = 0; i < serv->reactor_num; i++)
{
thread = &(serv->reactor_threads[i]);
param = SwooleG.memory_pool->alloc(SwooleG.memory_pool, sizeof(swThreadParam));
if (param == NULL)
{
swError("malloc failed");
return SW_ERR;
}
param->object = serv;
param->pti = i;
if (pthread_create(&pidt, NULL, (void * (*)(void *)) swReactorThread_loop, (void *) param) < 0)
{
swError("pthread_create[tcp_reactor] failed. Error: %s[%d]", strerror(errno), errno);
}
thread->thread_id = pidt;
}
return SW_OK;
}
pthread_create 创建reactor线程， 线程的循环函数是swReactorThread_loop
//src/network/ReactorThread.c 函数中间有省略，只粘贴了关键部分
static int swReactorThread_loop(swThreadParam *param)
{
//创建线程reactor对象
swReactor *reactor = &thread->reactor;
ret = swReactor_create(reactor, SW_REACTOR_MAXEVENTS);
//注册pipe管道的事件处理handle
reactor->setHandle(reactor, SW_FD_CLOSE, swReactorThread_onClose);
reactor->setHandle(reactor, SW_FD_PIPE | SW_EVENT_READ, swReactorThread_onPipeReceive);
reactor->setHandle(reactor, SW_FD_PIPE | SW_EVENT_WRITE, swReactorThread_onPipeWrite);
//set protocol function point 添加socket read write事件监听， 并注册协议处理函数
swReactorThread_set_protocol(serv, reactor);
// reactor线程循环
reactor->wait(reactor, NULL);
//shutdown
reactor->free(reactor);
return SW_OK;
}
void swReactorThread_set_protocol(swServer *serv, swReactor *reactor)
{
//UDP Packet
reactor->setHandle(reactor, SW_FD_UDP, swReactorThread_onPackage);
//Write socket 可写事件处理handle
reactor->setHandle(reactor, SW_FD_TCP | SW_EVENT_WRITE, swReactorThread_onWrite);
//Read socket 可读事件处理handle
reactor->setHandle(reactor, SW_FD_TCP | SW_EVENT_READ, swReactorThread_onRead);
swListenPort *ls;
//listen the all tcp port
LL_FOREACH(serv->listen_list, ls)
{
if (swSocket_is_dgram(ls->type))
{
continue;
}
swPort_set_protocol(ls); //协议处理解析， 比如http、http2、websocket、mqtt等
}
}
swPort_set_protocol 协议处理函数
//src/network/Port.c
void swPort_set_protocol(swListenPort *ls)
{
//Thread mode must copy the data.
//will free after onFinish
if (ls->open_eof_check)
{
if (ls->protocol.package_eof_len > sizeof(ls->protocol.package_eof))
{
ls->protocol.package_eof_len = sizeof(ls->protocol.package_eof);
}
ls->protocol.onPackage = swReactorThread_dispatch;
ls->onRead = swPort_onRead_check_eof;
}
else if (ls->open_length_check)
{
if (ls->protocol.package_length_type != '\0')
{
ls->protocol.get_package_length = swProtocol_get_package_length;
}
ls->protocol.onPackage = swReactorThread_dispatch;
ls->onRead = swPort_onRead_check_length;
}
else if (ls->open_http_protocol)
{
if (ls->open_websocket_protocol)
{
ls->protocol.get_package_length = swWebSocket_get_package_length;
ls->protocol.onPackage = swWebSocket_dispatch_frame;
ls->protocol.package_length_size = SW_WEBSOCKET_HEADER_LEN + SW_WEBSOCKET_MASK_LEN + sizeof(uint64_t);
}
#ifdef SW_USE_HTTP2
else if (ls->open_http2_protocol)
{
ls->protocol.get_package_length = swHttp2_get_frame_length;
ls->protocol.package_length_size = SW_HTTP2_FRAME_HEADER_SIZE;
ls->protocol.onPackage = swReactorThread_dispatch;
}
#endif
ls->onRead = swPort_onRead_http; //**http协议处理**
}
else if (ls->open_mqtt_protocol)
{
ls->protocol.get_package_length = swMqtt_get_package_length;
ls->protocol.onPackage = swReactorThread_dispatch;
ls->onRead = swPort_onRead_check_length;
}
else if (ls->open_redis_protocol)
{
ls->protocol.onPackage = swReactorThread_dispatch;
ls->onRead = swPort_onRead_redis;
}
else
{
ls->onRead = swPort_onRead_raw;
}
}
网友评论