LBRYCommunity/lbrycrd
1
0
Fork 0
Code Issues 29 Pull requests 1 Projects Releases 32 Packages Wiki Activity

ZMQ: add options to configure outbound message high water mark, aka SNDHWM

Browse source
This commit is contained in:
mruddy 2018-08-24 20:42:03 -04:00
parent d387507aec
commit a4edb168b6
11 changed files with 59 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
contrib/zmq/zmq_sub.py
View file

@ -42,6 +42,7 @@ class ZMQHandler():
self.zmqContext = zmq.asyncio.Context()
self.zmqSubSocket = self.zmqContext.socket(zmq.SUB)
self.zmqSubSocket.setsockopt(zmq.RCVHWM, 0)
self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "hashblock")
self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "hashtx")
self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "rawblock")

1
contrib/zmq/zmq_sub3.4.py
View file

@ -46,6 +46,7 @@ class ZMQHandler():
self.zmqContext = zmq.asyncio.Context()
self.zmqSubSocket = self.zmqContext.socket(zmq.SUB)
self.zmqSubSocket.setsockopt(zmq.RCVHWM, 0)
self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "hashblock")
self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "hashtx")
self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "rawblock")

13
doc/zmq.md
View file

@ -66,10 +66,21 @@ Currently, the following notifications are supported:
The socket type is PUB and the address must be a valid ZeroMQ socket
address. The same address can be used in more than one notification.
The option to set the PUB socket's outbound message high water mark
(SNDHWM) may be set individually for each notification:
-zmqpubhashtxhwm=n
-zmqpubhashblockhwm=n
-zmqpubrawblockhwm=n
-zmqpubrawtxhwm=n
The high water mark value must be an integer greater than or equal to 0.
For instance:
$ bitcoind -zmqpubhashtx=tcp://127.0.0.1:28332 \
-zmqpubrawtx=ipc:///tmp/bitcoind.tx.raw
-zmqpubrawtx=ipc:///tmp/bitcoind.tx.raw \
-zmqpubhashtxhwm=10000
Each PUB notification has a topic and body, where the header
corresponds to the notification type. For instance, for the

9
src/init.cpp
View file

@ -62,6 +62,7 @@
#include <openssl/crypto.h>
#if ENABLE_ZMQ
#include <zmq/zmqabstractnotifier.h>
#include <zmq/zmqnotificationinterface.h>
#include <zmq/zmqrpc.h>
#endif
@ -418,11 +419,19 @@ void SetupServerArgs()
gArgs.AddArg("-zmqpubhashtx=<address>", "Enable publish hash transaction in <address>", false, OptionsCategory::ZMQ);
gArgs.AddArg("-zmqpubrawblock=<address>", "Enable publish raw block in <address>", false, OptionsCategory::ZMQ);
gArgs.AddArg("-zmqpubrawtx=<address>", "Enable publish raw transaction in <address>", false, OptionsCategory::ZMQ);
gArgs.AddArg("-zmqpubhashblockhwm=<n>", strprintf("Set publish hash block outbound message high water mark (default: %d)", CZMQAbstractNotifier::DEFAULT_ZMQ_SNDHWM), false, OptionsCategory::ZMQ);
gArgs.AddArg("-zmqpubhashtxhwm=<n>", strprintf("Set publish hash transaction outbound message high water mark (default: %d)", CZMQAbstractNotifier::DEFAULT_ZMQ_SNDHWM), false, OptionsCategory::ZMQ);
gArgs.AddArg("-zmqpubrawblockhwm=<n>", strprintf("Set publish raw block outbound message high water mark (default: %d)", CZMQAbstractNotifier::DEFAULT_ZMQ_SNDHWM), false, OptionsCategory::ZMQ);
gArgs.AddArg("-zmqpubrawtxhwm=<n>", strprintf("Set publish raw transaction outbound message high water mark (default: %d)", CZMQAbstractNotifier::DEFAULT_ZMQ_SNDHWM), false, OptionsCategory::ZMQ);
#else
hidden_args.emplace_back("-zmqpubhashblock=<address>");
hidden_args.emplace_back("-zmqpubhashtx=<address>");
hidden_args.emplace_back("-zmqpubrawblock=<address>");
hidden_args.emplace_back("-zmqpubrawtx=<address>");
hidden_args.emplace_back("-zmqpubhashblockhwm=<n>");
hidden_args.emplace_back("-zmqpubhashtxhwm=<n>");
hidden_args.emplace_back("-zmqpubrawblockhwm=<n>");
hidden_args.emplace_back("-zmqpubrawtxhwm=<n>");
#endif
gArgs.AddArg("-checkblocks=<n>", strprintf("How many blocks to check at startup (default: %u, 0 = all)", DEFAULT_CHECKBLOCKS), true, OptionsCategory::DEBUG_TEST);

1
src/zmq/zmqabstractnotifier.cpp
View file

