aboutsummaryrefslogtreecommitdiff
path: root/contrib/epee/include
diff options
context:
space:
mode:
Diffstat (limited to 'contrib/epee/include')
-rw-r--r--contrib/epee/include/byte_slice.h145
-rw-r--r--contrib/epee/include/net/abstract_tcp_server2.h6
-rw-r--r--contrib/epee/include/net/abstract_tcp_server2.inl63
-rw-r--r--contrib/epee/include/net/connection_basic.hpp3
-rw-r--r--contrib/epee/include/net/http_protocol_handler.inl9
-rw-r--r--contrib/epee/include/net/levin_protocol_handler.h5
-rw-r--r--contrib/epee/include/net/levin_protocol_handler_async.h145
-rw-r--r--contrib/epee/include/net/net_utils_base.h5
8 files changed, 234 insertions, 147 deletions
diff --git a/contrib/epee/include/byte_slice.h b/contrib/epee/include/byte_slice.h
new file mode 100644
index 000000000..1fbba101e
--- /dev/null
+++ b/contrib/epee/include/byte_slice.h
@@ -0,0 +1,145 @@
+// Copyright (c) 2019, The Monero Project
+//
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without modification, are
+// permitted provided that the following conditions are met:
+//
+// 1. Redistributions of source code must retain the above copyright notice, this list of
+// conditions and the following disclaimer.
+//
+// 2. Redistributions in binary form must reproduce the above copyright notice, this list
+// of conditions and the following disclaimer in the documentation and/or other
+// materials provided with the distribution.
+//
+// 3. Neither the name of the copyright holder nor the names of its contributors may be
+// used to endorse or promote products derived from this software without specific
+// prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
+// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
+// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "span.h"
+
+namespace epee
+{
+ struct byte_slice_data;
+
+ struct release_byte_slice
+ {
+ void operator()(byte_slice_data*) const noexcept;
+ };
+
+ /*! Inspired by slices in golang. Storage is thread-safe reference counted,
+ allowing for cheap copies or range selection on the bytes. The bytes
+ owned by this class are always immutable.
+
+ The functions `operator=`, `take_slice` and `remove_prefix` may alter the
+ reference count for the backing store, which will invalidate pointers
+ previously returned if the reference count is zero. Be careful about
+ "caching" pointers in these circumstances. */
+ class byte_slice
+ {
+ /* A custom reference count is used instead of shared_ptr because it allows
+ for an allocation optimization for the span constructor. This also
+ reduces the size of this class by one pointer. */
+ std::unique_ptr<byte_slice_data, release_byte_slice> storage_;
+ span<const std::uint8_t> portion_; // within storage_
+
+ //! Internal use only; use to increase `storage` reference count.
+ byte_slice(byte_slice_data* storage, span<const std::uint8_t> portion) noexcept;
+
+ struct adapt_buffer{};
+
+ template<typename T>
+ explicit byte_slice(const adapt_buffer, T&& buffer);
+
+ public:
+ using value_type = std::uint8_t;
+ using size_type = std::size_t;
+ using difference_type = std::ptrdiff_t;
+ using pointer = const std::uint8_t*;
+ using const_pointer = const std::uint8_t*;
+ using reference = std::uint8_t;
+ using const_reference = std::uint8_t;
+ using iterator = pointer;
+ using const_iterator = const_pointer;
+
+ //! Construct empty slice.
+ byte_slice() noexcept
+ : storage_(nullptr), portion_()
+ {}
+
+ //! Construct empty slice
+ byte_slice(std::nullptr_t) noexcept
+ : byte_slice()
+ {}
+
+ //! Scatter-gather (copy) multiple `sources` into a single allocated slice.
+ explicit byte_slice(std::initializer_list<span<const std::uint8_t>> sources);
+
+ //! Convert `buffer` into a slice using one allocation for shared count.
+ explicit byte_slice(std::vector<std::uint8_t>&& buffer);
+
+ //! Convert `buffer` into a slice using one allocation for shared count.
+ explicit byte_slice(std::string&& buffer);
+
+ byte_slice(byte_slice&& source) noexcept;
+ ~byte_slice() noexcept = default;
+
+ //! \note May invalidate previously retrieved pointers.
+ byte_slice& operator=(byte_slice&&) noexcept;
+
+ //! \return A shallow (cheap) copy of the data from `this` slice.
+ byte_slice clone() const noexcept { return {storage_.get(), portion_}; }
+
+ iterator begin() const noexcept { return portion_.begin(); }
+ const_iterator cbegin() const noexcept { return portion_.begin(); }
+
+ iterator end() const noexcept { return portion_.end(); }
+ const_iterator cend() const noexcept { return portion_.end(); }
+
+ bool empty() const noexcept { return storage_ == nullptr; }
+ const std::uint8_t* data() const noexcept { return portion_.data(); }
+ std::size_t size() const noexcept { return portion_.size(); }
+
+ /*! Drop bytes from the beginning of `this` slice.
+
+ \note May invalidate previously retrieved pointers.
+ \post `this->size() = this->size() - std::min(this->size(), max_bytes)`
+ \post `if (this->size() <= max_bytes) this->data() = nullptr`
+ \return Number of bytes removed. */
+ std::size_t remove_prefix(std::size_t max_bytes) noexcept;
+
+ /*! "Take" bytes from the beginning of `this` slice.
+
+ \note May invalidate previously retrieved pointers.
+ \post `this->size() = this->size() - std::min(this->size(), max_bytes)`
+ \post `if (this->size() <= max_bytes) this->data() = nullptr`
+ \return Slice containing the bytes removed from `this` slice. */
+ byte_slice take_slice(std::size_t max_bytes) noexcept;
+
+ /*! Return a shallow (cheap) copy of a slice from `begin` and `end` offsets.
+
+ \throw std::out_of_range If `end < begin`.
+ \throw std::out_of_range If `size() < end`.
+ \return Slice starting at `data() + begin` of size `end - begin`. */
+ byte_slice get_slice(std::size_t begin, std::size_t end) const;
+ };
+} // epee
+
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;
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-18 09:19:41 +0000