![]() |
libzmq master
The Intelligent Transport Layer
|
00001 /* 00002 Copyright (c) 2009-2011 250bpm s.r.o. 00003 Copyright (c) 2007-2009 iMatix Corporation 00004 Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file 00005 00006 This file is part of 0MQ. 00007 00008 0MQ is free software; you can redistribute it and/or modify it under 00009 the terms of the GNU Lesser General Public License as published by 00010 the Free Software Foundation; either version 3 of the License, or 00011 (at your option) any later version. 00012 00013 0MQ is distributed in the hope that it will be useful, 00014 but WITHOUT ANY WARRANTY; without even the implied warranty of 00015 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 00016 GNU Lesser General Public License for more details. 00017 00018 You should have received a copy of the GNU Lesser General Public License 00019 along with this program. If not, see <http://www.gnu.org/licenses/>. 00020 */ 00021 00022 #include "select.hpp" 00023 #if defined ZMQ_USE_SELECT 00024 00025 #include "platform.hpp" 00026 #if defined ZMQ_HAVE_WINDOWS 00027 #include "windows.hpp" 00028 #elif defined ZMQ_HAVE_HPUX 00029 #include <sys/param.h> 00030 #include <sys/types.h> 00031 #include <sys/time.h> 00032 #elif defined ZMQ_HAVE_OPENVMS 00033 #include <sys/types.h> 00034 #include <sys/time.h> 00035 #else 00036 #include <sys/select.h> 00037 #endif 00038 00039 #include <string.h> 00040 #include <algorithm> 00041 00042 #include "err.hpp" 00043 #include "config.hpp" 00044 #include "i_poll_events.hpp" 00045 00046 zmq::select_t::select_t () : 00047 maxfd (retired_fd), 00048 retired (false), 00049 stopping (false) 00050 { 00051 // Clear file descriptor sets. 00052 FD_ZERO (&source_set_in); 00053 FD_ZERO (&source_set_out); 00054 FD_ZERO (&source_set_err); 00055 } 00056 00057 zmq::select_t::~select_t () 00058 { 00059 worker.stop (); 00060 } 00061 00062 zmq::select_t::handle_t zmq::select_t::add_fd (fd_t fd_, i_poll_events *events_) 00063 { 00064 // Store the file descriptor. 00065 fd_entry_t entry = {fd_, events_}; 00066 fds.push_back (entry); 00067 00068 // Ensure we do not attempt to select () on more than FD_SETSIZE 00069 // file descriptors. 00070 zmq_assert (fds.size () <= FD_SETSIZE); 00071 00072 // Start polling on errors. 00073 FD_SET (fd_, &source_set_err); 00074 00075 // Adjust maxfd if necessary. 00076 if (fd_ > maxfd) 00077 maxfd = fd_; 00078 00079 // Increase the load metric of the thread. 00080 adjust_load (1); 00081 00082 return fd_; 00083 } 00084 00085 void zmq::select_t::rm_fd (handle_t handle_) 00086 { 00087 // Mark the descriptor as retired. 00088 fd_set_t::iterator it; 00089 for (it = fds.begin (); it != fds.end (); ++it) 00090 if (it->fd == handle_) 00091 break; 00092 zmq_assert (it != fds.end ()); 00093 it->fd = retired_fd; 00094 retired = true; 00095 00096 // Stop polling on the descriptor. 00097 FD_CLR (handle_, &source_set_in); 00098 FD_CLR (handle_, &source_set_out); 00099 FD_CLR (handle_, &source_set_err); 00100 00101 // Discard all events generated on this file descriptor. 00102 FD_CLR (handle_, &readfds); 00103 FD_CLR (handle_, &writefds); 00104 FD_CLR (handle_, &exceptfds); 00105 00106 // Adjust the maxfd attribute if we have removed the 00107 // highest-numbered file descriptor. 00108 if (handle_ == maxfd) { 00109 maxfd = retired_fd; 00110 for (fd_set_t::iterator it = fds.begin (); it != fds.end (); ++it) 00111 if (it->fd > maxfd) 00112 maxfd = it->fd; 00113 } 00114 00115 // Decrease the load metric of the thread. 00116 adjust_load (-1); 00117 } 00118 00119 void zmq::select_t::set_pollin (handle_t handle_) 00120 { 00121 FD_SET (handle_, &source_set_in); 00122 } 00123 00124 void zmq::select_t::reset_pollin (handle_t handle_) 00125 { 00126 FD_CLR (handle_, &source_set_in); 00127 } 00128 00129 void zmq::select_t::set_pollout (handle_t handle_) 00130 { 00131 FD_SET (handle_, &source_set_out); 00132 } 00133 00134 void zmq::select_t::reset_pollout (handle_t handle_) 00135 { 00136 FD_CLR (handle_, &source_set_out); 00137 } 00138 00139 void zmq::select_t::start () 00140 { 00141 worker.start (worker_routine, this); 00142 } 00143 00144 void zmq::select_t::stop () 00145 { 00146 stopping = true; 00147 } 00148 00149 void zmq::select_t::loop () 00150 { 00151 while (!stopping) { 00152 00153 // Execute any due timers. 00154 int timeout = (int) execute_timers (); 00155 00156 // Intialise the pollsets. 00157 memcpy (&readfds, &source_set_in, sizeof source_set_in); 00158 memcpy (&writefds, &source_set_out, sizeof source_set_out); 00159 memcpy (&exceptfds, &source_set_err, sizeof source_set_err); 00160 00161 // Wait for events. 00162 struct timeval tv = {(long) (timeout / 1000), 00163 (long) (timeout % 1000 * 1000)}; 00164 #ifdef ZMQ_HAVE_WINDOWS 00165 int rc = select (0, &readfds, &writefds, &exceptfds, 00166 timeout ? &tv : NULL); 00167 wsa_assert (rc != SOCKET_ERROR); 00168 #else 00169 int rc = select (maxfd + 1, &readfds, &writefds, &exceptfds, 00170 timeout ? &tv : NULL); 00171 if (rc == -1 && errno == EINTR) 00172 continue; 00173 errno_assert (rc != -1); 00174 #endif 00175 00176 // If there are no events (i.e. it's a timeout) there's no point 00177 // in checking the pollset. 00178 if (rc == 0) 00179 continue; 00180 00181 for (fd_set_t::size_type i = 0; i < fds.size (); i ++) { 00182 if (fds [i].fd == retired_fd) 00183 continue; 00184 if (FD_ISSET (fds [i].fd, &exceptfds)) 00185 fds [i].events->in_event (); 00186 if (fds [i].fd == retired_fd) 00187 continue; 00188 if (FD_ISSET (fds [i].fd, &writefds)) 00189 fds [i].events->out_event (); 00190 if (fds [i].fd == retired_fd) 00191 continue; 00192 if (FD_ISSET (fds [i].fd, &readfds)) 00193 fds [i].events->in_event (); 00194 } 00195 00196 // Destroy retired event sources. 00197 if (retired) { 00198 fds.erase (std::remove_if (fds.begin (), fds.end (), 00199 zmq::select_t::is_retired_fd), fds.end ()); 00200 retired = false; 00201 } 00202 } 00203 } 00204 00205 void zmq::select_t::worker_routine (void *arg_) 00206 { 00207 ((select_t*) arg_)->loop (); 00208 } 00209 00210 bool zmq::select_t::is_retired_fd (const fd_entry_t &entry) 00211 { 00212 return (entry.fd == retired_fd); 00213 } 00214 00215 #endif