Job flow WIP.

CMakeLists.txt

@@ -84,7 +84,7 @@ endif()
 
 add_definitions(/D_GNU_SOURCE)
 add_definitions(/DUNICODE)
-add_definitions(/DAPP_DEBUG)
+#add_definitions(/DAPP_DEBUG)
 
 set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} "${CMAKE_SOURCE_DIR}/cmake")
 

src/App.cpp

@@ -74,7 +74,7 @@ App::exec()
     Mem::allocate(m_options->algo(), m_options->threads(), m_options->doubleHash());
     Summary::print();
 
-    Workers::start(m_options->threads());
+    Workers::start(m_options->threads(), m_options->affinity(), m_options->nicehash());
 
     m_network->connect();
 

src/Console.cpp

@@ -97,12 +97,12 @@ void Console::message(Console::Level level, const char* fmt, ...)
             m_colors ? kCL_N : ""
         );
 
-    uv_mutex_lock(&m_mutex);
+    pthread_mutex_lock(&m_mutex);
 
     vfprintf(stdout, buf, ap);
     fflush(stdout);
 
-    uv_mutex_unlock(&m_mutex);
+    pthread_mutex_unlock(&m_mutex);
 
     va_end(ap);
 }
@@ -121,12 +121,12 @@ void Console::text(const char* fmt, ...)
             m_colors ? kCL_N : ""
         );
 
-    uv_mutex_lock(&m_mutex);
+    pthread_mutex_lock(&m_mutex);
 
     vfprintf(stdout, buf, ap);
     fflush(stdout);
 
-    uv_mutex_unlock(&m_mutex);
+    pthread_mutex_unlock(&m_mutex);
 
     va_end(ap);
 }
@@ -135,5 +135,5 @@ void Console::text(const char* fmt, ...)
 Console::Console() :
     m_colors(true)
 {
-    uv_mutex_init(&m_mutex);
+    pthread_mutex_init(&m_mutex, nullptr);
 }

src/Console.h

@@ -25,7 +25,7 @@
 #define __CONSOLE_H__
 
 
-#include <uv.h>
+#include <pthread.h>
 
 
 class Console
@@ -61,7 +61,7 @@ private:
 
     static Console *m_self;
     bool m_colors;
-    uv_mutex_t m_mutex;
+    pthread_mutex_t m_mutex;
 };
 
 

src/net/Job.h

@@ -40,6 +40,7 @@ public:
     inline const char *id() const      { return m_id; }
     inline const uint8_t *blob() const { return m_blob; }
     inline int poolId() const          { return m_poolId; }
+    inline uint32_t *nonce()           { return reinterpret_cast<uint32_t*>(m_blob + 39); }
     inline uint32_t diff() const       { return m_diff; }
     inline uint32_t size() const       { return m_size; }
     inline uint64_t target() const     { return m_target; }

src/net/Network.cpp

@@ -30,6 +30,7 @@
 #include "net/Network.h"
 #include "net/Url.h"
 #include "Options.h"
+#include "workers/Workers.h"
 
 
 Network::Network(const Options *options) :
@@ -83,6 +84,7 @@ void Network::onClose(Client *client, int failures)
 
     if (m_pool == id) {
         m_pool = 0;
+        Workers::pause();
     }
 
     if (id == 1 && m_pools.size() > 2 && failures == m_options->retries()) {
@@ -152,6 +154,8 @@ void Network::setJob(Client *client, const Job &job)
     else {
         LOG_INFO("new job from \"%s:%d\", diff: %d", client->host(), client->port(), job.diff());
     }
+
+    Workers::setJob(job);
 }
 
 

src/workers/Handle.cpp

@@ -25,8 +25,10 @@
 #include "workers/Handle.h"
 
 
-Handle::Handle(int id) :
-    m_id(id),
+Handle::Handle(int threadId, int64_t affinity, bool nicehash) :
+    m_nicehash(nicehash),
+    m_threadId(threadId),
+    m_affinity(affinity),
     m_worker(nullptr)
 {
 }

src/workers/Handle.h

@@ -26,6 +26,7 @@
 
 
 #include <pthread.h>
+#include <stdint.h>
 
 
 class IWorker;
@@ -34,14 +35,18 @@ class IWorker;
 class Handle
 {
 public:
-    Handle(int id);
+    Handle(int threadId, int64_t affinity, bool nicehash);
     void start(void *(*callback) (void *));
 
-    inline int id() const                  { return m_id; }
+    inline bool nicehash() const           { return m_nicehash; }
+    inline int threadId() const            { return m_threadId; }
+    inline int64_t affinity() const        { return m_affinity; }
     inline void setWorker(IWorker *worker) { m_worker = worker; }
 
 private:
-    int m_id;
+    bool m_nicehash;
+    int m_threadId;
+    int64_t m_affinity;
     IWorker *m_worker;
     pthread_t m_thread;
 };

src/workers/SingleWorker.cpp

@@ -22,6 +22,7 @@
  */
 
 
+#include <atomic>
 #include <thread>
 #include <pthread.h>
 
