ZeroMQ support for solo mining

Gets new blocks from daemon immediately without polling, saving ~0.5 seconds on average when daemon gets new block from the network. Also saves some CPU cycles because it doesn't need to poll daemon every second.

Testing: add "daemon-zmq-port": 28083 to xmrig's pool config in config.json and run ./monerod --testnet --zmq-pub tcp://127.0.0.1:28083

src/base/base.cmake

@@ -70,6 +70,7 @@ set(HEADERS_BASE
     src/base/net/tools/Storage.h
     src/base/tools/Arguments.h
     src/base/tools/Baton.h
+    src/base/tools/bswap_64.h
     src/base/tools/Buffer.h
     src/base/tools/Chrono.h
     src/base/tools/Cvt.h

src/base/kernel/config/BaseTransform.cpp

@@ -242,13 +242,14 @@ void xmrig::BaseTransform::transform(rapidjson::Document &doc, int key, const ch
         return set(doc, BaseConfig::kTls, TlsConfig::kGen, arg);
 #   endif
 
-    case IConfig::RetriesKey:     /* --retries */
-    case IConfig::RetryPauseKey:  /* --retry-pause */
-    case IConfig::PrintTimeKey:   /* --print-time */
-    case IConfig::HttpPort:       /* --http-port */
-    case IConfig::DonateLevelKey: /* --donate-level */
-    case IConfig::DaemonPollKey:  /* --daemon-poll-interval */
-    case IConfig::DnsTtlKey:      /* --dns-ttl */
+    case IConfig::RetriesKey:       /* --retries */
+    case IConfig::RetryPauseKey:    /* --retry-pause */
+    case IConfig::PrintTimeKey:     /* --print-time */
+    case IConfig::HttpPort:         /* --http-port */
+    case IConfig::DonateLevelKey:   /* --donate-level */
+    case IConfig::DaemonPollKey:    /* --daemon-poll-interval */
+    case IConfig::DnsTtlKey:        /* --dns-ttl */
+    case IConfig::DaemonZMQPortKey: /* --daemon-zmq-port */
         return transformUint64(doc, key, static_cast<uint64_t>(strtol(arg, nullptr, 10)));
 
     case IConfig::BackgroundKey:  /* --background */
@@ -359,6 +360,9 @@ void xmrig::BaseTransform::transformUint64(rapidjson::Document &doc, int key, ui
 #   ifdef XMRIG_FEATURE_HTTP
     case IConfig::DaemonPollKey:  /* --daemon-poll-interval */
         return add(doc, Pools::kPools, Pool::kDaemonPollInterval, arg);
+
+    case IConfig::DaemonZMQPortKey:  /* --daemon-zmq-port */
+        return add(doc, Pools::kPools, Pool::kDaemonZMQPort, arg);
 #   endif
 
     default:

src/base/kernel/interfaces/IConfig.h

@@ -85,6 +85,7 @@ public:
         DnsIPv6Key           = 1053,
         DnsTtlKey            = 1054,
         SpendSecretKey       = 1055,
+        DaemonZMQPortKey     = 1056,
 
         // xmrig common
         CPUPriorityKey       = 1021,

src/base/net/stratum/DaemonClient.cpp

@@ -31,10 +31,14 @@
 #include "base/io/json/JsonRequest.h"
 #include "base/io/log/Log.h"
 #include "base/kernel/interfaces/IClientListener.h"
+#include "base/net/dns/Dns.h"
+#include "base/net/dns/DnsRecords.h"
 #include "base/net/http/Fetch.h"
 #include "base/net/http/HttpData.h"
 #include "base/net/http/HttpListener.h"
 #include "base/net/stratum/SubmitResult.h"
+#include "base/net/tools/NetBuffer.h"
+#include "base/tools/bswap_64.h"
 #include "base/tools/Cvt.h"
 #include "base/tools/Timer.h"
 #include "base/tools/cryptonote/Signatures.h"
@@ -48,7 +52,11 @@
 
 namespace xmrig {
 
-static const char *kBlocktemplateBlob       = "blocktemplate_blob";
+
+Storage<DaemonClient> DaemonClient::m_storage;
+
+
+static const char* kBlocktemplateBlob = "blocktemplate_blob";
 static const char *kGetHeight               = "/getheight";
 static const char *kGetInfo                 = "/getinfo";
 static const char *kHash                    = "hash";
@@ -57,6 +65,12 @@ static const char *kJsonRPC                 = "/json_rpc";
 
 static constexpr size_t kBlobReserveSize    = 8;
 
+static const char kZMQGreeting[64] = { -1, 0, 0, 0, 0, 0, 0, 0, 0, 127, 3, 0, 'N', 'U', 'L', 'L' };
+static constexpr size_t kZMQGreetingSize1 = 11;
+
+static const char kZMQHandshake[] = "\4\x19\5READY\xbSocket-Type\0\0\0\3SUB";
+static const char kZMQSubscribe[] = "\0\x18\1json-minimal-chain_main";
+
 }
 
 
@@ -65,12 +79,25 @@ xmrig::DaemonClient::DaemonClient(int id, IClientListener *listener) :
 {
     m_httpListener  = std::make_shared<HttpListener>(this);
     m_timer         = new Timer(this);
+    m_key           = m_storage.add(this);
 }
 
 
 xmrig::DaemonClient::~DaemonClient()
 {
     delete m_timer;
+    delete m_ZMQSocket;
+}
+
+
+void xmrig::DaemonClient::deleteLater()
+{
+    if (m_pool.zmq_port() >= 0) {
+        ZMQClose(true);
+    }
+    else {
+        delete this;
+    }
 }
 
 
@@ -159,7 +186,13 @@ void xmrig::DaemonClient::connect()
     }
 
     setState(ConnectingState);
-    getBlockTemplate();
+
+    if (m_pool.zmq_port() >= 0) {
+        m_dns = Dns::resolve(m_pool.host(), this);
+    }
+    else {
+        getBlockTemplate();
+    }
 }
 
 
@@ -238,7 +271,7 @@ void xmrig::DaemonClient::onHttpData(const HttpData &data)
 void xmrig::DaemonClient::onTimer(const Timer *)
 {
     if (m_state == ConnectingState) {
-        getBlockTemplate();
+        connect();
     }
     else if (m_state == ConnectedState) {
         if (m_apiVersion == API_DERO) {
@@ -251,6 +284,43 @@ void xmrig::DaemonClient::onTimer(const Timer *)
 }
 
 
+void xmrig::DaemonClient::onResolved(const DnsRecords& records, int status, const char* error)
+{
+    m_dns.reset();
+
+    if (status < 0 && records.isEmpty()) {
+        if (!isQuiet()) {
+            LOG_ERR("%s " RED("DNS error: ") RED_BOLD("\"%s\""), tag(), error);
+        }
+
+        retry();
+        return;
+    }
+
+    if (m_ZMQSocket) {
+        delete m_ZMQSocket;
+    }
+
+    const auto& record = records.get();
+    m_ip = record.ip();
+
+    uv_connect_t* req = new uv_connect_t;
+    req->data = m_storage.ptr(m_key);
+
+    m_ZMQSocket = new uv_tcp_t;
+    m_ZMQSocket->data = m_storage.ptr(m_key);
+
+    uv_tcp_init(uv_default_loop(), m_ZMQSocket);
+    uv_tcp_nodelay(m_ZMQSocket, 1);
+
+#   ifndef WIN32
+    uv_tcp_keepalive(m_ZMQSocket, 1, 60);
+#   endif
+
+    uv_tcp_connect(req, m_ZMQSocket, record.addr(m_pool.zmq_port()), onZMQConnect);
+}
+
+
 bool xmrig::DaemonClient::isOutdated(uint64_t height, const char *hash) const
 {
     return m_job.height() != height || m_prevHash != hash;
@@ -452,7 +522,9 @@ bool xmrig::DaemonClient::parseResponse(int64_t id, const rapidjson::Value &resu
     }
 
     if (handleSubmitResponse(id, error_msg)) {
-        getBlockTemplate();
+        if (error_msg || (m_pool.zmq_port() < 0)) {
+            getBlockTemplate();
+        }
         return true;
     }
 
@@ -504,6 +576,10 @@ void xmrig::DaemonClient::retry()
         setState(ConnectingState);
     }
 
+    if ((m_ZMQConnectionState != ZMQ_NOT_CONNECTED) && (m_ZMQConnectionState != ZMQ_DISCONNECTING)) {
+        uv_close(reinterpret_cast<uv_handle_t*>(m_ZMQSocket), onZMQClose);
+    }
+
     m_timer->stop();
     m_timer->start(m_retryPause, 0);
 }
@@ -531,8 +607,10 @@ void xmrig::DaemonClient::setState(SocketState state)
             m_failures = 0;
             m_listener->onLoginSuccess(this);
 
-            const uint64_t interval = std::max<uint64_t>(20, m_pool.pollInterval());
-            m_timer->start(interval, interval);
+            if (m_pool.zmq_port() < 0) {
+                const uint64_t interval = std::max<uint64_t>(20, m_pool.pollInterval());
+                m_timer->start(interval, interval);
+            }
         }
         break;
 
@@ -545,3 +623,282 @@ void xmrig::DaemonClient::setState(SocketState state)
         break;
     }
 }
