Added class RxQueue, class Rx now thin static wrapper on top of RxQueue.

cmake/randomx.cmake

@@ -8,6 +8,7 @@ if (WITH_RANDOMX)
         src/crypto/rx/RxCache.h
         src/crypto/rx/RxConfig.h
         src/crypto/rx/RxDataset.h
+        src/crypto/rx/RxQueue.h
         src/crypto/rx/RxSeed.h
         src/crypto/rx/RxVm.h
     )
@@ -38,6 +39,7 @@ if (WITH_RANDOMX)
         src/crypto/rx/RxCache.cpp
         src/crypto/rx/RxConfig.cpp
         src/crypto/rx/RxDataset.cpp
+        src/crypto/rx/RxQueue.cpp
         src/crypto/rx/RxVm.cpp
     )
 

src/core/Miner.cpp

@@ -72,12 +72,8 @@ class MinerPrivate
 public:
     XMRIG_DISABLE_COPY_MOVE_DEFAULT(MinerPrivate)
 
-    inline MinerPrivate(Controller *controller) : controller(controller)
+
-    {
+    inline MinerPrivate(Controller *controller) : controller(controller) {}
-#       ifdef XMRIG_ALGO_RANDOMX
-        Rx::init();
-#       endif
-    }
 
 
     inline ~MinerPrivate()
@@ -232,9 +228,9 @@ public:
 
 
 #   ifdef XMRIG_ALGO_RANDOMX
-    bool initRX(IRxListener *listener)
+    inline bool initRX()
     {
-        return Rx::init(job, controller->config()->rx(), controller->config()->cpu().isHugePages(), listener);
+        return Rx::init(job, controller->config()->rx(), controller->config()->cpu().isHugePages());
     }
 #   endif
 
@@ -261,6 +257,10 @@ public:
 xmrig::Miner::Miner(Controller *controller)
     : d_ptr(new MinerPrivate(controller))
 {
+#   ifdef XMRIG_ALGO_RANDOMX
+    Rx::init(this);
+#   endif
+
     controller->addListener(this);
 
 #   ifdef XMRIG_FEATURE_API
@@ -402,7 +402,7 @@ void xmrig::Miner::setJob(const Job &job, bool donate)
     }
 
 #   ifdef XMRIG_ALGO_RANDOMX
-    const bool ready = d_ptr->initRX(this);
+    const bool ready = d_ptr->initRX();
 #   else
     constexpr const bool ready = true;
 #   endif

src/crypto/rx/Rx.cpp

@@ -26,36 +26,10 @@
 
 
 #include "crypto/rx/Rx.h"
-
-#include "backend/common/interfaces/IRxListener.h"
-#include "backend/common/interfaces/IRxStorage.h"
 #include "backend/common/Tags.h"
-#include "backend/cpu/Cpu.h"
 #include "base/io/log/Log.h"
-#include "base/kernel/Platform.h"
-#include "base/net/stratum/Job.h"
-#include "base/tools/Buffer.h"
-#include "base/tools/Chrono.h"
-#include "base/tools/Handle.h"
-#include "base/tools/Object.h"
-#include "crypto/rx/RxAlgo.h"
-#include "crypto/rx/RxBasicStorage.h"
-#include "crypto/rx/RxCache.h"
 #include "crypto/rx/RxConfig.h"
