Job package for Ultimate++
---------------------------

This template class implements a scope bound, single worker thread based on RAII principle. 
It provides a return semantics for result gathering, functionally similar to promise/future 
pattern (including void type specialization). Also it provides a convenient error management 
and exception propagation mechanisms for worker threads, and it is compatible with U++ 
single-threaded mode.

Note that while Job is a general purpose multithreading tool, for high performance loop 
parallelization scenarios CoWork would be a more suitable option. This class is mainly designed 
to allow applications and libraries to gain an easily managable, optional non-blocking behavior 
where high latency is expected such as network operations and file I/O, and a safe, 
container-style access to the data processed by the worker threads is preferred.

Features and Highlights
-----------------------

- A safe way to gather results from worker threads.
- Simple and easy-to-use thread halting, and error reporting mechanism.
- Exception propagation.
- External blocking is possible.
- Optional constant reference access to job results.
- Compatible with U++ single-threaded environment.
- All Job instances are scope bound and will forced to finish job when they get out of scope.

Known Issues
-----------------------

- Currently none.

History
-----------------------

- 2017-10-07: Compatibility with U++ single-threaded mode is added.

- 2017-10-01: Global variables moved into JobGlobal namespace in order to avoid multiple
              definitions error. Accordingly, global functions are defined in Job.cpp.

- 2017-09-22: Exception propagation mechanism for job is properly added. From now on
              worker threads will pass exceptions to their caller.
              Void template specialization is re-implemented (without using future/promise).
              Constant reference access operator is added. This is especially useful
              where the data is a container with iterators (such as Vector or Array).

- 2017-09-19: std::exception class exceptions are handled, and treaated as errors. 
              (For the time being.)
              void instantiation is now possible.
              Jobs will notify their workers on shutdown. 
              Clean up & cosmetics...

- 2017-09-18: Clear() method is added. Worker id generator is using int64. Documentation 
              updated.

- 2017-09-17: Future/promise mechanism, and std template library code completely removed.
              From now on Job has its own result gathering mechanism with zero copy/move
              overhead.

- 2017-09-16: Job is redesigned. It is now a proper worker thread.

- 2017-09-10: Initial public beta version is released.

	{
		// Print a message to stdout.
		// Returns void.
		CoWork job;
		job & [=] { Cout() << "Hello world!\n"; };
		job.Finish();
	}
	
	{
		// Print all the divisors of random numbers.
		// Returns a String.
		CoWork jobs;
		Vector<String> results;
		for(int i = 0; i < 5; i++)
			jobs & [=, &results] { String h = GetDivisors2(); CoWork::FinLock(); results.At(i) = GetDivisors(); };
		jobs.Finish();
		for(auto r : results)
			Cout() << r << '\n';
	}

#include <CtrlLib/CtrlLib.h>

#include <future>

using namespace Upp;

template< class Function, class... Args>
std::future<std::result_of_t<std::decay_t<Function>(std::decay_t<Args>...)>>
Async(Function&& f, Args&&... args )
{
	std::promise<std::result_of_t<std::decay_t<Function>(std::decay_t<Args>...)>> p;
	auto ftr = p.get_future();
	CoWork::Schedule([=, p = pick(p)]() mutable {
		p.set_value(f(args...));
	});
	return ftr;
}

GUI_APP_MAIN
{
	DDUMP(Async([] { return "Hello world"; }).get());
}

	{
		Job<String> filejob([=]{
			FileIn fi("/home/test_session/Very_Large_Test_File.txt");
			if(fi.IsError())
				JobError(fi.GetError(), fi.GetErrorText());
			Cout() << "Reading file...";
			return LoadStream(fi);
		});
		while(!filejob.IsFinished()) Cout() << "....";
		Cout() << (filejob.IsError() ? filejob.GetErrorDesc() : filejob.GetResult()) << '\n';
	}

TIMING Job            : 137.00 ms - 137.00 ms (137.00 ms / 1 ), min: 137.00 ms, max: 137.00 ms, nesting: 1 - 1
TIMING CoWork         : 206.00 ms - 206.00 ms (206.00 ms / 1 ), min: 206.00 ms, max: 206.00 ms, nesting: 1 - 1

