rados bench 是 Ceph 自带的用来测试存储池性能的压测工具，其 main 函数在 src/tools/rados/rados.cc 中。rados.cc 集成了 rados 所有 bash 命令，可以通过 rados -h 查看帮助。开篇点题，让我们看看执行 rados bench 时，哪些线程被偷偷启动了。

#在一个终端开启 rados bench
[root@localhost build]# ./bin/rados -p rbd bench 60 write#在另一个终端监控 rados 启动哪些线程
[root@localhost ~]# ps -ef | grep rados
root      17796  10743  0 16:08 pts/0    00:00:00 ./bin/rados bench -p rbd 60 write
root      17808  10765  0 16:09 pts/1    00:00:00 grep --color=auto rados
[root@localhost ~]# top -Hp 17796 -d 0.5
top - 16:19:51 up  6:36,  2 users,  load average: 2.54, 0.73, 0.35
Threads:  16 total,   0 running,  16 sleeping,   0 stopped,   0 zombie
%Cpu(s):  0.0 us,  0.3 sy,  0.0 ni, 87.4 id, 12.3 wa,  0.0 hi,  0.0 si,  0.0 st
KiB Mem :  3861288 total,   201816 free,  1232584 used,  2426888 buff/cache
KiB Swap:  4063228 total,  4063228 free,        0 used.  2327116 avail Mem PID USER      PR  NI    VIRT    RES    SHR S %CPU %MEM     TIME+ COMMAND                        20207 root      20   0 1328848  95972  20544 S  0.0  2.5   0:00.28 rados                          20208 root      20   0 1328848  95972  20544 S  0.0  2.5   0:00.00 msgr-worker-0                  20209 root      20   0 1328848  95972  20544 S  0.0  2.5   0:00.49 msgr-worker-1                  20210 root      20   0 1328848  95972  20544 S  0.0  2.5   0:01.60 msgr-worker-2                  20215 root      20   0 1328848  95972  20544 S  0.0  2.5   0:00.00 log                            20216 root      20   0 1328848  95972  20544 S  0.0  2.5   0:00.00 service                        20217 root      20   0 1328848  95972  20544 S  0.0  2.5   0:00.00 admin_socket                   20222 root      20   0 1328848  95972  20544 S  0.0  2.5   0:00.00 rados                          20223 root      20   0 1328848  95972  20544 S  0.0  2.5   0:00.00 ms_dispatch                    20224 root      20   0 1328848  95972  20544 S  0.0  2.5   0:00.00 ms_local                       20225 root      20   0 1328848  95972  20544 S  0.0  2.5   0:00.00 safe_timer                     20226 root      20   0 1328848  95972  20544 S  0.0  2.5   0:00.00 fn_anonymous                   20227 root      20   0 1328848  95972  20544 S  0.0  2.5   0:00.00 safe_timer                     20228 root      20   0 1328848  95972  20544 S  0.0  2.5   0:00.00 safe_timer                     20229 root      20   0 1328848  95972  20544 S  0.0  2.5   0:00.01 fn-radosclient                 20230 root      20   0 1328848  95972  20544 S  0.0  2.5   0:00.00 write_stat

可以看到除了本身执行 rados bench 命令的线程外，还有十几个线程被创建出来。简单介绍下，线程号为20207的 rados 线程是 rados bench 命令的主线程，msgr-worker-0~3 是用来网络通信的三个工作线程，log 是用来记录日志的日志线程，service 是监控本地性能参数的线程，admin_socket 是对外提供查询接口的线程，ms_dispatch 和 ms_local 是用来处理消息分发的线程，safer_timer 是执行定时任务的线程，fn_anonymous 和 fn-radosclient 则是处理所有模块收尾工作的线程。

1. rados main()

以下给出了 rados 模块的 CMakeLists.txt 文件，可以看到 rados 一共包含了哪些源文件和链接了什么库。

# rados 源文件
set(rados_srcs rados/rados.ccRadosDump.ccrados/RadosImport.ccrados/PoolDump.cc${PROJECT_SOURCE_DIR}/src/common/util.cc${PROJECT_SOURCE_DIR}/src/common/obj_bencher.cc${PROJECT_SOURCE_DIR}/src/osd/ECUtil.cc)
add_executable(rados ${rados_srcs})
# 链接库
target_link_libraries(rados librados global ${BLKID_LIBRARIES} ${CMAKE_DL_LIBS})
if(WITH_LIBRADOSSTRIPER)target_link_libraries(rados radosstriper)
else()target_link_libraries(rados cls_lock_client)
endif()
install(TARGETS rados DESTINATION bin)

/src/tools/rados.cc 中提供了 rados 工具的入口 main() 函数。其实，main()函数主要进行了4步工作：参数解析、全局初始化、初始化 CephContest和调用 rados_tool_common() 函数。在传入 main() 的参数中：argc 表示传入的参数个数，argv 以二维数组的方式记录了参数值。函数argv_to_vec(argc, argc, args)功能是把参数解析，并存入 args vector 中。例如，输入命令：rados bench -p rbd 10 write。agrs解析得到后的结果[“bench”, “-p”, “rbd”, “10”, “write”]。后续的的 if-else 循环中，依然是参数解析，这一步的作用是解析参数名和参数值，把结果放在 opts map 中。例如“-p rbd”被解析成“pool : rbd”。

global_init()，即全局初始化，参数 CEPH_ENTITY_TYPE_CLIENT 和 CODE_ENVIRONMENT_UTILITY 分别表示该模块是 CLIENT 和代码环境为应用程序。此外还有 MON 、MDS、OSD等模块，DEAMON、LIBRARY 等代码环境。

后文将逐步介绍 global_init()、common_init_finish()、rados_tool_common()。

//rados.ccint main(int argc, const char **argv){...;argv_to_vec(argc, argv, args); //参数解析...;auto cct = global_init(NULL, args, CEPH_ENTITY_TYPE_CLIENT, CODE_ENVIRONMENT_UTILITY, 0);common_init_finish(g_ceph_context);//初始化 CephContext...;else if (ceph_argparse_flag(args, i, "--force-full", (char*)NULL)) //参数解析，放入 map 中opts["force-full"] = "true";...;return rados_tool_common(opts, args);//根据 args 调用相关 rados 命令
}

2. global_init()

global_init() 是全局初始化函数，所有的 ceph 相关进程（rados，ceph，rbd等）都需要执行该操作，msgr-worker 和 log 线程都是在这一步创建的。该函数主要目的是进行参数的初始化解析工作、创建 CephContext、开启一些基础线程以及执行一些通用的预处理工作。

run_pre_init 标志位默认是 true，所以通常情况下，都会执行 global_pre_init()。下文会详细介绍。

block_signals() ，此函数屏蔽了 siglist[] 中的信号，这里是 SIGPIPE 信号。SIGPIPE 信号是服务端断开链接后发送给客户端的信号，一般来说客户端收到 SIGPIPE 信号会立刻中断进程。Ceph 不希望出现客户端突然中断，所以这里屏蔽了该信号。install_standard_sighandlers() 则是制定了某些信号的处理程序，包括：SIGSEGV、SIGABRT、SIGBUS、SIGILL、SIGFPE、SIGXCPU、SIGCFSZ、SIGSYS，具体处理方法见 handle_fatal_signal() 函数（本文这里不做分析）。

