Source · fairmq/shmem/
Poller.hpublic header
Defines class fair::mq::shmem::Poller.
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_SHMEM_POLLER_H_10#define FAIR_MQ_SHMEM_POLLER_H_11 12#include <fairlogger/Logger.h>13#include <fairmq/Channel.h>14#include <fairmq/Poller.h>15#include <fairmq/shmem/Socket.h>16#include <fairmq/tools/Strings.h>17#include <unordered_map>18#include <vector>19#include <zmq.h>20 21namespace fair::mq::shmem22{23 24class Poller final : public fair::mq::Poller25{26 public:27 Poller(const std::vector<Channel>& channels)28 : fItems()29 , fNumItems(0)30 {31 fNumItems = channels.size();32 fItems = new zmq_pollitem_t[fNumItems];33 34 for (int i = 0; i < fNumItems; ++i) {35 fItems[i].socket = static_cast<const Socket*>(&(channels.at(i).GetSocket()))->GetSocket();36 fItems[i].fd = 0;37 fItems[i].revents = 0;38 39 int type = 0;40 size_t size = sizeof(type);41 zmq_getsockopt(static_cast<const Socket*>(&(channels.at(i).GetSocket()))->GetSocket(), ZMQ_TYPE, &type, &size);42 43 SetItemEvents(fItems[i], type);44 }45 }46 47 Poller(const std::vector<Channel*>& channels)48 : fItems()49 , fNumItems(0)50 {51 fNumItems = channels.size();52 fItems = new zmq_pollitem_t[fNumItems];53 54 for (int i = 0; i < fNumItems; ++i) {55 fItems[i].socket = static_cast<const Socket*>(&(channels.at(i)->GetSocket()))->GetSocket();56 fItems[i].fd = 0;57 fItems[i].revents = 0;58 59 int type = 0;60 size_t size = sizeof(type);61 zmq_getsockopt(static_cast<const Socket*>(&(channels.at(i)->GetSocket()))->GetSocket(), ZMQ_TYPE, &type, &size);62 63 SetItemEvents(fItems[i], type);64 }65 }66 67 Poller(const std::unordered_map<std::string, std::vector<Channel>>& channelsMap, const std::vector<std::string>& channelList)68 : fItems()69 , fNumItems(0)70 {71 try {72 int offset = 0;73 // calculate offsets and the total size of the poll item set74 for (const std::string& channel : channelList) {75 fOffsetMap[channel] = offset;76 offset += channelsMap.at(channel).size();77 fNumItems += channelsMap.at(channel).size();78 }79 80 fItems = new zmq_pollitem_t[fNumItems];81 82 int index = 0;83 for (const std::string& channel : channelList) {84 for (unsigned int i = 0; i < channelsMap.at(channel).size(); ++i) {85 index = fOffsetMap[channel] + i;86 87 fItems[index].socket = static_cast<const Socket*>(&(channelsMap.at(channel).at(i).GetSocket()))->GetSocket();88 fItems[index].fd = 0;89 fItems[index].revents = 0;90 91 int type = 0;92 size_t size = sizeof(type);93 zmq_getsockopt(static_cast<const Socket*>(&(channelsMap.at(channel).at(i).GetSocket()))->GetSocket(), ZMQ_TYPE, &type, &size);94 95 SetItemEvents(fItems[index], type);96 }97 }98 } catch (const std::out_of_range& oor) {99 LOG(error) << "At least one of the provided channel keys for poller initialization is invalid." << " Out of range error: " << oor.what();100 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()));101 }102 }103 104 Poller(const Poller&) = delete;105 Poller(Poller&&) = delete;106 Poller& operator=(const Poller&) = delete;107 Poller& operator=(Poller&&) = delete;108 109 void SetItemEvents(zmq_pollitem_t& item, int type)110 {111 if (type == ZMQ_REQ || type == ZMQ_REP || type == ZMQ_PAIR || type == ZMQ_DEALER || type == ZMQ_ROUTER) {112 item.events = ZMQ_POLLIN | ZMQ_POLLOUT;113 } else if (type == ZMQ_PUSH || type == ZMQ_PUB || type == ZMQ_XPUB) {114 item.events = ZMQ_POLLOUT;115 } else if (type == ZMQ_PULL || type == ZMQ_SUB || type == ZMQ_XSUB) {116 item.events = ZMQ_POLLIN;117 } else {118 LOG(error) << "invalid poller configuration, exiting.";119 throw fair::mq::PollerError("Invalid poller configuration, exiting.");120 }121 }122 123 void Poll(int timeout) override124 {125 while (true) {126 if (zmq_poll(fItems, fNumItems, timeout) < 0) {127 if (errno == ETERM) {128 LOG(debug) << "polling exited, reason: " << zmq_strerror(errno);129 return;130 } else if (errno == EINTR) {131 LOG(debug) << "polling interrupted by system call";132 continue;133 } else {134 LOG(error) << "polling failed, reason: " << zmq_strerror(errno);135 throw fair::mq::PollerError(fair::mq::tools::ToString("Polling failed, reason: ", zmq_strerror(errno)));136 }137 }138 break;139 }140 }141 142 bool CheckInput(int index) override143 {144 if (fItems[index].revents & ZMQ_POLLIN) {145 return true;146 }147 148 return false;149 }150 151 bool CheckOutput(int index) override152 {153 if (fItems[index].revents & ZMQ_POLLOUT) {154 return true;155 }156 157 return false;158 }159 160 bool CheckInput(const std::string& channelKey, int index) override161 {162 try {163 if (fItems[fOffsetMap.at(channelKey) + index].revents & ZMQ_POLLIN) {164 return true;165 }166 167 return false;168 } catch (const std::out_of_range& oor) {169 LOG(error) << "invalid channel key: '" << channelKey << "'";170 LOG(error) << "out of range error: " << oor.what();171 throw fair::mq::PollerError(fair::mq::tools::ToString("Invalid channel key '", channelKey, "'. Out of range error: ", oor.what()));172 }173 }174 175 bool CheckOutput(const std::string& channelKey, int index) override176 {177 try {178 if (fItems[fOffsetMap.at(channelKey) + index].revents & ZMQ_POLLOUT) {179 return true;180 }181 182 return false;183 } catch (const std::out_of_range& oor) {184 LOG(error) << "invalid channel key: '" << channelKey << "'";185 LOG(error) << "out of range error: " << oor.what();186 throw fair::mq::PollerError(fair::mq::tools::ToString("Invalid channel key '", channelKey, "'. Out of range error: ", oor.what()));187 }188 }189 190 ~Poller() override { delete[] fItems; }191 192 private:193 zmq_pollitem_t* fItems;194 int fNumItems;195 196 std::unordered_map<std::string, int> fOffsetMap;197};198 199} // namespace fair::mq::shmem200 201#endif /* FAIR_MQ_SHMEM_POLLER_H_ */