diff --git a/src/zmq/zmqnotificationinterface.cpp b/src/zmq/zmqnotificationinterface.cpp index 0939071e0..634fd795c 100644 --- a/src/zmq/zmqnotificationinterface.cpp +++ b/src/zmq/zmqnotificationinterface.cpp @@ -1,188 +1,171 @@ // Copyright (c) 2015-2018 The Bitcoin Core developers // Distributed under the MIT software license, see the accompanying // file COPYING or http://www.opensource.org/licenses/mit-license.php. #include #include #include #include void zmqError(const char *str) { LogPrint(BCLog::ZMQ, "zmq: Error: %s, errno=%s\n", str, zmq_strerror(errno)); } CZMQNotificationInterface::CZMQNotificationInterface() : pcontext(nullptr) {} CZMQNotificationInterface::~CZMQNotificationInterface() { Shutdown(); - - for (std::list::iterator i = notifiers.begin(); - i != notifiers.end(); ++i) { - delete *i; - } } std::list CZMQNotificationInterface::GetActiveNotifiers() const { std::list result; - for (const auto *n : notifiers) { - result.push_back(n); + for (const auto &n : notifiers) { + result.push_back(n.get()); } return result; } CZMQNotificationInterface *CZMQNotificationInterface::Create() { - CZMQNotificationInterface *notificationInterface = nullptr; std::map factories; - std::list notifiers; - factories["pubhashblock"] = CZMQAbstractNotifier::Create; factories["pubhashtx"] = CZMQAbstractNotifier::Create; factories["pubrawblock"] = CZMQAbstractNotifier::Create; factories["pubrawtx"] = CZMQAbstractNotifier::Create; + std::list> notifiers; for (const auto &entry : factories) { std::string arg("-zmq" + entry.first); if (gArgs.IsArgSet(arg)) { CZMQNotifierFactory factory = entry.second; std::string address = gArgs.GetArg(arg, ""); CZMQAbstractNotifier *notifier = factory(); notifier->SetType(entry.first); notifier->SetAddress(address); notifier->SetOutboundMessageHighWaterMark( static_cast(gArgs.GetArg( arg + "hwm", CZMQAbstractNotifier::DEFAULT_ZMQ_SNDHWM))); - notifiers.push_back(notifier); + notifiers.emplace_back(notifier); } } if (!notifiers.empty()) { - notificationInterface = new CZMQNotificationInterface(); - notificationInterface->notifiers = notifiers; + std::unique_ptr notificationInterface( + new CZMQNotificationInterface()); + notificationInterface->notifiers = std::move(notifiers); - if (!notificationInterface->Initialize()) { - delete notificationInterface; - notificationInterface = nullptr; + if (notificationInterface->Initialize()) { + return notificationInterface.release(); } } - return notificationInterface; + return nullptr; } // Called at startup to conditionally set up ZMQ socket(s) bool CZMQNotificationInterface::Initialize() { int major = 0, minor = 0, patch = 0; zmq_version(&major, &minor, &patch); LogPrint(BCLog::ZMQ, "zmq: version %d.%d.%d\n", major, minor, patch); LogPrint(BCLog::ZMQ, "zmq: Initialize notification interface\n"); assert(!pcontext); pcontext = zmq_ctx_new(); if (!pcontext) { zmqError("Unable to initialize context"); return false; } - std::list::iterator i = notifiers.begin(); - for (; i != notifiers.end(); ++i) { - CZMQAbstractNotifier *notifier = *i; + for (auto ¬ifier : notifiers) { if (notifier->Initialize(pcontext)) { LogPrint(BCLog::ZMQ, "zmq: Notifier %s ready (address = %s)\n", notifier->GetType(), notifier->GetAddress()); } else { LogPrint(BCLog::ZMQ, "zmq: Notifier %s failed (address = %s)\n", notifier->GetType(), notifier->GetAddress()); - break; + return false; } } - if (i != notifiers.end()) { - return false; - } - return true; } // Called during shutdown sequence void CZMQNotificationInterface::Shutdown() { LogPrint(BCLog::ZMQ, "zmq: Shutdown notification interface\n"); if (pcontext) { - for (std::list::iterator i = notifiers.begin(); - i != notifiers.end(); ++i) { - CZMQAbstractNotifier *notifier = *i; + for (auto ¬ifier : notifiers) { LogPrint(BCLog::ZMQ, "zmq: Shutdown notifier %s at %s\n", notifier->GetType(), notifier->GetAddress()); notifier->Shutdown(); } zmq_ctx_term(pcontext); pcontext = nullptr; } } void CZMQNotificationInterface::UpdatedBlockTip(const CBlockIndex *pindexNew, const CBlockIndex *pindexFork, bool fInitialDownload) { // In IBD or blocks were disconnected without any new ones if (fInitialDownload || pindexNew == pindexFork) { return; } - for (std::list::iterator i = notifiers.begin(); - i != notifiers.end();) { - CZMQAbstractNotifier *notifier = *i; + for (auto i = notifiers.begin(); i != notifiers.end();) { + CZMQAbstractNotifier *notifier = i->get(); if (notifier->NotifyBlock(pindexNew)) { i++; } else { notifier->Shutdown(); i = notifiers.erase(i); } } } void CZMQNotificationInterface::TransactionAddedToMempool( const CTransactionRef &ptx) { // Used by BlockConnected and BlockDisconnected as well, because they're all // the same external callback. const CTransaction &tx = *ptx; - for (std::list::iterator i = notifiers.begin(); - i != notifiers.end();) { - CZMQAbstractNotifier *notifier = *i; + for (auto i = notifiers.begin(); i != notifiers.end();) { + CZMQAbstractNotifier *notifier = i->get(); if (notifier->NotifyTransaction(tx)) { i++; } else { notifier->Shutdown(); i = notifiers.erase(i); } } } void CZMQNotificationInterface::BlockConnected( const std::shared_ptr &pblock, const CBlockIndex *pindexConnected) { for (const CTransactionRef &ptx : pblock->vtx) { // Do a normal notify for each transaction added in the block TransactionAddedToMempool(ptx); } } void CZMQNotificationInterface::BlockDisconnected( const std::shared_ptr &pblock, const CBlockIndex *pindexDisconnected) { for (const CTransactionRef &ptx : pblock->vtx) { // Do a normal notify for each transaction removed in block // disconnection TransactionAddedToMempool(ptx); } } CZMQNotificationInterface *g_zmq_notification_interface = nullptr; diff --git a/src/zmq/zmqnotificationinterface.h b/src/zmq/zmqnotificationinterface.h index 117d5bc3b..e0f3fa371 100644 --- a/src/zmq/zmqnotificationinterface.h +++ b/src/zmq/zmqnotificationinterface.h @@ -1,46 +1,47 @@ // Copyright (c) 2015-2018 The Bitcoin Core developers // Distributed under the MIT software license, see the accompanying // file COPYING or http://www.opensource.org/licenses/mit-license.php. #ifndef BITCOIN_ZMQ_ZMQNOTIFICATIONINTERFACE_H #define BITCOIN_ZMQ_ZMQNOTIFICATIONINTERFACE_H #include #include +#include class CBlockIndex; class CZMQAbstractNotifier; class CZMQNotificationInterface final : public CValidationInterface { public: virtual ~CZMQNotificationInterface(); std::list GetActiveNotifiers() const; static CZMQNotificationInterface *Create(); protected: bool Initialize(); void Shutdown(); // CValidationInterface void TransactionAddedToMempool(const CTransactionRef &tx) override; void BlockConnected(const std::shared_ptr &pblock, const CBlockIndex *pindexConnected) override; void BlockDisconnected(const std::shared_ptr &pblock, const CBlockIndex *pindexDisconnected) override; void UpdatedBlockTip(const CBlockIndex *pindexNew, const CBlockIndex *pindexFork, bool fInitialDownload) override; private: CZMQNotificationInterface(); void *pcontext; - std::list notifiers; + std::list> notifiers; }; extern CZMQNotificationInterface *g_zmq_notification_interface; #endif // BITCOIN_ZMQ_ZMQNOTIFICATIONINTERFACE_H diff --git a/src/zmq/zmqpublishnotifier.cpp b/src/zmq/zmqpublishnotifier.cpp index 4ff10648d..c412f9626 100644 --- a/src/zmq/zmqpublishnotifier.cpp +++ b/src/zmq/zmqpublishnotifier.cpp @@ -1,217 +1,222 @@ // Copyright (c) 2015-2016 The Bitcoin Core developers // Distributed under the MIT software license, see the accompanying // file COPYING or http://www.opensource.org/licenses/mit-license.php. #include #include #include #include #include #include #include #include #include #include +#include #include +#include +#include +#include +#include static std::multimap mapPublishNotifiers; static const char *MSG_HASHBLOCK = "hashblock"; static const char *MSG_HASHTX = "hashtx"; static const char *MSG_RAWBLOCK = "rawblock"; static const char *MSG_RAWTX = "rawtx"; // Internal function to send multipart message static int zmq_send_multipart(void *sock, const void *data, size_t size, ...) { va_list args; va_start(args, size); while (1) { zmq_msg_t msg; int rc = zmq_msg_init_size(&msg, size); if (rc != 0) { zmqError("Unable to initialize ZMQ msg"); va_end(args); return -1; } void *buf = zmq_msg_data(&msg); memcpy(buf, data, size); data = va_arg(args, const void *); rc = zmq_msg_send(&msg, sock, data ? ZMQ_SNDMORE : 0); if (rc == -1) { zmqError("Unable to send ZMQ msg"); zmq_msg_close(&msg); va_end(args); return -1; } zmq_msg_close(&msg); if (!data) { break; } size = va_arg(args, size_t); } va_end(args); return 0; } bool CZMQAbstractPublishNotifier::Initialize(void *pcontext) { assert(!psocket); // check if address is being used by other publish notifier std::multimap::iterator i = mapPublishNotifiers.find(address); if (i == mapPublishNotifiers.end()) { psocket = zmq_socket(pcontext, ZMQ_PUB); if (!psocket) { zmqError("Failed to create socket"); return false; } LogPrint(BCLog::ZMQ, "zmq: Outbound message high water mark for %s at %s is %d\n", type, address, outbound_message_high_water_mark); int rc = zmq_setsockopt(psocket, ZMQ_SNDHWM, &outbound_message_high_water_mark, sizeof(outbound_message_high_water_mark)); if (rc != 0) { zmqError("Failed to set outbound message high water mark"); zmq_close(psocket); return false; } rc = zmq_bind(psocket, address.c_str()); if (rc != 0) { zmqError("Failed to bind address"); zmq_close(psocket); return false; } // register this notifier for the address, so it can be reused for other // publish notifier mapPublishNotifiers.insert(std::make_pair(address, this)); return true; } else { LogPrint(BCLog::ZMQ, "zmq: Reusing socket for address %s\n", address); LogPrint(BCLog::ZMQ, "zmq: Outbound message high water mark for %s at %s is %d\n", type, address, outbound_message_high_water_mark); psocket = i->second->psocket; mapPublishNotifiers.insert(std::make_pair(address, this)); return true; } } void CZMQAbstractPublishNotifier::Shutdown() { // Early return if Initialize was not called if (!psocket) { return; } int count = mapPublishNotifiers.count(address); // remove this notifier from the list of publishers using this address typedef std::multimap::iterator iterator; std::pair iterpair = mapPublishNotifiers.equal_range(address); for (iterator it = iterpair.first; it != iterpair.second; ++it) { if (it->second == this) { mapPublishNotifiers.erase(it); break; } } if (count == 1) { LogPrint(BCLog::ZMQ, "zmq: Close socket at address %s\n", address); int linger = 0; zmq_setsockopt(psocket, ZMQ_LINGER, &linger, sizeof(linger)); zmq_close(psocket); } psocket = nullptr; } bool CZMQAbstractPublishNotifier::SendMessage(const char *command, const void *data, size_t size) { assert(psocket); /* send three parts, command & data & a LE 4byte sequence number */ uint8_t msgseq[sizeof(uint32_t)]; WriteLE32(&msgseq[0], nSequence); int rc = zmq_send_multipart(psocket, command, strlen(command), data, size, msgseq, (size_t)sizeof(uint32_t), nullptr); if (rc == -1) { return false; } /* increment memory only sequence number after sending */ nSequence++; return true; } bool CZMQPublishHashBlockNotifier::NotifyBlock(const CBlockIndex *pindex) { BlockHash hash = pindex->GetBlockHash(); LogPrint(BCLog::ZMQ, "zmq: Publish hashblock %s\n", hash.GetHex()); char data[32]; for (unsigned int i = 0; i < 32; i++) { data[31 - i] = hash.begin()[i]; } return SendMessage(MSG_HASHBLOCK, data, 32); } bool CZMQPublishHashTransactionNotifier::NotifyTransaction( const CTransaction &transaction) { TxId txid = transaction.GetId(); LogPrint(BCLog::ZMQ, "zmq: Publish hashtx %s\n", txid.GetHex()); char data[32]; for (unsigned int i = 0; i < 32; i++) { data[31 - i] = txid.begin()[i]; } return SendMessage(MSG_HASHTX, data, 32); } bool CZMQPublishRawBlockNotifier::NotifyBlock(const CBlockIndex *pindex) { LogPrint(BCLog::ZMQ, "zmq: Publish rawblock %s\n", pindex->GetBlockHash().GetHex()); const Config &config = GetConfig(); CDataStream ss(SER_NETWORK, PROTOCOL_VERSION | RPCSerializationFlags()); { LOCK(cs_main); CBlock block; if (!ReadBlockFromDisk(block, pindex, config.GetChainParams().GetConsensus())) { zmqError("Can't read block from disk"); return false; } ss << block; } return SendMessage(MSG_RAWBLOCK, &(*ss.begin()), ss.size()); } bool CZMQPublishRawTransactionNotifier::NotifyTransaction( const CTransaction &transaction) { TxId txid = transaction.GetId(); LogPrint(BCLog::ZMQ, "zmq: Publish rawtx %s\n", txid.GetHex()); CDataStream ss(SER_NETWORK, PROTOCOL_VERSION | RPCSerializationFlags()); ss << transaction; return SendMessage(MSG_RAWTX, &(*ss.begin()), ss.size()); }