OceanBase使用libeasy原理源码分析:客户端
这篇主要写libeasy作为客户端的情形。结合OceanBase 0.4的mergeserver使用libeasy的情况进行分析,mergeserver请求chunkserver是异步请求,chunkserver使用update是同步请求,在libeasy中,同步请求是通过异步请求来实现。使用libeasy作为服务器端的情况见《OceanBase使用libeasy原理源码分析:服务器端》,作为客户端使用时,会涉及到一些数据结构,easy_session_t, easy_client_t, easy_hash_t, easy_hash_list_t等。
easy_session_t用来封装一个要发出去的请求,easy_client_t用来封装一个TCP连接的发起端,easy_hash_t是一个哈希表,easy_hash_list_t是采用开放寻址解决hash冲突的链表节点。
一、异步请求
在OceanBase 0.4 中,mergeserver请求chunkserver是异步的方式。实际使用了libeasy的如下两个上层接口:
easy_session_t *easy_session_create(int64_t asize)
int easy_client_dispatch(easy_io_t *eio, easy_addr_t addr, easy_session_t *s)
首先看看session这个关键的数据结构:
#define EASY_MESSAGE_SESSION_HEADER \
easy_connection_t *c; \
easy_pool_t *pool; \
int8_t type; \
int8_t async; \ //1 代表这是个异步session
int8_t status; \
int8_t error;
struct easy_session_t {
EASY_MESSAGE_SESSION_HEADER;
ev_tstamp timeout, now;
ev_timer timeout_watcher;
// 用于串入session list的链表节点
easy_list_t session_list_node;
//为了快速根据packet id定位到发送队列中的session,将session加入到发送队列中时,同时,将其加入到一个hash表中,hash表采用链表的方式将同一个bucket的元素连接起来,
//链表元素就是easy_hash_list_t类型
easy_hash_list_t send_queue_hash;
// 用于串入发送队列链表的节点
easy_list_t send_queue_list;
// 当session的包返回时处理响应的回调函数,对于OceanBase 0.4的mergeserver来说,实际上是int ObMergeCallback::sql_process(easy_request_t* r)
// 将响应事件进mergeserver的finish queue队列
easy_io_process_pt *process;
easy_io_cleanup_pt *cleanup;
easy_addr_t addr;
easy_list_t *nextb;<br> // 用于作为session的hash key,用于收到session响应时快速的从发送队列中删除相应的session
uint64_t packet_id;
// 各种回调函数的集合
void *thread_ptr;
//该session封装的请求,一个session只封装一个请求
easy_request_t r;
char data[0];
};
libeasy作为客户端时,将每个发往libeasy服务器端的请求包封装成一个session(easy_session_t),客户端将这个session放入连接的队列中然后返回,随后收到包后,将相应的session从连接的发送队列中删除,使用了hash加快查找,以packet id为hash key。在超时处理上,tbnet的包超时是基于连接的,即在一个连接上的所有的包的超时时间是一样的。libeasy的超时粒度基于包,得益于底层libev实现的watcher机制。
发包:
mergeserver给chunkserver发异步请求时,会首先分配一块内存,用来容纳easy_session_t和请求包,通过easy_session_t *easy_session_create(int64_t asize)函数来实现,这个函数主要就做了一件事:分配一个内存池easy_pool_t,在内存池头部放置一个easy_session_t,剩下部分存放实际的数据包ObPacket,然后将session的type设置为EASY_TYPE_SESSION。
随后,将实际包挂在这个session包装的request的opacket上,设置session的目标server地址,最后通过easy_session_set_timeout设置这个session的超时时间timeout,这也就是这个请求的超时时间。
然后调用int easy_client_dispatch(easy_io_t *eio, easy_addr_t addr, easy_session_t *s)将这个session分发到具体TCP连接:
1. 调用easy_thread_pool_hash 从IO线程池中选一个IO线程来执行这次session的发送
2. 将这个session加入到这个IO线程的session list中
3. 唤醒这个IO线程开始干活
如上一篇中介绍,IO线程被唤醒的回调函数easy_connection_on_wakeup,继而调用easy_connection_send_session_list函数,这个函数主要做几件事:
1. 对于每个session,调用easy_connection_do_client,找到一条connection用于发送这个session
1.1 从IO线程的client_list(easy_hash_t)这个hash table中找出一个easy_client_t(封装一个connection的发起端),对端地址是session->addr
1.2 如果要发送的session是一个正常的session,则设置其回调函数process后,返回这个client封装的connection,如果是一个连接的session,并且没有现成的到对端的connection,则从IO线程的client_array(easy_array_t)中拿一个空闲的easy_client_t槽位,然后初始化它,并将其加入到IO线程的client_list中,再调用easy_connection_do_connect建立到对端的连接,最后返回该连接。如果是一个断开的session,则将这个client从client_list删除,并且将client占的槽位还给client_array。
2. 调用easy_connection_session_build:
2.1. 设置session的cleanup函数
2.2 调用easy_connection_get_packet_id(回调应用层定义的get_packet_id函数,对于OceanBase 0.4来说,对于packet的channel以一递增),为session封装的请求产生packet id,主要用于接收到session的响应时用于快速定位到发送队列将session删除。
2.3 回调应用层定义的encode函数,在这里,是int ObTbnetCallback::encode(easy_request_t *r, void *data)函数,在请求所在message的pool分配一个easy_buf_t,并且将包拷进去,然后调用easy_request_addbuf将这个新的buf挂在连接的output输出缓冲区链表中。
2.4 调用easy_hash_dlist_add(c->send_queue, s->packet_id, &s->send_queue_hash, &s->send_queue_list)将这个session添加到连接的发送队列中。这个函数将session添加到发送队列的同时,同时将相应的项添加到hash表的相应的bucket的链表头(easy_hash_t和easy_hash_list_t实现)
2.5 将session串入到连接的session list中。并且启动timeout watcher,事件为之前为session设置的timeout时间
3. 调用easy_connection_write_socket将所有这些session被分发到的connection的output输出缓冲区链表的数据写出。
收包:
如前所述,当fd可读时,会回调easy_connection_on_readable:
从内核缓冲区读入数据到应用层输入缓冲区中读入数据,一样,会初始化一个easy_message_t m存放请求,m的初始引用计数为1,然后调用easy_connection_do_response进行处理,libeasy作为服务器端时,调用的是easy_connection_do_request
easy_connection_do_response做几件事:
1. 回调应用层定义的decode的,从message的inputbuffer中解析出一个个的packet,将packet挂在session封装的请求r的ipacket上,并且将整个message挂在session封装的请求r的后面,相当于将请求和响应关联起来(s->r.request_list_node.next = (easy_list_t *) m)。对于OceanBase来说,是ObPacket, 由于是异步请求(s->async==1),将m的引用计数加1,并且同样,调用easy_connection_get_packet_id方法得到这个packet的packetid,然后调用s = (easy_session_t *) easy_hash_dlist_del(c->send_queue, packet_id)将这个packet所属的session从发送队列中摘下来
2. 由于这个session的响应收到了,将其timeout watcher停掉
3. 将这个session从连接的session list中删除
4. 调用easy_session_process:
4.1 回调应用层的process方法,在这里,是int ObMergeCallback::sql_process(easy_request_t* r),将包放入mergeserver的finish queue中。然后调用
easy_session_destroy(r->ms),该函数会调用easy_message_destroy(m, 0)将与这个session相关的响应message的引用计数减1,最后session使用的pool销毁。
5. 最后看是否message上是否还有request没处理完成,没有request要处理了,则调用easy_message_destroy(m, 1)将message的引用计数减1,将message从连接的message list删除,此时message引用计数为0,销毁message使用的pool。
二、同步请求
首先,同异步请求类似,也是先创建一个session,将数据包拷进去,不同的是,同步请求调用的是void *easy_client_send(easy_io_t *eio, easy_addr_t addr, easy_session_t *s)方法而不是int easy_client_dispatch(easy_io_t *eio, easy_addr_t addr, easy_session_t *s)
其实,同步是异步的一种特殊的情况,同步请求也是通过异步请求实现的。
void *easy_client_send(easy_io_t *eio, easy_addr_t addr, easy_session_t *s)方法封装了异步请求用到的easy_client_dispatch,easy_client_send将session
的process置为easy_client_wait_process方法,然后初始化一个easy_client_wait_t wobj,随后调用easy_client_dispatch方法走异步请求那一套,接着让自己wait在
wobj包装的一个信号量上,当这个请求收到包的时候回调easy_client_wait_process方法,signal这个信号量,然后返回session封装的请求的ipacket。