Source · fairmq/zeromq/
Context.hheader
Described in Stop a topology without losing messages.
1/********************************************************************************2 * Copyright (C) 2020 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_CONTEXT_H_10#define FAIR_MQ_ZMQ_CONTEXT_H_11 12#include <fairmq/tools/Strings.h>13#include <fairmq/UnmanagedRegion.h>14 15#include <fairlogger/Logger.h>16 17#include <zmq.h>18 19#include <atomic>20#include <condition_variable>21#include <functional>22#include <mutex>23#include <queue>24#include <stdexcept>25#include <string>26#include <thread>27#include <vector>28 29namespace fair::mq::zmq30{31 32struct ContextError : std::runtime_error { using std::runtime_error::runtime_error; };33 34class Context35{36 public:37 Context(int numIoThreads)38 : fZmqCtx(zmq_ctx_new())39 , fInterrupted(false)40 {41 if (!fZmqCtx) {42 throw ContextError(tools::ToString("failed creating context, reason: ", zmq_strerror(errno)));43 }44 45 if (zmq_ctx_set(fZmqCtx, ZMQ_MAX_SOCKETS, 10000) != 0) {46 LOG(error) << "failed configuring context, reason: " << zmq_strerror(errno);47 throw ContextError(tools::ToString("failed configuring context, reason: ", zmq_strerror(errno)));48 }49 50 if (zmq_ctx_set(fZmqCtx, ZMQ_IO_THREADS, numIoThreads) != 0) {51 LOG(error) << "failed configuring context, reason: " << zmq_strerror(errno);52 throw ContextError(tools::ToString("failed configuring context, reason: ", zmq_strerror(errno)));53 }54 55 fRegionEvents.emplace(true, 0, nullptr, 0, 0, RegionEvent::local_only);56 }57 58 Context(const Context&) = delete;59 Context(Context&&) = delete;60 Context& operator=(const Context&) = delete;61 Context& operator=(Context&&) = delete;62 63 void SubscribeToRegionEvents(RegionEventCallback callback)64 {65 if (fRegionEventThread.joinable()) {66 LOG(debug) << "Already subscribed. Overwriting previous subscription.";67 {68 std::lock_guard<std::mutex> lock(fMtx);69 fRegionEventsSubscriptionActive = false;70 }71 fRegionEventsCV.notify_one();72 fRegionEventThread.join();73 }74 std::lock_guard<std::mutex> lock(fMtx);75 fRegionEventCallback = callback;76 fRegionEventsSubscriptionActive = true;77 fRegionEventThread = std::thread(&Context::RegionEventsSubscription, this);78 }79 80 bool SubscribedToRegionEvents() const { return fRegionEventThread.joinable(); }81 82 void UnsubscribeFromRegionEvents()83 {84 if (fRegionEventThread.joinable()) {85 std::unique_lock<std::mutex> lock(fMtx);86 fRegionEventsSubscriptionActive = false;87 lock.unlock();88 fRegionEventsCV.notify_one();89 fRegionEventThread.join();90 lock.lock();91 fRegionEventCallback = nullptr;92 }93 }94 95 void RegionEventsSubscription()96 {97 std::unique_lock<std::mutex> lock(fMtx);98 while (fRegionEventsSubscriptionActive) {99 while (!fRegionEvents.empty()) {100 auto i = fRegionEvents.front();101 fRegionEventCallback(i);102 fRegionEvents.pop();103 }104 fRegionEventsCV.wait(lock, [&]() { return !fRegionEventsSubscriptionActive || !fRegionEvents.empty(); });105 }106 }107 108 std::vector<RegionInfo> GetRegionInfo() const109 {110 std::lock_guard<std::mutex> lock(fMtx);111 return fRegionInfos;112 }113 114 uint16_t RegionCount() const115 {116 std::lock_guard<std::mutex> lock(fMtx);117 return fRegionCounter;118 }119 120 void AddRegion(bool managed, uint16_t id, void* ptr, size_t size, int64_t userFlags, RegionEvent event)121 {122 {123 std::lock_guard<std::mutex> lock(fMtx);124 ++fRegionCounter;125 fRegionInfos.emplace_back(managed, id, ptr, size, userFlags, event);126 fRegionEvents.emplace(managed, id, ptr, size, userFlags, event);127 }128 fRegionEventsCV.notify_one();129 }130 131 void RemoveRegion(uint16_t id)132 {133 {134 std::lock_guard<std::mutex> lock(fMtx);135 auto it = find_if(fRegionInfos.begin(), fRegionInfos.end(), [id](const RegionInfo& i) {136 return i.id == id;137 });138 if (it != fRegionInfos.end()) {139 fRegionEvents.push(*it);140 fRegionEvents.back().event = RegionEvent::destroyed;141 fRegionInfos.erase(it);142 } else {143 LOG(error) << "RemoveRegion: given id (" << id << ") not found.";144 }145 }146 fRegionEventsCV.notify_one();147 }148 149 void Interrupt() { fInterrupted.store(true); }150 void Resume() { fInterrupted.store(false); }151 void Reset() {}152 bool Interrupted() { return fInterrupted.load(); }153 154 void* GetZmqCtx() { return fZmqCtx; }155 156 ~Context()157 {158 UnsubscribeFromRegionEvents();159 160 if (fZmqCtx) {161 while (true) {162 if (zmq_ctx_term(fZmqCtx) != 0) {163 if (errno == EINTR) {164 LOG(debug) << "zmq_ctx_term interrupted by system call, retrying";165 continue;166 } else {167 fZmqCtx = nullptr;168 }169 }170 break;171 }172 } else {173 LOG(error) << "context not available for shutdown";174 }175 }176 177 private:178 void* fZmqCtx;179 mutable std::mutex fMtx;180 std::atomic<bool> fInterrupted;181 182 uint16_t fRegionCounter{1};183 std::condition_variable fRegionEventsCV;184 std::vector<RegionInfo> fRegionInfos;185 std::queue<RegionInfo> fRegionEvents;186 std::thread fRegionEventThread;187 std::function<void(RegionInfo)> fRegionEventCallback;188 bool fRegionEventsSubscriptionActive;189};190 191} // namespace fair::mq::zmq192 193#endif /* FAIR_MQ_ZMQ_CONTEXT_H_ */