Commit 3efaa0c0 authored by Dominik Charousset's avatar Dominik Charousset

Streamline naming and use simpler buffer cache

parent dff483fd
......@@ -12,6 +12,7 @@ set(LIBCAF_NET_SRCS
src/connection_state.cpp
src/convert_ip_endpoint.cpp
src/datagram_socket.cpp
src/defaults.cpp
src/ec.cpp
src/endpoint_manager.cpp
src/header.cpp
......@@ -23,6 +24,7 @@ set(LIBCAF_NET_SRCS
src/net/endpoint_manager_queue.cpp
src/net/middleman.cpp
src/net/middleman_backend.cpp
src/net/packet_writer.cpp
src/network_socket.cpp
src/pipe_socket.cpp
src/pollset_updater.cpp
......@@ -32,7 +34,6 @@ set(LIBCAF_NET_SRCS
src/tcp_accept_socket.cpp
src/tcp_stream_socket.cpp
src/udp_datagram_socket.cpp
src/defaults.cpp
)
add_custom_target(libcaf_net)
......
......@@ -77,7 +77,7 @@ public:
manager_ = &parent.manager();
// Write handshake.
auto hdr = parent.next_header_buffer();
auto payload = parent.next_buffer();
auto payload = parent.next_payload_buffer();
if (auto err = generate_handshake(payload))
return err;
to_bytes(header{message_type::handshake,
......
......@@ -27,9 +27,10 @@ namespace defaults {
namespace middleman {
/// Maximum number of payload buffers to keep in transport
extern const size_t max_output_buffers;
/// Maximum number of header buffers to keep in transport
/// Maximum number of cached buffers for sending payloads.
extern const size_t max_payload_buffers;
/// Maximum number of cached buffers for sending headers.
extern const size_t max_header_buffers;
} // namespace middleman
......
......@@ -32,18 +32,18 @@ class packet_writer {
public:
using buffer_type = std::vector<byte>;
virtual ~packet_writer() = default;
virtual ~packet_writer();
/// Returns the next header_buffer from transport
/// Returns a buffer for writing header information.
virtual buffer_type next_header_buffer() = 0;
/// Returns the next payload_buffer from transport
virtual buffer_type next_buffer() = 0;
/// Returns a buffer for writing payload content.
virtual buffer_type next_payload_buffer() = 0;
/// Convenience function to write a packet consisting of multiple buffers.
/// @param buffers the buffers that make up the packet. first buffer should
/// contain the header.
/// @warning takes ownership of `buffers`.
/// @param buffers all buffers for the packet. The first buffer is a header
/// buffer, the other buffers are payload buffer.
/// @warning this function takes ownership of `buffers`.
template <class... Ts>
void write_packet(Ts&... buffers) {
buffer_type* bufs[] = {&buffers...};
......@@ -51,8 +51,8 @@ public:
}
protected:
/// Actual write_packet implementation.
/// @param buffers `span` containing all buffers of a packet.
/// Implementing function for `write_packet`.
/// @param buffers a `span` containing all buffers of a packet.
virtual void write_impl(span<buffer_type*> buffers) = 0;
};
......
......@@ -62,8 +62,8 @@ public:
return transport().next_header_buffer();
}
buffer_type next_buffer() override {
return transport().next_buffer();
buffer_type next_payload_buffer() override {
return transport().next_payload_buffer();
}
// -- member functions -------------------------------------------------------
......
......@@ -4,7 +4,7 @@
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *#
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
......@@ -50,6 +50,8 @@ public:
using buffer_type = std::vector<byte>;
using buffer_cache_type = std::vector<buffer_type>;
// -- constructors, destructors, and assignment operators --------------------
stream_transport(stream_socket handle, application_type application)
......@@ -92,10 +94,13 @@ public:
template <class Parent>
error init(Parent& parent) {
manager_ = &parent;
max_output_bufs_ = get_or(system().config(), "middleman.max-output-buffers",
defaults::middleman::max_output_buffers);
max_header_bufs_ = get_or(system().config(), "middleman.max-header-buffers",
defaults::middleman::max_header_buffers);
auto& cfg = system().config();
auto max_header_bufs = get_or(cfg, "middleman.max-header-buffers",
defaults::middleman::max_header_buffers);
header_bufs_.reserve(max_header_bufs);
auto max_payload_bufs = get_or(cfg, "middleman.max-payload-buffers",
defaults::middleman::max_payload_buffers);
payload_bufs_.reserve(max_payload_bufs);
if (auto err = worker_.init(*this))
return err;
return none;
......@@ -208,44 +213,44 @@ public:
}
void write_packet(unit_t, span<buffer_type*> buffers) {
// Sanity check
CAF_ASSERT(!buffers.empty());
auto it = buffers.begin();
if (write_queue_.empty())
manager().register_writing();
// move header by itself to keep things sorted.
write_queue_.emplace_back(true, std::move(**it++));
// payload buffers. just write them
for (; it != buffers.end(); ++it)
write_queue_.emplace_back(false, std::move(**it));
// By convention, the first buffer is a header buffer. Every other buffer is
// a payload buffer.
auto i = buffers.begin();
write_queue_.emplace_back(true, std::move(*(*i++)));
while (i != buffers.end())
write_queue_.emplace_back(false, std::move(*(*i++)));
}
// -- buffer recycling -------------------------------------------------------
// -- buffer management ------------------------------------------------------
buffer_type next_header_buffer() {
return next_buffer_impl(free_header_bufs_);
return next_buffer_impl(header_bufs_);
}
buffer_type next_buffer() {
return next_buffer_impl(free_bufs_);
buffer_type next_payload_buffer() {
return next_buffer_impl(payload_bufs_);
}
buffer_type next_buffer_impl(std::deque<buffer_type>& container) {
if (container.empty()) {
private:
// -- utility functions ------------------------------------------------------
static buffer_type next_buffer_impl(buffer_cache_type cache) {
if (cache.empty()) {
return {};
} else {
auto buf = std::move(container.front());
container.pop_front();
return buf;
}
auto buf = std::move(cache.back());
cache.pop_back();
return buf;
}
private:
// -- private member functions -----------------------------------------------
bool write_some() {
CAF_LOG_TRACE(CAF_ARG(handle_.id));
// helper to sort empty buffers back into the right queues
if (write_queue_.empty())
return false;
// Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() {
auto& front = write_queue_.front();
auto& is_header = front.first;
......@@ -253,16 +258,14 @@ private:
written_ = 0;
buf.clear();
if (is_header) {
if (free_header_bufs_.size() < max_header_bufs_)
free_header_bufs_.emplace_back(std::move(buf));
} else if (free_bufs_.size() < max_output_bufs_) {
free_bufs_.emplace_back(std::move(buf));
if (header_bufs_.size() < header_bufs_.capacity())
header_bufs_.emplace_back(std::move(buf));
} else if (payload_bufs_.size() < payload_bufs_.capacity()) {
payload_bufs_.emplace_back(std::move(buf));
}
write_queue_.pop_front();
};
// nothing to write
if (write_queue_.empty())
return false;
// Write buffers from the write_queue_ for as long as possible.
do {
auto& buf = write_queue_.front().second;
CAF_ASSERT(!buf.empty());
......@@ -294,10 +297,8 @@ private:
worker_type worker_;
stream_socket handle_;
std::deque<buffer_type> free_header_bufs_;
std::deque<buffer_type> free_bufs_;
size_t max_output_bufs_;
size_t max_header_bufs_;
buffer_cache_type header_bufs_;
buffer_cache_type payload_bufs_;
buffer_type read_buf_;
std::deque<std::pair<bool, buffer_type>> write_queue_;
......
......@@ -53,7 +53,7 @@ error application::write_message(
CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(ptr->msg != nullptr);
CAF_LOG_TRACE(CAF_ARG2("content", ptr->msg->content()));
auto payload_prefix = writer.next_buffer();
auto payload_prefix = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{system(), payload_prefix};
const auto& src = ptr->msg->sender;
const auto& dst = ptr->receiver;
......@@ -83,7 +83,7 @@ error application::write_message(
void application::resolve(packet_writer& writer, string_view path,
const actor& listener) {
CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(listener));
auto payload = writer.next_buffer();
auto payload = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{&executor_, payload};
if (auto err = sink(path)) {
CAF_LOG_ERROR("unable to serialize path" << CAF_ARG(err));
......@@ -109,7 +109,7 @@ void application::new_proxy(packet_writer& writer, actor_id id) {
void application::local_actor_down(packet_writer& writer, actor_id id,
error reason) {
auto payload = writer.next_buffer();
auto payload = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{system(), payload};
if (auto err = sink(reason))
CAF_RAISE_ERROR("unable to serialize an error");
......@@ -304,7 +304,7 @@ error application::handle_resolve_request(packet_writer& writer, header rec_hdr,
aid = 0;
}
// TODO: figure out how to obtain messaging interface.
auto payload = writer.next_buffer();
auto payload = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{&executor_, payload};
if (auto err = sink(aid, ifs))
return err;
......@@ -362,7 +362,7 @@ error application::handle_monitor_message(packet_writer& writer,
});
} else {
error reason = exit_reason::unknown;
auto payload = writer.next_buffer();
auto payload = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{&executor_, payload};
if (auto err = sink(reason))
return err;
......
......@@ -23,7 +23,8 @@ namespace defaults {
namespace middleman {
const size_t max_output_buffers = 100;
const size_t max_payload_buffers = 100;
const size_t max_header_buffers = 10;
} // namespace middleman
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/net/packet_writer.hpp"
namespace caf {
namespace net {
packet_writer::~packet_writer() {
// nop
}
} // namespace net
} // namespace caf
......@@ -111,7 +111,7 @@ struct fixture : test_coordinator_fixture<>,
CAF_FAIL("unexpected function call");
}
buffer_type next_buffer() override {
buffer_type next_payload_buffer() override {
return {};
}
......
......@@ -138,11 +138,11 @@ public:
return *this;
}
std::vector<byte> next_buffer() {
std::vector<byte> next_header_buffer() {
return {};
}
std::vector<byte> next_header_buffer() {
std::vector<byte> next_payload_buffer() {
return {};
}
......
......@@ -138,11 +138,11 @@ struct dummy_transport {
return *this;
}
buffer_type next_buffer() {
buffer_type next_header_buffer() {
return {};
}
buffer_type next_header_buffer() {
buffer_type next_payload_buffer() {
return {};
}
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment