feat redis: use userver sockets in mocks

Tests: протестировано локально и в CI
commit_hash:a7b16bee12544cf1eb0f5da1308f1da43c12ba8f

Author: apolukhin

redis/src/storages/redis/impl/mock_server_test.cpp

@@ -7,6 +7,7 @@
 #include <fmt/ranges.h>
 #include <boost/algorithm/string.hpp>
 
+#include <userver/engine/task/cancel.hpp>
 #include <userver/utest/assert_macros.hpp>
 #include <userver/utils/numeric_cast.hpp>
 #include <userver/utils/text.hpp>
@@ -20,15 +21,16 @@ const std::string kCrlf = "\r\n";
 }  // namespace
 
 MockRedisServerBase::MockRedisServerBase(int port)
-    : acceptor_(io_service_)
+    : client_tasks_()
 {
-    acceptor_.open(io::ip::tcp::v4());
-    const boost::asio::ip::tcp::acceptor::reuse_address option(true);
-    acceptor_.set_option(option);
-    acceptor_.bind(io::ip::tcp::endpoint(io::ip::tcp::v4(), port));
-    acceptor_.listen();
+    auto addr = engine::io::Sockaddr::MakeIPv4LoopbackAddress();
+    addr.SetPort(port);
+    listener_ = engine::io::Socket{engine::io::AddrDomain::kInet, engine::io::SocketType::kStream};
+    listener_.Bind(addr);
+    port_ = listener_.Getsockname().Port();
+    listener_.Listen();
 
-    thread_ = std::thread(&MockRedisServerBase::Work, this);
+    listener_task_ = engine::AsyncNoTracing([this] { AcceptLoop(); });
 }
 
 MockRedisServerBase::~MockRedisServerBase() { Stop(); }
@@ -45,19 +47,14 @@ void MockRedisServerBase::SendReplyData(ConnectionPtr connection, const storages
     SendReply(connection, ReplyDataToRedisProto(reply_data));
 }
 
-int MockRedisServerBase::GetPort() const { return acceptor_.local_endpoint().port(); }
+int MockRedisServerBase::GetPort() const { return port_; }
 
-void MockRedisServerBase::Stop() {
-    io_service_.stop();
-    if (thread_.joinable()) {
-        thread_.join();
-    }
-}
+void MockRedisServerBase::Stop() { listener_task_.SyncCancel(); }
 
 void MockRedisServerBase::SendReply(ConnectionPtr connection, const std::string& reply) {
     LOG_DEBUG() << "reply: " << reply;
-    // TODO: async?
-    io::write(connection->socket, io::buffer(reply.c_str(), reply.size()));
+    const auto size = connection->socket.SendAll(reply.data(), reply.size(), {});
+    UASSERT(size == reply.size());
 }
 
 std::string MockRedisServerBase::ReplyDataToRedisProto(const storages::redis::ReplyData& reply_data) {
@@ -84,62 +81,48 @@ std::string MockRedisServerBase::ReplyDataToRedisProto(const storages::redis::Re
     }
 }
 
