Ceph网络模块代码的示例分析

74次阅读

共计 2739 个字符，预计需要花费 7 分钟才能阅读完成。

这篇文章主要为大家展示了“Ceph 网络模块代码的示例分析”，内容简而易懂，条理清晰，希望能够帮助大家解决疑惑，下面让丸趣 TV 小编带领大家一起研究并学习一下“Ceph 网络模块代码的示例分析”这篇文章吧。

1. 创建消息管理类 Messenger.

Messenger *ms_public = Messenger::create(g_ceph_context,
 entity_name_t::OSD(whoami),  client ,
 getpid());

在这里函数这里创建了若干个 Messenger 实例,ms_public 实例只是其中的一个. 从这个函数参数 client 也能看出一些端倪, 说明这个 Messenger 是专门用于
处理客户端的消息的, 这里的客户端指的是诸如 RGW, RBD 这样的客户端. 而后面的 ms_cluster 则处理与其它 OSD 的消息。

2. 绑定端口和地址

然后在后面, 就像 Linux Socket 编程一样, 在这里也需要调用 bind() 函数完成端口以及地址的绑定, 只不过这里 bind() 函数是经过特殊封装后的.

ms_public- bind(g_conf- public_addr);

Messenger 是一个抽象类, ms_public 是一个基类指针.
而类 SimpleMessenger 继承自 Messenger.

最终调用的 bind() 函数是子类 SimpleMessenger::bind(const entity_addr_t bind_addr)

int SimpleMessenger::bind(const entity_addr_t  bind_addr)
 lock.Lock();
 if (started) { ldout(cct,10)    rank.bind already started    dendl;
 lock.Unlock();
 return -1; 
 }
 ldout(cct,10)    rank.bind     bind_addr   dendl;
 lock.Unlock();
 // bind to a socket
 set int  avoid_ports;
 int r = accepter.bind(bind_addr, avoid_ports);
 if (r  = 0)
 did_bind = true;
 return r;
}

这里的 entity_addr_t 实际是 socket 通信中需要用到的 sockaddr_in,AF_INET/AF_INET6 的集合.
继续调用成员 accepter.bind(bind_addr, avoid_ports);
在 msg/Accepter.cc 中可以看到:

int Accepter::bind(const entity_addr_t  bind_addr, const set int  avoid_ports)

这个函数执行了:

1) 创建一个 socket fd.

2) 执行 bind()

3) 执行 listen()
以上都是 socket 编程中基本操作步骤.

3. 启动服务端

ms_public- start();

实际上调用的是:

int SimpleMessenger::start()
 lock.Lock();
 ldout(cct,1)    messenger.start    dendl;
 // register at least one entity, first!
 assert(my_type  = 0);
 assert(!started);
 started = true;
 if (!did_bind) {
 my_inst.addr.nonce = nonce;
 init_local_connection();
 }
 lock.Unlock();
 reaper_started = true;
 reaper_thread.create();
 return 0;
}

从这里可以看出,reaper_thread.create() 创建了一个 reaper 线程. 找到它的 void *entry() 函数.//SimpleMessenger.h

reaper 线程实际上执行的是 void SimpleMessenger::reaper_entry(), 它最终执行 / 调用的是:

void SimpleMessenger::reaper()
 ldout(cct,10)    reaper    dendl;
 assert(lock.is_locked());
 while (!pipe_reap_queue.empty()) { Pipe *p = pipe_reap_queue.front();
 pipe_reap_queue.pop_front();
 ldout(cct,10)    reaper reaping pipe     p         p- get_peer_addr()   dendl;
 p- pipe_lock.Lock();
 p- discard_out_queue();
 if (p- connection_state) { // mark_down, mark_down_all, or fault() should have done this,
 // or accept() may have switch the Connection to a different
 // Pipe... but make sure!
 bool cleared = p- connection_state- clear_pipe(p);
 assert(!cleared);
 }
 p- pipe_lock.Unlock();
 p- unregister_pipe();
 assert(pipes.count(p));
 pipes.erase(p);
 p- join();
 if (p- sd  = 0)
 ::close(p- 
 ldout(cct,10)    reaper reaped pipe     p         p- get_peer_addr()   dendl;
 p- put();
 ldout(cct,10)    reaper deleted pipe     p   dendl;
 }
 ldout(cct,10)    reaper done    dendl;
}

从这里可以看出做出了清理 Pipe 的操作. (?)

4. 最后一步: