Unverified Commit 3e4f2dfa authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #1488

Drop unnecessary arguments in octet_stream::policy
parents bc88f45e 19ea85cd
...@@ -82,10 +82,9 @@ public: ...@@ -82,10 +82,9 @@ public:
connection_handle conn) override { connection_handle conn) override {
auto app = net::http::router::make(routes_); auto app = net::http::router::make(routes_);
auto serv = net::http::server::make(std::move(app)); auto serv = net::http::server::make(std::move(app));
auto fd = conn.fd();
auto transport = Transport::make(std::move(conn), std::move(serv)); auto transport = Transport::make(std::move(conn), std::move(serv));
transport->max_consecutive_reads(max_consecutive_reads_); transport->max_consecutive_reads(max_consecutive_reads_);
transport->active_policy().accept(fd); transport->active_policy().accept();
auto res = net::socket_manager::make(mpx, std::move(transport)); auto res = net::socket_manager::make(mpx, std::move(transport));
mpx->watch(res->as_disposable()); mpx->watch(res->as_disposable());
return res; return res;
......
...@@ -102,9 +102,8 @@ private: ...@@ -102,9 +102,8 @@ private:
auto bridge = bridge_t::make(std::move(a2s_pull), std::move(s2a_push)); auto bridge = bridge_t::make(std::move(a2s_pull), std::move(s2a_push));
auto bridge_ptr = bridge.get(); auto bridge_ptr = bridge.get();
auto impl = framing::make(std::move(bridge)); auto impl = framing::make(std::move(bridge));
auto fd = conn.fd();
auto transport = transport_t::make(std::move(conn), std::move(impl)); auto transport = transport_t::make(std::move(conn), std::move(impl));
transport->active_policy().connect(fd); transport->active_policy().connect();
auto ptr = socket_manager::make(cfg.mpx, std::move(transport)); auto ptr = socket_manager::make(cfg.mpx, std::move(transport));
bridge_ptr->self_ref(ptr->as_disposable()); bridge_ptr->self_ref(ptr->as_disposable());
cfg.mpx->start(ptr); cfg.mpx->start(ptr);
......
...@@ -99,10 +99,9 @@ public: ...@@ -99,10 +99,9 @@ public:
auto bridge = bridge_t::make(producer_); auto bridge = bridge_t::make(producer_);
auto bridge_ptr = bridge.get(); auto bridge_ptr = bridge.get();
auto impl = net::lp::framing::make(std::move(bridge)); auto impl = net::lp::framing::make(std::move(bridge));
auto fd = conn.fd();
auto transport = Transport::make(std::move(conn), std::move(impl)); auto transport = Transport::make(std::move(conn), std::move(impl));
transport->max_consecutive_reads(max_consecutive_reads_); transport->max_consecutive_reads(max_consecutive_reads_);
transport->active_policy().accept(fd); transport->active_policy().accept();
auto mgr = net::socket_manager::make(mpx, std::move(transport)); auto mgr = net::socket_manager::make(mpx, std::move(transport));
bridge_ptr->self_ref(mgr->as_disposable()); bridge_ptr->self_ref(mgr->as_disposable());
return mgr; return mgr;
......
...@@ -18,25 +18,28 @@ class CAF_NET_EXPORT policy { ...@@ -18,25 +18,28 @@ class CAF_NET_EXPORT policy {
public: public:
virtual ~policy(); virtual ~policy();
/// Returns the handle for the managed socket.
virtual stream_socket handle() const = 0;
/// Reads data from the socket into the buffer. /// Reads data from the socket into the buffer.
virtual ptrdiff_t read(stream_socket x, byte_span buf); virtual ptrdiff_t read(byte_span buf) = 0;
/// Writes data from the buffer to the socket. /// Writes data from the buffer to the socket.
virtual ptrdiff_t write(stream_socket x, const_byte_span buf); virtual ptrdiff_t write(const_byte_span buf) = 0;
/// Returns the last socket error on this thread. /// Returns the last socket error on this thread.
virtual errc last_error(stream_socket, ptrdiff_t); virtual errc last_error(ptrdiff_t) = 0;
/// Checks whether connecting a non-blocking socket was successful. /// Checks whether connecting a non-blocking socket was successful.
virtual ptrdiff_t connect(stream_socket x); virtual ptrdiff_t connect() = 0;
/// Convenience function that always returns 1. Exists to make writing code /// Convenience function that always returns 1. Exists to make writing code
/// against multiple policies easier by providing the same interface. /// against multiple policies easier by providing the same interface.
virtual ptrdiff_t accept(stream_socket); virtual ptrdiff_t accept() = 0;
/// Returns the number of bytes that are buffered internally and available /// Returns the number of bytes that are buffered internally and available
/// for immediate read. /// for immediate read.
virtual size_t buffered() const noexcept; virtual size_t buffered() const noexcept = 0;
}; };
} // namespace caf::net::octet_stream } // namespace caf::net::octet_stream
...@@ -43,6 +43,29 @@ public: ...@@ -43,6 +43,29 @@ public:
bool shutting_down : 1; bool shutting_down : 1;
}; };
class policy_impl : public octet_stream::policy {
public:
explicit policy_impl(stream_socket fd) : fd(std::move(fd)) {
// nop
}
stream_socket handle() const override;
ptrdiff_t read(byte_span) override;
ptrdiff_t write(const_byte_span) override;
octet_stream::errc last_error(ptrdiff_t) override;
ptrdiff_t connect() override;
ptrdiff_t accept() override;
size_t buffered() const noexcept override;
stream_socket fd;
};
// -- constants -------------------------------------------------------------- // -- constants --------------------------------------------------------------
static constexpr size_t default_buf_size = 4 * 1024; // 4 KiB static constexpr size_t default_buf_size = 4 * 1024; // 4 KiB
...@@ -51,12 +74,14 @@ public: ...@@ -51,12 +74,14 @@ public:
transport(stream_socket fd, upper_layer_ptr up); transport(stream_socket fd, upper_layer_ptr up);
transport(stream_socket fd, upper_layer_ptr up, policy* custom); transport(policy* custom, upper_layer_ptr up);
transport(const transport&) = delete; transport(const transport&) = delete;
transport& operator=(const transport&) = delete; transport& operator=(const transport&) = delete;
~transport();
// -- factories -------------------------------------------------------------- // -- factories --------------------------------------------------------------
static std::unique_ptr<transport> make(stream_socket fd, upper_layer_ptr up); static std::unique_ptr<transport> make(stream_socket fd, upper_layer_ptr up);
...@@ -148,9 +173,6 @@ protected: ...@@ -148,9 +173,6 @@ protected:
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
/// The socket file descriptor.
stream_socket fd_;
/// Stores temporary flags. /// Stores temporary flags.
flags_t flags_; flags_t flags_;
...@@ -190,13 +212,13 @@ protected: ...@@ -190,13 +212,13 @@ protected:
policy* policy_ = nullptr; policy* policy_ = nullptr;
/// Fallback policy. /// Fallback policy.
policy default_policy_; union {
policy_impl default_policy_;
};
/// Setting this to non-null informs the transport to replace `up_` with /// Setting this to non-null informs the transport to replace `up_` with
/// `next_`. /// `next_`.
upper_layer_ptr next_; upper_layer_ptr next_;
// TODO: add [[no_unique_address]] to default_policy_ when switching to C++20.
}; };
} // namespace caf::net::octet_stream } // namespace caf::net::octet_stream
...@@ -26,15 +26,17 @@ public: ...@@ -26,15 +26,17 @@ public:
public: public:
explicit policy_impl(connection conn); explicit policy_impl(connection conn);
ptrdiff_t read(stream_socket, byte_span) override; stream_socket handle() const override;
ptrdiff_t write(stream_socket, const_byte_span) override; ptrdiff_t read(byte_span) override;
octet_stream::errc last_error(stream_socket, ptrdiff_t) override; ptrdiff_t write(const_byte_span) override;
ptrdiff_t connect(stream_socket) override; octet_stream::errc last_error(ptrdiff_t) override;
ptrdiff_t accept(stream_socket) override; ptrdiff_t connect() override;
ptrdiff_t accept() override;
size_t buffered() const noexcept override; size_t buffered() const noexcept override;
...@@ -66,7 +68,7 @@ public: ...@@ -66,7 +68,7 @@ public:
private: private:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
transport(stream_socket fd, connection conn, upper_layer_ptr up); transport(connection conn, upper_layer_ptr up);
policy_impl policy_impl_; policy_impl policy_impl_;
}; };
......
...@@ -98,9 +98,8 @@ private: ...@@ -98,9 +98,8 @@ private:
auto bridge = bridge_t::make(std::move(a2s_pull), std::move(s2a_push)); auto bridge = bridge_t::make(std::move(a2s_pull), std::move(s2a_push));
auto bridge_ptr = bridge.get(); auto bridge_ptr = bridge.get();
auto impl = client::make(std::move(cfg.hs), std::move(bridge)); auto impl = client::make(std::move(cfg.hs), std::move(bridge));
auto fd = conn.fd();
auto transport = transport_t::make(std::move(conn), std::move(impl)); auto transport = transport_t::make(std::move(conn), std::move(impl));
transport->active_policy().connect(fd); transport->active_policy().connect();
auto ptr = socket_manager::make(cfg.mpx, std::move(transport)); auto ptr = socket_manager::make(cfg.mpx, std::move(transport));
bridge_ptr->self_ref(ptr->as_disposable()); bridge_ptr->self_ref(ptr->as_disposable());
cfg.mpx->start(ptr); cfg.mpx->start(ptr);
......
...@@ -129,10 +129,9 @@ public: ...@@ -129,10 +129,9 @@ public:
auto app = bridge_t::make(on_request_, producer_); auto app = bridge_t::make(on_request_, producer_);
auto app_ptr = app.get(); auto app_ptr = app.get();
auto ws = net::web_socket::server::make(std::move(app)); auto ws = net::web_socket::server::make(std::move(app));
auto fd = conn.fd();
auto transport = Transport::make(std::move(conn), std::move(ws)); auto transport = Transport::make(std::move(conn), std::move(ws));
transport->max_consecutive_reads(max_consecutive_reads_); transport->max_consecutive_reads(max_consecutive_reads_);
transport->active_policy().accept(fd); transport->active_policy().accept();
auto mgr = net::socket_manager::make(mpx, std::move(transport)); auto mgr = net::socket_manager::make(mpx, std::move(transport));
app_ptr->self_ref(mgr->as_disposable()); app_ptr->self_ref(mgr->as_disposable());
return mgr; return mgr;
......
...@@ -9,30 +9,4 @@ policy::~policy() { ...@@ -9,30 +9,4 @@ policy::~policy() {
// nop // nop
} }
ptrdiff_t policy::read(stream_socket x, byte_span buf) {
return net::read(x, buf);
}
ptrdiff_t policy::write(stream_socket x, const_byte_span buf) {
return net::write(x, buf);
}
errc policy::last_error(stream_socket, ptrdiff_t) {
return last_socket_error_is_temporary() ? errc::temporary : errc::permanent;
}
ptrdiff_t policy::connect(stream_socket x) {
// A connection is established if the OS reports a socket as ready for read
// or write and if there is no error on the socket.
return probe(x) ? 1 : -1;
}
ptrdiff_t policy::accept(stream_socket) {
return 1;
}
size_t policy::buffered() const noexcept {
return 0;
}
} // namespace caf::net::octet_stream } // namespace caf::net::octet_stream
...@@ -9,20 +9,61 @@ ...@@ -9,20 +9,61 @@
#include "caf/net/receive_policy.hpp" #include "caf/net/receive_policy.hpp"
#include "caf/net/socket_manager.hpp" #include "caf/net/socket_manager.hpp"
#include <new>
namespace caf::net::octet_stream { namespace caf::net::octet_stream {
// -- nested types -------------------------------------------------------------
stream_socket transport::policy_impl::handle() const {
return fd;
}
ptrdiff_t transport::policy_impl::read(byte_span buf) {
return net::read(fd, buf);
}
ptrdiff_t transport::policy_impl::write(const_byte_span buf) {
return net::write(fd, buf);
}
errc transport::policy_impl::last_error(ptrdiff_t) {
return last_socket_error_is_temporary() ? errc::temporary : errc::permanent;
}
ptrdiff_t transport::policy_impl::connect() {
// A connection is established if the OS reports a socket as ready for read
// or write and if there is no error on the socket.
return probe(fd) ? 1 : -1;
}
ptrdiff_t transport::policy_impl::accept() {
return 1;
}
size_t transport::policy_impl::buffered() const noexcept {
return 0;
}
// -- constructors, destructors, and assignment operators ---------------------- // -- constructors, destructors, and assignment operators ----------------------
transport::transport(stream_socket fd, upper_layer_ptr up) transport::transport(stream_socket fd, upper_layer_ptr up)
: fd_(fd), up_(std::move(up)), policy_(&default_policy_) { : up_(std::move(up)) {
memset(&flags_, 0, sizeof(flags_t)); memset(&flags_, 0, sizeof(flags_t));
new (&default_policy_) policy_impl(fd);
policy_ = &default_policy_;
} }
transport::transport(stream_socket fd, upper_layer_ptr up, policy* policy) transport::transport(policy* policy, upper_layer_ptr up)
: fd_(fd), up_(std::move(up)), policy_(policy) { : up_(std::move(up)), policy_(policy) {
memset(&flags_, 0, sizeof(flags_t)); memset(&flags_, 0, sizeof(flags_t));
} }
transport::~transport() {
if (policy_ == &default_policy_)
default_policy_.~policy_impl();
}
// -- factories ---------------------------------------------------------------- // -- factories ----------------------------------------------------------------
std::unique_ptr<transport> transport::make(stream_socket fd, std::unique_ptr<transport> transport::make(stream_socket fd,
...@@ -110,7 +151,7 @@ error transport::start(socket_manager* owner) { ...@@ -110,7 +151,7 @@ error transport::start(socket_manager* owner) {
// CAF_LOG_ERROR("nodelay failed: " << err); // CAF_LOG_ERROR("nodelay failed: " << err);
// return err; // return err;
// } // }
if (auto socket_buf_size = send_buffer_size(fd_)) { if (auto socket_buf_size = send_buffer_size(policy_->handle())) {
max_write_buf_size_ = *socket_buf_size; max_write_buf_size_ = *socket_buf_size;
CAF_ASSERT(max_write_buf_size_ > 0); CAF_ASSERT(max_write_buf_size_ > 0);
write_buf_.reserve(max_write_buf_size_ * 2); write_buf_.reserve(max_write_buf_size_ * 2);
...@@ -122,11 +163,11 @@ error transport::start(socket_manager* owner) { ...@@ -122,11 +163,11 @@ error transport::start(socket_manager* owner) {
} }
socket transport::handle() const { socket transport::handle() const {
return fd_; return policy_->handle();
} }
void transport::handle_read_event() { void transport::handle_read_event() {
CAF_LOG_TRACE(CAF_ARG2("socket", fd_.id)); CAF_LOG_TRACE(CAF_ARG2("socket", handle()));
// Resume a write operation if the transport waited for the socket to be // Resume a write operation if the transport waited for the socket to be
// readable from the last call to handle_write_event. // readable from the last call to handle_write_event.
if (flags_.wanted_read_from_write_event) { if (flags_.wanted_read_from_write_event) {
...@@ -150,11 +191,11 @@ void transport::handle_read_event() { ...@@ -150,11 +191,11 @@ void transport::handle_read_event() {
if (read_buf_.size() < max_read_size_) if (read_buf_.size() < max_read_size_)
read_buf_.resize(max_read_size_); read_buf_.resize(max_read_size_);
// Fill up our buffer. // Fill up our buffer.
auto rd = policy_->read(fd_, make_span(read_buf_.data() + buffered_, auto rd = policy_->read(
read_buf_.size() - buffered_)); make_span(read_buf_.data() + buffered_, read_buf_.size() - buffered_));
// Stop if we failed to get more data. // Stop if we failed to get more data.
if (rd < 0) { if (rd < 0) {
switch (policy_->last_error(fd_, rd)) { switch (policy_->last_error(rd)) {
case errc::temporary: case errc::temporary:
case errc::want_read: case errc::want_read:
// Try again later. // Try again later.
...@@ -178,8 +219,8 @@ void transport::handle_read_event() { ...@@ -178,8 +219,8 @@ void transport::handle_read_event() {
if (auto policy_buffered = policy_->buffered(); policy_buffered > 0) { if (auto policy_buffered = policy_->buffered(); policy_buffered > 0) {
if (auto n = buffered_ + policy_buffered; n > read_buf_.size()) if (auto n = buffered_ + policy_buffered; n > read_buf_.size())
read_buf_.resize(n); read_buf_.resize(n);
auto rd2 = policy_->read(fd_, make_span(read_buf_.data() + buffered_, auto rd2
policy_buffered)); = policy_->read(make_span(read_buf_.data() + buffered_, policy_buffered));
if (rd2 != static_cast<ptrdiff_t>(policy_buffered)) { if (rd2 != static_cast<ptrdiff_t>(policy_buffered)) {
CAF_LOG_ERROR("failed to read buffered data from the policy"); CAF_LOG_ERROR("failed to read buffered data from the policy");
return fail(make_error(sec::socket_operation_failed)); return fail(make_error(sec::socket_operation_failed));
...@@ -267,7 +308,7 @@ void transport::fail(const error& reason) { ...@@ -267,7 +308,7 @@ void transport::fail(const error& reason) {
} }
void transport::handle_write_event() { void transport::handle_write_event() {
CAF_LOG_TRACE(CAF_ARG2("socket", fd_.id)); CAF_LOG_TRACE(CAF_ARG2("socket", handle()));
// Resume a read operation if the transport waited for the socket to be // Resume a read operation if the transport waited for the socket to be
// writable from the last call to handle_read_event. // writable from the last call to handle_read_event.
if (flags_.wanted_write_from_read_event) { if (flags_.wanted_write_from_read_event) {
...@@ -291,7 +332,7 @@ void transport::handle_write_event() { ...@@ -291,7 +332,7 @@ void transport::handle_write_event() {
// Allow the upper layer to add extra data to the write buffer. // Allow the upper layer to add extra data to the write buffer.
up_->prepare_send(); up_->prepare_send();
} }
auto write_res = policy_->write(fd_, write_buf_); auto write_res = policy_->write(write_buf_);
if (write_res > 0) { if (write_res > 0) {
write_buf_.erase(write_buf_.begin(), write_buf_.begin() + write_res); write_buf_.erase(write_buf_.begin(), write_buf_.begin() + write_res);
if (write_buf_.empty() && up_->done_sending()) { if (write_buf_.empty() && up_->done_sending()) {
...@@ -305,7 +346,7 @@ void transport::handle_write_event() { ...@@ -305,7 +346,7 @@ void transport::handle_write_event() {
} else if (write_res < 0) { } else if (write_res < 0) {
// Try again later on temporary errors such as EWOULDBLOCK and // Try again later on temporary errors such as EWOULDBLOCK and
// stop writing to the socket on hard errors. // stop writing to the socket on hard errors.
switch (policy_->last_error(fd_, write_res)) { switch (policy_->last_error(write_res)) {
case errc::temporary: case errc::temporary:
case errc::want_write: case errc::want_write:
return; return;
......
...@@ -53,7 +53,7 @@ public: ...@@ -53,7 +53,7 @@ public:
up_->abort(make_error(sec::connection_closed)); up_->abort(make_error(sec::connection_closed));
owner_->deregister(); owner_->deregister();
} else { } else {
auto err = policy_.last_error(policy_.conn.fd(), res); auto err = policy_.last_error(res);
switch (err) { switch (err) {
case octet_stream::errc::want_read: case octet_stream::errc::want_read:
case octet_stream::errc::temporary: case octet_stream::errc::temporary:
...@@ -81,7 +81,7 @@ public: ...@@ -81,7 +81,7 @@ public:
up_->abort(make_error(sec::connection_closed)); up_->abort(make_error(sec::connection_closed));
owner_->deregister(); owner_->deregister();
} else { } else {
switch (policy_.last_error(policy_.conn.fd(), res)) { switch (policy_.last_error(res)) {
case octet_stream::errc::want_write: case octet_stream::errc::want_write:
case octet_stream::errc::temporary: case octet_stream::errc::temporary:
break; break;
...@@ -133,16 +133,19 @@ transport::policy_impl::policy_impl(connection conn) : conn(std::move(conn)) { ...@@ -133,16 +133,19 @@ transport::policy_impl::policy_impl(connection conn) : conn(std::move(conn)) {
// nop // nop
} }
ptrdiff_t transport::policy_impl::read(stream_socket, byte_span buf) { stream_socket transport::policy_impl::handle() const {
return conn.fd();
}
ptrdiff_t transport::policy_impl::read(byte_span buf) {
return conn.read(buf); return conn.read(buf);
} }
ptrdiff_t transport::policy_impl::write(stream_socket, const_byte_span buf) { ptrdiff_t transport::policy_impl::write(const_byte_span buf) {
return conn.write(buf); return conn.write(buf);
} }
octet_stream::errc transport::policy_impl::last_error(stream_socket fd, octet_stream::errc transport::policy_impl::last_error(ptrdiff_t ret) {
ptrdiff_t ret) {
switch (conn.last_error(ret)) { switch (conn.last_error(ret)) {
case errc::none: case errc::none:
case errc::want_accept: case errc::want_accept:
...@@ -151,7 +154,7 @@ octet_stream::errc transport::policy_impl::last_error(stream_socket fd, ...@@ -151,7 +154,7 @@ octet_stream::errc transport::policy_impl::last_error(stream_socket fd,
return octet_stream::errc::temporary; return octet_stream::errc::temporary;
case errc::syscall_failed: case errc::syscall_failed:
// Need to consult errno, which we just leave to the default policy. // Need to consult errno, which we just leave to the default policy.
return octet_stream::policy{}.last_error(fd, ret); return super::policy_impl{conn.fd()}.last_error(ret);
case errc::want_read: case errc::want_read:
return octet_stream::errc::want_read; return octet_stream::errc::want_read;
case errc::want_write: case errc::want_write:
...@@ -163,11 +166,11 @@ octet_stream::errc transport::policy_impl::last_error(stream_socket fd, ...@@ -163,11 +166,11 @@ octet_stream::errc transport::policy_impl::last_error(stream_socket fd,
} }
} }
ptrdiff_t transport::policy_impl::connect(stream_socket) { ptrdiff_t transport::policy_impl::connect() {
return conn.connect(); return conn.connect();
} }
ptrdiff_t transport::policy_impl::accept(stream_socket) { ptrdiff_t transport::policy_impl::accept() {
return conn.accept(); return conn.accept();
} }
...@@ -180,8 +183,7 @@ size_t transport::policy_impl::buffered() const noexcept { ...@@ -180,8 +183,7 @@ size_t transport::policy_impl::buffered() const noexcept {
std::unique_ptr<transport> transport::make(connection conn, std::unique_ptr<transport> transport::make(connection conn,
upper_layer_ptr up) { upper_layer_ptr up) {
// Note: can't use make_unique because the constructor is private. // Note: can't use make_unique because the constructor is private.
auto fd = conn.fd(); auto ptr = new transport(std::move(conn), std::move(up));
auto ptr = new transport(fd, std::move(conn), std::move(up));
return std::unique_ptr<transport>{ptr}; return std::unique_ptr<transport>{ptr};
} }
...@@ -199,8 +201,8 @@ transport::worker_ptr transport::make_client(connection conn, ...@@ -199,8 +201,8 @@ transport::worker_ptr transport::make_client(connection conn,
// -- constructors, destructors, and assignment operators ---------------------- // -- constructors, destructors, and assignment operators ----------------------
transport::transport(stream_socket fd, connection conn, upper_layer_ptr up) transport::transport(connection conn, upper_layer_ptr up)
: super(fd, std::move(up), &policy_impl_), policy_impl_(std::move(conn)) { : super(&policy_impl_, std::move(up)), policy_impl_(std::move(conn)) {
// nop // nop
} }
......
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