-#include "crypto/rx/RxDataset.h"
+#include "crypto/rx/RxQueue.h"
-#include "crypto/rx/RxSeed.h"
-
-
-#ifdef XMRIG_FEATURE_HWLOC
-#   include "crypto/rx/RxNUMAStorage.h"
-#endif
-
-
-#include <atomic>
-#include <map>
-#include <mutex>
-#include <thread>
-#include <uv.h>
 
 
 namespace xmrig {
@@ -66,104 +40,14 @@ class RxPrivate;
 
 static const char *tag  = BLUE_BG(WHITE_BOLD_S " rx  ") " ";
 static RxPrivate *d_ptr = nullptr;
-static std::mutex mutex;
 
 
 class RxPrivate
 {
 public:
-    XMRIG_DISABLE_COPY_MOVE(RxPrivate)
+    inline RxPrivate(IRxListener *listener) : queue(listener) {}
 
-    inline RxPrivate() :
+    RxQueue queue;
-        m_pending(0)
-    {
-        m_async = new uv_async_t;
-        m_async->data = this;
-
-        uv_async_init(uv_default_loop(), m_async, [](uv_async_t *handle) { static_cast<RxPrivate *>(handle->data)->onReady(); });
-    }
-
-
-    inline ~RxPrivate()
-    {
-        m_pending = std::numeric_limits<uint32_t>::max();
-
-        std::lock_guard<std::mutex> lock(mutex);
-        Handle::close(m_async);
-
-        delete m_storage;
-    }
-
-
-    inline bool isReady(const Job &job) const                   { return pending() == 0 && m_seed == job; }
-    inline RxDataset *dataset(const Job &job, uint32_t nodeId)  { return m_storage ? m_storage->dataset(job, nodeId) : nullptr; }
-    inline std::pair<uint32_t, uint32_t> hugePages()            { return m_storage ? m_storage->hugePages() : std::pair<uint32_t, uint32_t>(0u, 0u); }
-    inline uint64_t pending() const                             { return m_pending.load(std::memory_order_relaxed); }
-    inline void asyncSend()                                     { --m_pending; if (pending() == 0) { uv_async_send(m_async); } }
-
-
-    inline IRxStorage *storage(const std::vector<uint32_t> &nodeset)
-    {
-        if (!m_storage) {
-#           ifdef XMRIG_FEATURE_HWLOC
-            if (!nodeset.empty()) {
-                m_storage = new RxNUMAStorage(nodeset);
-            }
-            else
-#           endif
-            {
-                m_storage = new RxBasicStorage();
-            }
-        }
-
-        return m_storage;
-    }
-
-
-    static void initDataset(const RxSeed &seed, const std::vector<uint32_t> &nodeset, uint32_t threads, bool hugePages)
-    {
-        std::lock_guard<std::mutex> lock(mutex);
-
-        if (d_ptr->pending() > std::numeric_limits<uint16_t>::max()) {
-            return;
-        }
-
-        LOG_INFO("%s" MAGENTA_BOLD("init dataset%s") " algo " WHITE_BOLD("%s (") CYAN_BOLD("%u") WHITE_BOLD(" threads)") BLACK_BOLD(" seed %s..."),
-                 tag,
-                 nodeset.size() > 1 ? "s" : "",
-                 seed.algorithm().shortName(),
-                 threads,
-                 Buffer::toHex(seed.data().data(), 8).data()
-                 );
-
-        d_ptr->storage(nodeset)->init(seed, threads, hugePages);
-        d_ptr->asyncSend();
-    }
-
-
-    inline void setState(const Job &job, IRxListener *listener)
-    {
-        m_listener  = listener;
-        m_seed      = job;
-
-        ++m_pending;
-    }
-
-
-private:
-    inline void onReady()
-    {
-        if (m_listener && pending() == 0) {
-            m_listener->onDatasetReady();
-        }
-    }
-
-
-    IRxListener *m_listener = nullptr;
-    IRxStorage *m_storage   = nullptr;
-    RxSeed m_seed;
-    std::atomic<uint64_t> m_pending;
-    uv_async_t *m_async     = nullptr;
 };
 
 
@@ -176,20 +60,17 @@ const char *xmrig::rx_tag()
 }
 
 
