Bitcoin Core  21.99.0
P2P Digital Currency
zmqpublishnotifier.cpp
Go to the documentation of this file.
1 // Copyright (c) 2015-2020 The Bitcoin Core developers
2 // Distributed under the MIT software license, see the accompanying
3 // file COPYING or http://www.opensource.org/licenses/mit-license.php.
4 
6 
7 #include <chain.h>
8 #include <chainparams.h>
9 #include <node/blockstorage.h>
10 #include <rpc/server.h>
11 #include <streams.h>
12 #include <util/system.h>
13 #include <validation.h> // For cs_main
14 #include <zmq/zmqutil.h>
15 
16 #include <zmq.h>
17 
18 #include <cstdarg>
19 #include <cstddef>
20 #include <map>
21 #include <optional>
22 #include <string>
23 #include <utility>
24 
25 static std::multimap<std::string, CZMQAbstractPublishNotifier*> mapPublishNotifiers;
26 
27 static const char *MSG_HASHBLOCK = "hashblock";
28 static const char *MSG_HASHTX = "hashtx";
29 static const char *MSG_RAWBLOCK = "rawblock";
30 static const char *MSG_RAWTX = "rawtx";
31 static const char *MSG_SEQUENCE = "sequence";
32 
33 // Internal function to send multipart message
34 static int zmq_send_multipart(void *sock, const void* data, size_t size, ...)
35 {
36  va_list args;
37  va_start(args, size);
38 
39  while (1)
40  {
41  zmq_msg_t msg;
42 
43  int rc = zmq_msg_init_size(&msg, size);
44  if (rc != 0)
45  {
46  zmqError("Unable to initialize ZMQ msg");
47  va_end(args);
48  return -1;
49  }
50 
51  void *buf = zmq_msg_data(&msg);
52  memcpy(buf, data, size);
53 
54  data = va_arg(args, const void*);
55 
56  rc = zmq_msg_send(&msg, sock, data ? ZMQ_SNDMORE : 0);
57  if (rc == -1)
58  {
59  zmqError("Unable to send ZMQ msg");
60  zmq_msg_close(&msg);
61  va_end(args);
62  return -1;
63  }
64 
65  zmq_msg_close(&msg);
66 
67  if (!data)
68  break;
69 
70  size = va_arg(args, size_t);
71  }
72  va_end(args);
73  return 0;
74 }
75 
77 {
78  assert(!psocket);
79 
80  // check if address is being used by other publish notifier
81  std::multimap<std::string, CZMQAbstractPublishNotifier*>::iterator i = mapPublishNotifiers.find(address);
82 
83  if (i==mapPublishNotifiers.end())
84  {
85  psocket = zmq_socket(pcontext, ZMQ_PUB);
86  if (!psocket)
87  {
88  zmqError("Failed to create socket");
89  return false;
90  }
91 
92  LogPrint(BCLog::ZMQ, "zmq: Outbound message high water mark for %s at %s is %d\n", type, address, outbound_message_high_water_mark);
93 
94  int rc = zmq_setsockopt(psocket, ZMQ_SNDHWM, &outbound_message_high_water_mark, sizeof(outbound_message_high_water_mark));
95  if (rc != 0)
96  {
97  zmqError("Failed to set outbound message high water mark");
98  zmq_close(psocket);
99  return false;
100  }
101 
102  const int so_keepalive_option {1};
103  rc = zmq_setsockopt(psocket, ZMQ_TCP_KEEPALIVE, &so_keepalive_option, sizeof(so_keepalive_option));
104  if (rc != 0) {
105  zmqError("Failed to set SO_KEEPALIVE");
106  zmq_close(psocket);
107  return false;
108  }
109 
110  rc = zmq_bind(psocket, address.c_str());
111  if (rc != 0)
112  {
113  zmqError("Failed to bind address");
114  zmq_close(psocket);
115  return false;
116  }
117 
118  // register this notifier for the address, so it can be reused for other publish notifier
119  mapPublishNotifiers.insert(std::make_pair(address, this));
120  return true;
121  }
122  else
123  {
124  LogPrint(BCLog::ZMQ, "zmq: Reusing socket for address %s\n", address);
125  LogPrint(BCLog::ZMQ, "zmq: Outbound message high water mark for %s at %s is %d\n", type, address, outbound_message_high_water_mark);
126 
127  psocket = i->second->psocket;
128  mapPublishNotifiers.insert(std::make_pair(address, this));
129 
130  return true;
131  }
132 }
133 
135 {
136  // Early return if Initialize was not called
137  if (!psocket) return;
138 
139  int count = mapPublishNotifiers.count(address);
140 
141  // remove this notifier from the list of publishers using this address
142  typedef std::multimap<std::string, CZMQAbstractPublishNotifier*>::iterator iterator;
143  std::pair<iterator, iterator> iterpair = mapPublishNotifiers.equal_range(address);
144 
145  for (iterator it = iterpair.first; it != iterpair.second; ++it)
146  {
147  if (it->second==this)
148  {
149  mapPublishNotifiers.erase(it);
150  break;
151  }
152  }
153 
154  if (count == 1)
155  {
156  LogPrint(BCLog::ZMQ, "zmq: Close socket at address %s\n", address);
157  int linger = 0;
158  zmq_setsockopt(psocket, ZMQ_LINGER, &linger, sizeof(linger));
159  zmq_close(psocket);
160  }
161 
162  psocket = nullptr;
163 }
164 
165 bool CZMQAbstractPublishNotifier::SendZmqMessage(const char *command, const void* data, size_t size)
166 {
167  assert(psocket);
168 
169  /* send three parts, command & data & a LE 4byte sequence number */
170  unsigned char msgseq[sizeof(uint32_t)];
171  WriteLE32(&msgseq[0], nSequence);
172  int rc = zmq_send_multipart(psocket, command, strlen(command), data, size, msgseq, (size_t)sizeof(uint32_t), nullptr);
173  if (rc == -1)
174  return false;
175 
176  /* increment memory only sequence number after sending */
177  nSequence++;
178 
179  return true;
180 }
181 
183 {
184  uint256 hash = pindex->GetBlockHash();
185  LogPrint(BCLog::ZMQ, "zmq: Publish hashblock %s to %s\n", hash.GetHex(), this->address);
186  char data[32];
187  for (unsigned int i = 0; i < 32; i++)
188  data[31 - i] = hash.begin()[i];
189  return SendZmqMessage(MSG_HASHBLOCK, data, 32);
190 }
191 
193 {
194  uint256 hash = transaction.GetHash();
195  LogPrint(BCLog::ZMQ, "zmq: Publish hashtx %s to %s\n", hash.GetHex(), this->address);
196  char data[32];
197  for (unsigned int i = 0; i < 32; i++)
198  data[31 - i] = hash.begin()[i];
199  return SendZmqMessage(MSG_HASHTX, data, 32);
200 }
201 
203 {
204  LogPrint(BCLog::ZMQ, "zmq: Publish rawblock %s to %s\n", pindex->GetBlockHash().GetHex(), this->address);
205 
206  const Consensus::Params& consensusParams = Params().GetConsensus();
208  {
209  LOCK(cs_main);
210  CBlock block;
211  if(!ReadBlockFromDisk(block, pindex, consensusParams))
212  {
213  zmqError("Can't read block from disk");
214  return false;
215  }
216 
217  ss << block;
218  }
219 
220  return SendZmqMessage(MSG_RAWBLOCK, &(*ss.begin()), ss.size());
221 }
222 
224 {
225  uint256 hash = transaction.GetHash();
226  LogPrint(BCLog::ZMQ, "zmq: Publish rawtx %s to %s\n", hash.GetHex(), this->address);
228  ss << transaction;
229  return SendZmqMessage(MSG_RAWTX, &(*ss.begin()), ss.size());
230 }
231 
232 // Helper function to send a 'sequence' topic message with the following structure:
233 // <32-byte hash> | <1-byte label> | <8-byte LE sequence> (optional)
234 static bool SendSequenceMsg(CZMQAbstractPublishNotifier& notifier, uint256 hash, char label, std::optional<uint64_t> sequence = {})
235 {
236  unsigned char data[sizeof(hash) + sizeof(label) + sizeof(uint64_t)];
237  for (unsigned int i = 0; i < sizeof(hash); ++i) {
238  data[sizeof(hash) - 1 - i] = hash.begin()[i];
239  }
240  data[sizeof(hash)] = label;
241  if (sequence) WriteLE64(data + sizeof(hash) + sizeof(label), *sequence);
242  return notifier.SendZmqMessage(MSG_SEQUENCE, data, sequence ? sizeof(data) : sizeof(hash) + sizeof(label));
243 }
244 
246 {
247  uint256 hash = pindex->GetBlockHash();
248  LogPrint(BCLog::ZMQ, "zmq: Publish sequence block connect %s to %s\n", hash.GetHex(), this->address);
249  return SendSequenceMsg(*this, hash, /* Block (C)onnect */ 'C');
250 }
251 
253 {
254  uint256 hash = pindex->GetBlockHash();
255  LogPrint(BCLog::ZMQ, "zmq: Publish sequence block disconnect %s to %s\n", hash.GetHex(), this->address);
256  return SendSequenceMsg(*this, hash, /* Block (D)isconnect */ 'D');
257 }
258 
259 bool CZMQPublishSequenceNotifier::NotifyTransactionAcceptance(const CTransaction &transaction, uint64_t mempool_sequence)
260 {
261  uint256 hash = transaction.GetHash();
262  LogPrint(BCLog::ZMQ, "zmq: Publish hashtx mempool acceptance %s to %s\n", hash.GetHex(), this->address);
263  return SendSequenceMsg(*this, hash, /* Mempool (A)cceptance */ 'A', mempool_sequence);
264 }
265 
266 bool CZMQPublishSequenceNotifier::NotifyTransactionRemoval(const CTransaction &transaction, uint64_t mempool_sequence)
267 {
268  uint256 hash = transaction.GetHash();
269  LogPrint(BCLog::ZMQ, "zmq: Publish hashtx mempool removal %s to %s\n", hash.GetHex(), this->address);
270  return SendSequenceMsg(*this, hash, /* Mempool (R)emoval */ 'R', mempool_sequence);
271 }
CZMQPublishSequenceNotifier::NotifyBlockDisconnect
bool NotifyBlockDisconnect(const CBlockIndex *pindex) override
Definition: zmqpublishnotifier.cpp:252
MSG_RAWBLOCK
static const char * MSG_RAWBLOCK
Definition: zmqpublishnotifier.cpp:29
count
static int count
Definition: tests.c:35
BCLog::ZMQ
@ ZMQ
Definition: logging.h:43
CZMQPublishSequenceNotifier::NotifyBlockConnect
bool NotifyBlockConnect(const CBlockIndex *pindex) override
Definition: zmqpublishnotifier.cpp:245
streams.h
CDataStream::begin
const_iterator begin() const
Definition: streams.h:251
WriteLE32
static void WriteLE32(unsigned char *ptr, uint32_t x)
Definition: common.h:44
WriteLE64
static void WriteLE64(unsigned char *ptr, uint64_t x)
Definition: common.h:50
CChainParams::GetConsensus
const Consensus::Params & GetConsensus() const
Definition: chainparams.h:77
CZMQAbstractNotifier::type
std::string type
Definition: zmqabstractnotifier.h:61
zmqError
void zmqError(const char *str)
Definition: zmqutil.cpp:11
chainparams.h
cs_main
RecursiveMutex cs_main
Mutex to guard access to validation specific variables, such as reading or changing the chainstate.
Definition: validation.cpp:131
CZMQAbstractPublishNotifier::Shutdown
void Shutdown() override
Definition: zmqpublishnotifier.cpp:134
CTransaction
The basic transaction that is broadcasted on the network and contained in blocks.
Definition: transaction.h:259
CZMQAbstractPublishNotifier::nSequence
uint32_t nSequence
upcounting per message sequence number
Definition: zmqpublishnotifier.h:15
CZMQPublishRawBlockNotifier::NotifyBlock
bool NotifyBlock(const CBlockIndex *pindex) override
Definition: zmqpublishnotifier.cpp:202
Consensus::Params
Parameters that influence chain consensus.
Definition: params.h:56
zmqutil.h
CZMQPublishSequenceNotifier::NotifyTransactionAcceptance
bool NotifyTransactionAcceptance(const CTransaction &transaction, uint64_t mempool_sequence) override
Definition: zmqpublishnotifier.cpp:259
CZMQAbstractNotifier::psocket
void * psocket
Definition: zmqabstractnotifier.h:60
CZMQAbstractNotifier::outbound_message_high_water_mark
int outbound_message_high_water_mark
Definition: zmqabstractnotifier.h:63
base_blob::GetHex
std::string GetHex() const
Definition: uint256.cpp:20
CZMQPublishHashTransactionNotifier::NotifyTransaction
bool NotifyTransaction(const CTransaction &transaction) override
Definition: zmqpublishnotifier.cpp:192
ReadBlockFromDisk
bool ReadBlockFromDisk(CBlock &block, const FlatFilePos &pos, const Consensus::Params &consensusParams)
Functions for disk access for blocks.
Definition: blockstorage.cpp:44
uint256
256-bit opaque blob.
Definition: uint256.h:124
LogPrint
#define LogPrint(category,...)
Definition: logging.h:187
CBlockIndex::GetBlockHash
uint256 GetBlockHash() const
Definition: chain.h:246
CZMQAbstractNotifier::address
std::string address
Definition: zmqabstractnotifier.h:62
zmqpublishnotifier.h
CDataStream::size
size_type size() const
Definition: streams.h:255
system.h
CBlock
Definition: block.h:62
MSG_HASHTX
static const char * MSG_HASHTX
Definition: zmqpublishnotifier.cpp:28
CZMQAbstractPublishNotifier
Definition: zmqpublishnotifier.h:12
LOCK
#define LOCK(cs)
Definition: sync.h:232
CZMQAbstractPublishNotifier::Initialize
bool Initialize(void *pcontext) override
Definition: zmqpublishnotifier.cpp:76
blockstorage.h
CZMQPublishRawTransactionNotifier::NotifyTransaction
bool NotifyTransaction(const CTransaction &transaction) override
Definition: zmqpublishnotifier.cpp:223
RPCSerializationFlags
int RPCSerializationFlags()
Definition: server.cpp:540
mapPublishNotifiers
static std::multimap< std::string, CZMQAbstractPublishNotifier * > mapPublishNotifiers
Definition: zmqpublishnotifier.cpp:25
Params
const CChainParams & Params()
Return the currently selected parameters.
Definition: chainparams.cpp:538
MSG_RAWTX
static const char * MSG_RAWTX
Definition: zmqpublishnotifier.cpp:30
CDataStream
Double ended buffer combining vector and stream-like interfaces.
Definition: streams.h:204
SendSequenceMsg
static bool SendSequenceMsg(CZMQAbstractPublishNotifier &notifier, uint256 hash, char label, std::optional< uint64_t > sequence={})
Definition: zmqpublishnotifier.cpp:234
SER_NETWORK
@ SER_NETWORK
Definition: serialize.h:166
CTransaction::GetHash
const uint256 & GetHash() const
Definition: transaction.h:302
CZMQAbstractPublishNotifier::SendZmqMessage
bool SendZmqMessage(const char *command, const void *data, size_t size)
Definition: zmqpublishnotifier.cpp:165
CZMQPublishSequenceNotifier::NotifyTransactionRemoval
bool NotifyTransactionRemoval(const CTransaction &transaction, uint64_t mempool_sequence) override
Definition: zmqpublishnotifier.cpp:266
server.h
CBlockIndex
The block chain is a tree shaped structure starting with the genesis block at the root,...
Definition: chain.h:137
assert
assert(std::addressof(::ChainstateActive().CoinsTip())==std::addressof(coins_cache))
CZMQPublishHashBlockNotifier::NotifyBlock
bool NotifyBlock(const CBlockIndex *pindex) override
Definition: zmqpublishnotifier.cpp:182
base_blob::begin
unsigned char * begin()
Definition: uint256.h:58
MSG_HASHBLOCK
static const char * MSG_HASHBLOCK
Definition: zmqpublishnotifier.cpp:27
zmq_send_multipart
static int zmq_send_multipart(void *sock, const void *data, size_t size,...)
Definition: zmqpublishnotifier.cpp:34
it
auto it
Definition: validation.cpp:399
PROTOCOL_VERSION
static const int PROTOCOL_VERSION
network protocol versioning
Definition: version.h:12
MSG_SEQUENCE
static const char * MSG_SEQUENCE
Definition: zmqpublishnotifier.cpp:31