zmqpublishnotifier.cpp

Go to the documentation of this file.

// 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 <zmq/zmqpublishnotifier.h>
 
#include <chain.h>
#include <chainparams.h>
#include <common/system.h>
#include <config.h>
#include <logging.h>
#include <node/blockstorage.h>
#include <primitives/blockhash.h>
#include <primitives/txid.h>
#include <rpc/server.h>
#include <streams.h>
#include <zmq/zmqutil.h>
 
#include <zmq.h>
 
#include <cstdarg>
#include <cstddef>
#include <map>
#include <string>
#include <utility>
 
static std::multimap<std::string, CZMQAbstractPublishNotifier *>
    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";
static const char *MSG_SEQUENCE = "sequence";
 
// 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<std::string, CZMQAbstractPublishNotifier *>::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;
        }
 
        const int so_keepalive_option{1};
        rc = zmq_setsockopt(psocket, ZMQ_TCP_KEEPALIVE, &so_keepalive_option,
                            sizeof(so_keepalive_option));
        if (rc != 0) {
            zmqError("Failed to set SO_KEEPALIVE");
            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<std::string, CZMQAbstractPublishNotifier *>::iterator
        iterator;
    std::pair<iterator, iterator> 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::SendZmqMessage(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, 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 to %s\n", hash.GetHex(),
             this->address);
    uint8_t data[32];
    for (unsigned int i = 0; i < 32; i++) {
        data[31 - i] = hash.begin()[i];
    }
    return SendZmqMessage(MSG_HASHBLOCK, data, 32);
}
 
bool CZMQPublishHashTransactionNotifier::NotifyTransaction(
    const CTransaction &transaction) {
    TxId txid = transaction.GetId();
    LogPrint(BCLog::ZMQ, "zmq: Publish hashtx %s to %s\n", txid.GetHex(),
             this->address);
    uint8_t data[32];
    for (unsigned int i = 0; i < 32; i++) {
        data[31 - i] = txid.begin()[i];
    }
    return SendZmqMessage(MSG_HASHTX, data, 32);
}
 
bool CZMQPublishRawBlockNotifier::NotifyBlock(const CBlockIndex *pindex) {
    LogPrint(BCLog::ZMQ, "zmq: Publish rawblock %s to %s\n",
             pindex->GetBlockHash().GetHex(), this->address);
 
    CDataStream ss(SER_NETWORK, PROTOCOL_VERSION | RPCSerializationFlags());
    CBlock block;
    if (!m_get_block_by_index(block, *pindex)) {
        zmqError("Can't read block from disk");
        return false;
    }
 
    ss << block;
 
    return SendZmqMessage(MSG_RAWBLOCK, &(*ss.begin()), ss.size());
}
 
bool CZMQPublishRawTransactionNotifier::NotifyTransaction(
    const CTransaction &transaction) {
    TxId txid = transaction.GetId();
    LogPrint(BCLog::ZMQ, "zmq: Publish rawtx %s to %s\n", txid.GetHex(),
             this->address);
    CDataStream ss(SER_NETWORK, PROTOCOL_VERSION | RPCSerializationFlags());
    ss << transaction;
    return SendZmqMessage(MSG_RAWTX, &(*ss.begin()), ss.size());
}
 
// TODO: Dedup this code to take label char, log string
bool CZMQPublishSequenceNotifier::NotifyBlockConnect(
    const CBlockIndex *pindex) {
    BlockHash hash = pindex->GetBlockHash();
    LogPrint(BCLog::ZMQ, "zmq: Publish sequence block connect %s to %s\n",
             hash.GetHex(), this->address);
    char data[sizeof(BlockHash) + 1];
    for (unsigned int i = 0; i < sizeof(BlockHash); i++) {
        data[sizeof(BlockHash) - 1 - i] = hash.begin()[i];
    }
    // Block (C)onnect
    data[sizeof(data) - 1] = 'C';
    return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data));
}
 
bool CZMQPublishSequenceNotifier::NotifyBlockDisconnect(
    const CBlockIndex *pindex) {
    BlockHash hash = pindex->GetBlockHash();
    LogPrint(BCLog::ZMQ, "zmq: Publish sequence block disconnect %s to %s\n",
             hash.GetHex(), this->address);
    char data[sizeof(BlockHash) + 1];
    for (unsigned int i = 0; i < sizeof(BlockHash); i++) {
        data[sizeof(BlockHash) - 1 - i] = hash.begin()[i];
    }
    // Block (D)isconnect
    data[sizeof(data) - 1] = 'D';
    return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data));
}
 
bool CZMQPublishSequenceNotifier::NotifyTransactionAcceptance(
    const CTransaction &transaction, uint64_t mempool_sequence) {
    TxId txid = transaction.GetId();
    LogPrint(BCLog::ZMQ, "zmq: Publish hashtx mempool acceptance %s to %s\n",
             txid.GetHex(), this->address);
    uint8_t data[sizeof(TxId) + sizeof(mempool_sequence) + 1];
    for (unsigned int i = 0; i < sizeof(TxId); i++) {
        data[sizeof(TxId) - 1 - i] = txid.begin()[i];
    }
    // Mempool (A)cceptance
    data[sizeof(TxId)] = 'A';
    WriteLE64(data + sizeof(TxId) + 1, mempool_sequence);
    return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data));
}
 
bool CZMQPublishSequenceNotifier::NotifyTransactionRemoval(
    const CTransaction &transaction, uint64_t mempool_sequence) {
    TxId txid = transaction.GetId();
    LogPrint(BCLog::ZMQ, "zmq: Publish hashtx mempool removal %s to %s\n",
             txid.GetHex(), this->address);
    uint8_t data[sizeof(TxId) + sizeof(mempool_sequence) + 1];
    for (unsigned int i = 0; i < sizeof(TxId); i++) {
        data[sizeof(TxId) - 1 - i] = txid.begin()[i];
    }
    // Mempool (R)emoval
    data[sizeof(TxId)] = 'R';
    WriteLE64(data + sizeof(TxId) + 1, mempool_sequence);
    return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data));
}