Commit c3eb1365 authored by Dominik Charousset's avatar Dominik Charousset

Fix flag handling in the multiplexer

parent 31e73412
......@@ -57,6 +57,7 @@ caf_incubator_add_component(
net.actor_shell
net.consumer_adapter
net.length_prefix_framing
net.operation
net.producer_adapter
net.typed_actor_shell
net.web_socket.client
......
......@@ -17,31 +17,94 @@ namespace caf::net {
/// Values for representing bitmask of I/O operations.
enum class operation {
none = 0x00,
read = 0x01,
write = 0x02,
read_write = 0x03,
shutdown = 0x04,
none = 0b0000,
read = 0b0001,
write = 0b0010,
block_read = 0b0100,
block_write = 0b1000,
read_write = 0b0011,
read_only = 0b1001,
write_only = 0b0110,
shutdown = 0b1100,
};
/// @relates operation
constexpr operation operator|(operation x, operation y) {
return static_cast<operation>(static_cast<int>(x) | static_cast<int>(y));
[[nodiscard]] constexpr int to_integer(operation x) noexcept {
return static_cast<int>(x);
}
/// Adds the `read` flag to `x` unless the `block_read` bit is set.
/// @relates operation
constexpr operation operator&(operation x, operation y) {
return static_cast<operation>(static_cast<int>(x) & static_cast<int>(y));
[[nodiscard]] constexpr operation add_read_flag(operation x) noexcept {
if (auto bits = to_integer(x); !(bits & 0b0100))
return static_cast<operation>(bits | 0b0001);
else
return x;
}
/// Adds the `write` flag to `x` unless the `block_write` bit is set.
/// @relates operation
constexpr operation operator^(operation x, operation y) {
return static_cast<operation>(static_cast<int>(x) ^ static_cast<int>(y));
[[nodiscard]] constexpr operation add_write_flag(operation x) noexcept {
if (auto bits = to_integer(x); !(bits & 0b1000))
return static_cast<operation>(bits | 0b0010);
else
return x;
}
/// Removes the `read` flag from `x`.
/// @relates operation
constexpr operation operator~(operation x) {
return static_cast<operation>(~static_cast<int>(x));
[[nodiscard]] constexpr operation remove_read_flag(operation x) noexcept {
return static_cast<operation>(to_integer(x) & 0b1110);
}
/// Removes the `write` flag from `x`.
/// @relates operation
[[nodiscard]] constexpr operation remove_write_flag(operation x) noexcept {
return static_cast<operation>(to_integer(x) & 0b1101);
}
/// Adds the `block_read` flag to `x` and removes the `read` flag if present.
/// @relates operation
[[nodiscard]] constexpr operation block_reads(operation x) noexcept {
auto bits = to_integer(x) | 0b0100;;
return static_cast<operation>(bits & 0b1110);
}
/// Adds the `block_write` flag to `x` and removes the `write` flag if present.
/// @relates operation
[[nodiscard]] constexpr operation block_writes(operation x) noexcept {
auto bits = to_integer(x) | 0b1000;;
return static_cast<operation>(bits & 0b1101);
}
/// Returns whether the read flag is present in `x`.
/// @relates operation
[[nodiscard]] constexpr bool is_reading(operation x) noexcept {
return (to_integer(x) & 0b0001) == 0b0001;
}
/// Returns whether the write flag is present in `x`.
/// @relates operation
[[nodiscard]] constexpr bool is_writing(operation x) noexcept {
return (to_integer(x) & 0b0010) == 0b0010;
}
/// Returns `!is_reading(x) && !is_writing(x)`.
/// @relates operation
[[nodiscard]] constexpr bool is_idle(operation x) noexcept {
return (to_integer(x) & 0b0011) == 0b0000;
}
/// Returns whether the block read flag is present in `x`.
/// @relates operation
[[nodiscard]] constexpr bool is_read_blocked(operation x) noexcept {
return (to_integer(x) & 0b0100) == 0b0100;
}
/// Returns whether the block write flag is present in `x`.
/// @relates operation
[[nodiscard]] constexpr bool is_write_blocked(operation x) noexcept {
return (to_integer(x) & 0b1000) == 0b1000;
}
/// @relates operation
......
......@@ -81,15 +81,38 @@ public:
return mask_;
}
/// Adds given flag(s) to the event mask.
/// @returns `false` if `mask() | flag == mask()`, `true` otherwise.
/// @pre `flag != operation::none`
bool mask_add(operation flag) noexcept;
/// Convenience function for checking whether `mask()` contains the read bit.
bool is_reading() const noexcept {
return net::is_reading(mask_);
}
/// Convenience function for checking whether `mask()` contains the write bit.
bool is_writing() const noexcept {
return net::is_writing(mask_);
}
/// Tries to add the read flag to the event mask.
/// @returns `true` if the flag was added, `false` if this call had no effect.
bool set_read_flag() noexcept;
/// Tries to add the write flag to the event mask.
/// @returns `true` if the flag was added, `false` if this call had no effect.
bool set_write_flag() noexcept;
/// Removes the read flag from the event mask if present.
bool unset_read_flag() noexcept;
/// Tries to clear given flag(s) from the event mask.
/// @returns `false` if `mask() & ~flag == mask()`, `true` otherwise.
/// @pre `flag != operation::none`
bool mask_del(operation flag) noexcept;
/// Removes the write flag from the event mask if present.
bool unset_write_flag() noexcept;
/// Adds the `block_read` flag to the event mask.
void block_reads() noexcept;
/// Adds the `block_write` flag to the event mask.
void block_writes() noexcept;
/// Blocks reading and writing in the event mask.
void block_reads_and_writes() noexcept;
const error& abort_reason() const noexcept {
return abort_reason_;
......@@ -205,14 +228,17 @@ public:
// -- event callbacks --------------------------------------------------------
bool handle_read_event() override {
CAF_LOG_TRACE("");
return protocol_.handle_read_event(this);
}
bool handle_write_event() override {
CAF_LOG_TRACE("");
return protocol_.handle_write_event(this);
}
void handle_error(sec code) override {
CAF_LOG_TRACE(CAF_ARG(code));
abort_reason_ = code;
return protocol_.abort(this, abort_reason_);
}
......
......@@ -51,17 +51,9 @@ const short error_mask = POLLRDHUP | POLLERR | POLLHUP | POLLNVAL;
const short output_mask = POLLOUT;
short to_bitmask(operation op) {
switch (op) {
case operation::read:
return input_mask;
case operation::write:
return output_mask;
case operation::read_write:
return input_mask | output_mask;
default:
return 0;
}
short to_bitmask(operation mask) {
return static_cast<short>((is_reading(mask) ? input_mask : 0)
| (is_writing(mask) ? output_mask : 0));
}
} // namespace
......@@ -138,13 +130,12 @@ void multiplexer::register_reading(const socket_manager_ptr& mgr) {
if (std::this_thread::get_id() == tid_) {
if (shutting_down_ || mgr->abort_reason()) {
// nop
} else if (mgr->mask() != operation::none) {
if (auto index = index_of(mgr);
index != -1 && mgr->mask_add(operation::read)) {
} else if (!is_idle(mgr->mask())) {
if (auto index = index_of(mgr); index != -1 && mgr->set_read_flag()) {
auto& fd = pollset_[index];
fd.events |= input_mask;
}
} else if (mgr->mask_add(operation::read)) {
} else if (mgr->set_read_flag()) {
add(mgr);
}
} else {
......@@ -155,15 +146,14 @@ void multiplexer::register_reading(const socket_manager_ptr& mgr) {
void multiplexer::register_writing(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) {
if (shutting_down_) {
if (mgr->abort_reason()) {
// nop
} else if (mgr->mask() != operation::none) {
if (auto index = index_of(mgr);
index != -1 && mgr->mask_add(operation::write)) {
} else if (!is_idle(mgr->mask())) {
if (auto index = index_of(mgr); index != -1 && mgr->set_write_flag()) {
auto& fd = pollset_[index];
fd.events |= output_mask;
}
} else if (mgr->mask_add(operation::write)) {
} else if (mgr->set_write_flag()) {
add(mgr);
}
} else {
......@@ -192,7 +182,7 @@ void multiplexer::shutdown_reading(const socket_manager_ptr& mgr) {
if (shutting_down_) {
// nop
} else if (auto index = index_of(mgr); index != -1) {
mgr->mask_del(operation::read);
mgr->block_reads();
auto& entry = pollset_[index];
std::ignore = shutdown_read(socket{entry.fd});
entry.events &= ~input_mask;
......@@ -210,7 +200,7 @@ void multiplexer::shutdown_writing(const socket_manager_ptr& mgr) {
if (shutting_down_) {
// nop
} else if (auto index = index_of(mgr); index != -1) {
mgr->mask_del(operation::write);
mgr->block_writes();
auto& entry = pollset_[index];
std::ignore = shutdown_write(socket{entry.fd});
entry.events &= ~output_mask;
......@@ -329,8 +319,9 @@ void multiplexer::set_thread_id() {
void multiplexer::run() {
CAF_LOG_TRACE("");
while (!pollset_.empty())
while (!shutting_down_ || pollset_.size() > 1)
poll_once(true);
close_pipe();
}
void multiplexer::shutdown() {
......@@ -341,16 +332,16 @@ void multiplexer::shutdown() {
// 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)) {
if (mgr->unset_read_flag()) {
auto& fd = pollset_[index_of(mgr)];
fd.events &= ~input_mask;
}
if (mgr->mask() == operation::none)
mgr->block_reads();
if (is_idle(mgr->mask()))
del(i);
else
++i;
}
close_pipe();
} else {
CAF_LOG_DEBUG("push shutdown event to pipe");
write_to_pipe(pollset_updater::code::shutdown,
......@@ -362,20 +353,23 @@ void multiplexer::shutdown() {
short multiplexer::handle(const socket_manager_ptr& mgr, short events,
short revents) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle()));
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id)
<< CAF_ARG(events) << CAF_ARG(revents));
CAF_ASSERT(mgr != nullptr);
bool checkerror = true;
if ((revents & input_mask) != 0) {
// Note: we double-check whether the manager is actually reading because a
// previous action from the pipe may have called shutdown_reading.
if ((revents & input_mask) != 0 && mgr->is_reading()) {
checkerror = false;
if (!mgr->handle_read_event()) {
mgr->mask_del(operation::read);
}
if (!mgr->handle_read_event())
mgr->unset_read_flag();
}
if ((revents & output_mask) != 0) {
// Similar reasoning than before: double-check whether this event should still
// get dispatched.
if ((revents & output_mask) != 0 && mgr->is_writing()) {
checkerror = false;
if (!mgr->handle_write_event()) {
mgr->mask_del(operation::write);
}
if (!mgr->handle_write_event())
mgr->unset_write_flag();
}
if (checkerror && ((revents & error_mask) != 0)) {
if (revents & POLLNVAL)
......@@ -384,24 +378,11 @@ short multiplexer::handle(const socket_manager_ptr& mgr, short events,
mgr->handle_error(sec::socket_disconnected);
else
mgr->handle_error(sec::socket_operation_failed);
mgr->mask_del(operation::read_write);
events = 0;
mgr->block_reads_and_writes();
return 0;
} else {
switch (mgr->mask()) {
case operation::read:
events = input_mask;
break;
case operation::write:
events = output_mask;
break;
case operation::read_write:
events = input_mask | output_mask;
break;
default:
events = 0;
}
return to_bitmask(mgr->mask());
}
return events;
}
void multiplexer::add(socket_manager_ptr mgr) {
......
......@@ -25,34 +25,50 @@ actor_system& socket_manager::system() noexcept {
return parent_->system();
}
bool socket_manager::mask_add(operation flag) noexcept {
CAF_ASSERT(flag != operation::none);
auto x = mask();
if ((x & flag) == flag)
return false;
mask_ = x | flag;
return true;
bool socket_manager::set_read_flag() noexcept {
auto old = mask_;
mask_ = add_read_flag(mask_);
return old != mask_;
}
bool socket_manager::mask_del(operation flag) noexcept {
CAF_ASSERT(flag != operation::none);
auto x = mask();
if ((x & flag) == operation::none)
return false;
mask_ = x & ~flag;
return true;
bool socket_manager::set_write_flag() noexcept {
auto old = mask_;
mask_ = add_write_flag(mask_);
return old != mask_;
}
bool socket_manager::unset_read_flag() noexcept {
auto old = mask_;
mask_ = remove_read_flag(mask_);
return old != mask_;
}
bool socket_manager::unset_write_flag() noexcept {
auto old = mask_;
mask_ = remove_write_flag(mask_);
return old != mask_;
}
void socket_manager::block_reads() noexcept {
mask_ = net::block_reads(mask_);
}
void socket_manager::block_writes() noexcept {
mask_ = net::block_writes(mask_);
}
void socket_manager::block_reads_and_writes() noexcept {
mask_ = operation::shutdown;
}
void socket_manager::register_reading() {
if ((mask() & operation::read) == operation::read)
return;
parent_->register_reading(this);
if (!net::is_reading(mask_) && !is_read_blocked(mask_))
parent_->register_reading(this);
}
void socket_manager::register_writing() {
if ((mask() & operation::write) == operation::write)
return;
parent_->register_writing(this);
if (!net::is_writing(mask_) && !is_write_blocked(mask_))
parent_->register_writing(this);
}
void socket_manager::shutdown_reading() {
......
......@@ -6,7 +6,7 @@
#include "caf/net/basp/header.hpp"
#include "caf/test/dsl.hpp"
#include "net-test.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
......@@ -21,25 +21,26 @@ CAF_TEST(serialization) {
byte_buffer buf;
{
binary_serializer sink{nullptr, buf};
CAF_CHECK(sink.apply(x));
CHECK(sink.apply(x));
}
CAF_CHECK_EQUAL(buf.size(), basp::header_size);
CHECK_EQ(buf.size(), basp::header_size);
auto buf2 = to_bytes(x);
CAF_REQUIRE_EQUAL(buf.size(), buf2.size());
CAF_CHECK(std::equal(buf.begin(), buf.end(), buf2.begin()));
REQUIRE_EQ(buf.size(), buf2.size());
CHECK(std::equal(buf.begin(), buf.end(), buf2.begin()));
basp::header y;
{
binary_deserializer source{nullptr, buf};
CAF_CHECK(source.apply(y));
CHECK(source.apply(y));
}
CAF_CHECK_EQUAL(x, y);
CHECK_EQ(x, y);
auto z = basp::header::from_bytes(buf);
CAF_CHECK_EQUAL(x, z);
CAF_CHECK_EQUAL(y, z);
CHECK_EQ(x, z);
CHECK_EQ(y, z);
}
CAF_TEST(to_string) {
basp::header x{basp::message_type::handshake, 42, 4};
CAF_CHECK_EQUAL(deep_to_string(x),
"caf::net::basp::header(handshake, 42, 4)");
CHECK_EQ(
deep_to_string(x),
"caf::net::basp::header(caf::net::basp::message_type::handshake, 42, 4)");
}
#define CAF_SUITE net.operation
#include "caf/net/operation.hpp"
#include "net-test.hpp"
using namespace caf;
using namespace caf::net;
SCENARIO("add_read_flag adds the read bit unless block_read prevents it") {
GIVEN("a valid operation enum value") {
WHEN("calling add_read_flag") {
THEN("the read bit is set unless the block_read flag is present") {
auto add_rd = [](auto x) { return add_read_flag(x); };
CHECK_EQ(add_rd(operation::none), operation::read);
CHECK_EQ(add_rd(operation::read), operation::read);
CHECK_EQ(add_rd(operation::write), operation::read_write);
CHECK_EQ(add_rd(operation::block_read), operation::block_read);
CHECK_EQ(add_rd(operation::block_write), operation::read_only);
CHECK_EQ(add_rd(operation::read_write), operation::read_write);
CHECK_EQ(add_rd(operation::read_only), operation::read_only);
CHECK_EQ(add_rd(operation::write_only), operation::write_only);
CHECK_EQ(add_rd(operation::shutdown), operation::shutdown);
}
}
}
}
SCENARIO("add_write_flag adds the write bit unless block_write prevents it") {
GIVEN("a valid operation enum value") {
WHEN("calling add_write_flag") {
THEN("the write bit is set unless the block_write flag is present") {
auto add_wr = [](auto x) { return add_write_flag(x); };
CHECK_EQ(add_wr(operation::none), operation::write);
CHECK_EQ(add_wr(operation::read), operation::read_write);
CHECK_EQ(add_wr(operation::write), operation::write);
CHECK_EQ(add_wr(operation::block_read ), operation::write_only);
CHECK_EQ(add_wr(operation::block_write), operation::block_write);
CHECK_EQ(add_wr(operation::read_write), operation::read_write);
CHECK_EQ(add_wr(operation::read_only), operation::read_only);
CHECK_EQ(add_wr(operation::write_only), operation::write_only);
CHECK_EQ(add_wr(operation::shutdown), operation::shutdown);
}
}
}
}
SCENARIO("remove_read_flag erases the read flag") {
GIVEN("a valid operation enum value") {
WHEN("calling remove_read_flag") {
THEN("the read bit is removed if present") {
auto del_rd = [](auto x) { return remove_read_flag(x); };
CHECK_EQ(del_rd(operation::none), operation::none);
CHECK_EQ(del_rd(operation::read), operation::none);
CHECK_EQ(del_rd(operation::write), operation::write);
CHECK_EQ(del_rd(operation::block_read), operation::block_read);
CHECK_EQ(del_rd(operation::block_write), operation::block_write);
CHECK_EQ(del_rd(operation::read_write), operation::write);
CHECK_EQ(del_rd(operation::read_only), operation::block_write);
CHECK_EQ(del_rd(operation::write_only), operation::write_only);
CHECK_EQ(del_rd(operation::shutdown), operation::shutdown);
}
}
}
}
SCENARIO("remove_write_flag erases the write flag") {
GIVEN("a valid operation enum value") {
WHEN("calling remove_write_flag") {
THEN("the write bit is removed if present") {
auto del_wr = [](auto x) { return remove_write_flag(x); };
CHECK_EQ(del_wr(operation::none), operation::none);
CHECK_EQ(del_wr(operation::read), operation::read);
CHECK_EQ(del_wr(operation::write), operation::none);
CHECK_EQ(del_wr(operation::block_read), operation::block_read);
CHECK_EQ(del_wr(operation::block_write), operation::block_write);
CHECK_EQ(del_wr(operation::read_write), operation::read);
CHECK_EQ(del_wr(operation::read_only), operation::read_only);
CHECK_EQ(del_wr(operation::write_only), operation::block_read);
CHECK_EQ(del_wr(operation::shutdown), operation::shutdown);
}
}
}
}
SCENARIO("block_reads removes the read flag and sets the block read flag") {
GIVEN("a valid operation enum value") {
WHEN("calling block_reads") {
THEN("the read bit is removed if present and block_read is set") {
auto block_rd = [](auto x) { return block_reads(x); };
CHECK_EQ(block_rd(operation::none), operation::block_read);
CHECK_EQ(block_rd(operation::read), operation::block_read);
CHECK_EQ(block_rd(operation::write), operation::write_only);
CHECK_EQ(block_rd(operation::block_read), operation::block_read);
CHECK_EQ(block_rd(operation::block_write), operation::shutdown);
CHECK_EQ(block_rd(operation::read_write), operation::write_only);
CHECK_EQ(block_rd(operation::read_only), operation::shutdown);
CHECK_EQ(block_rd(operation::write_only), operation::write_only);
CHECK_EQ(block_rd(operation::shutdown), operation::shutdown);
}
}
}
}
SCENARIO("block_writes removes the write flag and sets the block write flag") {
GIVEN("a valid operation enum value") {
WHEN("calling block_writes") {
THEN("the write bit is removed if present and block_write is set") {
auto block_wr = [](auto x) { return block_writes(x); };
CHECK_EQ(block_wr(operation::none), operation::block_write);
CHECK_EQ(block_wr(operation::read), operation::read_only);
CHECK_EQ(block_wr(operation::write), operation::block_write);
CHECK_EQ(block_wr(operation::block_read), operation::shutdown);
CHECK_EQ(block_wr(operation::block_write), operation::block_write);
CHECK_EQ(block_wr(operation::read_write), operation::read_only);
CHECK_EQ(block_wr(operation::read_only), operation::read_only);
CHECK_EQ(block_wr(operation::write_only), operation::shutdown);
CHECK_EQ(block_wr(operation::shutdown), operation::shutdown);
}
}
}
}
SCENARIO("is-predicates check whether certain flags are present") {
GIVEN("a valid operation enum value") {
WHEN("calling is_reading") {
THEN("the predicate returns true if the read bit is present") {
CHECK_EQ(is_reading(operation::none), false);
CHECK_EQ(is_reading(operation::read), true);
CHECK_EQ(is_reading(operation::write), false);
CHECK_EQ(is_reading(operation::block_read), false);
CHECK_EQ(is_reading(operation::block_write), false);
CHECK_EQ(is_reading(operation::read_write), true);
CHECK_EQ(is_reading(operation::read_only), true);
CHECK_EQ(is_reading(operation::write_only), false);
CHECK_EQ(is_reading(operation::shutdown), false);
}
}
WHEN("calling is_writing") {
THEN("the predicate returns true if the write bit is present") {
CHECK_EQ(is_writing(operation::none), false);
CHECK_EQ(is_writing(operation::read), false);
CHECK_EQ(is_writing(operation::write), true);
CHECK_EQ(is_writing(operation::block_read), false);
CHECK_EQ(is_writing(operation::block_write), false);
CHECK_EQ(is_writing(operation::read_write), true);
CHECK_EQ(is_writing(operation::read_only), false);
CHECK_EQ(is_writing(operation::write_only), true);
CHECK_EQ(is_writing(operation::shutdown), false);
}
}
WHEN("calling is_idle") {
THEN("the predicate returns true when neither reading nor writing") {
CHECK_EQ(is_idle(operation::none), true);
CHECK_EQ(is_idle(operation::read), false);
CHECK_EQ(is_idle(operation::write), false);
CHECK_EQ(is_idle(operation::block_read), true);
CHECK_EQ(is_idle(operation::block_write), true);
CHECK_EQ(is_idle(operation::read_write), false);
CHECK_EQ(is_idle(operation::read_only), false);
CHECK_EQ(is_idle(operation::write_only), false);
CHECK_EQ(is_idle(operation::shutdown), true);
}
}
WHEN("calling is_read_blocked") {
THEN("the predicate returns when blocking reads") {
CHECK_EQ(is_read_blocked(operation::none), false);
CHECK_EQ(is_read_blocked(operation::read), false);
CHECK_EQ(is_read_blocked(operation::write), false);
CHECK_EQ(is_read_blocked(operation::block_read), true);
CHECK_EQ(is_read_blocked(operation::block_write), false);
CHECK_EQ(is_read_blocked(operation::read_write), false);
CHECK_EQ(is_read_blocked(operation::read_only), false);
CHECK_EQ(is_read_blocked(operation::write_only), true);
CHECK_EQ(is_read_blocked(operation::shutdown), true);
}
}
WHEN("calling is_write_blocked") {
THEN("the predicate returns when blocking writes") {
CHECK_EQ(is_write_blocked(operation::none), false);
CHECK_EQ(is_write_blocked(operation::read), false);
CHECK_EQ(is_write_blocked(operation::write), false);
CHECK_EQ(is_write_blocked(operation::block_read), false);
CHECK_EQ(is_write_blocked(operation::block_write), true);
CHECK_EQ(is_write_blocked(operation::read_write), false);
CHECK_EQ(is_write_blocked(operation::read_only), true);
CHECK_EQ(is_write_blocked(operation::write_only), false);
CHECK_EQ(is_write_blocked(operation::shutdown), true);
}
}
}
}
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