g_ceph_context->_log->set_flush_on_exit() 向退出函数注册 log_on_exit() 回调函数，该函数旨在退出时清空 log 指针，避免内存泄漏。

在 getuid() 循环体中，如果当前是 root 用户，则更改配置文件中的 userid 和 gruopid 属性为0（最高权限），并设置环境变量 HOME 为当前家目录。

globla_init() 最终返回一个 g_ceph_context 指针，该指针实际指向一个 CephContext 对象。CephContext表示由单个库用户持有的上下文。在同一个进程中可以有多个CephContexts。对于守护进程和实际程序（如本文最开始的 rados bench…），只有一个CephContext。CephContext包含配置、dout对象以及您可能希望在每次函数调用时传递给libcommon的任何其他内容。

boost::intrusive_ptr<CephContext>
global_init(const std::map<std::string,std::string> *defaults,std::vector < const char* >& args,uint32_t module_type, code_environment_t code_env,int flags,const char *data_dir_option, bool run_pre_init)
{//标志位，避免一个进程中多次执行 global_init()static bool first_run = true;//执行 global_pre_init()...if (run_pre_init) {ceph_assert(!g_ceph_context && first_run);global_pre_init(defaults, args, module_type, code_env, flags);} else {ceph_assert(g_ceph_context && first_run);}//更改标志位，避免重复执行global_init()first_run = false;//屏蔽 SIGPIPE 信号int siglist[] = { SIGPIPE, 0 };block_signals(siglist, NULL);//加载信号处理器if (g_conf()->fatal_signal_handlers) {install_standard_sighandlers();}//退出时删除 log 指针，注意是内存中的 log 指针，不是日志文件if (g_conf()->log_flush_on_exit)g_ceph_context->_log->set_flush_on_exit();//读取 --setuser 配置参数，如果当前不是 root 则什么也不做，如果时 root，则根据配置参数更改if (getuid() != 0) {...}else if (g_conf()->setgroup.length() ||g_conf()->setuser.length()) {...g_ceph_context->set_uid_gid(uid, gid);g_ceph_context->set_uid_gid_strings(uid_string, gid_string);...setenv("HOME", home_directory.c_str(), 1)...}//生效更改的配置// Expand metavariables. Invoke configuration observers. Open log file.g_conf().apply_changes(nullptr);// call all observers now.  this has the side-effect of configuring// and opening the log file immediately.g_conf().call_all_observers();//打印先前发生的错误到日志文件中g_conf().complain_about_parse_errors(g_ceph_context);...//内存泄漏检测if (g_conf()->debug_deliberately_leak_memory) {derr << "deliberately leaking some memory" << dendl;char *s = new char[1234567];(void)s;// cppcheck-suppress memleak} ...return boost::intrusive_ptr<CephContext>{g_ceph_context, false};
}

2.1 global_pre_init()

global_pre_init() 是 global_init() 的预执行步骤，主要解析配置、启动 MonClient、启动 log 线程。这里的解析配置的顺序是：部分命令行参数、默认配置、配置文件 ceph.conf、环境变量、命令行参数。后一步的配置可以覆盖前一步获得的配置参数。因此命令行参数的优先级最高，默认配置的优先级最低。

ceph_argparse_early_args() 就是用来预解释命令含参数的，包括：conf、cluster、id、name。此外遇到“–version”则会直接打印版本号，然后退出程序；遇到“–show_args”则会打印出所有的命令行参数。

common_preinit() 返回了一个 CephContext 实例，对于一个实例进程来说，贯穿一生的唯一 CephContext 其实就是在这里创建的，上文 global_init() 返回的 g_ceph_context 指针实际就是这个 CephContext 对象。其中包括了 log、配置参数、网络通信等等一个 Ceph 进程所包含的各种模块的信息或者实例指针。

conf.parse_config_files() 读取配置文件的参数。默认使用 /etc/ceph/ceph.conf。conf.parse_env() 从环境读取“CEPH_ARGS”参数。conf.parse_argv() 再从命令行读取配置参数。

cct->_log->start() 启动 log 线程，开启日志。

mc_bootstrap.get_monmap_and_config() 配置网络通信，并开启3个 msgr_worker 线程。

do_argv_commands() 执行 --show-config[-val] 命令，打印指定参数值。 g_conf().complain_about_parse_errors() 则是向日志中写入错误信息。因为日志刚刚开启，所以没有在一开始发生错误的时候就写日志，而是收集起来，最后一起写入日志文件。

void global_pre_init(const std::map<std::string,std::string> *defaults,std::vector < const char* >& args,uint32_t module_type, code_environment_t code_env,int flags)
{...//预读取命令行参数CephInitParameters iparams = ceph_argparse_early_args(args, module_type,&cluster, &conf_file_list);...//创建 CephContextCephContext *cct = common_preinit(iparams, code_env, flags);cct->_conf->cluster = cluster;global_init_set_globals(cct);auto& conf = cct->_conf;...//读取配置文件参数int ret = conf.parse_config_files(c_str_or_null(conf_file_list),&cerr, flags);...//读取环境变量参数conf.parse_env(cct->get_module_type());//读取命令行参数conf.parse_argv(args);//开启 logif (conf->log_early &&!cct->_log->is_started()) {cct->_log->start();}//开启网络通信if (!conf->no_mon_config) {conf.apply_changes(nullptr);MonClient mc_bootstrap(g_ceph_context);if (mc_bootstrap.get_monmap_and_config() < 0) {cct->_log->flush();cerr << "failed to fetch mon config (--no-mon-config to skip)"<< std::endl;_exit(1);}}// do the --show-config[-val], if present in argvconf.do_argv_commands();// Now we're ready to complain about config file parse errorsg_conf().complain_about_parse_errors(g_ceph_context);
}

2.1.1 log

Ceph 对每个子系统的日志都预先定义了日志级别，支持动态修改。每条 log 都带有日志级别，低于预先定义的级别才会被打印，预定义级别可以在配置文件中修改，并且可以针对某个模块，如 osd、rbd 等。log 模块有单独的线程，只需要向该线程提交 job 即可，是异步的过程，对系统的性能影响很小。

下面给出了 log 类的头文件，里面定义了 log 所需的变量以及方法。DEFAULT_MAX_NEW =100：表示新建日志的队列深度为100条，DEFAULT_MAX_RECENT=10000 表示最多记录最近写的10000条日志记录。

