Add ref-counted buffer byte_slice. Currently used for sending TCP data.

7 files changed, 89 insertions, 147 deletions

diff --git a/contrib/epee/include/net/abstract_tcp_server2.h b/contrib/epee/include/net/abstract_tcp_server2.h
index b38ab5399..373ef0dc6 100644
--- a/contrib/epee/include/net/abstract_tcp_server2.h
+++ b/contrib/epee/include/net/abstract_tcp_server2.h
@@ -53,6 +53,7 @@
 #include <boost/enable_shared_from_this.hpp>
 #include <boost/interprocess/detail/atomic.hpp>
 #include <boost/thread/thread.hpp>
+#include "byte_slice.h"
 #include "net_utils_base.h"
 #include "syncobj.h"
 #include "connection_basic.hpp"
@@ -135,8 +136,7 @@ namespace net_utils
     
   private:
     //----------------- i_service_endpoint ---------------------
-    virtual bool do_send(const void* ptr, size_t cb); ///< (see do_send from i_service_endpoint)
-    virtual bool do_send_chunk(const void* ptr, size_t cb); ///< will send (or queue) a part of data
+    virtual bool do_send(byte_slice message); ///< (see do_send from i_service_endpoint)
     virtual bool send_done();
     virtual bool close();
     virtual bool call_run_once_service_io();
@@ -145,6 +145,8 @@ namespace net_utils
     virtual bool add_ref();
     virtual bool release();
     //------------------------------------------------------
+    bool do_send_chunk(byte_slice chunk); ///< will send (or queue) a part of data. internal use only
+
     boost::shared_ptr<connection<t_protocol_handler> > safe_shared_from_this();
     bool shutdown();
     /// Handle completion of a receive operation.
diff --git a/contrib/epee/include/net/abstract_tcp_server2.inl b/contrib/epee/include/net/abstract_tcp_server2.inl
index 19e9c9af9..ebe1d5aaf 100644
--- a/contrib/epee/include/net/abstract_tcp_server2.inl
+++ b/contrib/epee/include/net/abstract_tcp_server2.inl
@@ -520,7 +520,7 @@ PRAGMA_WARNING_DISABLE_VS(4355)
   }
   //---------------------------------------------------------------------------------
     template<class t_protocol_handler>
