Merge pull request #7669

679d055 Remove payload copy in all outgoing p2p messages (Lee Clagett)

6 files changed, 127 insertions, 43 deletions

diff --git a/tests/fuzz/levin.cpp b/tests/fuzz/levin.cpp
index 78b7b6863..209a75221 100644
--- a/tests/fuzz/levin.cpp
+++ b/tests/fuzz/levin.cpp
@@ -68,13 +68,13 @@ namespace
     {
     }
 
-    virtual int invoke(int command, const epee::span<const uint8_t> in_buff, epee::byte_slice& buff_out, test_levin_connection_context& context)
+    virtual int invoke(int command, const epee::span<const uint8_t> in_buff, epee::byte_stream& buff_out, test_levin_connection_context& context)
     {
       m_invoke_counter.inc();
       boost::unique_lock<boost::mutex> lock(m_mutex);
       m_last_command = command;
       m_last_in_buf = std::string((const char*)in_buff.data(), in_buff.size());
-      buff_out = m_invoke_out_buf.clone();
+      buff_out.write(epee::to_span(m_invoke_out_buf));
       return m_return_code;
     }
 
diff --git a/tests/net_load_tests/net_load_tests.h b/tests/net_load_tests/net_load_tests.h
index baab07d31..59eef7bd1 100644
--- a/tests/net_load_tests/net_load_tests.h
+++ b/tests/net_load_tests/net_load_tests.h
@@ -67,7 +67,7 @@ namespace net_load_tests
     {
     }
 
