mirror of
https://github.com/NixOS/nix.git
synced 2024-11-21 22:32:26 +00:00
ThreadPool: Improve exception handling
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:
parent
a2740c9ca2
commit
8f6b347abd
@ -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->exception)
|
||||
std::rethrow_exception(state->exception);
|
||||
if (state->left.empty() && !state->pending) break;
|
||||
if (!state->active) {
|
||||
if (state->exception)
|
||||
std::rethrow_exception(state->exception);
|
||||
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());
|
||||
|
||||
if (didWork) {
|
||||
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 || state->exception) return;
|
||||
if (didWork) {
|
||||
assert(state->pending);
|
||||
state->pending--;
|
||||
didWork = false;
|
||||
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;
|
||||
|
@ -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;
|
||||
|
Loading…
Reference in New Issue
Block a user