Commit 323f42e1 authored by Eric Niebler's avatar Eric Niebler Committed by Facebook Github Bot

bring bit-rotted pushmi examples back from the dead

Summary: Several of the pushmi examples needed a working thread pool. Use the new one in pushmi/examples/pool.h, and bring the examples up-to-date with recent pushmi changes.

Reviewed By: yfeldblum

Differential Revision: D14563479

fbshipit-source-id: 70def2adc90fdf3eb5eef929245f39b294209cbe
parent 594ff7a5
...@@ -37,15 +37,21 @@ PUSHMI_INLINE_VAR constexpr struct bulk_fn { ...@@ -37,15 +37,21 @@ PUSHMI_INLINE_VAR constexpr struct bulk_fn {
Target&& driver, Target&& driver,
IF&& initFunc, IF&& initFunc,
RS&& selector) const { RS&& selector) const {
return [func, sb, se, driver, initFunc, selector](auto in) { return [func, sb, se, driver, initFunc, selector](auto in) mutable {
return make_single_sender( return make_single_sender(
[in, func, sb, se, driver, initFunc, selector](auto out) mutable { [in, func, sb, se, driver, initFunc, selector](auto out) mutable {
using Out = decltype(out);
struct data : Out {
data(Out out) : Out(std::move(out)) {}
bool empty = true;
};
submit( submit(
in, in,
make_receiver( make_receiver(
std::move(out), data{std::move(out)},
[func, sb, se, driver, initFunc, selector]( [func, sb, se, driver, initFunc, selector](
auto& out_, auto input) { auto& out_, auto input) mutable noexcept {
out_.empty = false;
driver( driver(
initFunc, initFunc,
selector, selector,
...@@ -53,8 +59,12 @@ PUSHMI_INLINE_VAR constexpr struct bulk_fn { ...@@ -53,8 +59,12 @@ PUSHMI_INLINE_VAR constexpr struct bulk_fn {
func, func,
sb, sb,
se, se,
std::move(out_)); std::move(static_cast<Out&>(out_)));
})); },
// forward to output
[](auto o, auto e) noexcept {set_error(o, e);},
// only pass done through when empty
[](auto o){ if (o.empty) { set_done(o); }}));
}); });
}; };
} }
......
...@@ -13,6 +13,7 @@ ...@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
#include <algorithm> #include <algorithm>
#include <cassert> #include <cassert>
#include <iostream> #include <iostream>
...@@ -41,32 +42,32 @@ void lisp(CPUExecutor cpu, IOExecutor io) { ...@@ -41,32 +42,32 @@ void lisp(CPUExecutor cpu, IOExecutor io) {
// f on cpu - g on cpu (implicit: a single task on the cpu executor runs all // f on cpu - g on cpu (implicit: a single task on the cpu executor runs all
// the functions) // the functions)
op::submit([](g_t) {})(op::transform([](f_t ft) { return g(ft); })( op::submit([](g_t) {})(op::transform([](f_t ft) { return g(ft); })(
op::transform([](auto) { return f(); })(cpu))); op::transform([](auto) { return f(); })(cpu.schedule())));
// f on cpu - g on cpu (explicit: the first cpu task runs f and a second cpu // f on cpu - g on cpu (explicit: the first cpu task runs f and a second cpu
// task runs g) // task runs g)
op::submit([](g_t) {})(op::transform([](f_t ft) { return g(ft); })( op::submit([](g_t) {})(op::transform([](f_t ft) { return g(ft); })(
op::via(mi::strands(cpu))(op::transform([](auto) { return f(); })(cpu)))); op::via(mi::strands(cpu))(op::transform([](auto) { return f(); })(cpu.schedule()))));
// f on io - g on cpu // f on io - g on cpu
op::submit([](g_t) {})(op::transform([](f_t ft) { return g(ft); })( op::submit([](g_t) {})(op::transform([](f_t ft) { return g(ft); })(
op::via(mi::strands(cpu))(op::transform([](auto) { return f(); })(io)))); op::via(mi::strands(cpu))(op::transform([](auto) { return f(); })(io.schedule()))));
} }
template <class CPUExecutor, class IOExecutor> template <class CPUExecutor, class IOExecutor>
void sugar(CPUExecutor cpu, IOExecutor io) { void sugar(CPUExecutor cpu, IOExecutor io) {
// f on cpu - g on cpu (implicit: a single task on the cpu executor runs all // f on cpu - g on cpu (implicit: a single task on the cpu executor runs all
// the functions) // the functions)
cpu | op::transform([](auto) { return f(); }) | cpu.schedule() | op::transform([](auto) { return f(); }) |
op::transform([](f_t ft) { return g(ft); }) | op::submit([](g_t) {}); op::transform([](f_t ft) { return g(ft); }) | op::submit([](g_t) {});
// f on cpu - g on cpu (explicit: the first cpu task runs f and a second cpu // f on cpu - g on cpu (explicit: the first cpu task runs f and a second cpu
// task runs g) // task runs g)
cpu | op::transform([](auto) { return f(); }) | op::via(mi::strands(cpu)) | cpu.schedule() | op::transform([](auto) { return f(); }) | op::via(mi::strands(cpu)) |
op::transform([](f_t ft) { return g(ft); }) | op::submit([](g_t) {}); op::transform([](f_t ft) { return g(ft); }) | op::submit([](g_t) {});
// f on io - g on cpu // f on io - g on cpu
io | op::transform([](auto) { return f(); }) | op::via(mi::strands(cpu)) | io.schedule() | op::transform([](auto) { return f(); }) | op::via(mi::strands(cpu)) |
op::transform([](f_t ft) { return g(ft); }) | op::submit([](g_t) {}); op::transform([](f_t ft) { return g(ft); }) | op::submit([](g_t) {});
} }
...@@ -75,7 +76,7 @@ void pipe(CPUExecutor cpu, IOExecutor io) { ...@@ -75,7 +76,7 @@ void pipe(CPUExecutor cpu, IOExecutor io) {
// f on cpu - g on cpu (implicit: a single task on the cpu executor runs all // f on cpu - g on cpu (implicit: a single task on the cpu executor runs all
// the functions) // the functions)
mi::pipe( mi::pipe(
cpu, cpu.schedule(),
op::transform([](auto) { return f(); }), op::transform([](auto) { return f(); }),
op::transform([](f_t ft) { return g(ft); }), op::transform([](f_t ft) { return g(ft); }),
op::submit([](g_t) {})); op::submit([](g_t) {}));
...@@ -83,7 +84,7 @@ void pipe(CPUExecutor cpu, IOExecutor io) { ...@@ -83,7 +84,7 @@ void pipe(CPUExecutor cpu, IOExecutor io) {
// f on cpu - g on cpu (explicit: the first cpu task runs f and a second cpu // f on cpu - g on cpu (explicit: the first cpu task runs f and a second cpu
// task runs g) // task runs g)
mi::pipe( mi::pipe(
cpu, cpu.schedule(),
op::transform([](auto) { return f(); }), op::transform([](auto) { return f(); }),
op::via(mi::strands(cpu)), op::via(mi::strands(cpu)),
op::transform([](f_t ft) { return g(ft); }), op::transform([](f_t ft) { return g(ft); }),
...@@ -91,7 +92,7 @@ void pipe(CPUExecutor cpu, IOExecutor io) { ...@@ -91,7 +92,7 @@ void pipe(CPUExecutor cpu, IOExecutor io) {
// f on io - g on cpu // f on io - g on cpu
mi::pipe( mi::pipe(
io, io.schedule(),
op::transform([](auto) { return f(); }), op::transform([](auto) { return f(); }),
op::via(mi::strands(cpu)), op::via(mi::strands(cpu)),
op::transform([](f_t ft) { return g(ft); }), op::transform([](f_t ft) { return g(ft); }),
......
...@@ -13,26 +13,23 @@ ...@@ -13,26 +13,23 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
#include <algorithm> #include <cstdio>
#include <cassert>
#include <iostream> #include <iostream>
#include <vector>
#include <folly/experimental/pushmi/examples/pool.h>
#include <folly/experimental/pushmi/strand.h> #include <folly/experimental/pushmi/strand.h>
#include <folly/experimental/pushmi/o/request_via.h> #include <folly/experimental/pushmi/o/request_via.h>
#include <folly/experimental/pushmi/o/tap.h> #include <folly/experimental/pushmi/o/tap.h>
#include <folly/experimental/pushmi/o/transform.h> #include <folly/experimental/pushmi/o/transform.h>
#include <folly/experimental/pushmi/examples/pool.h>
using namespace folly::pushmi::aliases; using namespace folly::pushmi::aliases;
template <class Io> template <class Io>
auto io_operation(Io io) { auto io_operation(Io io) {
return io | op::transform([](auto) { return 42; }) | return io.schedule() | op::transform([](auto) { return 42; }) |
op::tap([](int v) { printf("io pool producing, %d\n", v); }) | op::tap([](int v) { std::printf("io pool producing, %d\n", v); }) |
op::request_via(); op::request_via();
} }
...@@ -44,15 +41,18 @@ int main() { ...@@ -44,15 +41,18 @@ int main() {
auto cpu = cpuPool.executor(); auto cpu = cpuPool.executor();
io_operation(io).via(mi::strands(cpu)) | io_operation(io).via(mi::strands(cpu)) |
op::tap([](int v) { printf("cpu pool processing, %d\n", v); }) | op::tap([](int v) { std::printf("cpu pool processing, %d\n", v); }) |
op::submit(); op::submit();
// when the caller is not going to process the result (only side-effect // when the caller is not going to process the result (only side-effect
// matters) or the caller is just going to push the result into a queue. // matters) or the caller is just going to push the result into a queue,
// provide a way to skip the transition to a different executor and make it // provide a way to skip the transition to a different executor and make it
// stand out so that it has to be justified in code reviews. // stand out so that it has to be justified in code reviews.
mi::via_cast<mi::is_sender<>>(io_operation(io)) | op::submit(); mi::via_cast<mi::is_sender<>>(io_operation(io)) | op::submit();
io = mi::pool_executor{};
cpu = mi::pool_executor{};
ioPool.wait(); ioPool.wait();
cpuPool.wait(); cpuPool.wait();
......
...@@ -32,7 +32,7 @@ auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) { ...@@ -32,7 +32,7 @@ auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) {
auto&& func, auto&& func,
auto sb, auto sb,
auto se, auto se,
auto out) { auto out) mutable {
using RS = decltype(selector); using RS = decltype(selector);
using F = std::conditional_t< using F = std::conditional_t<
std::is_lvalue_reference<decltype(func)>::value, std::is_lvalue_reference<decltype(func)>::value,
...@@ -42,63 +42,70 @@ auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) { ...@@ -42,63 +42,70 @@ auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) {
try { try {
typename std::allocator_traits<Allocator>::template rebind_alloc<char> typename std::allocator_traits<Allocator>::template rebind_alloc<char>
allocState(a); allocState(a);
auto shared_state = std::allocate_shared<std::tuple< using Acc = decltype(init(input));
std::exception_ptr, // first exception struct shared_state_type {
Out, // destination std::exception_ptr first_exception_{};
RS, // selector Out destination_;
F, // func RS selector_;
std::atomic<decltype(init(input))>, // accumulation F func_;
std::atomic<std::size_t>, // pending std::atomic<Acc> accumulation_;
std::atomic<std::size_t> // exception count (protects assignment to std::atomic<std::size_t> pending_{1};
// first exception) std::atomic<std::size_t> exception_count_{0}; // protects assignment to
>>( // first exception
shared_state_type(Out&& destination, RS&& selector, F&& func, Acc acc)
: destination_((Out&&) destination)
, selector_((RS&&) selector)
, func_((F&&) func)
, accumulation_(acc)
{}
};
auto shared_state = std::allocate_shared<shared_state_type>(
allocState, allocState,
std::exception_ptr{},
std::move(out), std::move(out),
std::move(selector), std::move(selector),
(decltype(func)&&)func, (decltype(func)&&)func,
init(std::move(input)), init(std::move(input)));
1, e.schedule() | op::submit([e, sb, se, shared_state](auto) mutable {
0);
e | op::submit([e, sb, se, shared_state](auto) {
auto stepDone = [](auto shared_state) { auto stepDone = [](auto shared_state) {
// pending // pending
if (--std::get<5>(*shared_state) == 0) { if (--shared_state->pending_ == 0) {
// first exception // first exception
if (std::get<0>(*shared_state)) { if (shared_state->first_exception_) {
mi::set_error( mi::set_error(
std::get<1>(*shared_state), std::get<0>(*shared_state)); shared_state->destination_, shared_state->first_exception_);
return; return;
} }
try { try {
// selector(accumulation) // selector(accumulation)
auto result = std::get<2>(*shared_state)( auto result = shared_state->selector_(
std::move(std::get<4>(*shared_state).load())); std::move(shared_state->accumulation_.load()));
mi::set_value(std::get<1>(*shared_state), std::move(result)); mi::set_value(shared_state->destination_, std::move(result));
mi::set_done(shared_state->destination_);
} catch (...) { } catch (...) {
mi::set_error( mi::set_error(
std::get<1>(*shared_state), std::current_exception()); shared_state->destination_, std::current_exception());
} }
} }
}; };
for (decltype(sb) idx{sb}; idx != se; for (decltype(sb) idx{sb}; idx != se; ++idx) {
++idx, ++std::get<5>(*shared_state)) { ++shared_state->pending_;
e | op::submit([shared_state, idx, stepDone](auto ex) { e.schedule() | op::submit([shared_state, idx, stepDone](auto) {
try { try {
// this indicates to me that bulk is not the right abstraction // this indicates to me that bulk is not the right abstraction
auto old = std::get<4>(*shared_state).load(); auto old = shared_state->accumulation_.load();
auto step = old; Acc step;
do { do {
step = old; step = old;
// func(accumulation, idx) // func(accumulation, idx)
std::get<3> (*shared_state)(step, idx); shared_state->func_(step, idx);
} while (!std::get<4>(*shared_state) } while (!shared_state->accumulation_
.compare_exchange_strong(old, step)); .compare_exchange_strong(old, step));
} catch (...) { } catch (...) {
// exception count // exception count
if (std::get<6>(*shared_state)++ == 0) { if (shared_state->exception_count_++ == 0) {
// store first exception // store first exception
std::get<0>(*shared_state) = std::current_exception(); shared_state->first_exception_ = std::current_exception();
} // else eat the exception } // else eat the exception
} }
stepDone(shared_state); stepDone(shared_state);
...@@ -107,7 +114,7 @@ auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) { ...@@ -107,7 +114,7 @@ auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) {
stepDone(shared_state); stepDone(shared_state);
}); });
} catch (...) { } catch (...) {
e | e.schedule() |
op::submit([out = std::move(out), ep = std::current_exception()]( op::submit([out = std::move(out), ep = std::current_exception()](
auto) mutable { mi::set_error(out, ep); }); auto) mutable { mi::set_error(out, ep); });
} }
......
...@@ -37,6 +37,7 @@ auto inline_bulk_target() { ...@@ -37,6 +37,7 @@ auto inline_bulk_target() {
} }
auto result = selector(std::move(acc)); auto result = selector(std::move(acc));
mi::set_value(out, std::move(result)); mi::set_value(out, std::move(result));
mi::set_done(out);
} catch (...) { } catch (...) {
mi::set_error(out, std::current_exception()); mi::set_error(out, std::current_exception());
} }
......
...@@ -4,7 +4,7 @@ ...@@ -4,7 +4,7 @@
I wrote an operator called `bulk()` and implemented for_each and reduce in terms of it. I departed from the `bulk_execute()` signature and tried to model the reduce signature on my `bulk` operator. I am not satisfied with the result and would need to invest more to get an abstraction for bulk that I was confident was minimal and efficient. I wrote an operator called `bulk()` and implemented for_each and reduce in terms of it. I departed from the `bulk_execute()` signature and tried to model the reduce signature on my `bulk` operator. I am not satisfied with the result and would need to invest more to get an abstraction for bulk that I was confident was minimal and efficient.
# Background # Background
## bulk_execute ## bulk_execute
...@@ -43,13 +43,13 @@ auto bulk( ...@@ -43,13 +43,13 @@ auto bulk(
RS&& selector); RS&& selector);
``` ```
The `bulk` function packages the parameters and returns an adapter function. The `bulk` function packages the parameters and returns an adapter function.
> A Sender is an object with a `submit()` method > A Sender is an object with a `submit()` method
> An Adapter is a function that takes a Sender and returns a Sender. Adapters are used for composition. > An Adapter is a function that takes a Sender and returns a Sender. Adapters are used for composition.
When called, the Adapter from `bulk()` will package the Adapter parameter with the original parameters and return a Sender. When called, the Adapter from `bulk()` will package the Adapter parameter with the original parameters and return a Sender.
> A Receiver is an object that has methods like `value()`, `error()` and `done()`. A Receiver is like a Promise. > A Receiver is an object that has methods like `value()`, `error()` and `done()`. A Receiver is like a Promise.
...@@ -78,16 +78,17 @@ void inline_driver( ...@@ -78,16 +78,17 @@ void inline_driver(
} }
auto result = selector(std::move(acc)); auto result = selector(std::move(acc));
mi::set_value(out, std::move(result)); mi::set_value(out, std::move(result));
mi::set_done(out);
} catch(...) { } catch(...) {
mi::set_error(out, std::current_exception()); mi::set_error(out, std::current_exception());
} }
}; };
``` ```
> ways to improve bulk: > ways to improve bulk:
> - merge ShapeBegin and ShapeEnd into a Range. > - merge ShapeBegin and ShapeEnd into a Range.
> - pass out to selector so that it can deliver an error or a success. > - pass out to selector so that it can deliver an error or a success.
> - initFunc multiple times to have context local state that does not need locking or CAS loop. > - initFunc multiple times to have context local state that does not need locking or CAS loop.
> - compose the driver implementations from operators rather than each having a bespoke implementation > - compose the driver implementations from operators rather than each having a bespoke implementation
# for_each # for_each
...@@ -97,18 +98,18 @@ implementing for_each was straight-forward with the interface. ...@@ -97,18 +98,18 @@ implementing for_each was straight-forward with the interface.
```cpp ```cpp
template<class ExecutionPolicy, class RandomAccessIterator, class Function> template<class ExecutionPolicy, class RandomAccessIterator, class Function>
void for_each( void for_each(
ExecutionPolicy&& policy, ExecutionPolicy&& policy,
RandomAccessIterator begin, RandomAccessIterator begin,
RandomAccessIterator end, RandomAccessIterator end,
Function f) Function f)
{ {
operators::just(0) | operators::just(0) |
operators::bulk( operators::bulk(
[f](auto& acc, auto cursor){ f(*cursor); }, [f](auto& acc, auto cursor){ f(*cursor); },
begin, begin,
end, end,
policy, policy,
[](auto&& args){ return args; }, [](auto&& args){ return args; },
[](auto&& acc){ return 0; }) | [](auto&& acc){ return 0; }) |
operators::blocking_submit(); operators::blocking_submit();
} }
...@@ -116,7 +117,7 @@ void for_each( ...@@ -116,7 +117,7 @@ void for_each(
The oddity is that bulk is expecting a shared state value and a value as input and a value result. Since for_each does not have shared state, this is overhead that becomes obvious and disturbing when looking at the naive concurrent driver in the code (there is a CAS loop around the call to the state update function even though the state is not updated here). The oddity is that bulk is expecting a shared state value and a value as input and a value result. Since for_each does not have shared state, this is overhead that becomes obvious and disturbing when looking at the naive concurrent driver in the code (there is a CAS loop around the call to the state update function even though the state is not updated here).
# reduce # reduce
implementing reduce took more effort and some of the code in the drivers and parameters to the driver were modified to get reduce working. implementing reduce took more effort and some of the code in the drivers and parameters to the driver were modified to get reduce working.
...@@ -124,23 +125,23 @@ implementing reduce took more effort and some of the code in the drivers and par ...@@ -124,23 +125,23 @@ implementing reduce took more effort and some of the code in the drivers and par
template<class ExecutionPolicy, class ForwardIt, class T, class BinaryOp> template<class ExecutionPolicy, class ForwardIt, class T, class BinaryOp>
T reduce( T reduce(
ExecutionPolicy&& policy, ExecutionPolicy&& policy,
ForwardIt begin, ForwardIt begin,
ForwardIt end, ForwardIt end,
T init, T init,
BinaryOp binary_op){ BinaryOp binary_op){
return operators::just(std::move(init)) | return operators::just(std::move(init)) |
operators::bulk( operators::bulk(
[binary_op](auto& acc, auto cursor){ acc = binary_op(acc, *cursor); }, [binary_op](auto& acc, auto cursor){ acc = binary_op(acc, *cursor); },
begin, begin,
end, end,
policy, policy,
[](auto&& args){ return args; }, [](auto&& args){ return args; },
[](auto&& acc){ return acc; }) | [](auto&& acc){ return acc; }) |
operators::get<T>; operators::get<T>;
} }
``` ```
Based on examples that I have been shown, the existing bulk_execute would expect the bulk_execute caller to provide the synchronization for the shared state. In the case of reduce it is important to synchronize when the execution is concurrent and equally important not to synchronize when it is not concurrent. Using if constexpr to implement reduce with and without sync in the parameters to bulk would add a lot of unsafe bespoke work and complication into every algorithm using bulk. In this bulk design the driver owns synchronization instead. Based on examples that I have been shown, the existing bulk_execute would expect the bulk_execute caller to provide the synchronization for the shared state. In the case of reduce it is important to synchronize when the execution is concurrent and equally important not to synchronize when it is not concurrent. Using if constexpr to implement reduce with and without sync in the parameters to bulk would add a lot of unsafe bespoke work and complication into every algorithm using bulk. In this bulk design the driver owns synchronization instead.
> NOTE - in any case, if a high-level async library design requires manual lock or lock-free primitive usage for correct behavior, then the design needs to be changed. > NOTE - in any case, if a high-level async library design requires manual lock or lock-free primitive usage for correct behavior, then the design needs to be changed.
......
...@@ -34,7 +34,7 @@ auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) { ...@@ -34,7 +34,7 @@ auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) {
auto&& func, auto&& func,
auto sb, auto sb,
auto se, auto se,
auto out) { auto out) mutable {
using RS = decltype(selector); using RS = decltype(selector);
using F = std::conditional_t< using F = std::conditional_t<
std::is_lvalue_reference<decltype(func)>::value, std::is_lvalue_reference<decltype(func)>::value,
...@@ -44,63 +44,70 @@ auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) { ...@@ -44,63 +44,70 @@ auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) {
try { try {
typename std::allocator_traits<Allocator>::template rebind_alloc<char> typename std::allocator_traits<Allocator>::template rebind_alloc<char>
allocState(a); allocState(a);
auto shared_state = std::allocate_shared<std::tuple< using Acc = decltype(init(input));
std::exception_ptr, // first exception struct shared_state_type {
Out, // destination std::exception_ptr first_exception_{};
RS, // selector Out destination_;
F, // func RS selector_;
std::atomic<decltype(init(input))>, // accumulation F func_;
std::atomic<std::size_t>, // pending std::atomic<Acc> accumulation_;
std::atomic<std::size_t> // exception count (protects assignment to std::atomic<std::size_t> pending_{1};
// first exception) std::atomic<std::size_t> exception_count_{0}; // protects assignment to
>>( // first exception
shared_state_type(Out&& destination, RS&& selector, F&& func, Acc acc)
: destination_((Out&&) destination)
, selector_((RS&&) selector)
, func_((F&&) func)
, accumulation_(acc)
{}
};
auto shared_state = std::allocate_shared<shared_state_type>(
allocState, allocState,
std::exception_ptr{},
std::move(out), std::move(out),
std::move(selector), std::move(selector),
(decltype(func)&&)func, (decltype(func)&&)func,
init(std::move(input)), init(std::move(input)));
1, e.schedule() | op::submit([e, sb, se, shared_state](auto) mutable {
0);
e | op::submit([e, sb, se, shared_state](auto) {
auto stepDone = [](auto shared_state) { auto stepDone = [](auto shared_state) {
// pending // pending
if (--std::get<5>(*shared_state) == 0) { if (--shared_state->pending_ == 0) {
// first exception // first exception
if (std::get<0>(*shared_state)) { if (shared_state->first_exception_) {
mi::set_error( mi::set_error(
std::get<1>(*shared_state), std::get<0>(*shared_state)); shared_state->destination_, shared_state->first_exception_);
return; return;
} }
try { try {
// selector(accumulation) // selector(accumulation)
auto result = std::get<2>(*shared_state)( auto result = shared_state->selector_(
std::move(std::get<4>(*shared_state).load())); std::move(shared_state->accumulation_.load()));
mi::set_value(std::get<1>(*shared_state), std::move(result)); mi::set_value(shared_state->destination_, std::move(result));
mi::set_done(shared_state->destination_);
} catch (...) { } catch (...) {
mi::set_error( mi::set_error(
std::get<1>(*shared_state), std::current_exception()); shared_state->destination_, std::current_exception());
} }
} }
}; };
for (decltype(sb) idx{sb}; idx != se; for (decltype(sb) idx{sb}; idx != se; ++idx) {
++idx, ++std::get<5>(*shared_state)) { ++shared_state->pending_;
e | op::submit([shared_state, idx, stepDone](auto ex) { e.schedule() | op::submit([shared_state, idx, stepDone](auto) {
try { try {
// this indicates to me that bulk is not the right abstraction // this indicates to me that bulk is not the right abstraction
auto old = std::get<4>(*shared_state).load(); auto old = shared_state->accumulation_.load();
auto step = old; Acc step;
do { do {
step = old; step = old;
// func(accumulation, idx) // func(accumulation, idx)
std::get<3> (*shared_state)(step, idx); shared_state->func_(step, idx);
} while (!std::get<4>(*shared_state) } while (!shared_state->accumulation_
.compare_exchange_strong(old, step)); .compare_exchange_strong(old, step));
} catch (...) { } catch (...) {
// exception count // exception count
if (std::get<6>(*shared_state)++ == 0) { if (shared_state->exception_count_++ == 0) {
// store first exception // store first exception
std::get<0>(*shared_state) = std::current_exception(); shared_state->first_exception_ = std::current_exception();
} // else eat the exception } // else eat the exception
} }
stepDone(shared_state); stepDone(shared_state);
...@@ -109,7 +116,7 @@ auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) { ...@@ -109,7 +116,7 @@ auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) {
stepDone(shared_state); stepDone(shared_state);
}); });
} catch (...) { } catch (...) {
e | e.schedule() |
op::submit([out = std::move(out), ep = std::current_exception()]( op::submit([out = std::move(out), ep = std::current_exception()](
auto) mutable { mi::set_error(out, ep); }); auto) mutable { mi::set_error(out, ep); });
} }
......
...@@ -38,6 +38,7 @@ auto inline_bulk_target() { ...@@ -38,6 +38,7 @@ auto inline_bulk_target() {
} }
auto result = selector(std::move(acc)); auto result = selector(std::move(acc));
mi::set_value(out, std::move(result)); mi::set_value(out, std::move(result));
mi::set_done(out);
} catch (...) { } catch (...) {
mi::set_error(out, std::current_exception()); mi::set_error(out, std::current_exception());
} }
......
...@@ -14,24 +14,25 @@ ...@@ -14,24 +14,25 @@
* limitations under the License. * limitations under the License.
*/ */
#include <algorithm> #include <algorithm>
#include <cassert>
#include <iostream>
#include <vector>
#include <futures.h>
#include <futures_static_thread_pool.h>
#include <atomic> #include <atomic>
#include <cassert>
#include <functional> #include <functional>
#include <iostream>
#include <memory> #include <memory>
#include <thread> #include <thread>
#include <utility> #include <utility>
#include <vector>
#include <folly/experimental/pushmi/examples/pool.h> #include <folly/experimental/pushmi/strand.h>
#include <folly/experimental/pushmi/o/just.h> #include <folly/experimental/pushmi/o/just.h>
#include <folly/experimental/pushmi/o/transform.h> #include <folly/experimental/pushmi/o/transform.h>
#include <folly/experimental/pushmi/o/via.h> #include <folly/experimental/pushmi/o/via.h>
#include <folly/experimental/pushmi/strand.h>
#include <folly/experimental/pushmi/examples/pool.h>
// // See https://github.com/executors/futures-impl
// #include <futures.h>
// #include <futures_static_thread_pool.h>
using namespace folly::pushmi::aliases; using namespace folly::pushmi::aliases;
...@@ -51,54 +52,54 @@ struct inline_executor { ...@@ -51,54 +52,54 @@ struct inline_executor {
void execute(Function f) const noexcept { void execute(Function f) const noexcept {
f(); f();
} }
constexpr bool query(std::experimental::execution::oneway_t) { // constexpr bool query(std::experimental::execution::oneway_t) {
return true; // return true;
} // }
constexpr bool query(std::experimental::execution::twoway_t) { // constexpr bool query(std::experimental::execution::twoway_t) {
return false; // return false;
} // }
constexpr bool query(std::experimental::execution::single_t) { // constexpr bool query(std::experimental::execution::single_t) {
return true; // return true;
} // }
}; };
namespace p1054 { // namespace p1054 {
// A promise refers to a promise and is associated with a future, // // A promise refers to a promise and is associated with a future,
// either through type-erasure or through construction of an // // either through type-erasure or through construction of an
// underlying promise with an overload of make_promise_contract(). // // underlying promise with an overload of make_promise_contract().
//
// make_promise_contract() cannot be written to produce a lazy future. // // make_promise_contract() cannot be written to produce a lazy future.
// the promise has to exist prior to .then() getting a continuation. // // The promise has to exist prior to .then() getting a continuation.
// there must be a shared allocation to connect the promise and future. // // there must be a shared allocation to connect the promise and future.
template <class T, class Executor> // template <class T, class Executor>
std::pair< // std::pair<
std::experimental::standard_promise<T>, // std::experimental::standard_promise<T>,
std::experimental::standard_future<T, std::decay_t<Executor>>> // std::experimental::standard_future<T, std::decay_t<Executor>>>
make_promise_contract(const Executor& e) { // make_promise_contract(const Executor& e) {
std::experimental::standard_promise<T> promise; // std::experimental::standard_promise<T> promise;
auto ex = e; // auto ex = e;
return {promise, promise.get_future(std::move(ex))}; // return {promise, promise.get_future(std::move(ex))};
} // }
//
template <class Executor, class Function, class Future> // template <class Executor, class Function, class Future>
std::experimental::standard_future< // std::experimental::standard_future<
std::result_of_t< // std::result_of_t<
Function(std::decay_t<typename std::decay_t<Future>::value_type>&&)>, // Function(std::decay_t<typename std::decay_t<Future>::value_type>&&)>,
std::decay_t<Executor>> // std::decay_t<Executor>>
then_execute(Executor&& e, Function&& f, Future&& pred) { // then_execute(Executor&& e, Function&& f, Future&& pred) {
using V = std::decay_t<typename std::decay_t<Future>::value_type>; // using V = std::decay_t<typename std::decay_t<Future>::value_type>;
using T = std::result_of_t<Function(V &&)>; // using T = std::result_of_t<Function(V &&)>;
auto pc = make_promise_contract<T>(e); // auto pc = make_promise_contract<T>(e);
auto p = std::get<0>(pc); // auto p = std::get<0>(pc);
auto r = std::get<1>(pc); // auto r = std::get<1>(pc);
((Future &&) pred).then([e, p, f](V v) mutable { // ((Future &&) pred).then([e, p, f](V v) mutable {
e.execute([p, f, v]() mutable { p.set_value(f(v)); }); // e.execute([p, f, v]() mutable { p.set_value(f(v)); });
return 0; // return 0;
}); // });
return r; // return r;
} // }
//
} // namespace p1054 // } // namespace p1054
namespace p1055 { namespace p1055 {
...@@ -113,24 +114,25 @@ auto then_execute(Executor&& e, Function&& f, Future&& pred) { ...@@ -113,24 +114,25 @@ auto then_execute(Executor&& e, Function&& f, Future&& pred) {
int main() { int main() {
mi::pool p{std::max(1u, std::thread::hardware_concurrency())}; mi::pool p{std::max(1u, std::thread::hardware_concurrency())};
std::experimental::futures_static_thread_pool sp{
std::max(1u, std::thread::hardware_concurrency())};
auto pc = p1054::make_promise_contract<int>(inline_executor{});
auto& pr = std::get<0>(pc);
auto& r = std::get<1>(pc);
auto f = p1054::then_execute(
sp.executor(), [](int v) { return v * 2; }, std::move(r));
pr.set_value(42);
f.get();
p1055::then_execute(p.executor(), [](int v) { return v * 2; }, op::just(21)) | p1055::then_execute(p.executor(), [](int v) { return v * 2; }, op::just(21)) |
op::get<int>; op::get<int>;
sp.stop();
sp.wait();
p.stop(); p.stop();
p.wait(); p.wait();
// std::experimental::futures_static_thread_pool sp{
// std::max(1u, std::thread::hardware_concurrency())};
//
// auto pc = p1054::make_promise_contract<int>(inline_executor{});
// auto& pr = std::get<0>(pc);
// auto& r = std::get<1>(pc);
// auto f = p1054::then_execute(
// sp.executor(), [](int v) { return v * 2; }, std::move(r));
// pr.set_value(42);
// f.get();
//
// sp.stop();
// sp.wait();
std::cout << "OK" << std::endl; std::cout << "OK" << std::endl;
} }
...@@ -14,60 +14,61 @@ ...@@ -14,60 +14,61 @@
* limitations under the License. * limitations under the License.
*/ */
#include <algorithm> #include <algorithm>
#include <cassert>
#include <iostream>
#include <vector>
#include <futures.h>
#include <atomic> #include <atomic>
#include <cassert>
#include <functional> #include <functional>
#include <iostream>
#include <memory> #include <memory>
#include <thread> #include <thread>
#include <utility> #include <utility>
#include <vector>
#include <folly/experimental/pushmi/examples/pool.h>
#include <folly/experimental/pushmi/o/transform.h> #include <folly/experimental/pushmi/o/transform.h>
using namespace folly::pushmi::aliases; #include <folly/experimental/pushmi/examples/pool.h>
namespace p1054 { using namespace folly::pushmi::aliases;
// A promise refers to a promise and is associated with a future,
// either through type-erasure or through construction of an
// underlying promise with an overload of make_promise_contract().
// make_promise_contract() cannot be written to produce a lazy future. // // See https://github.com/executors/futures-impl
// the promise has to exist prior to .then() getting a continuation. // #include <futures.h>
// there must be a shared allocation to connect the promise and future. //
template <class T, class Executor> // namespace p1054 {
std::pair< // // A promise refers to a promise and is associated with a future,
std::experimental::standard_promise<T>, // // either through type-erasure or through construction of an
std::experimental::standard_future<T, std::decay_t<Executor>>> // // underlying promise with an overload of make_promise_contract().
make_promise_contract(const Executor& e) { //
std::experimental::standard_promise<T> promise; // // make_promise_contract() cannot be written to produce a lazy future.
auto ex = e; // // the promise has to exist prior to .then() getting a continuation.
return {promise, promise.get_future(std::move(ex))}; // // there must be a shared allocation to connect the promise and future.
} // template <class T, class Executor>
// std::pair<
template <class Executor, class Function> // std::experimental::standard_promise<T>,
std::experimental::standard_future< // std::experimental::standard_future<T, std::decay_t<Executor>>>
std::result_of_t<std::decay_t<Function>()>, // make_promise_contract(const Executor& e) {
std::decay_t<Executor>> // std::experimental::standard_promise<T> promise;
twoway_execute(Executor&& e, Function&& f) { // auto ex = e;
using T = std::result_of_t<std::decay_t<Function>()>; // return {promise, promise.get_future(std::move(ex))};
auto pc = make_promise_contract<T>(e); // }
auto p = std::get<0>(pc); //
auto r = std::get<1>(pc); // template <class Executor, class Function>
e.execute([p, f]() mutable { p.set_value(f()); }); // std::experimental::standard_future<
return r; // std::result_of_t<std::decay_t<Function>()>,
} // std::decay_t<Executor>>
} // namespace p1054 // twoway_execute(Executor&& e, Function&& f) {
// using T = std::result_of_t<std::decay_t<Function>()>;
// auto pc = make_promise_contract<T>(e);
// auto p = std::get<0>(pc);
// auto r = std::get<1>(pc);
// e.execute([p, f]() mutable { p.set_value(f()); });
// return r;
// }
// } // namespace p1054
namespace p1055 { namespace p1055 {
template <class Executor, class Function> template <class Executor, class Function>
auto twoway_execute(Executor&& e, Function&& f) { auto twoway_execute(Executor&& e, Function&& f) {
return e | op::transform([f](auto) { return f(); }); return e.schedule() | op::transform([f](auto) { return f(); });
} }
} // namespace p1055 } // namespace p1055
...@@ -75,17 +76,18 @@ auto twoway_execute(Executor&& e, Function&& f) { ...@@ -75,17 +76,18 @@ auto twoway_execute(Executor&& e, Function&& f) {
int main() { int main() {
mi::pool p{std::max(1u, std::thread::hardware_concurrency())}; mi::pool p{std::max(1u, std::thread::hardware_concurrency())};
std::experimental::static_thread_pool sp{
std::max(1u, std::thread::hardware_concurrency())};
p1054::twoway_execute(sp.executor(), []() { return 42; }).get();
p1055::twoway_execute(p.executor(), []() { return 42; }) | op::get<int>; p1055::twoway_execute(p.executor(), []() { return 42; }) | op::get<int>;
sp.stop();
sp.wait();
p.stop(); p.stop();
p.wait(); p.wait();
// std::experimental::static_thread_pool sp{
// std::max(1u, std::thread::hardware_concurrency())};
//
// p1054::twoway_execute(sp.executor(), []() { return 42; }).get();
//
// sp.stop();
// sp.wait();
std::cout << "OK" << std::endl; std::cout << "OK" << std::endl;
} }
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