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

Get rid of dirty_managers_ indirection

parent 91b12761
...@@ -89,14 +89,11 @@ public: ...@@ -89,14 +89,11 @@ public:
/// Polls until no socket event handler remains. /// Polls until no socket event handler remains.
void run(); void run();
/// Processes all updates on the socket managers.
void handle_updates();
protected: protected:
// -- utility functions ------------------------------------------------------ // -- utility functions ------------------------------------------------------
/// Handles an I/O event on given manager. /// Handles an I/O event on given manager.
void handle(const socket_manager_ptr& mgr, int mask); void handle(pollfd& fd, const socket_manager_ptr& mgr);
/// Adds a new socket manager to the pollset. /// Adds a new socket manager to the pollset.
void add(socket_manager_ptr mgr); void add(socket_manager_ptr mgr);
...@@ -110,9 +107,6 @@ protected: ...@@ -110,9 +107,6 @@ protected:
/// order as their sockets appear in `pollset_`. /// order as their sockets appear in `pollset_`.
manager_list managers_; manager_list managers_;
/// Managers that updated their event mask and need updating.
manager_list dirty_managers_;
/// Stores the ID of the thread this multiplexer is running in. Set when /// Stores the ID of the thread this multiplexer is running in. Set when
/// calling `init()`. /// calling `init()`.
std::thread::id tid_; std::thread::id tid_;
......
...@@ -108,10 +108,14 @@ ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) { ...@@ -108,10 +108,14 @@ ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) {
void multiplexer::register_writing(const socket_manager_ptr& mgr) { void multiplexer::register_writing(const socket_manager_ptr& mgr) {
if (std::this_thread::get_id() == tid_) { if (std::this_thread::get_id() == tid_) {
if (mgr->mask_add(operation::write)) { if (mgr->mask() != operation::none) {
auto i = std::find(dirty_managers_.begin(), dirty_managers_.end(), mgr); CAF_ASSERT(index_of(mgr) != -1);
if (i == dirty_managers_.end()) if (mgr->mask_add(operation::write)) {
dirty_managers_.emplace_back(mgr); auto& fd = pollset_[index_of(mgr)];
fd.events |= output_mask;
}
} else if (mgr->mask_add(operation::write)) {
add(mgr);
} }
} else { } else {
mgr->ref(); mgr->ref();
...@@ -139,10 +143,13 @@ void multiplexer::close_pipe() { ...@@ -139,10 +143,13 @@ void multiplexer::close_pipe() {
void multiplexer::register_reading(const socket_manager_ptr& mgr) { void multiplexer::register_reading(const socket_manager_ptr& mgr) {
CAF_ASSERT(std::this_thread::get_id() == tid_); CAF_ASSERT(std::this_thread::get_id() == tid_);
if (mgr->mask_add(operation::read)) { if (mgr->mask() != operation::none) {
auto i = std::find(dirty_managers_.begin(), dirty_managers_.end(), mgr); CAF_ASSERT(index_of(mgr) != -1);
if (i == dirty_managers_.end()) mgr->mask_add(operation::read);
dirty_managers_.emplace_back(mgr); auto& fd = pollset_[index_of(mgr)];
fd.events |= input_mask;
} else if (mgr->mask_add(operation::read)) {
add(mgr);
} }
} }
...@@ -192,14 +199,20 @@ bool multiplexer::poll_once(bool blocking) { ...@@ -192,14 +199,20 @@ bool multiplexer::poll_once(bool blocking) {
return false; return false;
// Scan pollset for events. // Scan pollset for events.
CAF_LOG_DEBUG("scan pollset for socket events"); CAF_LOG_DEBUG("scan pollset for socket events");
for (size_t i = 0; i < pollset_.size() && presult > 0; ++i) { for (size_t i = 0; i < pollset_.size() && presult > 0;) {
auto& x = pollset_[i]; auto& x = pollset_[i];
auto& mgr = managers_[i];
if (x.revents != 0) { if (x.revents != 0) {
handle(managers_[i], x.revents); handle(x, mgr);
--presult; --presult;
if (x.events == 0) {
pollset_.erase(pollset_.begin() + i);
managers_.erase(managers_.begin() + i);
continue;
}
} }
++i;
} }
handle_updates();
return true; return true;
} }
} }
...@@ -214,37 +227,24 @@ void multiplexer::run() { ...@@ -214,37 +227,24 @@ void multiplexer::run() {
poll_once(true); poll_once(true);
} }
void multiplexer::handle_updates() { void multiplexer::handle(pollfd& fd, const socket_manager_ptr& mgr) {
for (auto mgr : dirty_managers_) { CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle()));
auto index = index_of(mgr.get());
if (index == -1) {
add(std::move(mgr));
} else {
// Update or remove an existing manager in the pollset.
if (mgr->mask() == operation::none) {
pollset_.erase(pollset_.begin() + index);
managers_.erase(managers_.begin() + index);
} else {
pollset_[index].events = to_bitmask(mgr->mask());
}
}
}
dirty_managers_.clear();
}
void multiplexer::handle(const socket_manager_ptr& mgr, int mask) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle()) << CAF_ARG(mask));
CAF_ASSERT(mgr != nullptr); CAF_ASSERT(mgr != nullptr);
bool checkerror = true; bool checkerror = true;
auto mask = fd.revents;
if ((mask & input_mask) != 0) { if ((mask & input_mask) != 0) {
checkerror = false; checkerror = false;
if (!mgr->handle_read_event() && mgr->mask_del(operation::read)) if (!mgr->handle_read_event()) {
dirty_managers_.emplace_back(mgr); mgr->mask_del(operation::read);
fd.events = fd.events & ~input_mask;
}
} }
if ((mask & output_mask) != 0) { if ((mask & output_mask) != 0) {
checkerror = false; checkerror = false;
if (!mgr->handle_write_event() && mgr->mask_del(operation::write)) if (!mgr->handle_write_event()) {
dirty_managers_.emplace_back(mgr); mgr->mask_del(operation::write);
fd.events = fd.events & ~output_mask;
}
} }
if (checkerror && ((mask & error_mask) != 0)) { if (checkerror && ((mask & error_mask) != 0)) {
if (mask & POLLNVAL) if (mask & POLLNVAL)
...@@ -254,11 +254,12 @@ void multiplexer::handle(const socket_manager_ptr& mgr, int mask) { ...@@ -254,11 +254,12 @@ void multiplexer::handle(const socket_manager_ptr& mgr, int mask) {
else else
mgr->handle_error(sec::socket_operation_failed); mgr->handle_error(sec::socket_operation_failed);
mgr->mask_del(operation::read_write); mgr->mask_del(operation::read_write);
dirty_managers_.emplace_back(mgr); fd.events = 0;
} }
} }
void multiplexer::add(socket_manager_ptr mgr) { void multiplexer::add(socket_manager_ptr mgr) {
CAF_ASSERT(index_of(mgr) == -1);
pollfd new_entry{socket_cast<socket_id>(mgr->handle()), pollfd new_entry{socket_cast<socket_id>(mgr->handle()),
to_bitmask(mgr->mask()), 0}; to_bitmask(mgr->mask()), 0};
pollset_.emplace_back(new_entry); pollset_.emplace_back(new_entry);
......
...@@ -54,7 +54,6 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -54,7 +54,6 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
} }
bool handle_io_event() override { bool handle_io_event() override {
mpx->handle_updates();
return mpx->poll_once(false); return mpx->poll_once(false);
} }
...@@ -178,7 +177,6 @@ CAF_TEST(send and receive) { ...@@ -178,7 +177,6 @@ CAF_TEST(send and receive) {
CAF_CHECK_EQUAL(mgr->mask(), operation::none); CAF_CHECK_EQUAL(mgr->mask(), operation::none);
CAF_CHECK_EQUAL(mgr->init(), none); CAF_CHECK_EQUAL(mgr->init(), none);
CAF_CHECK_EQUAL(mgr->mask(), operation::read_write); CAF_CHECK_EQUAL(mgr->mask(), operation::read_write);
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
CAF_CHECK_EQUAL(write(sockets.second, as_bytes(make_span(hello_manager))), CAF_CHECK_EQUAL(write(sockets.second, as_bytes(make_span(hello_manager))),
hello_manager.size()); hello_manager.size());
...@@ -202,7 +200,6 @@ CAF_TEST(resolve and proxy communication) { ...@@ -202,7 +200,6 @@ CAF_TEST(resolve and proxy communication) {
dummy_transport{sockets.first, buf}); dummy_transport{sockets.first, buf});
CAF_CHECK_EQUAL(mgr->init(), none); CAF_CHECK_EQUAL(mgr->init(), none);
CAF_CHECK_EQUAL(mgr->mask(), operation::read_write); CAF_CHECK_EQUAL(mgr->mask(), operation::read_write);
mpx->handle_updates();
run(); run();
CAF_CHECK_EQUAL(read(sockets.second, make_span(read_buf)), hello_test.size()); CAF_CHECK_EQUAL(read(sockets.second, make_span(read_buf)), hello_test.size());
mgr->resolve(unbox(make_uri("test:id/42")), self); mgr->resolve(unbox(make_uri("test:id/42")), self);
......
...@@ -167,11 +167,9 @@ CAF_TEST(send and receive) { ...@@ -167,11 +167,9 @@ CAF_TEST(send and receive) {
auto bob = make_counted<dummy_manager>(manager_count, sockets.second, mpx); auto bob = make_counted<dummy_manager>(manager_count, sockets.second, mpx);
alice->register_reading(); alice->register_reading();
bob->register_reading(); bob->register_reading();
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3u);
alice->send("hello bob"); alice->send("hello bob");
alice->register_writing(); alice->register_writing();
mpx->handle_updates();
exhaust(); exhaust();
CAF_CHECK_EQUAL(bob->receive(), "hello bob"); CAF_CHECK_EQUAL(bob->receive(), "hello bob");
} }
......
...@@ -165,8 +165,6 @@ struct fixture : planet_driver { ...@@ -165,8 +165,6 @@ struct fixture : planet_driver {
} }
bool handle_io_event() override { bool handle_io_event() override {
earth.mpx.handle_updates();
mars.mpx.handle_updates();
return earth.mpx.poll_once(false) || mars.mpx.poll_once(false); return earth.mpx.poll_once(false) || mars.mpx.poll_once(false);
} }
......
...@@ -81,7 +81,6 @@ struct fixture : test_coordinator_fixture<config>, host_fixture { ...@@ -81,7 +81,6 @@ struct fixture : test_coordinator_fixture<config>, host_fixture {
bool handle_io_event() override { bool handle_io_event() override {
auto mpx = sys.network_manager().mpx(); auto mpx = sys.network_manager().mpx();
mpx->handle_updates();
return mpx->poll_once(false); return mpx->poll_once(false);
} }
......
...@@ -52,7 +52,6 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -52,7 +52,6 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
} }
bool handle_io_event() override { bool handle_io_event() override {
mpx->handle_updates();
return mpx->poll_once(false); return mpx->poll_once(false);
} }
...@@ -152,7 +151,6 @@ CAF_TEST(receive) { ...@@ -152,7 +151,6 @@ CAF_TEST(receive) {
transport.configure_read(net::receive_policy::exactly(hello_manager.size())); transport.configure_read(net::receive_policy::exactly(hello_manager.size()));
auto mgr = make_endpoint_manager(mpx, sys, transport); auto mgr = make_endpoint_manager(mpx, sys, transport);
CAF_CHECK_EQUAL(mgr->init(), none); CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
CAF_CHECK_EQUAL(write(sockets.second, as_bytes(make_span(hello_manager))), CAF_CHECK_EQUAL(write(sockets.second, as_bytes(make_span(hello_manager))),
hello_manager.size()); hello_manager.size());
...@@ -174,7 +172,6 @@ CAF_TEST(resolve and proxy communication) { ...@@ -174,7 +172,6 @@ CAF_TEST(resolve and proxy communication) {
transport_type{sockets.first, transport_type{sockets.first,
dummy_application{buf}}); dummy_application{buf}});
CAF_CHECK_EQUAL(mgr->init(), none); CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
run(); run();
mgr->resolve(unbox(make_uri("test:/id/42")), self); mgr->resolve(unbox(make_uri("test:/id/42")), self);
run(); run();
......
...@@ -53,7 +53,6 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -53,7 +53,6 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
} }
bool handle_io_event() override { bool handle_io_event() override {
mpx->handle_updates();
return mpx->poll_once(false); return mpx->poll_once(false);
} }
...@@ -218,13 +217,11 @@ CAF_TEST(receive) { ...@@ -218,13 +217,11 @@ CAF_TEST(receive) {
transport_type{sockets.first, transport_type{sockets.first,
application_type{sys, buf}}); application_type{sys, buf}});
CAF_CHECK_EQUAL(mgr1->init(), none); CAF_CHECK_EQUAL(mgr1->init(), none);
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
auto mgr2 = make_endpoint_manager(mpx, sys, auto mgr2 = make_endpoint_manager(mpx, sys,
transport_type{sockets.second, transport_type{sockets.second,
application_type{sys, buf}}); application_type{sys, buf}});
CAF_CHECK_EQUAL(mgr2->init(), none); CAF_CHECK_EQUAL(mgr2->init(), none);
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3u);
CAF_MESSAGE("resolve actor-proxy"); CAF_MESSAGE("resolve actor-proxy");
mgr1->resolve(unbox(make_uri("test:/id/42")), self); mgr1->resolve(unbox(make_uri("test:/id/42")), self);
......
...@@ -152,7 +152,6 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -152,7 +152,6 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
} }
bool handle_io_event() override { bool handle_io_event() override {
mpx->handle_updates();
return mpx->poll_once(false); return mpx->poll_once(false);
} }
......
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