ZmqContext.hpp

Go to the documentation of this file.

#ifndef IPM_INCLUDE_IPM_ZMQCONTEXT_HPP_
#define IPM_INCLUDE_IPM_ZMQCONTEXT_HPP_
 
#include "ers/Issue.hpp"
#include "logging/Logging.hpp" // NOTE: if ISSUES ARE DECLARED BEFORE include logging/Logging.hpp, TLOG_DEBUG<<issue wont work.
#include "zmq.hpp"
 
namespace dunedaq {
 
ERS_DECLARE_ISSUE(ipm,
                  ZmqOperationError,
                  "An exception occured while calling " << operation << " on the ZMQ " << direction << " socket: "
                                                        << what << " (connection_string: " << connection_string << ")",
                  ((std::string)operation)((std::string)direction)((const char*)what)(
                    (std::string)connection_string)) // NOLINT
 
ERS_DECLARE_ISSUE(ipm,
                  ZmqSendError,
                  "An exception occurred while sending " << N << " bytes to " << topic << ": " << what,
                  ((const char*)what)((int)N)((std::string)topic)) // NOLINT
 
ERS_DECLARE_ISSUE(ipm,
                  ZmqReceiveError,
                  "An exception occured while receiving " << which << ": " << what,
                  ((const char*)what)((const char*)which)) // NOLINT
 
ERS_DECLARE_ISSUE(ipm,
                  ZmqSubscribeError,
                  "An execption occured while subscribing to " << topic << ": " << what,
                  ((const char*)what)((std::string)topic)) // NOLINT
 
ERS_DECLARE_ISSUE(ipm,
                  ZmqUnsubscribeError,
                  "An execption occured while unsubscribing from " << topic << ": " << what,
                  ((const char*)what)((std::string)topic)) // NOLINT
 
namespace ipm {
class ZmqContext
{
public:
  static ZmqContext& instance()
  {
    static ZmqContext s_ctx;
    return s_ctx;
  }
 
  zmq::context_t& GetContext() { return m_context; }
 
  void set_context_threads(int nthreads) { 
      TLOG_DEBUG(10) << "Setting ZMQ Context IO thread count to " << nthreads;
      m_context.set(zmq::ctxopt::io_threads, nthreads); }
  void set_context_maxsockets(int max_sockets) { 
      TLOG_DEBUG(10) << "Setting ZMQ Context max sockets to " << max_sockets;
      m_context.set(zmq::ctxopt::max_sockets, max_sockets); }
 
private:
  ZmqContext()
  {
    auto threads_c = getenv("IPM_ZMQ_IO_THREADS");
    if (threads_c != nullptr) {
      auto threads = std::atoi(threads_c);
      if (threads > 1) {
        set_context_threads(threads);
      }
    }
 
    bool sockets_set = false;
    auto sockets_c = getenv("IPM_ZMQ_MAX_SOCKETS");
    if (sockets_c != nullptr) {
      auto sockets = std::atoi(sockets_c);
      if (sockets > s_minimum_sockets) {
        set_context_maxsockets(sockets);
        sockets_set = true;
      }
    }
    if(!sockets_set) {
      set_context_maxsockets(s_minimum_sockets);
    }
    
  }
  ~ZmqContext() { m_context.close(); }
  zmq::context_t m_context;
  static constexpr int s_minimum_sockets = 16636;
 
  ZmqContext(ZmqContext const&) = delete;
  ZmqContext(ZmqContext&&) = delete;
  ZmqContext& operator=(ZmqContext const&) = delete;
  ZmqContext& operator=(ZmqContext&&) = delete;
};
} // namespace ipm
} // namespace dunedaq
 
#endif // IPM_INCLUDE_IPM_ZMQCONTEXT_HPP_