Ceph網絡模塊代碼的示例分析

這篇文章主要為大家展示了“Ceph網絡模塊代碼的示例分析”，內容簡而易懂，條理清晰，希望能夠幫助大家解決疑惑，下面讓小編帶領大家一起研究并學習一下“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->sd);
    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. 最后一步:

delete ms_public;

通過以上4步, OSD(服務端)關于網絡方面的工作算是初始化完畢了.工作只是準備階段,并不復雜.

以上是“Ceph網絡模塊代碼的示例分析”這篇文章的所有內容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內容對大家有所幫助，如果還想學習更多知識，歡迎關注億速云行業資訊頻道！