Commit 4500d0ad authored by Dominik Charousset's avatar Dominik Charousset

Allow brokers to trigger events, close #358

parent e23320b7
...@@ -103,6 +103,34 @@ public: ...@@ -103,6 +103,34 @@ public:
// -- modifiers -------------------------------------------------------------- // -- modifiers --------------------------------------------------------------
/// Suspends activities on `hdl` unconditionally.
template <class Handle>
void halt(Handle hdl) {
auto ref = by_id(hdl);
if (ref)
ref->halt();
}
/// Allows activities on `hdl` unconditionally (default).
template <class Handle>
void trigger(Handle hdl) {
auto ref = by_id(hdl);
if (ref)
ref->trigger();
}
/// Allows `num_events` activities on `hdl`.
template <class Handle>
void trigger(Handle hdl, size_t num_events) {
if (num_events > 0) {
auto ref = by_id(hdl);
if (ref)
ref->trigger(num_events);
return;
}
halt(hdl);
}
/// Modifies the receive policy for given connection. /// Modifies the receive policy for given connection.
/// @param hdl Identifies the affected connection. /// @param hdl Identifies the affected connection.
/// @param config Contains the new receive policy. /// @param config Contains the new receive policy.
......
...@@ -46,12 +46,31 @@ public: ...@@ -46,12 +46,31 @@ public:
return hdl_; return hdl_;
} }
void halt() {
activity_tokens_ = none;
this->remove_from_loop();
}
void trigger() {
activity_tokens_ = none;
this->add_to_loop();
}
void trigger(size_t num) {
CAF_ASSERT(num > 0);
if (activity_tokens_)
*activity_tokens_ += num;
else
activity_tokens_ = num;
this->add_to_loop();
}
protected: protected:
void detach_from(abstract_broker* ptr) override { void detach_from(abstract_broker* ptr) override {
ptr->erase(hdl_); ptr->erase(hdl_);
} }
void invoke_mailbox_element(execution_unit* ctx) { void invoke_mailbox_element_impl(execution_unit* ctx, mailbox_element& x) {
auto self = this->parent(); auto self = this->parent();
auto pfac = self->proxy_registry_ptr(); auto pfac = self->proxy_registry_ptr();
if (pfac) if (pfac)
...@@ -60,7 +79,29 @@ protected: ...@@ -60,7 +79,29 @@ protected:
if (pfac) if (pfac)
ctx->proxy_registry_ptr(nullptr); ctx->proxy_registry_ptr(nullptr);
}); });
self->activate(ctx, value_); self->activate(ctx, x);
}
bool invoke_mailbox_element(execution_unit* ctx) {
auto prev = activity_tokens_;
invoke_mailbox_element_impl(ctx, value_);
// only consume an activity token if actor did not produce them now
if (prev && activity_tokens_ && --(*activity_tokens_) == 0) {
// tell broker it entered passive mode, this can result in
// producing, why we check the condition again afterwards
using passiv_t =
typename std::conditional<
std::is_same<Handle, connection_handle>::value,
connection_passivated_msg,
acceptor_passivated_msg
>::type;
using tmp_t = mailbox_element_vals<passiv_t>;
tmp_t tmp{strong_actor_ptr{}, message_id::make(),
mailbox_element::forwarding_stack{}, passiv_t{hdl()}};
invoke_mailbox_element_impl(ctx, tmp);
return activity_tokens_ != size_t{0};
}
return true;
} }
SysMsgType& msg() { SysMsgType& msg() {
...@@ -69,6 +110,7 @@ protected: ...@@ -69,6 +110,7 @@ protected:
Handle hdl_; Handle hdl_;
mailbox_element_vals<SysMsgType> value_; mailbox_element_vals<SysMsgType> value_;
optional<size_t> activity_tokens_;
}; };
} // namespace io } // namespace io
......
...@@ -46,6 +46,10 @@ public: ...@@ -46,6 +46,10 @@ public:
void io_failure(execution_unit* ctx, network::operation op) override; void io_failure(execution_unit* ctx, network::operation op) override;
using doorman_base::new_connection;
bool new_connection(execution_unit* ctx, connection_handle hdl);
// needs to be launched explicitly // needs to be launched explicitly
virtual void launch() = 0; virtual void launch() = 0;
......
...@@ -219,18 +219,29 @@ public: ...@@ -219,18 +219,29 @@ public:
std::forward<Ts>(xs)...); std::forward<Ts>(xs)...);
} }
/// Spawns a new broker as server running on given `port` /// Spawns a new broker as server running on given `port`.
/// or an `error`.
/// @warning Blocks the caller until the server socket is initialized. /// @warning Blocks the caller until the server socket is initialized.
template <spawn_options Os = no_spawn_options, template <spawn_options Os = no_spawn_options,
class F = std::function<void(broker*)>, class... Ts> class F = std::function<void(broker*)>, class... Ts>
expected<typename infer_handle_from_fun<F>::type> expected<typename infer_handle_from_fun<F>::type>
spawn_server(F fun, uint16_t port, Ts&&... xs) { spawn_server(F fun, uint16_t& port, Ts&&... xs) {
using impl = typename infer_handle_from_fun<F>::impl; using impl = typename infer_handle_from_fun<F>::impl;
return spawn_server_impl<Os, impl>(std::move(fun), port, return spawn_server_impl<Os, impl>(std::move(fun), port,
std::forward<Ts>(xs)...); std::forward<Ts>(xs)...);
} }
/// Spawns a new broker as server running on given `port`.
/// @warning Blocks the caller until the server socket is initialized.
template <spawn_options Os = no_spawn_options,
class F = std::function<void(broker*)>, class... Ts>
expected<typename infer_handle_from_fun<F>::type>
spawn_server(F fun, const uint16_t& port, Ts&&... xs) {
uint16_t dummy = port;
using impl = typename infer_handle_from_fun<F>::impl;
return spawn_server_impl<Os, impl>(std::move(fun), dummy,
std::forward<Ts>(xs)...);
}
/// Returns a middleman using the default network backend. /// Returns a middleman using the default network backend.
static actor_system::module* make(actor_system&, detail::type_list<>); static actor_system::module* make(actor_system&, detail::type_list<>);
...@@ -276,13 +287,14 @@ private: ...@@ -276,13 +287,14 @@ private:
template <spawn_options Os, class Impl, class F, class... Ts> template <spawn_options Os, class Impl, class F, class... Ts>
expected<typename infer_handle_from_class<Impl>::type> expected<typename infer_handle_from_class<Impl>::type>
spawn_server_impl(F fun, uint16_t port, Ts&&... xs) { spawn_server_impl(F fun, uint16_t& port, Ts&&... xs) {
detail::init_fun_factory<Impl, F> fac; detail::init_fun_factory<Impl, F> fac;
auto init_fun = fac(std::move(fun), std::forward<Ts>(xs)...); auto init_fun = fac(std::move(fun), std::forward<Ts>(xs)...);
auto ehdl = backend().new_tcp_doorman(port); auto ehdl = backend().new_tcp_doorman(port);
if (!ehdl) if (!ehdl)
return ehdl.error(); return ehdl.error();
auto hdl = ehdl->first; auto hdl = ehdl->first;
port = ehdl->second;
actor_config cfg{&backend()}; actor_config cfg{&backend()};
cfg.init_fun = [hdl, init_fun](local_actor* ptr) -> behavior { cfg.init_fun = [hdl, init_fun](local_actor* ptr) -> behavior {
static_cast<abstract_broker*>(ptr)->assign_tcp_doorman(hdl); static_cast<abstract_broker*>(ptr)->assign_tcp_doorman(hdl);
......
...@@ -34,9 +34,11 @@ public: ...@@ -34,9 +34,11 @@ public:
~acceptor_manager(); ~acceptor_manager();
/// Called by the underlying IO device to indicate that /// Called by the underlying I/O device to indicate that
/// a new connection is awaiting acceptance. /// a new connection is awaiting acceptance.
virtual void new_connection() = 0; /// @returns `true` if the manager accepts further connections,
/// otherwise `false`.
virtual bool new_connection() = 0;
}; };
} // namespace network } // namespace network
......
...@@ -240,12 +240,12 @@ asio_multiplexer::add_tcp_doorman(abstract_broker* self, ...@@ -240,12 +240,12 @@ asio_multiplexer::add_tcp_doorman(abstract_broker* self,
acceptor_(am, s.get_io_service()) { acceptor_(am, s.get_io_service()) {
acceptor_.init(std::move(s)); acceptor_.init(std::move(s));
} }
void new_connection() override { bool new_connection() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
auto& am = acceptor_.backend(); auto& am = acceptor_.backend();
msg().handle auto hdl = am.add_tcp_scribe(parent(),
= am.add_tcp_scribe(parent(), std::move(acceptor_.accepted_socket())); std::move(acceptor_.accepted_socket()));
invoke_mailbox_element(&am); return doorman::new_connection(&am, hdl);
} }
void stop_reading() override { void stop_reading() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
......
...@@ -222,7 +222,13 @@ public: ...@@ -222,7 +222,13 @@ public:
/// Closes the read channel of the underlying socket. /// Closes the read channel of the underlying socket.
void close_read_channel(); void close_read_channel();
/// Removes the file descriptor from the event loop of the parent.
void passivate();
protected: protected:
/// Adds the file descriptor to the event loop of the parent.
void activate();
void set_fd_flags(); void set_fd_flags();
int eventbf_; int eventbf_;
...@@ -385,7 +391,10 @@ public: ...@@ -385,7 +391,10 @@ public:
stream(default_multiplexer& backend_ref, native_socket sockfd); stream(default_multiplexer& backend_ref, native_socket sockfd);
/// Starts reading data from the socket, forwarding incoming data to `mgr`. /// Starts reading data from the socket, forwarding incoming data to `mgr`.
void start(const manager_ptr& mgr); void start(stream_manager* mgr);
/// Activates the stream.
void activate(stream_manager* mgr);
/// Configures how much data will be provided for the next `consume` callback. /// Configures how much data will be provided for the next `consume` callback.
/// @warning Must not be called outside the IO multiplexers event loop /// @warning Must not be called outside the IO multiplexers event loop
...@@ -468,7 +477,10 @@ public: ...@@ -468,7 +477,10 @@ public:
/// Starts this acceptor, forwarding all incoming connections to /// Starts this acceptor, forwarding all incoming connections to
/// `manager`. The intrusive pointer will be released after the /// `manager`. The intrusive pointer will be released after the
/// acceptor has been closed or an IO error occured. /// acceptor has been closed or an IO error occured.
void start(const manager_ptr& mgr); void start(acceptor_manager* mgr);
/// Activates the acceptor.
void activate(acceptor_manager* mgr);
/// Closes the network connection and removes this handler from its parent. /// Closes the network connection and removes this handler from its parent.
void stop_reading(); void stop_reading();
......
...@@ -31,8 +31,8 @@ namespace caf { ...@@ -31,8 +31,8 @@ namespace caf {
namespace io { namespace io {
namespace network { namespace network {
/// A manager configures an IO device and provides callbacks /// A manager configures an I/O device and provides callbacks
/// for various IO operations. /// for various I/O operations.
class manager : public ref_counted { class manager : public ref_counted {
public: public:
manager(abstract_broker* parent_ptr); manager(abstract_broker* parent_ptr);
...@@ -55,17 +55,23 @@ public: ...@@ -55,17 +55,23 @@ public:
/// if `invoke_detach_message == true`. /// if `invoke_detach_message == true`.
void detach(execution_unit* ctx, bool invoke_detach_message); void detach(execution_unit* ctx, bool invoke_detach_message);
/// Causes the manager to stop read operations on its IO device. /// Causes the manager to stop read operations on its I/O device.
/// Unwritten bytes are still send before the socket will be closed. /// Unwritten bytes are still send before the socket will be closed.
virtual void stop_reading() = 0; virtual void stop_reading() = 0;
/// Called by the underlying IO device to report failures. /// Removes the I/O device to the event loop of the middleman.
virtual void remove_from_loop() = 0;
/// Adds the I/O device to the event loop of the middleman.
virtual void add_to_loop() = 0;
/// Called by the underlying I/O device to report failures.
virtual void io_failure(execution_unit* ctx, operation op) = 0; virtual void io_failure(execution_unit* ctx, operation op) = 0;
/// Get the address of the underlying IO device. /// Get the address of the underlying I/O device.
virtual std::string addr() const = 0; virtual std::string addr() const = 0;
/// Get the port of the underlying IO device. /// Get the port of the underlying I/O device.
virtual uint16_t port() const = 0; virtual uint16_t port() const = 0;
protected: protected:
......
...@@ -37,7 +37,8 @@ public: ...@@ -37,7 +37,8 @@ public:
~stream_manager(); ~stream_manager();
/// Called by the underlying I/O device whenever it received data. /// Called by the underlying I/O device whenever it received data.
virtual void consume(execution_unit* ctx, const void* buf, size_t bsize) = 0; /// @returns `true` if the manager accepts further reads, otherwise `false`.
virtual bool consume(execution_unit* ctx, const void* buf, size_t bsize) = 0;
/// Called by the underlying I/O device whenever it sent data. /// Called by the underlying I/O device whenever it sent data.
virtual void data_transferred(execution_unit* ctx, size_t num_bytes, virtual void data_transferred(execution_unit* ctx, size_t num_bytes,
......
...@@ -90,6 +90,9 @@ public: ...@@ -90,6 +90,9 @@ public:
/// for reading, `false` otherwise. /// for reading, `false` otherwise.
bool& stopped_reading(connection_handle hdl); bool& stopped_reading(connection_handle hdl);
/// Returns `true` if this handle is inactive, otherwise `false`.
bool& passive_mode(connection_handle hdl);
intrusive_ptr<scribe>& impl_ptr(connection_handle hdl); intrusive_ptr<scribe>& impl_ptr(connection_handle hdl);
uint16_t& port(accept_handle hdl); uint16_t& port(accept_handle hdl);
...@@ -98,6 +101,9 @@ public: ...@@ -98,6 +101,9 @@ public:
/// for reading, `false` otherwise. /// for reading, `false` otherwise.
bool& stopped_reading(accept_handle hdl); bool& stopped_reading(accept_handle hdl);
/// Returns `true` if this handle is inactive, otherwise `false`.
bool& passive_mode(accept_handle hdl);
intrusive_ptr<doorman>& impl_ptr(accept_handle hdl); intrusive_ptr<doorman>& impl_ptr(accept_handle hdl);
/// Stores `hdl` as a pending connection for `src`. /// Stores `hdl` as a pending connection for `src`.
...@@ -149,6 +155,7 @@ private: ...@@ -149,6 +155,7 @@ private:
buffer_type wr_buf; buffer_type wr_buf;
receive_policy::config recv_conf; receive_policy::config recv_conf;
bool stopped_reading = false; bool stopped_reading = false;
bool passive_mode = false;
intrusive_ptr<scribe> ptr; intrusive_ptr<scribe> ptr;
bool ack_writes = false; bool ack_writes = false;
}; };
...@@ -156,6 +163,7 @@ private: ...@@ -156,6 +163,7 @@ private:
struct doorman_data { struct doorman_data {
uint16_t port; uint16_t port;
bool stopped_reading = false; bool stopped_reading = false;
bool passive_mode = false;
intrusive_ptr<doorman> ptr; intrusive_ptr<doorman> ptr;
}; };
......
...@@ -61,7 +61,7 @@ public: ...@@ -61,7 +61,7 @@ public:
void io_failure(execution_unit* ctx, network::operation op) override; void io_failure(execution_unit* ctx, network::operation op) override;
void consume(execution_unit*, const void*, size_t) override; bool consume(execution_unit*, const void*, size_t) override;
void data_transferred(execution_unit*, size_t, size_t) override; void data_transferred(execution_unit*, size_t, size_t) override;
......
...@@ -47,18 +47,6 @@ typename Inspector::result_type inspect(Inspector& f, new_connection_msg& x) { ...@@ -47,18 +47,6 @@ typename Inspector::result_type inspect(Inspector& f, new_connection_msg& x) {
return f(meta::type_name("new_connection_msg"), x.source, x.handle); return f(meta::type_name("new_connection_msg"), x.source, x.handle);
} }
/// @relates new_connection_msg
inline bool operator==(const new_connection_msg& lhs,
const new_connection_msg& rhs) {
return lhs.source == rhs.source && lhs.handle == rhs.handle;
}
/// @relates new_connection_msg
inline bool operator!=(const new_connection_msg& lhs,
const new_connection_msg& rhs) {
return !(lhs == rhs);
}
/// Signalizes newly arrived data for a {@link broker}. /// Signalizes newly arrived data for a {@link broker}.
struct new_data_msg { struct new_data_msg {
/// Handle to the related connection. /// Handle to the related connection.
...@@ -73,16 +61,6 @@ typename Inspector::result_type inspect(Inspector& f, new_data_msg& x) { ...@@ -73,16 +61,6 @@ typename Inspector::result_type inspect(Inspector& f, new_data_msg& x) {
return f(meta::type_name("new_data_msg"), x.handle, x.buf); return f(meta::type_name("new_data_msg"), x.handle, x.buf);
} }
/// @relates new_data_msg
inline bool operator==(const new_data_msg& lhs, const new_data_msg& rhs) {
return lhs.handle == rhs.handle && lhs.buf == rhs.buf;
}
/// @relates new_data_msg
inline bool operator!=(const new_data_msg& lhs, const new_data_msg& rhs) {
return !(lhs == rhs);
}
/// Signalizes that a certain amount of bytes has been written. /// Signalizes that a certain amount of bytes has been written.
struct data_transferred_msg { struct data_transferred_msg {
/// Handle to the related connection. /// Handle to the related connection.
...@@ -100,21 +78,6 @@ typename Inspector::result_type inspect(Inspector& f, data_transferred_msg& x) { ...@@ -100,21 +78,6 @@ typename Inspector::result_type inspect(Inspector& f, data_transferred_msg& x) {
x.handle, x.written, x.remaining); x.handle, x.written, x.remaining);
} }
/// @relates data_transferred_msg
inline bool operator==(const data_transferred_msg& x,
const data_transferred_msg& y) {
return x.handle == y.handle
&& x.written == y.written
&& x.remaining == y.remaining;
}
/// @relates data_transferred_msg
inline bool operator!=(const data_transferred_msg& x,
const data_transferred_msg& y) {
return !(x == y);
}
/// Signalizes that a {@link broker} connection has been closed. /// Signalizes that a {@link broker} connection has been closed.
struct connection_closed_msg { struct connection_closed_msg {
/// Handle to the closed connection. /// Handle to the closed connection.
...@@ -127,18 +90,6 @@ typename Inspector::result_type inspect(Inspector& f, connection_closed_msg& x) ...@@ -127,18 +90,6 @@ typename Inspector::result_type inspect(Inspector& f, connection_closed_msg& x)
return f(meta::type_name("connection_closed_msg"), x.handle); return f(meta::type_name("connection_closed_msg"), x.handle);
} }
/// @relates connection_closed_msg
inline bool operator==(const connection_closed_msg& lhs,
const connection_closed_msg& rhs) {
return lhs.handle == rhs.handle;
}
/// @relates connection_closed_msg
inline bool operator!=(const connection_closed_msg& lhs,
const connection_closed_msg& rhs) {
return !(lhs == rhs);
}
/// Signalizes that a {@link broker} acceptor has been closed. /// Signalizes that a {@link broker} acceptor has been closed.
struct acceptor_closed_msg { struct acceptor_closed_msg {
/// Handle to the closed connection. /// Handle to the closed connection.
...@@ -151,16 +102,28 @@ typename Inspector::result_type inspect(Inspector& f, acceptor_closed_msg& x) { ...@@ -151,16 +102,28 @@ typename Inspector::result_type inspect(Inspector& f, acceptor_closed_msg& x) {
return f(meta::type_name("acceptor_closed_msg"), x.handle); return f(meta::type_name("acceptor_closed_msg"), x.handle);
} }
/// @relates acceptor_closed_msg /// Signalizes that a connection has entered passive mode.
inline bool operator==(const acceptor_closed_msg& lhs, struct connection_passivated_msg {
const acceptor_closed_msg& rhs) { connection_handle handle;
return lhs.handle == rhs.handle; };
/// @relates connection_passivated_msg
template <class Inspector>
typename Inspector::result_type
inspect(Inspector& f, connection_passivated_msg& x) {
return f(meta::type_name("connection_passivated_msg"), x.handle);
} }
/// @relates acceptor_closed_msg /// Signalizes that an acceptor has entered passive mode.
inline bool operator!=(const acceptor_closed_msg& lhs, struct acceptor_passivated_msg {
const acceptor_closed_msg& rhs) { accept_handle handle;
return !(lhs == rhs); };
/// @relates acceptor_passivated_msg
template <class Inspector>
typename Inspector::result_type
inspect(Inspector& f, acceptor_passivated_msg& x) {
return f(meta::type_name("acceptor_passivated_msg"), x.handle);
} }
} // namespace io } // namespace io
......
...@@ -783,6 +783,12 @@ connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self, ...@@ -783,6 +783,12 @@ connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self,
launched_ = true; launched_ = true;
stream_.start(this); stream_.start(this);
} }
void add_to_loop() override {
stream_.activate(this);
}
void remove_from_loop() override {
stream_.passivate();
}
private: private:
bool launched_; bool launched_;
stream stream_; stream stream_;
...@@ -803,12 +809,12 @@ accept_handle default_multiplexer::add_tcp_doorman(abstract_broker* self, ...@@ -803,12 +809,12 @@ accept_handle default_multiplexer::add_tcp_doorman(abstract_broker* self,
acceptor_(mx, sockfd) { acceptor_(mx, sockfd) {
// nop // nop
} }
void new_connection() override { bool new_connection() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
auto& dm = acceptor_.backend(); auto& dm = acceptor_.backend();
msg().handle auto hdl = dm.add_tcp_scribe(parent(),
= dm.add_tcp_scribe(parent(), std::move(acceptor_.accepted_socket())); std::move(acceptor_.accepted_socket()));
invoke_mailbox_element(&acceptor_.backend()); return doorman::new_connection(&dm, hdl);
} }
void stop_reading() override { void stop_reading() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
...@@ -831,6 +837,12 @@ accept_handle default_multiplexer::add_tcp_doorman(abstract_broker* self, ...@@ -831,6 +837,12 @@ accept_handle default_multiplexer::add_tcp_doorman(abstract_broker* self,
return 0; return 0;
return *x; return *x;
} }
void add_to_loop() override {
acceptor_.activate(this);
}
void remove_from_loop() override {
acceptor_.passivate();
}
private: private:
network::acceptor acceptor_; network::acceptor acceptor_;
}; };
...@@ -976,6 +988,14 @@ void event_handler::close_read_channel() { ...@@ -976,6 +988,14 @@ void event_handler::close_read_channel() {
read_channel_closed_ = true; read_channel_closed_ = true;
} }
void event_handler::passivate() {
backend().del(operation::read, fd(), this);
}
void event_handler::activate() {
backend().add(operation::read, fd(), this);
}
void event_handler::set_fd_flags() { void event_handler::set_fd_flags() {
if (fd_ == invalid_native_socket) if (fd_ == invalid_native_socket)
return; return;
...@@ -1047,11 +1067,17 @@ stream::stream(default_multiplexer& backend_ref, native_socket sockfd) ...@@ -1047,11 +1067,17 @@ stream::stream(default_multiplexer& backend_ref, native_socket sockfd)
configure_read(receive_policy::at_most(1024)); configure_read(receive_policy::at_most(1024));
} }
void stream::start(const manager_ptr& mgr) { void stream::start(stream_manager* mgr) {
CAF_ASSERT(mgr != nullptr); CAF_ASSERT(mgr != nullptr);
reader_ = mgr; activate(mgr);
backend().add(operation::read, fd(), this); }
prepare_next_read();
void stream::activate(stream_manager* mgr) {
if (!reader_) {
reader_.reset(mgr);
event_handler::activate();
prepare_next_read();
}
} }
void stream::configure_read(receive_policy::config config) { void stream::configure_read(receive_policy::config config) {
...@@ -1084,7 +1110,7 @@ void stream::flush(const manager_ptr& mgr) { ...@@ -1084,7 +1110,7 @@ void stream::flush(const manager_ptr& mgr) {
void stream::stop_reading() { void stream::stop_reading() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
close_read_channel(); close_read_channel();
backend().del(operation::read, fd(), this); passivate();
} }
void stream::removed_from_loop(operation op) { void stream::removed_from_loop(operation op) {
...@@ -1108,15 +1134,19 @@ void stream::handle_event(operation op) { ...@@ -1108,15 +1134,19 @@ void stream::handle_event(operation op) {
rd_buf_.data() + collected_, rd_buf_.data() + collected_,
rd_buf_.size() - collected_)) { rd_buf_.size() - collected_)) {
reader_->io_failure(&backend(), operation::read); reader_->io_failure(&backend(), operation::read);
backend().del(operation::read, fd(), this); passivate();
return; return;
} }
if (rb == 0) if (rb == 0)
return; return;
collected_ += rb; collected_ += rb;
if (collected_ >= read_threshold_) { if (collected_ >= read_threshold_) {
reader_->consume(&backend(), rd_buf_.data(), collected_); auto res = reader_->consume(&backend(), rd_buf_.data(), collected_);
prepare_next_read(); prepare_next_read();
if (!res) {
passivate();
return;
}
} }
} }
break; break;
...@@ -1194,17 +1224,23 @@ acceptor::acceptor(default_multiplexer& backend_ref, native_socket sockfd) ...@@ -1194,17 +1224,23 @@ acceptor::acceptor(default_multiplexer& backend_ref, native_socket sockfd)
// nop // nop
} }
void acceptor::start(const manager_ptr& mgr) { void acceptor::start(acceptor_manager* mgr) {
CAF_LOG_TRACE(CAF_ARG(fd())); CAF_LOG_TRACE(CAF_ARG(fd()));
CAF_ASSERT(mgr != nullptr); CAF_ASSERT(mgr != nullptr);
mgr_ = mgr; activate(mgr);
backend().add(operation::read, fd(), this); }
void acceptor::activate(acceptor_manager* mgr) {
if (!mgr_) {
mgr_.reset(mgr);
event_handler::activate();
}
} }
void acceptor::stop_reading() { void acceptor::stop_reading() {
CAF_LOG_TRACE(CAF_ARG(fd())); CAF_LOG_TRACE(CAF_ARG(fd()));
close_read_channel(); close_read_channel();
backend().del(operation::read, fd(), this); passivate();
} }
void acceptor::handle_event(operation op) { void acceptor::handle_event(operation op) {
......
...@@ -46,5 +46,10 @@ void doorman::io_failure(execution_unit* ctx, network::operation op) { ...@@ -46,5 +46,10 @@ void doorman::io_failure(execution_unit* ctx, network::operation op) {
detach(ctx, true); detach(ctx, true);
} }
bool doorman::new_connection(execution_unit* ctx, connection_handle x) {
msg().handle = x;
return invoke_mailbox_element(ctx);
}
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -347,7 +347,9 @@ void middleman::init(actor_system_config& cfg) { ...@@ -347,7 +347,9 @@ void middleman::init(actor_system_config& cfg) {
.add_message_type<connection_closed_msg>("@connection_closed_msg") .add_message_type<connection_closed_msg>("@connection_closed_msg")
.add_message_type<connection_handle>("@connection_handle") .add_message_type<connection_handle>("@connection_handle")
.add_message_type<new_connection_msg>("@new_connection_msg") .add_message_type<new_connection_msg>("@new_connection_msg")
.add_message_type<new_data_msg>("@new_data_msg"); .add_message_type<new_data_msg>("@new_data_msg")
.add_message_type<connection_passivated_msg>("@connection_passivated_msg")
.add_message_type<acceptor_passivated_msg>("@acceptor_passivated_msg");
// compute and set ID for this network node // compute and set ID for this network node
node_id this_node{node_id::data::create_singleton()}; node_id this_node{node_id::data::create_singleton()};
system().node_.swap(this_node); system().node_.swap(this_node);
......
...@@ -37,16 +37,15 @@ message scribe::detach_message() { ...@@ -37,16 +37,15 @@ message scribe::detach_message() {
return make_message(connection_closed_msg{hdl()}); return make_message(connection_closed_msg{hdl()});
} }
void scribe::consume(execution_unit* ctx, const void*, size_t num_bytes) { bool scribe::consume(execution_unit* ctx, const void*, size_t num_bytes) {
CAF_ASSERT(ctx != nullptr); CAF_ASSERT(ctx != nullptr);
CAF_LOG_TRACE(CAF_ARG(num_bytes)); CAF_LOG_TRACE(CAF_ARG(num_bytes));
if (detached()) { if (detached())
// we are already disconnected from the broker while the multiplexer // we are already disconnected from the broker while the multiplexer
// did not yet remove the socket, this can happen if an IO event causes // did not yet remove the socket, this can happen if an I/O event causes
// the broker to call close_all() while the pollset contained // the broker to call close_all() while the pollset contained
// further activities for the broker // further activities for the broker
return; return false;
}
// keep a strong reference to our parent until we leave scope // keep a strong reference to our parent until we leave scope
// to avoid UB when becoming detached during invocation // to avoid UB when becoming detached during invocation
auto guard = parent_; auto guard = parent_;
...@@ -56,10 +55,11 @@ void scribe::consume(execution_unit* ctx, const void*, size_t num_bytes) { ...@@ -56,10 +55,11 @@ void scribe::consume(execution_unit* ctx, const void*, size_t num_bytes) {
buf.resize(num_bytes); buf.resize(num_bytes);
auto& msg_buf = msg().buf; auto& msg_buf = msg().buf;
msg_buf.swap(buf); msg_buf.swap(buf);
invoke_mailbox_element(ctx); auto result = invoke_mailbox_element(ctx);
// swap buffer back to stream and implicitly flush wr_buf() // swap buffer back to stream and implicitly flush wr_buf()
msg_buf.swap(buf); msg_buf.swap(buf);
flush(); flush();
return result;
} }
void scribe::data_transferred(execution_unit* ctx, size_t written, void scribe::data_transferred(execution_unit* ctx, size_t written,
...@@ -67,10 +67,16 @@ void scribe::data_transferred(execution_unit* ctx, size_t written, ...@@ -67,10 +67,16 @@ void scribe::data_transferred(execution_unit* ctx, size_t written,
CAF_LOG_TRACE(CAF_ARG(written) << CAF_ARG(remaining)); CAF_LOG_TRACE(CAF_ARG(written) << CAF_ARG(remaining));
if (detached()) if (detached())
return; return;
data_transferred_msg tmp{hdl(), written, remaining}; using transferred_t = data_transferred_msg;
auto ptr = make_mailbox_element(nullptr, invalid_message_id, {}, tmp); using tmp_t = mailbox_element_vals<data_transferred_msg>;
parent()->context(ctx); tmp_t tmp{strong_actor_ptr{}, message_id::make(),
parent()->consume(std::move(ptr)); mailbox_element::forwarding_stack{},
transferred_t{hdl(), written, remaining}};
invoke_mailbox_element_impl(ctx, tmp);
//data_transferred_msg tmp{hdl(), written, remaining};
//auto ptr = make_mailbox_element(nullptr, invalid_message_id, {}, tmp);
//parent()->context(ctx);
//parent()->consume(std::move(ptr));
} }
void scribe::io_failure(execution_unit* ctx, network::operation op) { void scribe::io_failure(execution_unit* ctx, network::operation op) {
......
...@@ -93,6 +93,14 @@ expected<void> test_multiplexer::assign_tcp_scribe(abstract_broker* ptr, ...@@ -93,6 +93,14 @@ expected<void> test_multiplexer::assign_tcp_scribe(abstract_broker* ptr,
return static_cast<uint16_t>(hdl().id()); return static_cast<uint16_t>(hdl().id());
} }
void add_to_loop() override {
mpx_->passive_mode(hdl()) = false;
}
void remove_from_loop() override {
mpx_->passive_mode(hdl()) = true;
}
private: private:
test_multiplexer* mpx_; test_multiplexer* mpx_;
}; };
...@@ -141,14 +149,15 @@ expected<void> test_multiplexer::assign_tcp_doorman(abstract_broker* ptr, ...@@ -141,14 +149,15 @@ expected<void> test_multiplexer::assign_tcp_doorman(abstract_broker* ptr,
// nop // nop
} }
void new_connection() override { bool new_connection() override {
auto& mm = mpx_->pending_connects(); auto& mm = mpx_->pending_connects();
auto i = mm.find(hdl()); auto i = mm.find(hdl());
bool result = true;
if (i != mm.end()) { if (i != mm.end()) {
msg().handle = i->second; result = doorman::new_connection(mpx_, i->second);
invoke_mailbox_element(mpx_);
mm.erase(i); mm.erase(i);
} }
return result;
} }
void stop_reading() override { void stop_reading() override {
...@@ -168,6 +177,14 @@ expected<void> test_multiplexer::assign_tcp_doorman(abstract_broker* ptr, ...@@ -168,6 +177,14 @@ expected<void> test_multiplexer::assign_tcp_doorman(abstract_broker* ptr,
return mpx_->port(hdl()); return mpx_->port(hdl());
} }
void add_to_loop() override {
mpx_->passive_mode(hdl()) = false;
}
void remove_from_loop() override {
mpx_->passive_mode(hdl()) = true;
}
private: private:
test_multiplexer* mpx_; test_multiplexer* mpx_;
}; };
...@@ -245,6 +262,10 @@ bool& test_multiplexer::stopped_reading(connection_handle hdl) { ...@@ -245,6 +262,10 @@ bool& test_multiplexer::stopped_reading(connection_handle hdl) {
return scribe_data_[hdl].stopped_reading; return scribe_data_[hdl].stopped_reading;
} }
bool& test_multiplexer::passive_mode(connection_handle hdl) {
return scribe_data_[hdl].passive_mode;
}
intrusive_ptr<scribe>& test_multiplexer::impl_ptr(connection_handle hdl) { intrusive_ptr<scribe>& test_multiplexer::impl_ptr(connection_handle hdl) {
return scribe_data_[hdl].ptr; return scribe_data_[hdl].ptr;
} }
...@@ -257,6 +278,10 @@ bool& test_multiplexer::stopped_reading(accept_handle hdl) { ...@@ -257,6 +278,10 @@ bool& test_multiplexer::stopped_reading(accept_handle hdl) {
return doorman_data_[hdl].stopped_reading; return doorman_data_[hdl].stopped_reading;
} }
bool& test_multiplexer::passive_mode(accept_handle hdl) {
return doorman_data_[hdl].passive_mode;
}
intrusive_ptr<doorman>& test_multiplexer::impl_ptr(accept_handle hdl) { intrusive_ptr<doorman>& test_multiplexer::impl_ptr(accept_handle hdl) {
return doorman_data_[hdl].ptr; return doorman_data_[hdl].ptr;
} }
...@@ -276,14 +301,19 @@ bool test_multiplexer::has_pending_scribe(std::string x, uint16_t y) { ...@@ -276,14 +301,19 @@ bool test_multiplexer::has_pending_scribe(std::string x, uint16_t y) {
} }
void test_multiplexer::accept_connection(accept_handle hdl) { void test_multiplexer::accept_connection(accept_handle hdl) {
if (passive_mode(hdl))
return;
auto& dd = doorman_data_[hdl]; auto& dd = doorman_data_[hdl];
if (!dd.ptr) if (!dd.ptr)
throw std::logic_error("accept_connection: this doorman was not " throw std::logic_error("accept_connection: this doorman was not "
"assigned to a broker yet"); "assigned to a broker yet");
dd.ptr->new_connection(); if (!dd.ptr->new_connection())
passive_mode(hdl) = true;
} }
void test_multiplexer::read_data(connection_handle hdl) { void test_multiplexer::read_data(connection_handle hdl) {
if (passive_mode(hdl))
return;
flush_runnables(); flush_runnables();
scribe_data& sd = scribe_data_[hdl]; scribe_data& sd = scribe_data_[hdl];
while (!sd.ptr) while (!sd.ptr)
...@@ -296,14 +326,16 @@ void test_multiplexer::read_data(connection_handle hdl) { ...@@ -296,14 +326,16 @@ void test_multiplexer::read_data(connection_handle hdl) {
auto last = first + static_cast<ptrdiff_t>(sd.recv_conf.second); auto last = first + static_cast<ptrdiff_t>(sd.recv_conf.second);
sd.rd_buf.insert(sd.rd_buf.end(), first, last); sd.rd_buf.insert(sd.rd_buf.end(), first, last);
sd.xbuf.erase(first, last); sd.xbuf.erase(first, last);
sd.ptr->consume(this, sd.rd_buf.data(), sd.rd_buf.size()); if (!sd.ptr->consume(this, sd.rd_buf.data(), sd.rd_buf.size()))
passive_mode(hdl) = true;
} }
break; break;
case receive_policy_flag::at_least: case receive_policy_flag::at_least:
if (sd.xbuf.size() >= sd.recv_conf.second) { if (sd.xbuf.size() >= sd.recv_conf.second) {
sd.rd_buf.clear(); sd.rd_buf.clear();
sd.rd_buf.swap(sd.xbuf); sd.rd_buf.swap(sd.xbuf);
sd.ptr->consume(this, sd.rd_buf.data(), sd.rd_buf.size()); if (!sd.ptr->consume(this, sd.rd_buf.data(), sd.rd_buf.size()))
passive_mode(hdl) = true;
} }
break; break;
case receive_policy_flag::at_most: case receive_policy_flag::at_most:
...@@ -315,7 +347,8 @@ void test_multiplexer::read_data(connection_handle hdl) { ...@@ -315,7 +347,8 @@ void test_multiplexer::read_data(connection_handle hdl) {
auto last = (max_bytes < xbuf_size) ? first + max_bytes : sd.xbuf.end(); auto last = (max_bytes < xbuf_size) ? first + max_bytes : sd.xbuf.end();
sd.rd_buf.insert(sd.rd_buf.end(), first, last); sd.rd_buf.insert(sd.rd_buf.end(), first, last);
sd.xbuf.erase(first, last); sd.xbuf.erase(first, last);
sd.ptr->consume(this, sd.rd_buf.data(), sd.rd_buf.size()); if (!sd.ptr->consume(this, sd.rd_buf.data(), sd.rd_buf.size()))
passive_mode(hdl) = true;
} }
} }
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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/config.hpp"
#define CAF_SUITE io_resuming
#include "caf/test/unit_test.hpp"
#include <memory>
#include <iostream>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
using namespace std;
using namespace caf;
using namespace caf::io;
namespace {
// -- client implementation, used for both test servers ------------------------
behavior client(broker* self, connection_handle hdl) {
std::vector<char> buf;
buf.resize(200);
std::iota(buf.begin(), buf.end(), 0);
self->write(hdl, buf.size(), buf.data());
CAF_REQUIRE_EQUAL(self->wr_buf(hdl).size(), 200u);
self->configure_read(hdl, receive_policy::at_least(1));
self->flush(hdl);
return {
[=](const new_data_msg&) {
CAF_FAIL("server unexpectedly sent data");
},
[=](const connection_closed_msg&) {
self->quit();
}
};
}
// -- first test server --------------------------------------------------------
struct server1_state {
size_t received = 0;
connection_handle peer = invalid_connection_handle;
};
// consumes 5 more tokens, then waits for passivated message to shutdown
behavior server1_stage4(stateful_actor<server1_state, broker>* self) {
CAF_MESSAGE("enter server stage 4");
self->trigger(self->state.peer, 5);
return {
[=](const new_data_msg& dm) {
CAF_REQUIRE_EQUAL(dm.buf.size(), 10u);
self->state.received += 1;
},
[=](const connection_passivated_msg& cp) {
CAF_REQUIRE_EQUAL(cp.handle, self->state.peer);
CAF_REQUIRE_EQUAL(self->state.received, 15u);
CAF_REQUIRE_NOT_EQUAL(self->state.peer, invalid_connection_handle);
// delay new tokens to force MM to remove this broker from its loop
self->quit();
}
};
}
// consumes 5 more tokens, then waits for passivated message to send itself
// a message that generates 5 more (force MM to actually remove this broker
// from its event loop and then re-adding it)
behavior server1_stage3(stateful_actor<server1_state, broker>* self) {
CAF_MESSAGE("enter server stage 3");
self->trigger(self->state.peer, 5);
return {
[=](const new_data_msg& dm) {
CAF_REQUIRE_EQUAL(dm.buf.size(), 10u);
self->state.received += 1;
},
[=](const connection_passivated_msg& cp) {
CAF_REQUIRE_EQUAL(cp.handle, self->state.peer);
CAF_REQUIRE_EQUAL(self->state.received, 10u);
CAF_REQUIRE_NOT_EQUAL(self->state.peer, invalid_connection_handle);
// delay new tokens to force MM to remove this broker from its loop
self->send(self, ok_atom::value);
},
[=](ok_atom) {
self->become(server1_stage4(self));
}
};
}
// consumes 5 tokens, then waits for passivated message and generates 5 more
behavior server1_stage2(stateful_actor<server1_state, broker>* self) {
CAF_MESSAGE("enter server stage 2");
self->trigger(self->state.peer, 5);
return {
[=](const new_data_msg& dm) {
CAF_REQUIRE_EQUAL(dm.buf.size(), 10u);
self->state.received += 1;
},
[=](const connection_passivated_msg& cp) {
CAF_REQUIRE_EQUAL(cp.handle, self->state.peer);
CAF_REQUIRE_EQUAL(self->state.received, 5u);
CAF_REQUIRE_NOT_EQUAL(self->state.peer, invalid_connection_handle);
self->become(server1_stage3(self));
}
};
}
// waits for the connection to the client
behavior server1(stateful_actor<server1_state, broker>* self) {
return {
[=](const new_connection_msg& nc) {
CAF_REQUIRE_EQUAL(self->state.peer, invalid_connection_handle);
self->state.peer = nc.handle;
self->configure_read(nc.handle, receive_policy::exactly(10));
self->become(server1_stage2(self));
}
};
}
// -- second test server -------------------------------------------------------
struct server2_state {
size_t accepted = 0;
};
// consumes 5 more tokens, then waits for passivated message to shutdown
behavior server2_stage4(stateful_actor<server2_state, broker>* self) {
CAF_MESSAGE("enter server stage 4");
return {
[=](const new_connection_msg&) {
self->state.accepted += 1;
},
[=](const acceptor_passivated_msg&) {
CAF_REQUIRE_EQUAL(self->state.accepted, 16u);
// delay new tokens to force MM to remove this broker from its loop
self->quit();
}
};
}
// consumes 5 more tokens, then waits for passivated message to send itself
// a message that generates 5 more (force MM to actually remove this broker
// from its event loop and then re-adding it)
behavior server2_stage3(stateful_actor<server2_state, broker>* self) {
CAF_MESSAGE("enter server stage 3");
return {
[=](const new_connection_msg&) {
self->state.accepted += 1;
},
[=](const acceptor_passivated_msg& cp) {
CAF_REQUIRE_EQUAL(self->state.accepted, 11u);
// delay new tokens to force MM to remove this broker from its loop
self->send(self, ok_atom::value, cp.handle);
},
[=](ok_atom, accept_handle hdl) {
self->trigger(hdl, 5);
self->become(server2_stage4(self));
}
};
}
// consumes 5 tokens, then waits for passivated message and generates 5 more
behavior server2_stage2(stateful_actor<server2_state, broker>* self) {
CAF_MESSAGE("enter server stage 2");
CAF_REQUIRE_EQUAL(self->state.accepted, 1u);
return {
[=](const new_connection_msg&) {
self->state.accepted += 1;
},
[=](const acceptor_passivated_msg& cp) {
CAF_REQUIRE_EQUAL(self->state.accepted, 6u);
self->trigger(cp.handle, 5);
self->become(server2_stage3(self));
}
};
}
// waits for the connection to the client
behavior server2(stateful_actor<server2_state, broker>* self) {
return {
[=](const new_connection_msg& nc) {
self->state.accepted += 1;
self->trigger(nc.source, 5);
self->become(server2_stage2(self));
}
};
}
// -- config and fixture -------------------------------------------------------
struct config : actor_system_config {
config() {
auto argc = test::engine::argc();
auto argv = test::engine::argv();
load<io::middleman>();
parse(argc, argv);
}
};
struct fixture {
config client_cfg;
actor_system client_system;
config server_cfg;
actor_system server_system;
fixture() : client_system(client_cfg), server_system(server_cfg) {
// nop
}
};
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(trigger_tests, fixture)
CAF_TEST(trigger_connection) {
CAF_MESSAGE("spawn server");
uint16_t port = 0;
auto serv = server_system.middleman().spawn_server(server1, port);
CAF_REQUIRE(serv);
CAF_REQUIRE_NOT_EQUAL(port, 0);
CAF_MESSAGE("server spawned at port " << port);
std::thread child{[&] {
auto cl = client_system.middleman().spawn_client(client, "localhost", port);
CAF_REQUIRE(cl);
}};
child.join();
}
CAF_TEST(trigger_acceptor) {
CAF_MESSAGE("spawn server");
uint16_t port = 0;
auto serv = server_system.middleman().spawn_server(server2, port);
CAF_REQUIRE(serv);
CAF_REQUIRE_NOT_EQUAL(port, 0);
CAF_MESSAGE("server spawned at port " << port);
std::thread child{[&] {
// 16 clients will succeed to connect
for (int i = 0; i < 16; ++i) {
auto cl = client_system.middleman().spawn_client(client,
"localhost", port);
CAF_REQUIRE(cl);
}
// but no longer after that point
auto cl = client_system.middleman().spawn_client(client,
"localhost", port);
CAF_REQUIRE(!cl);
}};
child.join();
}
CAF_TEST_FIXTURE_SCOPE_END()
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