Commit a1bdf3c9 authored by Dominik Charousset's avatar Dominik Charousset

Ensure only one ASIO async operation is pending

parent b79a5b38
...@@ -142,7 +142,8 @@ public: ...@@ -142,7 +142,8 @@ public:
ack_writes_(false), ack_writes_(false),
fd_(ref.service()), fd_(ref.service()),
backend_(ref), backend_(ref),
rd_buf_ready_(false) { rd_buf_ready_(false),
async_read_pending_(false) {
configure_read(receive_policy::at_most(1024)); configure_read(receive_policy::at_most(1024));
} }
...@@ -230,6 +231,7 @@ public: ...@@ -230,6 +231,7 @@ public:
/// Activates the stream. /// Activates the stream.
void activate(stream_manager* mgr) { void activate(stream_manager* mgr) {
reading_ = true;
read_loop(mgr); read_loop(mgr);
} }
...@@ -254,7 +256,8 @@ private: ...@@ -254,7 +256,8 @@ private:
} }
void read_loop(manager_ptr mgr) { void read_loop(manager_ptr mgr) {
reading_ = true; if (async_read_pending_)
return;
if (rd_buf_ready_) { if (rd_buf_ready_) {
rd_buf_ready_ = false; rd_buf_ready_ = false;
if (read_one(mgr.get(), rd_buf_.size())) if (read_one(mgr.get(), rd_buf_.size()))
...@@ -262,6 +265,7 @@ private: ...@@ -262,6 +265,7 @@ private:
return; return;
} }
auto cb = [=](const error_code& ec, size_t read_bytes) mutable { auto cb = [=](const error_code& ec, size_t read_bytes) mutable {
async_read_pending_ = false;
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
if (!ec) { if (!ec) {
// bail out early in case broker passivated stream in the meantime // bail out early in case broker passivated stream in the meantime
...@@ -273,15 +277,16 @@ private: ...@@ -273,15 +277,16 @@ private:
}; };
switch (rd_flag_) { switch (rd_flag_) {
case receive_policy_flag::exactly: case receive_policy_flag::exactly:
if (rd_buf_.size() < rd_size_) { if (rd_buf_.size() < rd_size_)
rd_buf_.resize(rd_size_); rd_buf_.resize(rd_size_);
} async_read_pending_ = true;
boost::asio::async_read(fd_, boost::asio::buffer(rd_buf_, rd_size_), boost::asio::async_read(fd_, boost::asio::buffer(rd_buf_, rd_size_),
cb); cb);
break; break;
case receive_policy_flag::at_most: case receive_policy_flag::at_most:
if (rd_buf_.size() < rd_size_) if (rd_buf_.size() < rd_size_)
rd_buf_.resize(rd_size_); rd_buf_.resize(rd_size_);
async_read_pending_ = true;
fd_.async_read_some(boost::asio::buffer(rd_buf_, rd_size_), cb); fd_.async_read_some(boost::asio::buffer(rd_buf_, rd_size_), cb);
break; break;
case receive_policy_flag::at_least: { case receive_policy_flag::at_least: {
...@@ -321,9 +326,11 @@ private: ...@@ -321,9 +326,11 @@ private:
} }
void collect_data(manager_ptr mgr, size_t collected_bytes) { void collect_data(manager_ptr mgr, size_t collected_bytes) {
async_read_pending_ = true;
fd_.async_read_some(boost::asio::buffer(rd_buf_.data() + collected_bytes, fd_.async_read_some(boost::asio::buffer(rd_buf_.data() + collected_bytes,
rd_buf_.size() - collected_bytes), rd_buf_.size() - collected_bytes),
[=](const error_code& ec, size_t nb) mutable { [=](const error_code& ec, size_t nb) mutable {
async_read_pending_ = false;
CAF_LOG_TRACE(CAF_ARG(nb)); CAF_LOG_TRACE(CAF_ARG(nb));
if (!ec) { if (!ec) {
auto sum = collected_bytes + nb; auto sum = collected_bytes + nb;
...@@ -339,17 +346,42 @@ private: ...@@ -339,17 +346,42 @@ private:
}); });
} }
/// Set if read loop was started by user and unset if passivate is called.
bool reading_; bool reading_;
/// Set on flush, also indicates that an async_write is pending.
bool writing_; bool writing_;
/// Stores whether user requested ACK messages for async writes.
bool ack_writes_; bool ack_writes_;
/// TCP socket for this connection.
Socket fd_; Socket fd_;
/// Configures how chunk sizes are calculated.
receive_policy_flag rd_flag_; receive_policy_flag rd_flag_;
/// Minimum, maximum, or exact size of a chunk, depending on `rd_flag_`.
size_t rd_size_; size_t rd_size_;
/// Input buffer.
buffer_type rd_buf_; buffer_type rd_buf_;
/// Output buffer for ASIO.
buffer_type wr_buf_; buffer_type wr_buf_;
/// Swapped with `wr_buf_` before next write. Users write into this buffer
/// as long as `wr_buf_` is processed by ASIO.
buffer_type wr_offline_buf_; buffer_type wr_offline_buf_;
/// Reference to our I/O backend.
asio_multiplexer& backend_; asio_multiplexer& backend_;
/// Signalizes that a scribe was passivated while an async read was pending.
bool rd_buf_ready_; bool rd_buf_ready_;
/// Makes sure no more than one async_read is pending at any given time
bool async_read_pending_;
}; };
/// An acceptor is responsible for accepting incoming connections. /// An acceptor is responsible for accepting incoming connections.
...@@ -370,7 +402,8 @@ public: ...@@ -370,7 +402,8 @@ public:
backend_(am), backend_(am),
accept_fd_(io), accept_fd_(io),
fd_valid_(false), fd_valid_(false),
fd_(io) { fd_(io),
async_accept_pending_(false) {
// nop // nop
} }
...@@ -407,6 +440,7 @@ public: ...@@ -407,6 +440,7 @@ public:
/// Starts the accept loop. /// Starts the accept loop.
void activate(manager_type* mgr) { void activate(manager_type* mgr) {
accepting_ = true;
accept_loop(mgr); accept_loop(mgr);
} }
...@@ -429,7 +463,8 @@ private: ...@@ -429,7 +463,8 @@ private:
} }
void accept_loop(manager_ptr mgr) { void accept_loop(manager_ptr mgr) {
accepting_ = true; if (async_accept_pending_)
return;
// accept "cached" connection first // accept "cached" connection first
if (fd_valid_) { if (fd_valid_) {
fd_valid_ = false; fd_valid_ = false;
...@@ -437,8 +472,10 @@ private: ...@@ -437,8 +472,10 @@ private:
accept_loop(std::move(mgr)); accept_loop(std::move(mgr));
return; return;
} }
async_accept_pending_ = true;
accept_fd_.async_accept(fd_, [=](const error_code& ec) mutable { accept_fd_.async_accept(fd_, [=](const error_code& ec) mutable {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
async_accept_pending_ = false;
if (!ec) { if (!ec) {
// if broker has passivated this in the meantime, cache fd_ for later // if broker has passivated this in the meantime, cache fd_ for later
if (!accepting_) { if (!accepting_) {
...@@ -458,6 +495,8 @@ private: ...@@ -458,6 +495,8 @@ private:
SocketAcceptor accept_fd_; SocketAcceptor accept_fd_;
bool fd_valid_; bool fd_valid_;
socket_type fd_; socket_type fd_;
/// Makes sure no more than one async_accept is pending at any given time
bool async_accept_pending_;
}; };
} // namesapce network } // namesapce network
......
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