artemist/lix
0
0
Fork 0
forked from lix-project/lix
Code Pull requests Activity

ThreadPool: Improve exception handling

Browse source 
In particular, process() won't return as long as there are active
items. This prevents work item lambdas from referring to stack frames
that no longer exist.
This commit is contained in:
Eelco Dolstra 2017-09-08 14:40:27 +02:00
parent a2740c9ca2
commit 8f6b347abd
No known key found for this signature in database
GPG key ID: 8170B4726D7198DE
2 changed files with 53 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

71
src/libutil/thread-pool.cc
View file

@ -46,11 +46,17 @@ void ThreadPool::enqueue(const work_t & t)
void ThreadPool::process()
{
/* Loop until there are no active work items *and* there either
are no queued items or there is an exception. The
post-condition is that no new items will become active. */
while (true) {
auto state(state_.lock());
if (!state->active) {
if (state->exception)
std::rethrow_exception(state->exception);
if (state->left.empty() && !state->pending) break;
if (state->left.empty())
break;
}
state.wait(done);
}
}
@ -58,41 +64,64 @@ void ThreadPool::process()
void ThreadPool::workerEntry()
{
bool didWork = false;
std::exception_ptr exc;
while (true) {
work_t w;
{
auto state(state_.lock());
while (true) {
if (state->quit || state->exception) return;
if (didWork) {
assert(state->pending);
state->pending--;
didWork = false;
assert(state->active);
state->active--;
if (exc) {
if (!state->exception) {
state->exception = exc;
// Tell the other workers to quit.
state->quit = true;
work.notify_all();
} else {
/* Print the exception, since we can't
propagate it. */
try {
std::rethrow_exception(exc);
} catch (std::exception & e) {
if (!dynamic_cast<Interrupted*>(&e) &&
!dynamic_cast<ThreadPoolShutDown*>(&e))
ignoreException();
} catch (...) {
}
}
}
}
/* Wait until a work item is available or another thread
had an exception or we're asked to quit. */
while (true) {
if (state->quit) {
if (!state->active)
done.notify_one();
return;
}
if (!state->left.empty()) break;
if (!state->pending)
done.notify_all();
if (!state->active) {
done.notify_one();
return;
}
state.wait(work);
}
w = state->left.front();
w = std::move(state->left.front());
state->left.pop();
state->pending++;
state->active++;
}
try {
w();
} catch (std::exception & e) {
auto state(state_.lock());
if (state->exception) {
if (!dynamic_cast<Interrupted*>(&e) &&
!dynamic_cast<ThreadPoolShutDown*>(&e))
printError(format("error: %s") % e.what());
} else {
state->exception = std::current_exception();
work.notify_all();
done.notify_all();
}
} catch (...) {
exc = std::current_exception();
}
didWork = true;

2
src/libutil/thread-pool.hh
View file

@ -44,7 +44,7 @@ private:
struct State
{
std::queue<work_t> left;
size_t pending = 0;
size_t active = 0;
std::exception_ptr exception;
std::vector<std::thread> workers;
bool quit = false;