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