log 类其实可以简单的通过三个队列来理解。m_new 储存新加入的待写日志，m_recent 存储最近写完的日志，m_flush 表示即将写入日志文件的日志。就像是3个杯子，首先向 new 杯子倒水，如果不把 new 杯子里的水倒掉，最多倒100次 new 杯子就满了。new 杯子的水则倒向向 flush 杯子，flush杯子像是一个临时的中转杯，水在这里短暂停留，一旦用小本本记下了这杯水，就立刻把 flush 杯子的水倒入 recent 杯子。看起来 recent 杯子好像是个垃圾桶，明明都已经记载在小本上了，却还要留着这杯水，这是为了方便我们后续可以直接通过 recent 杯子查看倒了多少水、什么水。但是 recent 杯子也不是无底洞，它最多可以存10000次，满了就把杯底的开关打开，把水从底部放掉，这样可以优先把最不新鲜的水倒掉。

log 线程则通过 start() 来创建，具体调用链：cct->_log->start() -> Thread::create() -> Thread::try_create() -> pthread_create()。线程具体方法则是 *Log::entry()。

class Log : private Thread
{...static const std::size_t DEFAULT_MAX_NEW = 100;static const std::size_t DEFAULT_MAX_RECENT = 10000;...//三个队列，分别是最新的entries（待写日志）队列，最近写完的日志队列和即将写的日志队列（m_new flush后得到）EntryVector m_new;    ///< new entriesEntryRing m_recent; ///< recent (less new) entries we've already written at low detailEntryVector m_flush; ///< entries to be flushed (here to optimize heap allocations)...  //线程入口void *entry() override;...public:using Thread::is_started;Log(const SubsystemMap *s);~Log() override;//退出时 flush 一次，清空 m_new，保证写入所有日志void set_flush_on_exit();...//刷新void flush();//打印最近的日志void dump_recent();...//提交日志void submit_entry(Entry&& e);//开启线程和终止线程void start();void stop();...
}

注意上面 log 的构造函数： Log(const SubsystemMap *s)，创建 Log 实例需要传入 SubsystemMap 。这个 map 实际也是在 CephContext 构造函数初始化列表中创建的。主要用来描述单个子系统的日志信息（即子模块，类似 rbd、rbd bench）。此处的日志级别区别与每条日志的级别，只有当日志级别小于等于该子系统日志级别时才会被打印，可在配置文件中更改子系统日志级别。

struct Subsystem {int log_level; // 日志级别int gather_level; // gather级别std::string name; // 子系统名称Subsystem() : log_level(0), gather_level(0) {}
};

下面给出了日志写入文件操作的部分源代码。简单叙述整个过程，就是把 m_new 队列数据转移到 m_flush，读取 m_flush 队列中的每条记录，判断是否满足日志级别小于子系统级别的要求，同时把数据复制到 m_recent 队列，然后经过层层调用，最终通过 write() 写入日志文件。

