Branch data Line data Source code
1 : : // Copyright (c) 2015-present 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 : :
5 : : #include <zmq/zmqpublishnotifier.h>
6 : :
7 : : #include <chain.h>
8 : : #include <crypto/common.h>
9 : : #include <logging.h>
10 : : #include <netaddress.h>
11 : : #include <netbase.h>
12 : : #include <primitives/transaction.h>
13 : : #include <serialize.h>
14 : : #include <streams.h>
15 : : #include <uint256.h>
16 : : #include <util/check.h>
17 : : #include <zmq/zmqutil.h>
18 : :
19 : : #include <zmq.h>
20 : :
21 : : #include <cstdarg>
22 : : #include <cstddef>
23 : : #include <cstdint>
24 : : #include <cstring>
25 : : #include <map>
26 : : #include <optional>
27 : : #include <span>
28 : : #include <string>
29 : : #include <utility>
30 : : #include <vector>
31 : :
32 : : static std::multimap<std::string, CZMQAbstractPublishNotifier*> mapPublishNotifiers;
33 : :
34 : : static const char *MSG_HASHBLOCK = "hashblock";
35 : : static const char *MSG_HASHTX = "hashtx";
36 : : static const char *MSG_RAWBLOCK = "rawblock";
37 : : static const char *MSG_RAWTX = "rawtx";
38 : : static const char *MSG_SEQUENCE = "sequence";
39 : :
40 : : // Internal function to send multipart message
41 : 0 : static int zmq_send_multipart(void *sock, const void* data, size_t size, ...)
42 : : {
43 : 0 : va_list args;
44 : 0 : va_start(args, size);
45 : :
46 : 0 : while (1)
47 : : {
48 : 0 : zmq_msg_t msg;
49 : :
50 : 0 : int rc = zmq_msg_init_size(&msg, size);
51 [ # # ]: 0 : if (rc != 0)
52 : : {
53 [ # # ]: 0 : zmqError("Unable to initialize ZMQ msg");
54 : 0 : va_end(args);
55 : 0 : return -1;
56 : : }
57 : :
58 : 0 : void *buf = zmq_msg_data(&msg);
59 : 0 : memcpy(buf, data, size);
60 : :
61 : 0 : data = va_arg(args, const void*);
62 : :
63 [ # # ]: 0 : rc = zmq_msg_send(&msg, sock, data ? ZMQ_SNDMORE : 0);
64 [ # # ]: 0 : if (rc == -1)
65 : : {
66 [ # # ]: 0 : zmqError("Unable to send ZMQ msg");
67 : 0 : zmq_msg_close(&msg);
68 : 0 : va_end(args);
69 : 0 : return -1;
70 : : }
71 : :
72 : 0 : zmq_msg_close(&msg);
73 : :
74 [ # # ]: 0 : if (!data)
75 : : break;
76 : :
77 : 0 : size = va_arg(args, size_t);
78 : 0 : }
79 : 0 : va_end(args);
80 : 0 : return 0;
81 : : }
82 : :
83 : 0 : static bool IsZMQAddressIPV6(const std::string &zmq_address)
84 : : {
85 : 0 : const std::string tcp_prefix = "tcp://";
86 [ # # ]: 0 : const size_t tcp_index = zmq_address.rfind(tcp_prefix);
87 : 0 : const size_t colon_index = zmq_address.rfind(':');
88 [ # # ]: 0 : if (tcp_index == 0 && colon_index != std::string::npos) {
89 [ # # ]: 0 : const std::string ip = zmq_address.substr(tcp_prefix.length(), colon_index - tcp_prefix.length());
90 [ # # # # ]: 0 : const std::optional<CNetAddr> addr{LookupHost(ip, false)};
91 [ # # # # ]: 0 : if (addr.has_value() && addr.value().IsIPv6()) return true;
92 : 0 : }
93 : : return false;
94 : 0 : }
95 : :
96 : 0 : bool CZMQAbstractPublishNotifier::Initialize(void *pcontext)
97 : : {
98 [ # # ]: 0 : assert(!psocket);
99 : :
100 : : // check if address is being used by other publish notifier
101 : 0 : std::multimap<std::string, CZMQAbstractPublishNotifier*>::iterator i = mapPublishNotifiers.find(address);
102 : :
103 [ # # ]: 0 : if (i==mapPublishNotifiers.end())
104 : : {
105 : 0 : psocket = zmq_socket(pcontext, ZMQ_PUB);
106 [ # # ]: 0 : if (!psocket)
107 : : {
108 [ # # ]: 0 : zmqError("Failed to create socket");
109 : 0 : return false;
110 : : }
111 : :
112 [ # # ]: 0 : LogDebug(BCLog::ZMQ, "Outbound message high water mark for %s at %s is %d\n", type, address, outbound_message_high_water_mark);
113 : :
114 : 0 : int rc = zmq_setsockopt(psocket, ZMQ_SNDHWM, &outbound_message_high_water_mark, sizeof(outbound_message_high_water_mark));
115 [ # # ]: 0 : if (rc != 0)
116 : : {
117 [ # # ]: 0 : zmqError("Failed to set outbound message high water mark");
118 : 0 : zmq_close(psocket);
119 : 0 : return false;
120 : : }
121 : :
122 : 0 : const int so_keepalive_option {1};
123 : 0 : rc = zmq_setsockopt(psocket, ZMQ_TCP_KEEPALIVE, &so_keepalive_option, sizeof(so_keepalive_option));
124 [ # # ]: 0 : if (rc != 0) {
125 [ # # ]: 0 : zmqError("Failed to set SO_KEEPALIVE");
126 : 0 : zmq_close(psocket);
127 : 0 : return false;
128 : : }
129 : :
130 : : // On some systems (e.g. OpenBSD) the ZMQ_IPV6 must not be enabled, if the address to bind isn't IPv6
131 [ # # ]: 0 : const int enable_ipv6 { IsZMQAddressIPV6(address) ? 1 : 0};
132 : 0 : rc = zmq_setsockopt(psocket, ZMQ_IPV6, &enable_ipv6, sizeof(enable_ipv6));
133 [ # # ]: 0 : if (rc != 0) {
134 [ # # ]: 0 : zmqError("Failed to set ZMQ_IPV6");
135 : 0 : zmq_close(psocket);
136 : 0 : return false;
137 : : }
138 : :
139 : 0 : rc = zmq_bind(psocket, address.c_str());
140 [ # # ]: 0 : if (rc != 0)
141 : : {
142 [ # # ]: 0 : zmqError("Failed to bind address");
143 : 0 : zmq_close(psocket);
144 : 0 : return false;
145 : : }
146 : :
147 : : // register this notifier for the address, so it can be reused for other publish notifier
148 [ # # # # ]: 0 : mapPublishNotifiers.insert(std::make_pair(address, this));
149 : 0 : return true;
150 : : }
151 : : else
152 : : {
153 [ # # ]: 0 : LogDebug(BCLog::ZMQ, "Reusing socket for address %s\n", address);
154 [ # # ]: 0 : LogDebug(BCLog::ZMQ, "Outbound message high water mark for %s at %s is %d\n", type, address, outbound_message_high_water_mark);
155 : :
156 [ # # ]: 0 : psocket = i->second->psocket;
157 [ # # # # ]: 0 : mapPublishNotifiers.insert(std::make_pair(address, this));
158 : :
159 : 0 : return true;
160 : : }
161 : : }
162 : :
163 : 0 : void CZMQAbstractPublishNotifier::Shutdown()
164 : : {
165 : : // Early return if Initialize was not called
166 [ # # ]: 0 : if (!psocket) return;
167 : :
168 : 0 : int count = mapPublishNotifiers.count(address);
169 : :
170 : : // remove this notifier from the list of publishers using this address
171 : 0 : typedef std::multimap<std::string, CZMQAbstractPublishNotifier*>::iterator iterator;
172 : 0 : std::pair<iterator, iterator> iterpair = mapPublishNotifiers.equal_range(address);
173 : :
174 [ # # ]: 0 : for (iterator it = iterpair.first; it != iterpair.second; ++it)
175 : : {
176 [ # # ]: 0 : if (it->second==this)
177 : : {
178 : 0 : mapPublishNotifiers.erase(it);
179 : 0 : break;
180 : : }
181 : : }
182 : :
183 [ # # ]: 0 : if (count == 1)
184 : : {
185 [ # # ]: 0 : LogDebug(BCLog::ZMQ, "Close socket at address %s\n", address);
186 : 0 : int linger = 0;
187 : 0 : zmq_setsockopt(psocket, ZMQ_LINGER, &linger, sizeof(linger));
188 : 0 : zmq_close(psocket);
189 : : }
190 : :
191 : 0 : psocket = nullptr;
192 : : }
193 : :
194 : 0 : bool CZMQAbstractPublishNotifier::SendZmqMessage(const char *command, const void* data, size_t size)
195 : : {
196 [ # # ]: 0 : assert(psocket);
197 : :
198 : : /* send three parts, command & data & a LE 4byte sequence number */
199 : 0 : unsigned char msgseq[sizeof(uint32_t)];
200 : 0 : WriteLE32(msgseq, nSequence);
201 : 0 : int rc = zmq_send_multipart(psocket, command, strlen(command), data, size, msgseq, (size_t)sizeof(uint32_t), nullptr);
202 [ # # ]: 0 : if (rc == -1)
203 : : return false;
204 : :
205 : : /* increment memory only sequence number after sending */
206 : 0 : nSequence++;
207 : :
208 : 0 : return true;
209 : : }
210 : :
211 : 0 : bool CZMQPublishHashBlockNotifier::NotifyBlock(const CBlockIndex *pindex)
212 : : {
213 : 0 : uint256 hash = pindex->GetBlockHash();
214 [ # # # # ]: 0 : LogDebug(BCLog::ZMQ, "Publish hashblock %s to %s\n", hash.GetHex(), this->address);
215 : : uint8_t data[32];
216 [ # # ]: 0 : for (unsigned int i = 0; i < 32; i++) {
217 : 0 : data[31 - i] = hash.begin()[i];
218 : : }
219 : 0 : return SendZmqMessage(MSG_HASHBLOCK, data, 32);
220 : : }
221 : :
222 : 0 : bool CZMQPublishHashTransactionNotifier::NotifyTransaction(const CTransaction &transaction)
223 : : {
224 : 0 : uint256 hash = transaction.GetHash().ToUint256();
225 [ # # # # ]: 0 : LogDebug(BCLog::ZMQ, "Publish hashtx %s to %s\n", hash.GetHex(), this->address);
226 : : uint8_t data[32];
227 [ # # ]: 0 : for (unsigned int i = 0; i < 32; i++) {
228 : 0 : data[31 - i] = hash.begin()[i];
229 : : }
230 : 0 : return SendZmqMessage(MSG_HASHTX, data, 32);
231 : : }
232 : :
233 : 0 : bool CZMQPublishRawBlockNotifier::NotifyBlock(const CBlockIndex *pindex)
234 : : {
235 [ # # # # ]: 0 : LogDebug(BCLog::ZMQ, "Publish rawblock %s to %s\n", pindex->GetBlockHash().GetHex(), this->address);
236 : :
237 : 0 : std::vector<std::byte> block{};
238 [ # # # # ]: 0 : if (!m_get_block_by_index(block, *pindex)) {
239 [ # # # # ]: 0 : zmqError("Can't read block from disk");
240 : 0 : return false;
241 : : }
242 : :
243 [ # # # # ]: 0 : return SendZmqMessage(MSG_RAWBLOCK, block.data(), block.size());
244 : 0 : }
245 : :
246 : 0 : bool CZMQPublishRawTransactionNotifier::NotifyTransaction(const CTransaction &transaction)
247 : : {
248 : 0 : uint256 hash = transaction.GetHash().ToUint256();
249 [ # # # # ]: 0 : LogDebug(BCLog::ZMQ, "Publish rawtx %s to %s\n", hash.GetHex(), this->address);
250 : 0 : DataStream ss;
251 [ # # ]: 0 : ss << TX_WITH_WITNESS(transaction);
252 [ # # # # ]: 0 : return SendZmqMessage(MSG_RAWTX, &(*ss.begin()), ss.size());
253 : 0 : }
254 : :
255 : : // Helper function to send a 'sequence' topic message with the following structure:
256 : : // <32-byte hash> | <1-byte label> | <8-byte LE sequence> (optional)
257 : 0 : static bool SendSequenceMsg(CZMQAbstractPublishNotifier& notifier, uint256 hash, char label, std::optional<uint64_t> sequence = {})
258 : : {
259 : 0 : unsigned char data[sizeof(hash) + sizeof(label) + sizeof(uint64_t)];
260 [ # # ]: 0 : for (unsigned int i = 0; i < sizeof(hash); ++i) {
261 : 0 : data[sizeof(hash) - 1 - i] = hash.begin()[i];
262 : : }
263 : 0 : data[sizeof(hash)] = label;
264 [ # # ]: 0 : if (sequence) WriteLE64(data + sizeof(hash) + sizeof(label), *sequence);
265 [ # # ]: 0 : return notifier.SendZmqMessage(MSG_SEQUENCE, data, sequence ? sizeof(data) : sizeof(hash) + sizeof(label));
266 : : }
267 : :
268 : 0 : bool CZMQPublishSequenceNotifier::NotifyBlockConnect(const CBlockIndex *pindex)
269 : : {
270 : 0 : uint256 hash = pindex->GetBlockHash();
271 [ # # # # ]: 0 : LogDebug(BCLog::ZMQ, "Publish sequence block connect %s to %s\n", hash.GetHex(), this->address);
272 : 0 : return SendSequenceMsg(*this, hash, /* Block (C)onnect */ 'C');
273 : : }
274 : :
275 : 0 : bool CZMQPublishSequenceNotifier::NotifyBlockDisconnect(const CBlockIndex *pindex)
276 : : {
277 : 0 : uint256 hash = pindex->GetBlockHash();
278 [ # # # # ]: 0 : LogDebug(BCLog::ZMQ, "Publish sequence block disconnect %s to %s\n", hash.GetHex(), this->address);
279 : 0 : return SendSequenceMsg(*this, hash, /* Block (D)isconnect */ 'D');
280 : : }
281 : :
282 : 0 : bool CZMQPublishSequenceNotifier::NotifyTransactionAcceptance(const CTransaction &transaction, uint64_t mempool_sequence)
283 : : {
284 : 0 : uint256 hash = transaction.GetHash().ToUint256();
285 [ # # # # ]: 0 : LogDebug(BCLog::ZMQ, "Publish hashtx mempool acceptance %s to %s\n", hash.GetHex(), this->address);
286 : 0 : return SendSequenceMsg(*this, hash, /* Mempool (A)cceptance */ 'A', mempool_sequence);
287 : : }
288 : :
289 : 0 : bool CZMQPublishSequenceNotifier::NotifyTransactionRemoval(const CTransaction &transaction, uint64_t mempool_sequence)
290 : : {
291 : 0 : uint256 hash = transaction.GetHash().ToUint256();
292 [ # # # # ]: 0 : LogDebug(BCLog::ZMQ, "Publish hashtx mempool removal %s to %s\n", hash.GetHex(), this->address);
293 : 0 : return SendSequenceMsg(*this, hash, /* Mempool (R)emoval */ 'R', mempool_sequence);
294 : : }
|