-void MockRedisServerBase::Accept() {
-    auto connection = std::make_shared<Connection>(io_service_);
-    connection
-        ->reader = std::unique_ptr<redisReader, decltype(&redisReaderFree)>(redisReaderCreate(), &redisReaderFree);
-    acceptor_.async_accept(connection->socket, [connection, this](auto item) {
-        OnAccept(connection, std::move(item));
-        Accept();
-    });
-}
-
-void MockRedisServerBase::Work() {
-    Accept();
-    UEXPECT_NO_THROW(io_service_.run());
-}
-
-void MockRedisServerBase::OnAccept(ConnectionPtr connection, boost::system::error_code ec) {
-    LOG_DEBUG() << "accept(2): " << ec;
-    OnConnected(connection);
-    DoRead(connection);
-}
-
-void MockRedisServerBase::OnRead(ConnectionPtr connection, boost::system::error_code ec, size_t count) {
-    LOG_DEBUG() << "read " << ec << " count=" << count;
-    if (ec) {
-        LOG_DEBUG() << "read(2) error: " << ec;
-        OnDisconnected(connection);
-        connection->socket.close();
-        return;
-    }
+void MockRedisServerBase::AcceptLoop() {
+    while (!engine::current_task::ShouldCancel()) {
+        engine::io::Socket client_socket = listener_.Accept({});
+        if (engine::current_task::ShouldCancel()) {
+            return;
+        }
 
-    auto ret = redisReaderFeed(connection->reader.get(), connection->data.data(), count);
-    if (ret != REDIS_OK) {
-        throw std::runtime_error("redisReaderFeed() returned error: " + std::string(connection->reader->errstr));
+        auto connection = std::make_shared<Connection>(std::move(client_socket));
+        connection
+            ->reader = std::unique_ptr<redisReader, decltype(&redisReaderFree)>(redisReaderCreate(), &redisReaderFree);
+        OnConnected(connection);
+        client_tasks_.AsyncDetach("mock-redis-client", [this, connection] { HandleConnection(connection); });
     }
+}
 
-    void* hiredis_reply = nullptr;
-    while (redisReaderGetReply(connection->reader.get(), &hiredis_reply) == REDIS_OK && hiredis_reply) {
-        auto reply = std::make_shared<
-            storages::redis::Reply>("", static_cast<redisReply*>(hiredis_reply), storages::redis::ReplyStatus::kOk);
-        LOG_DEBUG() << "command: " << reply->data.ToDebugString();
+void MockRedisServerBase::HandleConnection(ConnectionPtr connection) {
+    while (!engine::current_task::ShouldCancel()) {
+        const auto count = connection->socket.RecvSome(connection->data.data(), connection->data.size(), {});
+        LOG_DEBUG() << "read count=" << count;
+        if (count == 0) {
+            LOG_DEBUG() << "read: connection closed";
+            OnDisconnected(connection);
+            connection->socket.Close();
+            return;
+        }
 
-        OnCommand(connection, reply);
-        freeReplyObject(hiredis_reply);
-        hiredis_reply = nullptr;
-    }
+        auto ret = redisReaderFeed(connection->reader.get(), connection->data.data(), count);
+        if (ret != REDIS_OK) {
+            throw std::runtime_error("redisReaderFeed() returned error: " + std::string(connection->reader->errstr));
+        }
 
-    DoRead(connection);
-}
+        void* hiredis_reply = nullptr;
+        while (redisReaderGetReply(connection->reader.get(), &hiredis_reply) == REDIS_OK && hiredis_reply) {
+            auto reply = std::make_shared<
+                storages::redis::Reply>("", static_cast<redisReply*>(hiredis_reply), storages::redis::ReplyStatus::kOk);
+            LOG_DEBUG() << "command: " << reply->data.ToDebugString();
 
-void MockRedisServerBase::DoRead(ConnectionPtr connection) {
-    connection->socket.async_read_some(
-        io::buffer(connection->data),
-        [connection, this](boost::system::error_code error_code, size_t count) {
-            OnRead(connection, error_code, count);
+            OnCommand(connection, reply);
+            freeReplyObject(hiredis_reply);
+            hiredis_reply = nullptr;
         }
-    );
+    }
 }
 
 MockRedisServer::~MockRedisServer() { Stop(); }
@@ -361,7 +344,7 @@ void MockRedisServer::RegisterHandlerFunc(
     const std::vector<std::string>& args_prefix,
     HandlerFunc handler
 ) {
-    const std::lock_guard<std::mutex> lock(mutex_);
+    const std::lock_guard lock{mutex_};
     AddHandlerFunc(handlers_[boost::algorithm::to_lower_copy(cmd)], args_prefix, std::move(handler));
 }
 
@@ -410,16 +393,16 @@ MockRedisServer::HandlerPtr MockRedisServer::DoRegisterTimeoutHandler(
 }
 
 size_t MockRedisServer::Handler::GetReplyCount() const {
-    const std::lock_guard<std::mutex> lock(mutex_);
+    const std::lock_guard lock{mutex_};
     return reply_count_;
 }
 
 void MockRedisServer::Handler::AccountReply() {
     {
-        const std::lock_guard<std::mutex> lock(mutex_);
+        const std::lock_guard lock{mutex_};
         ++reply_count_;
     }
-    cv_.notify_one();
+    cv_.NotifyOne();
 }
 
 MockRedisServer::CommonMasterSlaveInfo::CommonMasterSlaveInfo(

redis/src/storages/redis/impl/mock_server_test.hpp

@@ -1,45 +1,38 @@
 #pragma once
 #include <gtest/gtest.h>
 
+#include <atomic>
 #include <chrono>
-#include <condition_variable>
 #include <memory>
 #include <mutex>
 #include <stdexcept>
 #include <unordered_map>
 
 #include <hiredis/hiredis.h>
 
+#include <userver/concurrent/background_task_storage.hpp>
+#include <userver/engine/async.hpp>
+#include <userver/engine/condition_variable.hpp>
+#include <userver/engine/io/sockaddr.hpp>
+#include <userver/engine/io/socket.hpp>
+#include <userver/engine/mutex.hpp>
+#include <userver/engine/task/task.hpp>
 #include <userver/logging/log.hpp>
 
 #include <storages/redis/impl/redis.hpp>
 #include <userver/storages/redis/base.hpp>
 #include <userver/storages/redis/reply.hpp>
 
-#ifdef __clang__
-#pragma clang diagnostic push
-#pragma clang diagnostic ignored "-Wimplicit-fallthrough"
-#endif
-#include <boost/asio.hpp>
-#ifdef __clang__
-#pragma clang diagnostic pop
-#endif
-
-#include <boost/version.hpp>
-
 USERVER_NAMESPACE_BEGIN
 
-namespace io = boost::asio;
-
 class MockRedisServerBase {
 public:
     struct Connection {
-        template <class IoService>
-        Connection(IoService& ios)
-            : socket(ios)
+        explicit Connection(engine::io::Socket socket)
+            : socket(std::move(socket))
         {}
 
-        io::ip::tcp::socket socket;
+        engine::io::Socket socket;
         std::array<char, 1024> data{};
         std::unique_ptr<redisReader, decltype(&redisReaderFree)> reader{nullptr, &redisReaderFree};
     };
@@ -63,25 +56,18 @@ class MockRedisServerBase {
     virtual void OnCommand(ConnectionPtr /*connection*/, std::shared_ptr<storages::redis::Reply> cmd) {
         LOG_DEBUG() << "Got command: " << cmd->data.ToDebugString();
     }
-    void Accept();
 
 private:
     static std::string ReplyDataToRedisProto(const storages::redis::ReplyData& reply_data);
-    void Work();
-
-    void OnAccept(ConnectionPtr connection, boost::system::error_code ec);
-    void OnRead(ConnectionPtr connection, boost::system::error_code ec, size_t count);
-    void DoRead(ConnectionPtr connection);
+    void AcceptLoop();
+    void HandleConnection(ConnectionPtr connection);
 
     void SendReply(ConnectionPtr connection, const std::string& reply);
 
-#if BOOST_VERSION >= 107400
-    io::io_context io_service_;
-#else
-    io::io_service io_service_;
-#endif
-    io::ip::tcp::acceptor acceptor_;
-    std::thread thread_;
+    std::uint16_t port_{};
+    engine::io::Socket listener_;
+    concurrent::BackgroundTaskStorage client_tasks_;
+    engine::Task listener_task_;
 };
 
 class MockRedisServer : public MockRedisServerBase {
@@ -168,7 +154,7 @@ class MockRedisServer : public MockRedisServerBase {
     );
 
     const std::string description_;
-    std::mutex mutex_;
+    engine::Mutex mutex_;
     std::unordered_map<std::string, HandlerNode> handlers_;
     HandlerPtr ping_handler_;
 };
@@ -206,8 +192,8 @@ class MockRedisServer::Handler {
 private:
     void AccountReply();
 
-    mutable std::mutex mutex_;
-    std::condition_variable cv_;
+    mutable engine::Mutex mutex_;
+    engine::ConditionVariable cv_;
     size_t reply_count_{0};
 };
 
@@ -238,8 +224,8 @@ struct MockRedisServer::SlaveInfo : public MockRedisServer::CommonMasterSlaveInf
 
 template <typename Rep, typename Period>
 bool MockRedisServer::Handler::WaitForFirstReply(const std::chrono::duration<Rep, Period>& duration) {
-    std::unique_lock<std::mutex> lock(mutex_);
-    return cv_.wait_for(lock, duration, [&] { return reply_count_; });
+    std::unique_lock lock{mutex_};
+    return cv_.WaitFor(lock, duration, [&] { return reply_count_; });
 }
 
 template <typename Rep, typename Period>