Line data    Source code

       1             : /*
       2             :     Copyright (c) 2007-2016 Contributors as noted in the AUTHORS file
       3             : 
       4             :     This file is part of libzmq, the ZeroMQ core engine in C++.
       5             : 
       6             :     libzmq is free software; you can redistribute it and/or modify it under
       7             :     the terms of the GNU Lesser General Public License (LGPL) as published
       8             :     by the Free Software Foundation; either version 3 of the License, or
       9             :     (at your option) any later version.
      10             : 
      11             :     As a special exception, the Contributors give you permission to link
      12             :     this library with independent modules to produce an executable,
      13             :     regardless of the license terms of these independent modules, and to
      14             :     copy and distribute the resulting executable under terms of your choice,
      15             :     provided that you also meet, for each linked independent module, the
      16             :     terms and conditions of the license of that module. An independent
      17             :     module is a module which is not derived from or based on this library.
      18             :     If you modify this library, you must extend this exception to your
      19             :     version of the library.
      20             : 
      21             :     libzmq is distributed in the hope that it will be useful, but WITHOUT
      22             :     ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
      23             :     FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
      24             :     License for more details.
      25             : 
      26             :     You should have received a copy of the GNU Lesser General Public License
      27             :     along with this program.  If not, see <http://www.gnu.org/licenses/>.
      28             : */
      29             : 
      30             : #include "precompiled.hpp"
      31             : #include <new>
      32             : 
      33             : #include "macros.hpp"
      34             : #include "io_thread.hpp"
      35             : #include "platform.hpp"
      36             : #include "err.hpp"
      37             : #include "ctx.hpp"
      38             : 
      39         603 : zmq::io_thread_t::io_thread_t (ctx_t *ctx_, uint32_t tid_) :
      40        1206 :     object_t (ctx_, tid_)
      41             : {
      42         603 :     poller = new (std::nothrow) poller_t (*ctx_);
      43         603 :     alloc_assert (poller);
      44             : 
      45         603 :     mailbox_handle = poller->add_fd (mailbox.get_fd (), this);
      46         603 :     poller->set_pollin (mailbox_handle);
      47         603 : }
      48             : 
      49        2412 : zmq::io_thread_t::~io_thread_t ()
      50             : {
      51         603 :     LIBZMQ_DELETE(poller);
      52        1206 : }
      53             : 
      54         603 : void zmq::io_thread_t::start ()
      55             : {
      56             :     //  Start the underlying I/O thread.
      57         603 :     poller->start ();
      58         603 : }
      59             : 
      60         603 : void zmq::io_thread_t::stop ()
      61             : {
      62         603 :     send_stop ();
      63         603 : }
      64             : 
      65         603 : zmq::mailbox_t *zmq::io_thread_t::get_mailbox ()
      66             : {
      67         603 :     return &mailbox;
      68             : }
      69             : 
      70       65047 : int zmq::io_thread_t::get_load ()
      71             : {
      72       65047 :     return poller->get_load ();
      73             : }
      74             : 
      75       11312 : void zmq::io_thread_t::in_event ()
      76             : {
      77             :     //  TODO: Do we want to limit number of commands I/O thread can
      78             :     //  process in a single go?
      79             : 
      80             :     command_t cmd;
      81       11312 :     int rc = mailbox.recv (&cmd, 0);
      82             : 
      83       87850 :     while (rc == 0 || errno == EINTR) {
      84       76536 :         if (rc == 0)
      85       76536 :             cmd.destination->process_command (cmd);
      86       76538 :         rc = mailbox.recv (&cmd, 0);
      87             :     }
      88             : 
      89       11314 :     errno_assert (rc != 0 && errno == EAGAIN);
      90       11314 : }
      91             : 
      92           0 : void zmq::io_thread_t::out_event ()
      93             : {
      94             :     //  We are never polling for POLLOUT here. This function is never called.
      95           0 :     zmq_assert (false);
      96           0 : }
      97             : 
      98           0 : void zmq::io_thread_t::timer_event (int)
      99             : {
     100             :     //  No timers here. This function is never called.
     101           0 :     zmq_assert (false);
     102           0 : }
     103             : 
     104       18308 : zmq::poller_t *zmq::io_thread_t::get_poller ()
     105             : {
     106       18308 :     zmq_assert (poller);
     107       18308 :     return poller;
     108             : }
     109             : 
     110         603 : void zmq::io_thread_t::process_stop ()
     111             : {
     112         603 :     poller->rm_fd (mailbox_handle);
     113         603 :     poller->stop ();
     114         603 : }