Commit ad7642cf authored by Dominik Charousset's avatar Dominik Charousset

Re-implement make_source(dest, ...)

parent 7685fb63
...@@ -50,6 +50,7 @@ ...@@ -50,6 +50,7 @@
#include "caf/stream.hpp" #include "caf/stream.hpp"
#include "caf/stream_manager.hpp" #include "caf/stream_manager.hpp"
#include "caf/stream_result.hpp" #include "caf/stream_result.hpp"
#include "caf/stream_result_handler_trait.hpp"
#include "caf/stream_result_trait.hpp" #include "caf/stream_result_trait.hpp"
#include "caf/stream_source_trait.hpp" #include "caf/stream_source_trait.hpp"
#include "caf/to_string.hpp" #include "caf/to_string.hpp"
...@@ -429,12 +430,23 @@ public: ...@@ -429,12 +430,23 @@ public:
} }
/// Creates an output path for the given stage without any type checking. /// Creates an output path for the given stage without any type checking.
/// @private
template <class Out, class... Ts> template <class Out, class... Ts>
output_stream<Out, Ts...> add_unsafe_output_path(stream_manager_ptr mgr) { output_stream<Out, Ts...> add_unsafe_output_path(stream_manager_ptr mgr) {
auto slot = assign_next_pending_slot(mgr); auto slot = assign_next_pending_slot(mgr);
return {0, slot, std::move(mgr)}; return {0, slot, std::move(mgr)};
} }
/// Creates an output path for the given stage without any type checking.
/// @pre `next != nullptr`
/// @pre `pending_stream_managers_[slot] == mgr`
/// @pre `mgr->out().terminal() == false`
/// @private
void add_unsafe_output_path(stream_manager_ptr mgr, strong_actor_ptr next,
stream_slot slot, strong_actor_ptr origin,
mailbox_element::forwarding_stack stages,
message_id mid);
/// Creates an input path for given stage. /// Creates an input path for given stage.
template <class In, class Result, class Out, class Scatterer, class... Ts> template <class In, class Result, class Out, class Scatterer, class... Ts>
make_sink_result<In, Result> make_sink_result<In, Result>
...@@ -445,6 +457,7 @@ public: ...@@ -445,6 +457,7 @@ public:
} }
/// Creates an input path for given stage. /// Creates an input path for given stage.
/// @private
template <class Result, class In> template <class Result, class In>
stream_result<Result> add_unsafe_input_path(const stream<In>&, stream_result<Result> add_unsafe_input_path(const stream<In>&,
stream_manager_ptr mgr) { stream_manager_ptr mgr) {
...@@ -452,7 +465,7 @@ public: ...@@ -452,7 +465,7 @@ public:
return {slot, std::move(mgr)}; return {slot, std::move(mgr)};
} }
/// Creates a new stream source and starts streaming to `dest`. /// Creates a new stream source from `Driver`.
/// @param dest Actor handle to the stream destination. /// @param dest Actor handle to the stream destination.
/// @param xs User-defined handshake payload. /// @param xs User-defined handshake payload.
/// @param init Function object for initializing the state of the source. /// @param init Function object for initializing the state of the source.
...@@ -518,6 +531,54 @@ public: ...@@ -518,6 +531,54 @@ public:
scatterer_type); scatterer_type);
} }
/// Creates a new stream source and adds `dest` as first outbound path to it.
template <class ActorHandle, class... Ts, class Init, class Pull, class Done,
class HandleResult,
class Scatterer =
broadcast_scatterer<typename stream_source_trait_t<Pull>::output>,
class Trait = stream_source_trait_t<Pull>>
detail::enable_if_t<detail::is_actor_handle<ActorHandle>::value,
typename make_source_result_t<Scatterer, Ts...>::ptr_type>
make_source(const ActorHandle& dest, std::tuple<Ts...> xs, Init init,
Pull pull, Done done, HandleResult handle_res,
policy::arg<Scatterer> scatterer_arg = {}) {
// TODO: type-check whether `dest` is a valid next stage
// Expect `handle_result` to have signature `void (expected<T>)`.
using handle_res_trait = stream_result_handler_trait_t<HandleResult>;
static_assert(handle_res_trait::valid,
"expected a result handler with signature "
"'void (expected<T>)'");
using handle_res_result = typename handle_res_trait::result;
auto result = make_source(std::move(xs), std::move(init), std::move(pull),
std::move(done), scatterer_arg);
auto mid = new_request_id(message_priority::normal);
add_unsafe_output_path(result.ptr(), actor_cast<strong_actor_ptr>(dest),
result.out(), actor_cast<strong_actor_ptr>(this),
no_stages, mid);
behavior tmp{
[=](handle_res_result& x) mutable { handle_res(std::move(x)); },
[=](error& err) mutable { handle_res(std::move(err)); }
};
add_multiplexed_response_handler(mid.response_id(), std::move(tmp));
return result.ptr();
}
/// Creates a new stream source and adds `dest` as first outbound path to it.
template <class ActorHandle, class Init, class Pull, class Done,
class HandleResult,
class Scatterer =
broadcast_scatterer<typename stream_source_trait_t<Pull>::output>,
class Trait = stream_source_trait_t<Pull>>
detail::enable_if_t<detail::is_actor_handle<ActorHandle>::value,
typename make_source_result_t<Scatterer>::ptr_type>
make_source(const ActorHandle& dest, Init init, Pull pull, Done done,
HandleResult handle_res,
policy::arg<Scatterer> scatterer_arg = {}) {
return make_source(dest, std::make_tuple(), std::move(init),
std::move(pull), std::move(done), std::move(handle_res),
scatterer_arg);
}
template <class Driver, class Input, class... Ts> template <class Driver, class Input, class... Ts>
stream_result<typename Driver::output_type> stream_result<typename Driver::output_type>
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://openresult_handler.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_RESULT_HANDLER_TRAIT_HPP
#define CAF_STREAM_RESULT_HANDLER_TRAIT_HPP
#include "caf/fwd.hpp"
#include "caf/detail/type_traits.hpp"
namespace caf {
/// Deduces the input type for a stream result handler from its signature.
template <class F>
struct stream_result_handler_trait {
static constexpr bool valid = false;
using result = void;
};
template <class T>
struct stream_result_handler_trait<void (expected<T>)> {
static constexpr bool valid = true;
using result = T;
};
/// Convenience alias for extracting the function signature from `Pull` and
/// passing it to `stream_result_handler_trait`.
template <class Pull>
using stream_result_handler_trait_t =
stream_result_handler_trait<typename detail::get_callable_trait<Pull>::fun_sig>;
} // namespace caf
#endif // CAF_STREAM_RESULT_HANDLER_TRAIT_HPP
...@@ -402,6 +402,19 @@ void scheduled_actor::quit(error x) { ...@@ -402,6 +402,19 @@ void scheduled_actor::quit(error x) {
// -- stream management -------------------------------------------------------- // -- stream management --------------------------------------------------------
void scheduled_actor::add_unsafe_output_path(
stream_manager_ptr mgr, strong_actor_ptr next, stream_slot slot,
strong_actor_ptr origin, mailbox_element::forwarding_stack stages,
message_id mid) {
CAF_ASSERT(next != nullptr);
CAF_ASSERT(mgr->out().terminal() == false);
CAF_ASSERT(pending_stream_managers_[slot] == mgr);
// Build pipeline by forwarding handshake along the path.
mgr->send_handshake(std::move(next), slot, std::move(origin),
std::move(stages), mid);
mgr->generate_messages();
}
sec scheduled_actor::build_pipeline(stream_slot in, stream_slot out, sec scheduled_actor::build_pipeline(stream_slot in, stream_slot out,
stream_manager_ptr mgr) { stream_manager_ptr mgr) {
CAF_LOG_TRACE(CAF_ARG(in) << CAF_ARG(out) << CAF_ARG(mgr)); CAF_LOG_TRACE(CAF_ARG(in) << CAF_ARG(out) << CAF_ARG(mgr));
...@@ -428,15 +441,12 @@ sec scheduled_actor::build_pipeline(stream_slot in, stream_slot out, ...@@ -428,15 +441,12 @@ sec scheduled_actor::build_pipeline(stream_slot in, stream_slot out,
mgr->add_promise(make_response_promise()); mgr->add_promise(make_response_promise());
} }
} }
if (out != 0) { if (out != 0 && mgr->out().path(out) == nullptr) {
CAF_ASSERT(mgr->out().terminal() == false);
CAF_ASSERT(pending_stream_managers_[out] == mgr);
if (next == nullptr) if (next == nullptr)
return fail(sec::no_downstream_stages_defined); return fail(sec::no_downstream_stages_defined);
// Build pipeline by forwarding handshake along the path. add_unsafe_output_path(mgr, std::move(next), out, current_sender(),
mgr->send_handshake(std::move(next), out, current_sender(), take_current_forwarding_stack(),
take_current_forwarding_stack(), current_message_id()); current_message_id());
mgr->generate_messages();
} }
return sec::none; return sec::none;
} }
......
...@@ -36,36 +36,67 @@ namespace { ...@@ -36,36 +36,67 @@ namespace {
TESTEE_SETUP(); TESTEE_SETUP();
VARARGS_TESTEE(file_reader, size_t buf_size) { using buf = std::deque<int>;
using buf = std::deque<int>;
return { std::function<void(buf&)> init(size_t buf_size) {
[=](string& fname) -> output_stream<int, string> { return [=](buf& xs) {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return self->make_source(
// forward file name in handshake to next stage
std::forward_as_tuple(std::move(fname)),
// initialize state
[=](buf& xs) {
xs.resize(buf_size); xs.resize(buf_size);
std::iota(xs.begin(), xs.end(), 1); std::iota(xs.begin(), xs.end(), 1);
}, };
// get next element }
[](buf& xs, downstream<int>& out, size_t num) {
void push_from_buf(buf& xs, downstream<int>& out, size_t num) {
CAF_MESSAGE("push " << num << " messages downstream"); CAF_MESSAGE("push " << num << " messages downstream");
auto n = std::min(num, xs.size()); auto n = std::min(num, xs.size());
for (size_t i = 0; i < n; ++i) for (size_t i = 0; i < n; ++i)
out.push(xs[i]); out.push(xs[i]);
xs.erase(xs.begin(), xs.begin() + static_cast<ptrdiff_t>(n)); xs.erase(xs.begin(), xs.begin() + static_cast<ptrdiff_t>(n));
}, }
// check whether we reached the end
[=](const buf& xs) { std::function<bool(const buf&)> is_done(scheduled_actor* self) {
return [=](const buf& xs) {
if (xs.empty()) { if (xs.empty()) {
CAF_MESSAGE(self->name() << " is done"); CAF_MESSAGE(self->name() << " is done sending");
return true; return true;
} }
return false; return false;
}); };
}
VARARGS_TESTEE(file_reader, size_t buf_size) {
return {
[=](string& fname) -> output_stream<int, string> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return self->make_source(
// forward file name in handshake to next stage
std::forward_as_tuple(std::move(fname)),
init(buf_size),
push_from_buf,
is_done(self)
);
},
[=](string& fname, actor dest) {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
auto rp = self->make_response_promise();
self->make_source(
// next stage in the stream
dest,
// forward file name in handshake to next stage
std::forward_as_tuple(std::move(fname)),
init(buf_size),
push_from_buf,
is_done(self),
[=](expected<int> x) mutable {
CAF_MESSAGE(self->name() << " received the result");
if (x)
rp.deliver(*x);
else
rp.deliver(std::move(x.error()));
}
);
return rp;
} }
}; };
} }
...@@ -234,6 +265,30 @@ CAF_TEST(depth_2_pipeline_50_items) { ...@@ -234,6 +265,30 @@ CAF_TEST(depth_2_pipeline_50_items) {
CAF_CHECK_EQUAL(fail_state(src), exit_reason::normal); CAF_CHECK_EQUAL(fail_state(src), exit_reason::normal);
} }
CAF_TEST(depth_2_pipeline_setup2_50_items) {
auto src = sys.spawn(file_reader, 50);
auto snk = sys.spawn(sum_up);
CAF_MESSAGE(CAF_ARG(self) << CAF_ARG(src) << CAF_ARG(snk));
CAF_MESSAGE("initiate stream handshake");
self->send(src, "numbers.txt", snk);
expect((string, actor), from(self).to(src).with("numbers.txt", snk));
expect((open_stream_msg), from(src).to(snk));
expect((upstream_msg::ack_open), from(snk).to(src));
CAF_MESSAGE("start data transmission (a single batch)");
expect((downstream_msg::batch), from(src).to(snk));
sched.clock().current_time += cycle;
sched.dispatch();
expect((timeout_msg), from(snk).to(snk));
expect((timeout_msg), from(src).to(src));
expect((upstream_msg::ack_batch), from(snk).to(src));
CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::close), from(src).to(snk));
expect((int), from(snk).to(src).with(1275));
expect((int), from(src).to(self).with(1275));
CAF_CHECK_EQUAL(fail_state(snk), exit_reason::normal);
CAF_CHECK_EQUAL(fail_state(src), exit_reason::normal);
}
CAF_TEST(delayed_depth_2_pipeline_50_items) { CAF_TEST(delayed_depth_2_pipeline_50_items) {
auto src = sys.spawn(file_reader, 50); auto src = sys.spawn(file_reader, 50);
auto snk = sys.spawn(delayed_sum_up); auto snk = sys.spawn(delayed_sum_up);
......
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