Merge branch '2025/10/optimistic'

Author: Sjors

src/sv2/connman.cpp

@@ -113,37 +113,27 @@ bool Sv2Connman::EventNewConnectionAccepted(NodeId node_id,
     return true;
 }
 
-void Sv2Connman::EventReadyToSend(NodeId node_id, bool& cancel_recv)
+std::pair<size_t, bool> Sv2Connman::SendMessagesAsBytes(Sv2Client& client)
 {
-    AssertLockNotHeld(m_clients_mutex);
-
-    auto client{WITH_LOCK(m_clients_mutex, return GetClientById(node_id);)};
-    if (client == nullptr) {
-        cancel_recv = true;
-        return;
-    }
-
-    LOCK(client->cs_send);
-    auto it = client->m_send_messages.begin();
+    auto it = client.m_send_messages.begin();
     std::optional<bool> expected_more;
-
     size_t total_sent = 0;
 
-    while(true) {
-        if (it != client->m_send_messages.end()) {
+    while (true) {
+        if (it != client.m_send_messages.end()) {
             // If possible, move one message from the send queue to the transport.
             // This fails when there is an existing message still being sent,
             // or when the handshake has not yet completed.
             //
             // Wrap Sv2NetMsg inside CSerializedNetMsg for transport
             CSerializedNetMsg net_msg{*it};
-            if (client->m_transport->SetMessageToSend(net_msg)) {
+            if (client.m_transport->SetMessageToSend(net_msg)) {
                 ++it;
             }
         }
 
-        const auto& [data, more, _m_message_type] = client->m_transport->GetBytesToSend(/*have_next_message=*/it != client->m_send_messages.end());
-
+        const auto& [data, more, _m_message_type] =
+            client.m_transport->GetBytesToSend(/*have_next_message=*/it != client.m_send_messages.end());
 
         // We rely on the 'more' value returned by GetBytesToSend to correctly predict whether more
         // bytes are still to be sent, to correctly set the MSG_MORE flag. As a sanity check,
@@ -155,29 +145,53 @@ void Sv2Connman::EventReadyToSend(NodeId node_id, bool& cancel_recv)
         std::string errmsg;
 
         if (!data.empty()) {
-            LogPrintLevel(BCLog::SV2, BCLog::Level::Trace, "Send %d bytes to client id=%zu\n",
-                            data.size() - total_sent, node_id);
+            LogPrintLevel(BCLog::SV2, BCLog::Level::Trace,
+                          "Send %d bytes to client id=%zu\n", data.size(), client.m_id);
 
-            sent = SendBytes(node_id, data, more, errmsg);
+            sent = SendBytes(client.m_id, data, more, errmsg);
         }
 
         if (sent > 0) {
-            client->m_transport->MarkBytesSent(sent);
+            client.m_transport->MarkBytesSent(sent);
+            total_sent += sent;
             if (static_cast<size_t>(sent) != data.size()) {
                 // could not send full message; stop sending more
                 break;
             }
         } else {
             if (sent < 0) {
-                LogDebug(BCLog::NET, "socket send error for peer=%d: %s\n", node_id, errmsg);
-                CloseConnection(node_id);
+                LogDebug(BCLog::NET, "socket send error for peer=%d: %s\n",
+                         client.m_id, errmsg);
+                CloseConnection(client.m_id);
             }
             break;
         }
     }
 
     // Clear messages that have been handed to transport from the queue
-    client->m_send_messages.erase(client->m_send_messages.begin(), it);
+    client.m_send_messages.erase(client.m_send_messages.begin(), it);
+
+    return {total_sent, expected_more.value_or(false)};
+}
+
+void Sv2Connman::TryOptimisticSend(Sv2Client& client)
+{
+    AssertLockHeld(client.cs_send);
+    SendMessagesAsBytes(client);
+}
+
+void Sv2Connman::EventReadyToSend(NodeId node_id, bool& cancel_recv)
+{
+    AssertLockNotHeld(m_clients_mutex);
+
+    auto client{WITH_LOCK(m_clients_mutex, return GetClientById(node_id);)};
+    if (client == nullptr) {
+        cancel_recv = true;
+        return;
+    }
+
+    LOCK(client->cs_send);
+    const auto [total_sent, more_pending] = SendMessagesAsBytes(*client);
 
     // If both receiving and (non-optimistic) sending were possible, we first attempt
     // sending. If that succeeds, but does not fully drain the send queue, do not
@@ -186,9 +200,7 @@ void Sv2Connman::EventReadyToSend(NodeId node_id, bool& cancel_recv)
     // sending actually succeeded to make sure progress is always made; otherwise a
     // deadlock would be possible when both sides have data to send, but neither is
     // receiving.
-    //
-    // TODO: decide if this is useful for Sv2
-    cancel_recv = total_sent > 0; // && more;
+    cancel_recv = total_sent > 0 && more_pending;
 }
 
 void Sv2Connman::EventGotData(Id id, std::span<const uint8_t> data)

src/sv2/connman.h

@@ -10,6 +10,8 @@
 #include <sv2/transport.h>
 #include <pubkey.h>
 
+#include <utility>
+
 namespace {
     /*
      * Supported Stratum v2 subprotocols
@@ -144,6 +146,9 @@ class Sv2Connman : SockMan
 
     void DisconnectFlagged() EXCLUSIVE_LOCKS_REQUIRED(m_clients_mutex);
 
+    std::pair<size_t, bool> SendMessagesAsBytes(Sv2Client& client)
+        EXCLUSIVE_LOCKS_REQUIRED(client.cs_send);
+
     /**
      * Create a `Sv2Client` object and add it to the `m_sv2_clients` member.
      * @param[in] node_id Id of the newly accepted connection.
@@ -211,6 +216,13 @@ class Sv2Connman : SockMan
      */
     void ProcessSv2Message(const node::Sv2NetMsg& sv2_header, Sv2Client& client);
 
+    /**
+     * Attempt to flush the send queue immediately, without waiting for the
+     * socket layer to signal readiness. Falls back to the regular send path if
+     * the socket would block.
+     */
+    void TryOptimisticSend(Sv2Client& client) EXCLUSIVE_LOCKS_REQUIRED(client.cs_send);
+
     std::shared_ptr<Sv2Client> GetClientById(NodeId node_id) const EXCLUSIVE_LOCKS_REQUIRED(m_clients_mutex);
 
     using Sv2ClientFn = std::function<void(Sv2Client&)>;

src/sv2/template_provider.cpp

@@ -437,6 +437,7 @@ void Sv2TemplateProvider::RequestTransactionData(Sv2Client& client, node::Sv2Req
                     client.m_id);
     LOCK(client.cs_send);
     client.m_send_messages.emplace_back(request_tx_data_success);
+    m_connman->TryOptimisticSend(client);
 }
 
 void Sv2TemplateProvider::SubmitSolution(node::Sv2SubmitSolutionMsg solution)
@@ -561,8 +562,6 @@ bool Sv2TemplateProvider::SendWork(Sv2Client& client, uint64_t template_id, Bloc
                                         template_id,
                                         future_template};
 
-    // TODO: use optimistic send instead of adding to the queue
-
     LogPrintLevel(BCLog::SV2, BCLog::Level::Debug, "Send 0x71 NewTemplate id=%lu future=%d to client id=%zu\n", template_id, future_template, client.m_id);
     {
         LOCK(client.cs_send);
@@ -573,6 +572,8 @@ bool Sv2TemplateProvider::SendWork(Sv2Client& client, uint64_t template_id, Bloc
             LogPrintLevel(BCLog::SV2, BCLog::Level::Debug, "Send 0x72 SetNewPrevHash to client id=%zu\n", client.m_id);
             client.m_send_messages.emplace_back(new_prev_hash);
         }
+
+        m_connman->TryOptimisticSend(client);
     }
 
     CAmount total_fees{0};