Thread management fixes for Farm.

exp/main.cpp

@@ -125,33 +125,30 @@ int main()
 		f.setWork(bi);
 		for (unsigned i = 0; !completed && i < timeout * 10; ++i, cout << f.miningProgress() << "\r" << flush)
 			this_thread::sleep_for(chrono::milliseconds(100));
+		cout << endl << flush;
 		cdebug << bi.mixHash << bi.nonce << (Ethash::verify(bi) ? "GOOD" : "bad");
 	};
 
+	Ethash::prep(genesis);
+
+	genesis.difficulty = u256(1) << 40;
+	genesis.noteDirty();
 	f.startCPU();
 	mine(f, genesis, 10);
-	mine(f, genesis, 10);
+
 	f.startGPU();
 
 	cdebug << "Good:";
 	genesis.difficulty = 1 << 18;
 	genesis.noteDirty();
-	mine(f, genesis, 3);
+	mine(f, genesis, 30);
 
 	cdebug << "Bad:";
 	genesis.difficulty = (u256(1) << 40);
 	genesis.noteDirty();
-	mine(f, genesis, 3);
+	mine(f, genesis, 30);
 
-	cdebug << "Good:";
+	f.stop();
-	genesis.difficulty = 1 << 18;
-	genesis.noteDirty();
-	mine(f, genesis, 3);
-
-	cdebug << "Bad:";
-	genesis.difficulty = (u256(1) << 40);
-	genesis.noteDirty();
-	mine(f, genesis, 3);
 
 	return 0;
 }

libdevcore/Worker.cpp

@@ -27,12 +27,20 @@
 using namespace std;
 using namespace dev;
 
-void Worker::startWorking()
+void Worker::startWorking(IfRunning _ir)
 {
 	cnote << "startWorking for thread" << m_name;
 	Guard l(x_work);
-	if (m_work)
+
-		return;
+	if (m_work && m_work->joinable())
+		try {
+			if (_ir == IfRunning::Detach)
+				m_work->detach();
+			else if (_ir == IfRunning::Join)
+				m_work->join();
+			else
+				return;
+		} catch (...) {}
 	cnote << "Spawning" << m_name;
 	m_stop = false;
 	m_work.reset(new thread([&]()
@@ -40,6 +48,7 @@ void Worker::startWorking()
 		setThreadName(m_name.c_str());
 		startedWorking();
 		workLoop();
+		m_work->detach();
 		cnote << "Finishing up worker thread";
 		doneWorking();
 	}));
@@ -49,11 +58,14 @@ void Worker::stopWorking()
 {
 	cnote << "stopWorking for thread" << m_name;
 	Guard l(x_work);
-	if (!m_work)
+	if (!m_work || !m_work->joinable())
 		return;
 	cnote << "Stopping" << m_name;
 	m_stop = true;
-	m_work->join();
+	try {
+		m_work->join();
+	}
+	catch (...) {}
 	m_work.reset();
 	cnote << "Stopped" << m_name;
 }

libdevcore/Worker.h

@@ -23,11 +23,19 @@
 
 #include <string>
 #include <thread>