-  bool connection<t_protocol_handler>::do_send(const void* ptr, size_t cb) {
+  bool connection<t_protocol_handler>::do_send(byte_slice message) {
     TRY_ENTRY();
 
     // Use safe_shared_from_this, because of this is public method and it can be called on the object being deleted
@@ -529,6 +529,9 @@ PRAGMA_WARNING_DISABLE_VS(4355)
     if (m_was_shutdown) return false;
 		// TODO avoid copy
 
+		std::uint8_t const* const message_data = message.data();
+		const std::size_t message_size = message.size();
+
 		const double factor = 32; // TODO config
 		typedef long long signed int t_safe; // my t_size to avoid any overunderflow in arithmetic
 		const t_safe chunksize_good = (t_safe)( 1024 * std::max(1.0,factor) );
@@ -538,13 +541,11 @@ PRAGMA_WARNING_DISABLE_VS(4355)
         CHECK_AND_ASSERT_MES(! (chunksize_max<0), false, "Negative chunksize_max" ); // make sure it is unsigned before removin sign with cast:
         long long unsigned int chunksize_max_unsigned = static_cast<long long unsigned int>( chunksize_max ) ;
 
-        if (allow_split && (cb > chunksize_max_unsigned)) {
+        if (allow_split && (message_size > chunksize_max_unsigned)) {
 			{ // LOCK: chunking
     		epee::critical_region_t<decltype(m_chunking_lock)> send_guard(m_chunking_lock); // *** critical *** 
 
-				MDEBUG("do_send() will SPLIT into small chunks, from packet="<<cb<<" B for ptr="<<ptr);
-				t_safe all = cb; // all bytes to send 
-				t_safe pos = 0; // current sending position
+				MDEBUG("do_send() will SPLIT into small chunks, from packet="<<message_size<<" B for ptr="<<message_data);
 				// 01234567890 
 				// ^^^^        (pos=0, len=4)     ;   pos:=pos+len, pos=4
 				//     ^^^^    (pos=4, len=4)     ;   pos:=pos+len, pos=8
@@ -554,40 +555,25 @@ PRAGMA_WARNING_DISABLE_VS(4355)
 				// char* buf = new char[ bufsize ];
 
 				bool all_ok = true;
-				while (pos < all) {
-					t_safe lenall = all-pos; // length from here to end
-					t_safe len = std::min( chunksize_good , lenall); // take a smaller part
-					CHECK_AND_ASSERT_MES(len<=chunksize_good, false, "len too large");
-					// pos=8; len=4; all=10;	len=3;
-
-                    CHECK_AND_ASSERT_MES(! (len<0), false, "negative len"); // check before we cast away sign:
-                    unsigned long long int len_unsigned = static_cast<long long int>( len );
-                    CHECK_AND_ASSERT_MES(len>0, false, "len not strictly positive"); // (redundant)
-                    CHECK_AND_ASSERT_MES(len_unsigned < std::numeric_limits<size_t>::max(), false, "Invalid len_unsigned");   // yeap we want strong < then max size, to be sure
-					
-					void *chunk_start = ((char*)ptr) + pos;
-					MDEBUG("chunk_start="<<chunk_start<<" ptr="<<ptr<<" pos="<<pos);
-					CHECK_AND_ASSERT_MES(chunk_start >= ptr, false, "Pointer wraparound"); // not wrapped around address?
-					//std::memcpy( (void*)buf, chunk_start, len);
-
-					MDEBUG("part of " << lenall << ": pos="<<pos << " len="<<len);
-
-					bool ok = do_send_chunk(chunk_start, len); // <====== ***
+				while (!message.empty()) {
+					byte_slice chunk = message.take_slice(chunksize_good);
+
+					MDEBUG("chunk_start="<<chunk.data()<<" ptr="<<message_data<<" pos="<<(chunk.data() - message_data));
+					MDEBUG("part of " << message.size() << ": pos="<<(chunk.data() - message_data) << " len="<<chunk.size());
+
+					bool ok = do_send_chunk(std::move(chunk)); // <====== ***
 
 					all_ok = all_ok && ok;
 					if (!all_ok) {
-						MDEBUG("do_send() DONE ***FAILED*** from packet="<<cb<<" B for ptr="<<ptr);
+						MDEBUG("do_send() DONE ***FAILED*** from packet="<<message_size<<" B for ptr="<<message_data);
 						MDEBUG("do_send() SEND was aborted in middle of big package - this is mostly harmless "
-							<< " (e.g. peer closed connection) but if it causes trouble tell us at #monero-dev. " << cb);
+							<< " (e.g. peer closed connection) but if it causes trouble tell us at #monero-dev. " << message_size);
 						return false; // partial failure in sending
 					}
-					pos = pos+len;
-					CHECK_AND_ASSERT_MES(pos >0, false, "pos <= 0");
-
 					// (in catch block, or uniq pointer) delete buf;
 				} // each chunk
 
-				MDEBUG("do_send() DONE SPLIT from packet="<<cb<<" B for ptr="<<ptr);
+				MDEBUG("do_send() DONE SPLIT from packet="<<message_size<<" B for ptr="<<message_data);
 
                 MDEBUG("do_send() m_connection_type = " << m_connection_type);
 
@@ -595,7 +581,7 @@ PRAGMA_WARNING_DISABLE_VS(4355)
 			} // LOCK: chunking
 		} // a big block (to be chunked) - all chunks
 		else { // small block
-			return do_send_chunk(ptr,cb); // just send as 1 big chunk
+			return do_send_chunk(std::move(message)); // just send as 1 big chunk
 		}
 
     CATCH_ENTRY_L0("connection<t_protocol_handler>::do_send", false);
@@ -603,7 +589,7 @@ PRAGMA_WARNING_DISABLE_VS(4355)
 
   //---------------------------------------------------------------------------------
   template<class t_protocol_handler>
-  bool connection<t_protocol_handler>::do_send_chunk(const void* ptr, size_t cb)
+  bool connection<t_protocol_handler>::do_send_chunk(byte_slice chunk)
   {
     TRY_ENTRY();
     // Use safe_shared_from_this, because of this is public method and it can be called on the object being deleted
@@ -623,7 +609,7 @@ PRAGMA_WARNING_DISABLE_VS(4355)
 
     //_info("[sock " << socket().native_handle() << "] SEND " << cb);
     context.m_last_send = time(NULL);
-    context.m_send_cnt += cb;
+    context.m_send_cnt += chunk.size();
     //some data should be wrote to stream
     //request complete
     
@@ -644,7 +630,7 @@ PRAGMA_WARNING_DISABLE_VS(4355)
         }*/
 
         long int ms = 250 + (rand()%50);
-        MDEBUG("Sleeping because QUEUE is FULL, in " << __FUNCTION__ << " for " << ms << " ms before packet_size="<<cb); // XXX debug sleep
+        MDEBUG("Sleeping because QUEUE is FULL, in " << __FUNCTION__ << " for " << ms << " ms before packet_size="<<chunk.size()); // XXX debug sleep
         m_send_que_lock.unlock();
         boost::this_thread::sleep(boost::posix_time::milliseconds( ms ) );
         m_send_que_lock.lock();
@@ -657,12 +643,11 @@ PRAGMA_WARNING_DISABLE_VS(4355)
         }
     }
 
-    m_send_que.resize(m_send_que.size()+1);
-    m_send_que.back().assign((const char*)ptr, cb);
-    
+    m_send_que.push_back(std::move(chunk));
+
     if(m_send_que.size() > 1)
     { // active operation should be in progress, nothing to do, just wait last operation callback
-        auto size_now = cb;
+        auto size_now = m_send_que.back().size();
         MDEBUG("do_send_chunk() NOW just queues: packet="<<size_now<<" B, is added to queue-size="<<m_send_que.size());
         //do_send_handler_delayed( ptr , size_now ); // (((H))) // empty function
       
@@ -680,7 +665,7 @@ PRAGMA_WARNING_DISABLE_VS(4355)
         auto size_now = m_send_que.front().size();
         MDEBUG("do_send_chunk() NOW SENSD: packet="<<size_now<<" B");
         if (speed_limit_is_enabled())
-			do_send_handler_write( ptr , size_now ); // (((H)))
+			do_send_handler_write( chunk.data(), chunk.size() ); // (((H)))
 
         CHECK_AND_ASSERT_MES( size_now == m_send_que.front().size(), false, "Unexpected queue size");
         reset_timer(get_default_timeout(), false);
diff --git a/contrib/epee/include/net/connection_basic.hpp b/contrib/epee/include/net/connection_basic.hpp
index 2acc6cdda..baeb18e27 100644
--- a/contrib/epee/include/net/connection_basic.hpp
+++ b/contrib/epee/include/net/connection_basic.hpp
@@ -49,6 +49,7 @@
 #include <boost/asio.hpp>
 #include <boost/asio/ssl.hpp>
 
+#include "byte_slice.h"
 #include "net/net_utils_base.h"
 #include "net/net_ssl.h"
 #include "syncobj.h"
@@ -108,7 +109,7 @@ class connection_basic { // not-templated base class for rapid developmet of som
     volatile uint32_t m_want_close_connection;
     std::atomic<bool> m_was_shutdown;
     critical_section m_send_que_lock;
-    std::list<std::string> m_send_que;
+    std::deque<byte_slice> m_send_que;
     volatile bool m_is_multithreaded;
     /// Strand to ensure the connection's handlers are not called concurrently.
     boost::asio::io_service::strand strand_;
diff --git a/contrib/epee/include/net/http_protocol_handler.inl b/contrib/epee/include/net/http_protocol_handler.inl
index 790d0f3b1..19bdf4ff0 100644
--- a/contrib/epee/include/net/http_protocol_handler.inl
+++ b/contrib/epee/include/net/http_protocol_handler.inl
@@ -591,11 +591,12 @@ namespace net_utils
 		std::string response_data = get_response_header(response);
 		//LOG_PRINT_L0("HTTP_SEND: << \r\n" << response_data + response.m_body);
 
-    LOG_PRINT_L3("HTTP_RESPONSE_HEAD: << \r\n" << response_data);
-		
-		m_psnd_hndlr->do_send((void*)response_data.data(), response_data.size());
+		LOG_PRINT_L3("HTTP_RESPONSE_HEAD: << \r\n" << response_data);
+
 		if ((response.m_body.size() && (query_info.m_http_method != http::http_method_head)) || (query_info.m_http_method == http::http_method_options))
-			m_psnd_hndlr->do_send((void*)response.m_body.data(), response.m_body.size());
+			response_data += response.m_body;
+
+		m_psnd_hndlr->do_send(byte_slice{std::move(response_data)});
 		m_psnd_hndlr->send_done();
 		return res;
 	}
diff --git a/contrib/epee/include/net/levin_protocol_handler.h b/contrib/epee/include/net/levin_protocol_handler.h
index 791766762..c510cfd79 100644
--- a/contrib/epee/include/net/levin_protocol_handler.h
+++ b/contrib/epee/include/net/levin_protocol_handler.h
@@ -157,10 +157,9 @@ namespace levin
 						m_current_head.m_return_code = m_config.m_pcommands_handler->invoke(m_current_head.m_command, buff_to_invoke, return_buff, m_conn_context);
 						m_current_head.m_cb = return_buff.size();
 						m_current_head.m_have_to_return_data = false;
-						std::string send_buff((const char*)&m_current_head, sizeof(m_current_head));
-						send_buff += return_buff;
 
-						if(!m_psnd_hndlr->do_send(send_buff.data(), send_buff.size()))
+						return_buff.insert(0, (const char*)&m_current_head, sizeof(m_current_head));
+						if(!m_psnd_hndlr->do_send(byte_slice{std::move(return_buff)}))
 							return false;
 
 					}
diff --git a/contrib/epee/include/net/levin_protocol_handler_async.h b/contrib/epee/include/net/levin_protocol_handler_async.h
index 8d7ffb2c2..3f734a72b 100644
--- a/contrib/epee/include/net/levin_protocol_handler_async.h
+++ b/contrib/epee/include/net/levin_protocol_handler_async.h
@@ -136,7 +136,6 @@ public:
   critical_section m_local_inv_buff_lock;
   std::string m_local_inv_buff;
 
-  critical_section m_send_lock;
   critical_section m_call_lock;
 
   volatile uint32_t m_wait_count;
@@ -260,6 +259,34 @@ public:
     return handler->is_timer_started();
   }
   template<class callback_t> friend struct anvoke_handler;
+
+  static bucket_head2 make_header(uint32_t command, uint64_t msg_size, uint32_t flags, bool expect_response) noexcept
+  {
+    bucket_head2 head = {0};
+    head.m_signature = SWAP64LE(LEVIN_SIGNATURE);
+    head.m_have_to_return_data = expect_response;
+    head.m_cb = SWAP64LE(msg_size);
+
+    head.m_command = SWAP32LE(command);
+    head.m_protocol_version = SWAP32LE(LEVIN_PROTOCOL_VER_1);
+    head.m_flags = SWAP32LE(flags);
+    return head;
+  }
+
+  bool send_message(uint32_t command, epee::span<const uint8_t> in_buff, uint32_t flags, bool expect_response)
+  {
+    const bucket_head2 head = make_header(command, in_buff.size(), flags, expect_response);
+    if(!m_pservice_endpoint->do_send(byte_slice{as_byte_span(head), in_buff}))
+      return false;
+
+    MDEBUG(m_connection_context << "LEVIN_PACKET_SENT. [len=" << head.m_cb
+        << ", flags" << head.m_flags
+        << ", r?=" << head.m_have_to_return_data
+        <<", cmd = " << head.m_command
+        << ", ver=" << head.m_protocol_version);
+    return true;
+  }
+
 public:
   async_protocol_handler(net_utils::i_service_endpoint* psnd_hndlr, 
     config_type& config, 
@@ -458,37 +485,22 @@ public:
             if(m_current_head.m_have_to_return_data)
             {
               std::string return_buff;
-              m_current_head.m_return_code = m_config.m_pcommands_handler->invoke(
-                                                                  m_current_head.m_command, 
-                                                                  buff_to_invoke, 
-                                                                  return_buff, 
-                                                                  m_connection_context);
-              m_current_head.m_cb = return_buff.size();
-              m_current_head.m_have_to_return_data = false;
-              m_current_head.m_protocol_version = LEVIN_PROTOCOL_VER_1;
-              m_current_head.m_flags = LEVIN_PACKET_RESPONSE;
-#if BYTE_ORDER == LITTLE_ENDIAN
-              std::string send_buff((const char*)&m_current_head, sizeof(m_current_head));
-#else
-              bucket_head2 head = m_current_head;
-              head.m_signature = SWAP64LE(head.m_signature);
-              head.m_cb = SWAP64LE(head.m_cb);
-              head.m_command = SWAP32LE(head.m_command);
-              head.m_return_code = SWAP32LE(head.m_return_code);
-              head.m_flags = SWAP32LE(head.m_flags);
-              head.m_protocol_version = SWAP32LE(head.m_protocol_version);
-              std::string send_buff((const char*)&head, sizeof(head));
-#endif
-              send_buff += return_buff;
-              CRITICAL_REGION_BEGIN(m_send_lock);
-              if(!m_pservice_endpoint->do_send(send_buff.data(), send_buff.size()))
+              const uint32_t return_code = m_config.m_pcommands_handler->invoke(
+                m_current_head.m_command, buff_to_invoke, return_buff, m_connection_context
+              );
+
+	      bucket_head2 head = make_header(m_current_head.m_command, return_buff.size(), LEVIN_PACKET_RESPONSE, false);
+	      head.m_return_code = SWAP32LE(return_code);
+	      return_buff.insert(0, reinterpret_cast<const char*>(&head), sizeof(head));
+
+              if(!m_pservice_endpoint->do_send(byte_slice{std::move(return_buff)}))
                 return false;
-              CRITICAL_REGION_END();
-              MDEBUG(m_connection_context << "LEVIN_PACKET_SENT. [len=" << m_current_head.m_cb
-                << ", flags" << m_current_head.m_flags 
-                << ", r?=" << m_current_head.m_have_to_return_data 
-                <<", cmd = " << m_current_head.m_command 
-                << ", ver=" << m_current_head.m_protocol_version);
+
+              MDEBUG(m_connection_context << "LEVIN_PACKET_SENT. [len=" << head.m_cb
+                << ", flags" << head.m_flags
+                << ", r?=" << head.m_have_to_return_data
+                <<", cmd = " << head.m_command
+                << ", ver=" << head.m_protocol_version);
             }
             else
               m_config.m_pcommands_handler->notify(m_current_head.m_command, buff_to_invoke, m_connection_context);
@@ -584,26 +596,10 @@ public:
         break;
       }
 
