Commit 5fc5bd29 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'issue/906'

parents 921ab3c7 401c48ff
...@@ -123,7 +123,6 @@ set(CAF_CORE_SOURCES ...@@ -123,7 +123,6 @@ set(CAF_CORE_SOURCES
src/policy/work_stealing.cpp src/policy/work_stealing.cpp
src/proxy_registry.cpp src/proxy_registry.cpp
src/raise_error.cpp src/raise_error.cpp
src/raw_event_based_actor.cpp
src/ref_counted.cpp src/ref_counted.cpp
src/replies_to.cpp src/replies_to.cpp
src/response_promise.cpp src/response_promise.cpp
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/detail/core_export.hpp"
#include "caf/event_based_actor.hpp"
namespace caf {
/// A cooperatively raw scheduled actor is a dynamically typed actor that does
/// not handle any system messages. All handler for system messages as well as
/// the default handler are ignored. This actor type is for testing and
/// system-level actors.
/// @extends event_based_actor
class CAF_CORE_EXPORT raw_event_based_actor : public event_based_actor {
public:
// -- member types -----------------------------------------------------------
/// Required by `spawn` for type deduction.
using signatures = none_t;
/// Required by `spawn` for type deduction.
using behavior_type = behavior;
// -- constructors and destructors -------------------------------------------
explicit raw_event_based_actor(actor_config& cfg);
invoke_message_result consume(mailbox_element& x) override;
};
} // namespace caf
...@@ -30,16 +30,8 @@ ...@@ -30,16 +30,8 @@
#include <unordered_map> #include <unordered_map>
#include "caf/actor_traits.hpp" #include "caf/actor_traits.hpp"
#include "caf/broadcast_downstream_manager.hpp"
#include "caf/default_downstream_manager.hpp"
#include "caf/detail/behavior_stack.hpp" #include "caf/detail/behavior_stack.hpp"
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/detail/stream_sink_driver_impl.hpp"
#include "caf/detail/stream_sink_impl.hpp"
#include "caf/detail/stream_source_driver_impl.hpp"
#include "caf/detail/stream_source_impl.hpp"
#include "caf/detail/stream_stage_driver_impl.hpp"
#include "caf/detail/stream_stage_impl.hpp"
#include "caf/detail/tick_emitter.hpp" #include "caf/detail/tick_emitter.hpp"
#include "caf/detail/unordered_flat_map.hpp" #include "caf/detail/unordered_flat_map.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
...@@ -52,7 +44,6 @@ ...@@ -52,7 +44,6 @@
#include "caf/intrusive/wdrr_dynamic_multiplexed_queue.hpp" #include "caf/intrusive/wdrr_dynamic_multiplexed_queue.hpp"
#include "caf/intrusive/wdrr_fixed_multiplexed_queue.hpp" #include "caf/intrusive/wdrr_fixed_multiplexed_queue.hpp"
#include "caf/invoke_message_result.hpp" #include "caf/invoke_message_result.hpp"
#include "caf/is_actor_handle.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/make_sink_result.hpp" #include "caf/make_sink_result.hpp"
...@@ -71,11 +62,7 @@ ...@@ -71,11 +62,7 @@
#include "caf/response_handle.hpp" #include "caf/response_handle.hpp"
#include "caf/scheduled_actor.hpp" #include "caf/scheduled_actor.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/stream.hpp"
#include "caf/stream_manager.hpp" #include "caf/stream_manager.hpp"
#include "caf/stream_sink_trait.hpp"
#include "caf/stream_source_trait.hpp"
#include "caf/stream_stage_trait.hpp"
#include "caf/to_string.hpp" #include "caf/to_string.hpp"
namespace caf { namespace caf {
...@@ -423,205 +410,6 @@ public: ...@@ -423,205 +410,6 @@ public:
} }
#endif // CAF_NO_EXCEPTIONS #endif // CAF_NO_EXCEPTIONS
// -- stream management ------------------------------------------------------
template <class Driver, class... Ts, class Init, class Pull, class Done,
class Finalize = unit_t>
[[deprecated("use attach_stream_source instead")]] make_source_result_t<
typename Driver::downstream_manager_type, Ts...>
make_source(std::tuple<Ts...> xs, Init init, Pull pull, Done done,
Finalize fin = {}) {
using detail::make_stream_source;
auto mgr = make_stream_source<Driver>(
this, std::move(init), std::move(pull), std::move(done), std::move(fin));
auto slot = mgr->add_outbound_path(std::move(xs));
return {slot, std::move(mgr)};
}
template <class... Ts, class Init, class Pull, class Done,
class Finalize = unit_t,
class DownstreamManager = broadcast_downstream_manager<
typename stream_source_trait_t<Pull>::output>>
[[deprecated("use attach_stream_source instead")]] make_source_result_t<
DownstreamManager, Ts...>
make_source(std::tuple<Ts...> xs, Init init, Pull pull, Done done,
Finalize fin = {}, policy::arg<DownstreamManager> = {}) {
using driver = detail::stream_source_driver_impl<DownstreamManager, Pull,
Done, Finalize>;
return make_source<driver>(std::move(xs), std::move(init), std::move(pull),
std::move(done), std::move(fin));
}
template <class Init, class Pull, class Done, class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Pull>,
class Trait = stream_source_trait_t<Pull>>
[[deprecated("use attach_stream_source instead")]] detail::enable_if_t<
!is_actor_handle<Init>::value && Trait::valid,
make_source_result_t<DownstreamManager>>
make_source(Init init, Pull pull, Done done, Finalize finalize = {},
policy::arg<DownstreamManager> token = {}) {
return make_source(std::make_tuple(), init, pull, done, finalize, token);
}
template <class ActorHandle, class... Ts, class Init, class Pull, class Done,
class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Pull>,
class Trait = stream_source_trait_t<Pull>>
[[deprecated("use attach_stream_source instead")]] detail::enable_if_t<
is_actor_handle<ActorHandle>::value,
make_source_result_t<DownstreamManager>>
make_source(const ActorHandle& dest, std::tuple<Ts...> xs, Init init,
Pull pull, Done done, Finalize fin = {},
policy::arg<DownstreamManager> = {}) {
// TODO: type checking of dest
using driver = detail::stream_source_driver_impl<DownstreamManager, Pull,
Done, Finalize>;
auto mgr = detail::make_stream_source<driver>(
this, std::move(init), std::move(pull), std::move(done), std::move(fin));
auto slot = mgr->add_outbound_path(dest, std::move(xs));
return {slot, std::move(mgr)};
}
template <class ActorHandle, class Init, class Pull, class Done,
class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Pull>,
class Trait = stream_source_trait_t<Pull>>
[[deprecated("use attach_stream_source instead")]] detail::enable_if_t<
is_actor_handle<ActorHandle>::value && Trait::valid,
make_source_result_t<DownstreamManager>>
make_source(const ActorHandle& dest, Init init, Pull pull, Done done,
Finalize fin = {}, policy::arg<DownstreamManager> token = {}) {
return make_source(dest, std::make_tuple(), std::move(init),
std::move(pull), std::move(done), std::move(fin), token);
}
template <class Driver, class Init, class Pull, class Done,
class Finalize = unit_t>
[[deprecated("use attach_continuous_stream_source instead")]] auto
make_continuous_source(Init init, Pull pull, Done done, Finalize fin = {}) {
using detail::make_stream_source;
auto mgr = make_stream_source<Driver>(
this, std::move(init), std::move(pull), std::move(done), std::move(fin));
mgr->continuous(true);
return mgr;
}
template <class Init, class Pull, class Done, class Finalize = unit_t,
class DownstreamManager = broadcast_downstream_manager<
typename stream_source_trait_t<Pull>::output>>
[[deprecated("use attach_continuous_stream_source instead")]] auto
make_continuous_source(Init init, Pull pull, Done done, Finalize fin = {},
policy::arg<DownstreamManager> = {}) {
using driver = detail::stream_source_driver_impl<DownstreamManager, Pull,
Done, Finalize>;
return make_continuous_source<driver>(std::move(init), std::move(pull),
std::move(done), std::move(fin));
}
template <class Driver, class... Ts>
[[deprecated("use attach_stream_sink instead")]] make_sink_result<
typename Driver::input_type>
make_sink(const stream<typename Driver::input_type>& src, Ts&&... xs) {
auto mgr = detail::make_stream_sink<Driver>(this, std::forward<Ts>(xs)...);
auto slot = mgr->add_inbound_path(src);
return {slot, std::move(mgr)};
}
template <class In, class Init, class Fun, class Finalize = unit_t,
class Trait = stream_sink_trait_t<Fun>>
[[deprecated("use attach_stream_sink instead")]] make_sink_result<In>
make_sink(const stream<In>& in, Init init, Fun fun, Finalize fin = {}) {
using driver = detail::stream_sink_driver_impl<In, Fun, Finalize>;
return make_sink<driver>(in, std::move(init), std::move(fun),
std::move(fin));
}
template <class Driver, class In, class... Ts, class... Us>
[[deprecated("use attach_stream_stage instead")]] make_stage_result_t<
In, typename Driver::downstream_manager_type, Ts...>
make_stage(const stream<In>& src, std::tuple<Ts...> xs, Us&&... ys) {
using detail::make_stream_stage;
auto mgr = make_stream_stage<Driver>(this, std::forward<Us>(ys)...);
auto in = mgr->add_inbound_path(src);
auto out = mgr->add_outbound_path(std::move(xs));
return {in, out, std::move(mgr)};
}
template <class In, class... Ts, class Init, class Fun,
class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Fun>,
class Trait = stream_stage_trait_t<Fun>>
[[deprecated("use attach_stream_stage instead")]] make_stage_result_t<
In, DownstreamManager, Ts...>
make_stage(const stream<In>& in, std::tuple<Ts...> xs, Init init, Fun fun,
Finalize fin = {}, policy::arg<DownstreamManager> token = {}) {
CAF_IGNORE_UNUSED(token);
CAF_ASSERT(current_mailbox_element() != nullptr);
CAF_ASSERT(current_mailbox_element()->content().types()
== make_type_id_list<open_stream_msg>());
using output_type = typename stream_stage_trait_t<Fun>::output;
using state_type = typename stream_stage_trait_t<Fun>::state;
static_assert(
std::is_same<void(state_type&),
typename detail::get_callable_trait<Init>::fun_sig>::value,
"Expected signature `void (State&)` for init function");
static_assert(
std::is_same<void(state_type&, downstream<output_type>&, In),
typename detail::get_callable_trait<Fun>::fun_sig>::value,
"Expected signature `void (State&, downstream<Out>&, In)` "
"for consume function");
using driver
= detail::stream_stage_driver_impl<typename Trait::input,
DownstreamManager, Fun, Finalize>;
return make_stage<driver>(in, std::move(xs), std::move(init),
std::move(fun), std::move(fin));
}
template <class In, class Init, class Fun, class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Fun>,
class Trait = stream_stage_trait_t<Fun>>
[[deprecated("use attach_stream_stage instead")]] make_stage_result_t<
In, DownstreamManager>
make_stage(const stream<In>& in, Init init, Fun fun, Finalize fin = {},
policy::arg<DownstreamManager> token = {}) {
return make_stage(in, std::make_tuple(), std::move(init), std::move(fun),
std::move(fin), token);
}
template <class Driver, class... Ts>
[[deprecated("use attach_continuous_stream_stage instead")]] auto
make_continuous_stage(Ts&&... xs) {
auto ptr = detail::make_stream_stage<Driver>(this, std::forward<Ts>(xs)...);
ptr->continuous(true);
return ptr;
}
template <class Init, class Fun, class Cleanup,
class DownstreamManager = default_downstream_manager_t<Fun>,
class Trait = stream_stage_trait_t<Fun>>
[[deprecated("use attach_continuous_stream_stage instead")]] auto
make_continuous_stage(Init init, Fun fun, Cleanup cleanup,
policy::arg<DownstreamManager> token = {}) {
CAF_IGNORE_UNUSED(token);
using input_type = typename Trait::input;
using output_type = typename Trait::output;
using state_type = typename Trait::state;
static_assert(
std::is_same<void(state_type&),
typename detail::get_callable_trait<Init>::fun_sig>::value,
"Expected signature `void (State&)` for init function");
static_assert(
std::is_same<void(state_type&, downstream<output_type>&, input_type),
typename detail::get_callable_trait<Fun>::fun_sig>::value,
"Expected signature `void (State&, downstream<Out>&, In)` "
"for consume function");
using driver
= detail::stream_stage_driver_impl<typename Trait::input,
DownstreamManager, Fun, Cleanup>;
return make_continuous_stage<driver>(std::move(init), std::move(fun),
std::move(cleanup));
}
/// @cond PRIVATE /// @cond PRIVATE
// -- timeout management ----------------------------------------------------- // -- timeout management -----------------------------------------------------
...@@ -653,7 +441,7 @@ public: ...@@ -653,7 +441,7 @@ public:
message_category categorize(mailbox_element& x); message_category categorize(mailbox_element& x);
/// Tries to consume `x`. /// Tries to consume `x`.
virtual invoke_message_result consume(mailbox_element& x); invoke_message_result consume(mailbox_element& x);
/// Tries to consume `x`. /// Tries to consume `x`.
void consume(mailbox_element_ptr x); void consume(mailbox_element_ptr x);
......
...@@ -28,7 +28,6 @@ ...@@ -28,7 +28,6 @@
#include "caf/policy/work_sharing.hpp" #include "caf/policy/work_sharing.hpp"
#include "caf/policy/work_stealing.hpp" #include "caf/policy/work_stealing.hpp"
#include "caf/raise_error.hpp" #include "caf/raise_error.hpp"
#include "caf/raw_event_based_actor.hpp"
#include "caf/scheduler/abstract_coordinator.hpp" #include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/scheduler/coordinator.hpp" #include "caf/scheduler/coordinator.hpp"
#include "caf/scheduler/test_coordinator.hpp" #include "caf/scheduler/test_coordinator.hpp"
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/raw_event_based_actor.hpp"
#include "caf/detail/default_invoke_result_visitor.hpp"
namespace caf {
raw_event_based_actor::raw_event_based_actor(actor_config& cfg)
: event_based_actor(cfg) {
// nop
}
invoke_message_result raw_event_based_actor::consume(mailbox_element& x) {
CAF_LOG_TRACE(CAF_ARG(x));
current_element_ = &x;
CAF_LOG_RECEIVE_EVENT(current_element_);
CAF_BEFORE_PROCESSING(this, x);
// Wrap the actual body for the function.
auto body = [this, &x] {
// short-circuit awaited responses
if (!awaited_responses_.empty()) {
auto& pr = awaited_responses_.front();
// skip all messages until we receive the currently awaited response
if (x.mid != pr.first)
return invoke_message_result::skipped;
if (!pr.second(x.content())) {
// try again with error if first attempt failed
auto msg = make_message(
make_error(sec::unexpected_response, std::move(x.payload)));
pr.second(msg);
}
awaited_responses_.pop_front();
return invoke_message_result::consumed;
}
// handle multiplexed responses
if (x.mid.is_response()) {
auto mrh = multiplexed_responses_.find(x.mid);
// neither awaited nor multiplexed, probably an expired timeout
if (mrh == multiplexed_responses_.end())
return invoke_message_result::dropped;
if (!mrh->second(x.content())) {
// try again with error if first attempt failed
auto msg = make_message(
make_error(sec::unexpected_response, std::move(x.payload)));
mrh->second(msg);
}
multiplexed_responses_.erase(mrh);
return invoke_message_result::consumed;
}
auto& content = x.content();
// handle timeout messages
if (content.match_elements<timeout_msg>()) {
auto& tm = content.get_as<timeout_msg>(0);
auto tid = tm.timeout_id;
CAF_ASSERT(x.mid.is_async());
if (is_active_receive_timeout(tid)) {
CAF_LOG_DEBUG("handle timeout message");
if (bhvr_stack_.empty())
return invoke_message_result::dropped;
bhvr_stack_.back().handle_timeout();
return invoke_message_result::consumed;
}
CAF_LOG_DEBUG("dropped expired timeout message");
return invoke_message_result::dropped;
}
// handle everything else as ordinary message
detail::default_invoke_result_visitor<event_based_actor> visitor{this};
auto had_timeout = getf(has_timeout_flag);
if (had_timeout)
unsetf(has_timeout_flag);
if (!bhvr_stack_.empty()) {
auto& bhvr = bhvr_stack_.back();
if (bhvr(visitor, x.content()))
return invoke_message_result::consumed;
}
auto sres = call_handler(default_handler_, this, x.payload);
auto f = detail::make_overload(
[&](auto& x) {
visitor(x);
return invoke_message_result::consumed;
},
[&](skip_t& x) {
// Restore timeout if message was skipped.
if (had_timeout)
setf(has_timeout_flag);
return invoke_message_result::skipped;
});
return visit(f, sres);
};
// Post-process the returned value from the function body.
auto result = body();
CAF_AFTER_PROCESSING(this, result);
CAF_LOG_SKIP_OR_FINALIZE_EVENT(result);
return result;
}
} // namespace caf
...@@ -580,10 +580,11 @@ scheduled_actor::categorize(mailbox_element& x) { ...@@ -580,10 +580,11 @@ scheduled_actor::categorize(mailbox_element& x) {
CAF_LOG_DEBUG("handle ordinary timeout message"); CAF_LOG_DEBUG("handle ordinary timeout message");
if (is_active_receive_timeout(tid) && !bhvr_stack_.empty()) if (is_active_receive_timeout(tid) && !bhvr_stack_.empty())
bhvr_stack_.back().handle_timeout(); bhvr_stack_.back().handle_timeout();
} else { } else if (tm.type == "stream") {
CAF_ASSERT(tm.type == "stream");
CAF_LOG_DEBUG("handle stream timeout message"); CAF_LOG_DEBUG("handle stream timeout message");
set_stream_timeout(advance_streams(clock().now())); set_stream_timeout(advance_streams(clock().now()));
} else {
// Drop. Other types not supported yet.
} }
return message_category::internal; return message_category::internal;
} }
......
...@@ -27,7 +27,6 @@ ...@@ -27,7 +27,6 @@
#include "caf/all.hpp" #include "caf/all.hpp"
#include "caf/detail/test_actor_clock.hpp" #include "caf/detail/test_actor_clock.hpp"
#include "caf/raw_event_based_actor.hpp"
using namespace caf; using namespace caf;
...@@ -39,8 +38,10 @@ struct testee_state { ...@@ -39,8 +38,10 @@ struct testee_state {
uint64_t timeout_id = 41; uint64_t timeout_id = 41;
}; };
behavior testee(stateful_actor<testee_state, raw_event_based_actor>* self, behavior testee(stateful_actor<testee_state>* self,
detail::test_actor_clock* t) { detail::test_actor_clock* t) {
self->set_exit_handler([self](exit_msg& x) { self->quit(x.reason); });
self->set_error_handler([](scheduled_actor*, error&) {});
return { return {
[=](ok_atom) { [=](ok_atom) {
CAF_LOG_TRACE("" << self->current_mailbox_element()->content()); CAF_LOG_TRACE("" << self->current_mailbox_element()->content());
...@@ -61,17 +62,11 @@ behavior testee(stateful_actor<testee_state, raw_event_based_actor>* self, ...@@ -61,17 +62,11 @@ behavior testee(stateful_actor<testee_state, raw_event_based_actor>* self,
auto mid = make_message_id(self->state.timeout_id).response_id(); auto mid = make_message_id(self->state.timeout_id).response_id();
t->set_request_timeout(n, self, mid); t->set_request_timeout(n, self, mid);
}, },
[](const timeout_msg&) { CAF_LOG_TRACE(""); },
[](const error&) { CAF_LOG_TRACE(""); },
[](const std::string&) { CAF_LOG_TRACE(""); }, [](const std::string&) { CAF_LOG_TRACE(""); },
[=](group& grp) { [=](group& grp) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
self->join(grp); self->join(grp);
}, },
[=](exit_msg& x) {
CAF_LOG_TRACE("");
self->quit(x.reason);
},
}; };
} }
......
...@@ -49,7 +49,6 @@ ...@@ -49,7 +49,6 @@
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
#include "caf/raw_event_based_actor.hpp"
#include "caf/scheduler/abstract_coordinator.hpp" #include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/scoped_actor.hpp" #include "caf/scoped_actor.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
......
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