String GetDivisors()
{
	String s;
	int number = (int) 1000;
	Vector<int> divisors;
	for(auto i = 1, j = 0; i < number + 1; i++) {
		auto d = number % i;
		if(d == 0){
			divisors.Add(i);
			j++;
		}
		if(i == number)
			s = Format("Worker Id: %d, Number: %d, Divisors (count: %d): %s",
						GetWorkerId(),
						number,
						j,
						divisors.ToString());
	}
	return pick(s);
}

CONSOLE_APP_MAIN
{
	{
		CoWork jobs;
		jobs.SetPoolSize(1500);
		Vector<String> results;
		TIMING("CoWork");
		for(int i = 0; i < 1500; i++)
			jobs & [=, &results] { String h = GetDivisors(); CoWork::FinLock(); results.At(i) = h; };
		jobs.Finish();
		for(auto r : results)
			Cout() << r << '\n';
	

	}
	
	{
		Array<Job<String>> jobs;
		jobs.SetCount(1500);
		TIMING("Job");
		for(int i = 0; i < 1500; i++)
		 jobs[i].Start([=]{ return GetDivisors(); });
		WaitForJobs();
		for(auto& job : jobs) 
				Cout() << job.GetResult()    << '\n';
		
	}

}

{
		CoWork jobs;
		Vector<String> results;
		TIMING("CoWork");
		jobs & [=, &results] {
			Vector<String> s;
			for(int i = 0; i < 50000; i++)
				 s.Add() = GetDivisors();
				CoWork::FinLock();
				results = pick(s);
		};
		jobs.Finish();
	}
	
	{
		Job<Vector<String>> job;
		TIMING("Job");
		job.Start([=]{
			Vector<String> s;
			for(int i = 0; i < 50000; i++)
				s.Add() = GetDivisors();
			return pick(s);
		});
		job.Finish();
		auto s = job.GetResult();
	}

TIMING Job            :  1.42 s  -  1.42 s  ( 1.42 s  / 1 ), min:  1.42 s , max:  1.42 s , nesting: 1 - 1
TIMING CoWork         :  1.39 s  -  1.39 s  ( 1.39 s  / 1 ), min:  1.39 s , max:  1.39 s , nesting: 1 - 1

This package contains a lightweight and easy-to-use multithreading tool for U++ framework: Job.
Job template class implements a scope bound, single worker thread based on RAII principle.
It provides a return semantics for result gathering functionally similar to promise/future
pattern but with three major differences:

1)  future/promise pair requires at least moving of the resulted data, which can be
    relatively expensive depending on the object type. On the other hand, Job acts as a simple
    container and uses a reference based result gathering method.  This makes it possible to
    reduce move/copy overhead involved (nearly down to zero).

2)  Job does not allow the T to be of plain void type (of course, void pointer is allowed).

3)  Trying to access the resulting data while it is still invalid will not throw.
    Resources are allocated during construction (including the job data).

Note that for higher performance loop parallelization scenarios, CoWork would be a more
suitable option. This class is mainly designed to allow the applications and libraries to gain 
an easily managable, optional non-blocking behaviour where high latency is expected (Such as
network operations and file I/O), and a safe "referential access" to the objects processed
by the worker threads is preferred.

class Client : public Job<String> {
public:
	Client&	Blocking(bool b = true)	{ blocking = b; return *this; }
	String	Request(const String& host, int port);
	
private:
	String	Run(Event<>&& cmd);
	bool	blocking = true;
};

String Client::Run(Event<>&& cmd)
{
	Start(pick(cmd));
	if(blocking) Finish();
	return blocking && !IsError() ? GetResult() : GetErrorDesc();
}

String Client::Request(const String& host, int port)
{
	auto cmd = [=]{
		TcpSocket	socket;
		auto& output = Job<String>::Data(); // This method allows referential access to the data of respective job.
		output = Format("Client #%d: ", GetWorkerId());
		
		INTERLOCKED {	Cout() << output << "Starting...\n"; }
		
		if(socket.Timeout(10000).Connect(host, port))
			output.Cat(socket.GetLine());
		if(socket.IsError())
			throw JobError(socket.GetError(), socket.GetErrorDesc());
	};
	return Run(cmd);
}

