hedged_request.hpp

Go to the documentation of this file.

#pragma once
 
/// @file userver/storages/redis/hedged_request.hpp
/// @brief
/// Classes and functions for performing hedged requests.
///
/// Use MakeHedgedRedisRequest method to perform hedged redis request.
///
/// Example(Sync):
/// auto result =
///     ::redis::MakeHedgedRedisRequest<storages::redis::RequestHGet>(
///         redis_client_shared_ptr,
///         &storages::redis::Client::Hget,
///         redis_cc, hedging_settings,
///         key, field);
/// Example(Async):
/// auto future =
///     ::redis::MakeHedgedRedisRequestAsync<storages::redis::RequestHGet>(
///         redis_client_shared_ptr,
///         &storages::redis::Client::Hget,
///         redis_cc, hedging_settings,
///         key, field);
/// auto result = future.Get();
///
 
#include <optional>
 
#include <userver/storages/redis/client.hpp>
#include <userver/storages/redis/command_control.hpp>
#include <userver/utils/hedged_request.hpp>
 
USERVER_NAMESPACE_BEGIN
 
namespace redis {
namespace impl {
 
template <typename RedisRequestType>
struct RedisRequestStrategy {
public:
    using RequestType = RedisRequestType;
    using ReplyType = typename RequestType::Reply;
    using GenF = std::function<std::optional<RequestType>(int)>;
 
    explicit RedisRequestStrategy(GenF gen_callback) : gen_callback_(std::move(gen_callback)) {}
 
    RedisRequestStrategy(RedisRequestStrategy&& other) noexcept = default;
 
    /// @{
    /// Methods needed by HedgingStrategy
    std::optional<RequestType> Create(size_t try_count) { return gen_callback_(try_count); }
    std::optional<std::chrono::milliseconds> ProcessReply(RequestType&& request) {
        reply_ = std::move(request).Get();
        return std::nullopt;
    }
    std::optional<ReplyType> ExtractReply() { return std::move(reply_); }
    void Finish(RequestType&&) {}
    /// @}
 
private:
    GenF gen_callback_;
    std::optional<ReplyType> reply_;
};
 
}  // namespace impl
 
template <typename RedisRequestType>
using HedgedRedisRequest = utils::hedging::HedgedRequestFuture<impl::RedisRequestStrategy<RedisRequestType>>;
 
template <
    typename RedisRequestType,
    typename... Args,
    typename M = RedisRequestType (storages::redis::Client::*)(Args..., const redis::CommandControl&)>
HedgedRedisRequest<RedisRequestType> MakeHedgedRedisRequestAsync(
    std::shared_ptr<storages::redis::Client> redis,
    M method,
    const redis::CommandControl& cc,
    utils::hedging::HedgingSettings hedging_settings,
    Args... args
) {
    auto gen_request = [redis, method, cc{std::move(cc)}, args_tuple{std::tuple(std::move(args)...)}](int try_count
                       ) mutable -> std::optional<RedisRequestType> {
        cc.retry_counter = try_count;
        cc.max_retries = 1;  ///< We do retries ourselves
 
        return std::apply(
            [redis, method, cc](auto&&... args) { return (redis.get()->*method)(args..., cc); }, args_tuple
        );
    };
    return utils::hedging::HedgeRequestAsync(
        impl::RedisRequestStrategy<RedisRequestType>(std::move(gen_request)), hedging_settings
    );
}
 
template <typename RedisRequestType>
using HedgedRedisRequest = utils::hedging::HedgedRequestFuture<impl::RedisRequestStrategy<RedisRequestType>>;
 
template <
    typename RedisRequestType,
    typename... Args,
    typename M = RedisRequestType (storages::redis::Client::*)(Args..., const redis::CommandControl&)>
std::optional<typename RedisRequestType::Reply> MakeHedgedRedisRequest(
    std::shared_ptr<storages::redis::Client> redis,
    M method,
    const redis::CommandControl& cc,
    utils::hedging::HedgingSettings hedging_settings,
    Args... args
) {
    auto gen_request = [redis, method, cc{std::move(cc)}, args_tuple{std::tuple(std::move(args)...)}](int try_count
                       ) mutable -> std::optional<RedisRequestType> {
        cc.retry_counter = try_count;
        cc.max_retries = 1;  ///< We do retries ourselves
 
        return std::apply(
            [redis, method, cc](auto&&... args) { return (redis.get()->*method)(args..., cc); }, args_tuple
        );
    };
    return utils::hedging::HedgeRequest(
        impl::RedisRequestStrategy<RedisRequestType>(std::move(gen_request)), hedging_settings
    );
}
 
/// Same as MakeHedgedRedisRequest but accepting a vector of argument tuples
/// instead of a single set of arguments. And return vector of replies
/// corresponding to input args.
template <
    typename RedisRequestType,
    typename... Args,
    typename M = RedisRequestType (storages::redis::Client::*)(Args..., const redis::CommandControl&)>
std::vector<std::optional<typename RedisRequestType::Reply>> MakeBulkHedgedRedisRequest(
    std::shared_ptr<storages::redis::Client> redis,
    M method,
    const redis::CommandControl& cc,
    utils::hedging::HedgingSettings hedging_settings,
    std::vector<std::tuple<Args...>> args
) {
    std::vector<redis::impl::RedisRequestStrategy<RedisRequestType>> strategies;
    strategies.reserve(args.size());
 
    for (auto&& arg : args) {
        auto gen_request = [redis, method, cc{std::move(cc)}, args_tuple{std::move(arg)}](int try_count
                           ) mutable -> std::optional<RedisRequestType> {
            cc.retry_counter = try_count;
            cc.max_retries = 1;  ///< We do retries ourselves
 
            return std::apply(
                [redis, method, cc](auto&&... args) { return (redis.get()->*method)(args..., cc); }, args_tuple
            );
        };
        strategies.emplace_back(std::move(gen_request));
    }
 
    return utils::hedging::HedgeRequestsBulk(std::move(strategies), hedging_settings);
}
 
/// Same as MakeHedgedRedisRequestAsync but accepting a vector of argument
/// tuples instead of a single set of arguments. And return vector of replies
/// corresponding to input args.
template <
    typename RedisRequestType,
    typename... Args,
    typename M = RedisRequestType (storages::redis::Client::*)(Args..., const redis::CommandControl&)>
utils::hedging::HedgedRequestBulkFuture<redis::impl::RedisRequestStrategy<RedisRequestType>>
MakeBulkHedgedRedisRequestAsync(
    std::shared_ptr<storages::redis::Client> redis,
    M method,
    const redis::CommandControl& cc,
    utils::hedging::HedgingSettings hedging_settings,
    std::vector<std::tuple<Args...>> args
) {
    using RequestStrategy = redis::impl::RedisRequestStrategy<RedisRequestType>;
    std::vector<RequestStrategy> strategies;
    strategies.reserve(args.size());
 
    for (auto&& arg : args) {
        auto gen_request = [redis, method, cc{std::move(cc)}, args_tuple{std::move(arg)}](int try_count
                           ) mutable -> std::optional<RedisRequestType> {
            cc.retry_counter = try_count;
            cc.max_retries = 1;  ///< We do retries ourselves
 
            return std::apply(
                [redis, method, cc](auto&&... args) { return (redis.get()->*method)(args..., cc); }, args_tuple
            );
        };
        strategies.emplace_back(std::move(gen_request));
    }
 
    return utils::hedging::HedgeRequestsBulkAsync<RequestStrategy>(std::move(strategies), hedging_settings);
}
 
}  // namespace redis
 
USERVER_NAMESPACE_END