zeromq源码学习(4)——zmq_init源码分析(4)

7485阅读　2评论2011-12-13　GFree_Wind
分类：LINUX

作者：gfree.wind@gmail.com
博客：blog.focus-linux.net   linuxfocus.blog.chinaunix.net 
 
本文的copyleft归gfree.wind@gmail.com所有，使用GPL发布，可以自由拷贝，转载。但转载请保持文档的完整性，注明原作者及原链接，严禁用于任何商业用途。
======================================================================================================

有一段时间没有更新了，开始看《深入理解Linux内核架构》，感觉写得很不错。不过暂时还是在学习阶段，所以没有什么可写的。为了保持自己学习的动力，今天继续学习一下zeromq吧。

上次看到了zmq_init中的reaper->start ();以及其对应的工作线程，今天就继续后面的代码。

  1. zmq::ctx_t::ctx_t (uint32_t io_threads_) :
  2.     tag (0xbadcafe0),
  3.     terminating (false)
  4. {
     。。。。。。
      
  1.     // Create I/O thread objects and launch them.
  2.     /*
  3.     为什么i从2开始呢？这里的i是为slot的索引，在前面的代码中slot[0]为zmq_term对应的mailbox，即&term_     mailbox ，slot[1]为reaper线程对应的mailbox。所以I/O线程的mailbox的索引i只能从2开始。
    
     这里创建并启动了I/O线程，并加入到slots中。
  1.     */
  2.     for (uint32_t i = 2; i != io_threads_ + 2; i++) {
  3.         io_thread_t *io_thread = new (std::nothrow) io_thread_t (this, i);
  4.         alloc_assert (io_thread);
  5.         io_threads.push_back (io_thread);
  6.         slots [i] = io_thread->get_mailbox ();
  7.         io_thread->start ();
  8.     }

     //  In the unused part of the slot array, create a list of empty slots.
     /* 如注释所言，保存空的slot 索引*/
     for (int32_t i = (int32_t) slot_count - 1;
          i >= (int32_t) io_threads_ + 2; i--) {
         empty_slots.push_back (i);
         slots [i] = NULL;
     }

     //  Create the logging infrastructure.
     log_socket = create_socket (ZMQ_PUB);
     zmq_assert (log_socket);
     rc = log_socket->bind ("sys://log");
     zmq_assert (rc == 0);
  1. }

先看一下I/O thread的工作线程：
  1. void zmq::epoll_t::loop ()
  2. {
  3.     epoll_event ev_buf [max_io_events];

  5.     while (!stopping) {

  7.         // Execute any due timers.
  8.         int timeout = (int) execute_timers ();

  10.         // Wait for events.
  11.         int n = epoll_wait (epoll_fd, &ev_buf [0], max_io_events,
  12.             timeout ? timeout : -1);
  13.         if (n == -1 && errno == EINTR)
  14.             continue;
  15.         errno_assert (n != -1);

  17.         for (int i = 0; i < n; i ++) {
  18.             poll_entry_t *pe = ((poll_entry_t*) ev_buf [i].data.ptr);

  20.             if (pe->fd == retired_fd)
  21.                 continue;
  22.             if (ev_buf [i].events & (EPOLLERR | EPOLLHUP))
  23.                 pe->events->in_event ();
  24.             if (pe->fd == retired_fd)
  25.                continue;
  26.             if (ev_buf [i].events & EPOLLOUT)
  27.                 pe->events->out_event ();
  28.             if (pe->fd == retired_fd)
  29.                 continue;
  30.             if (ev_buf [i].events & EPOLLIN)
  31.                 pe->events->in_event ();
  32.         }

  34.         // Destroy retired event sources.
  35.         for (retired_t::iterator it = retired.begin (); it != retired.end ();
  36.               ++it)
  37.             delete *it;
  38.         retired.clear ();
  39.     }
  40. }
这个工作流程和机制和reaper线程的工作函数zmq::select_t::loop很相似。
1. 先执行到期的timer，并计算出合适的timeout；
2. 使用epoll监视事件，并根据事件类型执行不同的操作；
3. 销毁过期的poller对象；

再看一下bind的代码，这个zmq_init基本上就结束了。
  1. int zmq::socket_base_t::bind (const char *addr_)
  2. {
  3.     if (unlikely (ctx_terminated)) {
  4.         errno = ETERM;
  5.         return -1;
  6.     }

  8.     // Parse addr_ string.
  9.     std::string protocol;
  10.     std::string address;
  11.     /*
  12.     按照zeromq的格式即protocol://address解析uri
  13.     得到protocol和address
  14.     */
  15.     int rc = parse_uri (addr_, protocol, address);
  16.     if (rc != 0)
  17.         return -1;
     // 检测是否为zeromq支持的协议类型
  1.     rc = check_protocol (protocol);
  2.     if (rc != 0)
  3.         return -1;

  5.     if (protocol == "inproc" || protocol == "sys") {
  6.         /* 
  7.         为本机的一个进程间的通信，即线程间通信.
  8.         因为使用内存间的直接通信，所以无需使用I/O thread
  9.         */
  10.         endpoint_t endpoint = {this, options};
  11.         return register_endpoint (addr_, endpoint);
  12.     }

  14.     if (protocol == "tcp" || protocol == "ipc") {

  16.         // Choose I/O thread to run the listerner in.
  17.         /* 根据亲和性以及线程的负载load挑选一个合适的I/O线程 */
  18.         io_thread_t *io_thread = choose_io_thread (options.affinity);
  19.         if (!io_thread) {
  20.             errno = EMTHREAD;
  21.             return -1;
  22.         }

  24.         // Create and run the listener.
  25.         /* 根据协议和地址，创建一个listener，并将其bind到一个I/O线程。*/
  26.         zmq_listener_t *listener = new (std::nothrow) zmq_listener_t (
  27.             io_thread, this, options);
  28.         alloc_assert (listener);
  29.         int rc = listener->set_address (protocol.c_str(), address.c_str ());
  30.         if (rc != 0) {
  31.             delete listener;
  32.             return -1;
  33.         }
  34.         launch_child (listener);

  36.         return 0;
  37.     }

  39.     if (protocol == "pgm" || protocol == "epgm") {

  41.         // For convenience's sake, bind can be used interchageable with
  42.         // connect for PGM and EPGM transports.
  43.         return connect (addr_);
  44.     }

  46.     zmq_assert (false);
  47.     return -1;
  48. }

看到bind的实现，总算到了zeromq有点意思的地方了。今天对于bind看得很粗略，下次会仔细的学习一下bind的代码——特别是protocol==tcp和ipc

上一篇：提高程序并行效率的一点经验
下一篇：netfilter源码学习(1)——框架hook处理(1)

文章评论