CONSOLE_APP_MAIN
{
        //.....

        // Requesting in a simple, blocking way.
	{
		Cout() << "----- Processing individual blocking requests...\n";
		Cout() << c1.Request(host1, 21) << '\n';
		Cout() << c2.Request(host2, 21)	<< '\n';
	}
	
	// Reuse workers and make requests in a simple, non-blocking way.
	{
		Cout() << "----- Processing individual non-blocking requests...\n";
		// We can "clear" the data (String):
		c1.GetResult().Clear();
		c2.GetResult().Clear();

		c1.Blocking(false).Request(host1, 21);
		c2.Blocking(false).Request(host2, 21);

		while(!c1.IsFinished() || !c2.IsFinished())
			;
		if(c1.IsError()) Cerr() << c1.GetErrorDesc() << '\n';
		else Cout() << ~c1 << '\n';
		
		if(c2.IsError()) Cerr() << c2.GetErrorDesc() << '\n';
		else Cout() << ~c2 << '\n';

	}
        
        //....

}

    double sum_total = 0;
    Job<Vector<double>> job;
    job.Start(DoSomeHeavyCalculationAndFillTheVector);

    //....

    if(job.IsFinishe() && !job.IsError()) {
       for(auto& e : ~job)          // Reference access to Vector. Data is not copied or moved. Vector is filled via a direct -yet safe- access.
             sum_total += e;        // And retrieved in same fashion as well. Note also that there is no requirement for the data type to be contained.
                                    // In fact, using One or Any, or Value, we can even differentiate between the job 
                                    // results for given operations in a single Job easily. :)
    }

	Array<Job<String>> jobs;
	jobs.SetCount(CPU_Cores() + 2);

	CoWork cowork;
//	cowork.SetPoolSize(CPU_Cores() + 2);
	
	Vector<String> results;
	DUMP(CPU_Cores());
	{

		TIMING("CoWork -- With stdout output");
		for(int i = 0; i < 10000; i++)
			cowork & [=, &results] { String h = GetDivisors(); CoWork::FinLock(); results.At(i) = h; };
		cowork.Finish();
		// Stdout output section.
		for(auto& r : results)
			Cout() << r << '\n';

	}
	{
		TIMING("Job -- With stdout output");

		int i = 0;
		while(i < 10000) {
			for(auto& job : jobs) {
				if(!job.IsFinished()) {
					continue;
				}
				job & [=]{ Job<String>::Data() = GetDivisors(); };
				if(!(~job).IsEmpty()) {
					Cout() << ~job << '\n';
					if(++i == 10000) break;
				}
			}
	
		}
	}

CPU_Cores() = 6
TIMING Job -- With stdout output: 370.00 ms - 370.00 ms (370.00 ms / 1 ), min: 370.00 ms, max: 370.00 ms, nesting: 1 - 1
TIMING CoWork -- With stdout output: 461.00 ms - 461.00 ms (461.00 ms / 1 ), min: 461.00 ms, max: 461.00 ms, nesting: 1 - 1

CPU_Cores() = 6
TIMING Job -- Without stdout output: 228.00 ms - 228.00 ms (228.00 ms / 1 ), min: 228.00 ms, max: 228.00 ms, nesting: 1 - 1
TIMING CoWork -- Without stdout output: 234.00 ms - 234.00 ms (234.00 ms / 1 ), min: 234.00 ms, max: 234.00 ms, nesting: 1 - 1

CPU_Cores() = 6
TIMING Job -- With stdout output: 34.00 ms - 34.00 ms (34.00 ms / 1 ), min: 34.00 ms, max: 34.00 ms, nesting: 1 - 1
TIMING CoWork -- With stdout output: 53.00 ms - 53.00 ms (53.00 ms / 1 ), min: 53.00 ms, max: 53.00 ms, nesting: 1 - 1