-bool xmrig::Rx::init(const Job &job, const RxConfig &config, bool hugePages, IRxListener *listener)
+bool xmrig::Rx::init(const Job &job, const RxConfig &config, bool hugePages)
 {
     if (job.algorithm().family() != Algorithm::RANDOM_X) {
         return true;
     }
 
-    if (d_ptr->isReady(job)) {
+    if (isReady(job)) {
         return true;
     }
 
-    d_ptr->setState(job, listener);
+    d_ptr->queue.enqueue(job, config.nodeset(), config.threads(), hugePages);
-
-    std::thread thread(RxPrivate::initDataset, job, config.nodeset(), config.threads(), hugePages);
-    thread.detach();
 
     return false;
 }
@@ -197,23 +78,19 @@ bool xmrig::Rx::init(const Job &job, const RxConfig &config, bool hugePages, IRx
 
 bool xmrig::Rx::isReady(const Job &job)
 {
-    return d_ptr->isReady(job);
+    return d_ptr->queue.isReady(job);
 }
 
 
 xmrig::RxDataset *xmrig::Rx::dataset(const Job &job, uint32_t nodeId)
 {
-    std::lock_guard<std::mutex> lock(mutex);
+    return d_ptr->queue.dataset(job, nodeId);
-
-    return d_ptr->dataset(job, nodeId);
 }
 
 
 std::pair<uint32_t, uint32_t> xmrig::Rx::hugePages()
 {
-    std::lock_guard<std::mutex> lock(mutex);
+    return d_ptr->queue.hugePages();
-
-    return d_ptr->hugePages();
 }
 
 
@@ -225,7 +102,7 @@ void xmrig::Rx::destroy()
 }
 
 
-void xmrig::Rx::init()
+void xmrig::Rx::init(IRxListener *listener)
 {
-    d_ptr = new RxPrivate();
+    d_ptr = new RxPrivate(listener);
 }

src/crypto/rx/Rx.h

@@ -46,12 +46,12 @@ class RxDataset;
 class Rx
 {
 public:
-    static bool init(const Job &job, const RxConfig &config, bool hugePages, IRxListener *listener);
+    static bool init(const Job &job, const RxConfig &config, bool hugePages);
     static bool isReady(const Job &job);
     static RxDataset *dataset(const Job &job, uint32_t nodeId);
     static std::pair<uint32_t, uint32_t> hugePages();
     static void destroy();
-    static void init();
+    static void init(IRxListener *listener);
 };
 
 

src/crypto/rx/RxQueue.cpp

@@ -0,0 +1,181 @@
+/* XMRig
+ * Copyright 2010      Jeff Garzik <jgarzik@pobox.com>
+ * Copyright 2012-2014 pooler      <pooler@litecoinpool.org>
+ * Copyright 2014      Lucas Jones <https://github.com/lucasjones>
+ * Copyright 2014-2016 Wolf9466    <https://github.com/OhGodAPet>
+ * Copyright 2016      Jay D Dee   <jayddee246@gmail.com>
+ * Copyright 2017-2019 XMR-Stak    <https://github.com/fireice-uk>, <https://github.com/psychocrypt>
+ * Copyright 2018      Lee Clagett <https://github.com/vtnerd>
+ * Copyright 2018-2019 tevador     <tevador@gmail.com>
+ * Copyright 2018-2019 SChernykh   <https://github.com/SChernykh>
+ * Copyright 2016-2019 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/>.
+ */
+
+
+#include "crypto/rx/RxQueue.h"
+#include "backend/common/Tags.h"
+#include "base/io/log/Log.h"
+#include "crypto/rx/RxBasicStorage.h"
+#include "base/tools/Handle.h"
+#include "backend/common/interfaces/IRxListener.h"
+
+
+#ifdef XMRIG_FEATURE_HWLOC
+#   include "crypto/rx/RxNUMAStorage.h"
+#endif
+
+
+xmrig::RxQueue::RxQueue(IRxListener *listener) :
+    m_listener(listener)
+{
+    m_async = new uv_async_t;
+    m_async->data = this;
+
+    uv_async_init(uv_default_loop(), m_async, [](uv_async_t *handle) { static_cast<RxQueue *>(handle->data)->onReady(); });
+
+    m_thread = std::move(std::thread(&RxQueue::backgroundInit, this));
+}
+
+
+xmrig::RxQueue::~RxQueue()
+{
+    std::unique_lock<std::mutex> lock(m_mutex);
+    m_state = STATE_SHUTDOWN;
+    lock.unlock();
+
+    m_cv.notify_one();
+
+    m_thread.join();
+
+    delete m_storage;
+
+    Handle::close(m_async);
+}
+
+
+bool xmrig::RxQueue::isReady(const Job &job)
+{
+    std::lock_guard<std::mutex> lock(m_mutex);
+
+    return isReadyUnsafe(job);
+}
+
+
+xmrig::RxDataset *xmrig::RxQueue::dataset(const Job &job, uint32_t nodeId)
+{
+    std::lock_guard<std::mutex> lock(m_mutex);
+
+    if (isReadyUnsafe(job)) {
+        return m_storage->dataset(job, nodeId);
+    }
+
+    return nullptr;
+}
+
+
+std::pair<uint32_t, uint32_t> xmrig::RxQueue::hugePages()
+{
+    std::lock_guard<std::mutex> lock(m_mutex);
+
+    return m_storage && m_state == STATE_IDLE ? m_storage->hugePages() : std::pair<uint32_t, uint32_t>(0u, 0u);
+}
+
+
+void xmrig::RxQueue::enqueue(const RxSeed &seed, const std::vector<uint32_t> &nodeset, uint32_t threads, bool hugePages)
+{
+    std::unique_lock<std::mutex> lock(m_mutex);
+
+    if (!m_storage) {
+#       ifdef XMRIG_FEATURE_HWLOC
+        if (!nodeset.empty()) {
+            m_storage = new RxNUMAStorage(nodeset);
+        }
+        else
+#       endif
+        {
+            m_storage = new RxBasicStorage();
+        }
+    }
+
+    if (m_state == STATE_PENDING && m_seed == seed) {
+        return;
+    }
+
+    m_queue.emplace_back(seed, nodeset, threads, hugePages);
+    m_seed  = seed;
+    m_state = STATE_PENDING;
+
+    lock.unlock();
+
+    m_cv.notify_one();
+}
+
+
+bool xmrig::RxQueue::isReadyUnsafe(const Job &job) const
+{
+    return m_storage != nullptr && m_state == STATE_IDLE && m_seed == job;
+}
+
+
+void xmrig::RxQueue::backgroundInit()
+{
+    while (true) {
+        std::unique_lock<std::mutex> lock(m_mutex);
+
+        if (m_state == STATE_IDLE) {
+            m_cv.wait(lock, [this]{ return m_state != STATE_IDLE; });
+        }
+
+        if (m_state == STATE_SHUTDOWN) {
+            break;
+        }
+
+        const auto item = m_queue.back();
+        m_queue.clear();
+
+        lock.unlock();
+
+        LOG_INFO("%s" MAGENTA_BOLD("init dataset%s") " algo " WHITE_BOLD("%s (") CYAN_BOLD("%u") WHITE_BOLD(" threads)") BLACK_BOLD(" seed %s..."),
+                 rx_tag(),
+                 item.nodeset.size() > 1 ? "s" : "",
+                 item.seed.algorithm().shortName(),
+                 item.threads,
+                 Buffer::toHex(item.seed.data().data(), 8).data()
+                 );
+
+        m_storage->init(item.seed, item.threads, item.hugePages);
+
+        lock = std::move(std::unique_lock<std::mutex>(m_mutex));
+        if (!m_queue.empty()) {
+            continue;
+        }
+
+        m_state = STATE_IDLE;
+        uv_async_send(m_async);
+    }
+}
+
+
+void xmrig::RxQueue::onReady()
+{
+    std::unique_lock<std::mutex> lock(m_mutex);
+    const bool ready = m_listener && m_state == STATE_IDLE;
+    lock.unlock();
+
+    if (ready) {
+        m_listener->onDatasetReady();
+    }
+}

src/crypto/rx/RxQueue.h

@@ -0,0 +1,108 @@
+/* XMRig
+ * Copyright 2010      Jeff Garzik <jgarzik@pobox.com>
+ * Copyright 2012-2014 pooler      <pooler@litecoinpool.org>
+ * Copyright 2014      Lucas Jones <https://github.com/lucasjones>
+ * Copyright 2014-2016 Wolf9466    <https://github.com/OhGodAPet>
+ * Copyright 2016      Jay D Dee   <jayddee246@gmail.com>
+ * Copyright 2017-2019 XMR-Stak    <https://github.com/fireice-uk>, <https://github.com/psychocrypt>
+ * Copyright 2018      Lee Clagett <https://github.com/vtnerd>
+ * Copyright 2018-2019 tevador     <tevador@gmail.com>
+ * Copyright 2018-2019 SChernykh   <https://github.com/SChernykh>
+ * Copyright 2016-2019 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_RX_QUEUE_H
+#define XMRIG_RX_QUEUE_H
+
+
+#include "base/tools/Object.h"
+#include "crypto/rx/RxSeed.h"
+
+
+#include <condition_variable>
+#include <mutex>
+#include <thread>
+
+
+using uv_async_t = struct uv_async_s;
+
+
+namespace xmrig
+{
+
+
+class IRxListener;
+class IRxStorage;
+class RxDataset;
+
+
+class RxQueueItem
+{
+public:
+    RxQueueItem(const RxSeed &seed, const std::vector<uint32_t> &nodeset, uint32_t threads, bool hugePages) :
+        hugePages(hugePages),
+        seed(seed),
+        nodeset(nodeset),
+        threads(threads)
+    {}
+
+    const bool hugePages;
+    const RxSeed seed;
+    const std::vector<uint32_t> nodeset;
+    const uint32_t threads;
+};
+
+
+class RxQueue
+{
+public:
+    XMRIG_DISABLE_COPY_MOVE(RxQueue);
+
+    RxQueue(IRxListener *listener);
+    ~RxQueue();
+
+    bool isReady(const Job &job);
+    RxDataset *dataset(const Job &job, uint32_t nodeId);
+    std::pair<uint32_t, uint32_t> hugePages();
+    void enqueue(const RxSeed &seed, const std::vector<uint32_t> &nodeset, uint32_t threads, bool hugePages);
+
+private:
+    enum State {
+        STATE_IDLE,
+        STATE_PENDING,
+        STATE_SHUTDOWN
+    };
+
+    bool isReadyUnsafe(const Job &job) const;
+    void backgroundInit();
+    void onReady();
+
+    IRxListener *m_listener = nullptr;
+    IRxStorage *m_storage   = nullptr;
+    RxSeed m_seed;
+    State m_state = STATE_IDLE;
+    std::condition_variable m_cv;
+    std::mutex m_mutex;
+    std::thread m_thread;
+    std::vector<RxQueueItem> m_queue;
+    uv_async_t *m_async     = nullptr;
+};
+
+
+} /* namespace xmrig */
+
+
+#endif /* XMRIG_RX_QUEUE_H */