diff --git a/uppsrc/Core/CoWork.cpp b/uppsrc/Core/CoWork.cpp index 4b61d20f7..d9c976afd 100644 --- a/uppsrc/Core/CoWork.cpp +++ b/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 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&& 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&& 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 } - diff --git a/uppsrc/Core/CoWork.h b/uppsrc/Core/CoWork.h index 7e9d6eebe..b9107509e 100644 --- a/uppsrc/Core/CoWork.h +++ b/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 jobs; // global stack and CoWork stack as double-linked lists - int todo; + ConditionVariable waitforfinish; + Link 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 diff --git a/uppsrc/Core/src.tpp/CoWork$en-us.tpp b/uppsrc/Core/src.tpp/CoWork$en-us.tpp index 5d09b6099..e6b5b966f 100644 --- a/uppsrc/Core/src.tpp/CoWork$en-us.tpp +++ b/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]()&]