Commit aaa7fecf authored by Dominik Charousset's avatar Dominik Charousset

Allow users to manually advance multiplexers

parent 75f4f39c
...@@ -287,6 +287,7 @@ public: ...@@ -287,6 +287,7 @@ public:
size_t middleman_max_consecutive_reads; size_t middleman_max_consecutive_reads;
size_t middleman_heartbeat_interval; size_t middleman_heartbeat_interval;
bool middleman_detach_utility_actors; bool middleman_detach_utility_actors;
bool middleman_detach_multiplexer;
// -- config parameters of the OpenCL module --------------------------------- // -- config parameters of the OpenCL module ---------------------------------
......
...@@ -133,6 +133,7 @@ actor_system_config::actor_system_config() ...@@ -133,6 +133,7 @@ actor_system_config::actor_system_config()
middleman_enable_automatic_connections = false; middleman_enable_automatic_connections = false;
middleman_max_consecutive_reads = 50; middleman_max_consecutive_reads = 50;
middleman_heartbeat_interval = 0; middleman_heartbeat_interval = 0;
middleman_detach_multiplexer = true;
// fill our options vector for creating INI and CLI parsers // fill our options vector for creating INI and CLI parsers
opt_group{options_, "scheduler"} opt_group{options_, "scheduler"}
.add(scheduler_policy, "policy", .add(scheduler_policy, "policy",
...@@ -193,7 +194,9 @@ actor_system_config::actor_system_config() ...@@ -193,7 +194,9 @@ actor_system_config::actor_system_config()
.add(middleman_heartbeat_interval, "heartbeat-interval", .add(middleman_heartbeat_interval, "heartbeat-interval",
"sets the interval (ms) of heartbeat, 0 (default) means disabling it") "sets the interval (ms) of heartbeat, 0 (default) means disabling it")
.add(middleman_detach_utility_actors, "detach-utility-actors", .add(middleman_detach_utility_actors, "detach-utility-actors",
"enables or disables detaching of utility actors"); "enables or disables detaching of utility actors")
.add(middleman_detach_multiplexer, "detach-multiplexer",
"enables or disables background activity of the multiplexer");
opt_group(options_, "opencl") opt_group(options_, "opencl")
.add(opencl_device_ids, "device-ids", .add(opencl_device_ids, "device-ids",
"restricts which OpenCL devices are accessed by CAF"); "restricts which OpenCL devices are accessed by CAF");
......
...@@ -80,6 +80,10 @@ public: ...@@ -80,6 +80,10 @@ public:
supervisor_ptr make_supervisor() override; supervisor_ptr make_supervisor() override;
bool try_run_once() override;
void run_once() override;
void run() override; void run() override;
boost::asio::io_service* pimpl() override; boost::asio::io_service* pimpl() override;
......
...@@ -285,6 +285,21 @@ multiplexer::supervisor_ptr asio_multiplexer::make_supervisor() { ...@@ -285,6 +285,21 @@ multiplexer::supervisor_ptr asio_multiplexer::make_supervisor() {
return std::unique_ptr<asio_supervisor>(new asio_supervisor(service())); return std::unique_ptr<asio_supervisor>(new asio_supervisor(service()));
} }
bool asio_multiplexer::try_run_once() {
boost::system::error_code ec;
auto num = service().poll(ec);
if (ec)
CAF_RAISE_ERROR(ec.message());
return num > 0;
}
void asio_multiplexer::run_once() {
boost::system::error_code ec;
service().run_one(ec);
if (ec)
CAF_RAISE_ERROR(ec.message());
}
void asio_multiplexer::run() { void asio_multiplexer::run() {
CAF_LOG_TRACE("asio-based multiplexer"); CAF_LOG_TRACE("asio-based multiplexer");
boost::system::error_code ec; boost::system::error_code ec;
......
...@@ -300,6 +300,12 @@ public: ...@@ -300,6 +300,12 @@ public:
supervisor_ptr make_supervisor() override; supervisor_ptr make_supervisor() override;
bool poll_once(bool block);
bool try_run_once() override;
void run_once() override;
void run() override; void run() override;
void add(operation op, native_socket fd, event_handler* ptr); void add(operation op, native_socket fd, event_handler* ptr);
......
...@@ -96,7 +96,14 @@ public: ...@@ -96,7 +96,14 @@ public:
/// Creates an instance using the networking backend compiled with CAF. /// Creates an instance using the networking backend compiled with CAF.
static std::unique_ptr<multiplexer> make(actor_system& sys); static std::unique_ptr<multiplexer> make(actor_system& sys);
/// Runs the multiplexers event loop. /// Exectutes all pending events without blocking.
/// @returns `true` if at least one event was called, `false` otherwise.
virtual bool try_run_once() = 0;
/// Runs at least one event and blocks if needed.
virtual void run_once() = 0;
/// Runs events until all connection are closed.
virtual void run() = 0; virtual void run() = 0;
/// Invokes @p fun in the multiplexer's event loop, calling `fun()` /// Invokes @p fun in the multiplexer's event loop, calling `fun()`
......
...@@ -57,6 +57,10 @@ public: ...@@ -57,6 +57,10 @@ public:
supervisor_ptr make_supervisor() override; supervisor_ptr make_supervisor() override;
bool try_run_once() override;
void run_once() override;
void run() override; void run() override;
scribe_ptr new_scribe(connection_handle); scribe_ptr new_scribe(connection_handle);
......
...@@ -308,14 +308,15 @@ namespace network { ...@@ -308,14 +308,15 @@ namespace network {
} }
} }
void default_multiplexer::run() { bool default_multiplexer::poll_once(bool block) {
CAF_LOG_TRACE("epoll()-based multiplexer"); CAF_LOG_TRACE("epoll()-based multiplexer");
while (shadow_ > 0) { // Keep running in case of `EINTR`.
for (;;) {
int presult = epoll_wait(epollfd_, pollset_.data(), int presult = epoll_wait(epollfd_, pollset_.data(),
static_cast<int>(pollset_.size()), -1); static_cast<int>(pollset_.size()),
CAF_LOG_DEBUG("epoll_wait() on " << CAF_ARG(shadow_) block ? -1 : 0);
<< " sockets reported " << CAF_ARG(presult) CAF_LOG_DEBUG("epoll_wait() on" << shadow_
<< " event(s)"); << "sockets reported" << presult << "event(s)");
if (presult < 0) { if (presult < 0) {
switch (errno) { switch (errno) {
case EINTR: { case EINTR: {
...@@ -329,6 +330,8 @@ namespace network { ...@@ -329,6 +330,8 @@ namespace network {
} }
} }
} }
if (presult == 0)
return false;
auto iter = pollset_.begin(); auto iter = pollset_.begin();
auto last = iter + presult; auto last = iter + presult;
for (; iter != last; ++iter) { for (; iter != last; ++iter) {
...@@ -340,7 +343,22 @@ namespace network { ...@@ -340,7 +343,22 @@ namespace network {
handle(me); handle(me);
} }
events_.clear(); events_.clear();
return true;
}
}
bool default_multiplexer::try_run_once() {
return poll_once(false);
} }
bool default_multiplexer::run_once() {
return poll_once(true);
}
void default_multiplexer::run() {
CAF_LOG_TRACE("epoll()-based multiplexer");
while (shadow_ > 0)
poll_once(true);
} }
void default_multiplexer::handle(const default_multiplexer::event& e) { void default_multiplexer::handle(const default_multiplexer::event& e) {
...@@ -439,9 +457,7 @@ namespace network { ...@@ -439,9 +457,7 @@ namespace network {
shadow_.push_back(&pipe_reader_); shadow_.push_back(&pipe_reader_);
} }
void default_multiplexer::run() { bool default_multiplexer::poll_once(bool block) {
CAF_LOG_TRACE("poll()-based multiplexer; " << CAF_ARG(input_mask)
<< CAF_ARG(output_mask) << CAF_ARG(error_mask));
// we store the results of poll() in a separate vector , because // we store the results of poll() in a separate vector , because
// altering the pollset while traversing it is not exactly a // altering the pollset while traversing it is not exactly a
// bright idea ... // bright idea ...
...@@ -451,15 +467,15 @@ namespace network { ...@@ -451,15 +467,15 @@ namespace network {
event_handler* ptr; // nullptr in case of a pipe event event_handler* ptr; // nullptr in case of a pipe event
}; };
std::vector<fd_event> poll_res; std::vector<fd_event> poll_res;
while (!pollset_.empty()) { for(;;) {
int presult; int presult;
CAF_LOG_DEBUG(CAF_ARG(pollset_.size()));
# ifdef CAF_WINDOWS # ifdef CAF_WINDOWS
presult = ::WSAPoll(pollset_.data(), presult = ::WSAPoll(pollset_.data(),
static_cast<ULONG>(pollset_.size()), -1); static_cast<ULONG>(pollset_.size()),
block ? -1 : 0);
# else # else
presult = ::poll(pollset_.data(), presult = ::poll(pollset_.data(),
static_cast<nfds_t>(pollset_.size()), -1); static_cast<nfds_t>(pollset_.size()), block ? -1 : 0);
# endif # endif
if (presult < 0) { if (presult < 0) {
switch (last_socket_error()) { switch (last_socket_error()) {
...@@ -482,6 +498,8 @@ namespace network { ...@@ -482,6 +498,8 @@ namespace network {
} }
continue; // rince and repeat continue; // rince and repeat
} }
if (presult == 0)
return false;
// scan pollset for events first, because we might alter pollset_ // scan pollset for events first, because we might alter pollset_
// while running callbacks (not a good idea while traversing it) // while running callbacks (not a good idea while traversing it)
CAF_LOG_DEBUG("scan pollset for socket events"); CAF_LOG_DEBUG("scan pollset for socket events");
...@@ -508,7 +526,23 @@ namespace network { ...@@ -508,7 +526,23 @@ namespace network {
handle(me); handle(me);
} }
events_.clear(); events_.clear();
return true;
}
}
bool default_multiplexer::try_run_once() {
return poll_once(false);
} }
void default_multiplexer::run_once() {
poll_once(true);
}
void default_multiplexer::run() {
CAF_LOG_TRACE("poll()-based multiplexer:" << CAF_ARG(input_mask)
<< CAF_ARG(output_mask) << CAF_ARG(error_mask));
while (!pollset_.empty())
poll_once(true);
} }
void default_multiplexer::handle(const default_multiplexer::event& e) { void default_multiplexer::handle(const default_multiplexer::event& e) {
......
...@@ -257,10 +257,13 @@ void middleman::start() { ...@@ -257,10 +257,13 @@ void middleman::start() {
for (auto& f : system().config().hook_factories) for (auto& f : system().config().hook_factories)
hooks_.emplace_back(f(system_)); hooks_.emplace_back(f(system_));
// Launch backend. // Launch backend.
if (system_.config().middleman_detach_multiplexer)
backend_supervisor_ = backend().make_supervisor(); backend_supervisor_ = backend().make_supervisor();
if (!backend_supervisor_) { if (!backend_supervisor_) {
// The only backend that returns a `nullptr` is the `test_multiplexer` // The only backend that returns a `nullptr` is the `test_multiplexer`
// which does not have its own thread but uses the main thread instead. // which does not have its own thread but uses the main thread instead.
// Other backends can set `middleman_detach_multiplexer` to false to
// suppress creation of the supervisor.
backend().thread_id(std::this_thread::get_id()); backend().thread_id(std::this_thread::get_id());
} else { } else {
thread_ = std::thread{[this] { thread_ = std::thread{[this] {
......
...@@ -251,6 +251,14 @@ auto test_multiplexer::make_supervisor() -> supervisor_ptr { ...@@ -251,6 +251,14 @@ auto test_multiplexer::make_supervisor() -> supervisor_ptr {
return nullptr; return nullptr;
} }
bool test_multiplexer::try_run_once() {
return try_exec_runnable() || try_read_data() || try_accept_connection();
}
void test_multiplexer::run_once() {
try_run_once();
}
void test_multiplexer::run() { void test_multiplexer::run() {
// nop // nop
} }
......
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