Commit 317f972e authored by Dominik Charousset's avatar Dominik Charousset

Ignore incoming data on sockets about to be closed

Fixes a bug in the multiplexer that could cause brokers to receive data on a
connection handle that has been closed but is not yet removed from the
multiplexer's event loop.
parent 1e63cdaa
...@@ -313,6 +313,18 @@ class default_multiplexer : public multiplexer { ...@@ -313,6 +313,18 @@ class default_multiplexer : public multiplexer {
event_handler* ptr; event_handler* ptr;
}; };
struct event_less {
inline bool operator()(native_socket lhs, const event& rhs) const {
return lhs < rhs.fd;
}
inline bool operator()(const event& lhs, native_socket rhs) const {
return lhs.fd < rhs;
}
inline bool operator()(const event& lhs, const event& rhs) const {
return lhs.fd < rhs.fd;
}
};
connection_handle add_tcp_scribe(broker*, default_socket_acceptor&& sock); connection_handle add_tcp_scribe(broker*, default_socket_acceptor&& sock);
connection_handle add_tcp_scribe(broker*, native_socket fd) override; connection_handle add_tcp_scribe(broker*, native_socket fd) override;
...@@ -357,11 +369,8 @@ class default_multiplexer : public multiplexer { ...@@ -357,11 +369,8 @@ class default_multiplexer : public multiplexer {
CAF_LOG_TRACE(CAF_TARG(op, static_cast<int>) CAF_LOG_TRACE(CAF_TARG(op, static_cast<int>)
<< ", " << CAF_ARG(fd) << ", " CAF_ARG(ptr) << ", " << CAF_ARG(fd) << ", " CAF_ARG(ptr)
<< ", " << CAF_ARG(old_bf)); << ", " << CAF_ARG(old_bf));
auto event_less = [](const event& e, native_socket arg) -> bool {
return e.fd < arg;
};
auto last = m_events.end(); auto last = m_events.end();
auto i = std::lower_bound(m_events.begin(), last, fd, event_less); auto i = std::lower_bound(m_events.begin(), last, fd, event_less{});
if (i != last && i->fd == fd) { if (i != last && i->fd == fd) {
CAF_REQUIRE(ptr == i->ptr); CAF_REQUIRE(ptr == i->ptr);
// squash events together // squash events together
...@@ -391,6 +400,8 @@ class default_multiplexer : public multiplexer { ...@@ -391,6 +400,8 @@ class default_multiplexer : public multiplexer {
void handle(const event& event); void handle(const event& event);
bool socket_had_rd_shutdown_event(native_socket fd);
void handle_socket_event(native_socket fd, int mask, event_handler* ptr); void handle_socket_event(native_socket fd, int mask, event_handler* ptr);
void close_pipe(); void close_pipe();
......
...@@ -602,11 +602,24 @@ void default_multiplexer::close_pipe() { ...@@ -602,11 +602,24 @@ void default_multiplexer::close_pipe() {
del(operation::read, m_pipe.first, nullptr); del(operation::read, m_pipe.first, nullptr);
} }
bool default_multiplexer::socket_had_rd_shutdown_event(native_socket fd) {
auto last = m_events.end();
auto i = std::lower_bound(m_events.begin(), last, fd, event_less{});
if (i != last && i->fd == fd) {
// socket is about to be shut down for read if
// its new bitmask does not have the input_mask flag
return (i->mask & input_mask) == 0;
}
return false;
}
void default_multiplexer::handle_socket_event(native_socket fd, int mask, void default_multiplexer::handle_socket_event(native_socket fd, int mask,
event_handler* ptr) { event_handler* ptr) {
CAF_LOG_TRACE(CAF_ARG(fd) << ", " << CAF_ARG(mask)); CAF_LOG_TRACE(CAF_ARG(fd) << ", " << CAF_ARG(mask));
bool checkerror = true; bool checkerror = true;
if (mask & input_mask) { // ignore read events if a previous event caused
// this socket to be shut down for reading
if ((mask & input_mask) && !socket_had_rd_shutdown_event(fd)) {
checkerror = false; checkerror = false;
if (ptr) { if (ptr) {
ptr->handle_event(operation::read); ptr->handle_event(operation::read);
......
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