Add byte_stream for zero-copy serialization, and add support in ZMQ-JSON.

This commit is contained in:
Lee Clagett 2020-03-18 23:22:39 +00:00
parent 8185054db7
commit c26c93019a
14 changed files with 806 additions and 171 deletions

View file

@ -1,4 +1,4 @@
// Copyright (c) 2019, The Monero Project
// Copyright (c) 2019-2020, The Monero Project
//
// All rights reserved.
//
@ -39,6 +39,7 @@
namespace epee
{
struct byte_slice_data;
class byte_stream;
struct release_byte_slice
{
@ -50,6 +51,12 @@ namespace epee
}
};
//! Frees ref count + buffer allocated internally by `byte_buffer`.
struct release_byte_buffer
{
void operator()(std::uint8_t* buf) 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.
@ -104,6 +111,9 @@ namespace epee
//! Convert `buffer` into a slice using one allocation for shared count.
explicit byte_slice(std::string&& buffer);
//! Convert `stream` into a slice with zero allocations.
explicit byte_slice(byte_stream&& stream) noexcept;
byte_slice(byte_slice&& source) noexcept;
~byte_slice() noexcept = default;
@ -149,5 +159,19 @@ namespace epee
//! \post `empty()` \return Ownership of ref-counted buffer.
std::unique_ptr<byte_slice_data, release_byte_slice> take_buffer() noexcept;
};
//! Alias for a buffer that has space for a `byte_slice` ref count.
using byte_buffer = std::unique_ptr<std::uint8_t, release_byte_buffer>;
/*! \return `buf` with a new size of exactly `length`. New bytes not
initialized. A `nullptr` is returned on allocation failure. */
byte_buffer byte_buffer_resize(byte_buffer buf, std::size_t length) noexcept;
/*! Increase `buf` of size `current` by `more` bytes.
\throw std::range_error if `current + more` exceeds `size_t` bounds.
\return Buffer of `current + more` bytes. A `nullptr` is returned on
allocation failure. */
byte_buffer byte_buffer_increase(byte_buffer buf, std::size_t current, std::size_t more);
} // epee

View file

