Commit 48c2ff93 authored by Dominik Charousset's avatar Dominik Charousset

Port to latest CAF flow API

parent 914428fa
...@@ -60,9 +60,9 @@ caf_incubator_add_component( ...@@ -60,9 +60,9 @@ caf_incubator_add_component(
ip ip
multiplexer multiplexer
net.actor_shell net.actor_shell
net.consumer_adapter
net.length_prefix_framing net.length_prefix_framing
net.observer_adapter net.producer_adapter
net.publisher_adapter
net.typed_actor_shell net.typed_actor_shell
net.web_socket.client net.web_socket.client
net.web_socket.handshake net.web_socket.handshake
......
...@@ -87,7 +87,7 @@ public: ...@@ -87,7 +87,7 @@ public:
using abstract_actor::enqueue; using abstract_actor::enqueue;
void enqueue(mailbox_element_ptr ptr, execution_unit* eu) override; bool enqueue(mailbox_element_ptr ptr, execution_unit* eu) override;
mailbox_element* peek_at_next_mailbox_element() override; mailbox_element* peek_at_next_mailbox_element() override;
......
...@@ -18,7 +18,7 @@ public: ...@@ -18,7 +18,7 @@ public:
~actor_proxy_impl() override; ~actor_proxy_impl() override;
void enqueue(mailbox_element_ptr what, execution_unit* context) override; bool enqueue(mailbox_element_ptr what, execution_unit* context) override;
void kill_proxy(execution_unit* ctx, error rsn) override; void kill_proxy(execution_unit* ctx, error rsn) override;
......
...@@ -37,8 +37,8 @@ public: ...@@ -37,8 +37,8 @@ public:
// -- member functions ------------------------------------------------------- // -- member functions -------------------------------------------------------
template <class ParentPtr> template <class LowerLayerPtr>
error init(socket_manager* owner, ParentPtr parent, const settings& config) { error init(socket_manager* owner, LowerLayerPtr parent, const settings& config) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
owner_ = owner; owner_ = owner;
cfg_ = config; cfg_ = config;
...@@ -48,8 +48,8 @@ public: ...@@ -48,8 +48,8 @@ public:
return none; return none;
} }
template <class ParentPtr> template <class LowerLayerPtr>
bool handle_read_event(ParentPtr parent) { bool handle_read_event(LowerLayerPtr parent) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
if (auto x = accept(parent->handle())) { if (auto x = accept(parent->handle())) {
socket_manager_ptr child = factory_.make(*x, owner_->mpx_ptr()); socket_manager_ptr child = factory_.make(*x, owner_->mpx_ptr());
...@@ -66,14 +66,19 @@ public: ...@@ -66,14 +66,19 @@ public:
} }
} }
template <class ParentPtr> template <class LowerLayerPtr>
bool handle_write_event(ParentPtr) { static void continue_reading(LowerLayerPtr) {
// nop
}
template <class LowerLayerPtr>
bool handle_write_event(LowerLayerPtr) {
CAF_LOG_ERROR("connection_acceptor received write event"); CAF_LOG_ERROR("connection_acceptor received write event");
return false; return false;
} }
template <class ParentPtr> template <class LowerLayerPtr>
void abort(ParentPtr, const error& reason) { void abort(LowerLayerPtr, const error& reason) {
CAF_LOG_ERROR("connection_acceptor aborts due to an error: " << reason); CAF_LOG_ERROR("connection_acceptor aborts due to an error: " << reason);
factory_.abort(reason); factory_.abort(reason);
} }
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/async/consumer.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 : public async::consumer, public ref_counted {
public:
using buf_ptr = intrusive_ptr<Buffer>;
void on_producer_ready() override {
// nop
}
void on_producer_wakeup() override {
mgr_->mpx().register_writing(mgr_);
}
void ref_consumer() const noexcept override {
this->ref();
}
void deref_consumer() const noexcept override {
this->deref();
}
template <class Policy, class OnNext, class OnError = unit_t>
bool consume(Policy policy, size_t demand, OnNext&& on_next,
OnError on_error = OnError{}) {
return buf_->consume(policy, demand, std::forward<OnNext>(on_next),
std::move(on_error));
}
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};
buf->set_consumer(adapter);
return adapter;
} else {
return nullptr;
}
}
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)) {
// nop
}
intrusive_ptr<socket_manager> mgr_;
intrusive_ptr<Buffer> buf_;
};
template <class T>
using consumer_adapter_ptr = intrusive_ptr<consumer_adapter<T>>;
} // namespace caf::net
...@@ -19,6 +19,7 @@ ...@@ -19,6 +19,7 @@
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
#include "caf/tag/message_oriented.hpp" #include "caf/tag/message_oriented.hpp"
#include "caf/tag/no_auto_reading.hpp"
#include "caf/tag/stream_oriented.hpp" #include "caf/tag/stream_oriented.hpp"
namespace caf::net { namespace caf::net {
...@@ -53,6 +54,7 @@ public: ...@@ -53,6 +54,7 @@ public:
template <class LowerLayerPtr> template <class LowerLayerPtr>
error init(socket_manager* owner, LowerLayerPtr down, const settings& cfg) { error init(socket_manager* owner, LowerLayerPtr down, const settings& cfg) {
if constexpr (!std::is_base_of_v<tag::no_auto_reading, UpperLayer>)
down->configure_read(receive_policy::exactly(hdr_size)); down->configure_read(receive_policy::exactly(hdr_size));
return upper_layer_.init(owner, this_layer_ptr(down), cfg); return upper_layer_.init(owner, this_layer_ptr(down), cfg);
} }
...@@ -81,7 +83,7 @@ public: ...@@ -81,7 +83,7 @@ public:
template <class LowerLayerPtr> template <class LowerLayerPtr>
static void suspend_reading(LowerLayerPtr down) { static void suspend_reading(LowerLayerPtr down) {
return down->suspend_reading(); down->configure_read(receive_policy::stop());
} }
template <class LowerLayerPtr> template <class LowerLayerPtr>
...@@ -129,6 +131,11 @@ public: ...@@ -129,6 +131,11 @@ public:
// -- interface for the lower layer ------------------------------------------ // -- interface for the lower layer ------------------------------------------
template <class LowerLayerPtr>
void continue_reading(LowerLayerPtr down) {
down->configure_read(receive_policy::exactly(hdr_size));
}
template <class LowerLayerPtr> template <class LowerLayerPtr>
std::enable_if_t<detail::has_after_reading_v< std::enable_if_t<detail::has_after_reading_v<
UpperLayer, UpperLayer,
...@@ -180,6 +187,7 @@ public: ...@@ -180,6 +187,7 @@ public:
auto [msg_size, msg] = split(input); auto [msg_size, msg] = split(input);
if (msg_size == msg.size() && msg_size + hdr_size == input.size()) { if (msg_size == msg.size() && msg_size + hdr_size == input.size()) {
if (upper_layer_.consume(this_layer, msg) >= 0) { if (upper_layer_.consume(this_layer, msg) >= 0) {
if (!down->stopped())
down->configure_read(receive_policy::exactly(hdr_size)); down->configure_read(receive_policy::exactly(hdr_size));
return static_cast<ptrdiff_t>(input.size()); return static_cast<ptrdiff_t>(input.size());
} else { } else {
......
...@@ -30,6 +30,10 @@ public: ...@@ -30,6 +30,10 @@ public:
return lptr_->handle(llptr_); return lptr_->handle(llptr_);
} }
void suspend_reading() {
return lptr_->suspend_reading(llptr_);
}
void begin_binary_message() { void begin_binary_message() {
lptr_->begin_binary_message(llptr_); lptr_->begin_binary_message(llptr_);
} }
......
...@@ -74,6 +74,25 @@ public: ...@@ -74,6 +74,25 @@ public:
/// @thread-safe /// @thread-safe
void discard(const socket_manager_ptr& mgr); void discard(const socket_manager_ptr& mgr);
/// Stops further reading by `mgr`.
/// @thread-safe
void shutdown_reading(const socket_manager_ptr& mgr);
/// Stops further writing by `mgr`.
/// @thread-safe
void shutdown_writing(const socket_manager_ptr& mgr);
/// Schedules an action for execution on this multiplexer.
/// @thread-safe
void schedule(const action& what);
/// Schedules an action for execution on this multiplexer.
/// @thread-safe
template <class F>
void schedule_fn(F f) {
schedule(make_action(std::move(f)));
}
/// Registers `mgr` for initialization in the multiplexer's thread. /// Registers `mgr` for initialization in the multiplexer's thread.
/// @thread-safe /// @thread-safe
void init(const socket_manager_ptr& mgr); void init(const socket_manager_ptr& mgr);
...@@ -111,10 +130,6 @@ protected: ...@@ -111,10 +130,6 @@ protected:
/// Deletes a known socket manager from the pollset. /// Deletes a known socket manager from the pollset.
void del(ptrdiff_t index); void del(ptrdiff_t index);
/// Writes `opcode` and pointer to `mgr` the the pipe for handling an event
/// later via the pollset updater.
void write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr);
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
/// Bookkeeping data for managed sockets. /// Bookkeeping data for managed sockets.
...@@ -139,6 +154,17 @@ protected: ...@@ -139,6 +154,17 @@ protected:
/// Signals whether shutdown has been requested. /// Signals whether shutdown has been requested.
bool shutting_down_ = false; bool shutting_down_ = false;
private:
/// Writes `opcode` and pointer to `mgr` the the pipe for handling an event
/// later via the pollset updater.
template <class T>
void write_to_pipe(uint8_t opcode, T* ptr);
template <class Enum, class T>
std::enable_if_t<std::is_enum_v<Enum>> write_to_pipe(Enum opcode, T* ptr) {
write_to_pipe(static_cast<uint8_t>(opcode), ptr);
}
}; };
} // namespace caf::net } // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/async/observer_buffer.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.hpp"
namespace caf::net {
/// Connects a socket manager to an asynchronous publisher using a buffer.
/// Whenever the buffer becomes non-empty, the adapter registers the socket
/// manager for writing. The usual pattern for using the adapter then is to call
/// `poll` on the adapter in `prepare_send`.
template <class T>
class observer_adapter : public async::observer_buffer<T> {
public:
using super = async::observer_buffer<T>;
explicit observer_adapter(socket_manager* owner) : mgr_(owner) {
// nop
}
private:
void deinit(std::unique_lock<std::mutex>& guard) final {
wakeup(guard);
mgr_ = nullptr;
}
void wakeup(std::unique_lock<std::mutex>&) final {
mgr_->mpx().register_writing(mgr_);
}
intrusive_ptr<socket_manager> mgr_;
};
template <class T>
using observer_adapter_ptr = intrusive_ptr<observer_adapter<T>>;
} // namespace caf::net
...@@ -24,16 +24,16 @@ public: ...@@ -24,16 +24,16 @@ public:
// -- constants -------------------------------------------------------------- // -- constants --------------------------------------------------------------
static constexpr uint8_t register_reading_code = 0x00; enum class code : uint8_t {
register_reading,
static constexpr uint8_t register_writing_code = 0x01; register_writing,
init_manager,
static constexpr uint8_t init_manager_code = 0x02; discard_manager,
shutdown_reading,
static constexpr uint8_t discard_manager_code = 0x03; shutdown_writing,
run_action,
static constexpr uint8_t shutdown_code = 0x04; shutdown,
};
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
pollset_updater(pipe_socket read_handle, multiplexer* parent); pollset_updater(pipe_socket read_handle, multiplexer* parent);
...@@ -57,6 +57,8 @@ public: ...@@ -57,6 +57,8 @@ public:
void handle_error(sec code) override; void handle_error(sec code) override;
void continue_reading() override;
private: private:
msg_buf buf_; msg_buf buf_;
size_t buf_size_ = 0; size_t buf_size_ = 0;
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <memory>
#include <new>
#include "caf/async/producer.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 {
template <class Buffer>
class producer_adapter final : public 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;
void on_consumer_ready() override {
// nop
}
void on_consumer_cancel() override {
mgr_->mpx().schedule_fn([adapter = strong_this()] { //
adapter->on_cancel();
});
}
void on_consumer_demand(size_t new_demand) override {
auto prev = demand_.fetch_add(new_demand);
if (prev == 0)
mgr_->mpx().schedule_fn([adapter = strong_this()] {
adapter->continue_reading();
});
}
void ref_producer() const noexcept override {
this->ref();
}
void deref_producer() const noexcept override {
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};
buf->set_producer(adapter);
return adapter;
} else {
return nullptr;
}
}
/// Returns the current consumer demand.
size_t demand() const noexcept {
return demand_;
}
/// Makes `item` available to the consumer.
/// @returns the remaining demand.
/// @pre `demand() > 0`
size_t push(const value_type& item) {
CAF_ASSERT(demand_ > 0);
buf_->push(item);
return --demand_;
}
/// Makes `items` available to the consumer.
/// @returns the remaining demand.
/// @pre `demand() >= items.size()`
size_t push(span<const value_type> items) {
CAF_ASSERT(demand_ >= items.size());
buf_->push(items);
return demand_ -= items.size();
}
void close() {
if (buf_) {
buf_->close();
buf_ = nullptr;
mgr_ = nullptr;
}
}
void abort(error reason) {
if (buf_) {
buf_->abort(std::move(reason));
buf_ = nullptr;
mgr_ = nullptr;
}
}
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)
: demand_(0), mgr_(owner), buf_(std::move(buf)) {
// nop
}
void continue_reading() {
if (mgr_)
mgr_->continue_reading();
}
void on_cancel() {
if (buf_)
mgr_->mpx().shutdown_reading(mgr_);
}
auto strong_this() {
return intrusive_ptr{this};
}
atomic_count demand_;
char pad[CAF_CACHE_LINE_SIZE - sizeof(atomic_count)];
intrusive_ptr<socket_manager> mgr_;
intrusive_ptr<Buffer> buf_;
};
template <class T>
using producer_adapter_ptr = intrusive_ptr<producer_adapter<T>>;
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <memory>
#include <new>
#include "caf/async/publisher.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/subscription.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.hpp"
namespace caf::net {
template <class T>
class publisher_adapter final : public async::publisher<T>::impl,
public flow::subscription::impl {
public:
publisher_adapter(socket_manager* owner, uint32_t max_in_flight,
uint32_t batch_size)
: batch_size_(batch_size), max_in_flight_(max_in_flight), mgr_(owner) {
CAF_ASSERT(max_in_flight > batch_size);
buf_ = reinterpret_cast<T*>(malloc(sizeof(T) * max_in_flight * 2));
}
~publisher_adapter() {
auto first = buf_ + rd_pos_;
auto last = buf_ + wr_pos_;
std::destroy(first, last);
free(buf_);
}
void subscribe(flow::observer<T> sink) override {
if (std::unique_lock guard{mtx_}; !sink_) {
sink_ = std::move(sink);
auto ptr = intrusive_ptr<flow::subscription::impl>{this};
sink_.on_attach(flow::subscription{std::move(ptr)});
} else {
sink.on_error(
make_error(sec::downstream_already_exists,
"caf::net::publisher_adapter only allows one observer"));
}
}
void request(size_t n) override {
CAF_ASSERT(n > 0);
// Reactive Streams specification 1.0.3:
// > Subscription.request MUST place an upper bound on possible synchronous
// > recursion between Publisher and Subscriber.
std::unique_lock guard{mtx_};
if (!sink_)
return;
credit_ += static_cast<uint32_t>(n);
if (!in_request_body_) {
in_request_body_ = true;
auto n = std::min(size(), credit_);
// When full, we take whatever we can out of the buffer even if the client
// requests less than a batch. Otherwise, we try to wait until we have
// sufficient credit for a full batch.
if (n == 0) {
in_request_body_ = false;
return;
} else if (full()) {
wakeup();
} else if (n < batch_size_) {
in_request_body_ = false;
return;
}
auto m = std::min(n, batch_size_);
deliver(m);
n -= m;
while (sink_ && n >= batch_size_) {
deliver(batch_size_);
n -= batch_size_;
}
shift_elements();
in_request_body_ = false;
}
}
void cancel() override {
std::unique_lock guard{mtx_};
discard();
}
bool disposed() const noexcept override {
std::unique_lock guard{mtx_};
return mgr_ == nullptr;
}
void on_complete() {
std::unique_lock guard{mtx_};
if (sink_) {
sink_.on_complete();
sink_ = nullptr;
}
}
void on_error(const error& what) {
std::unique_lock guard{mtx_};
if (sink_) {
sink_.on_error(what);
sink_ = nullptr;
}
}
/// Enqueues a new element to the buffer.
/// @returns The remaining buffer capacity. If this function return 0, the
/// manager MUST suspend reading until the observer consumes at least
/// one element.
size_t push(T value) {
std::unique_lock guard{mtx_};
if (!mgr_)
return 0;
new (buf_ + wr_pos_) T(std::move(value));
++wr_pos_;
if (auto n = std::min(size(), credit_); n >= batch_size_) {
do {
deliver(n);
n -= batch_size_;
} while (n >= batch_size_);
shift_elements();
}
if (auto result = capacity(); result == 0 && credit_ > 0) {
// Can only reach here if batch_size_ > credit_.
deliver(credit_);
shift_elements();
return capacity();
} else {
return result;
}
}
/// Pushes any buffered items to the observer as long as there is any
/// available credit.
void flush() {
std::unique_lock guard{mtx_};
while (sink_) {
if (auto n = std::min({size(), credit_, batch_size_}); n > 0)
deliver(n);
else
break;
}
shift_elements();
}
private:
void discard() {
if (mgr_) {
sink_ = nullptr;
mgr_->mpx().discard(mgr_);
mgr_ = nullptr;
credit_ = 0;
}
}
/// @pre `mtx_` is locked
[[nodiscard]] uint32_t size() const noexcept {
return wr_pos_ - rd_pos_;
}
/// @pre `mtx_` is locked
[[nodiscard]] uint32_t capacity() const noexcept {
return max_in_flight_ - size();
}
/// @pre `mtx_` is locked
[[nodiscard]] bool full() const noexcept {
return capacity() == 0;
}
/// @pre `mtx_` is locked
[[nodiscard]] bool empty() const noexcept {
return wr_pos_ == rd_pos_;
}
/// @pre `mtx_` is locked
void wakeup() {
CAF_ASSERT(mgr_ != nullptr);
mgr_->mpx().register_reading(mgr_);
}
void deliver(uint32_t n) {
auto first = buf_ + rd_pos_;
auto last = first + n;
CAF_ASSERT(rd_pos_ + n <= wr_pos_);
rd_pos_ += n;
CAF_ASSERT(credit_ >= n);
credit_ -= n;
sink_.on_next(span<const T>{first, n});
std::destroy(first, last);
}
void shift_elements() {
if (rd_pos_ >= max_in_flight_) {
if (empty()) {
rd_pos_ = 0;
wr_pos_ = 0;
} else {
// No need to check for overlap: the first half of the buffer is empty.
auto first = buf_ + rd_pos_;
auto last = buf_ + wr_pos_;
std::uninitialized_move(first, last, buf_);
std::destroy(first, last);
wr_pos_ -= rd_pos_;
rd_pos_ = 0;
}
}
}
mutable std::recursive_mutex mtx_;
/// Allocated to max_in_flight_ * 2, but at most holds max_in_flight_ elements
/// at any point in time. We dynamically shift elements into the first half of
/// the buffer whenever rd_pos_ crosses the midpoint.
T* buf_;
uint32_t rd_pos_ = 0;
uint32_t wr_pos_ = 0;
uint32_t credit_ = 0;
uint32_t batch_size_;
uint32_t max_in_flight_;
bool in_request_body_ = false;
flow::observer<T> sink_;
intrusive_ptr<socket_manager> mgr_;
};
template <class T>
using publisher_adapter_ptr = intrusive_ptr<publisher_adapter<T>>;
} // namespace caf::net
...@@ -83,4 +83,12 @@ error CAF_NET_EXPORT child_process_inherit(socket x, bool new_value); ...@@ -83,4 +83,12 @@ error CAF_NET_EXPORT child_process_inherit(socket x, bool new_value);
/// @relates socket /// @relates socket
error CAF_NET_EXPORT nonblocking(socket x, bool new_value); error CAF_NET_EXPORT nonblocking(socket x, bool new_value);
/// Disallows further reads from the socket.
/// @relates socket
error CAF_NET_EXPORT shutdown_read(socket x);
/// Disallows further writes to the socket.
/// @relates socket
error CAF_NET_EXPORT shutdown_write(socket x);
} // namespace caf::net } // namespace caf::net
...@@ -48,6 +48,11 @@ public: ...@@ -48,6 +48,11 @@ public:
return handle_; return handle_;
} }
/// @private
void handle(socket new_handle) {
handle_ = new_handle;
}
/// Returns a reference to the hosting @ref actor_system instance. /// Returns a reference to the hosting @ref actor_system instance.
actor_system& system() noexcept; actor_system& system() noexcept;
...@@ -128,6 +133,10 @@ public: ...@@ -128,6 +133,10 @@ public:
void register_writing(); void register_writing();
void shutdown_reading();
void shutdown_writing();
// -- pure virtual member functions ------------------------------------------ // -- pure virtual member functions ------------------------------------------
virtual error init(const settings& config) = 0; virtual error init(const settings& config) = 0;
...@@ -142,6 +151,10 @@ public: ...@@ -142,6 +151,10 @@ public:
/// @param code The error code as reported by the operating system. /// @param code The error code as reported by the operating system.
virtual void handle_error(sec code) = 0; virtual void handle_error(sec code) = 0;
/// Restarts a socket manager that suspended reads. Calling this member
/// function on active managers is a no-op.
virtual void continue_reading() = 0;
protected: protected:
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
...@@ -204,6 +217,10 @@ public: ...@@ -204,6 +217,10 @@ public:
return protocol_.abort(this, abort_reason_); return protocol_.abort(this, abort_reason_);
} }
void continue_reading() override {
return protocol_.continue_reading(this);
}
auto& protocol() noexcept { auto& protocol() noexcept {
return protocol_; return protocol_;
} }
......
...@@ -23,10 +23,6 @@ public: ...@@ -23,10 +23,6 @@ public:
// nop // nop
} }
void suspend_reading() {
return lptr_->suspend_reading(llptr_);
}
bool can_send_more() const noexcept { bool can_send_more() const noexcept {
return lptr_->can_send_more(llptr_); return lptr_->can_send_more(llptr_);
} }
...@@ -59,6 +55,10 @@ public: ...@@ -59,6 +55,10 @@ public:
lptr_->configure_read(llptr_, policy); lptr_->configure_read(llptr_, policy);
} }
bool stopped() const noexcept {
return lptr_->stopped(llptr_);
}
private: private:
Layer* lptr_; Layer* lptr_;
LowerLayerPtr llptr_; LowerLayerPtr llptr_;
......
...@@ -49,11 +49,6 @@ public: ...@@ -49,11 +49,6 @@ public:
// -- interface for stream_oriented_layer_ptr -------------------------------- // -- interface for stream_oriented_layer_ptr --------------------------------
template <class ParentPtr>
void suspend_reading(ParentPtr) {
suspend_reading_ = true;
}
template <class ParentPtr> template <class ParentPtr>
bool can_send_more(ParentPtr) const noexcept { bool can_send_more(ParentPtr) const noexcept {
return write_buf_.size() < max_write_buf_size_; return write_buf_.size() < max_write_buf_size_;
...@@ -98,6 +93,11 @@ public: ...@@ -98,6 +93,11 @@ public:
max_read_size_ = policy.max_size; max_read_size_ = policy.max_size;
} }
template <class ParentPtr>
bool stopped(ParentPtr) const noexcept {
return max_read_size_ == 0;
}
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
auto& read_buffer() noexcept { auto& read_buffer() noexcept {
...@@ -233,13 +233,6 @@ public: ...@@ -233,13 +233,6 @@ public:
if (read_buf_.size() != max_read_size_) if (read_buf_.size() != max_read_size_)
if (offset_ < max_read_size_) if (offset_ < max_read_size_)
read_buf_.resize(max_read_size_); read_buf_.resize(max_read_size_);
// Upper layer may have called suspend_reading().
if (suspend_reading_) {
suspend_reading_ = false;
if constexpr (has_after_reading)
upper_layer_.after_reading(this_layer_ptr);
return false;
}
} else if (read_res < 0) { } else if (read_res < 0) {
// Try again later on temporary errors such as EWOULDBLOCK and // Try again later on temporary errors such as EWOULDBLOCK and
// stop reading on the socket on hard errors. // stop reading on the socket on hard errors.
...@@ -298,6 +291,14 @@ public: ...@@ -298,6 +291,14 @@ public:
} }
} }
template <class ParentPtr>
void continue_reading(ParentPtr parent) {
if (max_read_size_ == 0) {
auto this_layer_ptr = make_stream_oriented_layer_ptr(this, parent);
upper_layer_.continue_reading(this_layer_ptr);
}
}
template <class ParentPtr> template <class ParentPtr>
void abort(ParentPtr parent, const error& reason) { void abort(ParentPtr parent, const error& reason) {
auto this_layer_ptr = make_stream_oriented_layer_ptr(this, parent); auto this_layer_ptr = make_stream_oriented_layer_ptr(this, parent);
...@@ -323,9 +324,6 @@ private: ...@@ -323,9 +324,6 @@ private:
// Stores the offset in `read_buf_` since last calling `upper_layer_.consume`. // Stores the offset in `read_buf_` since last calling `upper_layer_.consume`.
ptrdiff_t delta_offset_ = 0; ptrdiff_t delta_offset_ = 0;
// Stores whether the user called `suspend_reading()`.
bool suspend_reading_ = false;
// Caches incoming data. // Caches incoming data.
byte_buffer read_buf_; byte_buffer read_buf_;
......
...@@ -90,6 +90,12 @@ public: ...@@ -90,6 +90,12 @@ public:
upper_layer_.abort(down, reason); upper_layer_.abort(down, reason);
} }
template <class LowerLayerPtr>
void continue_reading(LowerLayerPtr down) {
if (handshake_complete())
upper_layer_.continue_reading(down);
}
template <class LowerLayerPtr> template <class LowerLayerPtr>
ptrdiff_t consume(LowerLayerPtr down, byte_span input, byte_span delta) { ptrdiff_t consume(LowerLayerPtr down, byte_span input, byte_span delta) {
CAF_LOG_TRACE(CAF_ARG2("socket", down->handle().id) CAF_LOG_TRACE(CAF_ARG2("socket", down->handle().id)
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/async/publisher.hpp"
#include "caf/net/observer_adapter.hpp"
#include "caf/net/publisher_adapter.hpp"
namespace caf::net::web_socket::internal {
/// Implements a WebSocket application that uses two flows for bidirectional
/// communication: one input flow and one output flow.
template <class Reader, class Writer>
class bidir_app {
public:
using input_tag = tag::message_oriented;
using reader_output_type = typename Reader::value_type;
using writer_input_type = typename Writer::value_type;
bidir_app(Reader&& reader, Writer&& writer)
: reader_(std::move(reader)), writer_(std::move(writer)) {
// nop
}
async::publisher<reader_input_type>
connect_flows(net::socket_manager* mgr,
async::publisher<writer_input_type> in) {
using make_counted;
// Connect the writer adapter.
using writer_input_t = net::observer_adapter<writer_input_type>;
writer_input_ = make_counted<writer_input_t>(mgr);
in.subscribe(writer_input_->as_observer());
// Create the reader adapter.
using reader_output_t = net::publisher_adapter<node_message>;
reader_output_ = make_counted<reader_output_t>(mgr, reader_.buffer_size(),
reader_.batch_size());
return reader_output_->as_publisher();
}
template <class LowerLayerPtr>
error init(net::socket_manager* mgr, LowerLayerPtr&&, const settings&cfg) {
if (auto err = reader_.init(cfg))
return err;
if (auto err = writer_.init(cfg))
return err;
return none;
}
template <class LowerLayerPtr>
bool prepare_send(LowerLayerPtr down) {
while (down->can_send_more()) {
auto [val, done, err] = writer_input_->poll();
if (val) {
if (!write(down, *val)) {
down->abort_reason(make_error(ec::invalid_message));
return false;
}
} else if (done) {
if (err) {
down->abort_reason(*err);
return false;
}
} else {
break;
}
}
return true;
}
template <class LowerLayerPtr>
bool done_sending(LowerLayerPtr) {
return !writer_input_->has_data();
}
template <class LowerLayerPtr>
void abort(LowerLayerPtr, const error& reason) {
reader_output_->flush();
if (reason == sec::socket_disconnected || reason == sec::discarded)
reader_output_->on_complete();
else
reader_output_->on_error(reason);
}
template <class LowerLayerPtr>
void after_reading(LowerLayerPtr) {
reader_output_->flush();
}
template <class LowerLayerPtr>
ptrdiff_t consume_text(LowerLayerPtr down, caf::string_view text) {
reader_msgput_type msg;
if (reader_.deserialize_text(text, msg)) {
if (reader_output_->push(std::move(msg)) == 0)
down->suspend_reading();
return static_cast<ptrdiff_t>(text.size());
} else {
down->abort_reason(make_error(ec::invalid_message));
return -1;
}
}
template <class LowerLayerPtr>
ptrdiff_t consume_binary(LowerLayerPtr, caf::byte_span bytes) {
reader_msgput_type msg;
if (reader_.deserialize_binary(bytes, msg)) {
if (reader_output_->push(std::move(msg)) == 0)
down->suspend_reading();
return static_cast<ptrdiff_t>(text.size());
} else {
down->abort_reason(make_error(ec::invalid_message));
return -1;
}
}
private:
template <class LowerLayerPtr>
bool write(LowerLayerPtr down, const writer_input_type& msg) {
if (writer_.is_text_message(msg)) {
down->begin_text_message();
if (writer_.serialize_text(msg, down->text_message_buffer())) {
down->end_text_message();
return true;
} else {
return false;
}
} else {
down->begin_binary_message();
if (writer_.serialize_binary(msg, down->binary_message_buffer())) {
down->end_binary_message();
return true;
} else {
return false;
}
}
}
/// Deserializes text or binary messages from the socket.
Reader reader_;
/// Serializes text or binary messages to the socket.
Writer writer_;
/// Forwards outgoing messages to the peer. We write whatever we receive from
/// this channel to the socket.
net::observer_adapter_ptr<node_message> writer_input_;
/// After receiving messages from the socket, we publish to this adapter for
/// downstream consumers.
net::publisher_adapter_ptr<node_message> reader_output_;
};
} // namespace caf::net::web_socket::internal
namespace caf::net::web_socket {
/// Connects to a WebSocket server for bidirectional communication.
/// @param sys The enclosing actor system.
/// @param cfg Provides optional configuration parameters such as WebSocket
/// protocols and extensions for the handshake.
/// @param locator Identifies the WebSocket server.
/// @param writer_input Publisher of events that go out to the server.
/// @param reader Reads messages from the server and publishes them locally.
/// @param writer Writes messages from the @p writer_input to text or binary
/// messages for sending them to the server.
/// @returns a publisher that makes messages from the server accessible on
/// success, an error otherwise.
template <template <class> class Transport = stream_transport, class Reader,
class Writer>
expected<async::publisher<typename Reader::value_type>>
flow_connect_bidir(actor_system& sys, const settings& cfg, uri locator,
async::publisher writer_input, Reader reader,
Writer writer) {
using stack_t
= Transport<web_socket::client<internal::bidir_app<Reader, Writer>>>;
using impl = socket_manager_impl<stack_t>;
if (locator.empty()) {
return make_error(sec::invalid_argument, __func__,
"cannot connect to empty URI");
} else if (locator.scheme() != "ws") {
return make_error(sec::invalid_argument, __func__,
"malformed URI, expected format 'ws://<authority>'");
} else if (!locator.fragment().empty()) {
return make_error(sec::invalid_argument, __func__,
"query and fragment components are not supported");
} else if (locator.authority().empty()) {
return make_error(sec::invalid_argument, __func__,
"malformed URI, expected format 'ws://<authority>'");
} else if (auto sock = impl::connect_to(locator.authority())) {
web_socket::handshake hs;
hs.host(to_string(locator.authority()));
if (locator.path().empty())
hs.endpoint("/");
else
hs.endpoint(to_string(locator.path()));
if (auto protocols = get_as<std::string>(cfg, "protocols"))
hs.protocols(std::move(*protocols));
if (auto extensions = get_as<std::string>(cfg, "extensions"))
hs.extensions(std::move(*extensions));
auto mgr = make_counted<impl>(*sock, sys.network_manager().mpx_ptr(),
std::move(hs), std::move(reader),
std::move(writer));
auto out = mgr->upper_layer().connect_flows(std::move(writer_input));
if (auto err = mgr->init(cfg); !err) {
return {std::move(out)};
} else {
return {std::move(err)};
}
} else {
return {std::move(sock.error())};
}
}
} // namespace caf::net::web_socket
...@@ -86,6 +86,14 @@ public: ...@@ -86,6 +86,14 @@ public:
return parent->handle(); return parent->handle();
} }
template <class LowerLayerPtr>
static void suspend_reading(LowerLayerPtr) {
CAF_RAISE_ERROR("suspending / resuming a WebSocket not implemented yet");
// TODO: uncommenting this isn't enough since consume() also must make sure
// to not override the configure_read.
// down->configure_read(receive_policy::stop());
}
template <class LowerLayerPtr> template <class LowerLayerPtr>
static constexpr void begin_binary_message(LowerLayerPtr) { static constexpr void begin_binary_message(LowerLayerPtr) {
// nop // nop
...@@ -138,6 +146,11 @@ public: ...@@ -138,6 +146,11 @@ public:
return upper_layer_.done_sending(down); return upper_layer_.done_sending(down);
} }
template <class LowerLayerPtr>
static void continue_reading(LowerLayerPtr down) {
down->configure_read(receive_policy::up_to(2048));
}
template <class LowerLayerPtr> template <class LowerLayerPtr>
void abort(LowerLayerPtr down, const error& reason) { void abort(LowerLayerPtr down, const error& reason) {
upper_layer_.abort(down, reason); upper_layer_.abort(down, reason);
......
...@@ -75,6 +75,11 @@ public: ...@@ -75,6 +75,11 @@ public:
return handshake_complete_ ? upper_layer_.done_sending(down) : true; return handshake_complete_ ? upper_layer_.done_sending(down) : true;
} }
template <class LowerLayerPtr>
static void continue_reading(LowerLayerPtr down) {
down->configure_read(receive_policy::up_to(handshake::max_http_size));
}
template <class LowerLayerPtr> template <class LowerLayerPtr>
void abort(LowerLayerPtr down, const error& reason) { void abort(LowerLayerPtr down, const error& reason) {
if (handshake_complete_) if (handshake_complete_)
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
namespace caf::tag {
/// Tells message-oriented transports to *not* implicitly register an
/// application for reading as part of the initialization when used a base type.
struct no_auto_reading {};
} // namespace caf::tag
...@@ -6,6 +6,7 @@ ...@@ -6,6 +6,7 @@
#include <algorithm> #include <algorithm>
#include "caf/action.hpp"
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
...@@ -65,6 +66,24 @@ short to_bitmask(operation op) { ...@@ -65,6 +66,24 @@ short to_bitmask(operation op) {
} // namespace } // namespace
template <class T>
void multiplexer::write_to_pipe(uint8_t opcode, T* ptr) {
pollset_updater::msg_buf buf;
if (ptr)
intrusive_ptr_add_ref(ptr);
buf[0] = static_cast<byte>(opcode);
auto value = reinterpret_cast<intptr_t>(ptr);
memcpy(buf.data() + 1, &value, sizeof(intptr_t));
ptrdiff_t res = -1;
{ // Lifetime scope of guard.
std::lock_guard<std::mutex> guard{write_lock_};
if (write_handle_ != invalid_socket)
res = write(write_handle_, buf);
}
if (res <= 0 && ptr)
intrusive_ptr_release(ptr);
}
// -- constructors, destructors, and assignment operators ---------------------- // -- constructors, destructors, and assignment operators ----------------------
multiplexer::multiplexer(middleman* owner) : owner_(owner) { multiplexer::multiplexer(middleman* owner) : owner_(owner) {
...@@ -117,19 +136,19 @@ actor_system& multiplexer::system() { ...@@ -117,19 +136,19 @@ actor_system& multiplexer::system() {
void multiplexer::register_reading(const socket_manager_ptr& mgr) { void multiplexer::register_reading(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id)); CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) { if (std::this_thread::get_id() == tid_) {
if (shutting_down_) { if (shutting_down_ || mgr->abort_reason()) {
// nop // nop
} else if (mgr->mask() != operation::none) { } else if (mgr->mask() != operation::none) {
if (auto index = index_of(mgr); if (auto index = index_of(mgr);
index != -1 && mgr->mask_add(operation::read)) { index != -1 && mgr->mask_add(operation::read)) {
auto& fd = pollset_[index_of(mgr)]; auto& fd = pollset_[index];
fd.events |= input_mask; fd.events |= input_mask;
} }
} else if (mgr->mask_add(operation::read)) { } else if (mgr->mask_add(operation::read)) {
add(mgr); add(mgr);
} }
} else { } else {
write_to_pipe(pollset_updater::register_reading_code, mgr); write_to_pipe(pollset_updater::code::register_reading, mgr.get());
} }
} }
...@@ -141,14 +160,14 @@ void multiplexer::register_writing(const socket_manager_ptr& mgr) { ...@@ -141,14 +160,14 @@ void multiplexer::register_writing(const socket_manager_ptr& mgr) {
} else if (mgr->mask() != operation::none) { } else if (mgr->mask() != operation::none) {
if (auto index = index_of(mgr); if (auto index = index_of(mgr);
index != -1 && mgr->mask_add(operation::write)) { index != -1 && mgr->mask_add(operation::write)) {
auto& fd = pollset_[index_of(mgr)]; auto& fd = pollset_[index];
fd.events |= output_mask; fd.events |= output_mask;
} }
} else if (mgr->mask_add(operation::write)) { } else if (mgr->mask_add(operation::write)) {
add(mgr); add(mgr);
} }
} else { } else {
write_to_pipe(pollset_updater::register_writing_code, mgr); write_to_pipe(pollset_updater::code::register_writing, mgr.get());
} }
} }
...@@ -163,10 +182,51 @@ void multiplexer::discard(const socket_manager_ptr& mgr) { ...@@ -163,10 +182,51 @@ void multiplexer::discard(const socket_manager_ptr& mgr) {
del(mgr_index); del(mgr_index);
} }
} else { } else {
write_to_pipe(pollset_updater::discard_manager_code, mgr); write_to_pipe(pollset_updater::code::discard_manager, mgr.get());
}
}
void multiplexer::shutdown_reading(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) {
if (shutting_down_) {
// nop
} else if (auto index = index_of(mgr); index != -1) {
mgr->mask_del(operation::read);
auto& entry = pollset_[index];
std::ignore = shutdown_read(socket{entry.fd});
entry.events &= ~input_mask;
if (entry.events == 0)
del(index);
}
} else {
write_to_pipe(pollset_updater::code::shutdown_reading, mgr.get());
} }
} }
void multiplexer::shutdown_writing(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) {
if (shutting_down_) {
// nop
} else if (auto index = index_of(mgr); index != -1) {
mgr->mask_del(operation::write);
auto& entry = pollset_[index];
std::ignore = shutdown_write(socket{entry.fd});
entry.events &= ~output_mask;
if (entry.events == 0)
del(index);
}
} else {
write_to_pipe(pollset_updater::code::shutdown_writing, mgr.get());
}
}
void multiplexer::schedule(const action& what) {
CAF_LOG_TRACE("");
write_to_pipe(pollset_updater::code::run_action, what.ptr());
}
void multiplexer::init(const socket_manager_ptr& mgr) { void multiplexer::init(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id)); CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) { if (std::this_thread::get_id() == tid_) {
...@@ -181,7 +241,7 @@ void multiplexer::init(const socket_manager_ptr& mgr) { ...@@ -181,7 +241,7 @@ void multiplexer::init(const socket_manager_ptr& mgr) {
} }
} }
} else { } else {
write_to_pipe(pollset_updater::init_manager_code, mgr); write_to_pipe(pollset_updater::code::init_manager, mgr.get());
} }
} }
...@@ -293,7 +353,8 @@ void multiplexer::shutdown() { ...@@ -293,7 +353,8 @@ void multiplexer::shutdown() {
close_pipe(); close_pipe();
} else { } else {
CAF_LOG_DEBUG("push shutdown event to pipe"); CAF_LOG_DEBUG("push shutdown event to pipe");
write_to_pipe(4, nullptr); write_to_pipe(pollset_updater::code::shutdown,
static_cast<socket_manager*>(nullptr));
} }
} }
...@@ -358,26 +419,4 @@ void multiplexer::del(ptrdiff_t index) { ...@@ -358,26 +419,4 @@ void multiplexer::del(ptrdiff_t index) {
managers_.erase(managers_.begin() + index); managers_.erase(managers_.begin() + index);
} }
void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) {
CAF_ASSERT(opcode == pollset_updater::register_reading_code
|| opcode == pollset_updater::register_writing_code
|| opcode == pollset_updater::init_manager_code
|| opcode == pollset_updater::shutdown_code);
CAF_ASSERT(mgr != nullptr || opcode == pollset_updater::shutdown_code);
pollset_updater::msg_buf buf;
if (opcode != pollset_updater::shutdown_code)
mgr->ref();
buf[0] = static_cast<byte>(opcode);
auto value = reinterpret_cast<intptr_t>(mgr.get());
memcpy(buf.data() + 1, &value, sizeof(intptr_t));
ptrdiff_t res = -1;
{ // Lifetime scope of guard.
std::lock_guard<std::mutex> guard{write_lock_};
if (write_handle_ != invalid_socket)
res = write(write_handle_, buf);
}
if (res <= 0 && opcode != pollset_updater::shutdown_code)
mgr->deref();
}
} // namespace caf::net } // namespace caf::net
...@@ -96,7 +96,7 @@ void abstract_actor_shell::add_multiplexed_response_handler( ...@@ -96,7 +96,7 @@ void abstract_actor_shell::add_multiplexed_response_handler(
// -- overridden functions of abstract_actor ----------------------------------- // -- overridden functions of abstract_actor -----------------------------------
void abstract_actor_shell::enqueue(mailbox_element_ptr ptr, execution_unit*) { bool abstract_actor_shell::enqueue(mailbox_element_ptr ptr, execution_unit*) {
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(!getf(is_blocking_flag)); CAF_ASSERT(!getf(is_blocking_flag));
CAF_LOG_TRACE(CAF_ARG(*ptr)); CAF_LOG_TRACE(CAF_ARG(*ptr));
...@@ -118,7 +118,7 @@ void abstract_actor_shell::enqueue(mailbox_element_ptr ptr, execution_unit*) { ...@@ -118,7 +118,7 @@ void abstract_actor_shell::enqueue(mailbox_element_ptr ptr, execution_unit*) {
// skip any further processing. // skip any further processing.
if (owner_) if (owner_)
owner_->mpx().register_writing(owner_); owner_->mpx().register_writing(owner_);
break; return true;
} }
case intrusive::inbox_result::queue_closed: { case intrusive::inbox_result::queue_closed: {
CAF_LOG_REJECT_EVENT(); CAF_LOG_REJECT_EVENT();
...@@ -129,12 +129,12 @@ void abstract_actor_shell::enqueue(mailbox_element_ptr ptr, execution_unit*) { ...@@ -129,12 +129,12 @@ void abstract_actor_shell::enqueue(mailbox_element_ptr ptr, execution_unit*) {
detail::sync_request_bouncer f{exit_reason()}; detail::sync_request_bouncer f{exit_reason()};
f(sender, mid); f(sender, mid);
} }
break; return false;
} }
case intrusive::inbox_result::success: case intrusive::inbox_result::success:
// Enqueued to a running actors' mailbox: nothing to do. // Enqueued to a running actors' mailbox: nothing to do.
CAF_LOG_ACCEPT_EVENT(false); CAF_LOG_ACCEPT_EVENT(false);
break; return true;
} }
} }
......
...@@ -6,6 +6,7 @@ ...@@ -6,6 +6,7 @@
#include <cstring> #include <cstring>
#include "caf/action.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
...@@ -29,6 +30,21 @@ error pollset_updater::init(const settings&) { ...@@ -29,6 +30,21 @@ error pollset_updater::init(const settings&) {
return nonblocking(handle(), true); return nonblocking(handle(), true);
} }
namespace {
auto as_mgr(intptr_t ptr) {
CAF_LOG_TRACE(CAF_ARG(ptr));
return intrusive_ptr{reinterpret_cast<socket_manager*>(ptr), false};
};
void run_action(intptr_t ptr) {
CAF_LOG_TRACE(CAF_ARG(ptr));
auto f = action{intrusive_ptr{reinterpret_cast<action::impl*>(ptr), false}};
f.run();
};
} // namespace
bool pollset_updater::handle_read_event() { bool pollset_updater::handle_read_event() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
for (;;) { for (;;) {
...@@ -40,23 +56,32 @@ bool pollset_updater::handle_read_event() { ...@@ -40,23 +56,32 @@ bool pollset_updater::handle_read_event() {
if (buf_.size() == buf_size_) { if (buf_.size() == buf_size_) {
buf_size_ = 0; buf_size_ = 0;
auto opcode = static_cast<uint8_t>(buf_[0]); auto opcode = static_cast<uint8_t>(buf_[0]);
intptr_t value; intptr_t ptr;
memcpy(&value, buf_.data() + 1, sizeof(intptr_t)); memcpy(&ptr, buf_.data() + 1, sizeof(intptr_t));
socket_manager_ptr mgr{reinterpret_cast<socket_manager*>(value), false}; switch (static_cast<code>(opcode)) {
switch (opcode) { case code::register_reading:
case register_reading_code: parent_->register_reading(as_mgr(ptr));
parent_->register_reading(mgr); break;
case code::register_writing:
parent_->register_writing(as_mgr(ptr));
break; break;
case register_writing_code: case code::init_manager:
parent_->register_writing(mgr); parent_->init(as_mgr(ptr));
break; break;
case init_manager_code: case code::discard_manager:
parent_->init(mgr); parent_->discard(as_mgr(ptr));
break; break;
case discard_manager_code: case code::shutdown_reading:
parent_->discard(mgr); parent_->shutdown_reading(as_mgr(ptr));
break; break;
case shutdown_code: case code::shutdown_writing:
parent_->shutdown_writing(as_mgr(ptr));
break;
case code::run_action:
run_action(ptr);
break;
case code::shutdown:
CAF_ASSERT(ptr == 0);
parent_->shutdown(); parent_->shutdown();
break; break;
default: default:
...@@ -81,4 +106,8 @@ void pollset_updater::handle_error(sec) { ...@@ -81,4 +106,8 @@ void pollset_updater::handle_error(sec) {
// nop // nop
} }
void pollset_updater::continue_reading() {
register_reading();
}
} // namespace caf::net } // namespace caf::net
...@@ -179,4 +179,14 @@ error nonblocking(socket x, bool new_value) { ...@@ -179,4 +179,14 @@ error nonblocking(socket x, bool new_value) {
#endif // CAF_WINDOWS #endif // CAF_WINDOWS
error shutdown_read(socket x) {
CAF_NET_SYSCALL("shutdown", res, !=, 0, shutdown(x.id, 0));
return caf::none;
}
error shutdown_write(socket x) {
CAF_NET_SYSCALL("shutdown", res, !=, 0, shutdown(x.id, 1));
return caf::none;
}
} // namespace caf::net } // namespace caf::net
...@@ -55,4 +55,12 @@ void socket_manager::register_writing() { ...@@ -55,4 +55,12 @@ void socket_manager::register_writing() {
parent_->register_writing(this); parent_->register_writing(this);
} }
void socket_manager::shutdown_reading() {
parent_->shutdown_reading(this);
}
void socket_manager::shutdown_writing() {
parent_->shutdown_writing(this);
}
} // namespace caf::net } // namespace caf::net
...@@ -74,6 +74,10 @@ public: ...@@ -74,6 +74,10 @@ public:
CAF_FAIL("handle_error called with code " << code); CAF_FAIL("handle_error called with code " << code);
} }
void continue_reading() override {
CAF_FAIL("continue_reading called");
}
void send(string_view x) { void send(string_view x) {
auto x_bytes = as_bytes(make_span(x)); auto x_bytes = as_bytes(make_span(x));
wr_buf_.insert(wr_buf_.end(), x_bytes.begin(), x_bytes.end()); wr_buf_.insert(wr_buf_.end(), x_bytes.begin(), x_bytes.end());
......
...@@ -55,6 +55,10 @@ public: ...@@ -55,6 +55,10 @@ public:
abort_reason_ = std::move(reason); abort_reason_ = std::move(reason);
} }
bool stopped() const noexcept {
return max_read_size == 0;
}
void configure_read(caf::net::receive_policy policy) { void configure_read(caf::net::receive_policy policy) {
min_read_size = policy.min_size; min_read_size = policy.min_size;
max_read_size = policy.max_size; max_read_size = policy.max_size;
......
...@@ -59,6 +59,11 @@ struct app_t { ...@@ -59,6 +59,11 @@ struct app_t {
return true; return true;
} }
template <class LowerLayerPtr>
void continue_reading(LowerLayerPtr) {
CAF_FAIL("continue_reading called");
}
template <class LowerLayerPtr> template <class LowerLayerPtr>
bool done_sending(LowerLayerPtr) { bool done_sending(LowerLayerPtr) {
return self->try_block_mailbox(); return self->try_block_mailbox();
......
...@@ -2,13 +2,13 @@ ...@@ -2,13 +2,13 @@
// the main distribution directory for license terms and copyright or visit // the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE. // https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE net.observer_adapter #define CAF_SUITE net.consumer_adapter
#include "caf/net/observer_adapter.hpp" #include "caf/net/consumer_adapter.hpp"
#include "net-test.hpp" #include "net-test.hpp"
#include "caf/async/publisher.hpp" #include "caf/async/bounded_buffer.hpp"
#include "caf/net/middleman.hpp" #include "caf/net/middleman.hpp"
#include "caf/net/socket_guard.hpp" #include "caf/net/socket_guard.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
...@@ -40,8 +40,12 @@ public: ...@@ -40,8 +40,12 @@ public:
} }
} }
size_t remaining() const noexcept {
return buf_.size() - rd_pos_;
}
bool done() const noexcept { bool done() const noexcept {
return rd_pos_ == buf_.size(); return remaining() == 0;
} }
auto& buf() const { auto& buf() const {
...@@ -58,40 +62,55 @@ class app_t { ...@@ -58,40 +62,55 @@ class app_t {
public: public:
using input_tag = tag::stream_oriented; using input_tag = tag::stream_oriented;
explicit app_t(async::publisher<int32_t> input) : input_(std::move(input)) { using resource_type = async::consumer_resource<int32_t>;
using buffer_type = resource_type::buffer_type;
using adapter_ptr = net::consumer_adapter_ptr<buffer_type>;
using adapter_type = adapter_ptr::element_type;
explicit app_t(resource_type input) : input_(std::move(input)) {
// nop // nop
} }
template <class LowerLayerPtr> template <class LowerLayerPtr>
error init(net::socket_manager* owner, LowerLayerPtr, const settings&) { error init(net::socket_manager* mgr, LowerLayerPtr, const settings&) {
adapter_ = make_counted<net::observer_adapter<int32_t>>(owner); if (auto ptr = adapter_type::try_open(mgr, std::move(input_))) {
input_.subscribe(adapter_->as_observer()); adapter_ = std::move(ptr);
input_ = nullptr;
return none; return none;
} else {
FAIL("unable to open the resource");
}
} }
template <class LowerLayerPtr> template <class LowerLayerPtr>
bool prepare_send(LowerLayerPtr down) { bool prepare_send(LowerLayerPtr down) {
while (!done_ && down->can_send_more()) { bool run = !done_;
auto [val, done, err] = adapter_->poll(); while (run && down->can_send_more()) {
if (val) { bool on_next_called = false;
written_values_.emplace_back(*val); auto fin = adapter_->consume(
async::ignore_errors, 1,
[this, down, &on_next_called](span<const int32_t> items) {
REQUIRE_EQ(items.size(), 1u);
auto val = items[0];
written_values_.emplace_back(val);
auto offset = written_bytes_.size(); auto offset = written_bytes_.size();
binary_serializer sink{nullptr, written_bytes_}; binary_serializer sink{nullptr, written_bytes_};
if (!sink.apply(*val)) if (!sink.apply(val))
FAIL("sink.apply failed: " << sink.get_error()); FAIL("sink.apply failed: " << sink.get_error());
auto bytes = make_span(written_bytes_).subspan(offset); auto bytes = make_span(written_bytes_).subspan(offset);
down->begin_output(); down->begin_output();
auto& buf = down->output_buffer(); auto& buf = down->output_buffer();
buf.insert(buf.end(), bytes.begin(), bytes.end()); buf.insert(buf.end(), bytes.begin(), bytes.end());
down->end_output(); down->end_output();
} else if (done) { on_next_called = true;
});
if (fin) {
MESSAGE("adapter signaled end-of-buffer");
done_ = true; done_ = true;
if (err)
FAIL("flow error: " << *err);
} else {
break;
} }
run = !done_ && on_next_called;
} }
MESSAGE(written_bytes_.size() << " bytes written"); MESSAGE(written_bytes_.size() << " bytes written");
return true; return true;
...@@ -99,7 +118,12 @@ public: ...@@ -99,7 +118,12 @@ public:
template <class LowerLayerPtr> template <class LowerLayerPtr>
bool done_sending(LowerLayerPtr) { bool done_sending(LowerLayerPtr) {
return !adapter_->has_data(); return done_ || !adapter_->has_data();
}
template <class LowerLayerPtr>
void continue_reading(LowerLayerPtr) {
CAF_FAIL("continue_reading called");
} }
template <class LowerLayerPtr> template <class LowerLayerPtr>
...@@ -124,8 +148,8 @@ private: ...@@ -124,8 +148,8 @@ private:
bool done_ = false; bool done_ = false;
std::vector<int32_t> written_values_; std::vector<int32_t> written_values_;
std::vector<byte> written_bytes_; std::vector<byte> written_bytes_;
net::observer_adapter_ptr<int32_t> adapter_; adapter_ptr adapter_;
async::publisher<int32_t> input_; resource_type input_;
}; };
struct fixture : test_coordinator_fixture<>, host_fixture { struct fixture : test_coordinator_fixture<>, host_fixture {
...@@ -147,25 +171,31 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -147,25 +171,31 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
BEGIN_FIXTURE_SCOPE(fixture) BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("subscriber adapters wake up idle socket managers") { SCENARIO("subscriber adapters wake up idle socket managers") {
GIVEN("a publisher<T>") { GIVEN("an actor pushing into a buffer resource") {
static constexpr size_t num_items = 4211; static constexpr size_t num_items = 79;
auto src = async::publisher_from<event_based_actor>(sys, [](auto* self) { auto [rd, wr] = async::make_bounded_buffer_resource<int32_t>(8, 2);
return self->make_observable().repeat(42).take(num_items); sys.spawn([wr{wr}](event_based_actor* self) {
self->make_observable().repeat(42).take(num_items).subscribe(wr);
}); });
WHEN("sending items of the stream over a socket") { WHEN("draining the buffer resource and sending its items over a socket") {
auto [fd1, fd2] = unbox(net::make_stream_socket_pair()); auto [fd1, fd2] = unbox(net::make_stream_socket_pair());
if (auto err = nonblocking(fd1, true)) if (auto err = nonblocking(fd1, true))
FAIL("nonblocking(fd1) returned an error: " << err); FAIL("nonblocking(fd1) returned an error: " << err);
if (auto err = nonblocking(fd2, true)) if (auto err = nonblocking(fd2, true))
FAIL("nonblocking(fd2) returned an error: " << err); FAIL("nonblocking(fd2) returned an error: " << err);
auto mgr = net::make_socket_manager<app_t, net::stream_transport>( auto mgr = net::make_socket_manager<app_t, net::stream_transport>(
fd1, mm.mpx_ptr(), src); fd1, mm.mpx_ptr(), std::move(rd));
auto& app = mgr->top_layer(); auto& app = mgr->top_layer();
if (auto err = mgr->init(content(cfg))) if (auto err = mgr->init(content(cfg)))
FAIL("mgr->init() failed: " << err); FAIL("mgr->init() failed: " << err);
THEN("the reader receives all items before the connection closes") { THEN("the reader receives all items before the connection closes") {
reader rd{fd2, num_items * sizeof(int32_t)}; auto remaining = num_items * sizeof(int32_t);
reader rd{fd2, remaining};
while (!rd.done()) { while (!rd.done()) {
if (auto new_val = rd.remaining(); remaining != new_val) {
remaining = new_val;
MESSAGE("want " << remaining << " more bytes");
}
run(); run();
rd.read_some(); rd.read_some();
} }
......
...@@ -2,13 +2,12 @@ ...@@ -2,13 +2,12 @@
// the main distribution directory for license terms and copyright or visit // the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE. // https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE net.publisher_adapter #define CAF_SUITE net.producer_adapter
#include "caf/net/publisher_adapter.hpp" #include "caf/net/producer_adapter.hpp"
#include "net-test.hpp" #include "net-test.hpp"
#include "caf/async/publisher.hpp"
#include "caf/detail/network_order.hpp" #include "caf/detail/network_order.hpp"
#include "caf/net/length_prefix_framing.hpp" #include "caf/net/length_prefix_framing.hpp"
#include "caf/net/middleman.hpp" #include "caf/net/middleman.hpp"
...@@ -52,14 +51,30 @@ private: ...@@ -52,14 +51,30 @@ private:
net::socket_guard<net::stream_socket> sg_; net::socket_guard<net::stream_socket> sg_;
}; };
class app { class app_t {
public: public:
using input_tag = tag::message_oriented; using input_tag = tag::message_oriented;
using resource_type = async::producer_resource<int32_t>;
using buffer_type = resource_type::buffer_type;
using adapter_ptr = net::producer_adapter_ptr<buffer_type>;
using adapter_type = adapter_ptr::element_type;
explicit app_t(resource_type output) : output_(std::move(output)) {
// nop
}
template <class LowerLayerPtr> template <class LowerLayerPtr>
error init(net::socket_manager* owner, LowerLayerPtr, const settings&) { error init(net::socket_manager* mgr, LowerLayerPtr, const settings&) {
adapter = make_counted<net::publisher_adapter<int32_t>>(owner, 3, 2); if (auto ptr = adapter_type::try_open(mgr, std::move(output_))) {
adapter_ = std::move(ptr);
return none; return none;
} else {
FAIL("unable to open the resource");
}
} }
template <class LowerLayerPtr> template <class LowerLayerPtr>
...@@ -74,16 +89,16 @@ public: ...@@ -74,16 +89,16 @@ public:
template <class LowerLayerPtr> template <class LowerLayerPtr>
void abort(LowerLayerPtr, const error& reason) { void abort(LowerLayerPtr, const error& reason) {
adapter->flush(); if (reason == caf::sec::socket_disconnected
if (reason == caf::sec::socket_disconnected) || reason == caf::sec::discarded)
adapter->on_complete(); adapter_->close();
else else
adapter->on_error(reason); adapter_->abort(reason);
} }
template <class LowerLayerPtr> template <class LowerLayerPtr>
void after_reading(LowerLayerPtr) { void after_reading(LowerLayerPtr) {
adapter->flush(); // nop
} }
template <class LowerLayerPtr> template <class LowerLayerPtr>
...@@ -93,102 +108,68 @@ public: ...@@ -93,102 +108,68 @@ public:
if (auto err = detail::parse(str, val)) if (auto err = detail::parse(str, val))
FAIL("unable to parse input: " << err); FAIL("unable to parse input: " << err);
++received_messages; ++received_messages;
if (auto n = adapter->push(val); n == 0) if (auto capacity_left = adapter_->push(val); capacity_left == 0)
down->suspend_reading(); down->suspend_reading();
return static_cast<ptrdiff_t>(buf.size()); return static_cast<ptrdiff_t>(buf.size());
} }
size_t received_messages = 0; size_t received_messages = 0;
net::publisher_adapter_ptr<int32_t> adapter; adapter_ptr adapter_;
resource_type output_;
}; };
struct mock_observer : flow::observer<int32_t>::impl { struct fixture : test_coordinator_fixture<>, host_fixture {
void dispose() { fixture() : mm(sys) {
if (sub) { mm.mpx().set_thread_id();
sub.cancel(); if (auto err = mm.mpx().init())
sub = nullptr; CAF_FAIL("mpx.init() failed: " << err);
}
done = true;
}
bool disposed() const noexcept {
return done;
} }
void on_complete() { bool handle_io_event() override {
sub = nullptr; return mm.mpx().poll_once(false);
done = true;
} }
void on_error(const error& what) { net::middleman mm;
FAIL("observer received an error: " << what);
}
void on_attach(flow::subscription new_sub) {
REQUIRE(!sub);
sub = std::move(new_sub);
}
void on_next(span<const int32_t> items) {
buf.insert(buf.end(), items.begin(), items.end());
}
bool done = false;
flow::subscription sub;
std::vector<int32_t> buf;
};
struct fixture {
}; };
} // namespace } // namespace
CAF_TEST_FIXTURE_SCOPE(publisher_adapter_tests, fixture) BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("publisher adapters suspend reads if the buffer becomes full") { SCENARIO("publisher adapters suspend reads if the buffer becomes full") {
auto ls = [](auto... xs) { return std::vector<int32_t>{xs...}; }; GIVEN("an actor reading from a buffer resource") {
GIVEN("a writer and a message-based application") { static constexpr size_t num_items = 13;
std::vector<int32_t> outputs;
auto [rd, wr] = async::make_bounded_buffer_resource<int32_t>(8, 2);
sys.spawn([rd{rd}, &outputs](event_based_actor* self) {
self //
->make_observable()
.from_resource(rd)
.for_each([&outputs](int32_t x) { outputs.emplace_back(x); });
});
WHEN("a producer reads from a socket and publishes to the buffer"){
auto [fd1, fd2] = unbox(net::make_stream_socket_pair()); auto [fd1, fd2] = unbox(net::make_stream_socket_pair());
auto writer_thread = std::thread{[fd1{fd1}] { auto writer_thread = std::thread{[fd1{fd1}] {
writer out{fd1}; writer out{fd1};
for (int i = 0; i < 12; ++i) for (size_t i = 0; i < num_items; ++i)
out.write(std::to_string(i)); out.write(std::to_string(i));
}}; }};
net::multiplexer mpx{nullptr}; if (auto err = nonblocking(fd2, true))
if (auto err = mpx.init()) FAIL("nonblocking(fd2) returned an error: " << err);
FAIL("mpx.init failed: " << err); auto mgr = net::make_socket_manager<app_t, net::length_prefix_framing,
mpx.set_thread_id(); net::stream_transport>(
REQUIRE_EQ(mpx.num_socket_managers(), 1u); fd2, mm.mpx_ptr(), std::move(wr));
if (auto err = net::nonblocking(fd2, true)) if (auto err = mgr->init(content(cfg)))
CAF_FAIL("nonblocking returned an error: " << err); FAIL("mgr->init() failed: " << err);
auto mgr = net::make_socket_manager<app, net::length_prefix_framing, THEN("the actor receives all items from the writer (socket)") {
net::stream_transport>(fd2, &mpx); while (outputs.size() < num_items)
auto& st = mgr->top_layer(); run();
CHECK_EQ(mgr->init(settings{}), none); auto ls = [](auto... xs) { return std::vector<int32_t>{xs...}; };
REQUIRE_EQ(mpx.num_socket_managers(), 2u); CHECK_EQ(outputs, ls(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12));
CHECK_EQ(mgr->mask(), net::operation::read);
WHEN("the publisher adapter runs out of capacity") {
while (mpx.num_socket_managers() > 1u)
mpx.poll_once(true);
CHECK_EQ(mgr->mask(), net::operation::none);
CHECK_EQ(st.received_messages, 3u);
THEN("reading from the adapter registers the manager for reading again") {
auto obs = make_counted<mock_observer>();
st.adapter->subscribe(flow::observer<int32_t>{obs});
REQUIRE(obs->sub.valid());
obs->sub.request(1);
while (st.received_messages != 4u)
mpx.poll_once(true);
CHECK_EQ(obs->buf, ls(0));
obs->sub.request(20);
while (st.received_messages != 12u)
mpx.poll_once(true);
CHECK_EQ(obs->buf, ls(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11));
}
} }
writer_thread.join(); writer_thread.join();
} }
}
} }
CAF_TEST_FIXTURE_SCOPE_END() END_FIXTURE_SCOPE()
...@@ -67,6 +67,11 @@ struct app_t { ...@@ -67,6 +67,11 @@ struct app_t {
return self->try_block_mailbox(); return self->try_block_mailbox();
} }
template <class LowerLayerPtr>
void continue_reading(LowerLayerPtr) {
CAF_FAIL("continue_reading called");
}
template <class LowerLayerPtr> template <class LowerLayerPtr>
void abort(LowerLayerPtr, const error& reason) { void abort(LowerLayerPtr, const error& reason) {
CAF_FAIL("app::abort called: " << reason); CAF_FAIL("app::abort called: " << reason);
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE net.web_socket.flow
#include "caf/net/web_socket/flow.hpp"
#include "caf/test/dsl.hpp"
using namespace caf;
namespace {
struct fixture {
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(flow_tests, fixture)
CAF_TEST(todo) {
// implement me
}
CAF_TEST_FIXTURE_SCOPE_END()
...@@ -95,6 +95,11 @@ public: ...@@ -95,6 +95,11 @@ public:
return true; return true;
} }
template <class ParentPtr>
void continue_reading(ParentPtr) {
CAF_FAIL("continue_reading called");
}
template <class ParentPtr> template <class ParentPtr>
size_t consume(ParentPtr, span<const byte> data, span<const byte>) { size_t consume(ParentPtr, span<const byte> data, span<const byte>) {
recv_buf_->clear(); recv_buf_->clear();
......
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