-      bucket_head2 head = {0};
-      head.m_signature = SWAP64LE(LEVIN_SIGNATURE);
-      head.m_cb = SWAP64LE(in_buff.size());
-      head.m_have_to_return_data = true;
-
-      head.m_flags = SWAP32LE(LEVIN_PACKET_REQUEST);
-      head.m_command = SWAP32LE(command);
-      head.m_protocol_version = SWAP32LE(LEVIN_PROTOCOL_VER_1);
-
       boost::interprocess::ipcdetail::atomic_write32(&m_invoke_buf_ready, 0);
-      CRITICAL_REGION_BEGIN(m_send_lock);
-      CRITICAL_REGION_LOCAL1(m_invoke_response_handlers_lock);
-      if(!m_pservice_endpoint->do_send(&head, sizeof(head)))
-      {
-        LOG_ERROR_CC(m_connection_context, "Failed to do_send");
-        err_code = LEVIN_ERROR_CONNECTION;
-        break;
-      }
+      CRITICAL_REGION_BEGIN(m_invoke_response_handlers_lock);
 
-      if(!m_pservice_endpoint->do_send(in_buff.data(), in_buff.size()))
+      if(!send_message(command, in_buff, LEVIN_PACKET_REQUEST, true))
       {
         LOG_ERROR_CC(m_connection_context, "Failed to do_send");
         err_code = LEVIN_ERROR_CONNECTION;
@@ -642,35 +638,13 @@ public:
     if(m_deletion_initiated)
       return LEVIN_ERROR_CONNECTION_DESTROYED;
 
-    bucket_head2 head = {0};
-    head.m_signature = SWAP64LE(LEVIN_SIGNATURE);
-    head.m_cb = SWAP64LE(in_buff.size());
-    head.m_have_to_return_data = true;
-
-    head.m_flags = SWAP32LE(LEVIN_PACKET_REQUEST);
-    head.m_command = SWAP32LE(command);
-    head.m_protocol_version = SWAP32LE(LEVIN_PROTOCOL_VER_1);
-
     boost::interprocess::ipcdetail::atomic_write32(&m_invoke_buf_ready, 0);
-    CRITICAL_REGION_BEGIN(m_send_lock);
-    if(!m_pservice_endpoint->do_send(&head, sizeof(head)))
-    {
-      LOG_ERROR_CC(m_connection_context, "Failed to do_send");
-      return LEVIN_ERROR_CONNECTION;
-    }
 
-    if(!m_pservice_endpoint->do_send(in_buff.data(), in_buff.size()))
+    if (!send_message(command, in_buff, LEVIN_PACKET_REQUEST, true))
     {
-      LOG_ERROR_CC(m_connection_context, "Failed to do_send");
+      LOG_ERROR_CC(m_connection_context, "Failed to send request");
       return LEVIN_ERROR_CONNECTION;
     }
-    CRITICAL_REGION_END();
-
-    MDEBUG(m_connection_context << "LEVIN_PACKET_SENT. [len=" << head.m_cb
-                            << ", f=" << head.m_flags 
-                            << ", r?=" << head.m_have_to_return_data 
-                            << ", cmd = " << head.m_command 
-                            << ", ver=" << head.m_protocol_version);
 
     uint64_t ticks_start = misc_utils::get_tick_count();
     size_t prev_size = 0;
@@ -716,32 +690,11 @@ public:
     if(m_deletion_initiated)
       return LEVIN_ERROR_CONNECTION_DESTROYED;
 
-    bucket_head2 head = {0};
-    head.m_signature = SWAP64LE(LEVIN_SIGNATURE);
-    head.m_have_to_return_data = false;
-    head.m_cb = SWAP64LE(in_buff.size());
-
-    head.m_command = SWAP32LE(command);
-    head.m_protocol_version = SWAP32LE(LEVIN_PROTOCOL_VER_1);
-    head.m_flags = SWAP32LE(LEVIN_PACKET_REQUEST);
-    CRITICAL_REGION_BEGIN(m_send_lock);
-    if(!m_pservice_endpoint->do_send(&head, sizeof(head)))
-    {
-      LOG_ERROR_CC(m_connection_context, "Failed to do_send()");
-      return -1;
-    }
-
-    if(!m_pservice_endpoint->do_send(in_buff.data(), in_buff.size()))
+    if (!send_message(command, in_buff, LEVIN_PACKET_REQUEST, false))
     {
-      LOG_ERROR_CC(m_connection_context, "Failed to do_send()");
+      LOG_ERROR_CC(m_connection_context, "Failed to send notify message");
       return -1;
     }
-    CRITICAL_REGION_END();
-    LOG_DEBUG_CC(m_connection_context, "LEVIN_PACKET_SENT. [len=" << head.m_cb <<
-      ", f=" << head.m_flags << 
-      ", r?=" << head.m_have_to_return_data <<
-      ", cmd = " << head.m_command << 
-      ", ver=" << head.m_protocol_version);
 
     return 1;
   }
diff --git a/contrib/epee/include/net/net_utils_base.h b/contrib/epee/include/net/net_utils_base.h
index 5ae3e53b3..dd80fae8b 100644
--- a/contrib/epee/include/net/net_utils_base.h
+++ b/contrib/epee/include/net/net_utils_base.h
@@ -34,9 +34,10 @@
 #include <boost/asio/ip/address_v6.hpp>
 #include <typeinfo>
 #include <type_traits>
+#include "byte_slice.h"
 #include "enums.h"
-#include "serialization/keyvalue_serialization.h"
 #include "misc_log_ex.h"
+#include "serialization/keyvalue_serialization.h"
 
 #undef MONERO_DEFAULT_LOG_CATEGORY
 #define MONERO_DEFAULT_LOG_CATEGORY "net"
@@ -424,7 +425,7 @@ namespace net_utils
 	/************************************************************************/
 	struct i_service_endpoint
 	{
-		virtual bool do_send(const void* ptr, size_t cb)=0;
+		virtual bool do_send(byte_slice message)=0;
     virtual bool close()=0;
     virtual bool send_done()=0;
     virtual bool call_run_once_service_io()=0;