SSL non blocking read

conanfile.py

@@ -107,7 +107,7 @@ def set_version(self):
         major_version = re.search(r'set\(USERVER_MAJOR_VERSION (.*)\)', content).group(1).strip()
         minor_version = re.search(r'set\(USERVER_MINOR_VERSION (.*)\)', content).group(1).strip()
 
-        self.version = f'{major_version}.{minor_version}'  # pylint: disable=attribute-defined-outside-init
+        self.version = f'{major_version}.{minor_version}.4'  # pylint: disable=attribute-defined-outside-init
 
     def layout(self):
         cmake_layout(self)

core/include/userver/engine/io/tls_wrapper.hpp

@@ -70,6 +70,14 @@ class [[nodiscard]] TlsWrapper final : public RwBase {
     /// received any more, received bytes count otherwise.
     [[nodiscard]] size_t RecvSome(void* buf, size_t len, Deadline deadline);
 
+    /// @brief Receives up to len bytes from the socket
+    /// @returns
+    /// - nullopt on data absence
+    /// - optional{0} if socket is closed by peer.
+    /// - optional{data_bytes_available} otherwise,
+    ///    1 <= data_bytes_available <= len
+    [[nodiscard]] std::optional<size_t> RecvNoblock(void* buf, size_t len);
+
     /// @brief Receives exactly len bytes from the socket.
     /// @note Can return less than len if socket is closed by peer.
     [[nodiscard]] size_t RecvAll(void* buf, size_t len, Deadline deadline);
@@ -83,6 +91,16 @@ class [[nodiscard]] TlsWrapper final : public RwBase {
     ///   socket extraction if interrupted.
     [[nodiscard]] Socket StopTls(Deadline deadline);
 
+    /// @brief Receives up to len bytes from the stream
+    /// @returns
+    /// - nullopt on data absence
+    /// - optional{0} if socket is closed by peer or len is 0
+    /// - optional{data_bytes_available} otherwise,
+    ///    1 <= data_bytes_available <= len
+    [[nodiscard]] std::optional<size_t> ReadNoblock(void* buf, size_t len) override {
+        return RecvNoblock(buf, len);
+    }
+
     /// @brief Receives at least one byte from the socket.
     /// @returns 0 if connection is closed on one side and no data could be
     /// received any more, received bytes count otherwise.

core/include/userver/server/websocket/server.hpp

@@ -72,6 +72,12 @@ class WebSocketConnection {
 
     virtual ~WebSocketConnection();
 
+    /// Suspends current task until the stream has data available.
+    [[nodiscard]] virtual bool WaitReadable(engine::Deadline) = 0;
+
+    /// Suspends current task until the socket can accept more data.
+    [[nodiscard]] virtual bool WaitWriteable(engine::Deadline) = 0;
+
     /// @brief Read a message from websocket, handling pings under the hood.
     /// @param message input message
     /// @throws engine::io::IoException in case of socket errors

core/src/engine/io/tls_wrapper.cpp

@@ -329,6 +329,45 @@ class TlsWrapper::Impl {
         }
     }
 
+    template <typename SslIoFunc>
+    std::optional<size_t> PerformSslIoOptional(
+        SslIoFunc&& io_func,
+        void* buf,
+        size_t len,
+        const char* context
+    ) {
+        UASSERT(ssl);
+        if (!len) return 0;
+
+        char* const begin = static_cast<char*>(buf);
+        size_t chunk_size = 0;
+
+        bio_data.current_deadline = Deadline::Passed();
+
+        const int io_ret = io_func(ssl.get(), begin, len, &chunk_size);
+        if (io_ret == 1) {
+            return chunk_size;
+        }
+
+        const int ssl_error = SSL_get_error(ssl.get(), io_ret);
+        switch (ssl_error) {
+            case SSL_ERROR_WANT_READ:
+            case SSL_ERROR_WANT_WRITE:
+                return std::nullopt;
+
+            case SSL_ERROR_ZERO_RETURN:
+                return 0;
+
+            case SSL_ERROR_SYSCALL:
+            case SSL_ERROR_SSL:
+                ssl.reset();
+            [[fallthrough]];
+
+            default:
+                throw TlsException(crypto::FormatSslError(std::string{context} + " failed"));
+        }
+    }
+
     template <typename SslIoFunc>
     size_t PerformSslIo(
         SslIoFunc&& io_func,
@@ -584,10 +623,7 @@ bool TlsWrapper::IsValid() const { return impl_->ssl && !impl_->is_in_shutdown;
 
 bool TlsWrapper::WaitReadable(Deadline deadline) {
     impl_->CheckAlive();
-    char buf = 0;
-    return impl_->PerformSslIo(
-        &SSL_peek_ex, &buf, 1, impl::TransferMode::kOnce, InterruptAction::kPass, deadline, "WaitReadable"
-    );
+    return impl_->bio_data.socket.WaitReadable(deadline);
 }
 
 bool TlsWrapper::WaitWriteable(Deadline deadline) {
@@ -609,6 +645,13 @@ size_t TlsWrapper::RecvAll(void* buf, size_t len, Deadline deadline) {
     );
 }
 
+std::optional<size_t> TlsWrapper::RecvNoblock(void* buf, size_t len) {
+    impl_->CheckAlive();
+    return impl_->PerformSslIoOptional(
+        &SSL_read_ex, buf, len, "RecvNoblock"
+    );
+}
+
 size_t TlsWrapper::SendAll(const void* buf, size_t len, Deadline deadline) {
     impl_->CheckAlive();
     return impl_->PerformSslIo(

core/src/engine/io/tls_wrapper_test.cpp

@@ -790,4 +790,53 @@ UTEST(TlsWrapper, PeerDisconnect) {
     server_task.Get();
 }
 
+UTEST(TlsWrapper, RecvNoblock) {
+    const auto test_deadline = Deadline::FromDuration(utest::kMaxTestWaitTime);
+
+    TcpListener tcp_listener;
+    auto [server, client] = tcp_listener.MakeSocketPair(test_deadline);
+
+    auto server_task = engine::AsyncNoSpan(
+        [test_deadline](auto&& server) {
+            try {
+                auto tls_server = io::TlsWrapper::StartTlsServer(
+                    std::forward<decltype(server)>(server),
+                    crypto::LoadCertificatesChainFromString(cert),
+                    crypto::PrivateKey::LoadFromString(key),
+                    test_deadline
+                );
+                EXPECT_EQ(1, tls_server.SendAll("1", 1, test_deadline));
+
+                char c = 0;
+                std::optional<size_t> read_bytes;
+                while (true) {
+                    read_bytes = tls_server.RecvNoblock(&c, 1);
+                    if (read_bytes.has_value()) {
+                        EXPECT_EQ(1, read_bytes.value());
+                        EXPECT_EQ('2', c);
+                        return;
+                    }
+
+                    ASSERT_FALSE(test_deadline.IsReached());
+                }
+            }
+            catch (const std::exception& e) {
+                LOG_ERROR() << e;
+                FAIL() << e.what();
+            }
+        },
+        std::move(server)
+    );
+
+    auto tls_client = io::TlsWrapper::StartTlsClient(std::move(client), {}, test_deadline);
+
+    char c = 0;
+    size_t read_bytes = tls_client.RecvAll(&c, 1, test_deadline);
+    EXPECT_EQ(1, read_bytes);
+    EXPECT_EQ('1', c);
+
+    EXPECT_EQ(1, tls_client.SendAll("2", 1, test_deadline));
+    server_task.Get();
+}
+
 USERVER_NAMESPACE_END

core/src/server/websocket/server.cpp

@@ -184,6 +184,16 @@ class WebSocketConnectionImpl final : public WebSocketConnection {
         }
     }
 
+    bool WaitReadable(engine::Deadline deadline) override {
+        UASSERT(io);
+        return io->GetReadableBase().WaitReadable(deadline);
+    }
+
+    bool WaitWriteable(engine::Deadline deadline) override {
+        UASSERT(io);
+        return io->GetWritableBase().WaitWriteable(deadline);
+    }
+
     void Recv(Message& msg) override {
         const auto ok = RecvImpl(msg, /*do_not_wait_for_message_header*/ false);
         UASSERT(ok);