+#include <atomic>
 #include "Guards.h"
 
 namespace dev
 {
 
+enum class IfRunning
+{
+	Fail,
+	Join,
+	Detach
+};
+
 class Worker
 {
 protected:
@@ -45,7 +53,7 @@ protected:
 	void setName(std::string _n) { if (!isWorking()) m_name = _n; }
 
 	/// Starts worker thread; causes startedWorking() to be called.
-	void startWorking();
+	void startWorking(IfRunning _ir = IfRunning::Fail);
 	
 	/// Stop worker thread; causes call to stopWorking().
 	void stopWorking();

libethcore/Ethash.cpp

@@ -30,9 +30,10 @@
 #include <thread>
 #include <libdevcore/Guards.h>
 #include <libdevcore/Log.h>
+#include <libdevcore/Common.h>
+#include <libdevcore/CommonIO.h>
 #include <libdevcrypto/CryptoPP.h>
 #include <libdevcrypto/FileSystem.h>
-#include <libdevcore/Common.h>
 #include <libethash/ethash.h>
 #if ETH_ETHASHCL || !ETH_TRUE
 #include <libethash-cl/ethash_cl_miner.h>
@@ -123,10 +124,12 @@ void Ethash::CPUMiner::workLoop()
 	uint64_t tryNonce = (uint64_t)(u64)Nonce::random(s_eng);
 	ethash_return_value ethashReturn;
 
-	auto p = EthashAux::params(m_work.seedHash);
+	WorkPackage w = work();
-	void const* dagPointer = EthashAux::full(m_work.seedHash).data();
+
-	uint8_t const* headerHashPointer = m_work.headerHash.data();
+	auto p = EthashAux::params(w.seedHash);
-	h256 boundary = m_work.boundary;
+	void const* dagPointer = EthashAux::full(w.seedHash).data();
+	uint8_t const* headerHashPointer = w.headerHash.data();
+	h256 boundary = w.boundary;
 	unsigned hashCount = 1;
 	for (; !shouldStop(); tryNonce++, hashCount++)
 	{
@@ -149,7 +152,6 @@ public:
 	void abort()
 	{
 		Guard l(x_all);
-		m_owner->m_work.headerHash = h256();
 		if (m_aborted)
 			return;
 //		cdebug << "Attempting to abort";
@@ -158,13 +160,17 @@ public:
 			std::this_thread::sleep_for(chrono::milliseconds(30));
 //		if (!m_aborted)
 //			cwarn << "Couldn't abort. Abandoning OpenCL process.";
+	}
+
+	void reset()
+	{
 		m_aborted = m_abort = false;
 	}
 
 protected:
 	virtual bool found(uint64_t const* _nonces, uint32_t _count) override
 	{
-//		cdebug << "Found nonces: " << vector<uint64_t>(_nonces, _nonces + _count);
+//		dev::operator <<(std::cerr << "Found nonces: ", vector<uint64_t>(_nonces, _nonces + _count)) << std::endl;
 		for (uint32_t i = 0; i < _count; ++i)
 		{
 			if (m_owner->report(_nonces[i]))
@@ -179,7 +185,7 @@ protected:
 	virtual bool searched(uint64_t _startNonce, uint32_t _count) override
 	{
 		Guard l(x_all);
-//		cdebug << "Searched" << _count << "from" << _startNonce;
+//		std::cerr << "Searched " << _count << " from " << _startNonce << std::endl;
 		m_owner->accumulateHashes(_count);
 		m_last = _startNonce + _count;
 		if (m_abort)
@@ -206,29 +212,30 @@ Ethash::GPUMiner::GPUMiner(ConstructionInfo const& _ci):
 
 Ethash::GPUMiner::~GPUMiner()
 {
-	delete m_hook;
+	pause();
 	delete m_miner;
+	delete m_hook;
 }
 
 bool Ethash::GPUMiner::report(uint64_t _nonce)
 {
 	Nonce n = (Nonce)(u64)_nonce;
-	Result r = EthashAux::eval(m_work.seedHash, m_work.headerHash, n);
+	Result r = EthashAux::eval(work().seedHash, work().headerHash, n);
-	if (r.value < m_work.boundary)
+	if (r.value < work().boundary)
 		return submitProof(Solution{n, r.mixHash});
 	return false;
 }
 
-void Ethash::GPUMiner::kickOff(WorkPackage const& _work)
+void Ethash::GPUMiner::kickOff()
 {
-	m_work = _work;
+	m_hook->reset();
 	startWorking();
 }
 
 void Ethash::GPUMiner::workLoop()
 {
 	// take local copy of work since it may end up being overwritten by kickOff/pause.
-	WorkPackage w = m_work;
+	WorkPackage w = work();
 	if (!m_miner || m_minerSeed != w.seedHash)
 	{
 		m_minerSeed = w.seedHash;
@@ -249,7 +256,6 @@ void Ethash::GPUMiner::pause()
 {
 	m_hook->abort();
 	stopWorking();
-	m_work.headerHash = h256();
 }
 
 #endif

libethcore/Ethash.h

@@ -87,19 +87,16 @@ public:
 		static unsigned instances() { return std::thread::hardware_concurrency(); }
 
 	protected:
-		void kickOff(WorkPackage const& _work) override
+		void kickOff() override
 		{
 			stopWorking();
-			m_work = _work;
 			startWorking();
 		}
 
-		void pause() override { stopWorking(); m_work.reset(); }
+		void pause() override { stopWorking(); }
 
 	private:
 		void workLoop() override;
-
-		WorkPackage m_work;
 	};
 
 #if ETH_ETHASHCL || !ETH_TRUE
@@ -114,7 +111,7 @@ public:
 		static unsigned instances() { return 1; }
 
 	protected:
-		void kickOff(WorkPackage const& _work) override;
+		void kickOff() override;
 		void pause() override;
 
 	private:
@@ -127,7 +124,6 @@ public:
 		ethash_cl_miner* m_miner = nullptr;
 
 		h256 m_minerSeed;		///< Last seed in m_miner
-		WorkPackage m_work;		///< Work to be done by GPU, set with kickOff and picked up in workLoop.
 	};
 #else
 	using GPUMiner = CPUMiner;

libethcore/Miner.h

@@ -40,13 +40,13 @@ namespace eth
 struct MiningProgress
 {
 //	MiningProgress& operator+=(MiningProgress const& _mp) { hashes += _mp.hashes; ms = std::max(ms, _mp.ms); return *this; }
-	unsigned hashes = 0;		///< Total number of hashes computed.
+	uint64_t hashes = 0;		///< Total number of hashes computed.
-	unsigned ms = 0;			///< Total number of milliseconds of mining thus far.
+	uint64_t ms = 0;			///< Total number of milliseconds of mining thus far.
 };
 
 struct MineInfo: public MiningProgress {};
 
-inline std::ostream& operator<<(std::ostream& _out, MiningProgress const& _p)
+inline std::ostream& operator<<(std::ostream& _out, MiningProgress _p)
 {
 	_out << (_p.hashes * 1000 / _p.ms) << "H/s = " <<  _p.hashes << " hashes / " << (double(_p.ms) / 1000) << "s";
 	return _out;
@@ -97,17 +97,19 @@ public:
 	void setWork(WorkPackage const& _work = WorkPackage())
 	{
 		Guard l(x_work);
-		if (_work.headerHash == m_work.headerHash)
+		auto old = m_work;
-			return;
-		if (_work.headerHash != h256())
-			kickOff(_work);
-		else if (m_work.headerHash == h256() && _work.headerHash != h256())
-			pause();
 		m_work = _work;
+		if (!!m_work)
+		{
+			pause();
+			kickOff();
+		}
+		else if (!m_work && !!old)
+			pause();
 		m_hashCount = 0;
 	}
 
-	unsigned hashCount() { return m_hashCount; }
+	uint64_t hashCount() { return m_hashCount; }
 
 	unsigned index() const { return m_index; }
 
@@ -119,7 +121,7 @@ protected:
 	 * @brief Begin working on a given work package, discarding any previous work.
 	 * @param _work The package for which to find a solution.
 	 */
-	virtual void kickOff(WorkPackage const& _work) = 0;
+	virtual void kickOff() = 0;
 
 	/**
 	 * @brief No work left to be done. Pause until told to kickOff().
@@ -138,7 +140,10 @@ protected:
 		if (m_farm)
 		{
 			Guard l(x_work);
-			return m_farm->submitProof(_s, m_work, this);
+			if (!m_farm->submitProof(_s, m_work, this))
+				return false;
+			m_work.reset();
+			return true;
 		}
 		return true;
 	}
@@ -154,7 +159,7 @@ private:
 	Mutex x_work;
 	WorkPackage m_work;
 
-	unsigned m_hashCount = 0;
+	uint64_t m_hashCount = 0;
 };
 
 }

libethereum/Farm.h

@@ -50,6 +50,11 @@ public:
 	using Solution = typename PoW::Solution;
 	using Miner = GenericMiner<PoW>;
 
+	~GenericFarm()
+	{
+		stop();
+	}
+
 	/**
 	 * @brief Sets the current mining mission.
 	 * @param _bi The block (header) we wish to be mining.
@@ -142,7 +147,7 @@ private:
 			for (std::shared_ptr<Miner> const& m: m_miners)
 				if (m.get() != _m)
 					m->setWork();
-			m_work.headerHash = h256();
+			m_work.reset();
 			return true;
 		}
 		return false;