@ -5,6 +5,7 @@
#include <zmq/zmqabstractnotifier.h>
#include <util.h>
const int CZMQAbstractNotifier::DEFAULT_ZMQ_SNDHWM;
CZMQAbstractNotifier::~CZMQAbstractNotifier()
{

11
src/zmq/zmqabstractnotifier.h
View file

@ -15,7 +15,9 @@ typedef CZMQAbstractNotifier* (*CZMQNotifierFactory)();
class CZMQAbstractNotifier
{
public:
CZMQAbstractNotifier() : psocket(nullptr) { }
static const int DEFAULT_ZMQ_SNDHWM {1000};
CZMQAbstractNotifier() : psocket(nullptr), outbound_message_high_water_mark(DEFAULT_ZMQ_SNDHWM) { }
virtual ~CZMQAbstractNotifier();
template <typename T>
@ -28,6 +30,12 @@ public:
void SetType(const std::string &t) { type = t; }
std::string GetAddress() const { return address; }
void SetAddress(const std::string &a) { address = a; }
int GetOutboundMessageHighWaterMark() const { return outbound_message_high_water_mark; }
void SetOutboundMessageHighWaterMark(const int sndhwm) {
if (sndhwm >= 0) {
outbound_message_high_water_mark = sndhwm;
}
}
virtual bool Initialize(void *pcontext) = 0;
virtual void Shutdown() = 0;
@ -39,6 +47,7 @@ protected:
void *psocket;
std::string type;
std::string address;
int outbound_message_high_water_mark; // aka SNDHWM
};
#endif // BITCOIN_ZMQ_ZMQABSTRACTNOTIFIER_H

7
src/zmq/zmqnotificationinterface.cpp
View file

@ -59,6 +59,7 @@ CZMQNotificationInterface* CZMQNotificationInterface::Create()
CZMQAbstractNotifier *notifier = factory();
notifier->SetType(entry.first);
notifier->SetAddress(address);
notifier->SetOutboundMessageHighWaterMark(static_cast<int>(gArgs.GetArg(arg + "hwm", CZMQAbstractNotifier::DEFAULT_ZMQ_SNDHWM)));
notifiers.push_back(notifier);
}
}
@ -102,11 +103,11 @@ bool CZMQNotificationInterface::Initialize()
CZMQAbstractNotifier *notifier = *i;
if (notifier->Initialize(pcontext))
{
LogPrint(BCLog::ZMQ, " Notifier %s ready (address = %s)\n", notifier->GetType(), notifier->GetAddress());
LogPrint(BCLog::ZMQ, "zmq: Notifier %s ready (address = %s)\n", notifier->GetType(), notifier->GetAddress());
}
else
{
LogPrint(BCLog::ZMQ, " Notifier %s failed (address = %s)\n", notifier->GetType(), notifier->GetAddress());
LogPrint(BCLog::ZMQ, "zmq: Notifier %s failed (address = %s)\n", notifier->GetType(), notifier->GetAddress());
break;
}
}
@ -128,7 +129,7 @@ void CZMQNotificationInterface::Shutdown()
for (std::list<CZMQAbstractNotifier*>::iterator i=notifiers.begin(); i!=notifiers.end(); ++i)
{
CZMQAbstractNotifier *notifier = *i;
LogPrint(BCLog::ZMQ, " Shutdown notifier %s at %s\n", notifier->GetType(), notifier->GetAddress());
LogPrint(BCLog::ZMQ, "zmq: Shutdown notifier %s at %s\n", notifier->GetType(), notifier->GetAddress());
notifier->Shutdown();
}
zmq_ctx_term(pcontext);

14
src/zmq/zmqpublishnotifier.cpp
View file

@ -76,7 +76,17 @@ bool CZMQAbstractPublishNotifier::Initialize(void *pcontext)
return false;
}
int rc = zmq_bind(psocket, address.c_str());
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");
@ -120,7 +130,7 @@ void CZMQAbstractPublishNotifier::Shutdown()
if (count == 1)
{
LogPrint(BCLog::ZMQ, "Close socket at address %s\n", address);
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);

2
src/zmq/zmqpublishnotifier.h
View file

@ -12,7 +12,7 @@ class CBlockIndex;
class CZMQAbstractPublishNotifier : public CZMQAbstractNotifier
{
private:
uint32_t nSequence; //!< upcounting per message sequence number
uint32_t nSequence {0U}; //!< upcounting per message sequence number
public:

4
src/zmq/zmqrpc.cpp
View file

@ -22,7 +22,8 @@ UniValue getzmqnotifications(const JSONRPCRequest& request)
"[\n"
" { (json object)\n"
" \"type\": \"pubhashtx\", (string) Type of notification\n"
" \"address\": \"...\" (string) Address of the publisher\n"
" \"address\": \"...\", (string) Address of the publisher\n"
" \"hwm\": n (numeric) Outbound message high water mark\n"
" },\n"
" ...\n"
"]\n"
@ -38,6 +39,7 @@ UniValue getzmqnotifications(const JSONRPCRequest& request)
UniValue obj(UniValue::VOBJ);
obj.pushKV("type", n->GetType());
obj.pushKV("address", n->GetAddress());
obj.pushKV("hwm", n->GetOutboundMessageHighWaterMark());
result.push_back(obj);
}
}

8
test/functional/interface_zmq.py
View file

@ -121,10 +121,10 @@ class ZMQTest (BitcoinTestFramework):
self.log.info("Test the getzmqnotifications RPC")
assert_equal(self.nodes[0].getzmqnotifications(), [
{"type": "pubhashblock", "address": ADDRESS},
{"type": "pubhashtx", "address": ADDRESS},
{"type": "pubrawblock", "address": ADDRESS},
{"type": "pubrawtx", "address": ADDRESS},
{"type": "pubhashblock", "address": ADDRESS, "hwm": 1000},
{"type": "pubhashtx", "address": ADDRESS, "hwm": 1000},
{"type": "pubrawblock", "address": ADDRESS, "hwm": 1000},
{"type": "pubrawtx", "address": ADDRESS, "hwm": 1000},
])
assert_equal(self.nodes[1].getzmqnotifications(), [])