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

Include review feedback

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