+
+
+void xmrig::DaemonClient::onZMQConnect(uv_connect_t* req, int status)
+{
+    DaemonClient* client = getClient(req->data);
+    delete req;
+
+    if (!client) {
+        return;
+    }
+
+    if (status < 0) {
+        LOG_ERR("%s " RED("ZMQ connect error: ") RED_BOLD("\"%s\""), client->tag(), uv_strerror(status));
+        client->retry();
+        return;
+    }
+
+    client->ZMQConnected();
+}
+
+
+void xmrig::DaemonClient::onZMQRead(uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf)
+{
+    DaemonClient* client = getClient(stream->data);
+    if (client) {
+        client->ZMQRead(nread, buf);
+    }
+
+    NetBuffer::release(buf);
+}
+
+
+void xmrig::DaemonClient::onZMQClose(uv_handle_t* handle)
+{
+    DaemonClient* client = getClient(handle->data);
+    if (client) {
+#       ifdef APP_DEBUG
+        LOG_DEBUG(CYAN("tcp-zmq://%s:%u") BLACK_BOLD(" disconnected"), client->m_pool.host().data(), client->m_pool.zmq_port());
+#       endif
+        client->m_ZMQConnectionState = ZMQ_NOT_CONNECTED;
+    }
+}
+
+
+void xmrig::DaemonClient::onZMQShutdown(uv_handle_t* handle)
+{
+    DaemonClient* client = getClient(handle->data);
+    if (client) {
+#       ifdef APP_DEBUG
+        LOG_DEBUG(CYAN("tcp-zmq://%s:%u") BLACK_BOLD(" shutdown"), client->m_pool.host().data(), client->m_pool.zmq_port());
+#       endif
+        client->m_ZMQConnectionState = ZMQ_NOT_CONNECTED;
+        m_storage.remove(client->m_key);
+    }
+}
+
+
+void xmrig::DaemonClient::ZMQConnected()
+{
+#   ifdef APP_DEBUG
+    LOG_DEBUG(CYAN("tcp-zmq://%s:%u") BLACK_BOLD(" connected"), m_pool.host().data(), m_pool.zmq_port());
+#   endif
+
+    m_ZMQConnectionState = ZMQ_GREETING_1;
+    m_ZMQSendBuf.reserve(256);
+    m_ZMQRecvBuf.reserve(256);
+
+    if (ZMQWrite(kZMQGreeting, kZMQGreetingSize1)) {
+        uv_read_start(reinterpret_cast<uv_stream_t*>(m_ZMQSocket), NetBuffer::onAlloc, onZMQRead);
+    }
+}
+
+
+bool xmrig::DaemonClient::ZMQWrite(const char* data, size_t size)
+{
+    m_ZMQSendBuf.assign(data, data + size);
+
+    uv_buf_t buf;
+    buf.base = m_ZMQSendBuf.data();
+    buf.len = static_cast<uint32_t>(m_ZMQSendBuf.size());
+
+    const int rc = uv_try_write(reinterpret_cast<uv_stream_t*>(m_ZMQSocket), &buf, 1);
+
+    if (static_cast<size_t>(rc) == buf.len) {
+        return true;
+    }
+
+    LOG_ERR("%s " RED("ZMQ write failed, rc = %d"), tag(), rc);
+    ZMQClose();
+    return false;
+}
+
+
+void xmrig::DaemonClient::ZMQRead(ssize_t nread, const uv_buf_t* buf)
+{
+    if (nread <= 0) {
+        LOG_ERR("%s " RED("ZMQ read failed, nread = %" PRId64), tag(), nread);
+        ZMQClose();
+        return;
+    }
+
+    m_ZMQRecvBuf.insert(m_ZMQRecvBuf.end(), buf->base, buf->base + nread);
+
+    do {
+        switch (m_ZMQConnectionState) {
+        case ZMQ_GREETING_1:
+            if (m_ZMQRecvBuf.size() >= kZMQGreetingSize1) {
+                if ((m_ZMQRecvBuf[0] == -1) && (m_ZMQRecvBuf[9] == 127) && (m_ZMQRecvBuf[10] == 3)) {
+                    ZMQWrite(kZMQGreeting + kZMQGreetingSize1, sizeof(kZMQGreeting) - kZMQGreetingSize1);
+                    m_ZMQConnectionState = ZMQ_GREETING_2;
+                    break;
+                }
+                else {
+                    LOG_ERR("%s " RED("ZMQ handshake failed: invalid greeting format"), tag());
+                    ZMQClose();
+                }
+            }
+            return;
+
+        case ZMQ_GREETING_2:
+            if (m_ZMQRecvBuf.size() >= sizeof(kZMQGreeting)) {
+                if (memcmp(m_ZMQRecvBuf.data() + 12, kZMQGreeting + 12, 20) == 0) {
+                    m_ZMQConnectionState = ZMQ_HANDSHAKE;
+                    m_ZMQRecvBuf.erase(m_ZMQRecvBuf.begin(), m_ZMQRecvBuf.begin() + sizeof(kZMQGreeting));
+
+                    ZMQWrite(kZMQHandshake, sizeof(kZMQHandshake) - 1);
+                    break;
+                }
+                else {
+                    LOG_ERR("%s " RED("ZMQ handshake failed: invalid greeting format 2"), tag());
+                    ZMQClose();
+                }
+            }
+            return;
+
+        case ZMQ_HANDSHAKE:
+            if (m_ZMQRecvBuf.size() >= 2) {
+                if (m_ZMQRecvBuf[0] != 4) {
+                    LOG_ERR("%s " RED("ZMQ handshake failed: invalid handshake format"), tag());
+                    ZMQClose();
+                    return;
+                }
+
+                const size_t size = static_cast<unsigned char>(m_ZMQRecvBuf[1]);
+                if (size < 18) {
+                    LOG_ERR("%s " RED("ZMQ handshake failed: invalid handshake size"), tag());
+                    ZMQClose();
+                    return;
+                }
+
+                if (m_ZMQRecvBuf.size() < size + 2) {
+                    return;
+                }
+
+                if (memcmp(m_ZMQRecvBuf.data() + 2, kZMQHandshake + 2, 18) != 0) {
+                    LOG_ERR("%s " RED("ZMQ handshake failed: invalid handshake data"), tag());
+                    ZMQClose();
+                    return;
+                }
+
+                ZMQWrite(kZMQSubscribe, sizeof(kZMQSubscribe) - 1);
+
+                m_ZMQConnectionState = ZMQ_CONNECTED;
+                m_ZMQRecvBuf.erase(m_ZMQRecvBuf.begin(), m_ZMQRecvBuf.begin() + size + 2);
+
+                getBlockTemplate();
+                break;
+            }
+            return;
+
+        case ZMQ_CONNECTED:
+            ZMQParse();
+            return;
+
+        default:
+            return;
+        }
+    } while (true);
+}
+
+
+void xmrig::DaemonClient::ZMQParse()
+{
+#   ifdef APP_DEBUG
+    std::vector<char> msg;
+#   endif
+
+    size_t msg_size = 0;
+
+    char* data = m_ZMQRecvBuf.data();
+    size_t avail = m_ZMQRecvBuf.size();
+    bool more;
+
+    do {
+        if (avail < 1) {
+            return;
+        }
+
+        more                 = (data[0] & 1) != 0;
+        const bool long_size = (data[0] & 2) != 0;
+        const bool command   = (data[0] & 4) != 0;
+
+        ++data;
+        --avail;
+
+        uint64_t size = 0;
+        if (long_size)
+        {
+            if (avail < sizeof(uint64_t)) {
+                return;
+            }
+            size = bswap_64(*((uint64_t*)data));
+            data += sizeof(uint64_t);
+            avail -= sizeof(uint64_t);
+        }
+        else
+        {
+            if (avail < sizeof(uint8_t)) {
+                return;
+            }
+            size = static_cast<uint8_t>(*data);
+            ++data;
+            --avail;
+        }
+
+        if (size > 1024U - msg_size)
+        {
+            LOG_ERR("%s " RED("ZMQ message is too large, size = %" PRIu64 " bytes"), tag(), size);
+            ZMQClose();
+            return;
+        }
+
+        if (avail < size) {
+            return;
+        }
+
+        if (!command) {
+#           ifdef APP_DEBUG
+            msg.insert(msg.end(), data, data + size);
+#           endif
+
+            msg_size += size;
+        }
+
+        data += size;
+        avail -= size;
+    } while (more);
+
+    m_ZMQRecvBuf.erase(m_ZMQRecvBuf.begin(), m_ZMQRecvBuf.begin() + (data - m_ZMQRecvBuf.data()));
+
+#   ifdef APP_DEBUG
+    LOG_DEBUG(CYAN("tcp-zmq://%s:%u") BLACK_BOLD(" read ") CYAN_BOLD("%zu") BLACK_BOLD(" bytes") " %s", m_pool.host().data(), m_pool.zmq_port(), msg.size(), msg.data());
+#   endif
+
+    getBlockTemplate();
+}
+
+
+bool xmrig::DaemonClient::ZMQClose(bool shutdown)
+{
+    if ((m_ZMQConnectionState == ZMQ_NOT_CONNECTED) || (m_ZMQConnectionState == ZMQ_DISCONNECTING)) {
+        if (shutdown) {
+            m_storage.remove(m_key);
+        }
+        return false;
+    }
+
+    m_ZMQConnectionState = ZMQ_DISCONNECTING;
+
+    if (uv_is_closing(reinterpret_cast<uv_handle_t*>(m_ZMQSocket)) == 0) {
+        uv_close(reinterpret_cast<uv_handle_t*>(m_ZMQSocket), shutdown ? onZMQShutdown : onZMQClose);
+        if (!shutdown) {
+            retry();
+        }
+        return true;
+    }
+
+    return false;
+}