@@ -40,5 +41,35 @@ SingleWorker::SingleWorker(Handle *handle)
 
 void SingleWorker::start()
 {
-//    Workers::submit();
+    while (true) {
+        if (Workers::isPaused()) {
+            do {
+                LOG_ERR("SLEEP WAIT FOR WORK");
+                std::this_thread::sleep_for(std::chrono::milliseconds(200));
+            }
+            while (Workers::isPaused());
+
+            consumeJob();
+        }
+
+        while (!Workers::isOutdated(m_sequence)) {
+            LOG_ERR("WORK %lld %lld", Workers::sequence(), m_sequence);
+
+            std::this_thread::sleep_for(std::chrono::milliseconds(2000));
+
+            sched_yield();
+        }
+
+        consumeJob();
+    }
+}
+
+
+
+void SingleWorker::consumeJob()
+{
+    m_job = Workers::job();
+    m_sequence = Workers::sequence();
+
+    LOG_WARN("consumeJob");
 }

src/workers/SingleWorker.h

@@ -25,6 +25,7 @@
 #define __SINGLEWORKER_H__
 
 
+#include "net/Job.h"
 #include "workers/Worker.h"
 
 
@@ -37,6 +38,11 @@ public:
     SingleWorker(Handle *handle);
 
     void start() override;
+
+private:
+    void consumeJob();
+
+    Job m_job;
 };
 
 

src/workers/Worker.cpp

@@ -22,17 +22,23 @@
  */
 
 
+#include "Cpu.h"
+#include "Mem.h"
 #include "workers/Handle.h"
 #include "workers/Worker.h"
-#include "Mem.h"
 
 
 Worker::Worker(Handle *handle) :
+    m_nicehash(handle->nicehash()),
     m_handle(handle),
-    m_id(handle->id())
+    m_id(handle->threadId())
 {
     m_handle->setWorker(this);
 
+    if (Cpu::threads() > 1 && m_handle->affinity() != -1L) {
+        Cpu::setAffinity(m_id, m_handle->affinity());
+    }
+
     m_ctx = Mem::create(m_id);
 }
 

src/workers/Worker.h

@@ -25,6 +25,9 @@
 #define __WORKER_H__
 
 
+#include <stdint.h>
+
+
 #include "interfaces/IWorker.h"
 
 
@@ -39,9 +42,11 @@ public:
     ~Worker();
 
 protected:
+    bool m_nicehash;
     cryptonight_ctx *m_ctx;
     Handle *m_handle;
     int m_id;
+    uint64_t m_sequence;
 };
 
 

src/workers/Workers.cpp

@@ -21,27 +21,52 @@
  *   along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-
-#include <pthread.h>
-
-
 #include "Console.h"
 #include "workers/Handle.h"
 #include "workers/SingleWorker.h"
 #include "workers/Workers.h"
 
+
+Job Workers::m_job;
+pthread_rwlock_t Workers::m_rwlock;
+std::atomic<int> Workers::m_paused;
+std::atomic<uint64_t> Workers::m_sequence;
 std::vector<Handle*> Workers::m_workers;
 uv_async_t Workers::m_async;
 
 
-void Workers::start(int threads)
+Job Workers::job()
+{
+    pthread_rwlock_rdlock(&m_rwlock);
+    Job job = m_job;
+    pthread_rwlock_unlock(&m_rwlock);
+
+    return std::move(job);
+}
+
+
+void Workers::setJob(const Job &job)
+{
+    pthread_rwlock_wrlock(&m_rwlock);
+    m_job = job;
+    pthread_rwlock_unlock(&m_rwlock);
+
+    m_sequence++;
+    m_paused = 0;
+}
+
+
+void Workers::start(int threads, int64_t affinity, bool nicehash)
 {
     LOG_NOTICE("start %d", pthread_self());
 
+    m_sequence = 0;
+    m_paused   = 1;
+
     uv_async_init(uv_default_loop(), &m_async, Workers::onResult);
 
     for (int i = 0; i < threads; ++i) {
-        Handle *handle = new Handle(i);
+        Handle *handle = new Handle(i, affinity, nicehash);
         m_workers.push_back(handle);
         handle->start(Workers::onReady);
     }

src/workers/Workers.h

@@ -25,8 +25,13 @@
 #define __WORKERS_H__
 
 
-#include <vector>
+#include <atomic>
+#include <pthread.h>
 #include <uv.h>
+#include <vector>
+
+
+#include "net/Job.h"
 
 
 class Handle;
@@ -35,13 +40,24 @@ class Handle;
 class Workers
 {
 public:
-    static void start(int threads);
+    static Job job();
+    static void setJob(const Job &job);
+    static void start(int threads, int64_t affinity, bool nicehash);
     static void submit();
 
+    static inline bool isOutdated(uint64_t sequence) { return m_sequence.load(std::memory_order_relaxed) != sequence; }
+    static inline bool isPaused()                    { return m_paused.load(std::memory_order_relaxed) == 1; }
+    static inline uint64_t sequence()                { return m_sequence.load(std::memory_order_relaxed); }
+    static inline void pause()                       { m_paused = 1; }
+
 private:
     static void *onReady(void *arg);
     static void onResult(uv_async_t *handle);
 
+    static Job m_job;
+    static pthread_rwlock_t m_rwlock;
+    static std::atomic<int> m_paused;
+    static std::atomic<uint64_t> m_sequence;
     static std::vector<Handle*> m_workers;
     static uv_async_t m_async;
 };