Commit 1f41b225 authored by Kirk Shoop's avatar Kirk Shoop Committed by Facebook Github Bot

fix synchronization issue and update benchmarks

fbshipit-source-id: 517a2a9c83e19c539a4ccdd6fe696a8ce2d9a2d1
parent aac47db4
...@@ -22,80 +22,78 @@ struct countdownsingle { ...@@ -22,80 +22,78 @@ struct countdownsingle {
template <class ExecutorRef> template <class ExecutorRef>
void operator()(ExecutorRef exec) { void operator()(ExecutorRef exec) {
if (--*counter > 0) { if (--*counter > 0) {
exec | op::submit(*this); exec | op::submit(mi::single{*this});
} }
} }
}; };
struct inline_executor {
using properties = mi::property_set<mi::is_sender<>, mi::is_single<>>;
template<class Out>
void submit(Out out) {
::mi::set_value(out, *this);
}
};
#define concept Concept #define concept Concept
#include <nonius/nonius.h++> #include <nonius/nonius.h++>
NONIUS_BENCHMARK("trampoline virtual derecursion 10,000", [](nonius::chronometer meter){ NONIUS_BENCHMARK("inline 10", [](nonius::chronometer meter){
int counter = 0; int counter = 0;
auto tr = mi::trampoline(); auto ie = inline_executor{};
using TR = decltype(tr); using IE = decltype(ie);
std::function<void(mi::any_time_executor_ref<> exec)> recurse; countdownsingle single{counter};
recurse = [&](mi::any_time_executor_ref<> tr) {
if (--counter <= 0)
return;
tr | op::submit(recurse);
};
meter.measure([&]{ meter.measure([&]{
counter = 10'000; counter = 10;
return tr | op::submit([&](auto exec) { recurse(exec); }); ie | op::submit(mi::single{single});
return counter;
}); });
}) })
NONIUS_BENCHMARK("trampoline static derecursion 10,000", [](nonius::chronometer meter){ NONIUS_BENCHMARK("trampoline static derecursion 10", [](nonius::chronometer meter){
int counter = 0; int counter = 0;
auto tr = mi::trampoline(); auto tr = mi::trampoline();
using TR = decltype(tr); using TR = decltype(tr);
countdownsingle single{counter}; countdownsingle single{counter};
meter.measure([&]{ meter.measure([&]{
counter = 10'000; counter = 10;
return tr | op::submit(single); tr | op::submit(single);
return counter;
}); });
}) })
NONIUS_BENCHMARK("new thread 10 blocking_submits", [](nonius::chronometer meter){ NONIUS_BENCHMARK("trampoline virtual derecursion 10", [](nonius::chronometer meter){
auto nt = mi::new_thread(); int counter = 0;
using NT = decltype(nt); auto tr = mi::trampoline();
using TR = decltype(tr);
std::function<void(mi::any_time_executor_ref<> exec)> recurse{countdownsingle{counter}};
meter.measure([&]{ meter.measure([&]{
return nt | counter = 10;
op::blocking_submit() | tr | op::submit([&](auto exec) { recurse(exec); });
op::blocking_submit() | return counter;
op::blocking_submit() |
op::blocking_submit() |
op::blocking_submit() |
op::blocking_submit() |
op::blocking_submit() |
op::blocking_submit() |
op::blocking_submit() |
op::transform([](auto nt){
return v::now(nt);
}) |
op::get<std::chrono::system_clock::time_point>;
}); });
}) })
NONIUS_BENCHMARK("pool 10 blocking_submits", [](nonius::chronometer meter){ NONIUS_BENCHMARK("pool 1 blocking_submit 10", [](nonius::chronometer meter){
mi::pool pl{std::max(1u,std::thread::hardware_concurrency())}; mi::pool pl{std::max(1u,std::thread::hardware_concurrency())};
auto pe = pl.executor(); auto pe = pl.executor();
using PE = decltype(pe); using PE = decltype(pe);
int counter = 0;
countdownsingle single{counter};
meter.measure([&]{
counter = 10;
return pe | op::blocking_submit(single);
});
})
NONIUS_BENCHMARK("new thread blocking_submit 10", [](nonius::chronometer meter){
auto nt = mi::new_thread();
using NT = decltype(nt);
int counter = 0;
countdownsingle single{counter};
meter.measure([&]{ meter.measure([&]{
return pe | counter = 10;
op::blocking_submit() | return nt | op::blocking_submit(single);
op::blocking_submit() |
op::blocking_submit() |
op::blocking_submit() |
op::blocking_submit() |
op::blocking_submit() |
op::blocking_submit() |
op::blocking_submit() |
op::blocking_submit() |
op::transform([](auto pe){
return mi::now(pe);
}) |
op::get<std::chrono::system_clock::time_point>;
}); });
}) })
This diff is collapsed.
...@@ -5393,6 +5393,7 @@ private: ...@@ -5393,6 +5393,7 @@ private:
template <bool IsTimeSender, class In> template <bool IsTimeSender, class In>
In impl_(In in) { In impl_(In in) {
bool done = false; bool done = false;
std::mutex lock;
std::condition_variable signaled; std::condition_variable signaled;
auto out{::pushmi::detail::out_from_fn<In>()( auto out{::pushmi::detail::out_from_fn<In>()(
std::move(args_), std::move(args_),
...@@ -5408,6 +5409,7 @@ private: ...@@ -5408,6 +5409,7 @@ private:
) else ( ) else (
::pushmi::set_value(out, id((V&&) v)); ::pushmi::set_value(out, id((V&&) v));
)) ))
std::unique_lock<std::mutex> guard{lock};
done = true; done = true;
signaled.notify_all(); signaled.notify_all();
} }
...@@ -5415,6 +5417,7 @@ private: ...@@ -5415,6 +5417,7 @@ private:
on_error(constrain(pushmi::lazy::NoneReceiver<_1, _2>, on_error(constrain(pushmi::lazy::NoneReceiver<_1, _2>,
[&](auto out, auto e) noexcept { [&](auto out, auto e) noexcept {
::pushmi::set_error(out, std::move(e)); ::pushmi::set_error(out, std::move(e));
std::unique_lock<std::mutex> guard{lock};
done = true; done = true;
signaled.notify_all(); signaled.notify_all();
} }
...@@ -5422,6 +5425,7 @@ private: ...@@ -5422,6 +5425,7 @@ private:
on_done(constrain(pushmi::lazy::Receiver<_1>, on_done(constrain(pushmi::lazy::Receiver<_1>,
[&](auto out){ [&](auto out){
::pushmi::set_done(out); ::pushmi::set_done(out);
std::unique_lock<std::mutex> guard{lock};
done = true; done = true;
signaled.notify_all(); signaled.notify_all();
} }
...@@ -5432,7 +5436,6 @@ private: ...@@ -5432,7 +5436,6 @@ private:
) else ( ) else (
id(::pushmi::submit)(in, std::move(out)); id(::pushmi::submit)(in, std::move(out));
)) ))
std::mutex lock;
std::unique_lock<std::mutex> guard{lock}; std::unique_lock<std::mutex> guard{lock};
signaled.wait(guard, [&]{ signaled.wait(guard, [&]{
return done; return done;
......
...@@ -125,6 +125,7 @@ private: ...@@ -125,6 +125,7 @@ private:
template <bool IsTimeSender, class In> template <bool IsTimeSender, class In>
In impl_(In in) { In impl_(In in) {
bool done = false; bool done = false;
std::mutex lock;
std::condition_variable signaled; std::condition_variable signaled;
auto out{::pushmi::detail::out_from_fn<In>()( auto out{::pushmi::detail::out_from_fn<In>()(
std::move(args_), std::move(args_),
...@@ -140,6 +141,7 @@ private: ...@@ -140,6 +141,7 @@ private:
) else ( ) else (
::pushmi::set_value(out, id((V&&) v)); ::pushmi::set_value(out, id((V&&) v));
)) ))
std::unique_lock<std::mutex> guard{lock};
done = true; done = true;
signaled.notify_all(); signaled.notify_all();
} }
...@@ -147,6 +149,7 @@ private: ...@@ -147,6 +149,7 @@ private:
on_error(constrain(pushmi::lazy::NoneReceiver<_1, _2>, on_error(constrain(pushmi::lazy::NoneReceiver<_1, _2>,
[&](auto out, auto e) noexcept { [&](auto out, auto e) noexcept {
::pushmi::set_error(out, std::move(e)); ::pushmi::set_error(out, std::move(e));
std::unique_lock<std::mutex> guard{lock};
done = true; done = true;
signaled.notify_all(); signaled.notify_all();
} }
...@@ -154,6 +157,7 @@ private: ...@@ -154,6 +157,7 @@ private:
on_done(constrain(pushmi::lazy::Receiver<_1>, on_done(constrain(pushmi::lazy::Receiver<_1>,
[&](auto out){ [&](auto out){
::pushmi::set_done(out); ::pushmi::set_done(out);
std::unique_lock<std::mutex> guard{lock};
done = true; done = true;
signaled.notify_all(); signaled.notify_all();
} }
...@@ -164,7 +168,6 @@ private: ...@@ -164,7 +168,6 @@ private:
) else ( ) else (
id(::pushmi::submit)(in, std::move(out)); id(::pushmi::submit)(in, std::move(out));
)) ))
std::mutex lock;
std::unique_lock<std::mutex> guard{lock}; std::unique_lock<std::mutex> guard{lock};
signaled.wait(guard, [&]{ signaled.wait(guard, [&]{
return done; return done;
......
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