Commit 4a069f4b authored by Dominik Charousset's avatar Dominik Charousset

Drastically reduce complexity of the multiplexer

When we started with caf-net in the incubator, socket managers had
minimal control flow and the read and write handlers could simply return
whether they wished to be called again later again. With SSL and its
`want_read` and `want_write` shenanigans, things got a whole lot more
complicated and the old return-next-operation model stopped being a good
fit. And finally, asynchronous events that trigger reads and writes
without prior socket events made this approach pretty much fell apart
and there were in fact two paths to influence the event mask for a
socket manager: the return value of the handlers and
`register_(reading|writing)` functions.

With this redesign, read and write handlers no longer return anything.
All bookkeeping for the event mask exclusively relies on
`register_(reading|writing)` functions. The new design also drops
`resume_reading` and simply adds more capabilities to lower layers to
(re-) start reading at any time using actions.
parent 787d4259
......@@ -21,10 +21,6 @@ public:
using factory_type = Factory;
using read_result = typename socket_event_layer::read_result;
using write_result = typename socket_event_layer::write_result;
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
......@@ -54,40 +50,34 @@ public:
return none;
}
read_result handle_read_event() override {
void handle_read_event() override {
CAF_LOG_TRACE("");
if (auto x = accept(fd_)) {
socket_manager_ptr child = factory_.make(owner_->mpx_ptr(), *x);
if (!child) {
CAF_LOG_ERROR("factory failed to create a new child");
return read_result::abort;
owner_->deregister();
return;
}
if (auto err = child->init(cfg_)) {
CAF_LOG_ERROR("failed to initialize new child:" << err);
return read_result::abort;
owner_->deregister();
return;
}
if (limit_ == 0) {
return read_result::again;
} else {
return ++accepted_ < limit_ ? read_result::again : read_result::stop;
if (limit_ != 0 && ++accepted_ == limit_) {
// TODO: ask owner to close socket.
owner_->deregister();
return;
}
} else {
CAF_LOG_ERROR("accept failed:" << x.error());
return read_result::stop;
owner_->deregister();
}
}
read_result handle_buffered_data() override {
return read_result::again;
}
read_result handle_continue_reading() override {
return read_result::again;
}
write_result handle_write_event() override {
void handle_write_event() override {
CAF_LOG_ERROR("connection_acceptor received write event");
return write_result::stop;
owner_->deregister_writing();
}
void abort(const error& reason) override {
......
......@@ -5,27 +5,28 @@
#pragma once
#include "caf/async/consumer.hpp"
#include "caf/detail/atomic_ref_counted.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/ref_counted.hpp"
namespace caf::net {
/// Connects a socket manager to an asynchronous consumer resource. Whenever new
/// data becomes ready, the adapter registers the socket manager for writing.
template <class Buffer>
class consumer_adapter final : public ref_counted, public async::consumer {
class consumer_adapter final : public detail::atomic_ref_counted,
public async::consumer {
public:
using buf_ptr = intrusive_ptr<Buffer>;
using ptr_type = intrusive_ptr<consumer_adapter>;
void on_producer_ready() override {
// nop
}
void on_producer_wakeup() override {
mgr_->mpx().schedule_fn([adapter = strong_this()] { //
adapter->on_wakeup();
});
mgr_->schedule(do_wakeup_);
}
void ref_consumer() const noexcept override {
......@@ -44,22 +45,20 @@ public:
void cancel() {
buf_->cancel();
buf_ = nullptr;
mgr_ = nullptr;
do_wakeup_.dispose();
do_wakeup_ = nullptr;
}
bool has_data() const noexcept {
return buf_->has_data();
}
/// Tries to open the resource for reading.
/// @returns a connected adapter that reads from the resource on success,
/// `nullptr` otherwise.
template <class Resource>
static intrusive_ptr<consumer_adapter>
try_open(socket_manager* owner, Resource src) {
CAF_ASSERT(owner != nullptr);
if (auto buf = src.try_open()) {
using ptr_type = intrusive_ptr<consumer_adapter>;
auto adapter = ptr_type{new consumer_adapter(owner, buf), false};
static ptr_type make(buf_ptr buf, socket_manager_ptr mgr, action do_wakeup) {
if (buf) {
auto adapter = ptr_type{new consumer_adapter(buf, mgr,
std::move(do_wakeup)), //
false};
buf->set_consumer(adapter);
return adapter;
} else {
......@@ -67,17 +66,11 @@ public:
}
}
friend void intrusive_ptr_add_ref(const consumer_adapter* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const consumer_adapter* ptr) noexcept {
ptr->deref();
}
private:
consumer_adapter(socket_manager* owner, buf_ptr buf)
: mgr_(owner), buf_(std::move(buf)) {
consumer_adapter(buf_ptr buf, socket_manager_ptr mgr, action do_wakeup)
: buf_(std::move(buf)),
mgr_(std::move(mgr)),
do_wakeup_(std::move(do_wakeup)) {
// nop
}
......@@ -91,8 +84,10 @@ private:
}
}
intrusive_ptr<socket_manager> mgr_;
intrusive_ptr<Buffer> buf_;
socket_manager_ptr mgr_;
action do_wakeup_;
action do_cancel_;
};
template <class T>
......
......@@ -5,6 +5,7 @@
#pragma once
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
namespace caf::net {
......@@ -17,12 +18,18 @@ public:
/// Queries whether the output device can accept more data straight away.
[[nodiscard]] virtual bool can_send_more() const noexcept = 0;
/// Halt receiving of additional bytes or messages.
virtual void suspend_reading() = 0;
/// Queries whether the lower layer is currently configured to halt receiving
/// of additional bytes or messages.
[[nodiscard]] virtual bool stopped_reading() const noexcept = 0;
[[nodiscard]] virtual bool is_reading() const noexcept = 0;
/// Shuts down any connection or session gracefully. Any pending data gets
/// flushed before closing the socket.
virtual void close() = 0;
/// Shuts down any connection or session du to an error. Any pending data gets
/// flushed before closing the socket. Protocols with a dedicated closing
/// handshake such as WebSocket may send the close reason to the peer.
virtual void close(const error& reason);
};
} // namespace caf::net
......@@ -24,11 +24,6 @@ public:
/// function to decide whether it has to wait for write events on the socket.
[[nodiscard]] virtual bool done_sending() = 0;
/// Called by the lower layer after some event triggered re-registering the
/// socket manager for read operations after it has been stopped previously
/// by the read policy. May restart consumption of bytes or messages.
virtual void continue_reading() = 0;
/// Called by the lower layer for cleaning up any state in case of an error.
virtual void abort(const error& reason) = 0;
};
......
......@@ -18,6 +18,12 @@ class CAF_NET_EXPORT lower_layer : public generic_lower_layer {
public:
virtual ~lower_layer();
/// Start or re-start reading data from the client.
virtual void request_messages() = 0;
/// Stops reading messages until calling `request_messages`.
virtual void suspend_reading() = 0;
/// Sends the next header to the client.
virtual bool
send_header(context ctx, status code, const header_fields_map& fields)
......
......@@ -80,9 +80,13 @@ public:
bool can_send_more() const noexcept override;
void suspend_reading() override;
bool is_reading() const noexcept override;
void close() override;
bool stopped_reading() const noexcept override;
void request_messages() override;
void suspend_reading() override;
bool send_header(context, status code,
const header_fields_map& fields) override;
......@@ -106,8 +110,6 @@ public:
bool done_sending() override;
void continue_reading() override;
ptrdiff_t consume(byte_span input, byte_span) override;
private:
......
......@@ -81,8 +81,6 @@ public:
ptrdiff_t consume(byte_span buffer, byte_span delta) override;
void continue_reading() override;
bool prepare_send() override;
bool done_sending() override;
......@@ -95,7 +93,7 @@ public:
void suspend_reading() override;
bool stopped_reading() const noexcept override;
bool is_reading() const noexcept override;
void begin_message() override;
......@@ -103,9 +101,7 @@ public:
bool end_message() override;
void send_close_message() override;
void send_close_message(const error& reason) override;
void close() override;
// -- utility functions ------------------------------------------------------
......
......@@ -116,9 +116,9 @@ public:
auto [again, consumed] = in_->pull(async::delay_errors, 1, helper);
if (!again) {
if (helper.err) {
down_->send_close_message(helper.err);
down_->close(helper.err);
} else {
down_->send_close_message();
down_->close();
}
in_ = nullptr;
} else if (helper.aborted) {
......
......@@ -47,6 +47,9 @@ public:
/// Pulls messages from the transport until calling `suspend_reading`.
virtual void request_messages() = 0;
/// Stops reading messages until calling `request_messages`.
virtual void suspend_reading() = 0;
/// Prepares the layer for an outgoing message, e.g., by allocating an
/// output buffer as necessary.
virtual void begin_message() = 0;
......@@ -62,17 +65,6 @@ public:
/// @note When returning `false`, clients must also call
/// `down.set_read_error(...)` with an appropriate error code.
virtual bool end_message() = 0;
/// Inform the remote endpoint that no more messages will arrive.
/// @note Not every protocol has a dedicated close message. Some
/// implementation may simply do nothing.
virtual void send_close_message() = 0;
/// Inform the remote endpoint that no more messages will arrive because of
/// an error.
/// @note Not every protocol has a dedicated close message. Some
/// implementation may simply do nothing.
virtual void send_close_message(const error& reason) = 0;
};
} // namespace caf::net::message_oriented
......@@ -73,10 +73,10 @@ public:
size_t num_socket_managers() const noexcept;
/// Returns the index of `mgr` in the pollset or `-1`.
ptrdiff_t index_of(const socket_manager_ptr& mgr);
ptrdiff_t index_of(const socket_manager_ptr& mgr) const noexcept;
/// Returns the index of `fd` in the pollset or `-1`.
ptrdiff_t index_of(socket fd);
ptrdiff_t index_of(socket fd) const noexcept;
/// Returns the owning @ref middleman instance.
middleman& owner();
......@@ -89,22 +89,6 @@ public:
// -- thread-safe signaling --------------------------------------------------
/// Registers `mgr` for read events.
/// @thread-safe
void register_reading(const socket_manager_ptr& mgr);
/// Registers `mgr` for write events.
/// @thread-safe
void register_writing(const socket_manager_ptr& mgr);
/// Triggers a continue reading event for `mgr`.
/// @thread-safe
void continue_reading(const socket_manager_ptr& mgr);
/// Triggers a continue writing event for `mgr`.
/// @thread-safe
void continue_writing(const socket_manager_ptr& mgr);
/// Schedules a call to `mgr->handle_error(sec::discarded)`.
/// @thread-safe
void discard(const socket_manager_ptr& mgr);
......@@ -136,6 +120,29 @@ public:
/// @thread-safe
void shutdown();
// -- callbacks for socket managers ------------------------------------------
/// Registers `mgr` for read events.
void register_reading(socket_manager* mgr);
/// Registers `mgr` for write events.
void register_writing(socket_manager* mgr);
/// Deregisters `mgr` from read events.
void deregister_reading(socket_manager* mgr);
/// Deregisters `mgr` from write events.
void deregister_writing(socket_manager* mgr);
/// Deregisters @p mgr from read and write events.
void deregister(socket_manager* mgr);
/// Queries whether `mgr` is currently registered for reading.
bool is_reading(const socket_manager* mgr) const noexcept;
/// Queries whether `mgr` is currently registered for writing.
bool is_writing(const socket_manager* mgr) const noexcept;
// -- control flow -----------------------------------------------------------
/// Polls I/O activity once and runs all socket event handlers that become
......@@ -160,15 +167,12 @@ protected:
/// Handles an I/O event on given manager.
void handle(const socket_manager_ptr& mgr, short events, short revents);
/// Transfers socket ownership from one manager to another.
void do_handover(const socket_manager_ptr& mgr);
/// Returns a change entry for the socket at given index. Lazily creates a new
/// entry before returning if necessary.
poll_update& update_for(ptrdiff_t index);
/// Returns a change entry for the socket of the manager.
poll_update& update_for(const socket_manager_ptr& mgr);
poll_update& update_for(socket_manager* mgr);
/// Writes `opcode` and pointer to `mgr` the the pipe for handling an event
/// later via the pollset updater.
......@@ -182,13 +186,7 @@ protected:
}
/// Queries the currently active event bitmask for `mgr`.
short active_mask_of(const socket_manager_ptr& mgr);
/// Queries whether `mgr` is currently registered for reading.
bool is_reading(const socket_manager_ptr& mgr);
/// Queries whether `mgr` is currently registered for writing.
bool is_writing(const socket_manager_ptr& mgr);
short active_mask_of(const socket_manager* mgr) const noexcept;
// -- member variables -------------------------------------------------------
......@@ -226,12 +224,6 @@ private:
void do_register_reading(const socket_manager_ptr& mgr);
void do_register_writing(const socket_manager_ptr& mgr);
void do_continue_reading(const socket_manager_ptr& mgr);
void do_continue_writing(const socket_manager_ptr& mgr);
void do_discard(const socket_manager_ptr& mgr);
void do_shutdown_reading(const socket_manager_ptr& mgr);
......
......@@ -25,10 +25,6 @@ public:
using msg_buf = std::array<std::byte, sizeof(intptr_t) + 1>;
enum class code : uint8_t {
register_reading,
continue_reading,
register_writing,
continue_writing,
init_manager,
discard_manager,
shutdown_reading,
......@@ -49,18 +45,15 @@ public:
error init(socket_manager* owner, const settings& cfg) override;
read_result handle_read_event() override;
void handle_read_event() override;
read_result handle_buffered_data() override;
read_result handle_continue_reading() override;
write_result handle_write_event() override;
void handle_write_event() override;
void abort(const error& reason) override;
private:
pipe_socket fd_;
socket_manager* owner_ = nullptr;
multiplexer* mpx_ = nullptr;
msg_buf buf_;
size_t buf_size_ = 0;
......
......@@ -8,36 +8,35 @@
#include <new>
#include "caf/async/producer.hpp"
#include "caf/detail/atomic_ref_counted.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/subscription.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/ref_counted.hpp"
namespace caf::net {
/// Connects a socket manager to an asynchronous producer resource.
template <class Buffer>
class producer_adapter final : public ref_counted, public async::producer {
class producer_adapter final : public detail::atomic_ref_counted,
public async::producer {
public:
using atomic_count = std::atomic<size_t>;
using buf_ptr = intrusive_ptr<Buffer>;
using value_type = typename Buffer::value_type;
using ptr_type = intrusive_ptr<producer_adapter>;
void on_consumer_ready() override {
// nop
}
void on_consumer_cancel() override {
mgr_->mpx().schedule_fn([adapter = strong_this()] { //
adapter->on_cancel();
});
mgr_->schedule(do_cancel_);
}
void on_consumer_demand(size_t) override {
mgr_->continue_reading();
mgr_->schedule(do_resume_);
}
void ref_producer() const noexcept override {
......@@ -48,16 +47,13 @@ public:
this->deref();
}
/// Tries to open the resource for writing.
/// @returns a connected adapter that writes to the resource on success,
/// `nullptr` otherwise.
template <class Resource>
static intrusive_ptr<producer_adapter>
try_open(socket_manager* owner, Resource src) {
CAF_ASSERT(owner != nullptr);
if (auto buf = src.try_open()) {
using ptr_type = intrusive_ptr<producer_adapter>;
auto adapter = ptr_type{new producer_adapter(owner, buf), false};
static ptr_type make(buf_ptr buf, socket_manager_ptr mgr, action do_resume,
action do_cancel) {
if (buf) {
auto adapter
= ptr_type{new producer_adapter(buf, mgr, std::move(do_resume),
std::move(do_cancel)),
false};
buf->set_producer(adapter);
return adapter;
} else {
......@@ -88,47 +84,40 @@ public:
void close() {
if (buf_) {
buf_->close();
buf_ = nullptr;
mgr_ = nullptr;
reset();
}
}
void abort(error reason) {
if (buf_) {
buf_->abort(std::move(reason));
buf_ = nullptr;
mgr_ = nullptr;
reset();
}
}
friend void intrusive_ptr_add_ref(const producer_adapter* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const producer_adapter* ptr) noexcept {
ptr->deref();
}
private:
producer_adapter(socket_manager* owner, buf_ptr buf)
: mgr_(owner), buf_(std::move(buf)) {
producer_adapter(buf_ptr buf, socket_manager_ptr mgr, action do_resume,
action do_cancel)
: buf_(std::move(buf)),
mgr_(std::move(mgr)),
do_resume_(std::move(do_resume)),
do_cancel_(std::move(do_cancel)) {
// nop
}
void on_cancel() {
if (buf_) {
mgr_->mpx().shutdown_reading(mgr_);
buf_ = nullptr;
mgr_ = nullptr;
}
}
auto strong_this() {
return intrusive_ptr{this};
void reset() {
buf_ = nullptr;
mgr_ = nullptr;
do_resume_.dispose();
do_resume_ = nullptr;
do_cancel_.dispose();
do_cancel_ = nullptr;
}
intrusive_ptr<socket_manager> mgr_;
intrusive_ptr<Buffer> buf_;
intrusive_ptr<socket_manager> mgr_;
action do_resume_;
action do_cancel_;
};
template <class T>
......
......@@ -16,58 +16,14 @@ class CAF_NET_EXPORT socket_event_layer {
public:
virtual ~socket_event_layer();
/// Encodes how to proceed after a read operation.
enum class read_result {
/// Indicates that a manager wants to read again later.
again,
/// Indicates that a manager wants to stop reading until explicitly resumed.
stop,
/// Indicates that a manager wants to write to the socket instead of reading
/// from the socket.
want_write,
/// Indicates that the manager no longer reads from the socket.
close,
/// Indicates that the manager encountered a fatal error and stops both
/// reading and writing.
abort,
/// Indicates that a manager is done with the socket and hands ownership to
/// another manager.
handover,
};
/// Encodes how to proceed after a write operation.
enum class write_result {
/// Indicates that a manager wants to read again later.
again,
/// Indicates that a manager wants to stop reading until explicitly resumed.
stop,
/// Indicates that a manager wants to read from the socket instead of
/// writing to the socket.
want_read,
/// Indicates that the manager no longer writes to the socket.
close,
/// Indicates that the manager encountered a fatal error and stops both
/// reading and writing.
abort,
/// Indicates that a manager is done with the socket and hands ownership to
/// another manager.
handover,
};
/// Initializes the layer.
virtual error init(socket_manager* owner, const settings& cfg) = 0;
/// Handles a read event on the managed socket.
virtual read_result handle_read_event() = 0;
/// Handles internally buffered data.
virtual read_result handle_buffered_data() = 0;
/// Handles a request to continue reading on the socket.
virtual read_result handle_continue_reading() = 0;
virtual void handle_read_event() = 0;
/// Handles a write event on the managed socket.
virtual write_result handle_write_event() = 0;
virtual void handle_write_event() = 0;
/// Called after returning `handover` from a read or write handler.
virtual bool do_handover(std::unique_ptr<socket_event_layer>& next);
......
......@@ -4,6 +4,7 @@
#pragma once
#include "caf/action.hpp"
#include "caf/actor.hpp"
#include "caf/actor_system.hpp"
#include "caf/callback.hpp"
......@@ -28,10 +29,6 @@ class CAF_NET_EXPORT socket_manager : public ref_counted {
public:
// -- member types -----------------------------------------------------------
using read_result = socket_event_layer::read_result;
using write_result = socket_event_layer::write_result;
using event_handler_ptr = std::unique_ptr<socket_event_layer>;
/// Stores manager-related flags in a single block.
......@@ -122,31 +119,54 @@ public:
return flags_.write_closed;
}
/// Queries whether the manager is registered for reading.
bool is_reading() const noexcept;
/// Queries whether the manager is registered for writing.
bool is_writing() const noexcept;
// -- event loop management --------------------------------------------------
/// Registers the manager for read operations on the @ref multiplexer.
/// @thread-safe
/// Registers the manager for read operations.
void register_reading();
/// Registers the manager for write operations on the @ref multiplexer.
/// @thread-safe
/// Registers the manager for write operations.
void register_writing();
/// Schedules a call to `handle_continue_reading` on the @ref multiplexer.
/// This mechanism allows users to signal changes in the environment to the
/// manager that allow it to make progress, e.g., new demand in asynchronous
/// buffer that allow the manager to push available data downstream. The event
/// is a no-op if the manager is already registered for reading.
/// @thread-safe
void continue_reading();
/// Schedules a call to `handle_continue_reading` on the @ref multiplexer.
/// This mechanism allows users to signal changes in the environment to the
/// manager that allow it to make progress, e.g., new data for writing in an
/// asynchronous buffer. The event is a no-op if the manager is already
/// registered for writing.
/// @thread-safe
void continue_writing();
/// Deregisters the manager from read operations.
void deregister_reading();
/// Deregisters the manager from write operations.
void deregister_writing();
/// Deregisters the manager from both read and write operations.
void deregister();
/// Deregisters the manager from read operations and blocks any future
/// attempts to re-register it.
void shutdown_read();
/// Deregisters the manager from write operations and blocks any future
/// attempts to re-register it.
void shutdown_write();
/// Deregisters the manager from both read and write operations and blocks any
/// future attempts to re-register it.
void shutdown();
// -- callbacks for the handler ----------------------------------------------
/// Schedules a call to `do_handover` on the handler.
void schedule_handover();
/// Schedules @p what to run later.
void schedule(action what);
/// Schedules @p what to run later.
template <class F>
void schedule_fn(F&& what) {
schedule(make_action(std::forward<F>(what)));
}
// -- callbacks for the multiplexer ------------------------------------------
......@@ -160,30 +180,20 @@ public:
error init(const settings& cfg);
/// Called whenever the socket received new data.
read_result handle_read_event();
/// Called after handovers to allow the manager to process any data that is
/// already buffered at the transport policy and thus would not trigger a read
/// event on the socket.
read_result handle_buffered_data();
/// Restarts a socket manager that suspended reads. Calling this member
/// function on active managers is a no-op. This function also should read any
/// data buffered outside of the socket.
read_result handle_continue_reading();
void handle_read_event();
/// Called whenever the socket is allowed to send data.
write_result handle_write_event();
void handle_write_event();
/// Called when the remote side becomes unreachable due to an error.
/// @param code The error code as reported by the operating system.
void handle_error(sec code);
/// Performs a handover to another transport after `handle_read_event` or
/// `handle_read_event` returned `handover`.
[[nodiscard]] bool do_handover();
private:
// -- utility functions ------------------------------------------------------
socket_manager_ptr strong_this();
// -- member variables -------------------------------------------------------
/// Stores the socket file descriptor. The socket manager automatically closes
......
......@@ -45,6 +45,10 @@ class CAF_NET_EXPORT lower_layer : public generic_lower_layer {
public:
virtual ~lower_layer();
/// Queries whether the transport is currently configured to read from its
/// socket.
virtual bool is_reading() const noexcept = 0;
/// Configures threshold for the next receive operations. Policies remain
/// active until calling this function again.
/// @warning Calling this function in `consume` invalidates both byte spans.
......
......@@ -85,6 +85,9 @@ public:
/// Stores whether we left a write handler due to want_read.
bool wanted_write_from_read_event : 1;
/// Stores whether we shutdown write on the socket after flushing.
bool shutting_down : 1;
};
// -- constants --------------------------------------------------------------
......@@ -118,9 +121,9 @@ public:
bool end_output() override;
void suspend_reading() override;
bool is_reading() const noexcept override;
bool stopped_reading() const noexcept override;
void close() override;
// -- properties -------------------------------------------------------------
......@@ -152,17 +155,23 @@ public:
error init(socket_manager* owner, const settings& config) override;
read_result handle_read_event() override;
read_result handle_buffered_data() override;
void handle_read_event() override;
write_result handle_write_event() override;
read_result handle_continue_reading() override;
void handle_write_event() override;
void abort(const error& reason) override;
protected:
// -- utility functions ------------------------------------------------------
/// Consumes as much data from the buffer as possible.
void handle_buffered_data();
/// Calls abort on the upper layer and deregisters the transport from events.
void fail(const error& reason);
// -- member variables -------------------------------------------------------
/// The socket file descriptor.
stream_socket fd_;
......
......@@ -74,8 +74,6 @@ public:
ptrdiff_t consume(byte_span buffer, byte_span delta) override;
void continue_reading() override;
bool prepare_send() override;
bool done_sending() override;
......
......@@ -79,6 +79,10 @@ public:
}
};
bool running() const noexcept {
return in_ || out_;
}
// -- implementation of web_socket::lower_layer ------------------------------
error init(net::socket_manager* mgr, web_socket::lower_layer* down,
......@@ -86,38 +90,48 @@ public:
down_ = down;
auto [err, pull, push] = conn_->on_request(cfg);
if (!err) {
in_ = consumer_type::try_open(mgr, pull);
out_ = producer_type::try_open(mgr, push);
CAF_ASSERT(in_ != nullptr);
CAF_ASSERT(out_ != nullptr);
auto do_wakeup = make_action([this] {
prepare_send();
if (!running())
down_->close();
});
auto do_resume = make_action([this] { down_->request_messages(); });
auto do_cancel = make_action([this] {
out_ = nullptr;
if (!running())
down_->close();
});
in_ = consumer_type::make(pull.try_open(), mgr, std::move(do_wakeup));
out_ = producer_type::make(push.try_open(), mgr, std::move(do_resume),
std::move(do_cancel));
conn_ = nullptr;
return none;
if (running())
return none;
else
return make_error(sec::runtime_error,
"cannot init flow bridge: no buffers");
} else {
conn_ = nullptr;
return err;
}
}
void continue_reading() override {
// nop
}
bool prepare_send() override {
write_helper helper{this};
while (down_->can_send_more() && in_) {
auto [again, consumed] = in_->pull(async::delay_errors, 1, helper);
if (!again) {
if (helper.err) {
down_->send_close_message(helper.err);
down_->close(helper.err);
} else {
down_->send_close_message();
down_->close();
}
in_ = nullptr;
return false;
return true;
} else if (helper.aborted) {
in_->cancel();
in_ = nullptr;
return false;
return true;
} else if (consumed == 0) {
return true;
}
......
......@@ -77,11 +77,15 @@ public:
// -- web_socket::lower_layer implementation ---------------------------------
using web_socket::lower_layer::close;
bool can_send_more() const noexcept override;
void suspend_reading() override;
bool stopped_reading() const noexcept override;
bool is_reading() const noexcept override;
void close(status code, std::string_view desc) override;
void request_messages() override;
......@@ -97,17 +101,8 @@ public:
bool end_text_message() override;
void send_close_message() override;
void send_close_message(status code, std::string_view desc) override;
// -- interface for the lower layer ------------------------------------------
template <class LowerLayerPtr>
static void continue_reading(LowerLayerPtr down) {
down->configure_read(receive_policy::up_to(2048));
}
ptrdiff_t consume(byte_span input, byte_span);
private:
......@@ -117,10 +112,6 @@ private:
void ship_pong(byte_span payload);
void ship_close(uint16_t code, std::string_view msg);
void ship_close();
template <class T>
void ship_frame(std::vector<T>& buf);
......
......@@ -22,6 +22,9 @@ public:
/// Pulls messages from the transport until calling `suspend_reading`.
virtual void request_messages() = 0;
/// Stops reading messages until calling `request_messages`.
virtual void suspend_reading() = 0;
/// Begins transmission of a binary message.
virtual void begin_binary_message() = 0;
......@@ -42,14 +45,12 @@ public:
/// Seals the current text message buffer and ships a new WebSocket frame.
virtual bool end_text_message() = 0;
/// Sends the close message with @ref status `normal_close`.
virtual void send_close_message() = 0;
void close() override;
/// Sends the close message with an custom @ref status.
virtual void send_close_message(status code, std::string_view desc) = 0;
void close(const error& reason) override;
/// Sends the close message with @ref status `unexpected_condition`.
void send_close_message(const error& reason);
/// Sends a close message with custom @ref status @p code and @p msg text.
virtual void close(status code, std::string_view msg) = 0;
};
} // namespace caf::net::web_socket
......@@ -81,8 +81,6 @@ public:
ptrdiff_t consume(byte_span input, byte_span) override;
void continue_reading() override;
bool prepare_send() override;
bool done_sending() override;
......
......@@ -10,4 +10,8 @@ generic_lower_layer::~generic_lower_layer() {
// nop
}
void generic_lower_layer::close(const error&) {
close();
}
} // namespace caf::net
......@@ -14,12 +14,21 @@ bool server::can_send_more() const noexcept {
return down_->can_send_more();
}
void server::suspend_reading() {
return down_->suspend_reading();
bool server::is_reading() const noexcept {
return down_->is_reading();
}
void server::close() {
down_->close();
}
bool server::stopped_reading() const noexcept {
return down_->stopped_reading();
void server::request_messages() {
if (!down_->is_reading())
down_->configure_read(receive_policy::up_to(max_request_size_));
}
void server::suspend_reading() {
down_->configure_read(receive_policy::stop());
}
bool server::send_header(context, status code,
......@@ -69,11 +78,10 @@ void server::fin(context) {
error server::init(socket_manager* owner, stream_oriented::lower_layer* down,
const settings& cfg) {
down_ = down;
if (auto err = up_->init(owner, this, cfg))
return err;
if (auto max_size = get_as<uint32_t>(cfg, "http.max-request-size"))
max_request_size_ = *max_size;
down_->configure_read(receive_policy::up_to(max_request_size_));
if (auto err = up_->init(owner, this, cfg))
return err;
return none;
}
......@@ -89,10 +97,6 @@ bool server::done_sending() {
return up_->done_sending();
}
void server::continue_reading() {
down_->configure_read(receive_policy::up_to(max_request_size_));
}
ptrdiff_t server::consume(byte_span input, byte_span) {
using namespace literals;
CAF_LOG_TRACE(CAF_ARG2("bytes", input.size()));
......
......@@ -36,7 +36,7 @@ void length_prefix_framing::abort(const error& reason) {
}
ptrdiff_t length_prefix_framing::consume(byte_span input, byte_span) {
CAF_LOG_TRACE("got" << input.size() << "bytes");
CAF_LOG_TRACE("got" << input.size() << "bytes\n");
if (input.size() < sizeof(uint32_t)) {
CAF_LOG_ERROR("received too few bytes from underlying transport");
up_->abort(make_error(sec::logic_error,
......@@ -65,7 +65,7 @@ ptrdiff_t length_prefix_framing::consume(byte_span input, byte_span) {
if (msg_size == msg.size() && msg_size + hdr_size == input.size()) {
CAF_LOG_DEBUG("got message of size" << msg_size);
if (up_->consume(msg) >= 0) {
if (!down_->stopped_reading())
if (down_->is_reading())
down_->configure_read(receive_policy::exactly(hdr_size));
return static_cast<ptrdiff_t>(input.size());
} else {
......@@ -79,10 +79,6 @@ ptrdiff_t length_prefix_framing::consume(byte_span input, byte_span) {
}
}
void length_prefix_framing::continue_reading() {
up_->continue_reading();
}
bool length_prefix_framing::prepare_send() {
return up_->prepare_send();
}
......@@ -98,15 +94,15 @@ bool length_prefix_framing::can_send_more() const noexcept {
}
void length_prefix_framing::suspend_reading() {
down_->suspend_reading();
down_->configure_read(receive_policy::stop());
}
bool length_prefix_framing::stopped_reading() const noexcept {
return down_->stopped_reading();
bool length_prefix_framing::is_reading() const noexcept {
return down_->is_reading();
}
void length_prefix_framing::request_messages() {
if (down_->stopped_reading())
if (!down_->is_reading())
down_->configure_read(receive_policy::exactly(hdr_size));
}
......@@ -139,12 +135,8 @@ bool length_prefix_framing::end_message() {
}
}
void length_prefix_framing::send_close_message() {
// nop: this layer has no close message
}
void length_prefix_framing::send_close_message(const error&) {
// nop: this layer has no close message
void length_prefix_framing::close() {
down_->close();
}
// -- utility functions ------------------------------------------------------
......
......@@ -131,17 +131,18 @@ size_t multiplexer::num_socket_managers() const noexcept {
return managers_.size();
}
ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) {
ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) const noexcept {
auto first = managers_.begin();
auto last = managers_.end();
auto i = std::find(first, last, mgr);
return i == last ? -1 : std::distance(first, i);
}
ptrdiff_t multiplexer::index_of(socket fd) {
ptrdiff_t multiplexer::index_of(socket fd) const noexcept {
auto first = pollset_.begin();
auto last = pollset_.end();
auto i = std::find_if(first, last, [fd](pollfd& x) { return x.fd == fd.id; });
auto i = std::find_if(first, last,
[fd](const pollfd& x) { return x.fd == fd.id; });
return i == last ? -1 : std::distance(first, i);
}
......@@ -166,40 +167,6 @@ operation multiplexer::mask_of(const socket_manager_ptr& mgr) {
// -- thread-safe signaling ----------------------------------------------------
void multiplexer::register_reading(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) {
do_register_reading(mgr);
} else {
write_to_pipe(pollset_updater::code::register_reading, mgr.get());
}
}
void multiplexer::register_writing(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
CAF_ASSERT(mgr != nullptr);
if (std::this_thread::get_id() == tid_) {
do_register_writing(mgr);
} else {
write_to_pipe(pollset_updater::code::register_writing, mgr.get());
}
}
void multiplexer::continue_reading(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
write_to_pipe(pollset_updater::code::continue_reading, mgr.get());
}
void multiplexer::continue_writing(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
CAF_ASSERT(mgr != nullptr);
if (std::this_thread::get_id() == tid_) {
do_continue_writing(mgr);
} else {
write_to_pipe(pollset_updater::code::continue_writing, mgr.get());
}
}
void multiplexer::discard(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) {
......@@ -251,6 +218,41 @@ void multiplexer::shutdown() {
static_cast<socket_manager*>(nullptr));
}
// -- callbacks for socket managers --------------------------------------------
void multiplexer::register_reading(socket_manager* mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
update_for(mgr).events |= input_mask;
}
void multiplexer::register_writing(socket_manager* mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
update_for(mgr).events |= output_mask;
}
void multiplexer::deregister_reading(socket_manager* mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
update_for(mgr).events &= ~input_mask;
}
void multiplexer::deregister_writing(socket_manager* mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
update_for(mgr).events &= ~output_mask;
}
void multiplexer::deregister(socket_manager* mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
update_for(mgr).events = 0;
}
bool multiplexer::is_reading(const socket_manager* mgr) const noexcept {
return (active_mask_of(mgr) & input_mask) != 0;
}
bool multiplexer::is_writing(const socket_manager* mgr) const noexcept {
return (active_mask_of(mgr) & output_mask) != 0;
}
// -- control flow -------------------------------------------------------------
bool multiplexer::poll_once(bool blocking) {
......@@ -375,44 +377,16 @@ void multiplexer::handle(const socket_manager_ptr& mgr,
CAF_ASSERT(mgr != nullptr);
bool checkerror = true;
// Note: we double-check whether the manager is actually reading because a
// previous action from the pipe may have called shutdown_reading.
if ((events & revents & input_mask) != 0) {
// previous action from the pipe may have disabled reading.
if ((revents & input_mask) != 0 && is_reading(mgr.get())) {
checkerror = false;
switch (mgr->handle_read_event()) {
default: // socket_manager::read_result::again
// Nothing to do, bitmask may remain unchanged.
break;
case socket_manager::read_result::abort:
case socket_manager::read_result::stop:
update_for(mgr).events &= ~input_mask;
break;
case socket_manager::read_result::want_write:
update_for(mgr).events = output_mask;
break;
case socket_manager::read_result::handover: {
do_handover(mgr);
return;
}
}
mgr->handle_read_event();
}
// Similar reasoning than before: double-check whether this event should still
// get dispatched.
if ((events & revents & output_mask) != 0) {
if ((revents & output_mask) != 0 && is_writing(mgr.get())) {
checkerror = false;
switch (mgr->handle_write_event()) {
default: // socket_manager::write_result::again
break;
case socket_manager::write_result::abort:
case socket_manager::write_result::stop:
update_for(mgr).events &= ~output_mask;
break;
case socket_manager::write_result::want_read:
update_for(mgr).events = input_mask;
break;
case socket_manager::write_result::handover:
do_handover(mgr);
return;
}
mgr->handle_write_event();
}
if (checkerror && ((revents & error_mask) != 0)) {
if (revents & POLLNVAL)
......@@ -421,38 +395,7 @@ void multiplexer::handle(const socket_manager_ptr& mgr,
mgr->handle_error(sec::socket_disconnected);
else
mgr->handle_error(sec::socket_operation_failed);
update_for(mgr).events = 0;
}
}
void multiplexer::do_handover(const socket_manager_ptr& mgr) {
// Make sure to override the manager pointer in the update. Updates are
// associated to sockets, so the new manager is likely to modify this update
// again. Hence, it *must not* point to the old manager.
auto& update = update_for(mgr);
update.events = 0;
if (mgr->do_handover()) {
// If the new manager registered itself for reading, make sure it processes
// whatever data is available in buffers outside of the socket that may not
// trigger read events.
if ((update.events & input_mask)) {
switch (mgr->handle_buffered_data()) {
default: // socket_manager::read_result::again
// Nothing to do, bitmask may remain unchanged.
break;
case socket_manager::read_result::abort:
case socket_manager::read_result::stop:
update.events &= ~input_mask;
break;
case socket_manager::read_result::want_write:
update.events = output_mask;
break;
case socket_manager::read_result::handover: {
// Down the rabbit hole we go!
do_handover(mgr);
}
}
}
update_for(mgr.get()).events = 0;
}
}
......@@ -467,14 +410,13 @@ multiplexer::poll_update& multiplexer::update_for(ptrdiff_t index) {
}
}
multiplexer::poll_update&
multiplexer::update_for(const socket_manager_ptr& mgr) {
multiplexer::poll_update& multiplexer::update_for(socket_manager* mgr) {
auto fd = mgr->handle();
if (auto i = updates_.find(fd); i != updates_.end()) {
return i->second;
} else if (auto index = index_of(fd); index != -1) {
updates_.container().emplace_back(fd,
poll_update{pollset_[index].events, mgr});
updates_.container().emplace_back(fd, poll_update{pollset_[index].events,
socket_manager_ptr{mgr}});
return updates_.container().back().second;
} else {
updates_.container().emplace_back(fd, poll_update{0, mgr});
......@@ -501,7 +443,7 @@ void multiplexer::write_to_pipe(uint8_t opcode, T* ptr) {
intrusive_ptr_release(ptr);
}
short multiplexer::active_mask_of(const socket_manager_ptr& mgr) {
short multiplexer::active_mask_of(const socket_manager* mgr) const noexcept {
auto fd = mgr->handle();
if (auto i = updates_.find(fd); i != updates_.end()) {
return i->second.events;
......@@ -512,14 +454,6 @@ short multiplexer::active_mask_of(const socket_manager_ptr& mgr) {
}
}
bool multiplexer::is_reading(const socket_manager_ptr& mgr) {
return (active_mask_of(mgr) & input_mask) != 0;
}
bool multiplexer::is_writing(const socket_manager_ptr& mgr) {
return (active_mask_of(mgr) & output_mask) != 0;
}
// -- internal callbacks the pollset updater -----------------------------------
void multiplexer::do_shutdown() {
......@@ -537,61 +471,17 @@ void multiplexer::do_shutdown() {
apply_updates();
}
void multiplexer::do_register_reading(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
// When shutting down, no new reads are allowed.
if (shutting_down_)
mgr->close_read();
else if (!mgr->read_closed())
update_for(mgr).events |= input_mask;
}
void multiplexer::do_register_writing(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
// When shutting down, we do allow managers to write whatever is currently
// pending but we make sure that all read channels are closed.
if (shutting_down_)
mgr->close_read();
if (!mgr->write_closed())
update_for(mgr).events |= output_mask;
}
void multiplexer::do_continue_reading(const socket_manager_ptr& mgr) {
if (!is_reading(mgr)) {
switch (mgr->handle_continue_reading()) {
default: // socket_manager::read_result::(stop | abort)
update_for(mgr).events &= ~input_mask;
break;
case socket_manager::read_result::again:
update_for(mgr).events |= input_mask;
break;
case socket_manager::read_result::want_write:
update_for(mgr).events = output_mask;
break;
case socket_manager::read_result::handover: {
do_handover(mgr);
}
}
}
}
void multiplexer::do_continue_writing(const socket_manager_ptr& mgr) {
if (!is_writing(mgr)) {
update_for(mgr).events |= output_mask;
}
}
void multiplexer::do_discard(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
mgr->handle_error(sec::disposed);
update_for(mgr).events = 0;
update_for(mgr.get()).events = 0;
}
void multiplexer::do_shutdown_reading(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (!shutting_down_ && !mgr->read_closed()) {
mgr->close_read();
update_for(mgr).events &= ~input_mask;
update_for(mgr.get()).events &= ~input_mask;
}
}
......@@ -599,7 +489,7 @@ void multiplexer::do_shutdown_writing(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (!shutting_down_ && !mgr->write_closed()) {
mgr->close_write();
update_for(mgr).events &= ~output_mask;
update_for(mgr.get()).events &= ~output_mask;
}
}
......@@ -615,10 +505,8 @@ void multiplexer::do_init(const socket_manager_ptr& mgr) {
CAF_LOG_DEBUG("mgr->init failed: " << err);
// The socket manager should not register itself for any events if
// initialization fails. Purge any state just in case.
update_for(mgr).events = 0;
update_for(mgr.get()).events = 0;
}
// Else: no update since the manager is supposed to call continue_reading
// and continue_writing as necessary.
}
}
......
......@@ -31,11 +31,12 @@ std::unique_ptr<pollset_updater> pollset_updater::make(pipe_socket fd) {
error pollset_updater::init(socket_manager* owner, const settings&) {
CAF_LOG_TRACE("");
owner_ = owner;
mpx_ = owner->mpx_ptr();
return nonblocking(fd_, true);
}
pollset_updater::read_result pollset_updater::handle_read_event() {
void pollset_updater::handle_read_event() {
CAF_LOG_TRACE("");
auto as_mgr = [](intptr_t ptr) {
return intrusive_ptr{reinterpret_cast<socket_manager*>(ptr), false};
......@@ -56,18 +57,6 @@ pollset_updater::read_result pollset_updater::handle_read_event() {
intptr_t ptr;
memcpy(&ptr, buf_.data() + 1, sizeof(intptr_t));
switch (static_cast<code>(opcode)) {
case code::register_reading:
mpx_->do_register_reading(as_mgr(ptr));
break;
case code::continue_reading:
mpx_->do_continue_reading(as_mgr(ptr));
break;
case code::register_writing:
mpx_->do_register_writing(as_mgr(ptr));
break;
case code::continue_writing:
mpx_->do_continue_writing(as_mgr(ptr));
break;
case code::init_manager:
mpx_->do_init(as_mgr(ptr));
break;
......@@ -94,25 +83,20 @@ pollset_updater::read_result pollset_updater::handle_read_event() {
}
} else if (num_bytes == 0) {
CAF_LOG_DEBUG("pipe closed, assume shutdown");
return read_result::stop;
owner_->shutdown();
return;
} else if (last_socket_error_is_temporary()) {
return read_result::again;
return;
} else {
return read_result::stop;
CAF_LOG_DEBUG("pollset updater failed to read from the pipe");
owner_->shutdown();
return;
}
}
}
pollset_updater::read_result pollset_updater::handle_buffered_data() {
return read_result::again;
}
pollset_updater::read_result pollset_updater::handle_continue_reading() {
return read_result::again;
}
pollset_updater::write_result pollset_updater::handle_write_event() {
return write_result::stop;
void pollset_updater::handle_write_event() {
owner_->deregister_writing();
}
void pollset_updater::abort(const error&) {
......
......@@ -40,22 +40,73 @@ actor_system& socket_manager::system() noexcept {
return mpx_->system();
}
bool socket_manager::is_reading() const noexcept {
return mpx_->is_reading(this);
}
bool socket_manager::is_writing() const noexcept {
return mpx_->is_writing(this);
}
// -- event loop management ----------------------------------------------------
void socket_manager::register_reading() {
mpx_->register_reading(this);
if (!flags_.read_closed)
mpx_->register_reading(this);
}
void socket_manager::continue_reading() {
mpx_->continue_reading(this);
void socket_manager::register_writing() {
if (!flags_.write_closed)
mpx_->register_writing(this);
}
void socket_manager::register_writing() {
mpx_->register_writing(this);
void socket_manager::deregister_reading() {
mpx_->deregister_reading(this);
}
void socket_manager::deregister_writing() {
mpx_->deregister_writing(this);
}
void socket_manager::deregister() {
mpx_->deregister(this);
}
void socket_manager::shutdown_read() {
deregister_reading();
flags_.read_closed = true;
}
void socket_manager::shutdown_write() {
deregister_writing();
flags_.write_closed = true;
}
void socket_manager::shutdown() {
flags_.read_closed = true;
flags_.write_closed = true;
deregister();
}
void socket_manager::continue_writing() {
mpx_->continue_writing(this);
// -- callbacks for the handler ------------------------------------------------
void socket_manager::schedule_handover() {
deregister();
mpx_->schedule_fn([ptr = strong_this()] { //
event_handler_ptr next;
if (ptr->handler_->do_handover(next)) {
ptr->handler_.swap(next);
}
});
}
void socket_manager::schedule(action what) {
// Wrap the action to make sure the socket manager is still alive when running
// the action later.
mpx_->schedule_fn([ptr = strong_this(), f = std::move(what)]() mutable {
CAF_IGNORE_UNUSED(ptr);
f.run();
});
}
// -- callbacks for the multiplexer --------------------------------------------
......@@ -70,16 +121,6 @@ void socket_manager::close_write() noexcept {
flags_.write_closed = true;
}
bool socket_manager::do_handover() {
event_handler_ptr next;
if (handler_->do_handover(next)) {
handler_.swap(next);
return true;
} else {
return false;
}
}
error socket_manager::init(const settings& cfg) {
CAF_LOG_TRACE(CAF_ARG(cfg));
if (auto err = nonblocking(fd_, true)) {
......@@ -89,32 +130,12 @@ error socket_manager::init(const settings& cfg) {
return handler_->init(this, cfg);
}
socket_manager::read_result socket_manager::handle_read_event() {
auto result = handler_->handle_read_event();
switch (result) {
default:
break;
case read_result::close:
flags_.read_closed = true;
break;
case read_result::abort:
flags_.read_closed = true;
flags_.write_closed = true;
break;
}
return result;
}
socket_manager::read_result socket_manager::handle_buffered_data() {
return handler_->handle_buffered_data();
}
socket_manager::read_result socket_manager::handle_continue_reading() {
return handler_->handle_continue_reading();
void socket_manager::handle_read_event() {
handler_->handle_read_event();
}
socket_manager::write_result socket_manager::handle_write_event() {
return handler_->handle_write_event();
void socket_manager::handle_write_event() {
handler_->handle_write_event();
}
void socket_manager::handle_error(sec code) {
......@@ -124,4 +145,10 @@ void socket_manager::handle_error(sec code) {
}
}
// -- utility functions --------------------------------------------------------
socket_manager_ptr socket_manager::strong_this() {
return socket_manager_ptr{this};
}
} // namespace caf::net
......@@ -26,10 +26,6 @@ public:
using super = socket_event_layer;
using read_result = typename super::read_result;
using write_result = typename super::write_result;
using upper_layer_ptr = transport::upper_layer_ptr;
handshake_worker(connection conn, bool is_server, upper_layer_ptr up)
......@@ -46,55 +42,56 @@ public:
return caf::none;
}
read_result handle_read_event() override {
void handle_read_event() override {
if (auto res = advance_handshake(); res > 0) {
return read_result::handover;
owner_->deregister();
owner_->schedule_handover();
} else if (res == 0) {
up_->abort(make_error(sec::connection_closed));
return read_result::stop;
owner_->deregister();
} else {
auto err = policy_.last_error(policy_.conn.fd(), res);
switch (err) {
case stream_transport_error::want_read:
case stream_transport_error::temporary:
return read_result::again;
break;
case stream_transport_error::want_write:
return read_result::want_write;
default:
owner_->deregister_reading();
owner_->register_writing();
break;
default: {
auto err = make_error(sec::cannot_connect_to_node,
policy_.conn.last_error_string(res));
up_->abort(std::move(err));
return read_result::stop;
owner_->deregister();
}
}
}
}
read_result handle_buffered_data() override {
return read_result::again;
}
read_result handle_continue_reading() override {
return read_result::again;
}
write_result handle_write_event() override {
void handle_write_event() override {
if (auto res = advance_handshake(); res > 0) {
return write_result::handover;
owner_->deregister();
owner_->schedule_handover();
return;
} else if (res == 0) {
up_->abort(make_error(sec::connection_closed));
return write_result::stop;
owner_->deregister();
} else {
switch (policy_.last_error(policy_.conn.fd(), res)) {
case stream_transport_error::want_write:
case stream_transport_error::temporary:
return write_result::again;
break;
case stream_transport_error::want_read:
return write_result::want_read;
default:
owner_->deregister_writing();
owner_->register_reading();
break;
default: {
auto err = make_error(sec::cannot_connect_to_node,
policy_.conn.last_error_string(res));
up_->abort(std::move(err));
return write_result::stop;
owner_->deregister();
}
}
}
}
......
......@@ -49,14 +49,14 @@ size_t stream_transport::default_policy::buffered() const noexcept {
// -- constructors, destructors, and assignment operators ----------------------
stream_transport::stream_transport(stream_socket fd, upper_layer_ptr up)
: fd_(fd), up_(std::move(up)) {
policy_ = &default_policy_;
: fd_(fd), up_(std::move(up)), policy_(&default_policy_) {
memset(&flags_, 0, sizeof(flags_t));
}
stream_transport::stream_transport(stream_socket fd, upper_layer_ptr up,
policy* policy)
: fd_(fd), up_(std::move(up)), policy_(policy) {
// nop
memset(&flags_, 0, sizeof(flags_t));
}
// -- factories ----------------------------------------------------------------
......@@ -73,11 +73,25 @@ bool stream_transport::can_send_more() const noexcept {
}
void stream_transport::configure_read(receive_policy rd) {
if (rd.max_size > 0 && max_read_size_ == 0) {
parent_->register_reading();
}
auto restarting = rd.max_size > 0 && max_read_size_ == 0;
min_read_size_ = rd.min_size;
max_read_size_ = rd.max_size;
if (restarting && !parent_->is_reading()) {
if (buffered_ >= min_read_size_) {
// We can already make progress with the data we have. Hence, we need
// schedule a call to read from our buffer before we can wait for
// additional data from the peer.
parent_->schedule_fn([this] {
parent_->register_reading();
handle_buffered_data();
});
} else {
// Simply ask for more data.
parent_->register_reading();
}
} else if (max_read_size_ == 0) {
parent_->deregister_reading();
}
}
void stream_transport::begin_output() {
......@@ -93,12 +107,16 @@ bool stream_transport::end_output() {
return true;
}
void stream_transport::suspend_reading() {
configure_read(receive_policy::stop());
bool stream_transport::is_reading() const noexcept {
return max_read_size_ > 0;
}
bool stream_transport::stopped_reading() const noexcept {
return max_read_size_ == 0;
void stream_transport::close() {
parent_->shutdown_read();
if (write_buf_.empty())
parent_->shutdown_write();
else
flags_.shutting_down = true;
}
// -- implementation of transport ----------------------------------------------
......@@ -124,32 +142,28 @@ error stream_transport::init(socket_manager* owner, const settings& config) {
return up_->init(owner, this, config);
}
stream_transport::read_result stream_transport::handle_read_event() {
void stream_transport::handle_read_event() {
CAF_LOG_TRACE(CAF_ARG2("socket", fd_.id));
// Convenience lambda for failing the application.
auto fail = [this](error reason) {
CAF_LOG_DEBUG("read failed" << CAF_ARG(reason));
up_->abort(reason);
return read_result::abort;
};
// Resume a write operation if the transport waited for the socket to be
// readable from the last call to handle_write_event.
if (flags_.wanted_read_from_write_event) {
flags_.wanted_read_from_write_event = false;
switch (handle_write_event()) {
case write_result::want_read:
CAF_ASSERT(flags_.wanted_read_from_write_event);
return read_result::again;
case write_result::handover:
return read_result::handover;
case write_result::again:
parent_->register_writing();
break;
default:
break;
// The subsequent call to handle_write_event expects a writing manager.
parent_->register_writing();
handle_write_event();
if (!parent_->is_reading()) {
// The call to handle_write_event deregisters the manager from reading in
// case of an error. So we need to double-check that flag here.
return;
}
// Check if we have actually some reading to do.
if (max_read_size_ == 0) {
if (!flags_.wanted_read_from_write_event)
parent_->deregister_reading();
return;
}
}
// Make sure the buffer is large enough.
// Make sure our read buffer is large enough.
if (read_buf_.size() < max_read_size_)
read_buf_.resize(max_read_size_);
// Fill up our buffer.
......@@ -160,19 +174,23 @@ stream_transport::read_result stream_transport::handle_read_event() {
switch (policy_->last_error(fd_, rd)) {
case stream_transport_error::temporary:
case stream_transport_error::want_read:
return read_result::again;
// Try again later.
return;
case stream_transport_error::want_write:
// Wait for writable socket and then call handle_read_event again.
flags_.wanted_write_from_read_event = true;
return read_result::want_write;
parent_->register_writing();
parent_->deregister_reading();
return;
default:
return fail(sec::socket_operation_failed);
return fail(make_error(sec::socket_operation_failed));
}
} else if (rd == 0) {
return fail(sec::socket_disconnected);
return fail(make_error(sec::socket_disconnected));
}
// Make sure we actually have all data currently available to us and the
// policy is not holding on to some bytes. This may happen when using
// OpenSSL or any other transport policy operating on blocks.
// OpenSSL or any other transport policy that operates on blocks.
buffered_ += static_cast<size_t>(rd);
if (auto policy_buffered = policy_->buffered(); policy_buffered > 0) {
if (auto n = buffered_ + policy_buffered; n > read_buf_.size())
......@@ -181,23 +199,16 @@ stream_transport::read_result stream_transport::handle_read_event() {
policy_buffered));
if (rd2 != static_cast<ptrdiff_t>(policy_buffered)) {
CAF_LOG_ERROR("failed to read buffered data from the policy");
return fail(sec::socket_operation_failed);
return fail(make_error(sec::socket_operation_failed));
}
buffered_ += static_cast<size_t>(rd2);
}
// Try to make progress on the application and return control to the
// multiplexer to allow other sockets to run.
return handle_buffered_data();
// Read buffered data and then allow other sockets to run.
handle_buffered_data();
}
stream_transport::read_result stream_transport::handle_buffered_data() {
void stream_transport::handle_buffered_data() {
CAF_LOG_TRACE(CAF_ARG(buffered_));
// Convenience lambda for failing the application.
auto fail = [this](error reason) {
CAF_LOG_DEBUG("read failed" << CAF_ARG(reason));
up_->abort(reason);
return read_result::abort;
};
// Loop until we have drained the buffer as much as we can.
CAF_ASSERT(min_read_size_ <= max_read_size_);
while (max_read_size_ > 0 && buffered_ >= min_read_size_) {
......@@ -210,7 +221,8 @@ stream_transport::read_result stream_transport::handle_buffered_data() {
// Negative values indicate that the application encountered an
// unrecoverable error.
up_->abort(make_error(caf::sec::runtime_error, "consumed < 0"));
return read_result::abort;
parent_->deregister();
return;
} else if (consumed == 0) {
// Returning 0 means that the application wants more data. Note:
// max_read_size_ may have changed if the application realized it
......@@ -219,18 +231,19 @@ stream_transport::read_result stream_transport::handle_buffered_data() {
delta_offset_ = static_cast<ptrdiff_t>(n);
if (n == max_read_size_ || max_read_size_ < old_max_read_size) {
CAF_LOG_ERROR("application failed to make progress");
return fail(sec::runtime_error);
return fail(make_error(sec::runtime_error));
} else if (n == buffered_) {
// Either the application has increased max_read_size_ or we
// did not reach max_read_size_ the first time. In both cases, we
// cannot proceed without receiving more data.
return read_result::again;
return;
}
} else if (static_cast<size_t>(consumed) > n) {
// Must not happen. An application cannot handle more data then we pass
// to it.
up_->abort(make_error(sec::logic_error, "consumed > buffer.size"));
return read_result::abort;
parent_->deregister();
return;
} else {
// Shove the unread bytes to the beginning of the buffer and continue
// to the next loop iteration.
......@@ -245,75 +258,74 @@ stream_transport::read_result stream_transport::handle_buffered_data() {
}
}
}
return max_read_size_ > 0 ? read_result::again : read_result::stop;
if (max_read_size_ == 0)
parent_->deregister_reading();
}
void stream_transport::fail(const error& reason) {
CAF_LOG_TRACE(CAF_ARG(reason));
up_->abort(reason);
parent_->deregister();
}
stream_transport::write_result stream_transport::handle_write_event() {
void stream_transport::handle_write_event() {
CAF_LOG_TRACE(CAF_ARG2("socket", fd_.id));
auto fail = [this](sec code) {
CAF_LOG_DEBUG("read failed" << CAF_ARG(code));
up_->abort(make_error(code));
return write_result::abort;
};
// Resume a read operation if the transport waited for the socket to be
// writable from the last call to handle_read_event.
if (flags_.wanted_write_from_read_event) {
flags_.wanted_write_from_read_event = false;
switch (handle_read_event()) {
case read_result::want_write:
CAF_ASSERT(flags_.wanted_write_from_read_event);
return write_result::again;
case read_result::handover:
return write_result::handover;
case read_result::again:
parent_->register_reading();
break;
default:
break;
// The subsequent call to handle_read_event expects a writing manager.
parent_->register_writing();
handle_read_event();
if (!parent_->is_writing()) {
// The call to handle_read_event deregisters the manager from writing in
// case of an error. So we need to double-check that flag here.
return;
}
// Fall though and see if we also have something to write.
}
// Allow the upper layer to add extra data to the write buffer.
if (!up_->prepare_send()) {
up_->abort(make_error(caf::sec::runtime_error, "prepare_send failed"));
return write_result::abort;
CAF_LOG_DEBUG("prepare_send failed");
parent_->deregister();
return;
}
if (write_buf_.empty()) {
if (up_->done_sending()) {
if (!flags_.shutting_down)
parent_->deregister_writing();
else
parent_->shutdown_write();
}
return;
}
if (write_buf_.empty())
return !up_->done_sending() ? write_result::again : write_result::stop;
auto write_res = policy_->write(fd_, write_buf_);
if (write_res > 0) {
write_buf_.erase(write_buf_.begin(), write_buf_.begin() + write_res);
return !write_buf_.empty() || !up_->done_sending() ? write_result::again
: write_result::stop;
if (write_buf_.empty() && up_->done_sending()) {
if (!flags_.shutting_down)
parent_->deregister_writing();
else
parent_->shutdown_write();
}
return;
} else if (write_res < 0) {
// Try again later on temporary errors such as EWOULDBLOCK and
// stop writing to the socket on hard errors.
switch (policy_->last_error(fd_, write_res)) {
case stream_transport_error::temporary:
case stream_transport_error::want_write:
return write_result::again;
return;
case stream_transport_error::want_read:
flags_.wanted_read_from_write_event = true;
return write_result::want_read;
parent_->register_reading();
parent_->deregister_writing();
return;
default:
return fail(sec::socket_operation_failed);
return fail(make_error(sec::socket_operation_failed));
}
} else {
// write() returns 0 if the connection was closed.
return fail(sec::socket_disconnected);
}
}
stream_transport::read_result stream_transport::handle_continue_reading() {
if (max_read_size_ == 0) {
up_->continue_reading();
if (max_read_size_ != 0) {
return handle_buffered_data();
} else {
return read_result::stop;
}
} else {
return handle_buffered_data();
return fail(make_error(sec::socket_disconnected));
}
}
......
......@@ -95,11 +95,6 @@ ptrdiff_t client::consume(byte_span buffer, byte_span delta) {
}
}
void client::continue_reading() {
if (handshake_completed())
upper_layer().continue_reading();
}
bool client::prepare_send() {
return handshake_completed() ? upper_layer().prepare_send() : true;
}
......
......@@ -23,15 +23,34 @@ bool framing::can_send_more() const noexcept {
}
void framing::suspend_reading() {
down_->suspend_reading();
down_->configure_read(receive_policy::stop());
}
bool framing::stopped_reading() const noexcept {
return down_->stopped_reading();
bool framing::is_reading() const noexcept {
return down_->is_reading();
}
void framing::close(status code, std::string_view msg) {
auto code_val = static_cast<uint16_t>(code);
uint32_t mask_key = 0;
byte_buffer payload;
payload.reserve(msg.size() + 2);
payload.push_back(static_cast<std::byte>((code_val & 0xFF00) >> 8));
payload.push_back(static_cast<std::byte>(code_val & 0x00FF));
for (auto c : msg)
payload.push_back(static_cast<std::byte>(c));
if (mask_outgoing_frames) {
mask_key = static_cast<uint32_t>(rng_());
detail::rfc6455::mask_data(mask_key, payload);
}
down_->begin_output();
detail::rfc6455::assemble_frame(detail::rfc6455::connection_close, mask_key,
payload, down_->output_buffer());
down_->end_output();
}
void framing::request_messages() {
if (down_->stopped_reading())
if (!down_->is_reading())
down_->configure_read(receive_policy::up_to(2048));
}
......@@ -61,14 +80,6 @@ bool framing::end_text_message() {
return true;
}
void framing::send_close_message() {
ship_close();
}
void framing::send_close_message(status code, std::string_view desc) {
ship_close(static_cast<uint16_t>(code), desc);
}
// -- interface for the lower layer --------------------------------------------
ptrdiff_t framing::consume(byte_span input, byte_span) {
......@@ -99,9 +110,6 @@ ptrdiff_t framing::consume(byte_span input, byte_span) {
// Wait for more data if necessary.
size_t frame_size = hdr_bytes + hdr.payload_len;
if (buffer.size() < frame_size) {
// Ask for more data unless the upper layer called suspend_reading.
if (!down_->stopped_reading())
down_->configure_read(receive_policy::up_to(2048));
down_->configure_read(receive_policy::exactly(frame_size));
return consumed;
}
......@@ -170,7 +178,7 @@ ptrdiff_t framing::consume(byte_span input, byte_span) {
buffer = buffer.subspan(frame_size);
if (buffer.empty()) {
// Ask for more data unless the upper layer called suspend_reading.
if (!down_->stopped_reading())
if (down_->is_reading())
down_->configure_read(receive_policy::up_to(2048));
return consumed + static_cast<ptrdiff_t>(frame_size);
}
......@@ -215,40 +223,6 @@ void framing::ship_pong(byte_span payload) {
down_->end_output();
}
void framing::ship_close(uint16_t code, std::string_view msg) {
uint32_t mask_key = 0;
byte_buffer payload;
payload.reserve(msg.size() + 2);
payload.push_back(static_cast<std::byte>((code & 0xFF00) >> 8));
payload.push_back(static_cast<std::byte>(code & 0x00FF));
for (auto c : msg)
payload.push_back(static_cast<std::byte>(c));
if (mask_outgoing_frames) {
mask_key = static_cast<uint32_t>(rng_());
detail::rfc6455::mask_data(mask_key, payload);
}
down_->begin_output();
detail::rfc6455::assemble_frame(detail::rfc6455::connection_close, mask_key,
payload, down_->output_buffer());
down_->end_output();
}
void framing::ship_close() {
uint32_t mask_key = 0;
std::byte payload[] = {// Error code 1000: normal close.
std::byte{0x03}, std::byte{0xE8},
// "EOF" string as goodbye message.
std::byte{'E'}, std::byte{'O'}, std::byte{'F'}};
if (mask_outgoing_frames) {
mask_key = static_cast<uint32_t>(rng_());
detail::rfc6455::mask_data(mask_key, payload);
}
down_->begin_output();
detail::rfc6455::assemble_frame(detail::rfc6455::connection_close, mask_key,
payload, down_->output_buffer());
down_->end_output();
}
template <class T>
void framing::ship_frame(std::vector<T>& buf) {
uint32_t mask_key = 0;
......
......@@ -13,8 +13,13 @@ lower_layer::~lower_layer() {
// nop
}
void lower_layer::send_close_message(const error& reason) {
send_close_message(status::unexpected_condition, to_string(reason));
void lower_layer::close() {
using namespace std::literals;
close(status::normal_close, "EOF"sv);
}
void lower_layer::close(const error& reason) {
close(status::unexpected_condition, to_string(reason));
}
} // namespace caf::net::web_socket
......@@ -74,11 +74,6 @@ ptrdiff_t server::consume(byte_span input, byte_span delta) {
}
}
void server::continue_reading() {
auto rp = receive_policy::up_to(handshake::max_http_size);
lower_layer().configure_read(rp);
}
bool server::prepare_send() {
return handshake_complete_ ? upper_layer().prepare_send() : true;
}
......
......@@ -19,8 +19,12 @@ bool mock_stream_transport::can_send_more() const noexcept {
return true;
}
void mock_stream_transport::suspend_reading() {
configure_read(net::receive_policy::stop());
bool mock_stream_transport::is_reading() const noexcept {
return max_read_size > 0;
}
void mock_stream_transport::close() {
// nop
}
void mock_stream_transport::configure_read(net::receive_policy policy) {
......@@ -40,10 +44,6 @@ bool mock_stream_transport::end_output() {
return true;
}
bool mock_stream_transport::stopped_reading() const noexcept {
return max_read_size == 0;
}
ptrdiff_t mock_stream_transport::handle_input() {
ptrdiff_t result = 0;
while (max_read_size > 0) {
......
......@@ -32,7 +32,9 @@ public:
bool can_send_more() const noexcept override;
void suspend_reading() override;
bool is_reading() const noexcept override;
void close() override;
void configure_read(caf::net::receive_policy policy) override;
......@@ -42,8 +44,6 @@ public:
bool end_output() override;
bool stopped_reading() const noexcept override;
// -- initialization ---------------------------------------------------------
caf::error init(const caf::settings& cfg) {
......
......@@ -65,10 +65,6 @@ public:
return !self->terminated();
}
void continue_reading() override {
CAF_FAIL("continue_reading called");
}
bool done_sending() override {
return self->try_block_mailbox();
}
......
......@@ -81,8 +81,9 @@ public:
net::stream_oriented::lower_layer* down_ptr,
const settings&) override {
down = down_ptr;
if (auto ptr = adapter_type::try_open(mgr, std::move(input))) {
adapter = std::move(ptr);
if (auto buf = input.try_open()) {
auto do_wakeup = make_action([this] { prepare_send(); });
adapter = adapter_type::make(std::move(buf), mgr, std::move(do_wakeup));
return none;
} else {
FAIL("unable to open the resource");
......@@ -137,10 +138,6 @@ public:
return done || !adapter->has_data();
}
void continue_reading() override {
CAF_FAIL("continue_reading called");
}
void abort(const error& reason) override {
MESSAGE("app::abort: " << reason);
}
......
......@@ -51,6 +51,7 @@ public:
error init(net::socket_manager*, net::http::lower_layer* down_ptr,
const settings&) override {
down = down_ptr;
down->request_messages();
return none;
}
......@@ -76,10 +77,6 @@ public:
payload.assign(body.begin(), body.end());
return static_cast<ptrdiff_t>(body.size());
}
void continue_reading() override {
// nop
}
};
} // namespace
......
......@@ -45,10 +45,11 @@ public:
// nop
}
caf::error init(net::socket_manager*,
caf::error init(net::socket_manager* mgr_ptr,
net::message_oriented::lower_layer* down_ptr,
const settings&) override {
// Start reading immediately.
mgr = mgr_ptr;
down = down_ptr;
down->request_messages();
return none;
......@@ -66,23 +67,23 @@ public:
// nop
}
void continue_reading() override {
down->request_messages();
void continue_reading() {
mgr->schedule_fn([this] { down->request_messages(); });
}
ptrdiff_t consume(byte_span buf) override {
printf("app_t::consume %d\n", __LINE__);
auto printable = [](std::byte x) {
return ::isprint(static_cast<uint8_t>(x));
};
if (CHECK(std::all_of(buf.begin(), buf.end(), printable))) {
printf("app_t::consume %d\n", __LINE__);
auto str_buf = reinterpret_cast<char*>(buf.data());
inputs->emplace_back(std::string{str_buf, buf.size()});
printf("app_t::consume %d added %s\n", __LINE__, inputs->back().c_str());
MESSAGE("app: consumed " << inputs->back());
if constexpr (EnableSuspend)
if (inputs->back() == "pause")
if (inputs->back() == "pause") {
MESSAGE("app: suspend reading");
down->suspend_reading();
}
std::string response = "ok ";
response += std::to_string(inputs->size());
auto response_bytes = as_bytes(make_span(response));
......@@ -92,11 +93,12 @@ public:
CHECK(down->end_message());
return static_cast<ptrdiff_t>(buf.size());
} else {
printf("app_t::consume %d\n", __LINE__);
return -1;
}
}
net::socket_manager* mgr = nullptr;
net::message_oriented::lower_layer* down = nullptr;
shared_string_list inputs;
......@@ -157,7 +159,7 @@ SCENARIO("length-prefix framing reads data with 32-bit size headers") {
}
}
SCENARIO("calling suspend_reading removes message apps temporarily") {
SCENARIO("calling suspend_reading temporarily halts receiving of messages") {
using namespace std::literals;
GIVEN("a length_prefix_framing with an app that consumes strings") {
auto [fd1, fd2] = unbox(net::make_stream_socket_pair());
......@@ -185,6 +187,7 @@ SCENARIO("calling suspend_reading removes message apps temporarily") {
CAF_FAIL("nonblocking returned an error: " << err);
auto buf = std::make_shared<string_list>();
auto app = app_t<true>::make(buf);
auto app_ptr = app.get();
auto framing = net::length_prefix_framing::make(std::move(app));
auto transport = net::stream_transport::make(fd2, std::move(framing));
auto mgr = net::socket_manager::make(&mpx, fd2, std::move(transport));
......@@ -201,8 +204,8 @@ SCENARIO("calling suspend_reading removes message apps temporarily") {
CHECK_EQ(buf->at(1), "second");
CHECK_EQ(buf->at(2), "pause");
}
THEN("users can resume it via continue_reading ") {
mgr->continue_reading();
THEN("users can resume it manually") {
app_ptr->continue_reading();
mpx.apply_updates();
mpx.poll_once(true);
CHECK_EQ(mpx.mask_of(mgr), net::operation::read);
......
......@@ -63,53 +63,33 @@ public:
// -- implementation of socket_event_layer -----------------------------------
error init(net::socket_manager*, const settings&) override {
error init(net::socket_manager* mgr, const settings&) override {
mgr_ = mgr;
return none;
}
read_result handle_read_event() override {
// if (trigger_handover) {
// MESSAGE(name << " triggered a handover");
// return read_result::handover;
// }
void handle_read_event() override {
if (read_capacity() < 1024)
rd_buf_.resize(rd_buf_.size() + 2048);
auto num_bytes = read(fd_,
make_span(read_position_begin(), read_capacity()));
if (num_bytes > 0) {
CAF_ASSERT(num_bytes > 0);
rd_buf_pos_ += num_bytes;
return read_result::again;
} else if (num_bytes < 0 && net::last_socket_error_is_temporary()) {
return read_result::again;
} else {
return read_result::stop;
auto res = read(fd_, make_span(read_position_begin(), read_capacity()));
if (res > 0) {
CAF_ASSERT(res > 0);
rd_buf_pos_ += res;
} else if (res == 0 || !net::last_socket_error_is_temporary()) {
mgr_->deregister();
}
}
read_result handle_buffered_data() override {
return read_result::again;
}
read_result handle_continue_reading() override {
return read_result::again;
}
write_result handle_write_event() override {
// if (trigger_handover) {
// MESSAGE(name << " triggered a handover");
// return write_result::handover;
// }
if (wr_buf_.size() == 0)
return write_result::stop;
auto num_bytes = write(fd_, wr_buf_);
if (num_bytes > 0) {
wr_buf_.erase(wr_buf_.begin(), wr_buf_.begin() + num_bytes);
return wr_buf_.size() > 0 ? write_result::again : write_result::stop;
void handle_write_event() override {
if (wr_buf_.size() == 0) {
mgr_->deregister_writing();
} else if (auto res = write(fd_, wr_buf_); res > 0) {
wr_buf_.erase(wr_buf_.begin(), wr_buf_.begin() + res);
if (wr_buf_.size() == 0)
mgr_->deregister_writing();
} else if (res == 0 || !net::last_socket_error_is_temporary()) {
mgr_->deregister();
}
return num_bytes < 0 && net::last_socket_error_is_temporary()
? write_result::again
: write_result::stop;
}
void abort(const error& reason) override {
......@@ -130,7 +110,6 @@ private:
size_t read_capacity() const {
return rd_buf_.size() - rd_buf_pos_;
}
net::stream_socket fd_;
shared_count count_;
......@@ -140,6 +119,8 @@ private:
byte_buffer wr_buf_;
byte_buffer rd_buf_;
net::socket_manager* mgr_ = nullptr;
};
struct fixture {
......@@ -168,7 +149,9 @@ struct fixture {
make_manager(net::stream_socket fd, std::string name) {
auto mock = mock_event_layer::make(fd, std::move(name), manager_count);
auto mock_ptr = mock.get();
return {mock_ptr, net::socket_manager::make(&mpx, fd, std::move(mock))};
auto mgr = net::socket_manager::make(&mpx, fd, std::move(mock));
std::ignore = mgr->init(settings{});
return {mock_ptr, std::move(mgr)};
}
void init() {
......
......@@ -75,18 +75,19 @@ public:
const settings&) override {
down = down_ptr;
down->request_messages();
if (auto ptr = adapter_type::try_open(mgr, std::move(output_))) {
adapter_ = std::move(ptr);
if (auto buf = output_.try_open()) {
// Note: the execution unit is the owner of this object. As as long as the
// execution unit lives, accessing `this` is safe.
auto do_resume = make_action([this] { down->request_messages(); });
auto do_cancel = make_action([this] { down->close(); });
adapter_ = adapter_type::make(std::move(buf), mgr, std::move(do_resume),
std::move(do_cancel));
return none;
} else {
FAIL("unable to open the resource");
}
}
void continue_reading() override {
down->request_messages();
}
bool prepare_send() override {
return true;
}
......
......@@ -80,10 +80,6 @@ public:
return true;
}
void continue_reading() override {
// nop
}
ptrdiff_t consume(byte_span data, byte_span) override {
MESSAGE("dummy app received " << data.size() << " bytes");
// Store the received bytes.
......@@ -193,6 +189,7 @@ SCENARIO("ssl::transport::make_client performs the client handshake") {
THEN("CAF transparently calls SSL_connect") {
net::multiplexer mpx{nullptr};
mpx.set_thread_id();
std::ignore = mpx.init();
auto ctx = unbox(ssl::context::make_client(ssl::tls::any));
auto conn = unbox(ctx.new_connection(client_fd));
auto done = std::make_shared<bool>(false);
......@@ -230,6 +227,7 @@ SCENARIO("ssl::transport::make_server performs the server handshake") {
THEN("CAF transparently calls SSL_accept") {
net::multiplexer mpx{nullptr};
mpx.set_thread_id();
std::ignore = mpx.init();
auto ctx = unbox(ssl::context::make_server(ssl::tls::any));
REQUIRE(ctx.use_certificate_from_file(cert_1_pem_path, //
ssl::format::pem));
......
......@@ -93,10 +93,6 @@ public:
return static_cast<ptrdiff_t>(recv_buf_->size());
}
void continue_reading() override {
FAIL("continue_reading called");
}
bool prepare_send() override {
MESSAGE("prepare_send called");
auto& buf = down->output_buffer();
......
......@@ -65,10 +65,6 @@ public:
return !self->terminated();
}
void continue_reading() override {
CAF_FAIL("continue_reading called");
}
bool done_sending() override {
return self->try_block_mailbox();
}
......
......@@ -40,10 +40,6 @@ public:
return true;
}
void continue_reading() override {
// nop
}
void abort(const error& reason) override {
CAF_FAIL("app::abort called: " << reason);
}
......
......@@ -47,10 +47,6 @@ public:
return true;
}
void continue_reading() override {
// nop
}
void abort(const error& reason) override {
CAF_FAIL("app::abort called: " << reason);
}
......
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