/* 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" #ifdef ZMQ_HAVE_LINUX #include #include #include #include #include #include #include "epoll.hpp" #include "err.hpp" #include "config.hpp" #include "i_poll_events.hpp" zmq::epoll_t::epoll_t () : stopping (false) { epoll_fd = epoll_create (1); errno_assert (epoll_fd != -1); } zmq::epoll_t::~epoll_t () { // Wait till the worker thread exits. worker.stop (); // Make sure there are no fds registered on shutdown. zmq_assert (load.get () == 0); close (epoll_fd); for (retired_t::iterator it = retired.begin (); it != retired.end (); it ++) delete *it; } zmq::epoll_t::handle_t zmq::epoll_t::add_fd (fd_t fd_, i_poll_events *events_) { poll_entry_t *pe = new (std::nothrow) poll_entry_t; zmq_assert (pe != NULL); // 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_; int rc = epoll_ctl (epoll_fd, EPOLL_CTL_ADD, fd_, &pe->ev); errno_assert (rc != -1); // Increase the load metric of the thread. load.add (1); return pe; } void zmq::epoll_t::rm_fd (handle_t handle_) { poll_entry_t *pe = (poll_entry_t*) handle_; int rc = epoll_ctl (epoll_fd, EPOLL_CTL_DEL, pe->fd, &pe->ev); errno_assert (rc != -1); pe->fd = retired_fd; retired.push_back (pe); // Decrease the load metric of the thread. load.sub (1); } void zmq::epoll_t::set_pollin (handle_t handle_) { poll_entry_t *pe = (poll_entry_t*) handle_; pe->ev.events |= EPOLLIN; int rc = epoll_ctl (epoll_fd, EPOLL_CTL_MOD, pe->fd, &pe->ev); errno_assert (rc != -1); } void zmq::epoll_t::reset_pollin (handle_t handle_) { poll_entry_t *pe = (poll_entry_t*) handle_; pe->ev.events &= ~((short) EPOLLIN); int rc = epoll_ctl (epoll_fd, EPOLL_CTL_MOD, pe->fd, &pe->ev); errno_assert (rc != -1); } void zmq::epoll_t::set_pollout (handle_t handle_) { poll_entry_t *pe = (poll_entry_t*) handle_; pe->ev.events |= EPOLLOUT; int rc = epoll_ctl (epoll_fd, EPOLL_CTL_MOD, pe->fd, &pe->ev); errno_assert (rc != -1); } void zmq::epoll_t::reset_pollout (handle_t handle_) { poll_entry_t *pe = (poll_entry_t*) handle_; pe->ev.events &= ~((short) EPOLLOUT); int rc = epoll_ctl (epoll_fd, EPOLL_CTL_MOD, pe->fd, &pe->ev); errno_assert (rc != -1); } void zmq::epoll_t::add_timer (i_poll_events *events_) { timers.push_back (events_); } void zmq::epoll_t::cancel_timer (i_poll_events *events_) { timers_t::iterator it = std::find (timers.begin (), timers.end (), events_); if (it == timers.end ()) return; timers.erase (it); } int zmq::epoll_t::get_load () { return load.get (); } void zmq::epoll_t::start () { worker.start (worker_routine, this); } void zmq::epoll_t::stop () { stopping = true; } void zmq::epoll_t::loop () { epoll_event ev_buf [max_io_events]; while (!stopping) { // Wait for events. int n; while (true) { n = epoll_wait (epoll_fd, &ev_buf [0], max_io_events, timers.empty () ? -1 : max_timer_period); if (!(n == -1 && errno == EINTR)) { errno_assert (n != -1); break; } } // Handle timer. if (!n) { // 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 (int i = 0; i < n; i ++) { poll_entry_t *pe = ((poll_entry_t*) ev_buf [i].data.ptr); if (pe->fd == retired_fd) continue; if (ev_buf [i].events & (EPOLLERR | EPOLLHUP)) pe->events->in_event (); if (pe->fd == retired_fd) continue; if (ev_buf [i].events & EPOLLOUT) pe->events->out_event (); if (pe->fd == retired_fd) continue; if (ev_buf [i].events & EPOLLIN) pe->events->in_event (); } // Destroy retired event sources. for (retired_t::iterator it = retired.begin (); it != retired.end (); it ++) delete *it; retired.clear (); } } void zmq::epoll_t::worker_routine (void *arg_) { ((epoll_t*) arg_)->loop (); } #endif