Commit 1754baa5 authored by Dominik Charousset's avatar Dominik Charousset

Remove inbound path management from local actor

parent 89f9b6f3
...@@ -53,7 +53,7 @@ public: ...@@ -53,7 +53,7 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
broadcast_scatterer(local_actor* selfptr) : super(selfptr) { broadcast_scatterer(scheduled_actor* selfptr) : super(selfptr) {
// nop // nop
} }
......
...@@ -46,7 +46,7 @@ public: ...@@ -46,7 +46,7 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
buffered_scatterer(local_actor* self) : super(self) { buffered_scatterer(scheduled_actor* self) : super(self) {
// nop // nop
} }
......
...@@ -43,7 +43,7 @@ public: ...@@ -43,7 +43,7 @@ public:
using input_type = typename driver_type::input_type; using input_type = typename driver_type::input_type;
template <class... Ts> template <class... Ts>
stream_sink_impl(local_actor* self, Ts&&... xs) stream_sink_impl(scheduled_actor* self, Ts&&... xs)
: stream_manager(self), : stream_manager(self),
driver_(std::forward<Ts>(xs)...), driver_(std::forward<Ts>(xs)...),
out_(self) { out_(self) {
...@@ -84,7 +84,7 @@ private: ...@@ -84,7 +84,7 @@ private:
}; };
template <class Driver, class... Ts> template <class Driver, class... Ts>
stream_manager_ptr make_stream_sink(local_actor* self, Ts&&... xs) { stream_manager_ptr make_stream_sink(scheduled_actor* self, Ts&&... xs) {
using impl = stream_sink_impl<Driver>; using impl = stream_sink_impl<Driver>;
return make_counted<impl>(self, std::forward<Ts>(xs)...); return make_counted<impl>(self, std::forward<Ts>(xs)...);
} }
......
...@@ -43,7 +43,7 @@ public: ...@@ -43,7 +43,7 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
template <class... Ts> template <class... Ts>
stream_source_impl(local_actor* self, Ts&&... xs) stream_source_impl(scheduled_actor* self, Ts&&... xs)
: stream_manager(self), : stream_manager(self),
super(self), super(self),
at_end_(false), at_end_(false),
...@@ -87,7 +87,7 @@ private: ...@@ -87,7 +87,7 @@ private:
}; };
template <class Driver, class... Ts> template <class Driver, class... Ts>
typename Driver::source_ptr_type make_stream_source(local_actor* self, typename Driver::source_ptr_type make_stream_source(scheduled_actor* self,
Ts&&... xs) { Ts&&... xs) {
using impl = stream_source_impl<Driver>; using impl = stream_source_impl<Driver>;
return make_counted<impl>(self, std::forward<Ts>(xs)...); return make_counted<impl>(self, std::forward<Ts>(xs)...);
......
...@@ -49,7 +49,7 @@ public: ...@@ -49,7 +49,7 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
template <class... Ts> template <class... Ts>
stream_stage_impl(local_actor* self, Ts&&... xs) stream_stage_impl(scheduled_actor* self, Ts&&... xs)
: stream_manager(self), : stream_manager(self),
super(self), super(self),
driver_(std::forward<Ts>(xs)...) { driver_(std::forward<Ts>(xs)...) {
...@@ -93,7 +93,7 @@ private: ...@@ -93,7 +93,7 @@ private:
}; };
template <class Driver, class... Ts> template <class Driver, class... Ts>
typename Driver::stage_ptr_type make_stream_stage(local_actor* self, typename Driver::stage_ptr_type make_stream_stage(scheduled_actor* self,
Ts&&... xs) { Ts&&... xs) {
using impl = stream_stage_impl<Driver>; using impl = stream_stage_impl<Driver>;
return make_counted<impl>(self, std::forward<Ts>(xs)...); return make_counted<impl>(self, std::forward<Ts>(xs)...);
......
...@@ -116,7 +116,7 @@ public: ...@@ -116,7 +116,7 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
fused_scatterer(local_actor* self) fused_scatterer(scheduled_actor* self)
: super(self), : super(self),
nested_(self, detail::pack_repeat<Ts...>(self)) { nested_(self, detail::pack_repeat<Ts...>(self)) {
detail::init_ptr_array<0, sizeof...(Ts) + 1>::apply(ptrs_, nested_); detail::init_ptr_array<0, sizeof...(Ts) + 1>::apply(ptrs_, nested_);
......
...@@ -26,7 +26,7 @@ namespace caf { ...@@ -26,7 +26,7 @@ namespace caf {
/// Type-erased policy for dispatching data to sinks. /// Type-erased policy for dispatching data to sinks.
class invalid_stream_scatterer : public stream_scatterer { class invalid_stream_scatterer : public stream_scatterer {
public: public:
invalid_stream_scatterer(local_actor* self); invalid_stream_scatterer(scheduled_actor* self);
~invalid_stream_scatterer() override; ~invalid_stream_scatterer() override;
......
...@@ -400,27 +400,6 @@ public: ...@@ -400,27 +400,6 @@ public:
message_id new_request_id(message_priority mp); message_id new_request_id(message_priority mp);
/// Creates a new path for incoming stream traffic from `sender`.
virtual inbound_path* make_inbound_path(stream_manager_ptr mgr,
stream_slots slots,
strong_actor_ptr sender);
/// Silently closes incoming stream traffic on `slot`.
virtual void erase_inbound_path_later(stream_slot slot);
/// Closes incoming stream traffic on `slot`. Emits a drop message on the
/// path if `reason == none` and a `forced_drop` message otherwise.
virtual void erase_inbound_path_later(stream_slot slot, error reason);
/// Silently closes all inbound paths for `mgr`.
virtual void erase_inbound_paths_later(const stream_manager* mgr);
/// Closes all incoming stream traffic for a manager. Emits a drop message on
/// each path if `reason == none` and a `forced_drop` message on each path
/// otherwise.
virtual void erase_inbound_paths_later(const stream_manager* mgr,
error reason);
protected: protected:
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
......
...@@ -732,17 +732,26 @@ public: ...@@ -732,17 +732,26 @@ public:
// -- inbound_path management ------------------------------------------------ // -- inbound_path management ------------------------------------------------
inbound_path* make_inbound_path(stream_manager_ptr mgr, stream_slots slots, /// Creates a new path for incoming stream traffic from `sender`.
strong_actor_ptr sender) override; virtual inbound_path* make_inbound_path(stream_manager_ptr mgr,
stream_slots slots,
void erase_inbound_path_later(stream_slot slot) override; strong_actor_ptr sender);
void erase_inbound_path_later(stream_slot slot, error reason) override; /// Silently closes incoming stream traffic on `slot`.
virtual void erase_inbound_path_later(stream_slot slot);
void erase_inbound_paths_later(const stream_manager* mgr) override;
/// Closes incoming stream traffic on `slot`. Emits a drop message on the
void erase_inbound_paths_later(const stream_manager* mgr, /// path if `reason == none` and a `forced_drop` message otherwise.
error reason) override; virtual void erase_inbound_path_later(stream_slot slot, error reason);
/// Silently closes all inbound paths for `mgr`.
virtual void erase_inbound_paths_later(const stream_manager* mgr);
/// Closes all incoming stream traffic for a manager. Emits a drop message on
/// each path if `reason == none` and a `forced_drop` message on each path
/// otherwise.
virtual void erase_inbound_paths_later(const stream_manager* mgr,
error reason);
// -- handling of stream message --------------------------------------------- // -- handling of stream message ---------------------------------------------
......
...@@ -40,7 +40,7 @@ public: ...@@ -40,7 +40,7 @@ public:
using inbound_paths_list = std::vector<inbound_path*>; using inbound_paths_list = std::vector<inbound_path*>;
stream_manager(local_actor* selfptr, stream_manager(scheduled_actor* selfptr,
stream_priority prio = stream_priority::normal); stream_priority prio = stream_priority::normal);
~stream_manager() override; ~stream_manager() override;
...@@ -171,7 +171,7 @@ public: ...@@ -171,7 +171,7 @@ public:
} }
/// Returns the parent actor. /// Returns the parent actor.
inline local_actor* self() { inline scheduled_actor* self() {
return self_; return self_;
} }
...@@ -209,7 +209,7 @@ protected: ...@@ -209,7 +209,7 @@ protected:
virtual void output_closed(error reason); virtual void output_closed(error reason);
/// Points to the parent actor. /// Points to the parent actor.
local_actor* self_; scheduled_actor* self_;
/// Stores non-owning pointers to all input paths. /// Stores non-owning pointers to all input paths.
inbound_paths_list inbound_paths_; inbound_paths_list inbound_paths_;
......
...@@ -60,13 +60,13 @@ public: ...@@ -60,13 +60,13 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
explicit stream_scatterer(local_actor* self); explicit stream_scatterer(scheduled_actor* self);
virtual ~stream_scatterer(); virtual ~stream_scatterer();
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
local_actor* self() const { scheduled_actor* self() const {
return self_; return self_;
} }
...@@ -208,7 +208,7 @@ protected: ...@@ -208,7 +208,7 @@ protected:
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
local_actor* self_; scheduled_actor* self_;
}; };
} // namespace caf } // namespace caf
......
...@@ -41,7 +41,7 @@ public: ...@@ -41,7 +41,7 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
explicit stream_scatterer_impl(local_actor* self); explicit stream_scatterer_impl(scheduled_actor* self);
virtual ~stream_scatterer_impl(); virtual ~stream_scatterer_impl();
......
...@@ -37,7 +37,7 @@ public: ...@@ -37,7 +37,7 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
stream_sink(local_actor* self) : stream_manager(self) { stream_sink(scheduled_actor* self) : stream_manager(self) {
// nop // nop
} }
}; };
......
...@@ -37,7 +37,7 @@ public: ...@@ -37,7 +37,7 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
stream_source(local_actor* self) : stream_manager(self), out_(self) { stream_source(scheduled_actor* self) : stream_manager(self), out_(self) {
// nop // nop
} }
......
...@@ -40,7 +40,7 @@ public: ...@@ -40,7 +40,7 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
stream_stage(local_actor* self) stream_stage(scheduled_actor* self)
: stream_manager(self), : stream_manager(self),
left_super(self), left_super(self),
right_super(self) { right_super(self) {
......
...@@ -29,7 +29,7 @@ class terminal_stream_scatterer : public invalid_stream_scatterer { ...@@ -29,7 +29,7 @@ class terminal_stream_scatterer : public invalid_stream_scatterer {
public: public:
using super = invalid_stream_scatterer; using super = invalid_stream_scatterer;
terminal_stream_scatterer(local_actor* self); terminal_stream_scatterer(scheduled_actor* self);
~terminal_stream_scatterer() override; ~terminal_stream_scatterer() override;
......
...@@ -21,7 +21,7 @@ ...@@ -21,7 +21,7 @@
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/no_stages.hpp" #include "caf/no_stages.hpp"
#include "caf/local_actor.hpp" #include "caf/scheduled_actor.hpp"
namespace caf { namespace caf {
......
...@@ -24,7 +24,7 @@ ...@@ -24,7 +24,7 @@
namespace caf { namespace caf {
invalid_stream_scatterer::invalid_stream_scatterer(local_actor* self) invalid_stream_scatterer::invalid_stream_scatterer(scheduled_actor* self)
: stream_scatterer(self) { : stream_scatterer(self) {
// nop // nop
} }
......
...@@ -125,25 +125,4 @@ bool local_actor::cleanup(error&& fail_state, execution_unit* host) { ...@@ -125,25 +125,4 @@ bool local_actor::cleanup(error&& fail_state, execution_unit* host) {
return true; return true;
} }
inbound_path* local_actor::make_inbound_path(stream_manager_ptr, stream_slots,
strong_actor_ptr) {
return nullptr;
}
void local_actor::erase_inbound_path_later(stream_slot) {
CAF_LOG_ERROR("local_actor::erase_inbound_path_later called");
}
void local_actor::erase_inbound_path_later(stream_slot, error) {
CAF_LOG_ERROR("local_actor::erase_inbound_path_later called");
}
void local_actor::erase_inbound_paths_later(const stream_manager*) {
CAF_LOG_ERROR("local_actor::erase_inbound_paths_later called");
}
void local_actor::erase_inbound_paths_later(const stream_manager*, error) {
CAF_LOG_ERROR("local_actor::erase_inbound_paths_later called");
}
} // namespace caf } // namespace caf
...@@ -24,7 +24,7 @@ ...@@ -24,7 +24,7 @@
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/inbound_path.hpp" #include "caf/inbound_path.hpp"
#include "caf/local_actor.hpp" #include "caf/scheduled_actor.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/outbound_path.hpp" #include "caf/outbound_path.hpp"
...@@ -34,7 +34,7 @@ ...@@ -34,7 +34,7 @@
namespace caf { namespace caf {
stream_manager::stream_manager(local_actor* selfptr, stream_priority prio) stream_manager::stream_manager(scheduled_actor* selfptr, stream_priority prio)
: self_(selfptr), : self_(selfptr),
pending_handshakes_(0), pending_handshakes_(0),
priority_(prio), priority_(prio),
......
...@@ -22,7 +22,7 @@ ...@@ -22,7 +22,7 @@
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/local_actor.hpp" #include "caf/scheduled_actor.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/outbound_path.hpp" #include "caf/outbound_path.hpp"
...@@ -36,7 +36,7 @@ stream_scatterer::stream_scatterer::path_predicate::~path_predicate() { ...@@ -36,7 +36,7 @@ stream_scatterer::stream_scatterer::path_predicate::~path_predicate() {
// nop // nop
} }
stream_scatterer::stream_scatterer(local_actor* self) : self_(self) { stream_scatterer::stream_scatterer(scheduled_actor* self) : self_(self) {
// nop // nop
} }
......
...@@ -25,7 +25,8 @@ ...@@ -25,7 +25,8 @@
namespace caf { namespace caf {
stream_scatterer_impl::stream_scatterer_impl(local_actor* self) : super(self) { stream_scatterer_impl::stream_scatterer_impl(scheduled_actor* self)
: super(self) {
// nop // nop
} }
......
...@@ -22,7 +22,7 @@ ...@@ -22,7 +22,7 @@
namespace caf { namespace caf {
terminal_stream_scatterer::terminal_stream_scatterer(local_actor* self) terminal_stream_scatterer::terminal_stream_scatterer(scheduled_actor* self)
: super(self) { : super(self) {
// nop // nop
} }
......
...@@ -24,7 +24,7 @@ ...@@ -24,7 +24,7 @@
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/broadcast_scatterer.hpp" #include "caf/broadcast_scatterer.hpp"
#include "caf/local_actor.hpp" #include "caf/scheduled_actor.hpp"
#include "caf/mixin/sender.hpp" #include "caf/mixin/sender.hpp"
...@@ -35,11 +35,11 @@ using namespace caf; ...@@ -35,11 +35,11 @@ using namespace caf;
namespace { namespace {
// Mocks just enough of an actor to receive and send batches. // Mocks just enough of an actor to receive and send batches.
class entity : public extend<local_actor, entity>::with<mixin::sender> { class entity : public scheduled_actor {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using super = extend<local_actor, entity>::with<mixin::sender>; using super = scheduled_actor;
using signatures = none_t; using signatures = none_t;
......
...@@ -170,7 +170,7 @@ using scatterer = fused_scatterer<int_scatterer, string_scatterer>; ...@@ -170,7 +170,7 @@ using scatterer = fused_scatterer<int_scatterer, string_scatterer>;
class fused_stage : public stream_manager { class fused_stage : public stream_manager {
public: public:
fused_stage(local_actor* self) : stream_manager(self), out_(self) { fused_stage(scheduled_actor* self) : stream_manager(self), out_(self) {
continuous(true); continuous(true);
} }
......
...@@ -149,12 +149,12 @@ using mboxqueue = ...@@ -149,12 +149,12 @@ using mboxqueue =
// -- entity and mailbox visitor ----------------------------------------------- // -- entity and mailbox visitor -----------------------------------------------
class entity : public extend<local_actor, entity>::with<mixin::sender> { class entity : public scheduled_actor {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
/// Base type. /// Base type.
using super = extend<local_actor, entity>::with<mixin::sender>; using super = scheduled_actor;
/// Defines the messaging interface. /// Defines the messaging interface.
using signatures = none_t; using signatures = none_t;
......
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