@ -0,0 +1,224 @@
// Copyright (c) 2020, 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 <cassert>
#include <cstdint>
#include <cstring>
#include "byte_slice.h"
#include "span.h"
namespace epee
{
/*! \brief A partial drop-in replacement for `std::ostream`.
Only a few base `std::ostream` functions are implemented - enough for
rapidjson output currently.
Advantages over `std::stringstream` or `rapidjson::StringBuffer`:
- The internal buffer can be taken without a copy.
- The internal buffer can be given to `byte_slice` with zero
allocations for reference count.
- The internal buffer can be given to `zmq_msg_data_init` without a
copy or extra allocation.
an additional advantage over `std::stringstream`:
- Construction is significantly faster - the global `std::locale`
does not have to be acquired (global thread synchronization), and
an extra allocation for `std::stringbuf` is not needed (which is an
addition to the buffer inside of that object). */
class byte_stream
{
byte_buffer buffer_; //! Beginning of buffer
std::uint8_t* next_write_; //! Current write position
const std::uint8_t* end_; //! End of buffer
std::size_t increase_size_; //! Minimum buffer size increase
//! \post `requested <= available()`
void overflow(const std::size_t requested);
//! Ensures that at least `requested` bytes are available.
void check(const std::size_t requested)
{
const std::size_t remaining = available();
if (remaining < requested)
overflow(requested);
}
public:
using char_type = std::uint8_t;
using Ch = char_type;
//! \return Default minimum size increase on buffer overflow
static constexpr std::size_t default_increase() noexcept { return 4096; }
//! Increase internal buffer by at least `byte_stream_increase` bytes.
byte_stream() noexcept
: byte_stream(default_increase())
{}
//! Increase internal buffer by at least `increase` bytes.
explicit byte_stream(const std::size_t increase) noexcept
: buffer_(nullptr),
next_write_(nullptr),
end_(nullptr),
increase_size_(increase)
{}
byte_stream(byte_stream&& rhs) noexcept;
~byte_stream() noexcept = default;
byte_stream& operator=(byte_stream&& rhs) noexcept;
//! \return The minimum increase size on buffer overflow
std::size_t increase_size() const noexcept { return increase_size_; }
const std::uint8_t* data() const noexcept { return buffer_.get(); }
std::uint8_t* tellp() const noexcept { return next_write_; }
std::size_t available() const noexcept { return end_ - next_write_; }
std::size_t size() const noexcept { return next_write_ - buffer_.get(); }
std::size_t capacity() const noexcept { return end_ - buffer_.get(); }
//! Compatibility with rapidjson.
void Flush() const noexcept
{}
/*! Reserve at least `more` bytes.
\post `size() + more <= available()`.
\throw std::range_error if exceeding max `size_t` value.
\throw std::bad_alloc if allocation fails. */
void reserve(const std::size_t more)
{
check(more);
}
/*! Copy `length` bytes starting at `ptr` to end of stream.
\throw std::range_error If exceeding max size_t value.
\throw std::bad_alloc If allocation fails. */
void write(const std::uint8_t* ptr, const std::size_t length)
{
check(length);
std::memcpy(tellp(), ptr, length);
next_write_ += length;
}
/*! Copy `length` bytes starting at `ptr` to end of stream.
\throw std::range_error if exceeding max `size_t` value.
\throw std::bad_alloc if allocation fails. */
void write(const char* ptr, const std::size_t length)
{
write(reinterpret_cast<const std::uint8_t*>(ptr), length);
}
/*! Copy `source` to end of stream.
\throw std::range_error if exceeding max `size_t` value.
\throw std::bad_alloc if allocation fails. */
void write(const epee::span<const std::uint8_t> source)
{
write(source.data(), source.size());
}
/*! Copy `source` to end of stream.
\throw std::range_error if exceeding max `size_t` value.
\throw std::bad_alloc if allocation fails. */
void write(const epee::span<const char> source)
{
write(source.data(), source.size());
}
/*! Copy `ch` to end of stream.
\throw std::range_error if exceeding max `size_t` value.
\throw std::bad_alloc if allocation fails. */
void put(const std::uint8_t ch)
{
check(1);
put_unsafe(ch);
}
/*! Copy `ch` to end of stream. Provides rapidjson compatability.
\throw std::range_error if exceeding max `size_t` value.
\throw std::bad_alloc if allocation fails. */
void Put(const std::uint8_t ch)
{
put(ch);
}
/*! Writes `ch` to end of stream without runtime capacity checks. Must use
`reserve` before calling this function. Primarily for use with
rapidjson, which writes characters at a time but reserves memory in
blocks. Most applications want to use `put` or `write`. */
void put_unsafe(const std::uint8_t ch) noexcept
{
assert(1 <= available());
*(tellp()) = ch;
++next_write_;
}
/*! Write `ch` to end of stream `count` times.
\throw std::range_error if exceeding max `size_t` value.
\throw std::bad_alloc if allocation fails. */
void put_n(const std::uint8_t ch, const std::size_t count)
{
check(count);
std::memset(tellp(), count, ch);
next_write_ += count;
}
/*! Copy `ch` to end of stream.
\throw std::range_error if exceeding max `size_t` value.
\throw std::bad_alloc if allocation fails. */
void push_back(const std::uint8_t ch)
{
put(ch);
}
//! \return The internal buffer. \post `size() == capacity() == 0`.
byte_buffer take_buffer() noexcept;
};
//! Compatability/optimization for rapidjson.
inline void PutReserve(byte_stream& dest, const std::size_t length)
{
dest.reserve(length);
}
//! Compatability/optimization for rapidjson.
inline void PutUnsafe(byte_stream& dest, const std::uint8_t ch)
{
dest.put_unsafe(ch);
}
//! Compability/optimization for rapidjson.
inline void PutN(byte_stream& dest, const std::uint8_t ch, const std::size_t count)
{
dest.put_n(ch, count);
}
} // epee

View file

