From 48a0e9fef727ee56ddf8ef1ba1bcbc042579a1d4 Mon Sep 17 00:00:00 2001 From: Sam Atkins Date: Thu, 16 Apr 2026 14:13:12 +0100 Subject: [PATCH] RequestServer: Queue websocket writes under curl backpressure Add the same connect timeout used by regular requests to curl-backed websockets, buffer unsent websocket bytes, and resume them from a write notifier instead of retrying curl_easy_send() in a tight loop. This prevents CPU spins on CURLE_AGAIN and keeps large websocket sends from hanging when the peer applies backpressure. --- Services/RequestServer/WebSocketImplCurl.cpp | 69 +++++++++++++++++--- Services/RequestServer/WebSocketImplCurl.h | 4 ++ 2 files changed, 65 insertions(+), 8 deletions(-) diff --git a/Services/RequestServer/WebSocketImplCurl.cpp b/Services/RequestServer/WebSocketImplCurl.cpp index 78c61a9de7..f81ea831fe 100644 --- a/Services/RequestServer/WebSocketImplCurl.cpp +++ b/Services/RequestServer/WebSocketImplCurl.cpp @@ -4,12 +4,15 @@ * SPDX-License-Identifier: BSD-2-Clause */ +#include #include #include #include namespace RequestServer { +static constexpr long s_connect_timeout_seconds = 90L; + NonnullRefPtr WebSocketImplCurl::create(CURLM* multi_handle) { return adopt_ref(*new WebSocketImplCurl(multi_handle)); @@ -24,6 +27,8 @@ WebSocketImplCurl::~WebSocketImplCurl() { if (m_read_notifier) m_read_notifier->close(); + if (m_write_notifier) + m_write_notifier->close(); if (m_error_notifier) m_error_notifier->close(); @@ -64,6 +69,7 @@ void WebSocketImplCurl::connect(WebSocket::ConnectionInfo const& info) auto const& url = info.url(); set_option(CURLOPT_URL, url.to_byte_string().characters()); set_option(CURLOPT_PORT, url.port_or_default()); + set_option(CURLOPT_CONNECTTIMEOUT, s_connect_timeout_seconds); if (auto root_certs = info.root_certificates_path(); root_certs.has_value()) set_option(CURLOPT_CAINFO, root_certs->characters()); @@ -127,15 +133,17 @@ ErrorOr WebSocketImplCurl::read_line(size_t) bool WebSocketImplCurl::send(ReadonlyBytes bytes) { - size_t sent = 0; - CURLcode result = CURLE_OK; - do { - sent = 0; - result = curl_easy_send(m_easy_handle, bytes.data(), bytes.size(), &sent); - bytes = bytes.slice(sent); - } while (bytes.size() > 0 && (result == CURLE_OK || result == CURLE_AGAIN)); + if (auto const error = m_pending_write_buffer.try_append(bytes); error.is_error()) { + dbgln("Failed to queue WebSocket write: {}", error.error()); + on_connection_error(); + return false; + } - return result == CURLE_OK; + if (flush_pending_write_buffer()) + return true; + + on_connection_error(); + return false; } bool WebSocketImplCurl::eof() @@ -149,6 +157,10 @@ void WebSocketImplCurl::discard_connection() m_read_notifier->close(); m_read_notifier = nullptr; } + if (m_write_notifier) { + m_write_notifier->close(); + m_write_notifier = nullptr; + } if (m_error_notifier) { m_error_notifier->close(); m_error_notifier = nullptr; @@ -158,6 +170,8 @@ void WebSocketImplCurl::discard_connection() curl_easy_cleanup(m_easy_handle); m_easy_handle = nullptr; } + m_pending_write_buffer.clear(); + m_pending_write_buffer_offset = 0; } void WebSocketImplCurl::read_from_socket() @@ -202,6 +216,39 @@ void WebSocketImplCurl::read_from_socket() on_ready_to_read(); } +bool WebSocketImplCurl::flush_pending_write_buffer() +{ + while (m_pending_write_buffer_offset < m_pending_write_buffer.size()) { + auto pending_bytes = m_pending_write_buffer.bytes().slice(m_pending_write_buffer_offset); + + size_t sent = 0; + auto const result = curl_easy_send(m_easy_handle, pending_bytes.data(), pending_bytes.size(), &sent); + if (result != CURLE_OK && result != CURLE_AGAIN) { + dbgln("Failed to send to WebSocket: {}", curl_easy_strerror(result)); + return false; + } + + m_pending_write_buffer_offset += sent; + if (sent > 0) + continue; + + if (result == CURLE_AGAIN) { + if (m_write_notifier) + m_write_notifier->set_enabled(true); + return true; + } + + dbgln("Failed to make progress sending to WebSocket"); + return false; + } + + m_pending_write_buffer.clear(); + m_pending_write_buffer_offset = 0; + if (m_write_notifier) + m_write_notifier->set_enabled(false); + return true; +} + bool WebSocketImplCurl::did_connect() { curl_socket_t socket_fd = CURL_SOCKET_BAD; @@ -213,6 +260,12 @@ bool WebSocketImplCurl::did_connect() m_read_notifier->on_activation = [this] { read_from_socket(); }; + m_write_notifier = Core::Notifier::construct(socket_fd, Core::Notifier::Type::Write); + m_write_notifier->on_activation = [this] { + if (!flush_pending_write_buffer()) + on_connection_error(); + }; + m_write_notifier->set_enabled(false); m_error_notifier = Core::Notifier::construct(socket_fd, Core::Notifier::Type::Error | Core::Notifier::Type::HangUp); m_error_notifier->on_activation = [this] { on_connection_error(); diff --git a/Services/RequestServer/WebSocketImplCurl.h b/Services/RequestServer/WebSocketImplCurl.h index 098c25d924..496f4dbe8a 100644 --- a/Services/RequestServer/WebSocketImplCurl.h +++ b/Services/RequestServer/WebSocketImplCurl.h @@ -38,13 +38,17 @@ private: explicit WebSocketImplCurl(CURLM*); void read_from_socket(); + bool flush_pending_write_buffer(); CURLM* m_multi_handle { nullptr }; CURL* m_easy_handle { nullptr }; RefPtr m_read_notifier; + RefPtr m_write_notifier; RefPtr m_error_notifier; Vector m_curl_string_lists; AllocatingMemoryStream m_read_buffer; + ByteBuffer m_pending_write_buffer; + size_t m_pending_write_buffer_offset { 0 }; }; }