diff --git a/conanfile.py b/conanfile.py index 049147397b24..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}' # 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) diff --git a/core/include/userver/engine/io/tls_wrapper.hpp b/core/include/userver/engine/io/tls_wrapper.hpp index aaecf95080c1..00c214001e68 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 or len is 0 + /// - 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/include/userver/server/websocket/server.hpp b/core/include/userver/server/websocket/server.hpp index 3fc8332ccd9a..956975c4bb0b 100644 --- a/core/include/userver/server/websocket/server.hpp +++ b/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 diff --git a/core/src/engine/io/tls_wrapper.cpp b/core/src/engine/io/tls_wrapper.cpp index ccd4890aef31..f4178af43814 100644 --- a/core/src/engine/io/tls_wrapper.cpp +++ b/core/src/engine/io/tls_wrapper.cpp @@ -329,6 +329,45 @@ 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); + 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 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 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..edf24dbd5e01 100644 --- a/core/src/engine/io/tls_wrapper_test.cpp +++ b/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(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; + 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 diff --git a/core/src/server/websocket/server.cpp b/core/src/server/websocket/server.cpp index 610ece9941b2..9251c709ba42 100644 --- a/core/src/server/websocket/server.cpp +++ b/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);