Skip to content

Instantly share code, notes, and snippets.

@oliverepper

oliverepper/senders_and_receivers_error.cpp

Created November 21, 2022 20:02
Show Gist options
  • Save oliverepper/3fb53f7d551189c3a72a259aa0160f29 to your computer and use it in GitHub Desktop.
Save oliverepper/3fb53f7d551189c3a72a259aa0160f29 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
senders_and_receivers_error.cpp
#include <thread>
auto async_work() {
return [](auto p) {
std::thread { [p=std::move(p)]() mutable {
std::exception_ptr ep;
try {
throw std::runtime_error("Test Error");
} catch (...) {
ep = std::current_exception();
p.set_error(ep);
}
return 42;
}}.detach();
};
}
template <typename P, typename N>
struct then_sender {
P p;
N next;
void set_value(auto ...args) {
p.set_value(next(args...));
}
void set_error(auto e) {
p.set_error(e);
}
};
auto then(auto task, auto next) {
return [=](auto p) {
task(then_sender<decltype (p), decltype (next)>{p, next});
};
}
template <typename T>
struct state {
std::mutex mtx;
std::condition_variable cv;
std::variant<std::monostate, std::exception_ptr, T> data;
};
template <typename T>
struct promise {
state<T> *ps;
// set
template <int I>
void _set(auto ...args) {
std::unique_lock lck{ps->mtx};
ps->data.template emplace<I>(args...);
ps->cv.notify_one();
}
void set_value(auto ...args) {
_set<2>(args...);
}
void set_error(auto e) {
_set<1>(e);
}
};
template <typename T, typename Task>
T sync_wait(Task task) {
state<T> s;
task(promise<T>{&s});
// get
{
std::unique_lock lck{s.mtx};
s.cv.wait(lck, [&s]() { return s.data.index() != 0; });
}
if (s.data.index() == 1) std::rethrow_exception(std::get<1>(s.data));
return std::move(std::get<2>(s.data));
}
int main() {
auto start = async_work();
auto next = then(start, [](int i) {
return i / 2;
});
try {
printf("%d\n", sync_wait<int>(next));
} catch (const std::exception& e) {
printf("%s\n", e.what());
}
return 0;
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment