Commit 1b230f37 authored by Jakob Otto's avatar Jakob Otto

Include review feedback

parent 3ed1b937
...@@ -61,8 +61,6 @@ public: ...@@ -61,8 +61,6 @@ public:
/// Returns the index of `mgr` in the pollset or `-1`. /// Returns the index of `mgr` in the pollset or `-1`.
ptrdiff_t index_of(const socket_manager_ptr& mgr); ptrdiff_t index_of(const socket_manager_ptr& mgr);
bool is_same_thread();
// -- thread-safe signaling -------------------------------------------------- // -- thread-safe signaling --------------------------------------------------
/// Registers `mgr` for read events. /// Registers `mgr` for read events.
...@@ -73,10 +71,6 @@ public: ...@@ -73,10 +71,6 @@ public:
/// @thread-safe /// @thread-safe
void register_writing(const socket_manager_ptr& mgr); void register_writing(const socket_manager_ptr& mgr);
/// Unregisters `mgr` for read and write events.
/// @thread-safe
void unregister_manager(const socket_manager_ptr& mgr);
/// Closes the pipe for signaling updates to the multiplexer. After closing /// Closes the pipe for signaling updates to the multiplexer. After closing
/// the pipe, calls to `update` no longer have any effect. /// the pipe, calls to `update` no longer have any effect.
/// @thread-safe /// @thread-safe
...@@ -107,9 +101,6 @@ protected: ...@@ -107,9 +101,6 @@ protected:
/// 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);
/// Deletes a known socket manager from the pollset.
void del(const socket_manager_ptr& mgr);
/// Deletes a known socket manager from the pollset. /// Deletes a known socket manager from the pollset.
void del(ptrdiff_t index); void del(ptrdiff_t index);
...@@ -137,7 +128,7 @@ protected: ...@@ -137,7 +128,7 @@ protected:
std::mutex write_lock_; std::mutex write_lock_;
/// Signals shutdown has been requested. /// Signals shutdown has been requested.
bool will_shutdown_; bool shutting_down_;
}; };
/// @relates multiplexer /// @relates multiplexer
......
...@@ -28,8 +28,7 @@ enum class operation { ...@@ -28,8 +28,7 @@ enum class operation {
read = 0x01, read = 0x01,
write = 0x02, write = 0x02,
read_write = 0x03, read_write = 0x03,
unregister = 0x04, shutdown = 0x04,
shutdown = 0x05,
}; };
constexpr operation operator|(operation x, operation y) { constexpr operation operator|(operation x, operation y) {
......
...@@ -76,7 +76,7 @@ short to_bitmask(operation op) { ...@@ -76,7 +76,7 @@ short to_bitmask(operation op) {
} // namespace } // namespace
multiplexer::multiplexer() { multiplexer::multiplexer() : shutting_down_(false) {
// nop // nop
} }
...@@ -104,13 +104,11 @@ ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) { ...@@ -104,13 +104,11 @@ ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) {
return i == last ? -1 : std::distance(first, i); return i == last ? -1 : std::distance(first, i);
} }
bool multiplexer::is_same_thread() {
return std::this_thread::get_id() == tid_;
}
void multiplexer::register_reading(const socket_manager_ptr& mgr) { void multiplexer::register_reading(const socket_manager_ptr& mgr) {
if (std::this_thread::get_id() == tid_) { if (std::this_thread::get_id() == tid_) {
if (mgr->mask() != operation::none) { if (shutting_down_) {
// discard
} else if (mgr->mask() != operation::none) {
CAF_ASSERT(index_of(mgr) != -1); CAF_ASSERT(index_of(mgr) != -1);
if (mgr->mask_add(operation::read)) { if (mgr->mask_add(operation::read)) {
auto& fd = pollset_[index_of(mgr)]; auto& fd = pollset_[index_of(mgr)];
...@@ -126,7 +124,9 @@ void multiplexer::register_reading(const socket_manager_ptr& mgr) { ...@@ -126,7 +124,9 @@ void multiplexer::register_reading(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() != operation::none) { if (shutting_down_) {
// discard
} else if (mgr->mask() != operation::none) {
CAF_ASSERT(index_of(mgr) != -1); CAF_ASSERT(index_of(mgr) != -1);
if (mgr->mask_add(operation::write)) { if (mgr->mask_add(operation::write)) {
auto& fd = pollset_[index_of(mgr)]; auto& fd = pollset_[index_of(mgr)];
...@@ -140,16 +140,6 @@ void multiplexer::register_writing(const socket_manager_ptr& mgr) { ...@@ -140,16 +140,6 @@ void multiplexer::register_writing(const socket_manager_ptr& mgr) {
} }
} }
void multiplexer::unregister_manager(const socket_manager_ptr& mgr) {
if (std::this_thread::get_id() == tid_) {
del(mgr);
if (will_shutdown_ && managers_.size() == 1)
close_pipe();
} else {
write_to_pipe(4, mgr);
}
}
void multiplexer::close_pipe() { void multiplexer::close_pipe() {
std::lock_guard<std::mutex> guard{write_lock_}; std::lock_guard<std::mutex> guard{write_lock_};
if (write_handle_ != invalid_socket) { if (write_handle_ != invalid_socket) {
...@@ -236,14 +226,20 @@ void multiplexer::run() { ...@@ -236,14 +226,20 @@ void multiplexer::run() {
void multiplexer::shutdown() { void multiplexer::shutdown() {
if (std::this_thread::get_id() == tid_) { if (std::this_thread::get_id() == tid_) {
will_shutdown_ = true; shutting_down_ = true;
if (managers_.size() == 1) {
close_pipe();
} else {
// First manager is the pollset_updater. Skip it and delete later. // First manager is the pollset_updater. Skip it and delete later.
for (size_t i = 1; i < managers_.size(); ++i) for (size_t i = 1; i < managers_.size();) {
write_to_pipe(4, managers_[i]); auto& mgr = managers_[i];
if (mgr->mask_del(operation::read)) {
auto& fd = pollset_[index_of(mgr)];
fd.events &= ~input_mask;
} }
if (mgr->mask() == operation::none)
del(i);
else
++i;
}
close_pipe();
} else { } else {
write_to_pipe(5, nullptr); write_to_pipe(5, nullptr);
} }
...@@ -289,10 +285,6 @@ void multiplexer::add(socket_manager_ptr mgr) { ...@@ -289,10 +285,6 @@ void multiplexer::add(socket_manager_ptr mgr) {
managers_.emplace_back(std::move(mgr)); managers_.emplace_back(std::move(mgr));
} }
void multiplexer::del(const socket_manager_ptr& mgr) {
del(index_of(mgr));
}
void multiplexer::del(ptrdiff_t index) { void multiplexer::del(ptrdiff_t index) {
CAF_ASSERT(index != -1); CAF_ASSERT(index != -1);
pollset_.erase(pollset_.begin() + index); pollset_.erase(pollset_.begin() + index);
...@@ -301,7 +293,7 @@ void multiplexer::del(ptrdiff_t index) { ...@@ -301,7 +293,7 @@ void multiplexer::del(ptrdiff_t index) {
void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) { void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) {
CAF_ASSERT(opcode == 0 || opcode == 1 || opcode == 4 || opcode == 5); CAF_ASSERT(opcode == 0 || opcode == 1 || opcode == 4 || opcode == 5);
CAF_ASSERT(mgr != nullptr || opcode == 5); CAF_ASSERT(mgr != nullptr || opcode == 4);
pollset_updater::msg_buf buf; pollset_updater::msg_buf buf;
if (opcode != 5) if (opcode != 5)
mgr->ref(); mgr->ref();
...@@ -316,7 +308,7 @@ void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) { ...@@ -316,7 +308,7 @@ void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) {
else else
res = sec::socket_invalid; res = sec::socket_invalid;
} }
if (holds_alternative<sec>(res) && opcode != 5) if (holds_alternative<sec>(res) && opcode != 4)
mgr->deref(); mgr->deref();
} }
......
...@@ -56,10 +56,10 @@ void middleman::stop() { ...@@ -56,10 +56,10 @@ void middleman::stop() {
for (const auto& backend : backends_) for (const auto& backend : backends_)
backend->stop(); backend->stop();
mpx_->shutdown(); mpx_->shutdown();
if (mpx_->is_same_thread()) if (mpx_thread_.joinable())
mpx_->run();
else if (mpx_thread_.joinable())
mpx_thread_.join(); mpx_thread_.join();
else
mpx_->run();
} }
void middleman::init(actor_system_config& cfg) { void middleman::init(actor_system_config& cfg) {
......
...@@ -50,21 +50,16 @@ bool pollset_updater::handle_read_event() { ...@@ -50,21 +50,16 @@ bool pollset_updater::handle_read_event() {
auto opcode = static_cast<uint8_t>(buf_[0]); auto opcode = static_cast<uint8_t>(buf_[0]);
intptr_t value; intptr_t value;
memcpy(&value, buf_.data() + 1, sizeof(intptr_t)); memcpy(&value, buf_.data() + 1, sizeof(intptr_t));
socket_manager_ptr mgr{reinterpret_cast<socket_manager*>(value), false};
if (auto ptr = parent_.lock()) { if (auto ptr = parent_.lock()) {
switch (opcode) { switch (opcode) {
case 0: case 0:
ptr->register_reading( ptr->register_reading(mgr);
{reinterpret_cast<socket_manager*>(value), false});
break; break;
case 1: case 1:
ptr->register_writing( ptr->register_writing(mgr);
{reinterpret_cast<socket_manager*>(value), false});
break; break;
case 4: case 4:
ptr->unregister_manager(
{reinterpret_cast<socket_manager*>(value), false});
break;
case 5:
ptr->shutdown(); ptr->shutdown();
break; break;
default: default:
......
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