Commit d25f65bd authored by knowledge4igor's avatar knowledge4igor

Fix data races in Async: whenAll and whenAny for Args&& ...

parent 90595480
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#include <condition_variable> #include <condition_variable>
#include <stdexcept> #include <stdexcept>
#include <typeinfo> #include <typeinfo>
#include <iostream>
namespace Pistache { namespace Pistache {
...@@ -180,7 +181,7 @@ namespace Async { ...@@ -180,7 +181,7 @@ namespace Async {
{ } { }
bool allocated; bool allocated;
State state; std::atomic<State> state;
std::exception_ptr exc; std::exception_ptr exc;
/* /*
...@@ -1173,13 +1174,15 @@ namespace Async { ...@@ -1173,13 +1174,15 @@ namespace Async {
: total(_total) : total(_total)
, resolved(0) , resolved(0)
, rejected(false) , rejected(false)
, mtx()
, resolve(std::move(_resolver)) , resolve(std::move(_resolver))
, reject(std::move(_rejection)) , reject(std::move(_rejection))
{ } { }
const size_t total; const size_t total;
std::atomic<size_t> resolved; size_t resolved;
std::atomic<bool> rejected; bool rejected;
std::mutex mtx;
Resolver resolve; Resolver resolve;
Rejection reject; Rejection reject;
...@@ -1187,11 +1190,13 @@ namespace Async { ...@@ -1187,11 +1190,13 @@ namespace Async {
template<size_t Index, typename T, typename Data> template<size_t Index, typename T, typename Data>
static void resolveT(const T& val, Data& data) { static void resolveT(const T& val, Data& data) {
std::lock_guard<std::mutex> guard(data->mtx);
if (data->rejected) return; if (data->rejected) return;
// @Check thread-safety of std::get ? // @Check thread-safety of std::get ?
std::get<Index>(data->results) = val; std::get<Index>(data->results) = val;
data->resolved.fetch_add(1, std::memory_order_relaxed); data->resolved++;
if (data->resolved == data->total) { if (data->resolved == data->total) {
data->resolve(data->results); data->resolve(data->results);
...@@ -1200,9 +1205,11 @@ namespace Async { ...@@ -1200,9 +1205,11 @@ namespace Async {
template<typename Data> template<typename Data>
static void resolveVoid(Data& data) { static void resolveVoid(Data& data) {
std::lock_guard<std::mutex> guard(data->mtx);
if (data->rejected) return; if (data->rejected) return;
data->resolved.fetch_add(1, std::memory_order_relaxed); data->resolved++;
if (data->resolved == data->total) { if (data->resolved == data->total) {
data->resolve(data->results); data->resolve(data->results);
...@@ -1211,7 +1218,9 @@ namespace Async { ...@@ -1211,7 +1218,9 @@ namespace Async {
template<typename Data> template<typename Data>
static void reject(std::exception_ptr exc, Data& data) { static void reject(std::exception_ptr exc, Data& data) {
data->rejected.store(true); std::lock_guard<std::mutex> guard(data->mtx);
data->rejected = true;
data->reject(exc); data->reject(exc);
} }
}; };
...@@ -1221,11 +1230,13 @@ namespace Async { ...@@ -1221,11 +1230,13 @@ namespace Async {
struct Data { struct Data {
Data(size_t, Resolver resolver, Rejection rejection) Data(size_t, Resolver resolver, Rejection rejection)
: done(false) : done(false)
, mtx()
, resolve(std::move(resolver)) , resolve(std::move(resolver))
, reject(std::move(rejection)) , reject(std::move(rejection))
{ } { }
std::atomic<bool> done; bool done;
std::mutex mtx;
Resolver resolve; Resolver resolve;
Rejection reject; Rejection reject;
...@@ -1233,30 +1244,36 @@ namespace Async { ...@@ -1233,30 +1244,36 @@ namespace Async {
template<size_t Index, typename T, typename Data> template<size_t Index, typename T, typename Data>
static void resolveT(const T& val, Data& data) { static void resolveT(const T& val, Data& data) {
std::lock_guard<std::mutex> guard(data->mtx);
if (data->done) return; if (data->done) return;
// Instead of allocating a new core, ideally we could share the same core as // Instead of allocating a new core, ideally we could share the same core as
// the relevant promise but we do not have access to the promise here is so meh // the relevant promise but we do not have access to the promise here is so meh
auto core = std::make_shared<Private::CoreT<T>>(); auto core = std::make_shared<Private::CoreT<T>>();
core->template construct<T>(val); core->template construct<T>(val);
data->resolve(Async::Any(std::move(core))); data->resolve(Async::Any(core));
data->done = true; data->done = true;
} }
template<typename Data> template<typename Data>
static void resolveVoid(Data& data) { static void resolveVoid(Data& data) {
std::lock_guard<std::mutex> guard(data->mtx);
if (data->done) return; if (data->done) return;
auto core = std::make_shared<Private::CoreT<void>>(); auto core = std::make_shared<Private::CoreT<void>>();
data->resolve(Async::Any(std::move(core))); data->resolve(Async::Any(core));
data->done = true; data->done = true;
} }
template<typename Data> template<typename Data>
static void reject(std::exception_ptr exc, Data& data) { static void reject(std::exception_ptr exc, Data& data) {
data->done.store(true); std::lock_guard<std::mutex> guard(data->mtx);
data->done = true;
data->reject(exc); data->reject(exc);
} }
}; };
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment