Source · fairmq/zeromq/
Poller.hheader
Described in Poll several channels.
1/********************************************************************************2 * Copyright (C) 2014-2021 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *3 * *4 * This software is distributed under the terms of the *5 * GNU Lesser General Public Licence (LGPL) version 3, *6 * copied verbatim in the file "LICENSE" *7 ********************************************************************************/8 9#ifndef FAIR_MQ_ZMQ_POLLER_H10#define FAIR_MQ_ZMQ_POLLER_H11 12#include <fairlogger/Logger.h>13#include <fairmq/Channel.h>14#include <fairmq/Poller.h>15#include <fairmq/tools/Strings.h>16#include <fairmq/zeromq/Socket.h>17#include <unordered_map>18#include <vector>19#include <zmq.h>20 21namespace fair::mq::zmq22{23 24class Poller final : public fair::mq::Poller25{26 public:27 Poller() = default;28 Poller(const Poller&) = delete;29 Poller(Poller&&) = delete;30 Poller& operator=(const Poller&) = delete;31 Poller& operator=(Poller&&) = delete;32 33 Poller(const std::vector<Channel>& channels)34 : fItems()35 , fNumItems(0)36 {37 fNumItems = channels.size();38 fItems = new zmq_pollitem_t[fNumItems]; // TODO: fix me39 40 for (int i = 0; i < fNumItems; ++i) {41 fItems[i].socket = static_cast<const Socket*>(&(channels.at(i).GetSocket()))->GetSocket();42 fItems[i].fd = 0;43 fItems[i].revents = 0;44 45 int type = 0;46 size_t size = sizeof(type);47 zmq_getsockopt(static_cast<const Socket*>(&(channels.at(i).GetSocket()))->GetSocket(), ZMQ_TYPE, &type, &size);48 49 SetItemEvents(fItems[i], type);50 }51 }52 53 Poller(const std::vector<Channel*>& channels)54 : fItems()55 , fNumItems(0)56 {57 fNumItems = channels.size();58 fItems = new zmq_pollitem_t[fNumItems];59 60 for (int i = 0; i < fNumItems; ++i) {61 fItems[i].socket = static_cast<const Socket*>(&(channels.at(i)->GetSocket()))->GetSocket();62 fItems[i].fd = 0;63 fItems[i].revents = 0;64 65 int type = 0;66 size_t size = sizeof(type);67 zmq_getsockopt(static_cast<const Socket*>(&(channels.at(i)->GetSocket()))->GetSocket(), ZMQ_TYPE, &type, &size);68 69 SetItemEvents(fItems[i], type);70 }71 }72 73 Poller(const std::unordered_map<std::string, std::vector<Channel>>& channelsMap, const std::vector<std::string>& channelList)74 : fItems()75 , fNumItems(0)76 {77 try {78 int offset = 0;79 // calculate offsets and the total size of the poll item set80 for (std::string const & channel : channelList) {81 fOffsetMap[channel] = offset;82 offset += channelsMap.at(channel).size();83 fNumItems += channelsMap.at(channel).size();84 }85 86 fItems = new zmq_pollitem_t[fNumItems];87 88 int index = 0;89 for (const std::string& channel : channelList) {90 for (unsigned int i = 0; i < channelsMap.at(channel).size(); ++i) {91 index = fOffsetMap[channel] + i;92 93 fItems[index].socket = static_cast<const Socket*>(&(channelsMap.at(channel).at(i).GetSocket()))->GetSocket();94 fItems[index].fd = 0;95 fItems[index].revents = 0;96 97 int type = 0;98 size_t size = sizeof(type);99 zmq_getsockopt(static_cast<const Socket*>(&(channelsMap.at(channel).at(i).GetSocket()))->GetSocket(), ZMQ_TYPE, &type, &size);100 101 SetItemEvents(fItems[index], type);102 }103 }104 } catch (const std::out_of_range& oor) {105 LOG(error) << "at least one of the provided channel keys for poller initialization is invalid";106 LOG(error) << "out of range error: " << oor.what();107 throw fair::mq::PollerError(fair::mq::tools::ToString("At least one of the provided channel keys for poller initialization is invalid. ", "Out of range error: ", oor.what()));108 }109 }110 111 void SetItemEvents(zmq_pollitem_t& item, int type)112 {113 if (type == ZMQ_REQ || type == ZMQ_REP || type == ZMQ_PAIR || type == ZMQ_DEALER || type == ZMQ_ROUTER) {114 item.events = ZMQ_POLLIN | ZMQ_POLLOUT;115 } else if (type == ZMQ_PUSH || type == ZMQ_PUB || type == ZMQ_XPUB) {116 item.events = ZMQ_POLLOUT;117 } else if (type == ZMQ_PULL || type == ZMQ_SUB || type == ZMQ_XSUB) {118 item.events = ZMQ_POLLIN;119 } else {120 LOG(error) << "invalid poller configuration, exiting.";121 throw fair::mq::PollerError("Invalid poller configuration, exiting.");122 }123 }124 125 void Poll(int timeout) override126 {127 while (true) {128 if (zmq_poll(fItems, fNumItems, timeout) < 0) {129 if (errno == ETERM) {130 LOG(debug) << "polling exited, reason: " << zmq_strerror(errno);131 return;132 } else if (errno == EINTR) {133 LOG(debug) << "polling interrupted by system call";134 continue;135 } else {136 LOG(error) << "polling failed, reason: " << zmq_strerror(errno);137 throw fair::mq::PollerError(fair::mq::tools::ToString("Polling failed, reason: ", zmq_strerror(errno)));138 }139 }140 break;141 }142 }143 144 bool CheckInput(int index) override145 {146 if (fItems[index].revents & ZMQ_POLLIN) {147 return true;148 }149 150 return false;151 }152 153 bool CheckOutput(int index) override154 {155 if (fItems[index].revents & ZMQ_POLLOUT) {156 return true;157 }158 159 return false;160 }161 162 bool CheckInput(const std::string& channelKey, int index) override163 {164 try {165 if (fItems[fOffsetMap.at(channelKey) + index].revents & ZMQ_POLLIN) {166 return true;167 }168 169 return false;170 } catch (const std::out_of_range& oor) {171 LOG(error) << "invalid channel key: '" << channelKey << "'";172 LOG(error) << "out of range error: " << oor.what();173 throw fair::mq::PollerError(fair::mq::tools::ToString("Invalid channel key '", channelKey, "'. Out of range error: ", oor.what()));174 }175 }176 177 bool CheckOutput(const std::string& channelKey, int index) override178 {179 try {180 if (fItems[fOffsetMap.at(channelKey) + index].revents & ZMQ_POLLOUT) {181 return true;182 }183 184 return false;185 } catch (const std::out_of_range& oor) {186 LOG(error) << "invalid channel key: '" << channelKey << "'";187 LOG(error) << "out of range error: " << oor.what();188 throw fair::mq::PollerError(fair::mq::tools::ToString("Invalid channel key '", channelKey, "'. Out of range error: ", oor.what()));189 }190 }191 192 ~Poller() override { delete[] fItems; }193 194 private:195 zmq_pollitem_t* fItems;196 int fNumItems;197 198 std::unordered_map<std::string, int> fOffsetMap;199};200 201} // namespace fair::mq::zmq202 203#endif /* FAIR_MQ_ZMQ_POLLER_H */