From a15854bd92db69fcd0b4444fe1b8fe3610a7acf6 Mon Sep 17 00:00:00 2001 From: Martin Lucina Date: Mon, 23 Jan 2012 08:53:19 +0100 Subject: Imported Upstream version 2.0.7.dfsg --- src/poll.cpp | 207 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 207 insertions(+) create mode 100644 src/poll.cpp (limited to 'src/poll.cpp') diff --git a/src/poll.cpp b/src/poll.cpp new file mode 100644 index 0000000..4214195 --- /dev/null +++ b/src/poll.cpp @@ -0,0 +1,207 @@ +/* + Copyright (c) 2007-2010 iMatix Corporation + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the Lesser GNU 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 + Lesser GNU General Public License for more details. + + You should have received a copy of the Lesser GNU General Public License + along with this program. If not, see . +*/ + +#include "platform.hpp" + +#if defined ZMQ_HAVE_LINUX || defined ZMQ_HAVE_FREEBSD ||\ + defined ZMQ_HAVE_OPENBSD || defined ZMQ_HAVE_SOLARIS ||\ + defined ZMQ_HAVE_OSX || defined ZMQ_HAVE_QNXNTO ||\ + defined ZMQ_HAVE_HPUX || defined ZMQ_HAVE_AIX ||\ + defined ZMQ_HAVE_NETBSD + +#include +#include +#include +#include +#include + +#include "poll.hpp" +#include "err.hpp" +#include "config.hpp" +#include "i_poll_events.hpp" + +zmq::poll_t::poll_t () : + retired (false), + stopping (false) +{ + // Get the limit on open file descriptors. Resize fds so that it + // can hold all descriptors. + rlimit rl; + int rc = getrlimit (RLIMIT_NOFILE, &rl); + errno_assert (rc != -1); + fd_table.resize (rl.rlim_cur); + + for (rlim_t i = 0; i < rl.rlim_cur; i ++) + fd_table [i].index = retired_fd; +} + +zmq::poll_t::~poll_t () +{ + worker.stop (); + + // Make sure there are no fds registered on shutdown. + zmq_assert (load.get () == 0); +} + +zmq::poll_t::handle_t zmq::poll_t::add_fd (fd_t fd_, i_poll_events *events_) +{ + pollfd pfd = {fd_, 0, 0}; + pollset.push_back (pfd); + assert (fd_table [fd_].index == retired_fd); + + fd_table [fd_].index = pollset.size() - 1; + fd_table [fd_].events = events_; + + // Increase the load metric of the thread. + load.add (1); + + return fd_; +} + +void zmq::poll_t::rm_fd (handle_t handle_) +{ + fd_t index = fd_table [handle_].index; + assert (index != retired_fd); + + // Mark the fd as unused. + pollset [index].fd = retired_fd; + fd_table [handle_].index = retired_fd; + retired = true; + + // Decrease the load metric of the thread. + load.sub (1); +} + +void zmq::poll_t::set_pollin (handle_t handle_) +{ + int index = fd_table [handle_].index; + pollset [index].events |= POLLIN; +} + +void zmq::poll_t::reset_pollin (handle_t handle_) +{ + int index = fd_table [handle_].index; + pollset [index].events &= ~((short) POLLIN); +} + +void zmq::poll_t::set_pollout (handle_t handle_) +{ + int index = fd_table [handle_].index; + pollset [index].events |= POLLOUT; +} + +void zmq::poll_t::reset_pollout (handle_t handle_) +{ + int index = fd_table [handle_].index; + pollset [index].events &= ~((short) POLLOUT); +} + +void zmq::poll_t::add_timer (i_poll_events *events_) +{ + timers.push_back (events_); +} + +void zmq::poll_t::cancel_timer (i_poll_events *events_) +{ + timers_t::iterator it = std::find (timers.begin (), timers.end (), events_); + if (it != timers.end ()) + timers.erase (it); +} + +int zmq::poll_t::get_load () +{ + return load.get (); +} + +void zmq::poll_t::start () +{ + worker.start (worker_routine, this); +} + +void zmq::poll_t::stop () +{ + stopping = true; +} + +void zmq::poll_t::loop () +{ + while (!stopping) { + + // Wait for events. + int rc = poll (&pollset [0], pollset.size (), + timers.empty () ? -1 : max_timer_period); + if (rc == -1 && errno == EINTR) + continue; + errno_assert (rc != -1); + + // Handle timer. + if (!rc) { + + // Use local list of timers as timer handlers may fill new timers + // into the original array. + timers_t t; + std::swap (timers, t); + + // Trigger all the timers. + for (timers_t::iterator it = t.begin (); it != t.end (); it ++) + (*it)->timer_event (); + + continue; + } + + for (pollset_t::iterator it = pollset.begin (); + it != pollset.end (); it ++) { + + zmq_assert (!(it->revents & POLLNVAL)); + if (it->fd == retired_fd) + continue; + if (it->revents & (POLLERR | POLLHUP)) + fd_table [it->fd].events->in_event (); + if (it->fd == retired_fd) + continue; + if (it->revents & POLLOUT) + fd_table [it->fd].events->out_event (); + if (it->fd == retired_fd) + continue; + if (it->revents & POLLIN) + fd_table [it->fd].events->in_event (); + } + + // Clean up the pollset and update the fd_table accordingly. + if (retired) { + pollset_t::size_type i = 0; + while (i < pollset.size ()) { + if (pollset [i].fd == retired_fd) + pollset.erase (pollset.begin () + i); + else { + fd_table [pollset [i].fd].index = i; + i ++; + } + } + retired = false; + } + } +} + +void zmq::poll_t::worker_routine (void *arg_) +{ + ((poll_t*) arg_)->loop (); +} + +#endif -- cgit v1.2.3