-    virtual int invoke(int command, const epee::span<const uint8_t> in_buff, epee::byte_slice& buff_out, test_connection_context& context)
+    virtual int invoke(int command, const epee::span<const uint8_t> in_buff, epee::byte_stream& buff_out, test_connection_context& context)
     {
       //m_invoke_counter.inc();
       //std::unique_lock<std::mutex> lock(m_mutex);
diff --git a/tests/unit_tests/epee_boosted_tcp_server.cpp b/tests/unit_tests/epee_boosted_tcp_server.cpp
index 457c05c15..84fc0a29b 100644
--- a/tests/unit_tests/epee_boosted_tcp_server.cpp
+++ b/tests/unit_tests/epee_boosted_tcp_server.cpp
@@ -153,7 +153,7 @@ TEST(test_epee_connection, test_lifetime)
       delay(delay),
       on_connection_close_f(on_connection_close_f)
     {}
-    virtual int invoke(int, const epee::span<const uint8_t>, epee::byte_slice&, context_t&) override { epee::misc_utils::sleep_no_w(delay); return {}; }
+    virtual int invoke(int, const epee::span<const uint8_t>, epee::byte_stream&, context_t&) override { epee::misc_utils::sleep_no_w(delay); return {}; }
     virtual int notify(int, const epee::span<const uint8_t>, context_t&) override { return {}; }
     virtual void callback(context_t&) override {}
     virtual void on_connection_new(context_t&) override {}
@@ -282,7 +282,7 @@ TEST(test_epee_connection, test_lifetime)
     for (auto i = 0; i < N; ++i) {
       tag = create_connection();
       ASSERT_TRUE(shared_state->get_connections_count() == 1);
-      success = shared_state->invoke_async(1, {}, tag, [](int, const epee::span<const uint8_t>, context_t&){}, TIMEOUT);
+      success = shared_state->invoke_async(1, epee::levin::message_writer{}, tag, [](int, const epee::span<const uint8_t>, context_t&){}, TIMEOUT);
       ASSERT_TRUE(success);
       while (shared_state->sock_count == 1) {
         success = shared_state->foreach_connection([&shared_state, &tag](context_t&){
diff --git a/tests/unit_tests/epee_levin_protocol_handler_async.cpp b/tests/unit_tests/epee_levin_protocol_handler_async.cpp
index a499fa608..d9de99b8b 100644
--- a/tests/unit_tests/epee_levin_protocol_handler_async.cpp
+++ b/tests/unit_tests/epee_levin_protocol_handler_async.cpp
@@ -59,13 +59,13 @@ namespace
     {
     }
 
-    virtual int invoke(int command, const epee::span<const uint8_t> in_buff, epee::byte_slice& buff_out, test_levin_connection_context& context)
+    virtual int invoke(int command, const epee::span<const uint8_t> in_buff, epee::byte_stream& buff_out, test_levin_connection_context& context)
     {
       m_invoke_counter.inc();
       boost::unique_lock<boost::mutex> lock(m_mutex);
       m_last_command = command;
       m_last_in_buf = std::string((const char*)in_buff.data(), in_buff.size());
-      buff_out = m_invoke_out_buf.clone();
+      buff_out.write(epee::to_span(m_invoke_out_buf));
       return m_return_code;
     }
 
@@ -434,8 +434,11 @@ TEST_F(positive_test_connection_to_levin_protocol_handler_calls, handler_process
   const int expected_command = 4673261;
   const std::string in_data(256, 'e');
 
+  epee::levin::message_writer message{};
+  message.buffer.write(epee::to_span(in_data));
+
   const epee::byte_slice noise = epee::levin::make_noise_notify(1024);
-  const epee::byte_slice notify = epee::levin::make_notify(expected_command, epee::strspan<std::uint8_t>(in_data));
+  const epee::byte_slice notify = message.finalize_notify(expected_command);
 
   test_connection_ptr conn = create_connection();
 
@@ -468,11 +471,16 @@ TEST_F(positive_test_connection_to_levin_protocol_handler_calls, handler_process
   const int expected_command = 4673261;
   const int expected_fragmented_command = 46732;
   const std::string in_data(256, 'e');
-  std::string in_fragmented_data(1024 * 4, 'c');
+
+  epee::levin::message_writer message{};
+  message.buffer.write(epee::to_span(in_data));
+
+  epee::levin::message_writer in_fragmented_data;
+  in_fragmented_data.buffer.put_n('c', 1024 * 4);
 
   const epee::byte_slice noise = epee::levin::make_noise_notify(1024);
-  const epee::byte_slice notify = epee::levin::make_notify(expected_command, epee::strspan<std::uint8_t>(in_data));
-  epee::byte_slice fragmented = epee::levin::make_fragmented_notify(noise, expected_fragmented_command, epee::strspan<std::uint8_t>(in_fragmented_data));
+  const epee::byte_slice notify = message.finalize_notify(expected_command);
+  epee::byte_slice fragmented = epee::levin::make_fragmented_notify(noise.size(), expected_fragmented_command, std::move(in_fragmented_data));
 
   EXPECT_EQ(5u, fragmented.size() / 1024);
   EXPECT_EQ(0u, fragmented.size() % 1024);
@@ -497,11 +505,13 @@ TEST_F(positive_test_connection_to_levin_protocol_handler_calls, handler_process
     ASSERT_TRUE(conn->m_protocol_handler.handle_recv(next.data(), next.size()));
   }
 
-  in_fragmented_data.resize(((1024 - sizeof(epee::levin::bucket_head2)) * 5) - sizeof(epee::levin::bucket_head2)); // add padding zeroes
+  std::string compare_buffer(1024 * 4, 'c');
+  compare_buffer.resize(((1024 - sizeof(epee::levin::bucket_head2)) * 5) - sizeof(epee::levin::bucket_head2)); // add padding zeroes
+
   ASSERT_EQ(4u, m_commands_handler.notify_counter());
   ASSERT_EQ(0u, m_commands_handler.invoke_counter());
   ASSERT_EQ(expected_fragmented_command, m_commands_handler.last_command());
-  ASSERT_EQ(in_fragmented_data, m_commands_handler.last_in_buf());
+  ASSERT_EQ(compare_buffer, m_commands_handler.last_in_buf());
   ASSERT_EQ(0u, conn->send_counter());
   ASSERT_TRUE(conn->last_send_data().empty());
 
diff --git a/tests/unit_tests/levin.cpp b/tests/unit_tests/levin.cpp
index eee9e224d..30d6f8133 100644
--- a/tests/unit_tests/levin.cpp
+++ b/tests/unit_tests/levin.cpp
@@ -245,9 +245,9 @@ namespace
             return out;
         }
 
-        virtual int invoke(int command, const epee::span<const uint8_t> in_buff, epee::byte_slice& buff_out, cryptonote::levin::detail::p2p_context& context) override final
+        virtual int invoke(int command, const epee::span<const uint8_t> in_buff, epee::byte_stream& buff_out, cryptonote::levin::detail::p2p_context& context) override final
         {
-            buff_out = nullptr;
+            buff_out.clear();
             invoked_.push_back(
                 {context.m_connection_id, command, std::string{reinterpret_cast<const char*>(in_buff.data()), in_buff.size()}}
             );
@@ -384,21 +384,50 @@ TEST(make_header, expect_return)
     EXPECT_EQ(0u, header1.m_flags);
 }
 
-TEST(make_notify, empty_payload)
+TEST(message_writer, invoke_with_empty_payload)
 {
-    const epee::byte_slice message = epee::levin::make_notify(443, nullptr);
+    const epee::byte_slice message = epee::levin::message_writer{}.finalize_invoke(443);
+    const epee::levin::bucket_head2 header =
+        epee::levin::make_header(443, 0, LEVIN_PACKET_REQUEST, true);
+    ASSERT_EQ(sizeof(header), message.size());
+    EXPECT_TRUE(std::memcmp(std::addressof(header), message.data(), sizeof(header)) == 0);
+}
+
+TEST(message_writer, invoke_with_payload)
+{
+    std::string bytes(100, 'a');
+    std::generate(bytes.begin(), bytes.end(), crypto::random_device{});
+
+    epee::levin::message_writer writer{};
+    writer.buffer.write(epee::to_span(bytes));
+
+    const epee::byte_slice message = writer.finalize_invoke(443);
+    const epee::levin::bucket_head2 header =
+        epee::levin::make_header(443, bytes.size(), LEVIN_PACKET_REQUEST, true);
+
+    ASSERT_EQ(sizeof(header) + bytes.size(), message.size());
+    EXPECT_TRUE(std::memcmp(std::addressof(header), message.data(), sizeof(header)) == 0);
+    EXPECT_TRUE(std::memcmp(bytes.data(), message.data() + sizeof(header), bytes.size()) == 0);
+}
+
+TEST(message_writer, notify_with_empty_payload)
+{
+    const epee::byte_slice message = epee::levin::message_writer{}.finalize_notify(443);
     const epee::levin::bucket_head2 header =
         epee::levin::make_header(443, 0, LEVIN_PACKET_REQUEST, false);
     ASSERT_EQ(sizeof(header), message.size());
     EXPECT_TRUE(std::memcmp(std::addressof(header), message.data(), sizeof(header)) == 0);
 }
 
-TEST(make_notify, with_payload)
+TEST(message_writer, notify_with_payload)
 {
     std::string bytes(100, 'a');
     std::generate(bytes.begin(), bytes.end(), crypto::random_device{});
 
-    const epee::byte_slice message = epee::levin::make_notify(443, epee::strspan<std::uint8_t>(bytes));
+    epee::levin::message_writer writer{};
+    writer.buffer.write(epee::to_span(bytes));
+
+    const epee::byte_slice message = writer.finalize_notify(443);
     const epee::levin::bucket_head2 header =
         epee::levin::make_header(443, bytes.size(), LEVIN_PACKET_REQUEST, false);
 
@@ -407,6 +436,44 @@ TEST(make_notify, with_payload)
     EXPECT_TRUE(std::memcmp(bytes.data(), message.data() + sizeof(header), bytes.size()) == 0);
 }
 
+TEST(message_writer, response_with_empty_payload)
+{
+    const epee::byte_slice message = epee::levin::message_writer{}.finalize_response(443, 1);
+    epee::levin::bucket_head2 header =
+        epee::levin::make_header(443, 0, LEVIN_PACKET_RESPONSE, false);
+    header.m_return_code = SWAP32LE(1);
+    ASSERT_EQ(sizeof(header), message.size());
+    EXPECT_TRUE(std::memcmp(std::addressof(header), message.data(), sizeof(header)) == 0);
+}
+
+TEST(message_writer, response_with_payload)
+{
+    std::string bytes(100, 'a');
+    std::generate(bytes.begin(), bytes.end(), crypto::random_device{});
+
+    epee::levin::message_writer writer{};
+    writer.buffer.write(epee::to_span(bytes));
+
+    const epee::byte_slice message = writer.finalize_response(443, 6450);
+    epee::levin::bucket_head2 header =
+        epee::levin::make_header(443, bytes.size(), LEVIN_PACKET_RESPONSE, false);
+    header.m_return_code = SWAP32LE(6450);
+
+    ASSERT_EQ(sizeof(header) + bytes.size(), message.size());
+    EXPECT_TRUE(std::memcmp(std::addressof(header), message.data(), sizeof(header)) == 0);
+    EXPECT_TRUE(std::memcmp(bytes.data(), message.data() + sizeof(header), bytes.size()) == 0);
+}
+
+TEST(message_writer, error)
+{
+    epee::levin::message_writer writer{};
+    writer.buffer.clear();
+
+    EXPECT_THROW(writer.finalize_invoke(0), std::runtime_error);
+    EXPECT_THROW(writer.finalize_notify(0), std::runtime_error);
+    EXPECT_THROW(writer.finalize_response(0, 0), std::runtime_error);
+}
+
 TEST(make_noise, invalid)
 {
     EXPECT_TRUE(epee::levin::make_noise_notify(sizeof(epee::levin::bucket_head2) - 1).empty());
@@ -428,13 +495,13 @@ TEST(make_noise, valid)
 
 TEST(make_fragment, invalid)
 {
-    EXPECT_TRUE(epee::levin::make_fragmented_notify(nullptr, 0, nullptr).empty());
+    EXPECT_TRUE(epee::levin::make_fragmented_notify(0, 0, epee::levin::message_writer{}).empty());
 }
 
 TEST(make_fragment, single)
 {
     const epee::byte_slice noise = epee::levin::make_noise_notify(1024);
-    const epee::byte_slice fragment = epee::levin::make_fragmented_notify(noise, 11, nullptr);
+    const epee::byte_slice fragment = epee::levin::make_fragmented_notify(noise.size(), 11, epee::levin::message_writer{});
     const epee::levin::bucket_head2 header =
         epee::levin::make_header(11, 1024 - sizeof(epee::levin::bucket_head2), LEVIN_PACKET_REQUEST, false);
 
@@ -449,8 +516,13 @@ TEST(make_fragment, multiple)
     std::string bytes(1024 * 3 - 150, 'a');
     std::generate(bytes.begin(), bytes.end(), crypto::random_device{});
 
+    epee::levin::message_writer message;
+    message.buffer.write(epee::to_span(bytes));
+
     const epee::byte_slice noise = epee::levin::make_noise_notify(1024);
-    epee::byte_slice fragment = epee::levin::make_fragmented_notify(noise, 114, epee::strspan<std::uint8_t>(bytes));
+    epee::byte_slice fragment = epee::levin::make_fragmented_notify(noise.size(), 114, std::move(message));
+
+    EXPECT_EQ(1024 * 3, fragment.size());
 
     epee::levin::bucket_head2 header =
         epee::levin::make_header(0, 1024 - sizeof(epee::levin::bucket_head2), LEVIN_PACKET_BEGIN, false);
@@ -497,6 +569,7 @@ TEST(make_fragment, multiple)
 
     fragment.take_slice(bytes.size());
 
+    EXPECT_EQ(18, fragment.size());
     EXPECT_EQ(18, std::count(fragment.cbegin(), fragment.cend(), 0));
 }
 
@@ -2164,20 +2237,31 @@ TEST_F(levin_notify, command_max_bytes)
 
     add_connection(true);
 
-    std::string bytes(4096, 'h');
+    std::string payload(4096, 'h');
+    epee::byte_slice bytes;
+    {
+        epee::levin::message_writer dest{};
+        dest.buffer.write(epee::to_span(payload));
+        bytes = dest.finalize_notify(ping_command);
+    }
 
-    EXPECT_EQ(1, get_connections().notify(ping_command, epee::strspan<std::uint8_t>(bytes), contexts_.front().get_id()));
+    EXPECT_EQ(1, get_connections().send(bytes.clone(), contexts_.front().get_id()));
     EXPECT_EQ(1u, contexts_.front().process_send_queue(true));
     EXPECT_EQ(1u, receiver_.notified_size());
 
     const received_message msg = receiver_.get_raw_notification();
     EXPECT_EQ(ping_command, msg.command);
     EXPECT_EQ(contexts_.front().get_id(), msg.connection);
-    EXPECT_EQ(bytes, msg.payload);
+    EXPECT_EQ(payload, msg.payload);
 
-    bytes.push_back('e');
+    {
+        payload.push_back('h');
+        epee::levin::message_writer dest{};
+        dest.buffer.write(epee::to_span(payload));
+        bytes = dest.finalize_notify(ping_command);
+    }
 
-    EXPECT_EQ(1, get_connections().notify(ping_command, epee::strspan<std::uint8_t>(bytes), contexts_.front().get_id()));
+    EXPECT_EQ(1, get_connections().send(std::move(bytes), contexts_.front().get_id()));
     EXPECT_EQ(1u, contexts_.front().process_send_queue(false));
     EXPECT_EQ(0u, receiver_.notified_size());
 }
diff --git a/tests/unit_tests/node_server.cpp b/tests/unit_tests/node_server.cpp
index 775feace8..2c80acda5 100644
--- a/tests/unit_tests/node_server.cpp
+++ b/tests/unit_tests/node_server.cpp
@@ -449,7 +449,6 @@ TEST(cryptonote_protocol_handler, race_condition)
   };
   struct net_node_t: commands_handler_t, p2p_endpoint_t {
     using span_t = epee::span<const uint8_t>;
-    using string_t = std::string;
     using zone_t = epee::net_utils::zone;
     using uuid_t = boost::uuids::uuid;
     using relay_t = cryptonote::relay_method;
@@ -462,12 +461,9 @@ TEST(cryptonote_protocol_handler, race_condition)
       using subnets = std::map<epee::net_utils::ipv4_network_subnet, time_t>;
       using hosts = std::map<std::string, time_t>;
     };
-    struct slice {
-      using bytes = epee::byte_slice;
-    };
     shared_state_ptr shared_state;
     core_protocol_ptr core_protocol;
-    virtual int invoke(int command, const span_t in, slice::bytes &out, context_t &context) override {
+    virtual int invoke(int command, const span_t in, epee::byte_stream &out, context_t &context) override {
       if (core_protocol) {
         if (command == messages::handshake::ID) {
           return epee::net_utils::buff_to_t_adapter<void, typename messages::handshake::request, typename messages::handshake::response>(
@@ -491,7 +487,7 @@ TEST(cryptonote_protocol_handler, race_condition)
     virtual int notify(int command, const span_t in, context_t &context) override {
       if (core_protocol) {
         bool handled;
-        slice::bytes out;
+        epee::byte_stream out;
         return core_protocol->handle_invoke_map(true, command, in, out, context, handled);
       }
       else
@@ -527,22 +523,16 @@ TEST(cryptonote_protocol_handler, race_condition)
       else
         return {};
     }
-    virtual bool invoke_command_to_peer(int command, const span_t in, string_t& out, const contexts::basic& context) override {
-      if (shared_state)
-        return shared_state->invoke(command, in, out, context.m_connection_id);
-      else
-        return {};
-    }
-    virtual bool invoke_notify_to_peer(int command, const span_t in, const contexts::basic& context) override {
+    virtual bool invoke_notify_to_peer(int command, epee::levin::message_writer in, const contexts::basic& context) override {
       if (shared_state)
-        return shared_state->notify(command, in, context.m_connection_id);
+        return shared_state->send(in.finalize_notify(command), context.m_connection_id);
       else
         return {};
     }
-    virtual bool relay_notify_to_list(int command, const span_t in, connections_t connections) override {
+    virtual bool relay_notify_to_list(int command, epee::levin::message_writer in, connections_t connections) override {
       if (shared_state) {
         for (auto &e: connections)
-          shared_state->notify(command, in, e.second);
+          shared_state->send(in.finalize_notify(command), e.second);
       }
       return {};
     }