/* Copyright (c) 2007-2011 iMatix Corporation Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file This file is part of 0MQ. 0MQ is free software; you can redistribute it and/or modify it under the terms of the GNU Lesser General Public License as published by the Free Software Foundation; either version 3 of the License, or (at your option) any later version. 0MQ is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more details. You should have received a copy of the GNU Lesser General Public License along with this program. If not, see . */ #include #include "../include/zmq.h" #include "io_thread.hpp" #include "platform.hpp" #include "err.hpp" #include "ctx.hpp" zmq::io_thread_t::io_thread_t (ctx_t *ctx_, uint32_t tid_) : object_t (ctx_, tid_) { poller = new (std::nothrow) poller_t; alloc_assert (poller); mailbox_handle = poller->add_fd (mailbox.get_fd (), this); poller->set_pollin (mailbox_handle); } zmq::io_thread_t::~io_thread_t () { delete poller; } void zmq::io_thread_t::start () { // Start the underlying I/O thread. poller->start (); } void zmq::io_thread_t::stop () { send_stop (); } zmq::mailbox_t *zmq::io_thread_t::get_mailbox () { return &mailbox; } int zmq::io_thread_t::get_load () { return poller->get_load (); } void zmq::io_thread_t::in_event () { // TODO: Do we want to limit number of commands I/O thread can // process in a single go? while (true) { // Get the next command. If there is none, exit. command_t cmd; int rc = mailbox.recv (&cmd, 0); if (rc != 0 && errno == EINTR) continue; if (rc != 0 && errno == EAGAIN) break; errno_assert (rc == 0); // Process the command. cmd.destination->process_command (cmd); } } void zmq::io_thread_t::out_event () { // We are never polling for POLLOUT here. This function is never called. zmq_assert (false); } void zmq::io_thread_t::timer_event (int id_) { // No timers here. This function is never called. zmq_assert (false); } zmq::poller_t *zmq::io_thread_t::get_poller () { zmq_assert (poller); return poller; } void zmq::io_thread_t::process_stop () { poller->rm_fd (mailbox_handle); poller->stop (); }