diff options
Diffstat (limited to 'src/rpc/zmq_server.cpp')
| -rw-r--r-- | src/rpc/zmq_server.cpp | 171 |
1 files changed, 126 insertions, 45 deletions
diff --git a/src/rpc/zmq_server.cpp b/src/rpc/zmq_server.cpp index 1a9f49c01..4028df96a 100644 --- a/src/rpc/zmq_server.cpp +++ b/src/rpc/zmq_server.cpp @@ -29,10 +29,16 @@ #include "zmq_server.h" #include <chrono> -#include <cstdint> +#include <cstring> +#include <utility> +#include <stdexcept> #include <system_error> #include "byte_slice.h" +#include "rpc/zmq_pub.h" + +#undef MONERO_DEFAULT_LOG_CATEGORY +#define MONERO_DEFAULT_LOG_CATEGORY "net.zmq" namespace cryptonote { @@ -42,14 +48,57 @@ namespace constexpr const int num_zmq_threads = 1; constexpr const std::int64_t max_message_size = 10 * 1024 * 1024; // 10 MiB constexpr const std::chrono::seconds linger_timeout{2}; // wait period for pending out messages -} + + net::zmq::socket init_socket(void* context, int type, epee::span<const std::string> addresses) + { + if (context == nullptr) + throw std::logic_error{"NULL context provided"}; + + net::zmq::socket out{}; + out.reset(zmq_socket(context, type)); + if (!out) + { + MONERO_LOG_ZMQ_ERROR("Failed to create ZMQ socket"); + return nullptr; + } + + if (zmq_setsockopt(out.get(), ZMQ_MAXMSGSIZE, std::addressof(max_message_size), sizeof(max_message_size)) != 0) + { + MONERO_LOG_ZMQ_ERROR("Failed to set maximum incoming message size"); + return nullptr; + } + + static constexpr const int linger_value = std::chrono::milliseconds{linger_timeout}.count(); + if (zmq_setsockopt(out.get(), ZMQ_LINGER, std::addressof(linger_value), sizeof(linger_value)) != 0) + { + MONERO_LOG_ZMQ_ERROR("Failed to set linger timeout"); + return nullptr; + } + + for (const std::string& address : addresses) + { + if (zmq_bind(out.get(), address.c_str()) < 0) + { + MONERO_LOG_ZMQ_ERROR("ZMQ bind failed"); + return nullptr; + } + MINFO("ZMQ now listening at " << address); + } + + return out; + } +} // anonymous namespace rpc { ZmqServer::ZmqServer(RpcHandler& h) : handler(h), - context(zmq_init(num_zmq_threads)) + context(zmq_init(num_zmq_threads)), + rep_socket(nullptr), + pub_socket(nullptr), + relay_socket(nullptr), + shared_state(nullptr) { if (!context) MONERO_ZMQ_THROW("Unable to create ZMQ context"); @@ -64,22 +113,59 @@ void ZmqServer::serve() try { // socket must close before `zmq_term` will exit. - const net::zmq::socket socket = std::move(rep_socket); - if (!socket) + const net::zmq::socket rep = std::move(rep_socket); + const net::zmq::socket pub = std::move(pub_socket); + const net::zmq::socket relay = std::move(relay_socket); + const std::shared_ptr<listener::zmq_pub> state = std::move(shared_state); + + const unsigned init_count = unsigned(bool(pub)) + bool(relay) + bool(state); + if (!rep || (init_count && init_count != 3)) { - MERROR("ZMQ RPC server reply socket is null"); + MERROR("ZMQ RPC server socket is null"); return; } + MINFO("ZMQ Server started"); + + const int read_flags = pub ? ZMQ_DONTWAIT : 0; + std::array<zmq_pollitem_t, 3> sockets = + {{ + {relay.get(), 0, ZMQ_POLLIN, 0}, + {pub.get(), 0, ZMQ_POLLIN, 0}, + {rep.get(), 0, ZMQ_POLLIN, 0} + }}; + + /* This uses XPUB to watch for subscribers, to reduce CPU cycles for + serialization when the data will be dropped. This is important for block + serialization, which is done on the p2p threads currently (see + zmq_pub.cpp). + + XPUB sockets are not thread-safe, so the p2p thread cannot write into + the socket while we read here for subscribers. A ZMQ_PAIR socket is + used for inproc notification. No data is every copied to kernel, it is + all userspace messaging. */ + while (1) { - const std::string message = MONERO_UNWRAP(net::zmq::receive(socket.get())); - MDEBUG("Received RPC request: \"" << message << "\""); - epee::byte_slice response = handler.handle(message); + if (pub) + MONERO_UNWRAP(net::zmq::retry_op(zmq_poll, sockets.data(), sockets.size(), -1)); - const boost::string_ref response_view{reinterpret_cast<const char*>(response.data()), response.size()}; - MDEBUG("Sending RPC reply: \"" << response_view << "\""); - MONERO_UNWRAP(net::zmq::send(std::move(response), socket.get())); + if (sockets[0].revents) + state->relay_to_pub(relay.get(), pub.get()); + + if (sockets[1].revents) + state->sub_request(MONERO_UNWRAP(net::zmq::receive(pub.get(), ZMQ_DONTWAIT))); + + if (!pub || sockets[2].revents) + { + std::string message = MONERO_UNWRAP(net::zmq::receive(rep.get(), read_flags)); + MDEBUG("Received RPC request: \"" << message << "\""); + epee::byte_slice response = handler.handle(std::move(message)); + + const boost::string_ref response_view{reinterpret_cast<const char*>(response.data()), response.size()}; + MDEBUG("Sending RPC reply: \"" << response_view << "\""); + MONERO_UNWRAP(net::zmq::send(std::move(response), rep.get())); + } } } catch (const std::system_error& e) @@ -97,38 +183,12 @@ void ZmqServer::serve() } } -bool ZmqServer::addIPCSocket(const boost::string_ref address, const boost::string_ref port) -{ - MERROR("ZmqServer::addIPCSocket not yet implemented!"); - return false; -} - -bool ZmqServer::addTCPSocket(boost::string_ref address, boost::string_ref port) +void* ZmqServer::init_rpc(boost::string_ref address, boost::string_ref port) { if (!context) { MERROR("ZMQ RPC Server already shutdown"); - return false; - } - - rep_socket.reset(zmq_socket(context.get(), ZMQ_REP)); - if (!rep_socket) - { - MONERO_LOG_ZMQ_ERROR("ZMQ RPC Server socket create failed"); - return false; - } - - if (zmq_setsockopt(rep_socket.get(), ZMQ_MAXMSGSIZE, std::addressof(max_message_size), sizeof(max_message_size)) != 0) - { - MONERO_LOG_ZMQ_ERROR("Failed to set maximum incoming message size"); - return false; - } - - static constexpr const int linger_value = std::chrono::milliseconds{linger_timeout}.count(); - if (zmq_setsockopt(rep_socket.get(), ZMQ_LINGER, std::addressof(linger_value), sizeof(linger_value)) != 0) - { - MONERO_LOG_ZMQ_ERROR("Failed to set linger timeout"); - return false; + return nullptr; } if (address.empty()) @@ -141,12 +201,34 @@ bool ZmqServer::addTCPSocket(boost::string_ref address, boost::string_ref port) bind_address += ":"; bind_address.append(port.data(), port.size()); - if (zmq_bind(rep_socket.get(), bind_address.c_str()) < 0) + rep_socket = init_socket(context.get(), ZMQ_REP, {std::addressof(bind_address), 1}); + return bool(rep_socket) ? context.get() : nullptr; +} + +std::shared_ptr<listener::zmq_pub> ZmqServer::init_pub(epee::span<const std::string> addresses) +{ + try + { + shared_state = std::make_shared<listener::zmq_pub>(context.get()); + pub_socket = init_socket(context.get(), ZMQ_XPUB, addresses); + if (!pub_socket) + throw std::runtime_error{"Unable to initialize ZMQ_XPUB socket"}; + + const std::string relay_address[] = {listener::zmq_pub::relay_endpoint()}; + relay_socket = init_socket(context.get(), ZMQ_PAIR, relay_address); + if (!relay_socket) + throw std::runtime_error{"Unable to initialize ZMQ_PAIR relay"}; + } + catch (const std::runtime_error& e) { - MONERO_LOG_ZMQ_ERROR("ZMQ RPC Server bind failed"); - return false; + shared_state = nullptr; + pub_socket = nullptr; + relay_socket = nullptr; + MERROR("Failed to create ZMQ/Pub listener: " << e.what()); + return nullptr; } - return true; + + return shared_state; } void ZmqServer::run() @@ -163,7 +245,6 @@ void ZmqServer::stop() run_thread.join(); } - } // namespace cryptonote } // namespace rpc |