OceanBase server处理网络包的回调逻辑 – 刘浩de技术博客

OceanBase处理网络包的逻辑还是蛮绕的,这里以UPS为例,作为给自己的备忘。

 

UPS代码的main.cpp中调用ObUpdateServerMain的start启动server。start函数会调用ObUpdateServerMain的do_work函数,此函数调用ObUpdateServer类的start启动UPS。

 

ObUpdateServer继承自如下几个类:

1)common::ObBaseServer,基础的server类,要求派生类实现handlePacket和handleBatchPacket方法(后一个UPS不会使用),这两个是纯虚函数,同时有个虚函数start_service,派生类可以实现此方法用于初始化时做些server相关的逻辑;

2)ObPacketQueueHandler,  要求派生类实现handlePacketQueue接口;

3)common::IBatchPacketQueueHandler,要求派生类实现handleBatchPacketQueue接口;

 

ObUpdateServer类的start函数实际上调用的ObBaseServer的start函数,此函数做如下几件事:

1)初始化libeasy相关的一系列数据结构, 在特定的网络端口上监听等;

2)调用initialize函数(需要派生类实现);

3)调用start_service函数(派生类可自己实现);

 

ObUpdateServer的initialize函数,做一系列启动UPS内部所需的参数的初始化,除了必要的参数外,有几个需要特别注意下:

1)server_handler_,此变量的类型是easy_io_handler_pt,此处注册一系列回调,包括收包时的编解码(server_handler_.encode/server_handler_.decode),收包时的处理逻辑(server_handler_.process),断开连接时的处理等(server_handler_.on_disconnect)等;编解码相关等都是使用的所有OB server共用的ObTbnetCallback类的方法,版本升级时包的格式变化只需要统一改此common类的方法即可;包的处理逻辑调用的UPS自己的ObUpdateCallback::process方法;

2)client_manager_的初始化,后续server发包都需要用到client_manager_;

3)rpc相关的初始化;

4)一系列worker线程的初始化(如read_thread_queue_/write_thread_queue_/lease_thread_queue_等,具体看代码,指定不同类型线程的线程数,将this传给各个线程用作handler回调等);

 

ObUpdateServer的start_service函数,调用start_threads(此函数调用上述一系列worker线程的start方法,等候收到任务处理),start_timer_schedule设定一系列ObTimer设定一些定时器相关的处理逻辑。

 

如上和网络包处理相关的一系列函数说清楚了,然后具体说说网络包收到后是怎么处理的。

libeasy收到包会调用注册的回调方法ObUpdateCallback::process,此函数会调用ObUpdateServer的handlePacket方法,此函数由各server自己实现,如UPS,会根据packet_code不同,或者进行本地处理(因为libeasy有多个IO poll thread,所以本地可以处理一些耗时少的操作包),或是扔给不同的worker类处理。这里分别举个例子:

1)packet_code为OB_SET_OBI_ROLE类型的包,调用read_thread_queue_的push方法,交由read_thread_queue_线程处理;

2)packet_code为OB_UPS_CLEAR_FATAL_STATUS类型的包,调用lease_thread_queue_的push方法,交由lease_thread_queue_线程处理;

3)OB_WRITE/OB_MS_MUTATE/OB_FAKE_WRITE_FOR_KEEP_ALIVE/OB_UPS_PHY_PLAN_EXECUTE/OB_START_TRANSACTION/OB_END_TRANSACTION/OB_WRITE_DUMMY_LOG类型的包,直接在IO线程中调用trans_executor_的handle_packet函数处理;这里注意下,trans_executor_的handle_packet也会进一步判断包的类型,确定是直接在此函数中处理(transe_executor_类对不同类型的包实现了一系列的handle函数),还是通过调用push_task_交由trans_executor自己的一系列worker线程处理。

 

read_thread_queue_等worker线程是common::ObPacketQueueThread类型的(或者其他类似类型)实例,此类类型都是tbsys::CDefaultRunnable的派生类,其内部含有一个阻塞队列等消息到来,同时有一个用于回调的ObPacketQueueHandler类型的handler(注意ObUpdateServer就继承自ObPacketQueueHandler,实际上此处注册的handler就是ObUpdateServer类的对象)。此类worker线程start后真正调用的是ObPacketQueueThread::run函数,此函数即从内部队列中取消息,然后调用ObUpdateServer的handlePacketQueue函数,handlePacketQueue函数再根据包的类型调用ObUpdateServer类实现的各类方法对包进行处理。此处由于是worker线程,所有包都是线程本地处理的。

 

至此,整个处理包的逻辑应该被串起来了。其实主要是初看代码时handlePacket/handlePacketQueue/handleBatchPacket/handleBatchPacketQueue各由谁调用,以及在哪个线程空间中被调用不太好理解。

本文链接:OceanBase server处理网络包的回调逻辑,转载请注明。



You must enable javascript to see captcha here!

Copyright © All Rights Reserved · Green Hope Theme by Sivan & schiy · Proudly powered by WordPress

无觅相关文章插件,快速提升流量