trafficserver的Net模块源码注释
Net的启动流程：
main.cc
main()--->UnixNetProcessor::start()
Net模块的启动
int
UnixNetProcessor::start(int, size_t)
{
EventType etype = ET_NET;
//给NetHandler实例分配空间
netHandler_offset = eventProcessor.allocate(sizeof(NetHandler));
//给PollCont实例分配空间
pollCont_offset = eventProcessor.allocate(sizeof(PollCont));
//UnixNetProcessor对应的事件类型是ET_NET，如果是sslNetProcessor则对应的事件类型是ET_SSL
upgradeEtype(etype);
//从eventProcessor获得net的线程，这事在event模块初始化时做好的
n_netthreads = eventProcessor.n_threads_for_type[etype];
//从eventProcessor获得net的线程数量
netthreads = eventProcessor.eventthread[etype];
//初始化所有Net线程
for (int i = 0; i < n_netthreads; ++i) {
initialize_thread_for_net(netthreads[i]);
#ifndef STANDALONE_IOCORE
extern void initialize_thread_for_http_sessions(EThread *thread, int thread_index);
initialize_thread_for_http_sessions(netthreads[i], i);
#endif
}
RecData d;
d.rec_int = 0;
//设置网络链接数的阈值
change_net_connections_throttle(NULL, RECD_INT, d, NULL);
//sock相关，很少使用，这里先不介绍
if (!netProcessor.socks_conf_stuff) {
socks_conf_stuff = NEW(new socks_conf_struct);
loadSocksConfiguration(socks_conf_stuff);
if (!socks_conf_stuff->socks_needed && socks_conf_stuff->accept_enabled) {
Warning("We can not have accept_enabled and socks_needed turned off" " disabling Socks accept\n");
socks_conf_stuff->accept_enabled = 0;
} else {
socks_conf_stuff = netProcessor.socks_conf_stuff;
}
}
//在页面上显示Net相关的统计信息
#ifdef NON_MODULAR
extern Action *register_ShowNet(Continuation * c, HTTPHdr * h);
if (etype == ET_NET)
statPagesManager.register_http("net", register_ShowNet);
#endif
return 1;
}
main()--->UnixNetProcessor::start()--->initialize_thread_for_net()
顾名思义，这个函数的功能是为网络初始化一个线程，
void
initialize_thread_for_net(EThread *thread)
{
//创建NetHandler、PollCont实例
//NetHandler：用于处理Net相关的所有时间
//PollCont：是一个Poll的continuation（ats的设计思想），包含指向NetHandler和PollDescriptor的指针
//PollDescriptor：Poll的描述封装结构
new((ink_dummy_for_new *) get_NetHandler(thread)) NetHandler();
new((ink_dummy_for_new *) get_PollCont(thread)) PollCont(thread->mutex, get_NetHandler(thread));
get_NetHandler(thread)->mutex = new_ProxyMutex();
PollCont *pc = get_PollCont(thread);
PollDescriptor *pd = pc->pollDescriptor;
//调用NetHandler实例启动，最终会每秒调用NetHandler::mainNetEvent()函数
thread->schedule_imm(get_NetHandler(thread));
#ifndef INACTIVITY_TIMEOUT
//创建InactivityCop实例，InactivityCop会定时（1秒）判断每个链接vc是否可以关闭然后进行关闭处理
InactivityCop *inactivityCop = NEW(new InactivityCop(get_NetHandler(thread)->mutex));
//定时调度 inactivityCop的check_inactivity()函数
thread->schedule_every(inactivityCop, HRTIME_SECONDS(1));
#endif
//注册信号处理函数
thread->signal_hook = net_signal_hook_function;
//创建EventIO实例并初始化
thread->ep = (EventIO*)ats_malloc(sizeof(EventIO));
thread->ep->type = EVENTIO_ASYNC_SIGNAL;
#if HAVE_EVENTFD
//启动EventIO实例,使用epoll注册读事件（不知道epoll的先看一下啊）
thread->ep->start(pd, thread->evfd, 0, EVENTIO_READ);
#else
thread->ep->start(pd, thread->evpipe[0], 0, EVENTIO_READ);
#endif
}
NetHandler的初始化
main()--->UnixNetProcessor::start()--->NetHandler::NetHandler()
设置NetHandler的handler为NetHandler::startNetEvent
NetHandler::NetHandler():Continuation(NULL), trigger_event(0)
{
SET_HANDLER((NetContHandler) & NetHandler::startNetEvent);
}
设置NetHandler的handler为NetHandler::mainNetEvent，并定时调度该函数执行
int
NetHandler::startNetEvent(int event, Event *e)
{
(void) event;
SET_HANDLER((NetContHandler) & NetHandler::mainNetEvent);
e->schedule_every(NET_PERIOD);
trigger_event = e;
return EVENT_CONT;
}
PollCont的初始化
main()--->UnixNetProcessor::start()--->PollCont::PollCont()
PollCont::PollCont(ProxyMutex *m, NetHandler *nh, int pt):Continuation(m), net_handler(nh), poll_timeout(pt)
{
//创建PollDescriptor实例
pollDescriptor = NEW(new PollDescriptor);
//初始化PollDescriptor实例
pollDescriptor->init();
//设置PollCont的handler为 PollCont::pollEvent
SET_HANDLER(&PollCont::pollEvent);
}
PollDescriptor的初始化
main()--->UnixNetProcessor::start()--->PollCont::PollCont()--->init()
PollDescriptor *init()
{
result = 0;
#if TS_USE_EPOLL
nfds = 0;
//创建epoll用的文件描述符
epoll_fd = epoll_create(POLL_DESCRIPTOR_SIZE);
memset(ePoll_Triggered_Events, 0, sizeof(ePoll_Triggered_Events));
memset(pfd, 0, sizeof(pfd));
#endif
......
return this;
}
main()--->UnixNetProcessor::start()--->initialize_thread_for_net()--->NetHandler::mainNetEvent()
这个函数看起来有点长，先说一下它的功能：首先是调用epoll_wait()等待事件，再次是根据事件的类型做不同的处理，事件分为EVENTIO_READWRITE_VC（读写事件）、EVENTIO_DNS_CONNECTION（DNS的CONNECT事件）、EVENTIO_ASYNC_SIGNAL（同步信号事件），正常的HTTP请求的接收和响应属于EVENTIO_READWRITE_VC，DNS请求发送流程时说过，调用connect()函数来发送DNS请求时会调用epoll_ctl()来注册响应的事件，这就是EVENTIO_DNS_CONNECTION，我们先不关心
EVENTIO_ASYNC_SIGNAL。
最后是分别遍历Handler的可读和可写队列，并调用read和write进行读和写，然后通知上层
int
NetHandler::mainNetEvent(int event, Event *e)
{
ink_assert(trigger_event == e && (event == EVENT_INTERVAL || event == EVENT_POLL));
(void) event;
(void) e;
EventIO *epd = NULL;
int poll_timeout = net_config_poll_timeout;
//计数信息++
NET_INCREMENT_DYN_STAT(net_handler_run_stat);
//处理NetHandler的可读和可写队列上的时间UnixNetVConnection，这里你可以看作什么都不做
process_enabled_list(this);
if (likely(!read_ready_list.empty() || !write_ready_list.empty() || !read_enable_list.empty() || !write_enable_list.empty()))
poll_timeout = 0;
else
poll_timeout = net_config_poll_timeout;
PollDescriptor *pd = get_PollDescriptor(trigger_event->ethread);
UnixNetVConnection *vc = NULL;
#if TS_USE_EPOLL
//调用epoll事件
pd->result = epoll_wait(pd->epoll_fd, pd->ePoll_Triggered_Events, POLL_DESCRIPTOR_SIZE, poll_timeout);
NetDebug("iocore_net_main_poll", "[NetHandler::mainNetEvent] epoll_wait(%d,%d), result=%d", pd->epoll_fd,poll_timeout,pd->result);
......
//处理所有的事件
vc = NULL;
for (int x = 0; x < pd->result; x++) {
epd = (EventIO*) get_ev_data(pd,x);
// EVENTIO_READWRITE_VC事件的处理：如果是读事件则加入到NetHandler的可读链表read_ready_list，如果是写事件加入NetHandler的可读链表write_ready_list
if (epd->type == EVENTIO_READWRITE_VC) {
vc = epd->data.vc;
if (get_ev_events(pd,x) & (EVENTIO_READ|EVENTIO_ERROR)) {
vc->read.triggered = 1;
if (!read_ready_list.in(vc))
read_ready_list.enqueue(vc);
else if (get_ev_events(pd,x) & EVENTIO_ERROR) {
// check for unhandled epoll events that should be handled
Debug("iocore_net_main", "Unhandled epoll event on read: 0xx read.enabled=%d closed=%d read.netready_queue=%d",
get_ev_events(pd,x), vc->read.enabled, vc->closed, read_ready_list.in(vc));
}
}
vc = epd->data.vc;
if (get_ev_events(pd,x) & (EVENTIO_WRITE|EVENTIO_ERROR)) {
vc->write.triggered = 1;
if (!write_ready_list.in(vc))
write_ready_list.enqueue(vc);
else if (get_ev_events(pd,x) & EVENTIO_ERROR) {
Debug("iocore_net_main",
"Unhandled epoll event on write: 0xx write.enabled=%d closed=%d write.netready_queue=%d",
get_ev_events(pd,x), vc->write.enabled, vc->closed, write_ready_list.in(vc));
}
} else if (!get_ev_events(pd,x) & EVENTIO_ERROR) {
Debug("iocore_net_main", "Unhandled epoll event: 0xx", get_ev_events(pd,x));
}
//EVENTIO_DNS_CONNECTION事件的处理：加入DNSHandler的triggered队列
} else if (epd->type == EVENTIO_DNS_CONNECTION) {
if (epd->data.dnscon != NULL) {
epd->data.dnscon->trigger();
#if defined(USE_EDGE_TRIGGER)
epd->refresh(EVENTIO_READ);
#endif
}
} else if (epd->type == EVENTIO_ASYNC_SIGNAL)
net_signal_hook_callback(trigger_event->ethread);
ev_next_event(pd,x);
}
pd->result = 0;
#if defined(USE_EDGE_TRIGGER)
//遍历Handler的可读队列中的vc，调用net_read_io分别处理每个vc，而net_read_io的功能就是调用read去接收数据，然后通知上层（HttpSM）
while ((vc = read_ready_list.dequeue())) {
if (vc->closed)
close_UnixNetVConnection(vc, trigger_event->ethread);
else if (vc->read.enabled && vc->read.triggered)
vc->net_read_io(this, trigger_event->ethread);
else if (!vc->read.enabled) {
read_ready_list.remove(vc);
}
}
//遍历Handler的可写队列中的vc，调用write_to_net分别处理每个vc，而write_to_net的功能就是调用write去发送数据，然后通知上层（HttpSM）
while ((vc = write_ready_list.dequeue())) {
if (vc->closed)
close_UnixNetVConnection(vc, trigger_event->ethread);
else if (vc->write.enabled && vc->write.triggered)
write_to_net(this, vc, trigger_event->ethread);
else if (!vc->write.enabled) {
write_ready_list.remove(vc);
}
}
return EVENT_CONT;
}
别忘了InactivityCop这个结构
main()--->UnixNetProcessor::start()--->initialize_thread_for_net()--->InactivityCop()
设置handler为InactivityCop::check_inactivity，该函数被每秒中调用一次
struct InactivityCop : public Continuation {
InactivityCop(ProxyMutex *m):Continuation(m) {
SET_HANDLER(&InactivityCop::check_inactivity);
}
main()--->UnixNetProcessor::start()--->initialize_thread_for_net()--->InactivityCop()---> InactivityCop::check_inactivity()
int check_inactivity(int event, Event *e) {
(void) event;
ink_hrtime now = ink_get_hrtime();
NetHandler *nh = get_NetHandler(this_ethread());
//遍历NetHandler的链接队列，判断和本线程是不是统一线程，是的话加到NetHandler的cop_list队列
forl_LL(UnixNetVConnection, vc, nh->open_list) {
if (vc->thread == this_ethread())
nh->cop_list.push(vc);
}
while (UnixNetVConnection *vc = nh->cop_list.pop()) {
// If we cannot ge tthe lock don't stop just keep cleaning
MUTEX_TRY_LOCK(lock, vc->mutex, this_ethread());
if (!lock.lock_acquired) {
NET_INCREMENT_DYN_STAT(inactivity_cop_lock_acquire_failure_stat);
continue;
}
//如果该链接vc已设置为关闭状态，则调用close_UnixNetVConnection()进行关闭操作
if (vc->closed) {
close_UnixNetVConnection(vc, e->ethread);
continue;
}
if (vc->next_inactivity_timeout_at && vc->next_inactivity_timeout_at < now)
//调用vc的handler（UnixNetVConnection::mainEvent）进行处理
vc->handleEvent(EVENT_IMMEDIATE, e);
}
return 0;
}
好了，到此为止，NetProcessor的启动流程已经分析完成，可以简单得总结为：NetProcessor的启动主要任务是初始化几个线程定时调用epoll_wait()等待读写事件，如果有读事件到来时调用read进行读操作，然后把读到的数据传给上层处理，如果有写事件到来时调用write进行发送操作，发送完成后通知上层发送结果。那么读写事件是怎么来的呢？根据网络编程的经验，server在read和write之前一般都要accept，这里的读写事件正是来于accept，下面来分析NetProcessor的accept。NetProcessor的accept是在HttpProxyServer启动时调用的，正确的说是main_accept()函数。
main()--->start_HttpProxyServer()
void
start_HttpProxyServer()
{
//根据配置，每个端口(默认只有一个：8080)创建一个Acceptor
for ( int i = 0 , n = proxy_ports.length() ; i < n ; ++i ) {
HttpProxyAcceptor& acceptor = HttpProxyAcceptors[i];
HttpProxyPort& port = proxy_ports[i];
......
if (NULL == netProcessor.main_accept(acceptor._accept, port.m_fd, acceptor._net_opt))
return;
}
......
}
}
main()--->start_HttpProxyServer()--->NetProcessor::main_accept()
Action *
NetProcessor::main_accept(Continuation *cont, SOCKET fd, AcceptOptions const& opt)
{
UnixNetProcessor* this_unp = static_cast
Debug("iocore_net_processor", "NetProcessor::main_accept - port %d,recv_bufsize %d, send_bufsize %d, sockopt 0x%0x",
opt.local_port, opt.recv_bufsize, opt.send_bufsize, opt.sockopt_flags);
//直接调用UnixNetProcessor::accept_internal()
return this_unp->accept_internal(cont, fd, opt);
}
main()--->start_HttpProxyServer()--->NetProcessor::main_accept()--->UnixNetProcessor::accept_internal()
Action *
UnixNetProcessor::accept_internal(Continuation *cont, int fd, AcceptOptions const& opt)
{
EventType et = opt.etype;
//创建NetAccept实例
NetAccept *na = createNetAccept();
EThread *thread = this_ethread();
ProxyMutex *mutex = thread->mutex;
int accept_threads = opt.accept_threads;
IpEndpoint accept_ip;
upgradeEtype(et);
//opt是从配置里读出来的网络相关的配置，作为Net的选项，具体的配置想请看ATS配置说明
if (opt.accept_threads < 0) {
REC_ReadConfigInteger(accept_threads, "proxy.config.accept_threads");
}
NET_INCREMENT_DYN_STAT(net_accepts_currently_open_stat);
//根据配置的模式设置server的地址
if (opt.localhost_only) {
accept_ip.setToLoopback(opt.ip_family);
} else if (opt.local_ip.isValid()) {
accept_ip.assign(opt.local_ip);
} else {
accept_ip.setToAnyAddr(opt.ip_family);
}
ink_assert(0 < opt.local_port && opt.local_port < 65536);
accept_ip.port() = htons(opt.local_port);
na->accept_fn = net_accept;
na->server.fd = fd;
ats_ip_copy(&na->server.accept_addr, &accept_ip);
na->server.f_inbound_transparent = opt.f_inbound_transparent;
//透明代理
if (opt.f_inbound_transparent) {
Debug( "http_tproxy", "Marking accept server %p on port %d as inbound transparent", na, opt.local_port);
}
int should_filter_int = 0;
na->server.http_accept_filter = false;
//查看该配置项的说明，只有数据到来时才会accept，默认等45秒没来就放弃，是在下面调用setsockopt()通过设置socket的选项来实现的
REC_ReadConfigInteger(should_filter_int, "proxy.config.net.defer_accept");
if (should_filter_int > 0 && opt.etype == ET_NET)
na->server.http_accept_filter = true;
na->action_ = NEW(new NetAcceptAction());
*na->action_ = cont;//指向上层的continuation，如HttpAccept
//下面是初始化接收buffer大小等网络的一些参数
na->action_->server = &na->server;
na->callback_on_open = opt.f_callback_on_open;
na->recv_bufsize = opt.recv_bufsize;
na->send_bufsize = opt.send_bufsize;
na->sockopt_flags = opt.sockopt_flags;
na->packet_mark = opt.packet_mark;
na->packet_tos = opt.packet_tos;
na->etype = opt.etype;
na->backdoor = opt.backdoor;
if (na->callback_on_open)
na->mutex = cont->mutex;
//实时接收
if (opt.frequent_accept) {
//配置的accept的线程数
if (accept_threads > 0) {
//设置socket的选项
if (0 == na->do_listen(BLOCKING, opt.f_inbound_transparent)) {
NetAccept *a;
//循环为每个线程创建NetAccept实例，并把上面创建的na赋值给之，最后调用NetAccept的 init_accept_loop()函数进入循环地accept的状态
for (int i=1; i < accept_threads; ++i) {
a = createNetAccept();
*a = *na;
a->init_accept_loop();
Debug("iocore_net_accept", "Created accept thread #%d for port %d", i, ats_ip_port_host_order(&accept_ip));
}
Debug("iocore_net_accept", "Created accept thread #%d for port %d", accept_threads, ats_ip_port_host_order(&accept_ip));
na->init_accept_loop();
}
} else {
na->init_accept_per_thread();
}
} else
na->init_accept();
//查看该配置项的说明，只有数据到来时才会accept，默认等45秒没来就放弃，是在下面调用setsockopt()通过设置socket的选项来实现的
#ifdef TCP_DEFER_ACCEPT
if (should_filter_int > 0) {
setsockopt(na->server.fd, IPPROTO_TCP, TCP_DEFER_ACCEPT, &should_filter_int, sizeof(int));
}
#endif
return na->action_;
}
main()--->start_HttpProxyServer()--->NetProcessor::main_accept()--->UnixNetProcessor::accept_internal()--->NetAccept::init_accept_loop()
该函数的功能是创建一个线程，并设置线程的执行函数为NetAccept::acceptLoopEvent
void
NetAccept::init_accept_loop()
{
size_t stacksize;
//线程栈的大小
REC_ReadConfigInteger(stacksize, "proxy.config.thread.default.stacksize");
SET_CONTINUATION_HANDLER(this, &NetAccept::acceptLoopEvent);
eventProcessor.spawn_thread(this, "[ACCEPT]", stacksize);
}
main()--->start_HttpProxyServer()--->NetProcessor::main_accept()--->UnixNetProcessor::accept_internal()--->NetAccept::init_accept_loop()--->NetAccept::acceptLoopEvent()
int
NetAccept::acceptLoopEvent(int event, Event * e)
{
(void) event;
(void) e;
EThread *t = this_ethread();
//妈啊，终于看到死循环在accept了（当然listen已经在前面了采用accept）
while (1)
do_blocking_accept(t);
NET_DECREMENT_DYN_STAT(net_accepts_currently_open_stat);
delete this;
return EVENT_DONE;
}
main()--->start_HttpProxyServer()--->NetProcessor::main_accept()--->UnixNetProcessor::accept_internal()--->NetAccept::init_accept_loop()--->NetAccept::acceptLoopEvent()--->NetAccept::do_blocking_accept()
该函数的功能是循环地调用accept去接收请求，并让event系统调度去处理每个链接 UnixNetVConnection
int
NetAccept::do_blocking_accept(EThread * t)
{
int res = 0;
int loop = accept_till_done;
UnixNetVConnection *vc = NULL;
do {
//创建表示一个链接的UnixNetVConnection的实例
vc = (UnixNetVConnection *)alloc_cache;
if (likely(!vc)) {
vc = allocateGlobal();
vc->from_accept_thread = true;
vc->id = net_next_connection_number();
alloc_cache = vc;
}
//流量控制
ink_hrtime now = ink_get_hrtime();
while (!backdoor && check_net_throttle(ACCEPT, now)) {
check_throttle_warning();
if (!unix_netProcessor.throttle_error_message) {
safe_delay(NET_THROTTLE_DELAY);
} else if (send_throttle_message(this) < 0) {
goto Lerror;
}
now = ink_get_hrtime();
}
//调用accept去接收请求
if ((res = server.accept(&vc->con)) < 0) {
//错误处理
Lerror:
int seriousness = accept_error_seriousness(res);
if (seriousness >= 0) {
if (!seriousness)
check_transient_accept_error(res);
safe_delay(NET_THROTTLE_DELAY);
return 0;
}
if (!action_->cancelled) {
MUTEX_LOCK(lock, action_->mutex, t);
action_->continuation->handleEvent(EVENT_ERROR, (void *)(intptr_t)res);
MUTEX_UNTAKE_LOCK(action_->mutex, t);
Warning("accept thread received fatal error: errno = %d", errno);
}
return -1;
}
//流量控制
check_emergency_throttle(vc->con);
alloc_cache = NULL;
NET_SUM_GLOBAL_DYN_STAT(net_connections_currently_open_stat, 1);
//设置vc的时间和server的ip地址
vc->submit_time = now;
ats_ip_copy(&vc->server_addr, &vc->con.addr);
//透明代理标志位
vc->set_is_transparent(server.f_inbound_transparent);
vc->mutex = new_ProxyMutex();
vc->action_ = *action_;
//设置UnixNetVConnection的handler为UnixNetVConnection::acceptEvent
SET_CONTINUATION_HANDLER(vc, (NetVConnHandler) & UnixNetVConnection::acceptEvent);
//让event系统调度vc的handler执行
eventProcessor.schedule_imm_signal(vc, getEtype());
} while (loop);
return 1;
}
main()--->start_HttpProxyServer()--->NetProcessor::main_accept()--->UnixNetProcessor::accept_internal()--->NetAccept::init_accept_loop()--->NetAccept::acceptLoopEvent()--->NetAccept::do_blocking_accept()--->UnixNetVConnection::acceptEvent()
该函数的功能是接收一个链接，向NetHandler注册读写事件，让NetHandler去接收该链接的数据和发送对该请求的响应报文（NetHandler肯定是干这个事的啊），最后调用上层（HttpSM）的handler（HttpAccept::mainEvent）来接收该链接
int
UnixNetVConnection::acceptEvent(int event, Event *e)
{
thread = e->ethread;
MUTEX_TRY_LOCK(lock, get_NetHandler(thread)->mutex, e->ethread);
if (!lock) {
if (event == EVENT_NONE) {
thread->schedule_in(this, NET_RETRY_DELAY);
return EVENT_DONE;
} else {
e->schedule_in(NET_RETRY_DELAY);
return EVENT_CONT;
}
}
if (action_.cancelled) {
free(thread);
return EVENT_DONE;
}
//设置UnixNetVConnection的handler为UnixNetVConnection::mainEvent
SET_HANDLER((NetVConnHandler) & UnixNetVConnection::mainEvent);
//获取指向NetHandler的指针，NetHandler前面介绍过了
nh = get_NetHandler(thread);
//获取指向PollDescriptor的指针，PollDescriptor前面介绍过了
PollDescriptor *pd = get_PollDescriptor(thread);
//注册epoll读和写事件，这会和上面的流程联系起来了吧
if (ep.start(pd, this, EVENTIO_READ|EVENTIO_WRITE) < 0) {
Debug("iocore_net", "acceptEvent : failed EventIO::start\n");
close_UnixNetVConnection(this, e->ethread);
return EVENT_DONE;
}
//把vc加入NetHandler的开链接队列open_list
nh->open_list.enqueue(this);
//设置相应的超时时间用于关闭链接
if (inactivity_timeout_in)
UnixNetVConnection::set_inactivity_timeout(inactivity_timeout_in);
if (active_timeout_in)
UnixNetVConnection::set_active_timeout(active_timeout_in);
//调用上层的handler来处理本次链接，如HttpAccept::mainEvent，怎么处理留到HTTP流程再分析
action_.continuation->handleEvent(NET_EVENT_ACCEPT, this);
return EVENT_DONE;
}
HttpSM <----------------------HttpAccept
^ ^
| |
| |
| 注册READ/WRITE事件 |
NetHandler <------------------NetAccept
^ ^
| |
| |
| |
| |
read() write() accept()
此外，NetProcessor还有两个函数要说明，那就是NetProcessor::connect_s()和NetProcessor::connect_re()。这两个函数都是给上层提供的connect接口，他们的区别是一个是异步的一个是同步的，connect_s是同步，connect_re是异步。
