CoWork: Exception handling, IsCanceled

git-svn-id: svn://ultimatepp.org/upp/trunk@11344 f0d560ea-af0d-0410-9eb7-867de7ffcac7

uppsrc/Core/CoWork.cpp

@@ -4,13 +4,14 @@ namespace Upp {
 
 #ifdef _MULTITHREADED
 
-#define LLOG(x)      // DLOG(x)
-#define LDUMP(x)     // DDUMP(x)
+#define LLOG(x)       // DLOG(x)
+#define LDUMP(x)      // DDUMP(x)
 
-#define LHITCOUNT(x) // RHITCOUNT(x)
+#define LHITCOUNT(x)  // RHITCOUNT(x)
 
-thread_local bool CoWork::Pool::finlock;
-thread_local int  CoWork::worker_index = -1;
+thread_local bool    CoWork::Pool::finlock;
+thread_local int     CoWork::worker_index = -1;
+thread_local CoWork *CoWork::current;
 
 CoWork::Pool& CoWork::GetPool()
 {
@@ -37,9 +38,8 @@ void CoWork::Pool::ExitThreads()
 	quit = true;
 	lock.Leave();
 	waitforjob.Broadcast();
-	for(int i = 0; i < threads.GetCount(); i++) {
+	for(int i = 0; i < threads.GetCount(); i++)
 		threads[i].Wait();
-	}
 	threads.Clear();
 	lock.Enter();
 	quit = false;
@@ -88,13 +88,25 @@ void CoWork::Pool::DoJob(MJob& job)
 	finlock = false;
 	Function<void ()> fn = pick(job.fn);
 	CoWork *work = job.work;
+	CoWork::current = work;
 	Free(job);
 	lock.Leave();
-	fn();
+	std::exception_ptr exc;
+	try {
+		fn();
+	}
+	catch(...) {
+		LLOG("DoJob caught exception");
+		exc = std::current_exception();
+	}
 	if(!finlock)
 		lock.Enter();
 	if(!work)
 		return;
+	if(exc && !work->exc) {
+		work->Cancel0();
+		work->exc = exc;
+	}
 	if(--work->todo == 0) {
 		LLOG("Releasing waitforfinish of (CoWork " << FormatIntHex(work) << ")");
 		work->waitforfinish.Signal();
@@ -182,35 +194,51 @@ void CoWork::Do(Function<void ()>&& fn)
 	p.lock.Leave();
 }
 
-void CoWork::Cancel()
+void CoWork::Cancel0()
 {
+	LLOG("CoWork Cancel0");
+	canceled = true;
 	Pool& p = GetPool();
-	p.lock.Enter();
 	while(!jobs.IsEmpty(1)) {
+		LHITCOUNT("CoWork::Canceling scheduled Job");
 		MJob& job = *jobs.GetNext(1);
 		job.UnlinkAll();
 		p.Free(job);
 		--todo;
 	}
+}
+
+void CoWork::Finish0()
+{
+	Pool& p = GetPool();
 	while(todo) {
-		LLOG("Cancel (CoWork " << FormatIntHex(this) << ")");
+		LLOG("WaitForFinish (CoWork " << FormatIntHex(this) << ")");
 		waitforfinish.Wait(p.lock);
 	}
+	canceled = false;
+	if(exc) {
+		LLOG("CoWork rethrowing worker exception");
+		auto e = exc;
+		exc = nullptr;
+		p.lock.Leave();
+		std::rethrow_exception(e);
+	}
+}
+
+void CoWork::Cancel()
+{
+	Pool& p = GetPool();
+	p.lock.Enter();
+	Cancel0();
+	Finish0();
 	p.lock.Leave();
-	LLOG("CoWork " << FormatIntHex(this) << " finished");
+	LLOG("CoWork " << FormatIntHex(this) << " canceled and finished");
 }
 
 void CoWork::Finish() {
 	Pool& p = GetPool();
 	p.lock.Enter();
-	while(!jobs.IsEmpty(1)) {
-		LLOG("Finish: todo: " << todo << " (CoWork " << FormatIntHex(this) << ")");
-		p.DoJob(*jobs.GetNext(1));
-	}
-	while(todo) {
-		LLOG("WaitForFinish (CoWork " << FormatIntHex(this) << ")");
-		waitforfinish.Wait(p.lock);
-	}
+	Finish0();
 	p.lock.Leave();
 	LLOG("CoWork " << FormatIntHex(this) << " finished");
 }
@@ -261,13 +289,24 @@ void CoWork::Pipe(int stepi, Function<void ()>&& fn)
 	}
 }
 
+void CoWork::Reset()
+{
+	try {
+		Cancel();
+	}
+	catch(...) {}
+	todo = 0;
+	canceled = false;
+}
+
 CoWork::CoWork()
 {
 	LLOG("CoWork constructed " << FormatHex(this));
 	todo = 0;
+	canceled = false;
 }
 