CPU_Cores() = 6
TIMING Job -- Without stdout output: 24.00 ms - 24.00 ms (24.00 ms / 1 ), min: 24.00 ms, max: 24.00 ms, nesting: 1 - 1
TIMING CoWork -- Without stdout output: 31.00 ms - 31.00 ms (31.00 ms / 1 ), min: 31.00 ms, max: 31.00 ms, nesting: 1 - 1

	// Reuse workers and make requests in a simple, non-blocking way.
	{
		Cout() << "----- Processing individual non-blocking requests...\n";
		// We can "clear" the data. Useful especially if the data is "picked".
		c1.Clear();
		c2.Clear();

		c1.Blocking(false).Request(host1, 21);
		c2.Blocking(false).Request(host2, 21);

		while(!c1.IsFinished() || !c2.IsFinished())
			;
		if(c1.IsError()) Cerr() << c1.GetErrorDesc() << '\n';
		else Cout() << ~c1 << '\n';
		
		if(c2.IsError()) Cerr() << c2.GetErrorDesc() << '\n';
		else Cout() << ~c2 << '\n';

	}

while(!c1.IsFinished() || !c2.IsFinished()) {
if(c2.IsFinished())
    Cout() << second request is complete before the first\n";
   // Etc...

}

template<class T>
One<T>*& sData()
{
	static thread_local One<T>* sData = NULL;
	return sData;
}

#include <Core/Core.h>
#include <Job/Job.h>

using namespace Upp;

String GetDivisors()
{
	String s;
	int number = (int) 1000;
	Vector<int> divisors;
	for(auto i = 1, j = 0; i < number + 1; i++) {
		auto d = number % i;
		if(d == 0){
			divisors.Add(i);
			j++;
		}
		if(i == number)
			s = Format("Worker Id: %d, Number: %d, Divisors (count: %d): %s",
						GetWorkerId(),
						number,
						j,
						divisors.ToString());
	}
	return pick(s);
}

// #define OUTPUT
#define N 100000

CONSOLE_APP_MAIN
{
	Array<Job<String>> jobs;
	jobs.SetCount(CPU_Cores() + 2);

	CoWork cowork;
//	cowork.SetPoolSize(CPU_Cores() + 2);
	
	Vector<String> results;
	RDUMP(CPU_Cores());
	{

		RTIMING("CoWork -- With stdout output");
		for(int i = 0; i < N; i++)
			cowork & [=, &results] { String h = GetDivisors(); CoWork::FinLock(); results.At(i) = h; };
		cowork.Finish();
		// Stdout output section.
	#ifdef OUTPUT
		for(auto& r : results)
			Cout() << r << '\n';
	#endif
	}
	{
		RTIMING("Job -- With stdout output");

		int i = 0;
		while(i < N) {
			for(auto& job : jobs) {
				if(!job.IsFinished()) {
					continue;
				}
				job & [=]{ Job<String>::Data() = GetDivisors(); };
				if(!(~job).IsEmpty()) {
				#ifdef OUTPUT
					Cout() << ~job << '\n';
				#endif
					if(++i == N) break;
				}
			}
	
		}
	}
}

* d:\upp.out\MyApps\MSC15\JobBenchmark.exe 19.09.2017 10:01:27, user: cxl

CPU_Cores() = 8
TIMING Job -- With stdout output: 123.00 ms - 123.00 ms (123.00 ms / 1 ), min: 123.00 ms, max: 123.00 ms, nesting: 1 - 1
TIMING CoWork -- With stdout output: 103.00 ms - 103.00 ms (103.00 ms / 1 ), min: 103.00 ms, max: 103.00 ms, nesting: 1 - 1

#include <Core/Core.h>
#include <future>

using namespace Upp;

template <class D, class P, class F, class... Args>
void set_future_value(D *dummy, P&& p, F&& f, Args&&... args)
{
	p.set_value(f(args...));
}

template <class P, class F, class... Args>
void set_future_value(void *dummy, P&& p, F&& f, Args&&... args)
{
	f(args...);
	p.set_value();
}

template< class Function, class... Args>
std::future<std::result_of_t<std::decay_t<Function>(std::decay_t<Args>...)>>
Async(Function&& f, Args&&... args )
{
	std::promise<std::result_of_t<std::decay_t<Function>(std::decay_t<Args>...)>> p;
	auto ftr = p.get_future();
	CoWork::Schedule([=, p = pick(p)]() mutable {
		std::result_of_t<std::decay_t<Function>(std::decay_t<Args>...)> *dummy = NULL;
		set_future_value(dummy, pick(p), pick(f));
	});
	return ftr;
}

CONSOLE_APP_MAIN
{
// with reference
	String h;
	auto f = Async([&h] { h = "Hello world"; });
	f.get();
	DDUMP(h);

// using return value
	DDUMP(Async([] { return "Hello world"; }).get());
}

    template<class V = T>
    Job(V v) : Job()							{ *data = v; }

	int i;
	
	Job<int*> job(&i);
	job.Start([]{ auto& n = *Job<int*>::Data() = 100;}); 
	job.Finish();
	Cout() << "Number is " << i << '\n';

      auto result1 = Async([=]{ return GetDivisors(); })
      while(!result.IsReady())
             ; // Do something;
      Cout() << result.IsValid() ? result.Get() : result.GetErrorDesc();

      // and of course...
      Cout() << Async([=]{ return "Hello World\n"; }).Get();
             

Result::Get();

Result::IsReady();

Result::IsValid();

Result::GetError();

Result::GetErrorDesc():

Result::GetId();

Result::Cancel();

- Void type instantiation is re-introduced for good (does not rely on anything other than plain template specializaton rules.)

- Value return semantics is re-introduced for good.

- Constant reference access operator is added. (This is especially useful when using Job with containers such as Vector, or array (e.g. Job<Vector<int>>) in a loop. See JobExample test app provided)

- Exception propagation mechanism is added. Workers will propagate raised exceptions to their caller. (when one of the GeResult(), operator~(), or IsError() methods is called. )

- JobExample test application is added to the package. This example demonstrates the above mentioned traits.
  

void CancelJob()
{
	// Singlethreaded version. (non-blocking operation)
	// Below example is one of the simplest ways to achive non-blocking operations with Job in a
	// single-threaded environment. A finer-grained operation would involve handling of return
	// codes. (e.g. using Job<int>)
	
	Job<void> job;
	auto work = [=] {
		if(IsJobCancelled()) {
			Cout() << "Worker #" << GetWorkerId() << " received cancel signal. Cancelling job...\n";
			throw JobError("Operation cancelled.");
		}
	};
	
	Cout() << "Worker #" << GetWorkerId() << " started. (Waiting the cancellation signal...)\n";
	const int timeout = 5000;
	auto start_time = msecs();
	while(!job.IsError()) {
		job.Start(work);
		if(msecs(start_time) >= timeout)
			job.Cancel();	// Or if you have more than one non-blocking operation in progress,
		                  	// you can call CancelJobs() global function to cancel all running jobs.
	}
}

- WorkEntry -> std:promise   (This class is not explicitly called. It is allocated on the heap (using new), and its ownership is passed to WorkResult.)
- WorkResult -> std: future  (With pick semantics, void type specialization, error handling, exception propagation, and optional constant reference access to the result.)
- Work                       (Contains a CoWork instance, and exposes Do() and Finish() methods.)

//	Error handling, picking/moving.
	auto r1 = work.Do([=] { throw Work::Error("Worker failed."); });
	work.Finish();
	auto r2 = pick(r1); // r1 is picked.
	if(r2.IsError()) {
		Cout() << r2.GetErrorDesc() << '\n';
	}

-
- WorkResult (former WorkEntry): Unlike std::promise, this is called privately. I don't really see any use in calling it explicitly
- Work (former WorkResult):     This can be the U++ std::future counterpart (see the first prototype packageI provided above) This can be a helper class to CoWork, representing a single, isolated (semantically) thread.

- CoWork::Work():               This is the thread-starter method which will return, well, Work :)

#include <Core/Core.h>

using namespace Upp;

template <class Ret>
class AsyncWork {
	template <class Ret>
	struct Imp {
		CoWork co;
		Ret    ret;
	
		template< class Function, class... Args>
		void       Do(Function&& f, Args&&... args) { co.Do([=]() { ret = f(args...); }); }
		const Ret& Get()                            { return ret; }
	};

	template <>
	struct Imp<void> {
		CoWork co;
	
		template< class Function, class... Args>
		void       Do(Function&& f, Args&&... args) { co.Do([=]() { f(args...); }); }
		void       Get()                            {}
	};
	
	One<Imp<Ret>> imp;
	
public:
	template< class Function, class... Args>
	void  Do(Function&& f, Args&&... args)    { imp.Create().Do(f, args...); }

	void  Cancel()                            { if(imp) imp->co.Cancel(); }
	Ret   Get()                               { ASSERT(imp); imp->co.Finish(); return imp->Get(); }
	Ret   operator~()                         { return Get(); }
	
	AsyncWork(AsyncWork&&) = default;
	AsyncWork()                               {}
	~AsyncWork()                              { if(imp) imp->co.Cancel(); }
};

template< class Function, class... Args>
AsyncWork<std::result_of_t<std::decay_t<Function>(std::decay_t<Args>...)>>
Async(Function&& f, Args&&... args)
{
	AsyncWork<std::result_of_t<std::decay_t<Function>(std::decay_t<Args>...)>> h;
	h.Do(f, args...);
	return pick(h);
}

CONSOLE_APP_MAIN
{
	auto h = Async([] { return "Hello world"; });
	DDUMP(h.Get());
	
	DDUMP(~Async([](int x) { return x * x; }, 9));
	
	auto x = Async([] { throw "error"; });
	try {
		x.Get();
	}
	catch(...) {
		DLOG("Exception caught");
	}
}

bool IsFinished() { ASSERT(imp); return imp->co.IsFinished(); }

template <class Range>
ValueTypeOf<Range> ASum(const Range& rng, const ValueTypeOf<Range>& zero)
{
	int n = rng.GetCount();
	if(n == 1)
		return rng[0];
	if(n == 0)
		return 0;
	auto l = Async([&] { return ASum(SubRange(rng, 0, n / 2)); });
	auto r = Async([&] { return ASum(SubRange(rng, n / 2, n - n / 2)); });
	while(!l.IsFinished() || !r.IsFinished())
		Sleep(1);
	return l.Get() + r.Get();
}

\AsyncTest.cpp (8): error: declaration of template parameter 'Ret' shadows template parameter
\AsyncTest.cpp (18): error: explicit specialization in non-namespace scope 'class AsyncWork<Ret>'
\AsyncTest.cpp (19): error: template parameters not deducible in partial specialization:
\AsyncTest.cpp (31): error: too many template-parameter-lists
\AsyncTest.cpp (47): error: 'class AsyncWork<const char*>' has no member named 'Do' (): h.Do
(f, args...);
\AsyncTest.cpp: In instantiation of 'AsyncWork<typename std::result_of<typename std::decay<_Tp>::type(std::decay_t<Args>...)>::type> Async(Function&&, Args&& ...) [wit
h Function = ConsoleMainFn_()::<lambda(int)>; Args = {int}; typename std::result_of<typename std::decay<_Tp>::type(std::decay_t<Args>...)>::type = int]':
\AsyncTest.cpp (56): required from here
\AsyncTest.cpp (47): error: 'class AsyncWork<int>' has no member named 'Do'
\AsyncTest.cpp (47): error: 'class AsyncWork<void>' has no member named 'Do'
\AsyncTest.cpp (11): error: 'AsyncWork<Ret>::Imp<Ret>::ret' has incomplete type
\AsyncTest.cpp (11): error: invalid use of 'void'
\AsyncTest.cpp (15): error: forming reference to void
\AsyncTest.cpp (34): error: 'struct AsyncWork<void>::Imp<void>' has no member named 'Get'; did you mean 'ret'?
\AsyncTest.cpp (34): error: return-statement with a value, in function returning 'void' [-fpermissive]

CONSOLE_APP_MAIN
{
	auto h = Async([] { return "Hello world"; });
	DDUMP(h.Get());
	
	DDUMP(~Async([](int x) { return x * x; }, 9));
	
	auto x = Async([] { throw "error"; });
	try {
		x.Get();
	}
	catch(...) {
		DLOG("Exception caught");
	}
}

#include <Core/Core.h>

using namespace Upp;

Atomic h;

void Wait(int c)
{
	while(c--)
		h++;
}

CONSOLE_APP_MAIN
{
	StdLogSetup(LOG_COUT|LOG_FILE);

	for(int i = 0; i < 100000; i++) {
		if(i % 1000 == 0)
			LOG(i);
		bool b = false;
		auto x = Async([] { Wait(Random(3000)); throw "error"; });
		try {
			Wait(Random(3000));
			x.Get();
		}
		catch(...) {
			b = true;
		}
		ASSERT(b);
	}
	
	LOG("============ OK");
}