void *Log::entry()
{reopen_log_file(); //打开日志文件，返回 fd 文件号，记录在 Log.cc 中的 m_fd 局部变量中{...//循环刷新日志，m_new 转移到 m_flush，具体在 flush() 函数实现。while (!m_stop) {if (!m_new.empty()) {...flush();...continue;}...}...}//线程结束时，再次刷新flush();return NULL;
}void Log::flush()
{{...m_flush.swap(m_new);//把 m_new 队列数据转移到 m_flushm_cond_loggers.notify_all();//唤醒所有线程...}_flush(m_flush, true, false);//实例化 m_flush 队列，数据记录在 m_recent...
}void Log::_flush(EntryVector& t, bool requeue, bool crash)
{...for (auto& e : t) {//判断本条日志级别是否小于等于子系统日志级别，小于则为truebool should_log = crash || m_subs->get_log_level(sub) >= prio;//日志文件是否成功打开，是否写本条日志bool do_fd = m_fd >= 0 && should_log;bool do_syslog = m_syslog_crash >= prio && should_log;bool do_stderr = m_stderr_crash >= prio && should_log;bool do_graylog2 = m_graylog_crash >= prio && should_log;if (do_fd || do_syslog || do_stderr) {...if (do_syslog) {syslog(LOG_USER|LOG_INFO, "%s", pos);}if (do_stderr) {std::cerr << m_log_stderr_prefix << std::string_view(pos, used) << std::endl;}...if (do_fd) {...if (m_log_buf.size() > MAX_LOG_BUF) {_flush_logbuf();}}//把数据存储压入m_recentif (requeue) {m_recent.push_back(std::move(e));}}t.clear();_flush_logbuf();
}//再经过几次调用，最终通过 write() 函数把日志序列化到文件中。
void Log::_flush_logbuf()
{if (m_log_buf.size()) {_log_safe_write(std::string_view(m_log_buf.data(), m_log_buf.size()));m_log_buf.resize(0);}
}

大致了解了 log 线程的工作模式后，让我们来看看如何通过代码的方式记录日志，以及它是如何实现的？

随机挑选一条日志记录代码，具体如下。

ldout(cct, 20) << "Log depth=" << depth << " x=" << x << dendl;

可以看到和我们通常使用的 cout 命令有点像。<< 拼接字符串，dendl 结尾，ldout() 中传入两个参数，第一个是固定的 cct （CephContext），第二个代表日志级别：error = -1、warn = 0、info = 1、debug = 5、20 = trace。了解这些，已经能够完成记录日志的操作了。

ldout 主要在dout.h 文件中通过宏定义来实现，如下给出了3个日志输出方式：lsubdout、ldout、lderr，都通过 dout_impl() 方法实现。should_gather 标志位，判定是否满足输出级别要求。创建 MutableEntry 实例，通过submit_entr() 方法提交到 m_new 队列，交由 log 线程处理。

#define lsubdout(cct, sub, v)  dout_impl(cct, ceph_subsys_##sub, v) dout_prefix
#define ldout(cct, v)  dout_impl(cct, dout_subsys, v) dout_prefix
#define lderr(cct) dout_impl(cct, ceph_subsys_, -1) dout_prefix#define dout_impl(cct, sub, v)                       \do {                                   \const bool should_gather = [&](const auto cctX) {         \if constexpr (ceph::dout::is_dynamic<decltype(sub)>::value ||    \ceph::dout::is_dynamic<decltype(v)>::value) {        \return cctX->_conf->subsys.should_gather(sub, v);            \} else {                               \/* The parentheses are **essential** because commas in angle   \* brackets are NOT ignored on macro expansion! A language's   \* limitation, sorry. */                        \return (cctX->_conf->subsys.template should_gather<sub, v>()); \}                                  \}(cct);                                \\if (should_gather) {                          \ceph::logging::MutableEntry _dout_e(v, sub);                        \static_assert(std::is_convertible<decltype(&*cct),             \CephContext* >::value,      \"provided cct must be compatible with CephContext*"); \auto _dout_cct = cct;                        \std::ostream* _dout = &_dout_e.get_ostream();#define dendl_impl std::flush;                                          \_dout_cct->_log->submit_entry(std::move(_dout_e));                  \}                                                                     \} while (0)

2.1.2 msgr-worker-x

说到 msgr-worker，它实际是 Messenger 模块创建的通信线程，用来监听和发送对象。在《Ceph 源码分析》一书中有简单介绍 SimpleMessenger，通过 pipe 传递消息。但是在目前的N版中，主要使用 AsyncMessenger，AsynsMessenger 使用 epoll 来监听端口发送消息，有异步、多路IO复用等功能，能够提升端口监听的数量以及降低资源消耗。

msgr-worker 创建过程的具体调用链如下：
global_init() -> global_pre_init() -> mc_bootstrap.get_monmap_and_config() -> Messenger::create_client_messenger() -> Messenger::create() -> new AsyncMessenger -> Stack->start()。

网络通信模块实在是纷繁复杂，想要顺畅、完整地介绍它，本人还做不到。这里就简单的介绍下 AsyncMessenger，关于网络通信模块详细介绍留待后续。

就从 Messenger::create() 开始吧。Meseenger 支持三种模式：simple、async、xio，可以通过指定 --ms_type=XX 参数来选择模式。AsyncMessenger 除了支持 posix 传输模式外，还支持 rdma 和 dpdk，通过 --ms_type=async+XX来配置。AsyncMessenger 首先创建一个 StackSingleton 单例，再调用 single->ready()，再到 NetworkStack::create() 函数创建 NetworkStack。

AsyncMessenger::AsyncMessenger(CephContext *cct, entity_name_t name,const std::string &type, string mname, uint64_t _nonce): SimplePolicyMessenger(cct, name,mname, _nonce),dispatch_queue(cct, this, mname),lock("AsyncMessenger::lock"),nonce(_nonce), need_addr(true), did_bind(false),global_seq(0), deleted_lock("AsyncMessenger::deleted_lock"),cluster_protocol(0), stopped(true)
{std::string transport_type = "posix";//支持 rdma 和 dpdk 传输模式if (type.find("rdma") != std::string::npos)transport_type = "rdma";else if (type.find("dpdk") != std::string::npos)transport_type = "dpdk";//创建 StackSingleton 单例auto single = &cct->lookup_or_create_singleton_object<StackSingleton>("AsyncMessenger::NetworkStack::" + transport_type, true, cct);//创建 NetworkStacksingle->ready(transport_type);//创建 PosixWorkerstack = single->stack.get();//开启 msgr-worker 线程stack->start();...
}struct StackSingleton {...void ready(std::string &type) {if (!stack)stack = NetworkStack::create(cct, type);}...
};

NetworkStack 构造函数如下。NetworkStack 中的 workers 用来保存多个 worker，每个 worker 都会创建一个 epoll（大多的网络编程中，都会使用基于事件通知的异步网络 IO 方式来实现，比如 Epoll 和 Kqueue，ceph 的网络模块使用的是Epoll）。worker 的作用主要就是监听端口和建立连接，使用 epoll 来进行 socket 事件驱动。3个 worker 保证了负载均衡，每次调用 connection 都会寻找连接数最少的 worker 线程。

NetworkStack::NetworkStack(CephContext *c, const string &t): type(t), started(false), cct(c)
{//读取 msger_worker 线程的数量配置num_workers = cct->_conf->ms_async_op_threads;...//循环创建 woker，压入 workers 队列，在事件中心初始化for (unsigned i = 0; i < num_workers; ++i) {Worker *w = create_worker(cct, type, i);w->center.init(InitEventNumber, i, type);workers.push_back(w);}
}class PosixWorker : public Worker {//监听端口int listen(entity_addr_t &sa,unsigned addr_slot,const SocketOptions &opt,ServerSocket *socks) override;//连接端口int connect(const entity_addr_t &addr, const SocketOptions &opts, ConnectedSocket *socket) override;
};

在 stack->start() 方法中，add_thread() 创建了线程的方法体（线程开启时，执行该方法），spawn_worker() 中通过 std::thread() 方法创建线程。线程的数量 num_workers 根据配置中的 ms.async.op.threads 来决定，默认为3。

 void NetworkStack::start(){...for (unsigned i = 0; i < num_workers; ++i) { if (workers[i]->is_init())continue;//创建线程主体std::function<void ()> thread = add_thread(i);//开启线程spawn_worker(i, std::move(thread));}...}

add_thread() 中设置了当前线程的名称：msgr-worker-{num}，在事件中心注册 worker 线程，进行初始化工作。while 循环中则一直轮询处理 EventCenter 中的事件。EventCenter 是一个存储事件的容器，它通过注册的回调函数 EventCallbackRef 来针对的处理事件。事件类型有三种：time_events（定时事件）、external_events（外部事件，发送消息）、可读事件（epoll 监听的事件，接收消息）。

std::function<void ()> NetworkStack::add_thread(unsigned i)
{Worker *w = workers[i];return [this, w]() {char tp_name[16];sprintf(tp_name, "msgr-worker-%u", w->id);//线程改名：msgr-worker-xceph_pthread_setname(pthread_self(), tp_name);const unsigned EventMaxWaitUs = 30000000;w->center.set_owner();ldout(cct, 10) << __func__ << " starting" << dendl;w->initialize();w->init_done();//轮询事件中心，处理消息while (!w->done) {ldout(cct, 30) << __func__ << " calling event process" << dendl;ceph::timespan dur;int r = w->center.process_events(EventMaxWaitUs, &dur);if (r < 0) {ldout(cct, 20) << __func__ << " process events failed: "<< cpp_strerror(errno) << dendl;// TODO do something?}w->perf_logger->tinc(l_msgr_running_total_time, dur);}w->reset();w->destroy();};
}

至此，建立了网络通信模块的基础，后续需要连接某个 ip 端口时，需要从 stack 中申请一个负载最小的 msgr_worker 线程，通过 new AsyncConnection() 方式建立连接。可以在上述代码中看到消息处理的蛛丝马迹：w->center.process_events()，消息的传递都经由 EventCenter 事件中心，我们在实际操作中，只需要把消息封装成 message ，通过 dispatch_event_external() 提交给事件中心并注册为外部事件即可，然后由 connection 建立连接时指定的 msgr_worker 线程去处理消息的传递。

总结：global_init() 函数是通用了全局初始化函数，无论是客户端还是服务端，Ceph 的任意子系统都需要进行该初始化。主要内容包括：解析配置（默认配置、命令行参数、配置文件）、开启 log、开启网络通信、加载信号处理器，创建 CephContext。

3. common_init_finish()

common_init_finish() 是在 rados.cc 的 main 函数中，紧接 global_init() 函数出现的。主要作用是开启 service 线程和 admin_socket 线程，具体调用过程：common_init_finish() => cct->start_service_thread() => _service_thread->create()，_admin_socket->init()。其中 service 是一个定时任务，通过 _refresh_perf_values() 方法定时刷新 workers 和 un_healthy_workers 线程的数量，监控 mempool 内存池的容量。admin_socket 则提供对外接口，用于查看当前配置、进程状态、获取 log 等。

void common_init_finish(CephContext *cct)
{// only do this once per cctif (cct->_finished) {return;}cct->_finished = true;//初始化加密设置cct->init_crypto();ZTracer::ztrace_init();//开启日志线程if (!cct->_log->is_started()) {cct->_log->start();}int flags = cct->get_init_flags();if (!(flags & CINIT_FLAG_NO_DAEMON_ACTIONS))//开启 service、admin_socket 线程cct->start_service_thread();...
}void CephContext::start_service_thread()
{{//开启 service 线程_service_thread = new CephContextServiceThread(this);_service_thread->create("service");}// make logs flush on_exit()if (_conf->log_flush_on_exit)_log->set_flush_on_exit();// Trigger callbacks on any config observers that were waiting for// it to become safe to start threads._conf.set_safe_to_start_threads();_conf.call_all_observers();// start admin socketif (_conf->admin_socket.length())_admin_socket->init(_conf->admin_socket);
}

3.1 service

以下给出了 service 线程的入口方法 entry() 。每经过 heartbeat_interval 内部心跳时间，就刷新一次性能参数：l_cct_total_workers、l_cct_unhealthy_workers、mempool中参数，通过 CephContext::_refresh_perf_values() 方法。

class CephContextServiceThread : public Thread
{void *entry() override{while (1) {//定时触发if (_cct->_conf->heartbeat_interval) {auto interval = ceph::make_timespan(_cct->_conf->heartbeat_interval);_cond.wait_for(l, interval);} else_cond.wait(l);//是否重打开日志文件if (_reopen_logs) {_cct->_log->reopen_log_file();_reopen_logs = false;}_cct->_heartbeat_map->check_touch_file();//刷新性能计数器。// refresh the perf coutners_cct->_refresh_perf_values();}return NULL;}
}void CephContext::_refresh_perf_values()
{if (_cct_perf) {_cct_perf->set(l_cct_total_workers, _heartbeat_map->get_total_workers());_cct_perf->set(l_cct_unhealthy_workers, _heartbeat_map->get_unhealthy_workers());}unsigned l = l_mempool_first + 1;for (unsigned i = 0; i < mempool::num_pools; ++i) {mempool::pool_t& p = mempool::get_pool(mempool::pool_index_t(i));//byte 和 items 为原子变量，本身具有锁得特性，所以读写无需上锁。_mempool_perf->set(l++, p.allocated_bytes());_mempool_perf->set(l++, p.allocated_items());}
}

以下给出了 mempool 中性能参数列表，可以通过 ceph daemon osd.0 perf dump mempool 查询指定模块的性能参数。

"mempool": {"bloom_filter_bytes": 0,"bloom_filter_items": 0,"bluestore_alloc_bytes": 98720,"bluestore_alloc_items": 12340,"bluestore_cache_data_bytes": 0,"bluestore_cache_data_items": 0,"bluestore_cache_onode_bytes": 48600,"bluestore_cache_onode_items": 81,"bluestore_cache_other_bytes": 66492,"bluestore_cache_other_items": 8132,"bluestore_fsck_bytes": 0,"bluestore_fsck_items": 0,"bluestore_txc_bytes": 17664,"bluestore_txc_items": 24,"bluestore_writing_deferred_bytes": 425088,"bluestore_writing_deferred_items": 93,"bluestore_writing_bytes": 0,"bluestore_writing_items": 0,"bluefs_bytes": 4936,"bluefs_items": 84,"buffer_anon_bytes": 2232173,"buffer_anon_items": 112,"buffer_meta_bytes": 3784,"buffer_meta_items": 43,"osd_bytes": 286656,"osd_items": 24,"osd_mapbl_bytes": 0,"osd_mapbl_items": 0,"osd_pglog_bytes": 17600,"osd_pglog_items": 40,"osdmap_bytes": 35292,"osdmap_items": 290,"osdmap_mapping_bytes": 0,"osdmap_mapping_items": 0,"pgmap_bytes": 0,"pgmap_items": 0,"mds_co_bytes": 0,"mds_co_items": 0,"unittest_1_bytes": 0,"unittest_1_items": 0,"unittest_2_bytes": 0,"unittest_2_items": 0},

_cct_perf 和 _mempool_pref 都是 PerfCounter 类的实例化对象，PerfCounter 是一个容器，用来记录某种性能参数，根据注释所示，它可以追踪记录四种参数：

* 1) integer values & counters              //整数
* 2) floating-point values & counters       //浮点数
* 3) floating-point averages                //浮点数
* 4) 2D histograms of quantized value pairs //二维柱状图

此外还可以记录时间。以下给出了修改和获取参数的函数：

  //idx 为参数索引void inc(int idx, uint64_t v = 1); //加1void dec(int idx, uint64_t v = 1);//减1void set(int idx, uint64_t v);//设置为v值uint64_t get(int idx) const;//获取//修改时间的函数void tset(int idx, utime_t v);void tinc(int idx, utime_t v);void tinc(int idx, ceph::timespan v);utime_t tget(int idx) const;

注意的是，service 线程中，只定时维护了_cct_perf 和 _mempool_pref 中的参数更新，对于一个 ceph 模块来说还有很多其他的 PerfCounter。用户可以通过自定义的方式，添加性能监控，并手动维护参数更新。这里给出简单的使用方法。

//PerfCountersBuilder 是构造 PerfCounters 对象的类，其构造函数中创建 PerfCounters 对象。
PerfCountersBuilder plb(this, "cct", l_cct_first, l_cct_last);//添加要监控的性能参数名称、描述
plb.add_u64(l_cct_total_workers, "total_workers", "Total workers");
plb.add_u64(l_cct_unhealthy_workers, "unhealthy_workers", "Unhealthy workers");//获取动态分配的指针
_cct_perf = plb.create_perf_counters();//将 PerfCounters 对象加入 collection
_perf_counters_collection->add(_cct_perf);//在程序中手动设置监控参数的值，还有 inc()、dec() 方法。
_cct_perf->set(l_cct_total_workers, _heartbeat_map->get_total_workers());
_cct_perf->set(l_cct_unhealthy_workers, _heartbeat_map->get_unhealthy_workers());

3.2 admin_socket

admin_socket 线程是用来处理 ceph daemon 命令的线程，它和 service 线程一起提供了性能监控服务，service 线程更新各个模块性能参数，admin_socket 线程提供对外查询接口。

其对象 _admin_socket 在 CephContext 中进行初始化。同时，还新建了 _admin_hook，并向 _admin_socket 对象注册了很多个命令。这里简单介绍下 register_command，该函数的作用就是把命令参数和对应的方法关联起来，例如：config show 对应 _conf->show_config()，具体映射在钩子函数对象中 CephContextHook -> call() -> m_cct->do_command()。

  _admin_socket = new AdminSocket(this);_admin_hook = new CephContextHook(this);_admin_socket->register_command("assert", "assert", _admin_hook, "");_admin_socket->register_command("abort", "abort", _admin_hook, "");_admin_socket->register_command("perfcounters_dump", "perfcounters_dump", _admin_hook, "");_admin_socket->register_command("1", "1", _admin_hook, "");_admin_socket->register_command("perf dump", "perf dump name=logger,type=CephString,req=false name=counter,type=CephString,req=false", _admin_hook, "dump perfcounters value");...

这里演示下 ceph daemon … 命令。admin_socket 线程作用就是监听这类查询命令，返回查询结果。

[root@localhost build]# ./bin/ceph daemon /tmp/ceph-asok.UU2i8r/client.admin.6783.asok help
*** DEVELOPER MODE: setting PATH, PYTHONPATH and LD_LIBRARY_PATH ***
{"config diff": "dump diff of current config and default config","config diff get": "dump diff get <field>: dump diff of current and default config setting <field>","config get": "config get <field>: get the config value","config help": "get config setting schema and descriptions","config set": "config set <field> <val> [<val> ...]: set a config variable","config show": "dump current config settings","config unset": "config unset <field>: unset a config variable","dump_mempools": "get mempool stats","get_command_descriptions": "list available commands","git_version": "get git sha1","help": "list available commands","log dump": "dump recent log entries to log file","log flush": "flush log entries to log file","log reopen": "reopen log file","objecter_requests": "show in-progress osd requests","perf dump": "dump perfcounters value","perf histogram dump": "dump perf histogram values","perf histogram schema": "dump perf histogram schema","perf reset": "perf reset <name>: perf reset all or one perfcounter name","perf schema": "dump perfcounters schema","version": "get ceph version"
}

admin_socket 线程在 init() 函数中启动。首先是创建了管道，读取端的文件描述符记录在 m_shutdown_rd_fd 中，写入端的文件描述符记录在 m_shutdown_wr_fd 中。从变量名字也可以看出，该文件描述符的作用是收取关闭信息。退出的信号会写入管道的写入端，而线程会通过多路复用接口，监听读取端，一旦发现 m_shutdown_rd_fd 中读出内容，就关闭线程。线程的入口函数为 AdminSocket::entry()，功能就是循环监听端口，执行命令。

bool AdminSocket::init(const std::string& path)
{ldout(m_cct, 5) << "init " << path << dendl;/* Set up things for the new thread *///创建管道int pipe_rd = -1, pipe_wr = -1;err = create_shutdown_pipe(&pipe_rd, &pipe_wr);...//绑定端口监听端口int sock_fd;err = bind_and_listen(path, &sock_fd);.../* Create new thread */th = make_named_thread("admin_socket", &AdminSocket::entry, this);...return true;
}void AdminSocket::entry() noexcept
{ldout(m_cct, 5) << "entry start" << dendl;while (true) {...if (fds[0].revents & POLLIN) {// Send out some datado_accept();}if (fds[1].revents & POLLIN) {// Parent wants us to shut downreturn;}}
}

common_init_finish() 中就创建了 service 和 admin_socket 线程，功能已经在上文介绍过了，可以看出，common_init_finish() 函数更侧重于一个进程的基础查询服务，包括对内的性能参数监控和对外的参数配置查询接口，它像是在 global_init() 功能之外的一层包装，global_init() 做的更多的是一个个 ceph 子系统或者子模块的基本构，类似操作系统的内核，而 commmon_init_finish() 类似内核之外的监控模块。

4. rados_tool_common()

rados_tool_common() 中封装了每个 rados 命令的具体处理方式。通过解析命令参数，来判断具体的处理方法。rados_tool_common() 中也做了一些基础通用操作，用于初始化 rados 和与集群建立连接。

rados.init_with_context() 主要工作就是根据配置文件中的参数，初始化 rados。具体调用链：rados.init_with_context() -> rados_create_with_context() -> _rados_create_with_context() -> librados::RadosClient::RadosClient()。

rados_connect() 作用是与集群建立连接，在 librados.h 中有提示注意：在调用其他相关通讯函数之前，必须调用此函数，否则会导致进程崩溃。在建立连接的过程中，还创建了6种线程：rados、ms_dispatcher、ms_local、safe_timer、fn_anonymous、fn_radosclient。后几小节将具体介绍这6个线程创建的时机与作用。

static int rados_tool_common(const std::map < std::string, std::string > &opts,std::vector<const char*> &nargs)
{...;Rados rados; IoCtx io_ctx; //新建了 Rados 和 IoCtx 对象，方便后续对 librados 调用...;//解析参数，opts 是在 main 中构造的 map，里面保留了命令中所有的有效参数i = opts.find("create");if (i != opts.end()) {create_pool = true;}i = opts.find("pool");if (i != opts.end()) {pool_name = i->second.c_str();}...;//打开 radosret = rados.init_with_context(g_ceph_context);...;ret = rados.connect();...;//创建新 poolif (create_pool) {ret = rados.pool_create(pool_name);if (ret < 0) {cerr << "error creating pool " << pool_name << ": " << cpp_strerror(ret) << std::endl;return 1;}}...//打开io 上下文ret = pool_name ? rados.ioctx_create(pool_name, io_ctx) : rados.ioctx_create2(pgid->pool(), io_ctx);//通过判断传入的第一个参数命令，来决定调用什么方法，如果第一个参数为“bench”（rados bench ...），则进入一下代码模块。else if (strcmp(nargs[0], "bench") == 0) {...;}
}

4.1 rados.connect()

先来解析下 rados_connect() 的调用链，找到具体的函数方法。rados.connect() -> client->connect() -> librados::RadosClient::connect()。实际上，connect() 进行网络通信需要做的一些预处理，尽管它的名字叫做 connnect “链接”，但是实际的网络链接是发生在后面的操作中：_op_submit() -> _get_session() -> messenger->get_connection()。

这里简单介绍下一个完整的 client 端发送消息的流程。
第一步：创建 messenger 并设置策略，Messenger::create() 和 messenger->set_default_policy()。
第二步：创建 objecter，new Object()。
第三步：设置消息分发器， messenger->add_dispatcher_head() 和 messenger->add_dispatcher_tail()。
第四步：启动消息，messenger->start()。
第五步：发送请求 objecter->op_submit()，在发送请求之前，要把数据封装成 Objecter::Op 对象。

在 connect() 方法中，完成了前四步的过程，相当于为发送消息做了预处理，后续再发送请求时只需要调用 op_submit() 即可。此外，connect() 中还做了 monclient、mgrclient 和 timer 的初始化工作。其中 monclient 的初始化函数 init() 中将 monclient 实例对象注册到 dispatcher 中心（通过 add_dispatcher_head() 方法，后文有介绍），创建路由密钥，启动定时器 timer 和开启 finisher 线程等。mgrclient 的初始化函数 init() 中做的工作就简单许多，只是启动了定时器。

int librados::RadosClient::connect()
{//获取monmap并配置{MonClient mc_bootstrap(cct);err = mc_bootstrap.get_monmap_and_config();if (err < 0)return err; }// get monmaperr = monclient.build_initial_monmap();if (err < 0)goto out;//新建 messengermessenger = Messenger::create_client_messenger(cct, "radosclient");//设置messenger策略// require OSDREPLYMUX feature.  this means we will fail to talk to// old servers.  this is necessary because otherwise we won't know// how to decompose the reply data into its constituent pieces.messenger->set_default_policy(Messenger::Policy::lossy_client(CEPH_FEATURE_OSDREPLYMUX));//创建object，用来处理发送的数据objecter = new (std::nothrow) Objecter(cct, messenger, &monclient,&finisher,cct->_conf->rados_mon_op_timeout,cct->_conf->rados_osd_op_timeout);if (!objecter)goto out;// 根据配置，设置启用 throttle 来控制发出的op数objecter->set_balanced_budget();//为 monclient 和 mgrclient 提供消息通信层实例monclient.set_messenger(messenger);mgrclient.set_messenger(messenger);//初始化objecter perfcounter 并提供能够查询正在处理的op状况的钩子objecter->init();//为消息层实例添加Dispatchermessenger->add_dispatcher_head(&mgrclient);messenger->add_dispatcher_tail(objecter);messenger->add_dispatcher_tail(this);//启动messenger实例messenger->start();ldout(cct, 1) << "setting wanted keys" << dendl;monclient.set_want_keys(CEPH_ENTITY_TYPE_MON | CEPH_ENTITY_TYPE_OSD | CEPH_ENTITY_TYPE_MGR);ldout(cct, 1) << "calling monclient init" << dendl;//初始化 monclienterr = monclient.init();if (err) {ldout(cct, 0) << conf->name << " initialization error " << cpp_strerror(-err) << dendl;shutdown();goto out;}//验证客户端的权限err = monclient.authenticate(conf->client_mount_timeout);if (err) {ldout(cct, 0) << conf->name << " authentication error " << cpp_strerror(-err) << dendl;shutdown();goto out;}messenger->set_myname(entity_name_t::CLIENT(monclient.get_global_id()));// Detect older cluster, put mgrclient into compatible modemgrclient.set_mgr_optional(!get_required_monitor_features().contains_all(ceph::features::mon::FEATURE_LUMINOUS));// MgrClient needs this (it doesn't have MonClient reference itself)monclient.sub_want("mgrmap", 0, 0);monclient.renew_subs();if (service_daemon) {ldout(cct, 10) << __func__ << " registering as " << service_name << "."<< daemon_name << dendl;mgrclient.service_daemon_register(service_name, daemon_name,daemon_metadata);}//初始化 mgrclientmgrclient.init();objecter->set_client_incarnation(0);objecter->start();lock.Lock();//初始化timertimer.init();finisher.start();//状态更新为已连接state = CONNECTED;instance_id = monclient.get_global_id();return err;
}

4.1.1 ms_dispatcher 与 ms_local

dispatcher 是消息分发中心，所有收到的消息都经由该模块，并由该模块转发给相应的处理模块（moncliet、mdsclient、osd等）。其实现方式比较简单，就是把所有的模块及其处理消息的方法 handle 注册到分发中心，具体函数为 add_dispatcher_head/tail()，这样就像 dispaatcher_queue 中添加了指定模块。后续在分发消息时，对 dispatcher_queue 进行轮询，直到有一个处理模块能够处理该消息，通过 message->get_type() 来指定消息的处理函数。所有的消息分发都在 dispatcher 线程中完成。

在 add_dispatcher_head() 和 add_dispatcher_tail() 函数中，都做了 dispatcher 队列是否为空的判断（通过 dispatchers.empty() == true）。如果判定结果为空，说明需要重新创建 dispatcher 线程并绑定服务端地址，加入事件中心监听端口，具体方法在 ready() 中。

  void add_dispatcher_head(Dispatcher *d) {bool first = dispatchers.empty();dispatchers.push_front(d);if (d->ms_can_fast_dispatch_any())fast_dispatchers.push_front(d);if (first)ready();}

这里给出了 AsyncMessenger::ready() 方法。p->start()（Processor::start()）方法中监听 EVENT_READABLE 事件并把事件提交到 EventCenter 事件中心，由上文介绍的 msgr-worker-x 线程去轮询事件中心的队列，监听端口是否收到消息。收到的消息则由 dispatcher 线程分发给指定的处理程序，其分发消息的接口为 ms_dispatch() 和 ms_fast_dispatch()。

dispatch_queue.start() 中开启了消息分发线程，分别为处理外部消息的 ms_dispatch 线程和处理本地消息的 ms_local 线程。相应的，它们有各自的优先级队列（注意：分发消息的队列时有优先级的，优先级越高，发送时机越早），分别是存储外部消息的 mqueue 和本地消息队列的 local_messages。消息队列的添加方式也有两种：mqueue.enqueue() 和 local_queue.emplace()。

void AsyncMessenger::ready()
{ldout(cct,10) << __func__ << " " << get_myaddrs() << dendl;stack->ready();//绑定端口if (pending_bind) {int err = bindv(pending_bind_addrs);if (err) {lderr(cct) << __func__ << " postponed bind failed" << dendl;ceph_abort();}}Mutex::Locker l(lock);//调用 worker 线程，监听端口for (auto &&p : processors)  p->start();//开启 ms_dispatcher 和 ms_locla 线程dispatch_queue.start();
}void DispatchQueue::start()
{ceph_assert(!stop);ceph_assert(!dispatch_thread.is_started());//开启 ms_dispatch 和 ms_local 线程dispatch_thread.create("ms_dispatch");local_delivery_thread.create("ms_local");
}

4.1.2 safe_timer

在 rados.connect() 中，monclient、mgrclient、radsoclient 都创建了 safe_time 线程。这其实一个定时器，每个模块都有自己的定时器（例如 mgrclient 在类中声明了：SafeTime timer 对象），通过调用 SafeTime::init() 方法来开启线程启动定时器模块。SafeTimer::timer_thread() 是 safe_time 线程方法，可以看到内部采用while 死循环，重复轮询事件表 schedule，检查是否到达任务的执行事件。任务在 schedule 中按照事件升序排列。首先检查，如果第一任务没有到事件，后面的任务就不用检查，直接 break。如果任务到了事件，则执行 callback 任务，并在 schedule 中删除该定时任务，然后继续循环。

添加定时任务函数：add_event_after()、add_event_at()。

取消任务：cancel_event()、cancel_all_events()。

void SafeTimer::init()
{ldout(cct,10) << "init" << dendl;//创建并开启 timer，线程名为 safe_timerthread = new SafeTimerThread(this);thread->create("safe_timer");
}//线程方法
void SafeTimer::timer_thread()
{lock.lock();ldout(cct,10) << "timer_thread starting" << dendl;while (!stopping) {utime_t now = ceph_clock_now();while (!schedule.empty()) {scheduled_map_t::iterator p = schedule.begin();// is the future now?if (p->first > now)break;Context *callback = p->second;events.erase(callback);schedule.erase(p);ldout(cct,10) << "timer_thread executing " << callback << dendl;if (!safe_callbacks)lock.unlock();callback->complete(0);if (!safe_callbacks)lock.lock();}// recheck stopping if we dropped the lockif (!safe_callbacks && stopping)break;ldout(cct,20) << "timer_thread going to sleep" << dendl;if (schedule.empty())cond.Wait(lock);elsecond.WaitUntil(lock, schedule.begin()->first);ldout(cct,20) << "timer_thread awake" << dendl;}ldout(cct,10) << "timer_thread exiting" << dendl;lock.unlock();
}

4.1.3 fn_anonymous 与 fn-radosclient

fn_anonymous 线程是在 monclient() 构造函数中创建，在 monclient.init() 初始化函数中启动的线程。fn-radosclient 线程则是 radosclient() 构造函数中创建的线程，在 rados.connect() 中启动的线程。它们都归属于 finisher 线程，只是一个属于匿名对象，一个属于命名对象。finisher 类主要用来完成回调函数 context 的执行，通过开启新线程的方式，异步处理 context 的收尾工作。

MonClient::MonClient(CephContext *cct_) :...//创建 fn_anonymous 线程对象 finisher(cct_),...
{}librados::RadosClient::RadosClient(CephContext *cct_): Dispatcher(cct_->get()),...//fn-radosclient 线程在此初始化，后续调用 finisher.start() 开启线程finisher(cct, "radosclient", "fn-radosclient")
{
}

finisher 线程的方法实体为 Finisher::finisher_thread_entry()，其主要功能为循环处理 finisher_queue 队列中的结束任务。注意：这里把 finisher_queue 中的任务提取到局部变量 ls 队列中，减少了锁的使用，提高了性能。通过调用 context 的 complete 处理方法，来执行 context 的收尾函数。可以设置多种 context 实例，并个性化它们的 finish() 函数。

添加一个 context 至完成队列：Finisher::queue()。

void *Finisher::finisher_thread_entry()
{...while (!finisher_stop) {/// Every time we are woken up, we process the queue until it is empty.while (!finisher_queue.empty()) {// To reduce lock contention, we swap out the queue to process.// This way other threads can submit new contexts to complete// while we are working.//把 finisher_queue 中的任务提取到 ls。这样就不必锁住 finisher_queue，在finish 线程处理任务的同时，其他线程可以向 finisher_queue 提交任务，提高了性能。vector<pair<Context*,int>> ls;ls.swap(finisher_queue);...// Now actually process the contexts.for (auto p : ls) {//执行 context 收尾函数p.first->complete(p.second);}ldout(cct, 10) << "finisher_thread done with " << ls << dendl;ls.clear();...}...}// If we are exiting, we signal the thread waiting in stop(),// otherwise it would never unblockfinisher_empty_cond.notify_all();...return 0;
}

4.2 radso.ioctx_create()

创建 ioctx （io 上下文）有两种方式，第一种是有 pool_name，可以使用 rados.ioctx_create(pool_name, io_ctx) 函数；第二种是有 pgid，可以使用 rados.ioctx_create2(pgid->pool(), io_ctx)。如果两个参数都不知晓，那么将无法创建 ioctx。在实际应用中，一般提前在集群中创建 pool 或者在程序中使用 rados.pool_create(pool_name) 创建 pool，来获取 pool 名称。

在 ioctx_create() 可以看到，创建 ioctx 实例，实际上是创建了 IoctxImpl 实例，并且 ioctx 类中的所有方法的具体实现也是调用了 IoctxImpl 中的函数。所以，可以这样认为 ioctx 是 IoctxImpl 的封装，放在 librados 库中，专门用于提供对外的 api 接口。

int librados::Rados::ioctx_create(const char *name, IoCtx &io)
{rados_ioctx_t p;//创建 IoctxImplint ret = rados_ioctx_create((rados_t)client, name, &p);if (ret)return ret;io.close();//把 IoctxImpl 加入 ioctxio.io_ctx_impl = (IoCtxImpl*)p;return 0;
}//librados::IoCtx::write() 调用 librados::IoCtxImpl::write()
int librados::IoCtx::write(const std::string& oid, bufferlist& bl, size_t len, uint64_t off)
{object_t obj(oid);return io_ctx_impl->write(obj, bl, len, off);
}

以下是 IoCtxImpl 的结构体，给出了部分参数。

struct librados::IoCtxImpl {std::atomic<uint64_t> ref_cnt = { 0 };RadosClient *client;int64_t poolid;snapid_t snap_seq;::SnapContext snapc;uint64_t assert_ver;version_t last_objver;uint32_t notify_timeout;object_locator_t oloc;Mutex aio_write_list_lock;ceph_tid_t aio_write_seq;Cond aio_write_cond;xlist<AioCompletionImpl*> aio_write_list;map<ceph_tid_t, std::list<AioCompletionImpl*> > aio_write_waiters;Objecter *objecter;...
}

ioctx 的功能有：读写数据、读写属性、快照 pool、读取快照等。

rados tools相关推荐

No cached version of com.android.tools.build:gradle:2.0.0 available for offline mode.
异常场景从AS2.0升级到2.1,重新编译工程后,抛出了如下异常 Error:A problem occurred configuring root project 'AndroidStudioPr ...
Error:The SDK Build Tools revision (23.0.3) is too low for project ':app'. Minimum required is 25.0.
导入github上项目的时候出现 Error:The SDK Build Tools revision (23.0.3) is too low for project ':app'. Minimum ...
MAC和windows开发操作系统环境，解决Maven工程中报 Missing artifact jdk.tools:jdk.tools
同事使用的是苹果mac,而我们其他人的开发环境是windows jdk1.8 导致同事从git上pull下来的工程,pom文件是直接报错的, windows下的pom文件设置是这样的: 1 <d ...
[转]Getting Start With Node.JS Tools For Visual Studio
本文转自:http://www.c-sharpcorner.com/UploadFile/g_arora/getting-started-with-node-js-tools-for-visual-s ...
ceph rados命令使用
文章目录 Pool相关 Object相关导出资源池数据最近了解了rados命令的使用,感觉在对象操作这块还是非常实用.因为rados是属于底层存储核心,所以关于rados的命令针对对象的操作较多. ...
安装VMWare tools，以及解决安装后/mnt中有hgfs但没共享文件的方法
一.首先是安装VMWare tools 安装过程可参考:http://www.cnblogs.com/jiu0821/p/7559949.html 二.解决安装VMWare tools后/mn ...
wxpython有没有可视化设计_wxPython - GUI Builder工具( GUI Builder Tools)
wxPython - GUI Builder工具( GUI Builder Tools) 通过手动编码创建美观的GUI可能很乏味. 可视化GUI设计器工具总是很方便. 许多针对wxPython的GUI ...
Debugging Tools for Windows__from WDK7
1. 主要要用到两个工具: (1).WinDBG 这个主要用于非IDE下调试程序/查看信息等 (2).cdb.exe 这个主要是用在 Qt5.3.2 for VS10 的单步调试器 2. WDK7 ...
前端(移动端)开发利器Chrome Developer Tools秘籍(下)
之前有分享到我们可以通过 Command Line API 来提高我们的开发效率.除此之外,还有一些比较有趣的快捷键和调试方法,也能帮助提高大家的生产效率. 几个小事项: 1.文中提到的快捷键 com ...

rados tools