Commit fe87a2bc authored by Jakob Otto's avatar Jakob Otto

fix compilation

parent bafc23f5
......@@ -22,8 +22,7 @@
#include "caf/intrusive_ptr.hpp"
namespace caf {
namespace net {
namespace caf::net {
// -- templates ----------------------------------------------------------------
......@@ -33,6 +32,9 @@ class transport_worker;
template <class Transport, class IdType = unit_t>
class transport_worker_dispatcher;
template <class Application>
class stream_transport;
// -- classes ------------------------------------------------------------------
class endpoint_manager;
......@@ -58,5 +60,4 @@ using multiplexer_ptr = std::shared_ptr<multiplexer>;
using socket_manager_ptr = intrusive_ptr<socket_manager>;
using weak_multiplexer_ptr = std::weak_ptr<multiplexer>;
} // namespace net
} // namespace caf
} // namespace caf::net
......@@ -5,7 +5,7 @@
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* 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 *
......@@ -26,6 +26,7 @@
#include "caf/logger.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/transport_base.hpp"
......@@ -34,22 +35,22 @@
#include "caf/span.hpp"
#include "caf/variant.hpp"
namespace caf {
namespace net {
namespace caf::net {
template <class Application>
using transport_base_type = transport_base<transport_worker<Application>,
using transport_base_type = transport_base<stream_transport<Application>,
transport_worker<Application>,
stream_socket, Application, unit_t>;
/// Implements a stream_transport that manages a stream socket.
template <class Application>
class stream_transport : transport_base_type<Application> {
class stream_transport : public transport_base_type<Application> {
public:
// -- member types -----------------------------------------------------------
using application_type = Application;
using transport_type = stream_transport;
using transport_type = stream_transport<application_type>;
using worker_type = transport_worker<application_type>;
......@@ -57,10 +58,12 @@ public:
using buffer_cache_type = std::vector<buffer_type>;
using super = transport_base_type<application_type>;
// -- constructors, destructors, and assignment operators --------------------
stream_transport(stream_socket handle, application_type application)
: transport_base_type<application_type>(handle, std::move(application)) {
: super(handle, std::move(application)), written_(0) {
// nop
}
......@@ -68,32 +71,34 @@ public:
error init(endpoint_manager& parent) override {
// call init function from base class
transport_base_type<application_type>::init(parent);
return super::init(parent);
}
bool handle_read_event(endpoint_manager&) override {
auto buf = read_buf_.data() + collected_;
size_t len = read_threshold_ - collected_;
CAF_LOG_TRACE(CAF_ARG(this->handle_.id) << CAF_ARG(len));
auto ret = read(handle_, make_span(buf, len));
auto buf = super::read_buf_.data() + super::collected_;
size_t len = super::read_threshold_ - super::collected_;
CAF_LOG_TRACE(CAF_ARG(super::handle().id) << CAF_ARG(len));
auto ret = read(super::handle_, make_span(buf, len));
// Update state.
if (auto num_bytes = get_if<size_t>(&ret)) {
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
collected_ += *num_bytes;
if (collected_ >= read_threshold_) {
if (auto err = worker_.handle_data(*this, read_buf_)) {
CAF_LOG_DEBUG(CAF_ARG(len)
<< CAF_ARG(super::handle_.id) << CAF_ARG(*num_bytes));
super::collected_ += *num_bytes;
if (super::collected_ >= super::read_threshold_) {
if (auto err = super::next_layer_.handle_data(*this,
super::read_buf_)) {
CAF_LOG_WARNING("handle_data failed:" << CAF_ARG(err));
return false;
}
// TODO: Is this the proper way or would
// `transport_base_type<Application>::prepare_next_read()` be better?
this->prepare_next_read();
super::prepare_next_read();
}
} else {
auto err = get<sec>(ret);
if (err != sec::unavailable_or_would_block) {
CAF_LOG_DEBUG("receive failed" << CAF_ARG(err));
worker_.handle_error(err);
super::next_layer_.handle_error(err);
return false;
}
}
......@@ -107,7 +112,7 @@ public:
// TODO: dont read all messages at once - get one by one.
for (auto msg = parent.next_message(); msg != nullptr;
msg = parent.next_message()) {
worker_.write_message(*this, std::move(msg));
super::next_layer_.write_message(*this, std::move(msg));
}
// Write prepared data.
return write_some();
......@@ -115,45 +120,46 @@ public:
void write_packet(unit_t, span<buffer_type*> buffers) override {
CAF_ASSERT(!buffers.empty());
if (write_queue_.empty())
this->manager().register_writing();
if (super::write_queue_.empty())
super::manager().register_writing();
// 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++)));
super::write_queue_.emplace_back(true, std::move(*(*i++)));
while (i != buffers.end())
write_queue_.emplace_back(false, std::move(*(*i++)));
super::write_queue_.emplace_back(false, std::move(*(*i++)));
}
private:
// -- utility functions ------------------------------------------------------
bool write_some() {
CAF_LOG_TRACE(CAF_ARG(handle_.id));
CAF_LOG_TRACE(CAF_ARG(super::handle_.id));
// Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() {
auto& front = write_queue_.front();
auto& front = super::write_queue_.front();
auto& is_header = front.first;
auto& buf = front.second;
written_ = 0;
buf.clear();
if (is_header) {
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));
if (super::header_bufs_.size() < super::header_bufs_.capacity())
super::header_bufs_.emplace_back(std::move(buf));
} else if (super::payload_bufs_.size()
< super::payload_bufs_.capacity()) {
super::payload_bufs_.emplace_back(std::move(buf));
}
write_queue_.pop_front();
super::write_queue_.pop_front();
};
// Write buffers from the write_queue_ for as long as possible.
while (!write_queue_.empty()) {
auto& buf = write_queue_.front().second;
while (!super::write_queue_.empty()) {
auto& buf = super::write_queue_.front().second;
CAF_ASSERT(!buf.empty());
auto data = buf.data() + written_;
auto len = buf.size() - written_;
auto write_ret = write(this->next_layer_, make_span(data, len));
auto write_ret = write(super::handle(), make_span(data, len));
if (auto num_bytes = get_if<size_t>(&write_ret)) {
CAF_LOG_DEBUG(CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
CAF_LOG_DEBUG(CAF_ARG(super::handle_.id) << CAF_ARG(*num_bytes));
if (*num_bytes + written_ >= buf.size()) {
recycle();
written_ = 0;
......@@ -165,7 +171,7 @@ private:
auto err = get<sec>(write_ret);
if (err != sec::unavailable_or_would_block) {
CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
next_layer_.handle_error(err);
super::next_layer_.handle_error(err);
return false;
}
return true;
......@@ -174,23 +180,7 @@ private:
return false;
}
buffer_cache_type header_bufs_;
buffer_cache_type payload_bufs_;
buffer_type read_buf_;
std::deque<std::pair<bool, buffer_type>> write_queue_;
// TODO implement retries using this member!
// size_t max_consecutive_reads_;
size_t read_threshold_;
size_t collected_;
size_t max_;
receive_policy_flag rd_flag_;
size_t written_;
endpoint_manager* manager_;
};
} // namespace net
} // namespace caf
} // namespace caf::net
......@@ -5,7 +5,7 @@
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* 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 *
......@@ -33,10 +33,10 @@
#include "caf/span.hpp"
#include "caf/variant.hpp"
namespace caf {
namespace net {
namespace caf::net {
template <class NextLayer, class Handle, class Application, class IdType>
template <class Transport, class NextLayer, class Handle, class Application,
class IdType>
class transport_base {
public:
// -- member types -----------------------------------------------------------
......@@ -45,6 +45,8 @@ public:
using handle_type = Handle;
using transport_type = Transport;
using application_type = Application;
using id_type = IdType;
......@@ -82,8 +84,8 @@ public:
return next_layer_.application();
}
transport_base& transport() {
return *this;
transport_type& transport() {
return *reinterpret_cast<transport_type*>(this);
}
endpoint_manager& manager() {
......@@ -216,5 +218,4 @@ protected:
endpoint_manager* manager_;
};
} // namespace net
} // namespace caf
} // namespace caf::net
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