src/base/net/stratum/DaemonClient.h

@@ -27,11 +27,16 @@
 #define XMRIG_DAEMONCLIENT_H
 
 
+#include <uv.h>
+
+
+#include "base/kernel/interfaces/IDnsListener.h"
 #include "base/kernel/interfaces/IHttpListener.h"
 #include "base/kernel/interfaces/ITimerListener.h"
 #include "base/net/stratum/BaseClient.h"
 #include "base/tools/Object.h"
 #include "base/tools/cryptonote/BlockTemplate.h"
+#include "base/net/tools/Storage.h"
 
 
 #include <memory>
@@ -40,7 +45,10 @@
 namespace xmrig {
 
 
-class DaemonClient : public BaseClient, public ITimerListener, public IHttpListener
+class DnsRequest;
+
+
+class DaemonClient : public BaseClient, public IDnsListener, public ITimerListener, public IHttpListener
 {
 public:
     XMRIG_DISABLE_COPY_MOVE_DEFAULT(DaemonClient)
@@ -57,6 +65,7 @@ protected:
 
     void onHttpData(const HttpData &data) override;
     void onTimer(const Timer *timer) override;
+    void onResolved(const DnsRecords& records, int status, const char* error) override;
 
     inline bool hasExtension(Extension) const noexcept override         { return false; }
     inline const char *mode() const override                            { return "daemon"; }
@@ -64,7 +73,7 @@ protected:
     inline const char *tlsVersion() const override                      { return m_tlsVersion; }
     inline int64_t send(const rapidjson::Value &, Callback) override    { return -1; }
     inline int64_t send(const rapidjson::Value &) override              { return -1; }
-    inline void deleteLater() override                                  { delete this; }
+    void deleteLater() override;
     inline void tick(uint64_t) override                                 {}
 
 private:
@@ -95,6 +104,38 @@ private:
     String m_blocktemplateRequestHash;
 
     BlockTemplate m_blocktemplate;
+
+private:
+    static inline DaemonClient* getClient(void* data) { return m_storage.get(data); }
+
+    uintptr_t m_key = 0;
+    static Storage<DaemonClient> m_storage;
+
+    static void onZMQConnect(uv_connect_t* req, int status);
+    static void onZMQRead(uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf);
+    static void onZMQClose(uv_handle_t* handle);
+    static void onZMQShutdown(uv_handle_t* handle);
+
+    void ZMQConnected();
+    bool ZMQWrite(const char* data, size_t size);
+    void ZMQRead(ssize_t nread, const uv_buf_t* buf);
+    void ZMQParse();
+    bool ZMQClose(bool shutdown = false);
+
+    std::shared_ptr<DnsRequest> m_dns;
+    uv_tcp_t* m_ZMQSocket = nullptr;
+
+    enum {
+        ZMQ_NOT_CONNECTED,
+        ZMQ_GREETING_1,
+        ZMQ_GREETING_2,
+        ZMQ_HANDSHAKE,
+        ZMQ_CONNECTED,
+        ZMQ_DISCONNECTING,
+    } m_ZMQConnectionState = ZMQ_NOT_CONNECTED;
+
+    std::vector<char> m_ZMQSendBuf;
+    std::vector<char> m_ZMQRecvBuf;
 };
 
 

src/base/net/stratum/Pool.cpp

@@ -66,6 +66,7 @@ const char *Pool::kAlgo                   = "algo";
 const char *Pool::kCoin                   = "coin";
 const char *Pool::kDaemon                 = "daemon";
 const char *Pool::kDaemonPollInterval     = "daemon-poll-interval";
+const char *Pool::kDaemonZMQPort          = "daemon-zmq-port";
 const char *Pool::kEnabled                = "enabled";
 const char *Pool::kFingerprint            = "tls-fingerprint";
 const char *Pool::kKeepalive              = "keepalive";
@@ -127,6 +128,7 @@ xmrig::Pool::Pool(const rapidjson::Value &object) :
     m_coin           = Json::getString(object, kCoin);
     m_daemon         = Json::getString(object, kSelfSelect);
     m_proxy          = Json::getValue(object, kSOCKS5);
+    m_zmqPort        = Json::getInt(object, kDaemonZMQPort, m_zmqPort);
 
     m_flags.set(FLAG_ENABLED,  Json::getBool(object, kEnabled, true));
     m_flags.set(FLAG_NICEHASH, Json::getBool(object, kNicehash) || m_url.host().contains(kNicehashHost));
@@ -301,6 +303,7 @@ rapidjson::Value xmrig::Pool::toJSON(rapidjson::Document &doc) const
 
     if (m_mode == MODE_DAEMON) {
         obj.AddMember(StringRef(kDaemonPollInterval), m_pollInterval, allocator);
+        obj.AddMember(StringRef(kDaemonZMQPort), m_zmqPort, allocator);
     }
     else {
         obj.AddMember(StringRef(kSelfSelect),     m_daemon.url().toJSON(), allocator);
@@ -336,6 +339,9 @@ void xmrig::Pool::print() const
     LOG_NOTICE("url:       %s", url().data());
     LOG_DEBUG ("host:      %s", host().data());
     LOG_DEBUG ("port:      %d", static_cast<int>(port()));
+    if (m_zmqPort >= 0) {
+        LOG_DEBUG("zmq-port:  %d", m_zmqPort);
+    }
     LOG_DEBUG ("user:      %s", m_user.data());
     LOG_DEBUG ("pass:      %s", m_password.data());
     LOG_DEBUG ("rig-id     %s", m_rigId.data());

src/base/net/stratum/Pool.h

@@ -72,6 +72,7 @@ public:
     static const char *kUrl;
     static const char *kUser;
     static const char* kSpendSecretKey;
+    static const char* kDaemonZMQPort;
     static const char *kNicehashHost;
 
     constexpr static int kKeepAliveTimeout         = 60;
@@ -107,6 +108,7 @@ public:
     inline int keepAlive() const                        { return m_keepAlive; }
     inline Mode mode() const                            { return m_mode; }
     inline uint16_t port() const                        { return m_url.port(); }
+    inline int zmq_port() const                         { return m_zmqPort; }
     inline uint64_t pollInterval() const                { return m_pollInterval; }
     inline void setAlgo(const Algorithm &algorithm)     { m_algorithm = algorithm; }
     inline void setPassword(const String &password)     { m_password = password; }
@@ -155,6 +157,7 @@ private:
     uint64_t m_pollInterval         = kDefaultPollInterval;
     Url m_daemon;
     Url m_url;
+    int m_zmqPort                   = -1;
 
 #   ifdef XMRIG_FEATURE_BENCHMARK
     std::shared_ptr<BenchConfig> m_benchmark;

src/base/tools/bswap_64.h

@@ -0,0 +1,37 @@
+/* XMRig
+ * Copyright (c) 2018-2021 SChernykh   <https://github.com/SChernykh>
+ * Copyright (c) 2016-2021 XMRig       <https://github.com/xmrig>, <support@xmrig.com>
+ *
+ *   This program is free software: you can redistribute it and/or modify
+ *   it under the terms of the GNU General Public License as published by
+ *   the Free Software Foundation, either version 3 of the License, or
+ *   (at your option) any later version.
+ *
+ *   This program is distributed in the hope that it will be useful,
+ *   but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ *   GNU General Public License for more details.
+ *
+ *   You should have received a copy of the GNU General Public License
+ *   along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef XMRIG_BSWAP_64_H
+#define XMRIG_BSWAP_64_H
+
+#ifdef _MSC_VER
+
+#include <stdlib.h>
+#define bswap_64(x) _byteswap_uint64(x)
+
+#elif defined __GNUC__
+
+#define bswap_64(x) __builtin_bswap64(x)
+
+#else
+
+#include <byteswap.h>
+
+#endif
+
+#endif /* XMRIG_BSWAP_64_H */

src/base/tools/cryptonote/BlockTemplate.cpp

@@ -96,7 +96,7 @@ bool BlockTemplate::Init(const String& blockTemplate, Coin coin)
     tx_extra_nonce_index = 0;
 
     while (ar_extra.index() < extra_size) {
-        uint64_t extra_tag;
+        uint64_t extra_tag = 0;
         ar_extra(extra_tag);
 
         switch (extra_tag) {

src/core/config/Config_platform.h

@@ -52,6 +52,7 @@ static const option options[] = {
     { "daemon-poll-interval",  1, nullptr, IConfig::DaemonPollKey         },
     { "self-select",           1, nullptr, IConfig::SelfSelectKey         },
     { "submit-to-origin",      0, nullptr, IConfig::SubmitToOriginKey     },
+    { "daemon-zmq-port",       1, nullptr, IConfig::DaemonZMQPortKey      },
 #   endif
     { "av",                    1, nullptr, IConfig::AVKey                 },
     { "background",            0, nullptr, IConfig::BackgroundKey         },

src/crypto/astrobwt/AstroBWT.cpp

@@ -30,6 +30,7 @@
 #include "crypto/astrobwt/AstroBWT.h"
 #include "backend/cpu/Cpu.h"
 #include "base/crypto/sha3.h"
+#include "base/tools/bswap_64.h"
 #include "crypto/cn/CryptoNight.h"
 
 
@@ -54,21 +55,6 @@ __attribute__((ms_abi))
 void SHA3_256_AVX2_ASM(const void* in, size_t inBytes, void* out);
 #endif
 
-#ifdef _MSC_VER
-
-#include <stdlib.h>
-#define bswap_64(x) _byteswap_uint64(x)
-
-#elif defined __GNUC__
-
-#define bswap_64(x) __builtin_bswap64(x)
-
-#else
-
-#include <byteswap.h>
-
-#endif
-
 #ifdef XMRIG_ARM
 extern "C" {
 #include "salsa20_ref/ecrypt-sync.h"

src/net/Network.cpp

@@ -137,9 +137,14 @@ void xmrig::Network::onActive(IStrategy *strategy, IClient *client)
     }
 #   endif
 
+    char zmq_buf[32] = {};
+    if (client->pool().zmq_port() >= 0) {
+        snprintf(zmq_buf, sizeof(zmq_buf), " (ZMQ:%d)", client->pool().zmq_port());
+    }
+
     const char *tlsVersion = client->tlsVersion();
-    LOG_INFO("%s " WHITE_BOLD("use %s ") CYAN_BOLD("%s:%d ") GREEN_BOLD("%s") " " BLACK_BOLD("%s"),
-             Tags::network(), client->mode(), pool.host().data(), pool.port(), tlsVersion ? tlsVersion : "", client->ip().data());
+    LOG_INFO("%s " WHITE_BOLD("use %s ") CYAN_BOLD("%s:%d%s ") GREEN_BOLD("%s") " " BLACK_BOLD("%s"),
+             Tags::network(), client->mode(), pool.host().data(), pool.port(), zmq_buf, tlsVersion ? tlsVersion : "", client->ip().data());
 
     const char *fingerprint = client->tlsFingerprint();
     if (fingerprint != nullptr) {
@@ -272,8 +277,13 @@ void xmrig::Network::setJob(IClient *client, const Job &job, bool donate)
         uint64_t diff       = job.diff();;
         const char *scale   = NetworkState::scaleDiff(diff);
 
-        LOG_INFO("%s " MAGENTA_BOLD("new job") " from " WHITE_BOLD("%s:%d") " diff " WHITE_BOLD("%" PRIu64 "%s") " algo " WHITE_BOLD("%s") " height " WHITE_BOLD("%" PRIu64),
-                 Tags::network(), client->pool().host().data(), client->pool().port(), diff, scale, job.algorithm().shortName(), job.height());
+        char zmq_buf[32] = {};
+        if (client->pool().zmq_port() >= 0) {
+            snprintf(zmq_buf, sizeof(zmq_buf), " (ZMQ:%d)", client->pool().zmq_port());
+        }
+
+        LOG_INFO("%s " MAGENTA_BOLD("new job") " from " WHITE_BOLD("%s:%d%s") " diff " WHITE_BOLD("%" PRIu64 "%s") " algo " WHITE_BOLD("%s") " height " WHITE_BOLD("%" PRIu64),
+                 Tags::network(), client->pool().host().data(), client->pool().port(), zmq_buf, diff, scale, job.algorithm().shortName(), job.height());
     }
 
     if (!donate && m_donate) {