diff --git a/.ci/copy_headers.sh b/.ci/copy_headers.sh index fd4ec82b96e9d..a6587b6ee1dbb 100755 --- a/.ci/copy_headers.sh +++ b/.ci/copy_headers.sh @@ -10,6 +10,6 @@ cmake -DCMAKE_EXPORT_COMPILE_COMMANDS=ON -Dall=On -Dtesting=On -Dx11=Off -Dalien # We need to prebuild a minimal set of targets which are responsible for header copy # or generation. make -j4 move_headers intrinsics_gen clang-tablegen-targets ClangDriverOptions \ - googletest Dictgen BaseTROOT + googletest Dictgen BaseTROOT BUILTIN_cppzmq ln -s $PWD/compile_commands.json $PWD/../root/ diff --git a/builtins/zeromq/CMakeLists.txt b/builtins/zeromq/CMakeLists.txt new file mode 100644 index 0000000000000..11f9828297488 --- /dev/null +++ b/builtins/zeromq/CMakeLists.txt @@ -0,0 +1,51 @@ +include(ExternalProject) + +set(ZeroMQ_PREFIX ${CMAKE_CURRENT_BINARY_DIR}/BUILTIN_ZeroMQ-prefix) +set(ZeroMQ_LIBNAME ${CMAKE_STATIC_LIBRARY_PREFIX}zmq${CMAKE_STATIC_LIBRARY_SUFFIX}) + +set(ZeroMQ_LIBRARY ${ZeroMQ_PREFIX}/lib/${ZeroMQ_LIBNAME} CACHE INTERNAL "" FORCE) +set(ZeroMQ_LIBRARIES ${ZeroMQ_LIBRARY} CACHE INTERNAL "" FORCE) + +ExternalProject_Add(BUILTIN_ZeroMQ + URL https://github.com/zeromq/libzmq/archive/refs/tags/v4.3.2.tar.gz + URL_HASH SHA256=02ecc88466ae38cf2c8d79f09cfd2675ba299a439680b64ade733e26a349edeb + CMAKE_ARGS + -DCMAKE_INSTALL_PREFIX= + -DWITH_PERF_TOOL=OFF + -DZMQ_BUILD_TESTS=OFF + -DPOLLER=select + -DCMAKE_CXX_COMPILER=${CMAKE_CXX_COMPILER} + # TODO: remove -w flag when updating to v4.3.5 or higher, current warnings should be gone by then + -DCMAKE_CXX_FLAGS=${CMAKE_CXX_FLAGS}\ -w + BUILD_BYPRODUCTS ${ZeroMQ_LIBRARIES} + ) + +ExternalProject_Get_Property(BUILTIN_ZeroMQ install_dir) + +set(ZeroMQ_FOUND TRUE CACHE BOOL "" FORCE) + +set(ZeroMQ_INCLUDE_DIR ${ZeroMQ_PREFIX}/include CACHE INTERNAL "" FORCE) +set(ZeroMQ_INCLUDE_DIRS ${ZeroMQ_PREFIX}/include CACHE INTERNAL "" FORCE) + +add_library(ZeroMQ INTERFACE) +target_include_directories(ZeroMQ INTERFACE $) +target_link_libraries(ZeroMQ INTERFACE $) +add_dependencies(ZeroMQ BUILTIN_ZeroMQ) + +add_library(ZeroMQ::ZeroMQ ALIAS ZeroMQ) +add_library(libzmq ALIAS ZeroMQ) + +set(ZeroMQ_DIR ${ZeroMQ_PREFIX}/share/cmake/ZeroMQ CACHE INTERNAL "" FORCE) + +# also need source and build directories for ppoll in RooFitZMQ +set(ZeroMQ_SOURCE_DIR ${ZeroMQ_PREFIX}/src/BUILTIN_ZeroMQ/src CACHE INTERNAL "" FORCE) +set(ZeroMQ_BUILD_DIR ${ZeroMQ_PREFIX}/src/BUILTIN_ZeroMQ-build CACHE INTERNAL "" FORCE) + +ExternalProject_Add(BUILTIN_cppzmq + URL https://github.com/zeromq/cppzmq/archive/refs/tags/v4.3.0.tar.gz + URL_HASH SHA256=27d1f56406ba94ee779e639203218820975cf68174f92fbeae0f645df0fcada4 + CMAKE_ARGS + -DCMAKE_INSTALL_PREFIX=${install_dir} + -DCPPZMQ_BUILD_TESTS=OFF + DEPENDS BUILTIN_ZeroMQ + ) diff --git a/cmake/modules/RootBuildOptions.cmake b/cmake/modules/RootBuildOptions.cmake index 1cc9f48926367..cd157236e552e 100644 --- a/cmake/modules/RootBuildOptions.cmake +++ b/cmake/modules/RootBuildOptions.cmake @@ -107,6 +107,7 @@ ROOT_BUILD_OPTION(builtin_vdt OFF "Build VDT internally (requires network)") ROOT_BUILD_OPTION(builtin_veccore OFF "Build VecCore internally (requires network)") ROOT_BUILD_OPTION(builtin_xrootd OFF "Build XRootD internally (requires network)") ROOT_BUILD_OPTION(builtin_xxhash OFF "Build bundled copy of xxHash") +ROOT_BUILD_OPTION(builtin_zeromq ON "Build ZeroMQ internally (requires network)") ROOT_BUILD_OPTION(builtin_zlib OFF "Build bundled copy of zlib") ROOT_BUILD_OPTION(builtin_zstd OFF "Build included libzstd, or use system libzstd") ROOT_BUILD_OPTION(ccache OFF "Enable ccache usage for speeding up builds") @@ -299,6 +300,7 @@ if(builtin_all) set(builtin_veccore_defvalue ON) set(builtin_xrootd_defvalue ON) set(builtin_xxhash_defvalue ON) + set(builtin_zeromq_defvalue ON) set(builtin_zlib_defvalue ON) set(builtin_zstd_defvalue ON) endif() diff --git a/cmake/modules/SearchInstalledSoftware.cmake b/cmake/modules/SearchInstalledSoftware.cmake index c64ab05414e21..0624a00fbb3a3 100644 --- a/cmake/modules/SearchInstalledSoftware.cmake +++ b/cmake/modules/SearchInstalledSoftware.cmake @@ -1767,6 +1767,13 @@ if (mpi) endif() endif() +#---Check for ZeroMQ----------------------------------------------------------- + +if(builtin_zeromq) + list(APPEND ROOT_BUILTINS ZeroMQ) + add_subdirectory(builtins/zeromq) +endif() + #---Download googletest-------------------------------------------------------------- if (testing) # FIXME: Remove our version of gtest in roottest. We can reuse this one. diff --git a/roofit/CMakeLists.txt b/roofit/CMakeLists.txt index 890df0340dde2..cec391f22a510 100644 --- a/roofit/CMakeLists.txt +++ b/roofit/CMakeLists.txt @@ -5,6 +5,9 @@ # For the list of contributors see $ROOTSYS/README/CREDITS. add_subdirectory(batchcompute) +if (builtin_zeromq) + add_subdirectory(roofitZMQ) +endif(builtin_zeromq) add_subdirectory(roofitcore) add_subdirectory(roofit) if(mathmore) diff --git a/roofit/roofitZMQ/CMakeLists.txt b/roofit/roofitZMQ/CMakeLists.txt new file mode 100644 index 0000000000000..81fee8d871a92 --- /dev/null +++ b/roofit/roofitZMQ/CMakeLists.txt @@ -0,0 +1,38 @@ +############################################################################ +# CMakeLists.txt file for building ROOT roofitcore/ZMQ package +# @author Patrick Bos, NL eScience Center +############################################################################ + +#if(NOT ZeroMQ_FOUND) +# find_package(ZeroMQ REQUIRED) +#endif() + +ROOT_STANDARD_LIBRARY_PACKAGE(RooFitZMQ + HEADERS + RooFit_ZMQ/Utility.h + RooFit_ZMQ/ZeroMQSvc.h + RooFit_ZMQ/functions.h + RooFit_ZMQ/ZeroMQPoller.h + RooFit_ZMQ/ppoll.h + SOURCES + src/ZeroMQSvc.cpp + src/ZeroMQPoller.cpp + src/functions.cpp + src/ppoll.cpp + DICTIONARY_OPTIONS + -writeEmptyRootPCM + DEPENDENCIES + Core # TVersionCheck, RegisterModule, these are always called from G__RooFitZMQ.cxx.o + LIBRARIES + ZeroMQ + ) + + +target_link_libraries(RooFitZMQ PRIVATE ZeroMQ) +target_include_directories(RooFitZMQ PUBLIC $) +target_include_directories(RooFitZMQ PUBLIC $) +target_include_directories(RooFitZMQ PUBLIC $) + +if(testing) + ROOT_ADD_TEST_SUBDIRECTORY(test) +endif() diff --git a/roofit/roofitZMQ/inc/LinkDef.h b/roofit/roofitZMQ/inc/LinkDef.h new file mode 100644 index 0000000000000..5495493c90b99 --- /dev/null +++ b/roofit/roofitZMQ/inc/LinkDef.h @@ -0,0 +1 @@ +// something \ No newline at end of file diff --git a/roofit/roofitZMQ/inc/RooFit_ZMQ/Utility.h b/roofit/roofitZMQ/inc/RooFit_ZMQ/Utility.h new file mode 100644 index 0000000000000..58a9df24f6e7e --- /dev/null +++ b/roofit/roofitZMQ/inc/RooFit_ZMQ/Utility.h @@ -0,0 +1,34 @@ +#ifndef SERIALIZE_UTILITY_H +#define SERIALIZE_UTILITY_H 1 + +#if defined(__clang__) && (__clang_major__ > 3 || (__clang_major__ == 3 && __clang_minor__ >= 10)) +#define HAVE_TRIVIALLY_COPYABLE 1 +#elif defined(__GNUC__) && __GNUC__ >= 5 +#define HAVE_TRIVIALLY_COPYABLE 1 +#else +#undef HAVE_TRIVIALLY_COPYABLE +#endif + +#include + +namespace ZMQ { +namespace Detail { + +#if defined(HAVE_TRIVIALLY_COPYABLE) +template +using simple_object = std::is_trivially_copyable; +#else +template +using simple_object = std::is_pod; +#endif + +// is trivial +template +struct is_trivial + : std::conditional::type>::value, std::true_type, std::false_type>::type { +}; + +} // namespace Detail +} // namespace ZMQ + +#endif // SERIALIZE_UTILITY_H diff --git a/roofit/roofitZMQ/inc/RooFit_ZMQ/ZeroMQPoller.h b/roofit/roofitZMQ/inc/RooFit_ZMQ/ZeroMQPoller.h new file mode 100644 index 0000000000000..a4709aca2c5f4 --- /dev/null +++ b/roofit/roofitZMQ/inc/RooFit_ZMQ/ZeroMQPoller.h @@ -0,0 +1,46 @@ +#ifndef ZEROMQPOLLER_H +#define ZEROMQPOLLER_H 1 +#include +#include +#include +#include + +#include "RooFit_ZMQ/ZeroMQSvc.h" +#include "RooFit_ZMQ/functions.h" + +class ZeroMQPoller { +public: + using entry_t = std::tuple; + // The key is what zmq::socket_t stores inside, and what goes into + // pollitem_t through zmq::socket_t's conversion to void* operator + using sockets_t = std::unordered_map; + + using fd_entry_t = std::tuple; + using fds_t = std::unordered_map; + + using free_t = std::deque; + + ZeroMQPoller() = default; + + std::vector> poll(int timeo = -1); + std::vector> ppoll(int timeo, const sigset_t *sigmask_); + + size_t size() const; + + size_t register_socket(zmq::socket_t &socket, zmq::PollType type); + size_t register_socket(int fd, zmq::PollType type); + + size_t unregister_socket(zmq::socket_t &socket); + size_t unregister_socket(int fd); + +private: + // Vector of (socket, flags) + std::vector m_items; + sockets_t m_sockets; + fds_t m_fds; + + // free slots in items + free_t m_free; +}; + +#endif // ZEROMQPOLLER_H diff --git a/roofit/roofitZMQ/inc/RooFit_ZMQ/ZeroMQSvc.h b/roofit/roofitZMQ/inc/RooFit_ZMQ/ZeroMQSvc.h new file mode 100644 index 0000000000000..ef31e68f85b42 --- /dev/null +++ b/roofit/roofitZMQ/inc/RooFit_ZMQ/ZeroMQSvc.h @@ -0,0 +1,221 @@ +#ifndef ZEROMQ_IZEROMQSVC_H +#define ZEROMQ_IZEROMQSVC_H 1 + +// Include files +// from STL +#include +#include +#include +#include +#include +#include // std::cerr + +// ZeroMQ +#include +#include "RooFit_ZMQ/Utility.h" +#include "RooFit_ZMQ/functions.h" + +// debugging +#include // getpid + +namespace ZMQ { + +struct TimeOutException : std::exception { + TimeOutException() = default; + TimeOutException(const TimeOutException &) = default; + ~TimeOutException() = default; + TimeOutException &operator=(const TimeOutException &) = default; +}; + +struct MoreException : std::exception { + MoreException() = default; + MoreException(const MoreException &) = default; + ~MoreException() = default; + MoreException &operator=(const MoreException &) = default; +}; + +} // namespace ZMQ + +template +struct ZmqLingeringSocketPtrDeleter { + void operator()(zmq::socket_t *socket) + { + auto period = PERIOD; + + int tries = 0; + int max_tries = 3; + while (true) { + try { + // the actual work this function should do, plus the delete socket below: + if (socket) + socket->setsockopt(ZMQ_LINGER, &period, sizeof(period)); + break; + } catch (zmq::error_t &e) { + if (++tries == max_tries || e.num() == EINVAL || e.num() == ETERM || + e.num() == ENOTSOCK // not recoverable from here + ) { + std::cerr << "ERROR in ZeroMQSvc::socket: " << e.what() << " (errno: " << e.num() << ")\n"; + throw; + } + std::cerr << "RETRY " << tries << "/" << (max_tries - 1) + << " in ZmqLingeringSocketPtrDeleter: call interrupted (errno: " << e.num() << ")\n"; + } + } + + delete socket; + } +}; + +template +using ZmqLingeringSocketPtr = std::unique_ptr>; + +// We retry send and receive only on EINTR, all other errors are either fatal, or can only +// be handled at the caller. +template +auto retry_send(zmq::socket_t &socket, int max_tries, args_t... args) -> decltype(socket.send(args...)) +{ + int tries = 0; + while (true) { + try { + // the actual work this function should do, all the rest is error handling: + return socket.send(args...); + } catch (zmq::error_t &e) { + if (++tries == max_tries || e.num() != EINTR // only recoverable error + ) { + // std::cerr << "ERROR in ZeroMQSvc::send (retry_send) on pid " << getpid() << ": " << e.what() << " + // (errno: " << e.num() << ")\n"; + throw; + } + std::cerr << "RETRY " << tries << "/" << (max_tries - 1) << " in ZeroMQSvc::send (retry_send) on pid " + << getpid() << ": " << e.what() << ")\n"; + } + } +} + +template +auto retry_recv(zmq::socket_t &socket, int max_tries, args_t... args) -> decltype(socket.recv(args...)) +{ + int tries = 0; + while (true) { + try { + // the actual work this function should do, all the rest is error handling: + return socket.recv(args...); + } catch (zmq::error_t &e) { + if (++tries == max_tries || e.num() != EINTR // only recoverable error + ) { + // std::cerr << "ERROR in ZeroMQSvc::recv (retry_recv) on pid " << getpid() << ": " << e.what() << " + // (errno: " << e.num() << ")\n"; + throw; + } + std::cerr << "RETRY " << tries << "/" << (max_tries - 1) << " in ZeroMQSvc::recv (retry_recv) on pid " + << getpid() << ": " << e.what() << ")\n"; + } + } +} + +/** @class IZeroMQSvc IZeroMQSvc.h ZeroMQ/IZeroMQSvc.h + * + * + * @author + * @date 2015-06-22 + */ +class ZeroMQSvc { + // Note on error handling: + // Creating message_t can throw, but only when memory ran out (errno ENOMEM), + // and that is something only the caller can fix, so we don't catch it here. + +public: + enum Encoding { Text = 0, Binary }; + + Encoding encoding() const; + void setEncoding(const Encoding &e); + zmq::context_t &context() const; + zmq::socket_t socket(int type) const; + zmq::socket_t *socket_ptr(int type) const; + void close_context() const; + + // decode message with ZMQ, POD version + template ::value && ZMQ::Detail::is_trivial::value, T>::type + * = nullptr> + T decode(const zmq::message_t &msg) const + { + T object; + memcpy(&object, msg.data(), msg.size()); + return object; + } + + // decode ZMQ message, string version + template ::value, T>::type * = nullptr> + std::string decode(const zmq::message_t &msg) const + { + std::string r(msg.size() + 1, char{}); + r.assign(static_cast(msg.data()), msg.size()); + return r; + } + + // receive message with ZMQ, general version + // FIXME: what to do with flags=0.... more is a pointer, that might prevent conversion + template ::value), T>::type * = nullptr> + T receive(zmq::socket_t &socket, int flags = 0, bool *more = nullptr) const + { + // receive message + zmq::message_t msg; + auto nbytes = retry_recv(socket, 2, &msg, flags); + if (0 == nbytes) { + throw ZMQ::TimeOutException{}; + } + if (more) + *more = msg.more(); + + // decode message + return decode(msg); + } + + // receive message with ZMQ + template ::value, T>::type * = nullptr> + T receive(zmq::socket_t &socket, int flags = 0, bool *more = nullptr) const + { + // receive message + zmq::message_t msg; + auto nbytes = retry_recv(socket, 2, &msg, flags); + if (0 == nbytes) { + throw ZMQ::TimeOutException{}; + } + if (more) + *more = msg.more(); + return msg; + } + + // encode message to ZMQ + template ::value && ZMQ::Detail::is_trivial::value, T>::type + * = nullptr> + zmq::message_t encode(const T &item, std::function sizeFun = ZMQ::defaultSizeOf) const + { + size_t s = sizeFun(item); + zmq::message_t msg{s}; + memcpy((void *)msg.data(), &item, s); + return msg; + } + + zmq::message_t encode(const char *item) const; + zmq::message_t encode(const std::string &item) const; + + // Send message with ZMQ + template ::value, T>::type * = nullptr> + bool send(zmq::socket_t &socket, const T &item, int flags = 0) const + { + return retry_send(socket, 1, encode(item), flags); + } + + bool send(zmq::socket_t &socket, const char *item, int flags = 0) const; + bool send(zmq::socket_t &socket, zmq::message_t &msg, int flags = 0) const; + bool send(zmq::socket_t &socket, zmq::message_t &&msg, int flags = 0) const; + +private: + Encoding m_enc = Text; + mutable zmq::context_t *m_context = nullptr; +}; + +ZeroMQSvc &zmqSvc(); + +#endif // ZEROMQ_IZEROMQSVC_H diff --git a/roofit/roofitZMQ/inc/RooFit_ZMQ/functions.h b/roofit/roofitZMQ/inc/RooFit_ZMQ/functions.h new file mode 100644 index 0000000000000..f57a1da69e52d --- /dev/null +++ b/roofit/roofitZMQ/inc/RooFit_ZMQ/functions.h @@ -0,0 +1,102 @@ +#ifndef ZEROMQ_FUNCTIONS_H +#define ZEROMQ_FUNCTIONS_H 1 + +#include + +namespace zmq { + +enum SocketTypes { + PAIR = ZMQ_PAIR, + PUB = ZMQ_PUB, + SUB = ZMQ_SUB, + REQ = ZMQ_REQ, + REP = ZMQ_REP, + DEALER = ZMQ_DEALER, + ROUTER = ZMQ_ROUTER, + PULL = ZMQ_PULL, + PUSH = ZMQ_PUSH, + XPUB = ZMQ_XPUB, + XSUB = ZMQ_XSUB, + STREAM = ZMQ_STREAM +}; + +enum PollType : short { POLLIN = ZMQ_POLLIN, POLLOUT = ZMQ_POLLOUT }; + +enum SocketOptions { + AFFINITY = ZMQ_AFFINITY, + IDENTITY = ZMQ_IDENTITY, + SUBSCRIBE = ZMQ_SUBSCRIBE, + UNSUBSCRIBE = ZMQ_UNSUBSCRIBE, + RATE = ZMQ_RATE, + RECOVERY_IVL = ZMQ_RECOVERY_IVL, + SNDBUF = ZMQ_SNDBUF, + RCVBUF = ZMQ_RCVBUF, + RCVMORE = ZMQ_RCVMORE, + FD = ZMQ_FD, + EVENTS = ZMQ_EVENTS, + TYPE = ZMQ_TYPE, + LINGER = ZMQ_LINGER, + RECONNECT_IVL = ZMQ_RECONNECT_IVL, + BACKLOG = ZMQ_BACKLOG, + RECONNECT_IVL_MAX = ZMQ_RECONNECT_IVL_MAX, + MAXMSGSIZE = ZMQ_MAXMSGSIZE, + SNDHWM = ZMQ_SNDHWM, + RCVHWM = ZMQ_RCVHWM, + MULTICAST_HOPS = ZMQ_MULTICAST_HOPS, + RCVTIMEO = ZMQ_RCVTIMEO, + SNDTIMEO = ZMQ_SNDTIMEO, + LAST_ENDPOINT = ZMQ_LAST_ENDPOINT, + ROUTER_MANDATORY = ZMQ_ROUTER_MANDATORY, + TCP_KEEPALIVE = ZMQ_TCP_KEEPALIVE, + TCP_KEEPALIVE_CNT = ZMQ_TCP_KEEPALIVE_CNT, + TCP_KEEPALIVE_IDLE = ZMQ_TCP_KEEPALIVE_IDLE, + TCP_KEEPALIVE_INTVL = ZMQ_TCP_KEEPALIVE_INTVL, + IMMEDIATE = ZMQ_IMMEDIATE, + XPUB_VERBOSE = ZMQ_XPUB_VERBOSE, + ROUTER_RAW = ZMQ_ROUTER_RAW, + IPV6 = ZMQ_IPV6, + MECHANISM = ZMQ_MECHANISM, + PLAIN_SERVER = ZMQ_PLAIN_SERVER, + PLAIN_USERNAME = ZMQ_PLAIN_USERNAME, + PLAIN_PASSWORD = ZMQ_PLAIN_PASSWORD, + CURVE_SERVER = ZMQ_CURVE_SERVER, + CURVE_PUBLICKEY = ZMQ_CURVE_PUBLICKEY, + CURVE_SECRETKEY = ZMQ_CURVE_SECRETKEY, + CURVE_SERVERKEY = ZMQ_CURVE_SERVERKEY, + PROBE_ROUTER = ZMQ_PROBE_ROUTER, + REQ_CORRELATE = ZMQ_REQ_CORRELATE, + REQ_RELAXED = ZMQ_REQ_RELAXED, + CONFLATE = ZMQ_CONFLATE, + ZAP_DOMAIN = ZMQ_ZAP_DOMAIN, + ROUTER_HANDOVER = ZMQ_ROUTER_HANDOVER, + TOS = ZMQ_TOS, + CONNECT_RID = ZMQ_CONNECT_RID, + GSSAPI_SERVER = ZMQ_GSSAPI_SERVER, + GSSAPI_PRINCIPAL = ZMQ_GSSAPI_PRINCIPAL, + GSSAPI_SERVICE_PRINCIPAL = ZMQ_GSSAPI_SERVICE_PRINCIPAL, + GSSAPI_PLAINTEXT = ZMQ_GSSAPI_PLAINTEXT, + HANDSHAKE_IVL = ZMQ_HANDSHAKE_IVL, + SOCKS_PROXY = ZMQ_SOCKS_PROXY, + XPUB_NODROP = ZMQ_XPUB_NODROP +}; + +// Message options +enum MessageOptions { MORE = ZMQ_MORE, SRCFD = ZMQ_SRCFD, SHARED = ZMQ_SHARED }; + +// Send/recv options. +enum SendRecvOptions { DONTWAIT = ZMQ_DONTWAIT, SNDMORE = ZMQ_SNDMORE }; +} // namespace zmq + +namespace ZMQ { + +template +size_t defaultSizeOf(const T &) +{ + return sizeof(T); +} + +size_t stringLength(const char &cs); + +} // namespace ZMQ + +#endif // ZEROMQ_FUNCTIONS_H diff --git a/roofit/roofitZMQ/inc/RooFit_ZMQ/ppoll.h b/roofit/roofitZMQ/inc/RooFit_ZMQ/ppoll.h new file mode 100644 index 0000000000000..678dbe1d9ee04 --- /dev/null +++ b/roofit/roofitZMQ/inc/RooFit_ZMQ/ppoll.h @@ -0,0 +1,17 @@ +#ifndef ROOT_ROOFIT_ZMQ_ppoll +#define ROOT_ROOFIT_ZMQ_ppoll + +#include +#include + +namespace ZMQ { + +int zmq_ppoll(zmq_pollitem_t *items_, int nitems_, long timeout_, const sigset_t *sigmask_); +int ppoll(zmq_pollitem_t *items_, size_t nitems_, long timeout_, const sigset_t *sigmask_); +int ppoll(std::vector &items, long timeout_, const sigset_t *sigmask_); +class ppoll_error_t : public zmq::error_t { +}; + +} // namespace ZMQ + +#endif // ROOT_ROOFIT_ZMQ_ppoll diff --git a/roofit/roofitZMQ/src/ZeroMQPoller.cpp b/roofit/roofitZMQ/src/ZeroMQPoller.cpp new file mode 100644 index 0000000000000..d08ba2e70f20f --- /dev/null +++ b/roofit/roofitZMQ/src/ZeroMQPoller.cpp @@ -0,0 +1,171 @@ +/* + * ROOT + * Copyright (c) 2018, Patrick Bos + * Distributed under the terms of the BSD 3-Clause License. + * + * The full license is in the file LICENSE, distributed with this software. + */ + +#include "RooFit_ZMQ/ZeroMQPoller.h" + +#include + +#include "RooFit_ZMQ/ppoll.h" + +std::vector> ZeroMQPoller::poll(int timeo) +{ + std::vector> r; + if (m_items.empty()) { + throw std::runtime_error("No sockets registered"); + } + int n = 0; + while (true) { + try { + n = zmq::poll(&m_items[0], m_items.size(), timeo); + if (n == 0) + return r; + break; + } catch (const zmq::error_t &e) { + std::cerr << "in ZeroMQPoller::poll on PID " << getpid() << ": " << e.what() << std::endl; + if (e.num() != EINTR) { + throw; + } + } + } + // TODO: replace this with ranges::v3::zip + for (size_t i = 0; i < m_items.size(); ++i) { + void *socket = m_items[i].socket; + size_t index = 0; + int flags = 0; + if (socket == nullptr) { + // an fd was registered + std::tie(index, flags) = m_fds[m_items[i].fd]; + } else { + // a socket was registered + const zmq::socket_t *s; + std::tie(index, flags, s) = m_sockets[socket]; + } + if (m_items[i].revents & short(flags)) { + r.emplace_back(index, flags); + } + } + return r; +} + +// This function can throw (from inside ZMQ::ppoll), so wrap in try-catch! +std::vector> ZeroMQPoller::ppoll(int timeo, const sigset_t *sigmask_) +{ + if (m_items.empty()) { + throw std::runtime_error("No sockets registered"); + } + + std::vector> r; + + auto n = ZMQ::ppoll(m_items, timeo, sigmask_); + if (n == 0) + return r; + + for (auto &m_item : m_items) { + size_t index = 0; + int flags = 0; + if (m_item.socket == nullptr) { + // an fd was registered + std::tie(index, flags) = m_fds[m_item.fd]; + } else { + // a socket was registered + const zmq::socket_t *s; + std::tie(index, flags, s) = m_sockets[m_item.socket]; + } + if (m_item.revents & short(flags)) { + r.emplace_back(index, flags); + } + } + return r; +} + +size_t ZeroMQPoller::size() const +{ + return m_items.size(); +} + +size_t ZeroMQPoller::register_socket(zmq::socket_t &socket, zmq::PollType type) +{ + zmq::socket_t *s = &socket; + auto it = m_sockets.find(s); + if (it != m_sockets.end()) { + return std::get<0>(it->second); + } + size_t index = m_free.empty() ? m_items.size() : m_free.front(); + if (!m_free.empty()) + m_free.pop_front(); + // NOTE: tis uses the conversion-to-void* operator of + // zmq::socket_t, which returns the wrapped object + m_items.push_back({socket, 0, type, 0}); + + // We need to lookup by the pointer to the object wrapped by zmq::socket_t + m_sockets.emplace(m_items.back().socket, std::make_tuple(index, type, s)); + return index; +} + +size_t ZeroMQPoller::register_socket(int fd, zmq::PollType type) +{ + auto it = m_fds.find(fd); + if (it != m_fds.end()) { + return std::get<0>(it->second); + } + size_t index = m_free.empty() ? m_items.size() : m_free.front(); + if (!m_free.empty()) + m_free.pop_front(); + // NOTE: tis uses the conversion-to-void* operator of + // zmq::socket_t, which returns the wrapped object + m_items.push_back({nullptr, fd, type, 0}); + + // We need to lookup by the pointer to the object wrapped by zmq::socket_t + m_fds.emplace(fd, std::make_tuple(index, type)); + return index; +} + +size_t ZeroMQPoller::unregister_socket(zmq::socket_t &socket) +{ + if (!m_sockets.count(socket.operator void *())) { + throw std::out_of_range("Socket is not registered"); + } + // Remove from m_sockets + // Can't search by the key of m_sockets, as that is the wrapped + // object, but have to use the pointer to the wrapper + // (zmq::socket_t) + auto it = std::find_if(begin(m_sockets), end(m_sockets), [&socket](const decltype(m_sockets)::value_type &entry) { + return &socket == std::get<2>(entry.second); + }); + auto index = std::get<0>(it->second); + m_free.push_back(index); + m_sockets.erase(it); + + // Remove from m_items + auto iit = std::find_if(begin(m_items), end(m_items), + [&it](const zmq::pollitem_t &item) { return it->first == item.socket; }); + assert(iit != end(m_items)); + m_items.erase(iit); + + return index; +} + +size_t ZeroMQPoller::unregister_socket(int fd) +{ + if (!m_fds.count(fd)) { + throw std::out_of_range("fileno is not registered"); + } + // Remove from m_fds + auto it = m_fds.find(fd); + auto index = std::get<0>(it->second); + m_free.push_back(index); + m_fds.erase(it); + + // Remove from m_items + auto iit = + std::find_if(begin(m_items), end(m_items), [&it](const zmq::pollitem_t &item) { return it->first == item.fd; }); + assert(iit != end(m_items)); + m_items.erase(iit); + + return index; +} diff --git a/roofit/roofitZMQ/src/ZeroMQSvc.cpp b/roofit/roofitZMQ/src/ZeroMQSvc.cpp new file mode 100644 index 0000000000000..9c37d57b5fc4d --- /dev/null +++ b/roofit/roofitZMQ/src/ZeroMQSvc.cpp @@ -0,0 +1,114 @@ +#include "RooFit_ZMQ/ZeroMQSvc.h" + +#include // std::ref +#include + +ZeroMQSvc &zmqSvc() +{ + static std::unique_ptr svc; + if (!svc) { + svc = std::make_unique(); + } + return *svc; +} + +ZeroMQSvc::Encoding ZeroMQSvc::encoding() const +{ + return m_enc; +} + +void ZeroMQSvc::setEncoding(const ZeroMQSvc::Encoding &e) +{ + m_enc = e; +} + +zmq::context_t &ZeroMQSvc::context() const +{ + if (!m_context) { + try { + m_context = new zmq::context_t; + } catch (zmq::error_t &e) { + std::cerr << "ERROR: Creating ZeroMQ context failed. This only happens when PGM initialization failed or when " + "a nullptr was returned from zmq_ctx_new because the created context was invalid. Contact ZMQ " + "experts when this happens, because it shouldn't.\n"; + throw; + } + } + return *m_context; +} + +zmq::socket_t ZeroMQSvc::socket(int type) const +{ + try { + // the actual work this function should do, all the rest is error handling: + return zmq::socket_t{context(), type}; + } catch (zmq::error_t &e) { + // all zmq errors not recoverable from here, only at call site + std::cerr << "ERROR in ZeroMQSvc::socket: " << e.what() << " (errno: " << e.num() << ")\n"; + throw; + } +} + +zmq::socket_t *ZeroMQSvc::socket_ptr(int type) const +{ + try { + // the actual work this function should do, all the rest is error handling: + return new zmq::socket_t(context(), type); + } catch (zmq::error_t &e) { + // all zmq errors not recoverable from here, only at call site + std::cerr << "ERROR in ZeroMQSvc::socket_ptr: " << e.what() << " (errno: " << e.num() << ")\n"; + throw; + } +} + +void ZeroMQSvc::close_context() const +{ + if (m_context) { + delete m_context; + m_context = nullptr; + } +} + +zmq::message_t ZeroMQSvc::encode(const char *item) const +{ + std::function fun = ZMQ::stringLength; + return encode(*item, fun); +} + +zmq::message_t ZeroMQSvc::encode(const std::string &item) const +{ + return encode(item.c_str()); +} + +// zmq::message_t ZeroMQSvc::encode(const TObject& item) const { +// auto deleteBuffer = []( void* data, void* /* hint */ ) -> void { +// delete [] (char*)data; +// }; +// +// TBufferFile buffer(TBuffer::kWrite); +// buffer.WriteObject(&item); +// +// // Create message and detach buffer +// // This is the only ZMQ thing that can throw, and only when memory ran out +// // (errno ENOMEM), and that is something only the caller can fix, so we don't +// // catch it here: +// zmq::message_t message(buffer.Buffer(), buffer.Length(), deleteBuffer); +// buffer.DetachBuffer(); +// +// return message; +//} + +bool ZeroMQSvc::send(zmq::socket_t &socket, const char *item, int flags) const +{ + return retry_send(socket, 2, encode(item), flags); +} + +bool ZeroMQSvc::send(zmq::socket_t &socket, zmq::message_t &msg, int flags) const +{ + return retry_send(socket, 2, std::ref(msg), flags); +} + +bool ZeroMQSvc::send(zmq::socket_t &socket, zmq::message_t &&msg, int flags) const +{ + return retry_send(socket, 2, std::move(msg), flags); +} diff --git a/roofit/roofitZMQ/src/functions.cpp b/roofit/roofitZMQ/src/functions.cpp new file mode 100644 index 0000000000000..d43e79b79624d --- /dev/null +++ b/roofit/roofitZMQ/src/functions.cpp @@ -0,0 +1,11 @@ +#include "RooFit_ZMQ/functions.h" + +#include +#include "RooFit_ZMQ/ZeroMQSvc.h" + +namespace ZMQ { +size_t stringLength(const char &cs) +{ + return strlen(&cs); +} +} // namespace ZMQ diff --git a/roofit/roofitZMQ/src/ppoll.cpp b/roofit/roofitZMQ/src/ppoll.cpp new file mode 100644 index 0000000000000..32372698dbc2b --- /dev/null +++ b/roofit/roofitZMQ/src/ppoll.cpp @@ -0,0 +1,306 @@ +#include + +#include // pselect +#include // timespec +#include + +#include "likely.hpp" // unlikely +#include "clock.hpp" // zmq::clock_t +#include "err.hpp" // zmq_assert +#include "polling_util.hpp" // zmq::optimized_fd_set_t, zmq::compute_timeout, zmq::timeout_t +#include "fd.hpp" // zmq::fd_t + +namespace ZMQ { + +int zmq_ppoll(zmq_pollitem_t *items_, int nitems_, long timeout_, const sigset_t *sigmask_) +{ +#if defined ZMQ_HAVE_POLLER + // if poller is present, use that if there is at least 1 thread-safe socket, + // otherwise fall back to the previous implementation as it's faster. + for (int i = 0; i != nitems_; i++) { + if (items_[i].socket && as_socket_base_t(items_[i].socket)->is_thread_safe()) { + return zmq_poller_poll(items_, nitems_, timeout_); + } + } +#endif // ZMQ_HAVE_POLLER +#if defined ZMQ_POLL_BASED_ON_POLL || defined ZMQ_POLL_BASED_ON_SELECT + if (unlikely(nitems_ < 0)) { + errno = EINVAL; + return -1; + } + if (unlikely(nitems_ == 0)) { + if (timeout_ == 0) + return 0; +#if defined ZMQ_HAVE_WINDOWS + Sleep(timeout_ > 0 ? timeout_ : INFINITE); + return 0; +#elif defined ZMQ_HAVE_VXWORKS + struct timespec ns_; + ns_.tv_sec = timeout_ / 1000; + ns_.tv_nsec = timeout_ % 1000 * 1000000; + return nanosleep(&ns_, 0); +#else + return usleep(timeout_ * 1000); +#endif + } + if (!items_) { + errno = EFAULT; + return -1; + } + + zmq::clock_t clock; + uint64_t now = 0; + uint64_t end = 0; +#if defined ZMQ_POLL_BASED_ON_POLL + zmq::fast_vector_t pollfds(nitems_); + + // Build pollset for poll () system call. + for (int i = 0; i != nitems_; i++) { + // If the poll item is a 0MQ socket, we poll on the file descriptor + // retrieved by the ZMQ_FD socket option. + if (items_[i].socket) { + size_t zmq_fd_size = sizeof(zmq::fd_t); + if (zmq_getsockopt(items_[i].socket, ZMQ_FD, &pollfds[i].fd, &zmq_fd_size) == -1) { + return -1; + } + pollfds[i].events = items_[i].events ? POLLIN : 0; + } + // Else, the poll item is a raw file descriptor. Just convert the + // events to normal POLLIN/POLLOUT for poll (). + else { + pollfds[i].fd = items_[i].fd; + pollfds[i].events = (items_[i].events & ZMQ_POLLIN ? POLLIN : 0) | + (items_[i].events & ZMQ_POLLOUT ? POLLOUT : 0) | + (items_[i].events & ZMQ_POLLPRI ? POLLPRI : 0); + } + } +#else + // Ensure we do not attempt to select () on more than FD_SETSIZE + // file descriptors. + // TODO since this function is called by a client, we could return errno EINVAL/ENOMEM/... here + zmq_assert(nitems_ <= FD_SETSIZE); + + zmq::optimized_fd_set_t pollset_in(nitems_); + FD_ZERO(pollset_in.get()); + zmq::optimized_fd_set_t pollset_out(nitems_); + FD_ZERO(pollset_out.get()); + zmq::optimized_fd_set_t pollset_err(nitems_); + FD_ZERO(pollset_err.get()); + + zmq::fd_t maxfd = 0; + + // Build the fd_sets for passing to select (). + for (int i = 0; i != nitems_; i++) { + // If the poll item is a 0MQ socket we are interested in input on the + // notification file descriptor retrieved by the ZMQ_FD socket option. + if (items_[i].socket) { + size_t zmq_fd_size = sizeof(zmq::fd_t); + zmq::fd_t notify_fd; + if (zmq_getsockopt(items_[i].socket, ZMQ_FD, ¬ify_fd, &zmq_fd_size) == -1) + return -1; + if (items_[i].events) { + FD_SET(notify_fd, pollset_in.get()); + if (maxfd < notify_fd) + maxfd = notify_fd; + } + } + // Else, the poll item is a raw file descriptor. Convert the poll item + // events to the appropriate fd_sets. + else { + if (items_[i].events & ZMQ_POLLIN) + FD_SET(items_[i].fd, pollset_in.get()); + if (items_[i].events & ZMQ_POLLOUT) + FD_SET(items_[i].fd, pollset_out.get()); + if (items_[i].events & ZMQ_POLLERR) + FD_SET(items_[i].fd, pollset_err.get()); + if (maxfd < items_[i].fd) + maxfd = items_[i].fd; + } + } + + zmq::optimized_fd_set_t inset(nitems_); + zmq::optimized_fd_set_t outset(nitems_); + zmq::optimized_fd_set_t errset(nitems_); +#endif + + bool first_pass = true; + int nevents = 0; + + while (true) { +#if defined ZMQ_POLL_BASED_ON_POLL + + // Compute the timeout for the subsequent poll. + zmq::timeout_t timeout = zmq::compute_timeout(first_pass, timeout_, now, end); + + // Wait for events. + { + int rc = ppoll(&pollfds[0], nitems_, timeout); + if (rc == -1 && errno == EINTR) { + return -1; + } + errno_assert(rc >= 0); + } + // Check for the events. + for (int i = 0; i != nitems_; i++) { + items_[i].revents = 0; + + // The poll item is a 0MQ socket. Retrieve pending events + // using the ZMQ_EVENTS socket option. + if (items_[i].socket) { + size_t zmq_events_size = sizeof(uint32_t); + uint32_t zmq_events; + if (zmq_getsockopt(items_[i].socket, ZMQ_EVENTS, &zmq_events, &zmq_events_size) == -1) { + return -1; + } + if ((items_[i].events & ZMQ_POLLOUT) && (zmq_events & ZMQ_POLLOUT)) + items_[i].revents |= ZMQ_POLLOUT; + if ((items_[i].events & ZMQ_POLLIN) && (zmq_events & ZMQ_POLLIN)) + items_[i].revents |= ZMQ_POLLIN; + } + // Else, the poll item is a raw file descriptor, simply convert + // the events to zmq_pollitem_t-style format. + else { + if (pollfds[i].revents & POLLIN) + items_[i].revents |= ZMQ_POLLIN; + if (pollfds[i].revents & POLLOUT) + items_[i].revents |= ZMQ_POLLOUT; + if (pollfds[i].revents & POLLPRI) + items_[i].revents |= ZMQ_POLLPRI; + if (pollfds[i].revents & ~(POLLIN | POLLOUT | POLLPRI)) + items_[i].revents |= ZMQ_POLLERR; + } + + if (items_[i].revents) + nevents++; + } + +#else + + // Compute the timeout for the subsequent poll. + struct timespec timeout; + struct timespec *ptimeout; + if (first_pass) { + timeout.tv_sec = 0; + timeout.tv_nsec = 0; + ptimeout = &timeout; + } else if (timeout_ < 0) + ptimeout = NULL; + else { + timeout.tv_sec = static_cast((end - now) / 1000); + timeout.tv_nsec = static_cast((end - now) % 1000 * 1000000); + ptimeout = &timeout; + } + + // Wait for events. Ignore interrupts if there's infinite timeout. + while (true) { + memcpy(inset.get(), pollset_in.get(), zmq::valid_pollset_bytes(*pollset_in.get())); + memcpy(outset.get(), pollset_out.get(), zmq::valid_pollset_bytes(*pollset_out.get())); + memcpy(errset.get(), pollset_err.get(), zmq::valid_pollset_bytes(*pollset_err.get())); +#if defined ZMQ_HAVE_WINDOWS + int rc = select(0, inset.get(), outset.get(), errset.get(), ptimeout); + if (unlikely(rc == SOCKET_ERROR)) { + errno = zmq::wsa_error_to_errno(WSAGetLastError()); + wsa_assert(errno == ENOTSOCK); + return -1; + } +#else + int rc = pselect(maxfd + 1, inset.get(), outset.get(), errset.get(), ptimeout, sigmask_); + if (unlikely(rc == -1)) { + errno_assert(errno == EINTR || errno == EBADF); + return -1; + } +#endif + break; + } + + // Check for the events. + for (int i = 0; i != nitems_; i++) { + items_[i].revents = 0; + + // The poll item is a 0MQ socket. Retrieve pending events + // using the ZMQ_EVENTS socket option. + if (items_[i].socket) { + size_t zmq_events_size = sizeof(uint32_t); + uint32_t zmq_events; + if (zmq_getsockopt(items_[i].socket, ZMQ_EVENTS, &zmq_events, &zmq_events_size) == -1) + return -1; + if ((items_[i].events & ZMQ_POLLOUT) && (zmq_events & ZMQ_POLLOUT)) + items_[i].revents |= ZMQ_POLLOUT; + if ((items_[i].events & ZMQ_POLLIN) && (zmq_events & ZMQ_POLLIN)) + items_[i].revents |= ZMQ_POLLIN; + } + // Else, the poll item is a raw file descriptor, simply convert + // the events to zmq_pollitem_t-style format. + else { + if (FD_ISSET(items_[i].fd, inset.get())) + items_[i].revents |= ZMQ_POLLIN; + if (FD_ISSET(items_[i].fd, outset.get())) + items_[i].revents |= ZMQ_POLLOUT; + if (FD_ISSET(items_[i].fd, errset.get())) + items_[i].revents |= ZMQ_POLLERR; + } + + if (items_[i].revents) + nevents++; + } +#endif + + // If timeout is zero, exit immediately whether there are events or not. + if (timeout_ == 0) + break; + + // If there are events to return, we can exit immediately. + if (nevents) + break; + + // At this point we are meant to wait for events but there are none. + // If timeout is infinite we can just loop until we get some events. + if (timeout_ < 0) { + if (first_pass) + first_pass = false; + continue; + } + + // The timeout is finite and there are no events. In the first pass + // we get a timestamp of when the polling have begun. (We assume that + // first pass have taken negligible time). We also compute the time + // when the polling should time out. + if (first_pass) { + now = clock.now_ms(); + end = now + timeout_; + if (now == end) + break; + first_pass = false; + continue; + } + + // Find out whether timeout have expired. + now = clock.now_ms(); + if (now >= end) + break; + } + + return nevents; +#else + // Exotic platforms that support neither poll() nor select(). + errno = ENOTSUP; + return -1; +#endif +} + +// This function can throw, so wrap in try-catch! +int ppoll(zmq_pollitem_t *items_, size_t nitems_, long timeout_, const sigset_t *sigmask_) +{ + int rc = zmq_ppoll(items_, static_cast(nitems_), timeout_, sigmask_); + if (rc < 0) + throw ppoll_error_t(); + return rc; +} + +// This function can throw, so wrap in try-catch! +int ppoll(std::vector &items, long timeout_, const sigset_t *sigmask_) +{ + return ppoll(items.data(), items.size(), timeout_, sigmask_); +} + +} // namespace ZMQ diff --git a/roofit/roofitZMQ/test/CMakeLists.txt b/roofit/roofitZMQ/test/CMakeLists.txt new file mode 100644 index 0000000000000..b1da06f39da1f --- /dev/null +++ b/roofit/roofitZMQ/test/CMakeLists.txt @@ -0,0 +1,6 @@ +#find_package(ZeroMQ REQUIRED) + +ROOT_ADD_GTEST(test_RooFitZMQ test_ZMQ.cpp LIBRARIES RooFitZMQ ${ZeroMQ_LIBRARY}) +ROOT_ADD_GTEST(test_RooFitZMQ_polling test_polling.cxx LIBRARIES RooFitZMQ ${ZeroMQ_LIBRARY}) +ROOT_ADD_GTEST(test_RooFitZMQ_HWM test_HWM.cxx LIBRARIES RooFitZMQ ${ZeroMQ_LIBRARY}) +ROOT_ADD_GTEST(test_RooFitZMQ_load_balancing test_ZMQ_load_balancing.cxx LIBRARIES RooFitZMQ ${ZeroMQ_LIBRARY}) diff --git a/roofit/roofitZMQ/test/test_HWM.cxx b/roofit/roofitZMQ/test/test_HWM.cxx new file mode 100644 index 0000000000000..c6bfa7b29d4ec --- /dev/null +++ b/roofit/roofitZMQ/test/test_HWM.cxx @@ -0,0 +1,145 @@ +/* + * Project: RooFit + * Authors: + * PB, Patrick Bos, Netherlands eScience Center, p.bos@esciencecenter.nl + * + * Copyright (c) 2016-2021, Netherlands eScience Center + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + */ + +#include "gtest/gtest.h" + +#include // fork, usleep + +#include "RooFit_ZMQ/ZeroMQSvc.h" + +class HighWaterMarkTest : public ::testing::Test { +protected: + void SetUp() override + { + do { + child_pid = fork(); + } while (child_pid == -1); // retry if fork fails + + if (child_pid > 0) { // parent + pusher.reset(zmqSvc().socket_ptr(zmq::PUSH)); + if (set_hwm) { + auto rc = zmq_setsockopt(*pusher, ZMQ_SNDHWM, &hwm, sizeof hwm); + assert(rc == 0); + } + pusher->bind("ipc:///tmp/ZMQ_test_fork_polling_P2C.ipc"); + } else { // child + puller.reset(zmqSvc().socket_ptr(zmq::PULL)); + if (set_hwm) { + auto rc = zmq_setsockopt(*puller, ZMQ_RCVHWM, &hwm, sizeof hwm); + assert(rc == 0); + } + puller->connect("ipc:///tmp/ZMQ_test_fork_polling_P2C.ipc"); + } + } + + void TearDown() override + { + if (child_pid > 0) { // parent + // wait for child + int status = 0; + pid_t pid; + do { + pid = waitpid(child_pid, &status, 0); + } while (-1 == pid && EINTR == errno); // retry on interrupted system call + if (0 != status) { + if (WIFEXITED(status)) { + printf("exited, status=%d\n", WEXITSTATUS(status)); + } else if (WIFSIGNALED(status)) { + printf("killed by signal %d\n", WTERMSIG(status)); + } else if (WIFSTOPPED(status)) { + printf("stopped by signal %d\n", WSTOPSIG(status)); + } else if (WIFCONTINUED(status)) { + printf("continued\n"); + } + } + if (-1 == pid) { + throw std::runtime_error(std::string("waitpid, errno ") + std::to_string(errno)); + } + pusher.reset(nullptr); + zmqSvc().close_context(); + } else { // child + puller.reset(nullptr); + zmqSvc().close_context(); + } + } + + void run_test() + { + std::size_t max_sends = 2000; + + if (child_pid > 0) { // parent + // start test + for (std::size_t ix = 0; ix < max_sends; ++ix) { + zmqSvc().send(*pusher, 0.1f); + zmqSvc().send(*pusher, 1); + zmqSvc().send(*pusher, true); + if (ix % 100 == 0) { + printf("parent at ix = %lu\n", ix); + } + } + } else { // child + // wait a few seconds before reading to allow the parent to overflow the HWM + printf("child waiting for 2 seconds...\n"); + sleep(2); + printf("child starts receiving\n"); + + for (std::size_t ix = 0; ix < max_sends; ++ix) { + zmqSvc().receive(*puller); + zmqSvc().receive(*puller); + zmqSvc().receive(*puller); + if (ix % 100 == 0) { + printf("child at ix = %lu\n", ix); + } + } + } + } + + pid_t child_pid{0}; + ZmqLingeringSocketPtr<> pusher, puller; + bool set_hwm = false; + int hwm = 0; +}; + +TEST_F(HighWaterMarkTest, demonstrateHittingDefaultHWM) +{ + run_test(); +} + +class HighWaterMarkZeroTest : public HighWaterMarkTest { + void SetUp() override + { + set_hwm = true; + hwm = 0; + HighWaterMarkTest::SetUp(); + } +}; + +TEST_F(HighWaterMarkZeroTest, zeroHWM) +{ + run_test(); +} + +class HighWaterMark2kTest : public HighWaterMarkTest { + void SetUp() override + { + set_hwm = true; + hwm = 2000; + HighWaterMarkTest::SetUp(); + } +}; + +TEST_F(HighWaterMark2kTest, HWM2k) +{ + run_test(); +} diff --git a/roofit/roofitZMQ/test/test_ZMQ.cpp b/roofit/roofitZMQ/test/test_ZMQ.cpp new file mode 100644 index 0000000000000..d93d230f04a34 --- /dev/null +++ b/roofit/roofitZMQ/test/test_ZMQ.cpp @@ -0,0 +1,308 @@ +/***************************************************************************** + * Project: RooFit * + * Package: RooFitCore * + * @(#)root/roofitcore:$Id$ + * Authors: * + * PB, Patrick Bos, NL eScience Center, p.bos@esciencecenter.nl * + * * + * Redistribution and use in source and binary forms, * + * with or without modification, are permitted according to the terms * + * listed in LICENSE (http://roofit.sourceforge.net/license.txt) * + *****************************************************************************/ + +#include "gtest/gtest.h" + +#include // fork, usleep + +#include + +#include "RooFit_ZMQ/ZeroMQSvc.h" + +// N.B.: wait_for_child is identically defined in RooFit::MultiProcess, but we copy it here to reduce module +// interdependencies. It also requires an extra include: +#include // kill, SIGKILL + +int wait_for_child(pid_t child_pid, bool may_throw, int retries_before_killing) +{ + int status = 0; + int patience = retries_before_killing; + pid_t tmp; + do { + if (patience-- < 1) { + ::kill(child_pid, SIGKILL); + } + tmp = waitpid(child_pid, &status, WNOHANG); + } while (tmp == 0 // child has not yet changed state, try again + || (-1 == tmp && EINTR == errno) // retry on interrupted system call + ); + + if (patience < 1) { + std::cout << "Had to send PID " << child_pid << " " << (-patience + 1) << " SIGKILLs"; + } + + if (0 != status) { + if (WIFEXITED(status)) { + printf("exited, status=%d\n", WEXITSTATUS(status)); + } else if (WIFSIGNALED(status)) { + printf("killed by signal %d\n", WTERMSIG(status)); + } else if (WIFSTOPPED(status)) { + printf("stopped by signal %d\n", WSTOPSIG(status)); + } else if (WIFCONTINUED(status)) { + printf("continued\n"); + } + } + + if (-1 == tmp && may_throw) + throw std::runtime_error(std::string("waitpid, errno ") + std::to_string(errno)); + + return status; +} + +void elaborate_bind(const ZmqLingeringSocketPtr<> &socket, std::string name) +{ + try { + socket->bind(name); + } catch (const zmq::error_t &e) { + if (e.num() == EADDRINUSE) { + std::cerr << "address already in use, retrying bind in 500ms\n"; + usleep(500000); + try { + socket->bind(name); + } catch (const zmq::error_t &e2) { + if (e2.num() == EADDRINUSE) { + std::cerr + << "again: address already in use, aborting; please check whether there are any remaining improperly " + "exited processes (zombies) around or whether some other program is using port 6660\n"; + } + throw; + } + // Sometimes, the socket from the previous test needs some time to close, so + // we introduce a latency here. A more robust and fast approach might be to + // do the following on the bind side: + // 1. first try another port, e.g. increase by one + // 2. if that doesn't work, do the latency and retry the original port + // The connect side then also needs to change, because it doesn't know which + // port the bind side will bind to. The connect side could try connecting to + // both options asynchronously, and then in a loop check both for signs of + // life. If one comes alive, transfer ownership of that pointer to the pointer + // you want to eventually use (`socket`) and that's it. + } else { + throw; + } + } +} + +class AllSocketTypes + : public ::testing::TestWithParam, + std::pair /* socket_names */>> { +}; + +TEST_P(AllSocketTypes, forkHandshake) +{ + auto socket_names = std::get<2>(GetParam()); + pid_t child_pid{0}; + do { + child_pid = fork(); + } while (child_pid == -1); // retry if fork fails + + if (child_pid > 0) { // master + ZmqLingeringSocketPtr<> socket; + socket.reset(zmqSvc().socket_ptr(std::get<1>(GetParam()).first)); + elaborate_bind(socket, socket_names.second); + // bind is on the master process to avoid zombie children to hold on to binds + + // start test + zmqSvc().send(*socket, std::string("breaker breaker")); + + auto receipt = zmqSvc().receive(*socket); + + EXPECT_EQ(receipt, 1212); + + // socket->close(); // this gives exception of type zmq::error_t: Socket operation on non-socket + socket.reset(nullptr); + zmqSvc().close_context(); // if you don't close context in parent process as well, the next repeat will hang + + wait_for_child(child_pid, true, 5); + } else { // child + ZmqLingeringSocketPtr<> socket; + socket.reset(zmqSvc().socket_ptr(std::get<1>(GetParam()).second)); + socket->connect(socket_names.first); + + // start test + auto receipt = zmqSvc().receive(*socket); + if (receipt == "breaker breaker") { + zmqSvc().send(*socket, 1212); + } + // take care, don't just use _exit, it will not cleanly destroy context etc! + // if you really need to, at least close and destroy everything properly + // socket->close(); // this gives exception of type zmq::error_t: Socket operation on non-socket + socket.reset(nullptr); + zmqSvc().close_context(); + _Exit(0); + } +} + +std::string ipc{"ipc:///tmp/ZMQ_test_fork.ipc"}; +std::string tcp_server{"tcp://127.0.0.1:6660"}; +std::string tcp_client{"tcp://*:6660"}; +auto socket_name_options = ::testing::Values(std::make_pair(tcp_server, tcp_client), std::make_pair(ipc, ipc)); + +INSTANTIATE_TEST_SUITE_P(REQREP, AllSocketTypes, + ::testing::Combine(::testing::Range(0, 10), // repeat to probe connection stability + ::testing::Values(std::make_pair(zmq::REQ, zmq::REP)), + socket_name_options)); +INSTANTIATE_TEST_SUITE_P(PAIRPAIR, AllSocketTypes, + ::testing::Combine(::testing::Range(0, 10), // repeat to probe connection stability + ::testing::Values(std::make_pair(zmq::PAIR, zmq::PAIR)), + socket_name_options)); + +class AsyncSocketTypes + : public ::testing::TestWithParam< + std::tuple, + std::pair /* socket_names */, bool /* expect_throw */>> { +}; + +TEST_P(AsyncSocketTypes, forkMultiSendReceive) +{ + bool expect_throw = std::get<3>(GetParam()); + ZmqLingeringSocketPtr<> socket; + auto socket_names = std::get<2>(GetParam()); + pid_t child_pid{0}; + do { + child_pid = fork(); + } while (child_pid == -1); // retry if fork fails + + if (child_pid > 0) { // master + socket.reset(zmqSvc().socket_ptr(std::get<1>(GetParam()).first)); + elaborate_bind(socket, socket_names.second); + // bind is on the master process to avoid zombie children to hold on to binds + + // start test: send 2 things, receive 1, send 1 more, finish + zmqSvc().send(*socket, std::string("breaker breaker")); + + if (expect_throw) { + EXPECT_ANY_THROW(zmqSvc().send(*socket, std::string("anybody out there?"))); + // NOTE: also in case of a throw, be sure to properly close down the connection! + // Otherwise, you may get zombies waiting for a reply. + socket.reset(nullptr); + zmqSvc().close_context(); // if you don't close context in parent process as well, the next repeat will hang + wait_for_child(child_pid, true, 5); + return; + } else { + EXPECT_NO_THROW(zmqSvc().send(*socket, std::string("anybody out there?"))); + } + + auto receipt = zmqSvc().receive(*socket); + + EXPECT_EQ(receipt, 1212); + + zmqSvc().send(*socket, std::string("kthxbye")); + + // socket->close(); // this gives exception of type zmq::error_t: Socket operation on non-socket + socket.reset(nullptr); + zmqSvc().close_context(); // if you don't close context in parent process as well, the next repeat will hang + + wait_for_child(child_pid, true, 5); + } else { // child + socket.reset(zmqSvc().socket_ptr(std::get<1>(GetParam()).second)); + socket->connect(socket_names.first); + + // start test, receive something + auto receipt1 = zmqSvc().receive(*socket); + auto receipt2 = zmqSvc().receive(*socket); + if (receipt1 == "breaker breaker" && receipt2 == "anybody out there?") { + zmqSvc().send(*socket, 1212); + } + auto receipt3 = zmqSvc().receive(*socket); + if (receipt3 != "kthxbye") { + std::cerr << "did not receive final reply correctly\n"; + } + + // take care, don't just use _exit, it will not cleanly destroy context etc! + // if you really need to, at least close and destroy everything properly + // socket->close(); // this gives exception of type zmq::error_t: Socket operation on non-socket + socket.reset(nullptr); + zmqSvc().close_context(); + _Exit(0); + } +} + +TEST_P(AsyncSocketTypes, forkIgnoreSomeMessages) +{ + bool expect_throw = std::get<3>(GetParam()); + ZmqLingeringSocketPtr<> socket; + auto socket_names = std::get<2>(GetParam()); + pid_t child_pid{0}; + do { + child_pid = fork(); + } while (child_pid == -1); // retry if fork fails + + if (child_pid > 0) { // master + socket.reset(zmqSvc().socket_ptr(std::get<1>(GetParam()).first)); + elaborate_bind(socket, socket_names.second); + // bind is on the master process to avoid zombie children to hold on to binds + + // start test: send 2 things, receive 1, send 1 more, finish + zmqSvc().send(*socket, std::string("breaker breaker")); + + if (expect_throw) { + EXPECT_ANY_THROW(zmqSvc().send(*socket, std::string("anybody out there?"))); + // NOTE: also in case of a throw, be sure to properly close down the connection! + // Otherwise, you may get zombies waiting for a reply. + socket.reset(nullptr); + zmqSvc().close_context(); // if you don't close context in parent process as well, the next repeat will hang + wait_for_child(child_pid, true, 5); + return; + } else { + EXPECT_NO_THROW(zmqSvc().send(*socket, std::string("anybody out there?"))); + } + + auto receipt = zmqSvc().receive(*socket); + + EXPECT_EQ(receipt, 1212); + + zmqSvc().send(*socket, std::string("kthxbye")); + + // socket->close(); // this gives exception of type zmq::error_t: Socket operation on non-socket + socket.reset(nullptr); + zmqSvc().close_context(); // if you don't close context in parent process as well, the next repeat will hang + + wait_for_child(child_pid, true, 5); + } else { // child + socket.reset(zmqSvc().socket_ptr(std::get<1>(GetParam()).second)); + socket->connect(socket_names.first); + + // start test, receive first thing + auto receipt = zmqSvc().receive(*socket); + if (receipt == "breaker breaker") { + zmqSvc().send(*socket, 1212); + } + + // ignore the rest of the sent messages, but give the other end a second to + // actually send its stuff, instead of hanging in retry_send because the + // connection has died; a better solution would be if retry_send (in + // ZeroMQSvc::send) had a callback mechanism that could be used to break + // out when a child has died, but ok + sleep(1); + + // take care, don't just use _exit, it will not cleanly destroy context etc! + // if you really need to, at least close and destroy everything properly + // socket->close(); // this gives exception of type zmq::error_t: Socket operation on non-socket + socket.reset(nullptr); + zmqSvc().close_context(); + _Exit(0); + } +} + +INSTANTIATE_TEST_SUITE_P(PAIRPAIR, AsyncSocketTypes, + ::testing::Combine(::testing::Range(0, 10), // repeat to probe connection stability + ::testing::Values(std::make_pair(zmq::PAIR, zmq::PAIR)), + socket_name_options, + ::testing::Values(false) // don't expect throw + )); + +INSTANTIATE_TEST_SUITE_P(REQREP, AsyncSocketTypes, + ::testing::Combine(::testing::Values(0), // no repeats, we only care about the throw + ::testing::Values(std::make_pair(zmq::REQ, zmq::REP)), socket_name_options, + ::testing::Values(true) // expect throw + )); diff --git a/roofit/roofitZMQ/test/test_ZMQ_load_balancing.cxx b/roofit/roofitZMQ/test/test_ZMQ_load_balancing.cxx new file mode 100644 index 0000000000000..1acbd44117eab --- /dev/null +++ b/roofit/roofitZMQ/test/test_ZMQ_load_balancing.cxx @@ -0,0 +1,149 @@ +/* + * Project: RooFit + * Authors: + * PB, Patrick Bos, Netherlands eScience Center, p.bos@esciencecenter.nl + * + * Copyright (c) 2016-2021, Netherlands eScience Center + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + */ + +#include "gtest/gtest.h" + +#include // fork, usleep + +#include "RooFit_ZMQ/ZeroMQSvc.h" +#include "RooFit_ZMQ/ZeroMQPoller.h" + +class ZMQPushPullTest : public ::testing::Test { +protected: + std::size_t N_children = 4; + std::size_t max_sends = 20; + + void SetUp() override + { + for (std::size_t i = 0; i < N_children; ++i) { + do { + child_pid = fork(); + } while (child_pid == -1); // retry if fork fails + if (child_pid == 0) { // child + child_id = i; + break; + } else { + child_pids.push_back(child_pid); + } + } + + if (child_pid > 0) { // parent + pusher.reset(zmqSvc().socket_ptr(zmq::PUSH)); + pusher->bind("ipc:///tmp/ZMQ_test_push_pull_P2C.ipc"); + } else { // child + puller.reset(zmqSvc().socket_ptr(zmq::PULL)); + puller->connect("ipc:///tmp/ZMQ_test_push_pull_P2C.ipc"); + + poller.register_socket(*puller, zmq::POLLIN); + } + } + + void TearDown() override + { + if (child_pid > 0) { // parent + // wait for children + int status = 0; + pid_t pid; + for (pid_t child_pid_i : child_pids) { + do { + pid = waitpid(child_pid_i, &status, 0); + } while (-1 == pid && EINTR == errno); // retry on interrupted system call + if (0 != status) { + if (WIFEXITED(status)) { + printf("exited, status=%d\n", WEXITSTATUS(status)); + } else if (WIFSIGNALED(status)) { + printf("killed by signal %d\n", WTERMSIG(status)); + } else if (WIFSTOPPED(status)) { + printf("stopped by signal %d\n", WSTOPSIG(status)); + } else if (WIFCONTINUED(status)) { + printf("continued\n"); + } + } + if (-1 == pid) { + throw std::runtime_error(std::string("waitpid, errno ") + std::to_string(errno)); + } + } + pusher.reset(nullptr); + zmqSvc().close_context(); + } else { // child + puller.reset(nullptr); + zmqSvc().close_context(); + } + } + + void run_parent() + { + // start test + usleep(1000); // wait a second so that all pull sockets are connected for round-robin distribution + // if you don't wait a second above, the push socket will "round-robin" all the messages to just one or two + // connected sockets + for (std::size_t ix = 0; ix < max_sends; ++ix) { + zmqSvc().send(*pusher, 0); + } + for (std::size_t ix = 0; ix < N_children; ++ix) { + // end by sending some 1's to all children, to let them know the sending is over + zmqSvc().send(*pusher, 1); + } + } + + void run_child() + { + std::size_t count = 0; + for (std::size_t ix = 0; ix < max_sends; ++ix) { + auto r = poller.poll(2000); + if (r.empty()) { + printf("poller of child %d timed out after 2 seconds\n", child_id); + break; + } + auto value = zmqSvc().receive(*puller, ZMQ_DONTWAIT); + usleep(200); // "do some work" + printf("value on child %d: %d\n", child_id, value); + if (value == 1) { + printf("child %d got value %d, done here\n", child_id, value); + break; + } + ++count; + } + printf("child %d got %lu values\n", child_id, count); + } + + pid_t child_pid{0}; + int child_id = -1; + std::vector child_pids; + ZmqLingeringSocketPtr<> pusher, puller; + ZeroMQPoller poller; +}; + +/// This test shows how push-pull is unsuited for load balancing; messages are just sent to the first available pull +/// socket without any dynamic load balancing +TEST_F(ZMQPushPullTest, demoRoundRobin) +{ + if (child_pid > 0) { + run_parent(); + } else { + run_child(); + } +} + +/// This test tries to see whether push-pull can be made to work as a bit of a load balancer, using a low HWM at the +/// receiver +TEST_F(ZMQPushPullTest, demoHWM1LoadBalancing) +{ + if (child_pid > 0) { + run_parent(); + } else { + puller->setsockopt(ZMQ_RCVHWM, 1); + run_child(); + } +} diff --git a/roofit/roofitZMQ/test/test_polling.cxx b/roofit/roofitZMQ/test/test_polling.cxx new file mode 100644 index 0000000000000..9f6b889c85a9c --- /dev/null +++ b/roofit/roofitZMQ/test/test_polling.cxx @@ -0,0 +1,156 @@ +/* + * Project: RooFit + * Authors: + * PB, Patrick Bos, Netherlands eScience Center, p.bos@esciencecenter.nl + * + * Copyright (c) 2016-2019, Netherlands eScience Center + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + */ + +#include "gtest/gtest.h" + +#include // fork, usleep +#include // signal blocking +#include // strsignal() + +#include + +#include "RooFit_ZMQ/ZeroMQSvc.h" +#include "RooFit_ZMQ/ZeroMQPoller.h" + +static bool terminated = false; + +void handle_sigterm(int signum) +{ + terminated = true; + std::cout << "handled signal " << strsignal(signum) << " on PID " << getpid() << std::endl; +} + +TEST(Polling, doublePoll) +{ + pid_t child_pid{0}; + do { + child_pid = fork(); + } while (child_pid == -1); // retry if fork fails + + if (child_pid > 0) { // master + sigset_t sigmask, sigmask_old; + sigemptyset(&sigmask); + sigaddset(&sigmask, SIGCHLD); + sigprocmask(SIG_BLOCK, &sigmask, &sigmask_old); + + // std::cout << "master PID: " << getpid() << std::endl; + ZmqLingeringSocketPtr<> pusher, puller; + pusher.reset(zmqSvc().socket_ptr(zmq::PUSH)); + pusher->bind("ipc:///tmp/ZMQ_test_fork_polling_M2C.ipc"); + puller.reset(zmqSvc().socket_ptr(zmq::PULL)); + puller->bind("ipc:///tmp/ZMQ_test_fork_polling_C2M.ipc"); + + ZeroMQPoller poller1, poller2; + poller1.register_socket(*puller, zmq::POLLIN); + poller2.register_socket(*puller, zmq::POLLIN); + + // start test + zmqSvc().send(*pusher, std::string("breaker breaker")); + + auto result1a = poller1.poll(-1); + auto result1b = poller1.poll(-1); + auto result2 = poller2.poll(-1); + EXPECT_EQ(result1a.size(), result1b.size()); + EXPECT_EQ(result1a.size(), result2.size()); + + auto receipt = zmqSvc().receive(*puller, ZMQ_DONTWAIT); + + EXPECT_EQ(receipt, 1212); + + kill(child_pid, SIGTERM); + + // socket->close(); // this gives exception of type zmq::error_t: Socket operation on non-socket + pusher.reset(nullptr); + puller.reset(nullptr); + zmqSvc().close_context(); // if you don't close context in parent process as well, the next repeat will hang + + // wait for child + int status = 0; + pid_t pid; + do { + pid = waitpid(child_pid, &status, 0); + } while (-1 == pid && EINTR == errno); // retry on interrupted system call + + if (0 != status) { + if (WIFEXITED(status)) { + printf("exited, status=%d\n", WEXITSTATUS(status)); + } else if (WIFSIGNALED(status)) { + printf("killed by signal %d\n", WTERMSIG(status)); + } else if (WIFSTOPPED(status)) { + printf("stopped by signal %d\n", WSTOPSIG(status)); + } else if (WIFCONTINUED(status)) { + printf("continued\n"); + } + } + + if (-1 == pid) { + throw std::runtime_error(std::string("waitpid, errno ") + std::to_string(errno)); + } + + sigprocmask(SIG_SETMASK, &sigmask_old, nullptr); + } else { // child + // std::cout << "child PID: " << getpid() << std::endl; + + sigset_t sigmask, sigmask_old; + sigemptyset(&sigmask); + sigaddset(&sigmask, SIGTERM); + sigprocmask(SIG_BLOCK, &sigmask, &sigmask_old); + + struct sigaction sa; + memset(&sa, '\0', sizeof(sa)); + sa.sa_handler = handle_sigterm; + + if (sigaction(SIGTERM, &sa, NULL) < 0) { + std::perror("sigaction failed"); + std::exit(1); + } + + ZmqLingeringSocketPtr<> puller, pusher; + puller.reset(zmqSvc().socket_ptr(zmq::PULL)); + puller->connect("ipc:///tmp/ZMQ_test_fork_polling_M2C.ipc"); + pusher.reset(zmqSvc().socket_ptr(zmq::PUSH)); + pusher->connect("ipc:///tmp/ZMQ_test_fork_polling_C2M.ipc"); + + ZeroMQPoller poller1, poller2; + poller1.register_socket(*puller, zmq::POLLIN); + poller2.register_socket(*puller, zmq::POLLIN); + + // start test + auto result1a = poller1.poll(-1); + auto result1b = poller1.poll(-1); + auto result2 = poller2.poll(-1); + EXPECT_EQ(result1a.size(), result1b.size()); + EXPECT_EQ(result1a.size(), result2.size()); + + auto receipt = zmqSvc().receive(*puller, ZMQ_DONTWAIT); + if (receipt == "breaker breaker") { + zmqSvc().send(*pusher, 1212); + } + // take care, don't just use _exit, it will not cleanly destroy context etc! + // if you really need to, at least close and destroy everything properly + // socket->close(); // this gives exception of type zmq::error_t: Socket operation on non-socket + + sigprocmask(SIG_SETMASK, &sigmask_old, nullptr); + + while (!terminated) { + } + + // std::cout << "child terminated" << std::endl; + + puller.reset(nullptr); + pusher.reset(nullptr); + zmqSvc().close_context(); + _Exit(0); + } +}