-CoWork::~CoWork()
+CoWork::~CoWork() noexcept(false)
 {
 	Finish();
 	LLOG("~CoWork " << FormatIntHex(this));
@@ -276,4 +315,3 @@ CoWork::~CoWork()
 #endif
 
 }
-

uppsrc/Core/CoWork.h

@@ -30,7 +30,7 @@ public:
 		Pool();
 		~Pool();
 		
-		static thread__ bool finlock;
+		static thread_local bool    finlock;
 
 		bool DoJob();
 		static void ThreadRun(int tno);
@@ -41,10 +41,16 @@ public:
 	static Pool& GetPool();
 
 	static thread_local int worker_index;
+	static thread_local CoWork *current;
 
-	ConditionVariable waitforfinish;
-	Link<MJob, 2> jobs; // global stack and CoWork stack as double-linked lists
-	int todo;
+	ConditionVariable  waitforfinish;
+	Link<MJob, 2>      jobs; // global stack and CoWork stack as double-linked lists
+	int                todo;
+	bool               canceled;
+	std::exception_ptr exc;
+
+	void Cancel0();
+	void Finish0();
 
 // experimental pipe support
 	Mutex stepmutex;
@@ -68,10 +74,13 @@ public:
 	static void FinLock();
 	
 	void Cancel();
+	static bool IsCanceled()                                  { return current && current->canceled; }
 
 	void Finish();
 	
 	bool IsFinished();
+	
+	void Reset();
 
 	static bool IsWorker()                                    { return worker_index >= 0; }
 	static int  GetWorkerIndex()                              { return worker_index; }
@@ -79,7 +88,7 @@ public:
 	static void SetPoolSize(int n);
 
 	CoWork();
-	~CoWork();
+	~CoWork() noexcept(false);
 };
 
 template <class T>

uppsrc/Core/src.tpp/CoWork$en-us.tpp

@@ -27,6 +27,10 @@ also possible. Thread pool is normally terminated when the main
 thread finishes.&]
 [s9;%% No synchronization is required to access CoWork instances 
 from various threads (CoWork is internally synchronized).&]
+[s9;%% If an exception is thrown in worker thread, which is not handled 
+by worker thread, it is caught and rethrown in CoWork thread 
+in Finish routine. Any such exception also causes the Cancel 
+of the CoWork.&]
 [s9;%% [*/ Implementation notes: ]Current implementation has single 
 global FIFO stack for 2048 scheduled jobs. When there is no slot 
 available when scheduling the job, it is performed immediately 
@@ -89,14 +93,23 @@ not to cause congestion of CoWork scheduling.&]
 [s5;:Upp`:`:CoWork`:`:Cancel`(`): [@(0.0.255) void]_[* Cancel]()&]
 [s2;%% Removed all jobs scheduled by this thread that has not started 
 yet from the queue and then waits for any jobs already started 
-to finish.&]
+to finish. Rethrows the exception thrown in worker threads. If 
+more than single worker thread throws the exception, the first 
+exception thrown in absolute time is rethrown.&]
+[s3; &]
+[s4; &]
+[s5;:Upp`:`:CoWork`:`:IsCanceled`(`): [@(0.0.255) static] [@(0.0.255) bool]_[* IsCanceled](
+)&]
+[s2;%% This methods returns true in worker thread when the worker 
+thread is a part of some CoWork instance and that instance was 
+canceled.&]
 [s3; &]
 [s4; &]
 [s5;:CoWork`:`:Finish`(`): [@(0.0.255) void]_[* Finish]()&]
 [s2;%% Waits until all jobs scheduled using Do (or operator`&) are 
 finished. All changes to data performed by scheduled threads 
 are visible after Finish. While waiting, Finish can perform scheduled 
-jobs.&]
+jobs. Can &]
 [s3; &]
 [s4; &]
 [s5;:Upp`:`:CoWork`:`:IsFinished`(`): [@(0.0.255) bool]_[* IsFinished]()&]
@@ -106,8 +119,17 @@ are visible after IsFinished returns true (so this is basically
 non`-blocking variant of Finish).&]
 [s3;%% &]
 [s4; &]
+[s5;:Upp`:`:CoWork`:`:Reset`(`): [@(0.0.255) void]_[* Reset]()&]
+[s2;%% Calls Cancel, catches and ignores all exceptions eventually 
+thrown by worker threads. Then resets CoWork to the initial state 
+as if it was just constructed. Useful when using CoWork as nonlocal 
+variable.&]
+[s3; &]
+[s4; &]
 [s5;:CoWork`:`:`~CoWork`(`): [@(0.0.255) `~][* CoWork]()&]
-[s2;%% Calls Finish().&]
+[s2;%% Calls Finish(). Can eventually rethrow worker thread exception. 
+If there is a chance of destructor being involved in stack unwinding, 
+Finish should be called separately before destructor.&]
 [s3; &]
 [s4; &]
 [s5;:Upp`:`:CoWork`:`:IsWorker`(`): [@(0.0.255) static] [@(0.0.255) bool]_[* IsWorker]()&]