From 528fd79c1f01590ade4272e798581fc187028341 Mon Sep 17 00:00:00 2001 From: Denis Razinkin Date: Fri, 1 Aug 2025 19:05:46 +0300 Subject: [PATCH 1/6] TlsWrapper: non blocking recv --- conanfile.py | 2 +- .../include/userver/engine/io/tls_wrapper.hpp | 18 +++++++ core/src/engine/io/tls_wrapper.cpp | 48 +++++++++++++++++++ core/src/engine/io/tls_wrapper_test.cpp | 2 + 4 files changed, 69 insertions(+), 1 deletion(-) diff --git a/conanfile.py b/conanfile.py index 049147397b24..6f1e0fa38076 100644 --- a/conanfile.py +++ b/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}.2' # pylint: disable=attribute-defined-outside-init def layout(self): cmake_layout(self) diff --git a/core/include/userver/engine/io/tls_wrapper.hpp b/core/include/userver/engine/io/tls_wrapper.hpp index aaecf95080c1..41984f0966fe 100644 --- a/core/include/userver/engine/io/tls_wrapper.hpp +++ b/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 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. + /// - optional{data_bytes_available} otherwise, + /// 1 <= data_bytes_available <= len + [[nodiscard]] std::optional 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. diff --git a/core/src/engine/io/tls_wrapper.cpp b/core/src/engine/io/tls_wrapper.cpp index ccd4890aef31..5a441b7116d0 100644 --- a/core/src/engine/io/tls_wrapper.cpp +++ b/core/src/engine/io/tls_wrapper.cpp @@ -329,6 +329,46 @@ class TlsWrapper::Impl { } } + template + std::optional PerformSslIoOptional( + SslIoFunc&& io_func, + void* buf, + size_t len, + const char* context + ) { + UASSERT(ssl); + if (!len) return 0; + + char* const begin = static_cast(buf); + char* const end = begin + len; + size_t chunk_size = 0; + + bio_data.current_deadline = Deadline::Passed(); + + const int io_ret = io_func(ssl.get(), begin, end - begin, &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 size_t PerformSslIo( SslIoFunc&& io_func, @@ -595,6 +635,7 @@ bool TlsWrapper::WaitWriteable(Deadline deadline) { return impl_->bio_data.socket.WaitWriteable(deadline); } + size_t TlsWrapper::RecvSome(void* buf, size_t len, Deadline deadline) { impl_->CheckAlive(); return impl_->PerformSslIo( @@ -609,6 +650,13 @@ size_t TlsWrapper::RecvAll(void* buf, size_t len, Deadline deadline) { ); } +std::optional 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( diff --git a/core/src/engine/io/tls_wrapper_test.cpp b/core/src/engine/io/tls_wrapper_test.cpp index 68cfa30eb1a9..45119af39bf0 100644 --- a/core/src/engine/io/tls_wrapper_test.cpp +++ b/core/src/engine/io/tls_wrapper_test.cpp @@ -790,4 +790,6 @@ UTEST(TlsWrapper, PeerDisconnect) { server_task.Get(); } +// TODO tests with RecvNoblock + USERVER_NAMESPACE_END From 64d2648b7bd559eb5df1b2e0c087952dcae1a405 Mon Sep 17 00:00:00 2001 From: Denis Razinkin Date: Wed, 6 Aug 2025 17:10:08 +0300 Subject: [PATCH 2/6] test --- core/src/engine/io/tls_wrapper_test.cpp | 50 ++++++++++++++++++++++++- 1 file changed, 49 insertions(+), 1 deletion(-) diff --git a/core/src/engine/io/tls_wrapper_test.cpp b/core/src/engine/io/tls_wrapper_test.cpp index 45119af39bf0..2c609ddef6fa 100644 --- a/core/src/engine/io/tls_wrapper_test.cpp +++ b/core/src/engine/io/tls_wrapper_test.cpp @@ -790,6 +790,54 @@ UTEST(TlsWrapper, PeerDisconnect) { server_task.Get(); } -// TODO tests with RecvNoblock +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(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 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; + std::optional read_bytes = tls_client.RecvAll(&c, 1, test_deadline); + EXPECT_TRUE(read_bytes.has_value()); + EXPECT_EQ(1, read_bytes.value()); + EXPECT_EQ('1', c); + + EXPECT_EQ(1, tls_client.SendAll("2", 1, test_deadline)); + server_task.Get(); +} USERVER_NAMESPACE_END From 7199103967658fd22665154c3f7f2bd22755c7a6 Mon Sep 17 00:00:00 2001 From: Denis Razinkin Date: Thu, 7 Aug 2025 17:49:06 +0300 Subject: [PATCH 3/6] TlsWrapper & ws server: WaitReadable implementation --- core/include/userver/engine/io/tls_wrapper.hpp | 2 +- core/include/userver/server/websocket/server.hpp | 3 +++ core/src/engine/io/tls_wrapper.cpp | 9 ++------- core/src/engine/io/tls_wrapper_test.cpp | 5 ++--- core/src/server/websocket/server.cpp | 5 +++++ 5 files changed, 13 insertions(+), 11 deletions(-) diff --git a/core/include/userver/engine/io/tls_wrapper.hpp b/core/include/userver/engine/io/tls_wrapper.hpp index 41984f0966fe..00c214001e68 100644 --- a/core/include/userver/engine/io/tls_wrapper.hpp +++ b/core/include/userver/engine/io/tls_wrapper.hpp @@ -94,7 +94,7 @@ class [[nodiscard]] TlsWrapper final : public RwBase { /// @brief Receives up to len bytes from the stream /// @returns /// - nullopt on data absence - /// - optional{0} if socket is closed by peer. + /// - 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 ReadNoblock(void* buf, size_t len) override { diff --git a/core/include/userver/server/websocket/server.hpp b/core/include/userver/server/websocket/server.hpp index 3fc8332ccd9a..3672209fe374 100644 --- a/core/include/userver/server/websocket/server.hpp +++ b/core/include/userver/server/websocket/server.hpp @@ -72,6 +72,9 @@ class WebSocketConnection { virtual ~WebSocketConnection(); + /// Suspends current task until the stream has data available. + [[nodiscard]] virtual bool WaitReadable(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 diff --git a/core/src/engine/io/tls_wrapper.cpp b/core/src/engine/io/tls_wrapper.cpp index 5a441b7116d0..f4178af43814 100644 --- a/core/src/engine/io/tls_wrapper.cpp +++ b/core/src/engine/io/tls_wrapper.cpp @@ -340,12 +340,11 @@ class TlsWrapper::Impl { if (!len) return 0; char* const begin = static_cast(buf); - char* const end = begin + len; size_t chunk_size = 0; bio_data.current_deadline = Deadline::Passed(); - const int io_ret = io_func(ssl.get(), begin, end - begin, &chunk_size); + const int io_ret = io_func(ssl.get(), begin, len, &chunk_size); if (io_ret == 1) { return chunk_size; } @@ -624,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) { @@ -635,7 +631,6 @@ bool TlsWrapper::WaitWriteable(Deadline deadline) { return impl_->bio_data.socket.WaitWriteable(deadline); } - size_t TlsWrapper::RecvSome(void* buf, size_t len, Deadline deadline) { impl_->CheckAlive(); return impl_->PerformSslIo( diff --git a/core/src/engine/io/tls_wrapper_test.cpp b/core/src/engine/io/tls_wrapper_test.cpp index 2c609ddef6fa..edf24dbd5e01 100644 --- a/core/src/engine/io/tls_wrapper_test.cpp +++ b/core/src/engine/io/tls_wrapper_test.cpp @@ -831,9 +831,8 @@ UTEST(TlsWrapper, RecvNoblock) { auto tls_client = io::TlsWrapper::StartTlsClient(std::move(client), {}, test_deadline); char c = 0; - std::optional read_bytes = tls_client.RecvAll(&c, 1, test_deadline); - EXPECT_TRUE(read_bytes.has_value()); - EXPECT_EQ(1, read_bytes.value()); + 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)); diff --git a/core/src/server/websocket/server.cpp b/core/src/server/websocket/server.cpp index 610ece9941b2..a885639473ad 100644 --- a/core/src/server/websocket/server.cpp +++ b/core/src/server/websocket/server.cpp @@ -184,6 +184,11 @@ class WebSocketConnectionImpl final : public WebSocketConnection { } } + bool WaitReadable(engine::Deadline deadline) override { + UASSERT(io); + return io->GetReadableBase().WaitReadable(deadline); + } + void Recv(Message& msg) override { const auto ok = RecvImpl(msg, /*do_not_wait_for_message_header*/ false); UASSERT(ok); From 0085e01248321af63e57f5c9881b12c4396aff3e Mon Sep 17 00:00:00 2001 From: Denis Razinkin Date: Thu, 7 Aug 2025 17:51:14 +0300 Subject: [PATCH 4/6] hotfix 3 --- conanfile.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/conanfile.py b/conanfile.py index 6f1e0fa38076..d4dba4b44fb7 100644 --- a/conanfile.py +++ b/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}.2' # pylint: disable=attribute-defined-outside-init + self.version = f'{major_version}.{minor_version}.3' # pylint: disable=attribute-defined-outside-init def layout(self): cmake_layout(self) From 525af66b2443108c26587e8b087a01c2f50da03e Mon Sep 17 00:00:00 2001 From: Denis Razinkin Date: Fri, 8 Aug 2025 13:23:07 +0300 Subject: [PATCH 5/6] WaitWriteable for websocket --- core/include/userver/server/websocket/server.hpp | 3 +++ core/src/server/websocket/server.cpp | 5 +++++ 2 files changed, 8 insertions(+) diff --git a/core/include/userver/server/websocket/server.hpp b/core/include/userver/server/websocket/server.hpp index 3672209fe374..956975c4bb0b 100644 --- a/core/include/userver/server/websocket/server.hpp +++ b/core/include/userver/server/websocket/server.hpp @@ -75,6 +75,9 @@ class 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 diff --git a/core/src/server/websocket/server.cpp b/core/src/server/websocket/server.cpp index a885639473ad..9251c709ba42 100644 --- a/core/src/server/websocket/server.cpp +++ b/core/src/server/websocket/server.cpp @@ -189,6 +189,11 @@ class WebSocketConnectionImpl final : public WebSocketConnection { 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); From cac6dca37445563f8206f62a70de81042a5efa5d Mon Sep 17 00:00:00 2001 From: Denis Razinkin Date: Fri, 8 Aug 2025 13:25:54 +0300 Subject: [PATCH 6/6] version 4 --- conanfile.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/conanfile.py b/conanfile.py index d4dba4b44fb7..f48bd300ee4d 100644 --- a/conanfile.py +++ b/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}.3' # 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)