OceanBase使用libeasy原理源码分析:服务器端
libeasy是个网络框架,这个网络框架基于事件驱动模型,libeasy可以有多个网络I/O线程,每个网络I/O线程一个event loop,事件驱动模型基于开源的libev实现。
我认为,libeasy不同于其它的网络框架比如tbnet,muduo。tbnet,muduo等网络框架的目的就是向应用层暴露出简单的发包和收包的接口,让应用层从底层发包和收包的处理细节中解放出来,使得应用层能更加专注于业务逻辑的实现,为了做到这些,网络框架帮助应用程序管理连接,管理输入输出缓冲区,处理具体的发包收包等细节和错误的处理,处理流控,并且允许应用层注入封包,解包,新建连接时处理,断开连接时处理,收到包后处理包的逻辑等。libeasy与这些网络框架稍不同,从高层次看,libeasy中的线程分为业务逻辑线程和网络I/O线程,无论哪种线程,线程都有唯一的一个event loop,工作的时候,I/O线程和网络I/O线程有对应关系,他们之间通过pipe来实现线程的唤醒,原理就是业务逻辑线程的event loop监听pipe的读端,当I/O线程A接收到包的时候,往pipe的写端写入数据,业务逻辑线程event loop返回,从任务链表中取任务执行。可以说,libeasy已经超出了网络框架的范畴,但是libeasy也支持和普通的网络框架一样,仅仅使用网络 I/O线程部分。
下面分析libeasy的原理和实现,由于libeasy基于libev,对libev不了解的参见 这
这篇主要分析OceanBase 0.4的mergeserver使用libeasy作为服务器端的模式,客户端自然就是MySQL客户端。OceanBase仅仅使用了libeasy的IO线程部分,工作线程是使用了我们自己的线程池。
主要说如下几个方面:
一、OceanBase启动时的使用模式
二、 基础数据结构
2.1 easy_list_t
2.2 easy_pool_t
2.3 easy_buf_t
2.4 easy_connection_t
三、 连接建立
四、 同步处理(OceanBase少量使用这种模式)
五、 异步处理(OceanBase大量采用这种模式)
六、 资源管理
一、OceanBase启动时的使用模式
libeasy内与OceanBase使用模式相关的各个对象之间关系如下图所示:
OceanBase主要用到的是libeasy的IO线程池部分,没有用到libeasy内置的工作线程池。libeasy采用one event loop per thread的模式,即每个线程一个event loop,
内存资源每连接自管理,连接之间的资源互不干涉。每个连接(easy_connection_t)上有可以有多个消息(easy_message_t),通过链表连起来,每个消息可以由多个请求组成(easy_request_t),也通过链表连起来。
OceanBase 0.4的ObMySQLServer启动的时候,会初始化libeasy相关的东西,主要步骤如下:
//设置一堆给libeasy的回调函数
memset(&handler_, 0, sizeof(easy_io_handler_pt));
//以下都是对于OceanBase 0.4 mergeserver的obmysql端口来说的
// 将mergeserver需要回复给mysql客户端的结果以easy_buf_t(libeasy用来管理输入输出缓冲区的数据结构)的形式加到请求所属于的easy_connection_t(TCP连接)的输出缓冲区链表中
handler_.encode = ObMySQLCallback::encode;
// libeasy回调这个函数用于从该连接的输入缓冲区中反序列化出一个符合MySQL协议的包,然后吐给上层使用
handler_.decode = ObMySQLCallback::decode;
// 对于每个decode出来的包进行实际的处理,在OceanBase的实现中,我们大多数时候仅仅是将包push到我们自己的工作队列中,在这种情况下返回的不是EASY_OK这个错误码,因为目前我们还没有对
// 这个包进行实质上的处理,还没有为这个包产生结果。少数情况下,当我们接收到的包的大小超过2MB的时候,process这个函数会返回EASY_OK,并且会为这个请求产生一个响应结果,一个MySQL的
// error packet,将其挂在request->opacket上,当libeasy看到了返回EASY_OK后,就会调用encode方法将opacket给挂在连接的输出缓冲区链表中,随后将其发送出去
handler_.process = ObMySQLCallback::process;
handler_.get_packet_id = ObMySQLCallback::get_packet_id;
handler_.on_disconnect = ObMySQLCallback::on_disconnect;
// 登录逻辑,在libeasy发现listenfd上有读事件时,会将连接接下来,然后给MySQL客户端发送握手包,同时接受客户端发送过来的用户名密码等信息,最后进行服务器端的验证,这几次
//交互过程是不经过libeasy网络框架的
handler_.on_connect = ObMySQLCallback::on_connect;<br>// 用于当请求处理完毕后,告诉工作线程不要再发包了
handler_.cleanup = ObMySQLCallback::clean_up;
eio_ = easy_eio_create(eio_, io_thread_count_);
eio_->tcp_defer_accept = 0;
easy_listen_t* listen = easy_connection_add_listen(eio_, NULL, port_, &handler_);
rc = easy_eio_start(eio_);
easy_eio_wait(eio_);
easy_eio_create(eio_, io_thread_count_)做了如下几件事:
1. 分配一个easy_pool_t, 用来存放easy_io_t对象,io_thread_count_个io线程(easy_io_thread_t),初始化针对每个eio(一般系统就只有一个)的统计信息结构(easy_summary_t)
2. 设置一些tcp参数,比如tcp_defer_accept,tcp_nodelay,设置一些负载保护参数,比如EASY_CONN_DOING_REQ_CNT,表示每个连接同时正在处理的请求数不能
超过EASY_CONN_DOING_REQ_CNT
3. 初始化每个io线程:
3.1 初始化其各个链表节点成员,比如conn_list(已建立连接但是读写事件还没有监听的连接链表), connected_list(连接已建立并且事件已监听的连接链表),request_list(已处理完成但是还没有将结果发送出去的请求链表)等
3.2 统计信息初始化,例如io线程同时正在处理的请求数,已经处理的请求数
3.3 初始化成员变量listen_watcher, 每100ms触发一次对于listen的切换(回调函数easy_connection_on_listen),实际上,在刚启动的时候,是100ms,当有IO线程抢到listen的权利后,这个timer会被改成60s,随后,每60s进行一次listen的切换,而之前拥有listen权利的IO线程则会停掉它的read_watcher
3.4 设定io线程的执行体函数easy_io_on_thread_start
3.5 设定io线程被唤醒时的回调函数easy_connection_on_wakeup(设置为成员变量ev_async thread_watcher的cb上,当io线程被唤醒时,这个thread_watcher被触发,从而回调).
3.6 计算出io线程在io线程池中的下标放在io线程的成员变量idx中
3.7 为这个线程分配一个event loop
easy_listen_t * listen = easy_connection_add_listen(eio_, NULL, port_, &handler_) : 增加一个listen地址,并且设置回调函数
easy_listen_t定义如下:
struct easy_listen_t {
int fd;<br> // read_watcher的下标
int8_t cur, old;
int8_t hidden_sum;<br> //如果为1,则所有线程可以监听同一个地址
uint8_t reuseport : 1;
// 监听地址
easy_addr_t addr;
//各种回调函数的集合
easy_io_handler_pt *handler;
// 多个io线程竞争listen的锁
easy_atomic_t listen_lock;
//当前listen权利被哪个IO线程拥有
easy_io_thread_t *curr_ioth;
easy_io_thread_t *old_ioth;
easy_listen_t *next;
//有多少个IO线程就有多少个watcher, 每个watcher都监听fd上的EV_READ和EV_CLEANUP事件
ev_io read_watcher[0];
};
easy_connection_add_listen:
从eio->pool中为easy_listen_t和io线程个数个ev_io分配空间,开始监听某个地址,初始化每个read_watcher,关注listen fd上的读事件,设置其回调函数 为easy_connection_on_accept,在这里仅仅是初始化read_watcher, 还没有激活,激活在每个IO线程启动(easy_io_on_thread_start)的时候做。一旦激活后,当有连接到来的时候,触发easy_connection_on_accept
easy_eio_start(eio_):
将eio_ 上挂着的所有的线程池中的所有线程全部启动,每个IO线程的执行函数体easy_io_on_thread_start做如下几件事:
1. 可以选择设置是否屏蔽信号,可以设置CPU亲缘性
2. 选择一个线程listen:通过listen_watcher的方式,或者如果只有一个线程或者设置了socket的SO_REUSEPORT标记,则所有线程一起监听,同时将listen的线程的read_watcher激活,从而下次来新连接的时候,就可以调用回调函数easy_connection_on_accept来接收新连接了。
二、基础数据结构
在libeasy中,有如下一些重要的数据结构,分别为
2.1 easy_list_t:链表结构,只有两个指针next,prev,可以用于所有的元素类型,如同内核中的链表,假设有一个easy_request_t的元素的链表,遍历它的代码如下:
// 假设有一个指针easy_list_t *request_list指向的是一个easy_request_t链表的头
easy_request_t *r, *rn;
// request_list_node 它是easy_request_t结构体中的一个成员:easy_list_t *request_list_node,用于将easy_request_t串起来
easy_list_for_each_entry_safe(r, rn, request_list, request_list_node)
{
//r指向链表中第一个easy_request_t元素
//rn指向链表中第二个easy_request_t元素
}
这里用到easy_list_for_each_entry_safe和easy_list_entry两个宏,就以这个例子来解释一下这两个宏:
宏easy_list_for_each_entry_safe:
#define easy_list_for_each_entry_safe(pos, n, head, member) \
for (pos = easy_list_entry((head)->next, typeof(*pos), member), \
n = easy_list_entry(pos->member.next, typeof(*pos), member); \
&pos->member != (head); \
pos = n, n = easy_list_entry(n->member.next, typeof(*n), member))
宏easy_list_entry:这个宏的作用就是根据链表节点指针找到对应元素的首地址
#define easy_list_entry(ptr, type, member) ({ \
const typeof( ((type *)0)->member ) *__mptr = (ptr); \ // 得到easy_request_t中成员request_list_node的指针
(type *)( (char *)__mptr - offsetof(type,member) );}) // 指针减去 request_list_node成员在easy_request_t中的偏移就得到了easy_request_t的首地址
其中offsetof(type,member)是得到member这个这个成员在结构type中的偏移量(man offsetof),typeof(((type*)0)->member)得到参数的类型。
2.2 easy_pool_t :内存池,和nginx的内存池实现几乎一样,见http://www.alidata.org/archives/1390 它不是一个全局的内存池,libeasy中可以有很多个,比如对于每个新的连接产生一个easy_pool_t
2.3 easy_buf_t : 用于管理连接的输入输出缓冲区
#define EASY_BUF_DEFINE \
easy_list_t node; \
int flags; \
//当easy_buf_t不再使用时调用cleanup
easy_buf_cleanup_pt *cleanup; \
void *args;
struct easy_buf_t {
EASY_BUF_DEFINE;
//buf开始处
char *pos;
// 下次从这开始写,或者从已经读到这
char *last;
//buf结束处
char *end;
};
2.4 easy_connection_t:封装一个TCP连接,在libeasy中,一个easy_message_t可以包含一个或者多个easy_request_t,easy_request_t就相当于应用层的一个具体的包,例如
在OceanBase 0.4中,一个 easy_request_t 对应于一个mysql客户端发过来的mysql协议包
struct easy_connection_t {
//这个event loop监听这个连接的事件
struct ev_loop *loop;
//该连接是在这个内存池上分配的
easy_pool_t *pool;
// 该连接所属的io线程
easy_io_thread_t *ioth;
//用于链表
easy_connection_t *next;
// 用于串链表,比如连接建立后,通过这个链表节点,将其串到io线程的已连接链表中
easy_list_t conn_list_node;
// file description
// default_message_len 默认是16, 大小是8KB, first_message_len默认是2, 大小是1KB
// libeasy对于每个连接收数据的时候,如果上次对于这个连接已经收到了1个或者n个完整的包(easy_request_t对应于一个完整的包,在OB 0.4,对应一个mysql协议的包),
// 那么会重新分配一个8KB大小的easy_pool_t,然后再上面easy_message_t所需要的内存,最主要就是easy_message_t的输入缓冲区easy_buf_t *input所需要的内存,
// 这里,会为easy_buf_t分配1KB的内存,一个数据包所占用的内存是不会跨easy_buf_t的边界的,读连接的数据的时候,如果easy_buf_t里空间不够,会分配一个足够大的
// easy_buf_t, 然后将原来的数据拷贝过来
uint16_t first_message_len, default_message_len;
int reconn_time, reconn_fail;
int idle_time;
// fd: socket fd,
// seq: 这是系统accept的第seq个连接
int fd, seq;
easy_addr_t addr;
// 该连接fd的读事件的watcher
ev_io read_watcher;
// 该连接fd的写事件的watcher
ev_io write_watcher;
// 该连接fd的超时事件的watcher
ev_timer timeout_watcher;
// 将该连接接收到的所有的easy_message_t串在一起
easy_list_t message_list;
// 输出缓冲区链表,实际上就是easy_buf_t的链表
easy_list_t output;
//应用层注入的各种回调函数的集合,比如decode,在OceanBase 0.4中,就是解析mysql协议
//on_connect,在OceanBase 0.4中,on_connect会验证用户名和密码
easy_io_handler_pt *handler;
// 对连接数据的读,就是简单封装recv系统调用
easy_read_pt *read;
// 将output写出
easy_write_pt *write;<br> //用于libeasy作为客户端的时候,再下一篇中讲述
easy_client_t *client;
easy_list_t session_list;
easy_hash_t *send_queue;
void *user_data;
easy_uthread_t *uthread; //user thread
//一堆TCP的参数
uint32_t status : 4;
uint32_t event_status : 4;
uint32_t type : 1;
uint32_t async_conn : 1;
uint32_t conn_has_error : 1;
uint32_t tcp_cork_flag : 1;
uint32_t wait_close : 1;
uint32_t need_redispatch : 1;
uint32_t read_eof : 1;
uint32_t auto_reconn : 1;
uint32_t life_idle : 1;
//当前connection上同时正在处理的easy_request_t的个数
uint32_t doing_request_count;
ev_tstamp start_time, last_time;
ev_tstamp wait_client_time, wcs;
// 统计信息,比如连接上走的流量
easy_summary_node_t *con_summary; //add for summary
// 不关注
easy_ssl_connection_t *sc;
};
三、连接建立
如前所述,当listen fd上有可读事件时,IO线程相应的read_watcher会被触发,从而回调easy_connection_on_accept函数接受新连接。
easy_connection_on_accept主要做如下几件事:
1. accept将连接接下来
2. 调用easy_connection_new为返回的fd新建一个easy_connection_t
2.1 分配一个easy_pool_t,专门用来存easy_connection_t, 并且设置其各个成员:
// 表示这个连接是在pool上分配的,对于连接的引用计数也是记在pool上的<br> c->pool = pool;<br> // 用于重连的一个参数,100ms,用于libeasy作为客户端,现不关注
c->reconn_time = 100;<br>
c->idle_time = 60000;<br> // 在前面easy_connection_t中有解释
c->first_message_len = 2; // 1Kbyte
c->default_message_len = (EASY_IO_BUFFER_SIZE >> 9); // 8Kbyte<br> //往连接上读写的函数
c->read = easy_socket_read;
c->write = easy_socket_write;<br> //连接上message的链表
easy_list_init(&c->message_list);<br> // 用于libeasy IO线程唤醒工作线程(对于OceanBase 0.4来说,就是ObPacketQueueThread)<br> // 当OceanBase同步写数据给MySQL客户端时,工作线程每发一个包就阻塞住,直到IO线程将其唤醒再发下一个包,session_list就存放这些工作线程正在处理的request
easy_list_init(&c->session_list);<br> // 用于串连接到链表中
easy_list_init(&c->conn_list_node);<br> // 该连接上的输出缓冲区链表,每个元素是一个easy_buf_t
easy_list_init(&c->output);
3. 设置返回的socket fd为非阻塞
4.将回调函数集合easy_io_handler_pt, 调用回调函数on_connect(),在OB中,用于验证用户名密码。
5.初始化该连接的read_watcher, write_watcher,和timeout_watcher,并且设置其回调函数分别为easy_connection_on_readable,easy_connection_on_writable和easy_connection_on_timeout_conn(注意,仅仅是初始化,还没有激活)
6. 激活该连接的read_watcher(即让该连接所属的IO线程的event loop监听这个连接的读事件), 设置其回调函数为easy_connection_on_readable(这里有一些细节,例如如果设置了tcp_defer_accept参数,则如果连接上没有数据,则该连接不会返回,IO线程阻塞住)
7. 将该连接加入了所属IO线程的已连接链表中(connected_list)
至此连接建立了。
四、 同步处理
这里的同步是指libeasy的IO线程回调应用层的process函数后对这个包的业务逻辑处理即结束。不需要应用层的工作线程的参与。这种模式下,process函数应该直接返回EASY_OK. 以OceanBase 0.4的obmysql接口为例,当用户输入的包的大小大于2MB的时候,应用层process函数直接构造一个MySQL的error packet,挂在连接的输出缓冲区链表中(r->retcode默认为EASY_OK)。随后输出缓冲区中的数据被写出,请求结束。
下面就以从MySQL客户端接收到一个大于2MB的包为例:
显然每个连接上有可读事件的时候都会回调easy_connection_on_readable函数:该函数流程如下:
1. 检查当前IO线程同时正在处理的请求是否超过EASY_IOTH_DOING_REQ_CNT(8192),当前连接上的请求数是否超过EASY_CONN_DOING_REQ_CNT(1024),如果超过,则调用easy_connection_destroy(c)将连接销毁掉, 提供了一种负载保护机制
2. 检查上一次收到的message(easy_message_t)是不是完整的,即收到了一条或者多条(easy_request_t),一个easy_request_t相当于一个请求包,贯穿着请求的输入,处理和输出整个流程。如果收到的是一个完整的message(判断message的状态status, status == EASY_MESG_READ_AGAIN说明不完整),那么就调用easy_message_create函数创建一个8KB的easy_pool_t,然后在其上分配一个easy_message_t结构,再分配一个1KB大小的easy_buf_t作为输入缓冲区将其挂在easy_message_t的input成员上。然后设置message的pool,并将其引用计数初始化为1,并将next_read_len设置为1KB
#define EASY_MESSAGE_SESSION_HEADER
easy_connection_t *c; \
easy_pool_t *pool; \
int8_t type; \
int8_t async; \
int8_t status; \
int8_t error;
// 用于接收, 一个或多个easy_request_t
struct easy_message_t {
EASY_MESSAGE_SESSION_HEADER
int recycle_cnt;
//该连接的输入缓冲区
easy_buf_t *input;
//用于将一个连接的所有的message串起来
easy_list_t message_list_node;
easy_list_t request_list;
easy_list_t all_list;
//该message上request的个数
int request_list_count;
//下次需要读取的数据的长度
int next_read_len;
void *user_data;
};
3. 调用easy_buf_check_read_space检查连接的输入缓冲区input中是否有next_read_len个字节的空间,如果没有,则继续在message的pool上分配next_read_len个字节的空间,并且将原来的输入缓冲区的数据拷贝过来,将这个next_read_len字节大小的缓冲区作为新的输入缓冲区
4. 调用 c->read从连接读数据,实际上调用的是函数easy_socket_read,该函数只是简单的封装了recv()。设置连接的成员read_eof,如果读到的数据小于next_read_len,则将其设置为1,主要用于异常情况下关闭连接的情形
5. 更新连接的统计信息con_summary
6. 如果是作为服务器端,则调用easy_connection_do_request(m),否则调用 easy_connection_do_response(m),目前仅关注服务器端。
easy_connection_do_request(m)流程如下:
// ipacket放进来的包, opacket及出去的包
struct easy_request_t {
//所属的message
easy_message_session_t *ms;
easy_list_t request_list_node;
easy_list_t all_node;
int16_t retcode, status;
int reserved;
//请求开始时间
ev_tstamp start_time;
void *ipacket;
void *opacket;
void *args;
void *user_data;
// waitobj,用于IO线程唤醒工作线程,封装pthread_mutex_t和pthread_cond_t
easy_client_wait_t *client_wait;
};
1. 回调应用层实现的decode函数,然后在message的pool上分配一个easy_request_t,再将decode出来的packet(对于OceanBase 0.4来说,就是ObMySQLCommandPacket)挂在request的ipacket上. 然后将该请求加入到message的request_list中
2. 修改一些统计信息
3. 设置message的status,当输入缓冲区中还有空闲空间时,说明没有收到一个完整的request,则将message的status设置为EASY_MESG_READ_AGAIN,这正是前面用来判断是否接收到了一个完整的message的判断标记
4. 调用easy_connection_process_request函数对刚才decode出来的所有的request进行处理
easy_connection_process_request是处理请求的函数,主要流程如下:
1. 对于每个request,
1.1 将这个request从所在的message中摘下来(request_list_node)
1.2. 回调应用层传入的process:
process函数返回EASY_OK, 则说明应用已经处理完了这个请求了(这种情况出现在当OceanBase 0.4接收到了一个大于2MB的包的时候),接着调用easy_connection_request_done函数。
easy_connection_request_done函数处理:
1.2.1 回调应用层定义的encode,对于OceanBase 0.4来说,比如在create table这种回复只需要回复一个包的情形来说,就是将封装了MySQL回复包的easy_buf_t(r->opacket)给挂在该请求所在连接的输出缓冲区链表(c->output)后面。
1.2.2 调用easy_request_set_cleanup(r, &c->output), 将message的引用计数加1,以防结果还没有输出message就被析构了。然后设置一个cleanup(实际上是函数easy_request_cleanup)函数挂在输出缓冲区链表上的easy_buf_t(也就是刚刚挂上去的那个buf)上,当buf被写出去后不再使用的时候,会回调cleanup。
1.2.3 将request的status设置成EASY_REQUEST_DONE
1.3. 当连接的输出缓冲区链表中积累了128个没发出去,或者不使用tcp nagle算法的时候,则调用easy_connection_write_socket(一次写不完会再次启写事件)主动往连接写一次数据
1.4. 检查request所在的message上是否还有request,如果没有请求需要处理并且没有接收到了一半的请求,则调用easy_message_destroy(m, 1),将这个message从m所属的连接上的message链表中摘除,把m的status设置为EASY_MESG_DESTROY,最后将message的引用计数减1.如果message的引用计数变成了0,则将和message关联的easy_pool_t整个给释放掉
2. 调用easy_connection_write_socket将连接的输出缓冲区链表中的数据写出,这个函数调用c->write(实际调用easy_socket_write)将数据写出,easy_socket_write使用writev系统调用将数据写出,同时,对于已经写出去的easy_buf_t,调用easy_buf_destroy函数,它会回调buf的cleanup函数,如前所属,实际上是easy_request_cleanup函数.easy_request_cleanup函数做了如下几件事:
2.1 调用easy_request_server_done,修改一些统计信息,并且回调应用层定义的cleanup函数。对于我们这种同步处理的场景来说,什么都没做。
2.2 调用easy_message_destroy((easy_message_t *)r->ms, 0),将message的引用计数减1,到此message的引用计数变为0,将这个message的easy_pool_t销毁,从而内存被释放。
请求结束。
五、异步处理
异步处理流程:IO线程将数据接下来后,调用应用层定义的process方法,应用层process方法不返回EASY_OK,返回EASY_AGAIN,将packet丢到工作线程队列,同时IO线程将这个异步请求丢到连接的session链表中,并且启动写(start write watcher)。随后,应用层的工作线程从工作队列里面拿出packet,进行处理,检索出结果包,将其挂在r->opacket上,然后调用easy_request_wakeup(req), 将请求挂在IO线程的已完成的请求链表中,并且唤醒IO线程,随后工作线程阻塞在一个信号量上。IO线程被唤醒后,将所有的输出buf都写出,然后再次回调process函数,process函数看到请求的retcode等于EASY_OK,则signal信号量,从而工作线程被唤醒。
典型的,select请求这种需要回复多个包给MySQL客户端的场景使用的都是这种模式
其它诸如只需要回复一个包给MySQL客户端的DML操作,例如INSERT,UPDATE等也使用这种模式,只是工作线程由于只需要发一个包,所以不会阻塞在信号量上,直接就返回了
详细分析如下:
message的引用计数初始化为1
前面读取输入,decode过程都类似,不同的是easy_connection_process_request函数:
对于这种需要工作线程处理的request,
1.1 将这个request从所在的message中摘下来(request_list_node)
1.2 回调应用层定义的process,process将包放入到工作队列,然后process(message引用计数加1)返回EASY_AGAIN。
随后,工作线程从工作队列中拿到了请求包,处理,将结果包封装成一个easy_buf_t,挂在请求r的opacket上,随后调用easy_request_wakeup(r)将请求r挂在IO线程的已完成请求链表,并且唤醒IO线程,然后调用wait_client_obj(*wait_obj),wait在easy_client_wait_t的一个信号量上。
IO线程被唤醒:首先会做一些其他的事情(比如从自己的conn_list队列中取出从其它的IO线程迁移过来的连接,然后监听这个连接原有的读写事件后将其加到本IO线程的已连接链表中(connected_list))。然后调用easy_connection_send_response。
easy_connection_send_response:
遍历IO线程的已完成请求链表:将请求所在message的引用计数减1(此时引用计数为1),然后执行easy_connection_request_done
easy_connection_request_done:
和前面一样:回调应用层定义的encode方法将r->opacket挂在连接的输出缓冲区链表output中,然后调easy_request_set_cleanup,给刚才加入的输出buf设置cleanup回调函数,并且将message的引用计数加1(此时引用计数为2),由于此时请求r的retcode等于EASY_AGAIN,所以和同步处理不同的是,这里会将该请求加入到该连接的session_list链表中,然后激活该连接的写(ev_io_start(c->loop, &c->write_watcher))。函数结束。
随后调用easy_connection_write_socket,继而调用c->write(实际是easy_socket_write)将数据写入连接,同时回调buf的cleanup,message引用计数减1(此时引用计数等于1)由于fd可写,随后下次event loop会直接返回,回调
该连接write_watcher的回调函数easy_connection_on_writable方法:
1. 调用easy_connection_write_socket继续写数据,如果没有数据可写,将write_watcher停掉
2. 将连接中的session_list拿出来,对于每个请求调用应用层process方法,这是由于请求r的返回值retcode是EASY_AGAIN,所以在proces中,会调用easy_client_wait_wakeup_request(message引用计数加1,此时引用计数为2),从而signal信号量,将工作线程唤醒。这是工作线程发第一个包的情形,随后发第二个包,第三个包……如此反复。
当发最后一个包时,message的引用计数为2,process函数返回EASY_OK,将最后一个包挂在请求r的opacket上,然后执行easy_request_wakeup(r)将请求挂在IO线程的已完成链表中,然后唤醒IO线程,IO线程执行回调函数easy_connection_on_wakeup,在函数最后调用easy_connection_send_response处理这个IO线程所有已完成请求,处理流程如下:
对于每个请求,在这里只关注request的最后一个回复包:
1. 将请求r所在的message的引用计数减1(此时引用计数为1)
2. 执行easy_connection_request_done,将请求r的opacket挂在连接的输出缓冲区链表output上。
3. 调用easy_request_set_cleanup,设置buf的cleanup,并且将message引用计数加1(此时引用计数为2)
4. 由于请求r的返回码retcode为EASY_OK,则更新一些统计信息,将r的status状态值为EASY_REQUEST_DONE
5. 判断是否message上还有请求,这里已经没有了,调用easy_message_destroy(m, 1),将message从连接的message链表中删除,并将message的引用计数减1,,此时message的引用计数为1.
6. 调用easy_connection_write_socket将数据写出,然后调用easy_buf_destroy(buf),从而会调buf的cleanup函数easy_request_cleanup,将message的引用计数减1,此时引用计数变成0,将message的pool销毁,从而内存释放。
六、资源管理
可以看出,libeasy管理内存是以连接为单位的,更具体说,是基于message的。当新建一个连接的时候,会分配1+N个内存池,N指连接上message的个数。
1:用来存储连接本身的元信息(easy_connection_t)
N:每个完整的message会分配一个对应的内存池,这个内存池用来存储请求元信息(easy_request_t),这个message的输入缓冲区,输出缓冲区的内存可以应用层进行分配,也可以由这个内存池进行分配。
message的销毁和连接的销毁都是基于引用计数,当引用计数为0时,相应的内存池被销毁。
message的引用计数加1的场景:
1. 将结果包挂在连接的输出缓冲区时引用计数加1,因为结果包的元信息easy_buf_t是在message的内存池中分配的,结果没有返回,内存池不能销毁。
2. 异步处理模式下process时引用计数也要加1,因为请求显然还没有处理完成,而请求也是在message的内存池中分配的,内存池不能销毁。
3. 工作线程被IO线程唤醒,工作线程继续处理,相当于请求还没有结束,理由同2,message的引用计数继续+1,内存池不能被销毁
message的引用计数减1的场景:
1. 当结果buf被发出去后,调用cleanup函数将message的引用计数减1,与上面的1对应。
2. 当IO线程被唤醒后,处理被工作线程标记为已完成(easy_request_wakeup)的请求(easy_connection_send_response函数中)时,将message的引用计数减1,与上面的2对应。
3. IO线程处理完请求,确认是否message上面的请求是否都已经处理完成(ret == EASY_OK && m->request_list_count == 0 && m->status != EASY_MESG_READ_AGAIN),如果处理完成则将message的引用计数减1。需要注意的是,message的引用计数初始状态为1。这次的引用计数减1正是对应这个情况。
相对与message的引用计数,connection的引用计数简单很多,因为本身connection对应于的pool只存储connection相关的元数据。
connection的引用计数加1的场景:
1. 应用层的process函数中将连接的引用计数加1。
2. 当IO线程通过easy_client_wait_wakeup_request 唤醒工作线程时,将连接的引用计数加1。
connection的引用计数减1的场景:
1. 每次IO线程调用easy_connection_send_response给所有的已完成请求发应答后,将连接的引用计数减1
connection的销毁:
每次对连接进行读写,都会更新其last_time,如果连接的引用计数大于0的时候(异常情况下)执行了easy_connection_destroy(),则会将连接的读写watcher关掉,但是连接没有关掉,导致OS实际上还在接包,但是libeasy没有对其进行处理,导致超时,在这种情况下,会起一个周期性的timeout watcher,每0.5秒检查一下:对于状态不等于EASY_CONN_OK的连接,判断现在距离c->last_time是否超过了时间force_destroy_second,如果超过了,则将连接的引用计数强行置为0,随后close连接并且销毁。
上一篇:OceanBase使用libeasy原理源码分析:客户端
下一篇:【解决】mysql_real_connect: Can't connect to local MySQL server through socket '/tmp/mysql.sock'