Skip to content

Removes handshaker in favor of asio::deferred #291

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 1 commit into
base: develop
Choose a base branch
from
Draft

Removes handshaker in favor of asio::deferred #291

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 29 additions & 5 deletions include/boost/redis/connection.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
#include <boost/redis/operation.hpp>
#include <boost/redis/request.hpp>
#include <boost/redis/resp3/type.hpp>
#include <boost/redis/response.hpp>
#include <boost/redis/usage.hpp>

#include <boost/asio/any_completion_handler.hpp>
Expand Down Expand Up @@ -67,7 +68,6 @@ struct connection_impl {
Executor,
void(system::error_code, std::size_t)>;
using health_checker_type = detail::health_checker<Executor>;
using resp3_handshaker_type = detail::resp3_handshaker<Executor>;
using exec_notifier_type = asio::experimental::channel<
Executor,
void(system::error_code, std::size_t)>;
Expand All @@ -81,12 +81,13 @@ struct connection_impl {
timer_type reconnect_timer_; // to wait the reconnection period
receive_channel_type receive_channel_;
health_checker_type health_checker_;
resp3_handshaker_type handshaker_;

config cfg_;
multiplexer mpx_;
connection_logger logger_;
read_buffer read_buffer_;
request hello_req_;
generic_response hello_resp_;

using executor_type = Executor;

Expand Down Expand Up @@ -326,6 +327,27 @@ class run_op {

using order_t = std::array<std::size_t, 5>;

static system::error_code on_hello(connection_impl<Executor>& conn, system::error_code ec)
{
conn.logger_.on_hello(ec, conn.hello_resp_);
ec = check_hello_response(ec, conn.hello_resp_);
if (ec) {
conn.cancel(operation::run);
}
return ec;
}

template <class CompletionToken>
auto handshaker(CompletionToken&& token)
{
return conn_->async_exec(
conn_->hello_req_,
any_adapter(conn_->hello_resp_),
asio::deferred([&conn = *this->conn_](system::error_code hello_ec, std::size_t) {
return asio::deferred.values(on_hello(conn, hello_ec));
}))(std::forward<CompletionToken>(token));
}

template <class CompletionToken>
auto reader(CompletionToken&& token)
{
Expand Down Expand Up @@ -392,6 +414,9 @@ class run_op {
return;
}

// Set up the hello request, as it only depends on the config
push_hello(conn_->cfg_, conn_->hello_req_);

for (;;) {
// Try to connect
BOOST_ASIO_CORO_YIELD
Expand All @@ -401,14 +426,15 @@ class run_op {
if (!ec) {
conn_->read_buffer_.clear();
conn_->mpx_.reset();
clear_response(conn_->hello_resp_);

// Note: Order is important here because the writer might
// trigger an async_write before the async_hello thereby
// causing an authentication problem.
BOOST_ASIO_CORO_YIELD
asio::experimental::make_parallel_group(
[this](auto token) {
return conn_->handshaker_.async_hello(*conn_, token);
return this->handshaker(token);
},
[this](auto token) {
return conn_->health_checker_.async_ping(*conn_, token);
Expand Down Expand Up @@ -606,7 +632,6 @@ class basic_connection {
{
impl_->cfg_ = cfg;
impl_->health_checker_.set_config(cfg);
impl_->handshaker_.set_config(cfg);
impl_->read_buffer_.set_config({cfg.read_buffer_append_size, cfg.max_read_size});

return asio::async_compose<CompletionToken, void(system::error_code)>(
Expand Down Expand Up @@ -908,7 +933,6 @@ class basic_connection {
executor_type,
void(system::error_code, std::size_t)>;
using health_checker_type = detail::health_checker<Executor>;
using resp3_handshaker_type = detail::resp3_handshaker<executor_type>;

auto use_ssl() const noexcept { return impl_->cfg_.use_ssl; }

Expand Down
109 changes: 12 additions & 97 deletions include/boost/redis/detail/resp3_handshaker.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,110 +4,25 @@
* accompanying file LICENSE.txt)
*/

#ifndef BOOST_REDIS_RUNNER_HPP
#define BOOST_REDIS_RUNNER_HPP
#ifndef BOOST_REDIS_RESP3_HANDSHAKER_HPP
#define BOOST_REDIS_RESP3_HANDSHAKER_HPP

#include <boost/redis/config.hpp>
#include <boost/redis/detail/connection_logger.hpp>
#include <boost/redis/error.hpp>
#include <boost/redis/operation.hpp>
#include <boost/redis/request.hpp>
#include <boost/redis/response.hpp>

#include <boost/asio/compose.hpp>
#include <boost/asio/coroutine.hpp>

#include <string>

namespace boost::redis::detail {

void push_hello(config const& cfg, request& req);

// TODO: Can we avoid this whole function whose only purpose is to
// check for an error in the hello response and complete with an error
// so that the parallel group that starts it can exit?
template <class Handshaker, class ConnectionImpl>
struct hello_op {
Handshaker* handshaker_ = nullptr;
ConnectionImpl* conn_ = nullptr;
asio::coroutine coro_{};

template <class Self>
void operator()(Self& self, system::error_code ec = {}, std::size_t = 0)
{
BOOST_ASIO_CORO_REENTER(coro_)
{
handshaker_->add_hello();

BOOST_ASIO_CORO_YIELD
conn_->async_exec(
handshaker_->hello_req_,
any_adapter(handshaker_->hello_resp_),
std::move(self));
conn_->logger_.on_hello(ec, handshaker_->hello_resp_);

if (ec) {
conn_->cancel(operation::run);
self.complete(ec);
return;
}

if (handshaker_->has_error_in_response()) {
conn_->cancel(operation::run);
self.complete(error::resp3_hello);
return;
}

self.complete({});
}
}
};

template <class Executor>
class resp3_handshaker {
public:
void set_config(config const& cfg) { cfg_ = cfg; }

template <class ConnectionImpl, class CompletionToken>
auto async_hello(ConnectionImpl& conn, CompletionToken token)
{
return asio::async_compose<CompletionToken, void(system::error_code)>(
hello_op<resp3_handshaker, ConnectionImpl>{this, &conn},
token,
conn);
}

private:
template <class, class> friend struct hello_op;

void add_hello()
{
hello_req_.clear();
if (hello_resp_.has_value())
hello_resp_.value().clear();
push_hello(cfg_, hello_req_);
}

bool has_error_in_response() const noexcept
{
if (!hello_resp_.has_value())
return true;

auto f = [](auto const& e) {
switch (e.data_type) {
case resp3::type::simple_error:
case resp3::type::blob_error: return true;
default: return false;
}
};

return std::any_of(std::cbegin(hello_resp_.value()), std::cend(hello_resp_.value()), f);
}

request hello_req_;
generic_response hello_resp_;
config cfg_;
};
void push_hello(config const& cfg, request& req); // TODO: rename
system::error_code check_hello_response(system::error_code io_ec, const generic_response&);
// TODO: logging should be here, too
inline void clear_response(generic_response& res)
{
if (res.has_value())
res->clear();
else
res.emplace();
}

} // namespace boost::redis::detail

Expand Down
21 changes: 21 additions & 0 deletions include/boost/redis/impl/resp3_handshaker.ipp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ namespace boost::redis::detail {

void push_hello(config const& cfg, request& req)
{
req.clear();
if (!cfg.username.empty() && !cfg.password.empty() && !cfg.clientname.empty())
req.push("HELLO", "3", "AUTH", cfg.username, cfg.password, "SETNAME", cfg.clientname);
else if (cfg.password.empty() && cfg.clientname.empty())
Expand All @@ -23,4 +24,24 @@ void push_hello(config const& cfg, request& req)
req.push("SELECT", cfg.database_index.value());
}

system::error_code check_hello_response(system::error_code io_ec, const generic_response& resp)
{
if (io_ec)
return io_ec;

if (resp.has_error())
return error::resp3_hello;

auto f = [](auto const& e) {
switch (e.data_type) {
case resp3::type::simple_error:
case resp3::type::blob_error: return true;
default: return false;
}
};

bool has_error = std::any_of(resp->cbegin(), resp->cend(), f);
return has_error ? error::resp3_hello : system::error_code();
}

} // namespace boost::redis::detail
Loading