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

作者：[e-mail]

博客：blog.focus-linux.net     linuxfocus.blog.chinaunix.net

书接上回，继续的zmq_init的源码分析。上次学习到了zmq::reaper_t::reaper_t中。

zmq::reaper_t::reaper_t (class ctx_t *ctx_, uint32_t tid_) :
    object_t (ctx_, tid_),
    sockets (0),
    terminating (false)
{
    poller = new (std::nothrow) poller_t;
    alloc_assert (poller);

     

     /* 

     上次说到mailbox.get_fd()返回的是一个unix域的读取描述符 

     */

    mailbox_handle = poller->add_fd (mailbox.get_fd (), this);
    poller->set_pollin (mailbox_handle);
}

进入poller->add_fd

zmq::epoll_t::handle_t zmq::epoll_t::add_fd (fd_t fd_, i_poll_events *events_)
{
    /* 一个fd需要一个poll_entry_t */
    poll_entry_t *pe = new (std::nothrow) poll_entry_t;
    alloc_assert (pe);

     /* 

     如注释所说，这个memset为了防止某些编译器报warning

     平时我也遇到过同样的问题。虽然并没有真正使用未初始化的变量，但是某些编译器确实会报警告。

     */

    //  The memset is not actually needed. It's here to prevent debugging
    //  tools to complain about using uninitialised memory.
    memset (pe, 0, sizeof (poll_entry_t));

    pe->fd = fd_;
    pe->ev.events = 0;
    pe->ev.data.ptr = pe;
    pe->events = events_;

     /* 将fd_加入到epoll_fd对应的epoll的instance */

    int rc = epoll_ctl (epoll_fd, EPOLL_CTL_ADD, fd_, &pe->ev);
    errno_assert (rc != -1);

     /* 增加poller的负载计数 */

    //  Increase the load metric of the thread.
    adjust_load (1);

    return pe;
}

进入poller->set_pollin

void zmq::epoll_t::set_pollin (handle_t handle_)
{
    poll_entry_t *pe = (poll_entry_t*) handle_;    /* 告诉epoll_fd对应的epoll实例，pe->fd已经处于可读取状态 */
    pe->ev.events |= EPOLLIN;
    int rc = epoll_ctl (epoll_fd, EPOLL_CTL_MOD, pe->fd, &pe->ev);
    errno_assert (rc != -1);
}

接下来，回到ctx_t的构造函数：

zmq::ctx_t::ctx_t (uint32_t io_threads_) :
    tag (0xbadcafe0),
    terminating (false)
{
    int rc;

    ......  ......

    //  Create the reaper thread.
    reaper = new (std::nothrow) reaper_t (this, reaper_tid);
    alloc_assert (reaper);
    /* 保存reaper的mailbox，用于以后通信 */
    slots [reaper_tid] = reaper->get_mailbox ();
    /* 启动reaper的工作线程*/
    reaper->start ();

     

     ......  ......

}

reaper->start ==> zmq::reaper_t::start ==> zmq::epoll_t::start ==> zmq::thread_t::start

那么现在我们看看zmq::thread_t::start的实现：

extern "C"
{
    static void *thread_routine (void *arg_)
    {
#if !defined ZMQ_HAVE_OPENVMS
        //  Following code will guarantee more predictable latecnies as it'll
        //  disallow any signal handling in the I/O thread.
        sigset_t signal_set;
        int rc = sigfillset (&signal_set);
        errno_assert (rc == 0);
#   if !defined ZMQ_HAVE_ANDROID
        rc = pthread_sigmask (SIG_BLOCK, &signal_set, NULL);
        posix_assert (rc);
#   endif
#endif

 /* 本函数只是一个简单包装，这里直接调用thread_t的tfn，即调用者真正想启动的函数 */

        zmq::thread_t *self = (zmq::thread_t*) arg_;
        self->tfn (self->arg);
        return NULL;
    }
}

void zmq::thread_t::start (thread_fn *tfn_, void *arg_)
{
    /* 保存工作函数和参数 */
    tfn = tfn_;
    arg =arg_;

     /* 

     创建线程，但是这里线程的入口函数并不是参数tfn_,为什么呢？

     为什么不直接pthread_create(&descriptor, NULL, tfn_, arg)呢

     */

    int rc = pthread_create (&descriptor, NULL, thread_routine, this);
    posix_assert (rc);
}

今天遗留下一个问题，就是上面所说的，为什么要多一层thread_routine的封装呢？