@ -26,8 +26,9 @@
# 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.
add_library(epee STATIC byte_slice.cpp hex.cpp abstract_http_client.cpp http_auth.cpp mlog.cpp net_helper.cpp net_utils_base.cpp string_tools.cpp wipeable_string.cpp
levin_base.cpp memwipe.c connection_basic.cpp network_throttle.cpp network_throttle-detail.cpp mlocker.cpp buffer.cpp net_ssl.cpp
add_library(epee STATIC byte_slice.cpp byte_stream.cpp hex.cpp abstract_http_client.cpp http_auth.cpp mlog.cpp net_helper.cpp net_utils_base.cpp string_tools.cpp
wipeable_string.cpp levin_base.cpp memwipe.c connection_basic.cpp network_throttle.cpp network_throttle-detail.cpp mlocker.cpp buffer.cpp net_ssl.cpp
int-util.cpp)
if (USE_READLINE AND (GNU_READLINE_FOUND OR (DEPENDS AND NOT MINGW)))

View file

@ -1,4 +1,4 @@
// Copyright (c) 2019, The Monero Project
// Copyright (c) 2019-2020, The Monero Project
//
// All rights reserved.
//
@ -34,6 +34,7 @@
#include <utility>
#include "byte_slice.h"
#include "byte_stream.h"
namespace epee
{
@ -117,6 +118,12 @@ namespace epee
}
} // anonymous
void release_byte_buffer::operator()(std::uint8_t* buf) const noexcept
{
if (buf)
std::free(buf - sizeof(raw_byte_slice));
}
byte_slice::byte_slice(byte_slice_data* storage, span<const std::uint8_t> portion) noexcept
: storage_(storage), portion_(portion)
{
@ -163,6 +170,14 @@ namespace epee
: byte_slice(adapt_buffer{}, std::move(buffer))
{}
byte_slice::byte_slice(byte_stream&& stream) noexcept
: storage_(nullptr), portion_(stream.data(), stream.size())
{
std::uint8_t* const data = stream.take_buffer().release() - sizeof(raw_byte_slice);
new (data) raw_byte_slice{};
storage_.reset(reinterpret_cast<raw_byte_slice*>(data));
}
byte_slice::byte_slice(byte_slice&& source) noexcept
: storage_(std::move(source.storage_)), portion_(source.portion_)
{
@ -217,4 +232,29 @@ namespace epee
portion_ = nullptr;
return out;
}
byte_buffer byte_buffer_resize(byte_buffer buf, const std::size_t length) noexcept
{
if (std::numeric_limits<std::size_t>::max() - sizeof(raw_byte_slice) < length)
return nullptr;
std::uint8_t* data = buf.get();
if (data != nullptr)
data -= sizeof(raw_byte_slice);
data = static_cast<std::uint8_t*>(std::realloc(data, sizeof(raw_byte_slice) + length));
if (data == nullptr)
return nullptr;
buf.release();
buf.reset(data + sizeof(raw_byte_slice));
return buf;
}
byte_buffer byte_buffer_increase(byte_buffer buf, const std::size_t current, const std::size_t more)
{
if (std::numeric_limits<std::size_t>::max() - current < more)
throw std::range_error{"byte_buffer_increase size_t overflow"};
return byte_buffer_resize(std::move(buf), current + more);
}
} // epee

View file

@ -0,0 +1,93 @@
// Copyright (c) 2020, 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.
#include "byte_stream.h"
#include <algorithm>
#include <limits>
#include <utility>
#include <iostream>
namespace epee
{
void byte_stream::overflow(const std::size_t requested)
{
// Recalculating `need` bytes removes at least one instruction from every
// inlined `put` call in header
assert(available() < requested);
const std::size_t need = requested - available();
const std::size_t len = size();
const std::size_t cap = capacity();
const std::size_t increase = std::max(need, increase_size());
next_write_ = nullptr;
end_ = nullptr;
buffer_ = byte_buffer_increase(std::move(buffer_), cap, increase);
if (!buffer_)
throw std::bad_alloc{};
next_write_ = buffer_.get() + len;
end_ = buffer_.get() + cap + increase;
}
byte_stream::byte_stream(byte_stream&& rhs) noexcept
: buffer_(std::move(rhs.buffer_)),
next_write_(rhs.next_write_),
end_(rhs.end_),
increase_size_(rhs.increase_size_)
{
rhs.next_write_ = nullptr;
rhs.end_ = nullptr;
}
byte_stream& byte_stream::operator=(byte_stream&& rhs) noexcept
{
if (this != std::addressof(rhs))
{
buffer_ = std::move(rhs.buffer_);
next_write_ = rhs.next_write_;
end_ = rhs.end_;
increase_size_ = rhs.increase_size_;
rhs.next_write_ = nullptr;
rhs.end_ = nullptr;
}
return *this;
}
byte_buffer byte_stream::take_buffer() noexcept
{
byte_buffer out{std::move(buffer_)};
next_write_ = nullptr;
end_ = nullptr;
return out;
}
}