Commit 25d12e95 authored by Dominik Charousset's avatar Dominik Charousset

Use the new async loop abstraction in caf_net

parent 1dddf13d
......@@ -224,7 +224,10 @@ caf_add_component(
actor_system_config
actor_termination
aout
async.blocking_consumer
async.blocking_producer
async.consumer_adapter
async.producer_adapter
async.promise
async.spsc_buffer
behavior
......
......@@ -33,20 +33,24 @@ public:
template <class ErrorPolicy, class TimePoint>
read_result pull(ErrorPolicy policy, T& item, TimePoint timeout) {
if (!buf_) {
return abort_reason_ ? read_result::abort : read_result::stop;
}
val_ = &item;
std::unique_lock guard{buf_->mtx()};
if constexpr (std::is_same_v<TimePoint, none_t>) {
buf_->await_consumer_ready(guard, cv_);
} else {
if (!buf_->await_consumer_ready(guard, cv_, timeout))
return read_result::timeout;
return read_result::try_again_later;
}
auto [again, n] = buf_->pull_unsafe(guard, policy, 1u, *this);
CAF_IGNORE_UNUSED(again);
CAF_ASSERT(!again || n == 1);
if (!again) {
buf_ = nullptr;
}
if (n == 1) {
return read_result::ok;
} else if (buf_->abort_reason_unsafe()) {
} else if (abort_reason_) {
return read_result::abort;
} else {
return read_result::stop;
......@@ -58,19 +62,18 @@ public:
return pull(policy, item, none);
}
void on_next(span<const T> items) {
CAF_ASSERT(items.size() == 1);
*val_ = items[0];
void on_next(const T& item) {
*val_ = item;
}
void on_complete() {
}
void on_error(const caf::error&) {
// nop
void on_error(const caf::error& abort_reason) {
abort_reason_ = abort_reason;
}
void dispose() {
void cancel() {
if (buf_) {
buf_->cancel();
buf_ = nullptr;
......@@ -94,8 +97,8 @@ public:
deref();
}
error abort_reason() const {
buf_->abort_reason()();
const error& abort_reason() const {
return abort_reason_;
}
CAF_INTRUSIVE_PTR_FRIENDS(impl)
......@@ -104,6 +107,7 @@ public:
spsc_buffer_ptr<T> buf_;
std::condition_variable cv_;
T* val_ = nullptr;
error abort_reason_;
};
using impl_ptr = intrusive_ptr<impl>;
......@@ -122,6 +126,10 @@ public:
// nop
}
explicit blocking_consumer(spsc_buffer_ptr<T> buf) {
impl_.emplace(std::move(buf));
}
~blocking_consumer() {
if (impl_)
impl_->cancel();
......
......@@ -4,14 +4,15 @@
#pragma once
#include <condition_variable>
#include <type_traits>
#include "caf/async/producer.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/detail/atomic_ref_counted.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/make_counted.hpp"
#include "caf/ref_counted.hpp"
#include <condition_variable>
#include <optional>
#include <type_traits>
namespace caf::async {
......@@ -19,7 +20,7 @@ namespace caf::async {
template <class T>
class blocking_producer {
public:
class impl : public ref_counted, public producer {
class impl : public detail::atomic_ref_counted, public producer {
public:
impl() = delete;
impl(const impl&) = delete;
......@@ -99,8 +100,6 @@ public:
deref();
}
CAF_INTRUSIVE_PTR_FRIENDS(impl)
private:
spsc_buffer_ptr<T> buf_;
mutable std::mutex mtx_;
......@@ -169,18 +168,30 @@ public:
return impl_->canceled();
}
explicit operator bool() const noexcept {
return static_cast<bool>(impl_);
}
private:
intrusive_ptr<impl> impl_;
};
/// @pre `buf != nullptr`
/// @relates blocking_producer
template <class T>
blocking_producer<T> make_blocking_producer(spsc_buffer_ptr<T> buf) {
using impl_t = typename blocking_producer<T>::impl;
return blocking_producer<T>{make_counted<impl_t>(std::move(buf))};
}
/// @relates blocking_producer
template <class T>
expected<blocking_producer<T>>
std::optional<blocking_producer<T>>
make_blocking_producer(producer_resource<T> res) {
if (auto buf = res.try_open()) {
using impl_t = typename blocking_producer<T>::impl;
return {blocking_producer<T>{make_counted<impl_t>(std::move(buf))}};
return {make_blocking_producer(std::move(buf))};
} else {
return {make_error(sec::cannot_open_resource)};
return {};
}
}
......
// 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/async/execution_context.hpp"
#include "caf/async/read_result.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/detail/atomic_ref_counted.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/make_counted.hpp"
#include <optional>
namespace caf::async {
/// Integrates an SPSC buffer consumer into an asynchronous event loop.
template <class T>
class consumer_adapter {
public:
class impl : public detail::atomic_ref_counted, public consumer {
public:
impl() = delete;
impl(const impl&) = delete;
impl& operator=(const impl&) = delete;
impl(spsc_buffer_ptr<T> buf, execution_context_ptr ctx, action do_wakeup)
: buf_(std::move(buf)),
ctx_(std::move(ctx)),
do_wakeup_(std::move(do_wakeup)) {
buf_->set_consumer(this);
}
~impl() {
if (buf_) {
buf_->cancel();
do_wakeup_.dispose();
}
}
void cancel() {
if (buf_) {
buf_->cancel();
buf_ = nullptr;
do_wakeup_.dispose();
do_wakeup_ = nullptr;
}
}
template <class ErrorPolicy>
read_result pull(ErrorPolicy policy, T& item) {
if (buf_) {
val_ = &item;
auto [again, n] = buf_->pull(policy, 1u, *this);
if (!again) {
buf_ = nullptr;
}
if (n == 1) {
return read_result::ok;
} else if (again) {
CAF_ASSERT(n == 0);
return read_result::try_again_later;
} else {
CAF_ASSERT(n == 0);
return abort_reason_ ? read_result::abort : read_result::stop;
}
} else {
return abort_reason_ ? read_result::abort : read_result::stop;
}
}
const error& abort_reason() const noexcept {
return abort_reason_;
}
bool has_data() const noexcept {
return buf_ ? buf_->has_data() : false;
}
bool has_consumer_event() const noexcept {
return buf_ ? buf_->has_consumer_event() : false;
}
void on_next(const T& item) {
*val_ = item;
}
void on_complete() {
// nop
}
void on_error(const caf::error& what) {
abort_reason_ = what;
}
void on_producer_ready() override {
// nop
}
void on_producer_wakeup() override {
ctx_->schedule(do_wakeup_);
}
void ref_consumer() const noexcept override {
ref();
}
void deref_consumer() const noexcept override {
deref();
}
private:
spsc_buffer_ptr<T> buf_;
execution_context_ptr ctx_;
action do_wakeup_;
error abort_reason_;
T* val_ = nullptr;
};
using impl_ptr = intrusive_ptr<impl>;
consumer_adapter() = default;
consumer_adapter(const consumer_adapter&) = delete;
consumer_adapter& operator=(const consumer_adapter&) = delete;
consumer_adapter(consumer_adapter&&) = default;
consumer_adapter& operator=(consumer_adapter&&) = default;
explicit consumer_adapter(impl_ptr ptr) : impl_(std::move(ptr)) {
// nop
}
~consumer_adapter() {
if (impl_)
impl_->cancel();
}
template <class Policy>
read_result pull(Policy policy, T& result) {
if (impl_)
return impl_->pull(policy, result);
else
return read_result::abort;
}
void cancel() {
if (impl_) {
impl_->cancel();
impl_ = nullptr;
}
}
error abort_reason() const noexcept {
if (impl_)
return impl_->abort_reason();
else
return make_error(sec::disposed);
}
explicit operator bool() const noexcept {
return static_cast<bool>(impl_);
}
bool has_data() const noexcept {
return impl_ ? impl_->has_data() : false;
}
bool has_consumer_event() const noexcept {
return impl_ ? impl_->has_consumer_event() : false;
}
static consumer_adapter make(spsc_buffer_ptr<T> buf,
execution_context_ptr ctx, action do_wakeup) {
if (buf) {
using impl_t = typename consumer_adapter<T>::impl;
auto impl = make_counted<impl_t>(std::move(buf), std::move(ctx),
std::move(do_wakeup));
return consumer_adapter<T>{std::move(impl)};
} else {
return {};
}
}
static std::optional<consumer_adapter>
make(consumer_resource<T> res, execution_context_ptr ctx, action do_wakeup) {
if (auto buf = res.try_open()) {
return {make(std::move(buf), std::move(ctx), std::move(do_wakeup))};
} else {
return {};
}
}
private:
intrusive_ptr<impl> impl_;
};
/// @pre `buf != nullptr`
/// @relates consumer_adapter
template <class T>
consumer_adapter<T> make_consumer_adapter(spsc_buffer_ptr<T> buf,
execution_context_ptr ctx,
action do_wakeup) {
return consumer_adapter<T>::make(std::move(buf), std::move(ctx),
std::move(do_wakeup));
}
/// @relates consumer_adapter
template <class T>
std::optional<consumer_adapter<T>>
make_consumer_adapter(consumer_resource<T> res, execution_context_ptr ctx,
action do_wakeup) {
if (auto buf = res.try_open()) {
return {consumer_adapter<T>::make(std::move(buf), std::move(ctx),
std::move(do_wakeup))};
} else {
return {};
}
}
} // namespace caf::async
......@@ -109,12 +109,24 @@ public:
return {ctx, std::move(cell_)};
}
/// Binds this future to an @ref execution_context to run callbacks.
/// @pre `valid()`
bound_future<T> bind_to(execution_context& ctx) && {
return {&ctx, std::move(cell_)};
}
/// Binds this future to an @ref execution_context to run callbacks.
/// @pre `valid()`
bound_future<T> bind_to(execution_context* ctx) const& {
return {ctx, cell_};
}
/// Binds this future to an @ref execution_context to run callbacks.
/// @pre `valid()`
bound_future<T> bind_to(execution_context& ctx) const& {
return {&ctx, cell_};
}
/// Queries whether the result of the asynchronous computation is still
/// pending, i.e., neither `set_value` nor `set_error` has been called on the
/// @ref promise.
......
......@@ -4,7 +4,7 @@
#pragma once
#include <caf/fwd.hpp>
#include "caf/fwd.hpp"
namespace caf::async {
......@@ -12,6 +12,7 @@ namespace caf::async {
class batch;
class consumer;
class execution_context;
class producer;
// -- template classes ---------------------------------------------------------
......@@ -31,6 +32,10 @@ class promise;
template <class T>
class future;
// -- smart pointer aliases ----------------------------------------------------
using execution_context_ptr = intrusive_ptr<execution_context>;
// -- free function templates --------------------------------------------------
template <class T>
......
// 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/execution_context.hpp"
#include "caf/async/producer.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/detail/atomic_ref_counted.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/make_counted.hpp"
#include <optional>
namespace caf::async {
/// Integrates an SPSC buffer producer into an asynchronous event loop.
template <class T>
class producer_adapter {
public:
class impl : public detail::atomic_ref_counted, public producer {
public:
impl() = delete;
impl(const impl&) = delete;
impl& operator=(const impl&) = delete;
impl(spsc_buffer_ptr<T> buf, execution_context_ptr ctx, action do_resume,
action do_cancel)
: buf_(std::move(buf)),
ctx_(std::move(ctx)),
do_resume_(std::move(do_resume)),
do_cancel_(std::move(do_cancel)) {
buf_->set_producer(this);
}
size_t push(span<const T> items) {
return buf_->push(items);
}
bool push(const T& item) {
return buf_->push(item);
}
void close() {
if (buf_) {
buf_->close();
buf_ = nullptr;
do_resume_.dispose();
do_resume_ = nullptr;
do_cancel_.dispose();
do_cancel_ = nullptr;
}
}
void abort(error reason) {
if (buf_) {
buf_->abort(std::move(reason));
buf_ = nullptr;
do_resume_.dispose();
do_resume_ = nullptr;
do_cancel_.dispose();
do_cancel_ = nullptr;
}
}
void on_consumer_ready() override {
// nop
}
void on_consumer_cancel() override {
ctx_->schedule(do_cancel_);
}
void on_consumer_demand(size_t) override {
ctx_->schedule(do_resume_);
}
void ref_producer() const noexcept override {
ref();
}
void deref_producer() const noexcept override {
deref();
}
private:
spsc_buffer_ptr<T> buf_;
execution_context_ptr ctx_;
action do_resume_;
action do_cancel_;
};
using impl_ptr = intrusive_ptr<impl>;
producer_adapter() = default;
producer_adapter(const producer_adapter&) = delete;
producer_adapter& operator=(const producer_adapter&) = delete;
producer_adapter(producer_adapter&&) = default;
producer_adapter& operator=(producer_adapter&&) = default;
explicit producer_adapter(impl_ptr ptr) : impl_(std::move(ptr)) {
// nop
}
~producer_adapter() {
if (impl_)
impl_->close();
}
/// Makes `item` available to the consumer.
/// @returns the remaining demand.
size_t push(const T& item) {
if (!impl_)
CAF_RAISE_ERROR("cannot push to a closed producer adapter");
return impl_->push(item);
}
/// Makes `items` available to the consumer.
/// @returns the remaining demand.
size_t push(span<const T> items) {
if (!impl_)
CAF_RAISE_ERROR("cannot push to a closed producer adapter");
return impl_->push(items);
}
void close() {
if (impl_) {
impl_->close();
impl_ = nullptr;
}
}
void abort(error reason) {
if (impl_) {
impl_->abort(std::move(reason));
impl_ = nullptr;
}
}
explicit operator bool() const noexcept {
return static_cast<bool>(impl_);
}
/// @pre `buf != nullptr`
static producer_adapter make(spsc_buffer_ptr<T> buf,
execution_context_ptr ctx, action do_resume,
action do_cancel) {
if (buf) {
using impl_t = typename producer_adapter<T>::impl;
auto impl = make_counted<impl_t>(std::move(buf), std::move(ctx),
std::move(do_resume),
std::move(do_cancel));
return producer_adapter<T>{std::move(impl)};
} else {
return {};
}
}
static std::optional<producer_adapter>
make(producer_resource<T> res, execution_context_ptr ctx, action do_resume,
action do_cancel) {
if (auto buf = res.try_open()) {
return {make(std::move(buf), std::move(ctx), std::move(do_resume),
std::move(do_cancel))};
} else {
return {};
}
}
private:
intrusive_ptr<impl> impl_;
};
/// @pre `buf != nullptr`
/// @relates producer_adapter
template <class T>
producer_adapter<T> make_producer_adapter(spsc_buffer_ptr<T> buf,
execution_context_ptr ctx,
action do_resume, action do_cancel) {
return producer_adapter<T>::make(std::move(buf), std::move(ctx),
std::move(do_resume), std::move(do_cancel));
}
/// @relates producer_adapter
template <class T>
std::optional<producer_adapter<T>>
make_producer_adapter(producer_resource<T> res, execution_context_ptr ctx,
action do_resume, action do_cancel) {
if (auto buf = res.try_open()) {
return {producer_adapter<T>::make(std::move(buf), std::move(ctx),
std::move(do_resume),
std::move(do_cancel))};
} else {
return {};
}
}
} // namespace caf::async
......@@ -20,8 +20,9 @@ enum class read_result {
stop,
/// Signals that the source failed with an error.
abort,
/// Signals that the read operation timed out.
timeout,
/// Signals that the read operation cannot produce a result at the moment,
/// e.g., because of a timeout.
try_again_later,
};
/// @relates read_result
......
// 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 async.blocking_consumer
#include "caf/async/blocking_consumer.hpp"
#include "core-test.hpp"
#include "caf/async/blocking_producer.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <mutex>
using namespace caf;
namespace {
struct fixture {
actor_system_config cfg;
actor_system sys;
fixture()
: sys(cfg.set("caf.scheduler.max-threads", 2)
.set("caf.scheduler.policy", "sharing")) {
// nop
}
};
void produce(async::producer_resource<int> push) {
async::blocking_producer out{push.try_open()};
for (int i = 0; i < 5000; ++i)
out.push(i);
}
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("blocking consumers allow threads to receive data") {
GIVEN("a producers running in a separate thread") {
WHEN("consuming the generated value with a blocking consumer") {
THEN("the consumer receives all values in order") {
auto [pull, push] = async::make_spsc_buffer_resource<int>();
std::thread producer{produce, push};
async::blocking_consumer<int> in{pull.try_open()};
std::vector<int> got;
bool done = false;
while (!done) {
int tmp = 0;
switch (in.pull(async::delay_errors, tmp)) {
case async::read_result::ok:
got.push_back(tmp);
break;
case async::read_result::stop:
done = true;
break;
case async::read_result::abort:
CAF_FAIL("did not expect async::read_result::abort");
done = true;
break;
default:
CAF_FAIL("unexpected pull result");
done = true;
}
}
auto want = std::vector<int>(5000);
std::iota(want.begin(), want.end(), 0);
CHECK_EQ(got.size(), 5000u);
CHECK_EQ(got, want);
producer.join();
}
}
}
}
END_FIXTURE_SCOPE()
// 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 async.consumer_adapter
#include "caf/async/consumer_adapter.hpp"
#include "core-test.hpp"
#include "caf/async/blocking_producer.hpp"
#include "caf/flow/scoped_coordinator.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <mutex>
using namespace caf;
namespace {
struct fixture {
actor_system_config cfg;
actor_system sys;
fixture()
: sys(cfg.set("caf.scheduler.max-threads", 2)
.set("caf.scheduler.policy", "sharing")) {
// nop
}
};
void produce(async::producer_resource<int> push) {
async::blocking_producer out{push.try_open()};
for (int i = 0; i < 5000; ++i)
out.push(i);
}
class runner_t {
public:
runner_t(async::spsc_buffer_ptr<int> buf, async::execution_context_ptr ctx) {
do_wakeup_ = make_action([this] { resume(); });
in_ = make_consumer_adapter(buf, ctx, do_wakeup_);
}
disposable as_disposable() {
return do_wakeup_.as_disposable();
}
void resume() {
int tmp = 0;
for (;;) {
auto res = in_.pull(async::delay_errors, tmp);
switch (res) {
case async::read_result::ok:
values_.push_back(tmp);
break;
case async::read_result::stop:
do_wakeup_.dispose();
return;
case async::read_result::try_again_later:
return;
default:
CAF_FAIL("unexpected pull result: " << res);
do_wakeup_.dispose();
return;
}
}
}
void start() {
resume();
}
const std::vector<int>& values() {
return values_;
}
private:
async::consumer_adapter<int> in_;
action do_wakeup_;
std::vector<int> values_;
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("consumer adapters allow integrating consumers into event loops") {
GIVEN("a producers running in a separate thread") {
WHEN("consuming the generated value with a blocking consumer") {
THEN("the consumer receives all values in order") {
auto [pull, push] = async::make_spsc_buffer_resource<int>();
std::thread producer{produce, push};
auto loop = flow::make_scoped_coordinator();
runner_t runner{pull.try_open(), loop};
loop->watch(runner.as_disposable());
runner.start();
loop->run();
auto& got = runner.values();
auto want = std::vector<int>(5000);
std::iota(want.begin(), want.end(), 0);
CHECK_EQ(got.size(), 5000u);
CHECK_EQ(got, want);
producer.join();
}
}
}
}
END_FIXTURE_SCOPE()
// 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 async.producer_adapter
#include "caf/async/producer_adapter.hpp"
#include "core-test.hpp"
#include "caf/async/blocking_producer.hpp"
#include "caf/flow/scoped_coordinator.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <mutex>
using namespace caf;
namespace {
struct fixture {
actor_system_config cfg;
actor_system sys;
fixture() : sys(cfg.set("caf.scheduler.max-threads", 2)) {
// nop
}
};
struct sync_t {
std::mutex mtx;
std::condition_variable cv;
int available = 0;
void release() {
std::unique_lock guard{mtx};
if (++available == 1)
cv.notify_all();
}
void acquire() {
std::unique_lock guard{mtx};
while (available == 0)
cv.wait(guard);
--available;
}
};
using push_val_t = async::producer_resource<int>;
using pull_val_t = async::consumer_resource<int>;
using push_resource_t = async::producer_resource<pull_val_t>;
using pull_resource_t = async::consumer_resource<pull_val_t>;
class runner_t {
public:
runner_t(async::spsc_buffer_ptr<int> buf, async::execution_context_ptr ctx,
int first, int last)
: n(first), end(last) {
do_resume_ = make_action([this] { resume(); });
do_cancel_ = make_action([this] { do_resume_.dispose(); });
out = make_producer_adapter(buf, ctx, do_resume_, do_cancel_);
}
disposable as_disposable() {
return do_resume_.as_disposable();
}
void resume() {
while (n < end)
if (out.push(n++) == 0)
return;
do_resume_.dispose();
}
private:
int n;
int end;
async::producer_adapter<int> out;
action do_resume_;
action do_cancel_;
};
void do_push(sync_t* sync, push_val_t push, int begin, int end) {
auto buf = push.try_open();
if (!buf) {
CAF_RAISE_ERROR("push.try_open failed");
}
auto loop = flow::make_scoped_coordinator();
runner_t runner{buf, loop, begin, end};
loop->watch(runner.as_disposable());
loop->run();
sync->release();
}
std::pair<std::thread, pull_val_t> start_worker(sync_t* sync, int begin,
int end) {
auto [pull, push] = async::make_spsc_buffer_resource<int>();
return {std::thread{do_push, sync, std::move(push), begin, end},
std::move(pull)};
}
void run(push_resource_t push) {
auto buf = push.try_open();
if (!buf) {
CAF_RAISE_ERROR("push.try_open failed");
}
// Note: using the adapter here as well would be tricky since we would need to
// wait for available threads.
async::blocking_producer out{std::move(buf)};
sync_t sync;
std::vector<std::thread> threads;
auto add_worker = [&](int begin, int end) {
auto [hdl, pull] = start_worker(&sync, begin, end);
threads.push_back(std::move(hdl));
out.push(std::move(pull));
};
std::vector<std::pair<int, int>> ranges{{0, 1337}, {1337, 1338},
{1338, 1338}, {1338, 2777},
{2777, 3000}, {3000, 3003},
{3003, 3500}, {3500, 4000}};
add_worker(4000, 4007);
add_worker(4007, 4333);
add_worker(4333, 4500);
add_worker(4500, 5000);
for (auto [begin, end] : ranges) {
sync.acquire();
add_worker(begin, end);
}
for (auto& hdl : threads)
hdl.join();
}
void receiver_impl(event_based_actor* self, pull_resource_t inputs,
actor parent) {
inputs.observe_on(self)
.flat_map([self](const pull_val_t& in) { return in.observe_on(self); })
.to_vector()
.for_each([self, parent](const cow_vector<int>& values) { //
self->send(parent, values);
});
}
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("producers adapters allow integrating producers into event loops") {
GIVEN("a dynamic set of blocking producers") {
WHEN("consuming the generated values from an actor via flat_map") {
THEN("the actor merges all values from all buffers into one") {
auto [pull, push] = async::make_spsc_buffer_resource<pull_val_t>();
auto loop = flow::make_scoped_coordinator();
scoped_actor self{sys};
auto receiver = self->spawn(receiver_impl, std::move(pull),
actor{self});
std::thread runner{run, std::move(push)};
self->receive([](const cow_vector<int>& values) {
auto want = std::vector<int>(5000);
std::iota(want.begin(), want.end(), 0);
auto got = values.std();
std::sort(got.begin(), got.end());
CHECK_EQ(got.size(), 5000u);
CHECK_EQ(got, want);
});
runner.join();
}
}
}
}
END_FIXTURE_SCOPE()
......@@ -33,6 +33,9 @@ caf_add_component(
src/detail/rfc6455.cpp
src/net/abstract_actor_shell.cpp
src/net/actor_shell.cpp
src/net/binary/frame.cpp
src/net/binary/lower_layer.cpp
src/net/binary/upper_layer.cpp
src/net/datagram_socket.cpp
src/net/generic_lower_layer.cpp
src/net/generic_upper_layer.cpp
......@@ -48,7 +51,6 @@ caf_add_component(
src/net/http/v1.cpp
src/net/ip.cpp
src/net/length_prefix_framing.cpp
src/net/message_oriented.cpp
src/net/middleman.cpp
src/net/multiplexer.cpp
src/net/network_socket.cpp
......@@ -88,7 +90,6 @@ caf_add_component(
detail.rfc6455
net.accept_socket
net.actor_shell
net.consumer_adapter
net.datagram_socket
net.http.server
net.ip
......@@ -97,7 +98,6 @@ caf_add_component(
net.network_socket
net.operation
net.pipe_socket
net.producer_adapter
net.prometheus.server
net.socket
net.socket_guard
......
......@@ -5,6 +5,7 @@
#pragma once
#include "caf/actor_traits.hpp"
#include "caf/async/execution_context.hpp"
#include "caf/callback.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/extend.hpp"
......@@ -46,7 +47,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
abstract_actor_shell(actor_config& cfg, socket_manager* owner);
abstract_actor_shell(actor_config& cfg, async::execution_context_ptr loop);
~abstract_actor_shell() override;
......@@ -56,7 +57,7 @@ public:
// -- state modifiers --------------------------------------------------------
/// Detaches the shell from its owner and closes the mailbox.
/// Detaches the shell from its loop and closes the mailbox.
void quit(error reason);
/// Overrides the default handler for unexpected messages.
......@@ -94,11 +95,6 @@ public:
// -- message processing -----------------------------------------------------
/// Dequeues and processes the next message from the mailbox.
/// @returns `true` if a message was dequeued and processed, `false` if the
/// mailbox was empty.
bool consume_message();
/// Adds a callback for a multiplexed response.
void add_multiplexed_response_handler(message_id response_id, behavior bhvr);
......@@ -117,23 +113,35 @@ public:
bool cleanup(error&& fail_state, execution_unit* host) override;
protected:
// Stores incoming actor messages.
void set_behavior_impl(behavior bhvr) {
bhvr_ = std::move(bhvr);
}
private:
/// Stores incoming actor messages.
mailbox_type mailbox_;
// Guards access to owner_.
std::mutex owner_mtx_;
/// Guards access to loop_.
std::mutex loop_mtx_;
// Points to the owning manager (nullptr after quit was called).
socket_manager* owner_;
/// Points to the loop in which this "actor" runs (nullptr after calling
/// quit).
async::execution_context_ptr loop_;
// Handler for consuming messages from the mailbox.
/// Handler for consuming messages from the mailbox.
behavior bhvr_;
// Handler for unexpected messages.
/// Handler for unexpected messages.
fallback_handler fallback_;
// Stores callbacks for multiplexed responses.
/// Stores callbacks for multiplexed responses.
unordered_flat_map<message_id, behavior> multiplexed_responses_;
/// Callback for processing the next message on the event loop.
action resume_;
/// Dequeues and processes the next message from the mailbox.
bool consume_message();
};
} // namespace caf::net
......@@ -38,7 +38,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
actor_shell(actor_config& cfg, socket_manager* owner);
actor_shell(actor_config& cfg, async::execution_context_ptr loop);
~actor_shell() override;
......@@ -47,7 +47,7 @@ public:
/// Overrides the callbacks for incoming messages.
template <class... Fs>
void set_behavior(Fs... fs) {
bhvr_ = behavior{std::move(fs)...};
set_behavior_impl(behavior{std::move(fs)...});
}
// -- overridden functions of local_actor ------------------------------------
......@@ -61,7 +61,9 @@ class CAF_NET_EXPORT actor_shell_ptr {
public:
// -- friends ----------------------------------------------------------------
friend class socket_manager;
template <class Handle>
friend actor_shell_ptr_t<Handle>
make_actor_shell(actor_system&, async::execution_context_ptr);
// -- member types -----------------------------------------------------------
......
// 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_adapter.hpp"
#include "caf/async/producer_adapter.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/fwd.hpp"
#include "caf/net/web_socket/flow_connector.hpp"
#include "caf/net/web_socket/lower_layer.hpp"
#include "caf/net/web_socket/request.hpp"
#include "caf/net/web_socket/upper_layer.hpp"
#include "caf/sec.hpp"
#include <utility>
namespace caf::net::binary {
/// Translates between a message-oriented transport and data flows.
template <class Trait>
class flow_bridge : public web_socket::upper_layer {
public:
using input_type = typename Trait::input_type;
using output_type = typename Trait::output_type;
/// Type for the consumer adapter. We consume the output of the application.
using consumer_type = async::consumer_adapter<output_type>;
/// Type for the producer adapter. We produce the input of the application.
using producer_type = async::producer_adapter<input_type>;
using request_type = request<Trait>;
using connector_pointer = flow_connector_ptr<Trait>;
explicit flow_bridge(connector_pointer conn) : conn_(std::move(conn)) {
// nop
}
static std::unique_ptr<flow_bridge> make(connector_pointer conn) {
return std::make_unique<flow_bridge>(std::move(conn));
}
bool write(const output_type& item) {
down_->begin_message();
auto& bytes = down_->message_buffer();
return trait_.convert(item, bytes) && down_->end_message();
}
bool running() const noexcept {
return in_ || out_;
}
// -- implementation of web_socket::lower_layer ------------------------------
error init(net::socket_manager* mgr, web_socket::lower_layer* down,
const settings& cfg) override {
CAF_ASSERT(mgr != nullptr);
down_ = down;
auto [err, pull, push] = conn_->on_request(cfg);
if (!err) {
auto do_wakeup = make_action([this] {
prepare_send();
if (!running())
down_->shutdown();
});
auto do_resume = make_action([this] { down_->request_messages(); });
auto do_cancel = make_action([this] {
if (!running())
down_->shutdown();
});
in_ = consumer_type::make(pull.try_open(), mgr->mpx_ptr(),
std::move(do_wakeup));
out_ = producer_type::make(push.try_open(), mgr->mpx_ptr(),
std::move(do_resume), std::move(do_cancel));
conn_ = nullptr;
if (running())
return none;
else
return make_error(sec::runtime_error,
"cannot init flow bridge: no buffers");
} else {
conn_ = nullptr;
return err;
}
}
void prepare_send() override {
input_type tmp;
while (down_->can_send_more()) {
switch (in_.pull(async::delay_errors, tmp)) {
case async::read_result::ok:
if (!write(tmp)) {
down_->shutdown(trait_.last_error());
return;
}
break;
case async::read_result::stop:
down_->shutdown();
break;
case async::read_result::abort:
down_->shutdown(in_.abort_reason());
break;
default: // try later
return;
}
}
}
bool done_sending() override {
return !in_.has_consumer_event();
}
void abort(const error& reason) override {
CAF_LOG_TRACE(CAF_ARG(reason));
if (out_) {
if (reason == sec::connection_closed || reason == sec::socket_disconnected
|| reason == sec::disposed)
out_.close();
else
out_.abort(reason);
}
in_.cancel();
}
ptrdiff_t consume_binary(byte_span buf) override {
if (!out_)
return -1;
input_type val;
if (!trait_.convert(buf, val))
return -1;
if (out_.push(std::move(val)) == 0)
down_->suspend_reading();
return static_cast<ptrdiff_t>(buf.size());
}
ptrdiff_t consume_text(std::string_view buf) override {
if (!out_)
return -1;
input_type val;
if (!trait_.convert(buf, val))
return -1;
if (out_.push(std::move(val)) == 0)
down_->suspend_reading();
return static_cast<ptrdiff_t>(buf.size());
}
private:
binary::lower_layer* down_;
/// The output of the application. Serialized to the socket.
consumer_type in_;
/// The input to the application. Deserialized from the socket.
producer_type out_;
/// Converts between raw bytes and native C++ objects.
Trait trait_;
/// Initializes the bridge. Disposed (set to null) after initializing.
connector_pointer conn_;
};
} // namespace caf::net::binary
// 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/fwd.hpp"
#include "caf/byte_span.hpp"
#include "caf/config.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/net/web_socket/frame.hpp"
#include "caf/raise_error.hpp"
#include "caf/type_id.hpp"
namespace caf::net::binary {
/// An implicitly shared type for binary data frames.
class CAF_NET_EXPORT frame {
public:
frame() = default;
frame(frame&&) = default;
frame(const frame&) = default;
frame& operator=(frame&&) = default;
frame& operator=(const frame&) = default;
explicit frame(const_byte_span data);
explicit operator bool() const noexcept {
return static_cast<bool>(data_);
}
size_t size() const noexcept {
return data_ ? data_->size() : 0;
}
bool empty() const noexcept {
return data_ ? data_->size() == 0 : true;
}
void swap(frame& other) {
data_.swap(other.data_);
}
const_byte_span bytes() const noexcept {
return {data_->storage(), data_->size()};
}
private:
intrusive_ptr<web_socket::frame::data> data_;
};
} // namespace caf::net::binary
// 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::net::binary {
class frame;
class lower_layer;
class upper_layer;
} // namespace caf::net::binary
......@@ -7,39 +7,13 @@
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/binary/fwd.hpp"
#include "caf/net/generic_lower_layer.hpp"
#include "caf/net/generic_upper_layer.hpp"
namespace caf::net::message_oriented {
namespace caf::net::binary {
class lower_layer;
/// Consumes binary messages from the lower layer.
class CAF_NET_EXPORT upper_layer : public generic_upper_layer {
public:
virtual ~upper_layer();
/// Initializes the upper layer.
/// @param owner A pointer to the socket manager that owns the entire
/// protocol stack. Remains valid for the lifetime of the upper
/// layer.
/// @param down A pointer to the lower layer that remains valid for the
/// lifetime of the upper layer.
virtual error
init(socket_manager* owner, lower_layer* down, const settings& config)
= 0;
/// Consumes bytes from the lower layer.
/// @param payload Payload of the received message.
/// @returns The number of consumed bytes or a negative value to signal an
/// error.
/// @note Discarded data is lost permanently.
[[nodiscard]] virtual ptrdiff_t consume(byte_span payload) = 0;
};
/// Provides access to a resource that operates on the granularity of messages,
/// e.g., a UDP socket.
/// Provides access to a resource that operates on the granularity of binary
/// messages.
class CAF_NET_EXPORT lower_layer : public generic_lower_layer {
public:
virtual ~lower_layer();
......@@ -67,4 +41,4 @@ public:
virtual bool end_message() = 0;
};
} // namespace caf::net::message_oriented
} // namespace caf::net::binary
// 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/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/net/binary/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/generic_upper_layer.hpp"
namespace caf::net::binary {
/// Consumes binary messages from the lower layer.
class CAF_NET_EXPORT upper_layer : public generic_upper_layer {
public:
virtual ~upper_layer();
/// Initializes the upper layer.
/// @param down A pointer to the lower layer that remains valid for the
/// lifetime of the upper layer.
/// @param config Protocol-dependent configuration parameters.
virtual error init(lower_layer* down, const settings& config) = 0;
/// Consumes bytes from the lower layer.
/// @param payload Payload of the received message.
/// @returns The number of consumed bytes or a negative value to signal an
/// error.
/// @note Discarded data is lost permanently.
[[nodiscard]] virtual ptrdiff_t consume(byte_span payload) = 0;
};
} // namespace caf::net::binary
// 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/detail/atomic_ref_counted.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.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 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_->schedule(do_wakeup_);
}
void ref_consumer() const noexcept override {
this->ref();
}
void deref_consumer() const noexcept override {
this->deref();
}
template <class Policy, class Observer>
std::pair<bool, size_t> pull(Policy policy, size_t demand, Observer& dst) {
return buf_->pull(policy, demand, dst);
}
void cancel() {
buf_->cancel();
buf_ = nullptr;
mgr_ = nullptr;
do_wakeup_.dispose();
do_wakeup_ = nullptr;
}
bool has_data() const noexcept {
return buf_->has_data();
}
bool has_consumer_event() const noexcept {
return buf_->has_consumer_event();
}
static ptr_type make(buf_ptr buf, socket_manager_ptr mgr, action do_wakeup) {
if (buf) {
CAF_ASSERT(mgr);
CAF_ASSERT(do_wakeup);
auto adapter = ptr_type{new consumer_adapter(buf, std::move(mgr),
std::move(do_wakeup)), //
false};
buf->set_consumer(adapter);
return adapter;
} else {
return nullptr;
}
}
private:
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
}
intrusive_ptr<Buffer> buf_;
socket_manager_ptr mgr_;
action do_wakeup_;
action do_cancel_;
};
template <class T>
using consumer_adapter_ptr = intrusive_ptr<consumer_adapter<T>>;
} // namespace caf::net
......@@ -4,6 +4,8 @@
#pragma once
#include "caf/async/fwd.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/type_id.hpp"
......@@ -55,4 +57,26 @@ using socket_manager_ptr = intrusive_ptr<socket_manager>;
using text_buffer = std::vector<char>;
// -- factory functions --------------------------------------------------------
template <class>
struct actor_shell_ptr_oracle;
template <>
struct actor_shell_ptr_oracle<actor> {
using type = actor_shell_ptr;
};
template <class... Sigs>
struct actor_shell_ptr_oracle<typed_actor<Sigs...>> {
using type = typed_actor_shell_ptr<Sigs...>;
};
template <class Handle>
using actor_shell_ptr_t = typename actor_shell_ptr_oracle<Handle>::type;
template <class Handle = caf::actor>
actor_shell_ptr_t<Handle>
make_actor_shell(actor_system&, async::execution_context_ptr);
} // namespace caf::net
......@@ -18,10 +18,14 @@ public:
/// Queries whether the output device can accept more data straight away.
[[nodiscard]] virtual bool can_send_more() const noexcept = 0;
/// Queries whether the lower layer is currently configured to halt receiving
/// of additional bytes or messages.
/// Queries whether the lower layer is currently reading from its input
/// device.
[[nodiscard]] virtual bool is_reading() const noexcept = 0;
/// Triggers a write callback after the write device signals downstream
/// capacity. Does nothing if this layer is already writing.
virtual void write_later() = 0;
/// Shuts down any connection or session gracefully. Any pending data gets
/// flushed before closing the socket.
virtual void shutdown() = 0;
......
......@@ -52,8 +52,9 @@ using http_request_producer_ptr = intrusive_ptr<http_request_producer>;
class CAF_NET_EXPORT http_flow_adapter : public net::http::upper_layer {
public:
explicit http_flow_adapter(http_request_producer_ptr ptr)
: producer_(std::move(ptr)) {
explicit http_flow_adapter(async::execution_context_ptr loop,
http_request_producer_ptr ptr)
: loop_(std::move(loop)), producer_(std::move(ptr)) {
// nop
}
......@@ -63,18 +64,18 @@ public:
void abort(const error& reason) override;
error init(net::socket_manager* owner, net::http::lower_layer* down,
const settings& config) override;
error init(net::http::lower_layer* down, const settings& config) override;
ptrdiff_t consume(const net::http::header& hdr,
const_byte_span payload) override;
static auto make(http_request_producer_ptr ptr) {
return std::make_unique<http_flow_adapter>(ptr);
static auto make(async::execution_context_ptr loop,
http_request_producer_ptr ptr) {
return std::make_unique<http_flow_adapter>(loop, ptr);
}
private:
async::execution_context* parent_ = nullptr;
async::execution_context_ptr loop_;
net::http::lower_layer* down_ = nullptr;
std::vector<disposable> pending_;
http_request_producer_ptr producer_;
......@@ -94,7 +95,7 @@ public:
template <class Socket>
net::socket_manager_ptr make(net::multiplexer* mpx, Socket fd) {
auto app = http_flow_adapter::make(producer_);
auto app = http_flow_adapter::make(mpx, producer_);
auto serv = net::http::server::make(std::move(app));
auto transport = Transport::make(fd, std::move(serv));
auto res = net::socket_manager::make(mpx, fd, std::move(transport));
......
......@@ -81,6 +81,8 @@ public:
bool is_reading() const noexcept override;
void write_later() override;
void shutdown() override;
void request_messages() override;
......@@ -101,7 +103,7 @@ public:
// -- stream_oriented::upper_layer implementation ----------------------------
error init(socket_manager* owner, stream_oriented::lower_layer* down,
error init(stream_oriented::lower_layer* down,
const settings& config) override;
void abort(const error& reason) override;
......
......@@ -18,14 +18,9 @@ public:
virtual ~upper_layer();
/// Initializes the upper layer.
/// @param owner A pointer to the socket manager that owns the entire
/// protocol stack. Remains valid for the lifetime of the upper
/// layer.
/// @param down A pointer to the lower layer that remains valid for the
/// lifetime of the upper layer.
virtual error
init(socket_manager* owner, lower_layer* down, const settings& config)
= 0;
virtual error init(lower_layer* down, const settings& config) = 0;
/// Consumes an HTTP message.
/// @param hdr The header fields for the received message.
......
......@@ -6,7 +6,8 @@
#include "caf/byte_span.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/net/message_oriented.hpp"
#include "caf/net/binary/lower_layer.hpp"
#include "caf/net/binary/upper_layer.hpp"
#include "caf/net/stream_oriented.hpp"
#include <cstdint>
......@@ -23,11 +24,11 @@ namespace caf::net {
/// on 32-bit platforms.
class CAF_NET_EXPORT length_prefix_framing
: public stream_oriented::upper_layer,
public message_oriented::lower_layer {
public binary::lower_layer {
public:
// -- member types -----------------------------------------------------------
using upper_layer_ptr = std::unique_ptr<message_oriented::upper_layer>;
using upper_layer_ptr = std::unique_ptr<binary::upper_layer>;
// -- constants --------------------------------------------------------------
......@@ -45,36 +46,13 @@ public:
// -- high-level factory functions -------------------------------------------
// /// Runs a WebSocket server on the connected socket `fd`.
// /// @param mpx The multiplexer that takes ownership of the socket.
// /// @param fd A connected stream socket.
// /// @param cfg Additional configuration parameters for the protocol
// stack.
// /// @param in Inputs for writing to the socket.
// /// @param out Outputs from the socket.
// /// @param trait Converts between the native and the wire format.
// /// @relates length_prefix_framing
// template <template <class> class Transport = stream_transport, class
// Socket,
// class T, class Trait, class... TransportArgs>
// error run(actor_system&sys,Socket fd,
// const settings& cfg,
// async::consumer_resource<T> in,
// async::producer_resource<T> out,
// Trait trait, TransportArgs&&...
// args) {
// using app_t
// = Transport<length_prefix_framing<message_flow_bridge<T, Trait>>>;
// auto mgr = make_socket_manager<app_t>(fd, &mpx,
// std::forward<TransportArgs>(args)...,
// std::move(in), std::move(out),
// std::move(trait));
// return mgr->init(cfg);
// }
// disposable accept(actor_system& sys, Socket fd,
// acceptor_resource_t<Ts...> out, OnRequest on_request,
// size_t limit = 0);
// -- implementation of stream_oriented::upper_layer -------------------------
error init(socket_manager* owner, stream_oriented::lower_layer* down,
error init(stream_oriented::lower_layer* down,
const settings& config) override;
void abort(const error& reason) override;
......@@ -85,7 +63,7 @@ public:
bool done_sending() override;
// -- implementation of message_oriented::lower_layer ------------------------
// -- implementation of binary::lower_layer ----------------------------------
bool can_send_more() const noexcept override;
......@@ -95,6 +73,8 @@ public:
bool is_reading() const noexcept override;
void write_later() override;
void begin_message() override;
byte_buffer& message_buffer() override;
......
#include "caf/actor_system.hpp"
#include "caf/net/actor_shell.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/typed_actor_shell.hpp"
namespace caf::net {
template <class Handle>
actor_shell_ptr_t<Handle>
make_actor_shell(actor_system& sys, async::execution_context_ptr loop) {
auto f = [](abstract_actor_shell* self, message& msg) -> result<message> {
self->quit(make_error(sec::unexpected_message, std::move(msg)));
return make_error(sec::unexpected_message);
};
using ptr_type = actor_shell_ptr_t<Handle>;
using impl_type = typename ptr_type::element_type;
auto hdl = sys.spawn<impl_type>(loop);
auto ptr = ptr_type{actor_cast<strong_actor_ptr>(std::move(hdl))};
ptr->set_fallback(std::move(f));
return ptr;
}
} // namespace caf::net
......@@ -5,11 +5,11 @@
#pragma once
#include "caf/actor_system.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/net/consumer_adapter.hpp"
#include "caf/net/message_oriented.hpp"
#include "caf/async/consumer_adapter.hpp"
#include "caf/async/producer_adapter.hpp"
#include "caf/net/binary/lower_layer.hpp"
#include "caf/net/binary/upper_layer.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/producer_adapter.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/sec.hpp"
#include "caf/settings.hpp"
......@@ -32,7 +32,7 @@ namespace caf::net {
/// };
/// ~~~
template <class Trait>
class message_flow_bridge : public message_oriented::upper_layer {
class message_flow_bridge : public binary::upper_layer {
public:
/// The input type for the application.
using input_type = typename Trait::input_type;
......@@ -64,15 +64,17 @@ public:
// nop
}
error init(net::socket_manager* mgr, message_oriented::lower_layer* down,
error init(net::socket_manager* mgr, binary::lower_layer* down,
const settings&) {
down_ = down;
if (in_res_) {
in_ = consumer_adapter<pull_buffer_t>::try_open(mgr, in_res_);
if (auto in = make_consumer_adapter(in_res_, mgr->mpx_ptr(),
do_wakeup_cb())) {
in_ = std::move(*in);
in_res_ = nullptr;
}
if (out_res_) {
out_ = producer_adapter<push_buffer_t>::try_open(mgr, out_res_);
if (auto out = make_producer_adapter(out_res_, mgr->mpx_ptr(),
do_resume_cb(), do_cancel_cb())) {
out_ = std::move(*out);
out_res_ = nullptr;
}
if (!in_ && !out_)
......@@ -87,60 +89,39 @@ public:
return trait_.convert(item, buf) && down_->end_message();
}
struct write_helper {
message_flow_bridge* bridge;
bool aborted = false;
error err;
write_helper(message_flow_bridge* bridge) : bridge(bridge) {
// nop
}
void on_next(const input_type& item) {
if (!bridge->write(item))
aborted = true;
}
void on_complete() {
// nop
}
void on_error(const error& x) {
err = x;
}
};
void 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_->shutdown(helper.err);
} else {
input_type tmp;
while (down_->can_send_more()) {
switch (in_.pull(async::delay_errors, tmp)) {
case async::read_result::ok:
if (!write(tmp)) {
down_->shutdown(trait_.last_error());
return;
}
break;
case async::read_result::stop:
down_->shutdown();
}
in_ = nullptr;
} else if (helper.aborted) {
in_->cancel();
in_ = nullptr;
down_->shutdown(trait_.last_error());
break;
case async::read_result::abort:
down_->shutdown(in_.abort_reason());
break;
default: // try later
return;
}
}
}
bool done_sending() override {
return !in_ || !in_->has_data();
return !in_->has_consumer_event();
}
void abort(const error& reason) override {
CAF_LOG_TRACE(CAF_ARG(reason));
if (out_) {
if (reason == sec::socket_disconnected || reason == sec::disposed)
out_->close();
out_.close();
else
out_->abort(reason);
out_ = nullptr;
out_ > abort(reason);
}
if (in_) {
in_->cancel();
......@@ -160,14 +141,31 @@ public:
}
private:
action do_wakeup_cb() {
return make_action([this] { down_->write_later(); });
}
action do_resume_cb() {
return make_action([this] { down_->request_messages(); });
}
action do_cancel_cb() {
return make_action([this] {
if (out_) {
out_ = nullptr;
down_->shutdown();
}
});
}
/// Points to the next layer down the protocol stack.
message_oriented::lower_layer* down_ = nullptr;
binary::lower_layer* down_ = nullptr;
/// Incoming messages, serialized to the socket.
consumer_adapter_ptr<pull_buffer_t> in_;
async::consumer_adapter<input_type> in_;
/// Outgoing messages, deserialized from the socket.
producer_adapter_ptr<push_buffer_t> out_;
async::producer_adapter<output_type> out_;
/// Converts between raw bytes and items.
Trait trait_;
......
// 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/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"
namespace caf::net {
/// Connects a socket manager to an asynchronous producer resource.
template <class Buffer>
class producer_adapter final : public detail::atomic_ref_counted,
public async::producer {
public:
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_->schedule(do_cancel_);
}
void on_consumer_demand(size_t) override {
mgr_->schedule(do_resume_);
}
void ref_producer() const noexcept override {
this->ref();
}
void deref_producer() const noexcept override {
this->deref();
}
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 {
return nullptr;
}
}
/// Makes `item` available to the consumer.
/// @returns the remaining demand.
size_t push(const value_type& item) {
if (buf_) {
return buf_->push(item);
} else {
return 0;
}
}
/// Makes `items` available to the consumer.
/// @returns the remaining demand.
size_t push(span<const value_type> items) {
if (buf_) {
return buf_->push(items);
} else {
return 0;
}
}
void close() {
if (buf_) {
buf_->close();
reset();
}
}
void abort(error reason) {
if (buf_) {
buf_->abort(std::move(reason));
reset();
}
}
private:
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 reset() {
buf_ = nullptr;
mgr_ = nullptr;
do_resume_.dispose();
do_resume_ = nullptr;
do_cancel_.dispose();
do_cancel_ = nullptr;
}
intrusive_ptr<Buffer> buf_;
intrusive_ptr<socket_manager> mgr_;
action do_resume_;
action do_cancel_;
};
template <class T>
using producer_adapter_ptr = intrusive_ptr<producer_adapter<T>>;
} // namespace caf::net
......@@ -61,8 +61,7 @@ public:
void abort(const error& reason) override;
error init(socket_manager* owner, http::lower_layer* down,
const settings& config) override;
error init(http::lower_layer* down, const settings& config) override;
ptrdiff_t consume(const http::header& hdr, const_byte_span payload) override;
......
......@@ -44,6 +44,13 @@ public:
fd_ = x;
}
void reset() noexcept {
if (fd_.id != invalid_socket_id) {
close(fd_);
fd_.id = invalid_socket_id;
}
}
Socket release() noexcept {
auto sock = fd_;
fd_.id = invalid_socket_id;
......
......@@ -50,26 +50,6 @@ public:
static socket_manager_ptr make(multiplexer* mpx, socket handle,
event_handler_ptr handler);
template <class Handle = actor, class FallbackHandler>
detail::infer_actor_shell_ptr_type<Handle>
make_actor_shell(FallbackHandler f) {
using ptr_type = detail::infer_actor_shell_ptr_type<Handle>;
using impl_type = typename ptr_type::element_type;
auto hdl = system().spawn<impl_type>(this);
auto ptr = ptr_type{actor_cast<strong_actor_ptr>(std::move(hdl))};
ptr->set_fallback(std::move(f));
return ptr;
}
template <class Handle = actor>
auto make_actor_shell() {
auto f = [](abstract_actor_shell* self, message& msg) -> result<message> {
self->quit(make_error(sec::unexpected_message, std::move(msg)));
return make_error(sec::unexpected_message);
};
return make_actor_shell<Handle>(std::move(f));
}
// -- properties -------------------------------------------------------------
/// Returns the handle for the managed socket.
......
......@@ -21,14 +21,9 @@ public:
virtual ~upper_layer();
/// Initializes the upper layer.
/// @param owner A pointer to the socket manager that owns the entire
/// protocol stack. Remains valid for the lifetime of the upper
/// layer.
/// @param down A pointer to the lower layer that remains valid for the
/// lifetime of the upper layer.
virtual error
init(socket_manager* owner, lower_layer* down, const settings& config)
= 0;
virtual error init(lower_layer* down, const settings& config) = 0;
/// Consumes bytes from the lower layer.
/// @param buffer Available bytes to read.
......
......@@ -123,6 +123,8 @@ public:
bool is_reading() const noexcept override;
void write_later() override;
void shutdown() override;
// -- properties -------------------------------------------------------------
......
......@@ -55,7 +55,7 @@ public:
template <class... Fs>
void set_behavior(Fs... fs) {
auto new_bhvr = behavior_type{std::move(fs)...};
this->bhvr_ = std::move(new_bhvr.unbox());
this->set_behavior_impl(std::move(new_bhvr.unbox()));
}
// -- overridden functions of local_actor ------------------------------------
......@@ -72,7 +72,9 @@ class typed_actor_shell_ptr {
public:
// -- friends ----------------------------------------------------------------
friend class socket_manager;
template <class Handle>
friend actor_shell_ptr_t<Handle>
make_actor_shell(actor_system&, async::execution_context_ptr);
// -- member types -----------------------------------------------------------
......@@ -154,23 +156,3 @@ private:
};
} // namespace caf::net
namespace caf::detail {
template <class T>
struct typed_actor_shell_ptr_oracle;
template <class... Sigs>
struct typed_actor_shell_ptr_oracle<typed_actor<Sigs...>> {
using type = net::typed_actor_shell_ptr<Sigs...>;
};
} // namespace caf::detail
namespace caf::net {
template <class Handle>
using typed_actor_shell_ptr_t =
typename detail::typed_actor_shell_ptr_oracle<Handle>::type;
} // namespace caf::net
......@@ -33,7 +33,7 @@ public:
template <class Socket>
net::socket_manager_ptr make(net::multiplexer* mpx, Socket fd) {
auto app = net::web_socket::flow_bridge<Trait>::make(connector_);
auto app = net::web_socket::flow_bridge<Trait>::make(mpx, connector_);
auto ws = net::web_socket::server::make(std::move(app));
auto transport = Transport::make(fd, std::move(ws));
return net::socket_manager::make(mpx, fd, std::move(transport));
......
......@@ -71,7 +71,7 @@ public:
// -- implementation of stream_oriented::upper_layer -------------------------
error init(socket_manager* owner, stream_oriented::lower_layer* down,
error init(stream_oriented::lower_layer* down,
const settings& config) override;
void abort(const error& reason) override;
......@@ -95,9 +95,6 @@ private:
/// Stores the upper layer.
framing framing_;
/// Stores a pointer to the owning manager for the delayed initialization.
socket_manager* owner_ = nullptr;
settings cfg_;
};
......
......@@ -31,13 +31,13 @@ void connect(actor_system& sys, Socket fd, handshake hs, Init init) {
auto [ws_pull, app_push] = async::make_spsc_buffer_resource<frame>();
auto [app_pull, ws_push] = async::make_spsc_buffer_resource<frame>();
using conn_t = flow_connector_trivial_impl<trait_t>;
auto& mpx = sys.network_manager().mpx();
auto mpx = sys.network_manager().mpx_ptr();
auto conn = std::make_shared<conn_t>(std::move(ws_pull), std::move(ws_push));
auto bridge = flow_bridge<trait_t>::make(std::move(conn));
auto bridge = flow_bridge<trait_t>::make(mpx, std::move(conn));
auto impl = client::make(std::move(hs), std::move(bridge));
auto transport = Transport::make(fd, std::move(impl));
auto ptr = socket_manager::make(&mpx, fd, std::move(transport));
mpx.init(ptr);
auto ptr = socket_manager::make(mpx, fd, std::move(transport));
mpx->init(ptr);
init(connect_event_t{app_pull, app_push});
}
......
......@@ -4,10 +4,10 @@
#pragma once
#include "caf/async/consumer_adapter.hpp"
#include "caf/async/producer_adapter.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/fwd.hpp"
#include "caf/net/consumer_adapter.hpp"
#include "caf/net/producer_adapter.hpp"
#include "caf/net/web_socket/flow_connector.hpp"
#include "caf/net/web_socket/lower_layer.hpp"
#include "caf/net/web_socket/request.hpp"
......@@ -27,28 +27,24 @@ public:
using output_type = typename Trait::output_type;
/// Type for the consumer adapter. We consume the output of the application.
using consumer_type = consumer_adapter<async::spsc_buffer<output_type>>;
using consumer_type = async::consumer_adapter<output_type>;
/// Type for the producer adapter. We produce the input of the application.
using producer_type = producer_adapter<async::spsc_buffer<input_type>>;
using producer_type = async::producer_adapter<input_type>;
using request_type = request<Trait>;
using connector_pointer = flow_connector_ptr<Trait>;
explicit flow_bridge(connector_pointer conn) : conn_(std::move(conn)) {
explicit flow_bridge(async::execution_context_ptr loop,
connector_pointer conn)
: loop_(std::move(loop)), conn_(std::move(conn)) {
// nop
}
~flow_bridge() {
if (in_)
in_->cancel();
if (out_)
out_->close();
}
static std::unique_ptr<flow_bridge> make(connector_pointer conn) {
return std::make_unique<flow_bridge>(std::move(conn));
static std::unique_ptr<flow_bridge> make(async::execution_context_ptr loop,
connector_pointer conn) {
return std::make_unique<flow_bridge>(std::move(loop), std::move(conn));
}
bool write(const output_type& item) {
......@@ -63,38 +59,13 @@ public:
}
}
struct write_helper {
flow_bridge* thisptr;
bool aborted = false;
error err;
explicit write_helper(flow_bridge* thisptr) : thisptr(thisptr) {
// nop
}
void on_next(const output_type& item) {
if (!thisptr->write(item))
aborted = true;
}
void on_complete() {
// nop
}
void on_error(const error& x) {
err = x;
}
};
bool running() const noexcept {
return in_ || out_;
}
// -- implementation of web_socket::lower_layer ------------------------------
error init(net::socket_manager* mgr, web_socket::lower_layer* down,
const settings& cfg) override {
CAF_ASSERT(mgr != nullptr);
error init(web_socket::lower_layer* down, const settings& cfg) override {
down_ = down;
auto [err, pull, push] = conn_->on_request(cfg);
if (!err) {
......@@ -105,12 +76,11 @@ public:
});
auto do_resume = make_action([this] { down_->request_messages(); });
auto do_cancel = make_action([this] {
out_ = nullptr;
if (!running())
down_->shutdown();
});
in_ = consumer_type::make(pull.try_open(), mgr, std::move(do_wakeup));
out_ = producer_type::make(push.try_open(), mgr, std::move(do_resume),
in_ = consumer_type::make(pull.try_open(), loop_, std::move(do_wakeup));
out_ = producer_type::make(push.try_open(), loop_, std::move(do_resume),
std::move(do_cancel));
conn_ = nullptr;
if (running())
......@@ -125,28 +95,29 @@ public:
}
void 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_->shutdown(helper.err);
} else {
input_type tmp;
while (down_->can_send_more()) {
switch (in_.pull(async::delay_errors, tmp)) {
case async::read_result::ok:
if (!write(tmp)) {
down_->shutdown(trait_.last_error());
return;
}
break;
case async::read_result::stop:
down_->shutdown();
}
in_ = nullptr;
} else if (helper.aborted) {
in_->cancel();
in_ = nullptr;
down_->shutdown(trait_.last_error());
} else if (consumed == 0) {
return;
break;
case async::read_result::abort:
down_->shutdown(in_.abort_reason());
break;
default: // try later
return;
}
}
}
bool done_sending() override {
return !in_ || !in_->has_consumer_event();
return !in_.has_consumer_event();
}
void abort(const error& reason) override {
......@@ -154,15 +125,11 @@ public:
if (out_) {
if (reason == sec::connection_closed || reason == sec::socket_disconnected
|| reason == sec::disposed)
out_->close();
out_.close();
else
out_->abort(reason);
out_ = nullptr;
}
if (in_) {
in_->cancel();
in_ = nullptr;
out_.abort(reason);
}
in_.cancel();
}
ptrdiff_t consume_binary(byte_span buf) override {
......@@ -171,7 +138,7 @@ public:
input_type val;
if (!trait_.convert(buf, val))
return -1;
if (out_->push(std::move(val)) == 0)
if (out_.push(std::move(val)) == 0)
down_->suspend_reading();
return static_cast<ptrdiff_t>(buf.size());
}
......@@ -182,7 +149,7 @@ public:
input_type val;
if (!trait_.convert(buf, val))
return -1;
if (out_->push(std::move(val)) == 0)
if (out_.push(std::move(val)) == 0)
down_->suspend_reading();
return static_cast<ptrdiff_t>(buf.size());
}
......@@ -191,14 +158,17 @@ private:
web_socket::lower_layer* down_;
/// The output of the application. Serialized to the socket.
intrusive_ptr<consumer_type> in_;
consumer_type in_;
/// The input to the application. Deserialized from the socket.
intrusive_ptr<producer_type> out_;
producer_type out_;
/// Converts between raw bytes and native C++ objects.
Trait trait_;
/// Runs callbacks in the I/O event loop.
async::execution_context_ptr loop_;
/// Initializes the bridge. Disposed (set to null) after initializing.
connector_pointer conn_;
};
......
......@@ -72,7 +72,6 @@ public:
std::string_view as_text() const noexcept;
private:
class data {
public:
data() = delete;
......@@ -133,6 +132,7 @@ private:
std::byte storage_[];
};
private:
explicit frame(intrusive_ptr<data> ptr) : data_(std::move(ptr)) {
// nop
}
......
......@@ -49,7 +49,7 @@ public:
// -- initialization ---------------------------------------------------------
void init(socket_manager* owner, stream_oriented::lower_layer* down);
void init(stream_oriented::lower_layer* down);
// -- properties -------------------------------------------------------------
......@@ -85,6 +85,8 @@ public:
bool is_reading() const noexcept override;
void write_later() override;
void shutdown(status code, std::string_view desc) override;
void request_messages() override;
......
......@@ -74,7 +74,7 @@ public:
// -- stream_oriented::upper_layer implementation ----------------------------
error init(socket_manager* owner, stream_oriented::lower_layer* down,
error init(stream_oriented::lower_layer* down,
const settings& config) override;
void abort(const error& reason) override;
......@@ -98,9 +98,6 @@ private:
/// Stores the upper layer.
framing framing_;
/// Stores a pointer to the owning manager for the delayed initialization.
socket_manager* owner_ = nullptr;
/// Holds a copy of the settings in order to delay initialization of the upper
/// layer until the handshake completed. We also fill this dictionary with the
/// contents of the HTTP GET header.
......
......@@ -16,9 +16,7 @@ class CAF_NET_EXPORT upper_layer : public generic_upper_layer {
public:
virtual ~upper_layer();
virtual error
init(net::socket_manager* mgr, lower_layer* down, const settings& cfg)
= 0;
virtual error init(lower_layer* down, const settings& cfg) = 0;
virtual ptrdiff_t consume_binary(byte_span buf) = 0;
virtual ptrdiff_t consume_text(std::string_view buf) = 0;
};
......
......@@ -4,6 +4,7 @@
#include "caf/net/abstract_actor_shell.hpp"
#include "caf/action.hpp"
#include "caf/callback.hpp"
#include "caf/config.hpp"
#include "caf/detail/default_invoke_result_visitor.hpp"
......@@ -18,9 +19,15 @@ namespace caf::net {
// -- constructors, destructors, and assignment operators ----------------------
abstract_actor_shell::abstract_actor_shell(actor_config& cfg,
socket_manager* owner)
: super(cfg), mailbox_(policy::normal_messages{}), owner_(owner) {
async::execution_context_ptr loop)
: super(cfg), mailbox_(policy::normal_messages{}), loop_(loop) {
mailbox_.try_block();
resume_ = make_action([this] {
for (;;) {
if (!consume_message() && try_block_mailbox())
return;
}
});
}
abstract_actor_shell::~abstract_actor_shell() {
......@@ -117,13 +124,13 @@ bool abstract_actor_shell::enqueue(mailbox_element_ptr ptr, execution_unit*) {
switch (mailbox().push_back(std::move(ptr))) {
case intrusive::inbox_result::unblocked_reader: {
CAF_LOG_ACCEPT_EVENT(true);
std::unique_lock<std::mutex> guard{owner_mtx_};
// The owner can only be null if this enqueue succeeds, then we close the
// mailbox and reset owner_ in cleanup() before acquiring the mutex here.
std::unique_lock<std::mutex> guard{loop_mtx_};
// The loop can only be null if this enqueue succeeds, then we close the
// mailbox and reset loop_ in cleanup() before acquiring the mutex here.
// Hence, the mailbox element has already been disposed and we can simply
// skip any further processing.
if (owner_)
owner_->mpx().register_writing(owner_);
if (loop_)
loop_->schedule(resume_);
return true;
}
case intrusive::inbox_result::success:
......@@ -175,8 +182,9 @@ bool abstract_actor_shell::cleanup(error&& fail_state, execution_unit* host) {
}
// Detach from owner.
{
std::unique_lock<std::mutex> guard{owner_mtx_};
owner_ = nullptr;
std::unique_lock<std::mutex> guard{loop_mtx_};
loop_ = nullptr;
resume_.dispose();
}
// Dispatch to parent's `cleanup` function.
return super::cleanup(std::move(fail_state), host);
......
......@@ -8,8 +8,8 @@ namespace caf::net {
// -- actor_shell --------------------------------------------------------------
actor_shell::actor_shell(actor_config& cfg, socket_manager* owner)
: super(cfg, owner) {
actor_shell::actor_shell(actor_config& cfg, async::execution_context_ptr loop)
: super(cfg, loop) {
// nop
}
......
// 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.
#include "caf/net/binary/frame.hpp"
#include <cstring>
#include <new>
namespace caf::net::binary {
frame::frame(const_byte_span data) {
auto total_size = sizeof(web_socket::frame::data) + data.size();
auto vptr = malloc(total_size);
data_.reset(new (vptr) web_socket::frame::data(true, data.size()), false);
memcpy(data_->storage(), data.data(), data.size());
}
} // namespace caf::net::binary
......@@ -2,16 +2,12 @@
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/net/message_oriented.hpp"
#include "caf/net/binary/lower_layer.hpp"
namespace caf::net::message_oriented {
upper_layer::~upper_layer() {
// nop
}
namespace caf::net::binary {
lower_layer::~lower_layer() {
// nop
}
} // namespace caf::net::message_oriented
} // namespace caf::net::binary
// 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.
#include "caf/net/binary/upper_layer.hpp"
namespace caf::net::binary {
upper_layer::~upper_layer() {
// nop
}
} // namespace caf::net::binary
......@@ -48,9 +48,7 @@ void http_flow_adapter::abort(const error&) {
pending.dispose();
}
error http_flow_adapter::init(net::socket_manager* owner,
net::http::lower_layer* down, const settings&) {
parent_ = owner->mpx_ptr();
error http_flow_adapter::init(net::http::lower_layer* down, const settings&) {
down_ = down;
down_->request_messages();
return none;
......@@ -69,7 +67,7 @@ ptrdiff_t http_flow_adapter::consume(const net::http::header& hdr,
auto buf = std::vector<std::byte>{payload.begin(), payload.end()};
auto impl = request::impl{hdr, std::move(buf), std::move(prom)};
producer_->push(request{std::make_shared<request::impl>(std::move(impl))});
auto hdl = fut.bind_to(parent_).then(
auto hdl = fut.bind_to(*loop_).then(
[this](const response& res) {
down_->begin_header(res.code());
for (auto& [key, val] : res.header_fields())
......
......@@ -18,6 +18,10 @@ bool server::is_reading() const noexcept {
return down_->is_reading();
}
void server::write_later() {
down_->write_later();
}
void server::shutdown() {
down_->shutdown();
}
......@@ -76,14 +80,14 @@ bool server::send_end_of_chunks() {
// -- stream_oriented::upper_layer implementation ------------------------------
error server::init(socket_manager* owner, stream_oriented::lower_layer* down,
const settings& cfg) {
error server::init(stream_oriented::lower_layer* down, const settings& cfg) {
down_ = down;
if (auto max_size = get_as<uint32_t>(cfg, "http.max-request-size"))
max_request_size_ = *max_size;
if (auto err = up_->init(owner, this, cfg))
if (auto err = up_->init(this, cfg))
return err;
return none;
else
return none;
}
void server::abort(const error& reason) {
......
......@@ -24,11 +24,10 @@ length_prefix_framing::make(upper_layer_ptr up) {
// -- implementation of stream_oriented::upper_layer ---------------------------
error length_prefix_framing::init(socket_manager* owner,
stream_oriented::lower_layer* down,
error length_prefix_framing::init(stream_oriented::lower_layer* down,
const settings& cfg) {
down_ = down;
return up_->init(owner, this, cfg);
return up_->init(this, cfg);
}
void length_prefix_framing::abort(const error& reason) {
......@@ -48,8 +47,10 @@ ptrdiff_t length_prefix_framing::consume(byte_span input, byte_span) {
auto msg_size = static_cast<size_t>(detail::from_network_order(u32_size));
if (msg_size == 0) {
// Ignore empty messages.
CAF_LOG_DEBUG("received empty message");
return static_cast<ptrdiff_t>(input.size());
CAF_LOG_ERROR("received empty message");
up_->abort(make_error(sec::logic_error,
"received empty buffer from stream layer"));
return -1;
} else if (msg_size > max_message_length) {
CAF_LOG_DEBUG("exceeded maximum message size");
up_->abort(
......@@ -87,7 +88,7 @@ bool length_prefix_framing::done_sending() {
return up_->done_sending();
}
// -- implementation of message_oriented::lower_layer --------------------------
// -- implementation of binary::lower_layer ------------------------------------
bool length_prefix_framing::can_send_more() const noexcept {
return down_->can_send_more();
......@@ -101,6 +102,10 @@ bool length_prefix_framing::is_reading() const noexcept {
return down_->is_reading();
}
void length_prefix_framing::write_later() {
down_->write_later();
}
void length_prefix_framing::request_messages() {
if (!down_->is_reading())
down_->configure_read(receive_policy::exactly(hdr_size));
......
......@@ -36,7 +36,7 @@ void server::abort(const error&) {
// nop
}
error server::init(socket_manager*, http::lower_layer* down, const settings&) {
error server::init(http::lower_layer* down, const settings&) {
down_ = down;
down_->request_messages();
return caf::none;
......
......@@ -76,7 +76,7 @@ void socket_manager::deregister() {
void socket_manager::schedule_handover() {
deregister();
mpx_->schedule_fn([ptr = strong_this()] { //
mpx_->schedule_fn([ptr = strong_this()] {
event_handler_ptr next;
if (ptr->handler_->do_handover(next)) {
ptr->handler_.swap(next);
......
......@@ -77,7 +77,7 @@ void stream_transport::configure_read(receive_policy rd) {
min_read_size_ = rd.min_size;
max_read_size_ = rd.max_size;
if (restarting && !parent_->is_reading()) {
if (buffered_ >= min_read_size_) {
if (buffered_ > 0 && 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.
......@@ -111,6 +111,10 @@ bool stream_transport::is_reading() const noexcept {
return max_read_size_ > 0;
}
void stream_transport::write_later() {
parent_->register_writing();
}
void stream_transport::shutdown() {
if (write_buf_.empty()) {
parent_->shutdown();
......@@ -141,7 +145,7 @@ error stream_transport::init(socket_manager* owner, const settings& config) {
CAF_LOG_ERROR("send_buffer_size: " << socket_buf_size.error());
return std::move(socket_buf_size.error());
}
return up_->init(owner, this, config);
return up_->init(this, config);
}
void stream_transport::handle_read_event() {
......@@ -214,7 +218,6 @@ void stream_transport::handle_buffered_data() {
// 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_) {
auto old_max_read_size = max_read_size_;
auto n = std::min(buffered_, size_t{max_read_size_});
auto bytes = make_span(read_buf_.data(), n);
auto delta = bytes.subspan(delta_offset_);
......@@ -225,21 +228,6 @@ void stream_transport::handle_buffered_data() {
up_->abort(make_error(caf::sec::runtime_error, "consumed < 0"));
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
// requires more data to parse the input. It may of course only increase
// the max_read_size_ in this case, everything else makes no sense.
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(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;
}
} else if (static_cast<size_t>(consumed) > n) {
// Must not happen. An application cannot handle more data then we pass
// to it.
......@@ -252,7 +240,7 @@ void stream_transport::handle_buffered_data() {
auto del = static_cast<size_t>(consumed);
auto prev = buffered_;
buffered_ -= del;
delta_offset_ = 0;
delta_offset_ = static_cast<ptrdiff_t>(n - del);
if (buffered_ > 0) {
auto new_begin = read_buf_.begin() + del;
auto new_end = read_buf_.begin() + prev;
......@@ -267,7 +255,9 @@ void stream_transport::handle_buffered_data() {
void stream_transport::fail(const error& reason) {
CAF_LOG_TRACE(CAF_ARG(reason));
up_->abort(reason);
up_.reset();
parent_->deregister();
parent_->shutdown();
}
void stream_transport::handle_write_event() {
......
......@@ -38,16 +38,14 @@ std::unique_ptr<client> client::make(handshake_ptr hs, upper_layer_ptr up) {
// -- implementation of stream_oriented::upper_layer ---------------------------
error client::init(socket_manager* owner, stream_oriented::lower_layer* down,
const settings& cfg) {
error client::init(stream_oriented::lower_layer* down, const settings& cfg) {
CAF_ASSERT(hs_ != nullptr);
framing_.init(owner, down);
framing_.init(down);
if (!hs_->has_mandatory_fields())
return make_error(sec::runtime_error,
"handshake data lacks mandatory fields");
if (!hs_->has_valid_key())
hs_->randomize_key();
owner_ = owner;
cfg_ = cfg;
down->begin_output();
hs_->write_http_1_request(down->output_buffer());
......@@ -112,7 +110,7 @@ bool client::handle_header(std::string_view http) {
auto http_ok = hs_->is_valid_http_1_response(http);
hs_.reset();
if (http_ok) {
if (auto err = upper_layer().init(owner_, &framing_, cfg_)) {
if (auto err = upper_layer().init(&framing_, cfg_)) {
CAF_LOG_DEBUG("failed to initialize WebSocket framing layer");
return false;
} else {
......
// 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.
#include "caf/net/web_socket/frame.hpp"
#include <cstring>
......
......@@ -10,7 +10,7 @@ namespace caf::net::web_socket {
// -- initialization ---------------------------------------------------------
void framing::init(socket_manager*, stream_oriented::lower_layer* down) {
void framing::init(stream_oriented::lower_layer* down) {
std::random_device rd;
rng_.seed(rd());
down_ = down;
......@@ -30,6 +30,10 @@ bool framing::is_reading() const noexcept {
return down_->is_reading();
}
void framing::write_later() {
down_->write_later();
}
void framing::shutdown(status code, std::string_view msg) {
auto code_val = static_cast<uint16_t>(code);
uint32_t mask_key = 0;
......
......@@ -22,10 +22,8 @@ std::unique_ptr<server> server::make(upper_layer_ptr up) {
// -- stream_oriented::upper_layer implementation ------------------------------
error server::init(socket_manager* owner, stream_oriented::lower_layer* down,
const settings& cfg) {
framing_.init(owner, down);
owner_ = owner;
error server::init(stream_oriented::lower_layer* down, const settings& cfg) {
framing_.init(down);
cfg_ = cfg;
lower_layer().configure_read(receive_policy::up_to(handshake::max_http_size));
return none;
......@@ -128,7 +126,7 @@ bool server::handle_header(std::string_view http) {
put(fields, std::string{key}, std::string{val});
}
// Try to initialize the upper layer.
if (auto err = upper_layer().init(owner_, &framing_, cfg_)) {
if (auto err = upper_layer().init(&framing_, cfg_)) {
auto descr = to_string(err);
CAF_LOG_DEBUG("upper layer rejected a WebSocket connection:" << descr);
write_response(http::status::bad_request, descr);
......
......@@ -23,6 +23,10 @@ bool mock_stream_transport::is_reading() const noexcept {
return max_read_size > 0;
}
void mock_stream_transport::write_later() {
// nop
}
void mock_stream_transport::shutdown() {
// nop
}
......
......@@ -34,6 +34,8 @@ public:
bool is_reading() const noexcept override;
void write_later() override;
void shutdown() override;
void configure_read(caf::net::receive_policy policy) override;
......@@ -47,7 +49,7 @@ public:
// -- initialization ---------------------------------------------------------
caf::error init(const caf::settings& cfg) {
return up->init(nullptr, this, cfg);
return up->init(this, cfg);
}
caf::error init() {
......
......@@ -11,6 +11,7 @@
#include "caf/byte.hpp"
#include "caf/byte_span.hpp"
#include "caf/callback.hpp"
#include "caf/net/make_actor_shell.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/socket_manager.hpp"
......@@ -26,24 +27,20 @@ using svec = std::vector<std::string>;
class app_t : public net::stream_oriented::upper_layer {
public:
app_t() = default;
explicit app_t(actor hdl) : worker(std::move(hdl)) {
// nop
}
static auto make() {
return std::make_unique<app_t>();
explicit app_t(actor_system& sys, async::execution_context_ptr loop,
actor hdl = {})
: worker(std::move(hdl)) {
self = net::make_actor_shell(sys, loop);
}
static auto make(actor hdl) {
return std::make_unique<app_t>(std::move(hdl));
static auto make(actor_system& sys, async::execution_context_ptr loop,
actor hdl = {}) {
return std::make_unique<app_t>(sys, std::move(loop), std::move(hdl));
}
error init(net::socket_manager* mgr, net::stream_oriented::lower_layer* down,
error init(net::stream_oriented::lower_layer* down,
const settings&) override {
this->down = down;
self = mgr->make_actor_shell();
self->set_behavior([this](std::string& line) {
CAF_MESSAGE("received an asynchronous message: " << line);
lines.emplace_back(std::move(line));
......@@ -60,16 +57,15 @@ public:
}
void prepare_send() override {
while (!self->terminated() && self->consume_message())
; // repeat
// nop
}
bool done_sending() override {
return self->try_block_mailbox();
return true;
}
void abort(const error&) override {
// nop
void abort(const error& reason) override {
MESSAGE("abort: " << reason);
}
ptrdiff_t consume(byte_span buf, byte_span) override {
......@@ -197,7 +193,7 @@ CAF_TEST_FIXTURE_SCOPE(actor_shell_tests, fixture)
CAF_TEST(actor shells expose their mailbox to their owners) {
auto fd = testee_socket_guard.release();
auto app_uptr = app_t::make();
auto app_uptr = app_t::make(sys, mpx);
auto app = app_uptr.get();
auto transport = net::stream_transport::make(fd, std::move(app_uptr));
auto mgr = net::socket_manager::make(mpx.get(), fd, std::move(transport));
......@@ -209,6 +205,8 @@ CAF_TEST(actor shells expose their mailbox to their owners) {
anon_send(hdl, "line 3");
run_while([&] { return app->lines.size() != 3; });
CAF_CHECK_EQUAL(app->lines, svec({"line 1", "line 2", "line 3"}));
self_socket_guard.reset();
run_while([&] { return mpx->num_socket_managers() > 1; });
}
CAF_TEST(actor shells can send requests and receive responses) {
......@@ -218,7 +216,7 @@ CAF_TEST(actor shells can send requests and receive responses) {
};
});
auto fd = testee_socket_guard.release();
auto app_uptr = app_t::make(worker);
auto app_uptr = app_t::make(sys, mpx, worker);
auto app = app_uptr.get();
auto transport = net::stream_transport::make(fd, std::move(app_uptr));
auto mgr = net::socket_manager::make(mpx.get(), fd, std::move(transport));
......@@ -232,6 +230,8 @@ CAF_TEST(actor shells can send requests and receive responses) {
std::string_view received_response{reinterpret_cast<char*>(recv_buf.data()),
recv_buf.size()};
CAF_CHECK_EQUAL(received_response, expected_response);
self_socket_guard.reset();
run_while([&] { return mpx->num_socket_managers() > 1; });
}
CAF_TEST_FIXTURE_SCOPE_END()
// 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.consumer_adapter
#include "caf/net/consumer_adapter.hpp"
#include "net-test.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/scheduled_actor/flow.hpp"
using namespace caf;
namespace {
class reader {
public:
reader(net::stream_socket fd, size_t n) : sg_(fd) {
buf_.resize(n);
}
auto fd() {
return sg_.socket();
}
void read_some() {
if (rd_pos_ < buf_.size()) {
auto res = read(fd(), make_span(buf_).subspan(rd_pos_));
if (res > 0) {
rd_pos_ += static_cast<size_t>(res);
MESSAGE(rd_pos_ << " bytes received");
} else if (res < 0 && !net::last_socket_error_is_temporary()) {
FAIL("failed to read: " << net::last_socket_error_as_string());
}
}
}
size_t remaining() const noexcept {
return buf_.size() - rd_pos_;
}
bool done() const noexcept {
return remaining() == 0;
}
auto& buf() const {
return buf_;
}
private:
size_t rd_pos_ = 0;
byte_buffer buf_;
net::socket_guard<net::stream_socket> sg_;
};
class app_t : public net::stream_oriented::upper_layer {
public:
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
}
static auto make(resource_type input) {
return std::make_unique<app_t>(std::move(input));
}
error init(net::socket_manager* mgr,
net::stream_oriented::lower_layer* down_ptr,
const settings&) override {
down = down_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");
}
}
struct send_helper {
app_t* thisptr;
void on_next(int32_t item) {
thisptr->written_values.emplace_back(item);
auto offset = thisptr->written_bytes.size();
binary_serializer sink{nullptr, thisptr->written_bytes};
if (!sink.apply(item))
FAIL("sink.apply failed: " << sink.get_error());
auto bytes = make_span(thisptr->written_bytes).subspan(offset);
thisptr->down->begin_output();
auto& buf = thisptr->down->output_buffer();
buf.insert(buf.end(), bytes.begin(), bytes.end());
thisptr->down->end_output();
}
void on_complete() {
// nop
}
void on_error(const error&) {
// nop
}
};
void prepare_send() override {
if (!done && adapter) {
auto helper = send_helper{this};
while (down->can_send_more()) {
auto [again, consumed] = adapter->pull(async::delay_errors, 1, helper);
if (!again) {
MESSAGE("adapter signaled end-of-buffer");
adapter = nullptr;
done = true;
break;
} else if (consumed == 0) {
break;
}
}
MESSAGE(written_bytes.size() << " bytes written");
}
}
bool done_sending() override {
return done || !adapter->has_data();
}
void abort(const error& reason) override {
MESSAGE("app::abort: " << reason);
}
ptrdiff_t consume(byte_span, byte_span) override {
FAIL("app::consume called: unexpected data");
}
net::stream_oriented::lower_layer* down = nullptr;
bool done = false;
std::vector<int32_t> written_values;
byte_buffer written_bytes;
adapter_ptr adapter;
resource_type input;
};
struct fixture : test_coordinator_fixture<> {
fixture() : mm(sys) {
mm.mpx().set_thread_id();
if (auto err = mm.mpx().init())
CAF_FAIL("mpx.init() failed: " << err);
}
bool handle_io_event() override {
return mm.mpx().poll_once(false);
}
auto mpx() {
return mm.mpx_ptr();
}
net::middleman mm;
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("subscriber adapters wake up idle socket managers") {
GIVEN("an actor pushing into a buffer resource") {
static constexpr size_t num_items = 79;
auto [rd, wr] = async::make_spsc_buffer_resource<int32_t>(8, 2);
sys.spawn([wr{wr}](event_based_actor* self) {
self->make_observable().repeat(42).take(num_items).subscribe(wr);
});
WHEN("draining the buffer resource and sending its items over a socket") {
auto [fd1, fd2] = unbox(net::make_stream_socket_pair());
if (auto err = nonblocking(fd1, true))
FAIL("nonblocking(fd1) returned an error: " << err);
if (auto err = nonblocking(fd2, true))
FAIL("nonblocking(fd2) returned an error: " << err);
auto app = app_t::make(std::move(rd));
auto& state = *app;
auto transport = net::stream_transport::make(fd1, std::move(app));
auto mgr = net::socket_manager::make(mpx(), fd1, std::move(transport));
if (auto err = mgr->init(content(cfg)))
FAIL("mgr->init() failed: " << err);
THEN("the reader receives all items before the connection closes") {
auto remaining = num_items * sizeof(int32_t);
reader rd{fd2, remaining};
while (!rd.done()) {
if (auto new_val = rd.remaining(); remaining != new_val) {
remaining = new_val;
MESSAGE("want " << remaining << " more bytes");
}
run();
rd.read_some();
}
CHECK_EQ(state.written_values, std::vector<int32_t>(num_items, 42));
CHECK_EQ(state.written_bytes.size(), num_items * sizeof(int32_t));
CHECK_EQ(rd.buf().size(), num_items * sizeof(int32_t));
CHECK_EQ(state.written_bytes, rd.buf());
}
}
}
}
END_FIXTURE_SCOPE()
......@@ -48,8 +48,7 @@ public:
// -- implementation of http::upper_layer ------------------------------------
error init(net::socket_manager*, net::http::lower_layer* down_ptr,
const settings&) override {
error init(net::http::lower_layer* down_ptr, const settings&) override {
down = down_ptr;
down->request_messages();
return none;
......
......@@ -17,7 +17,8 @@
#include "caf/byte_buffer.hpp"
#include "caf/byte_span.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/net/message_oriented.hpp"
#include "caf/net/binary/lower_layer.hpp"
#include "caf/net/binary/upper_layer.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/socket_manager.hpp"
......@@ -35,21 +36,21 @@ using string_list = std::vector<std::string>;
using shared_string_list = std::shared_ptr<string_list>;
template <bool EnableSuspend>
class app_t : public net::message_oriented::upper_layer {
class app_t : public net::binary::upper_layer {
public:
static auto make(shared_string_list inputs) {
return std::make_unique<app_t>(std::move(inputs));
app_t(async::execution_context_ptr loop, shared_string_list ls_ptr)
: loop(std::move(loop)), inputs(std::move(ls_ptr)) {
// nop
}
app_t(shared_string_list ls_ptr) : inputs(std::move(ls_ptr)) {
// nop
static auto make(async::execution_context_ptr loop,
shared_string_list inputs) {
return std::make_unique<app_t>(std::move(loop), std::move(inputs));
}
caf::error init(net::socket_manager* mgr_ptr,
net::message_oriented::lower_layer* down_ptr,
caf::error init(net::binary::lower_layer* down_ptr,
const settings&) override {
// Start reading immediately.
mgr = mgr_ptr;
down = down_ptr;
down->request_messages();
return none;
......@@ -63,12 +64,12 @@ public:
return true;
}
void abort(const error&) override {
// nop
void abort(const error& err) override {
MESSAGE("abort: " << err);
}
void continue_reading() {
mgr->schedule_fn([this] { down->request_messages(); });
loop->schedule_fn([this] { down->request_messages(); });
}
ptrdiff_t consume(byte_span buf) override {
......@@ -97,9 +98,9 @@ public:
}
}
net::socket_manager* mgr = nullptr;
async::execution_context_ptr loop;
net::message_oriented::lower_layer* down = nullptr;
net::binary::lower_layer* down = nullptr;
shared_string_list inputs;
};
......@@ -140,7 +141,7 @@ SCENARIO("length-prefix framing reads data with 32-bit size headers") {
GIVEN("a length_prefix_framing with an app that consumes strings") {
WHEN("pushing data into the unit-under-test") {
auto buf = std::make_shared<string_list>();
auto app = app_t<false>::make(buf);
auto app = app_t<false>::make(nullptr, buf);
auto framing = net::length_prefix_framing::make(std::move(app));
auto uut = mock_stream_transport::make(std::move(framing));
CHECK_EQ(uut->init(), error{});
......@@ -186,7 +187,7 @@ SCENARIO("calling suspend_reading temporarily halts receiving of messages") {
if (auto err = net::nonblocking(fd2, true))
CAF_FAIL("nonblocking returned an error: " << err);
auto buf = std::make_shared<string_list>();
auto app = app_t<true>::make(buf);
auto app = app_t<true>::make(mpx, 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));
......
// 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.producer_adapter
#include "caf/net/producer_adapter.hpp"
#include "net-test.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/net/length_prefix_framing.hpp"
#include "caf/net/message_oriented.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/scheduled_actor/flow.hpp"
using namespace caf;
namespace {
class writer {
public:
explicit writer(net::stream_socket fd) : sg_(fd) {
// nop
}
auto fd() {
return sg_.socket();
}
byte_buffer encode(std::string_view msg) {
using detail::to_network_order;
auto prefix = to_network_order(static_cast<uint32_t>(msg.size()));
auto prefix_bytes = as_bytes(make_span(&prefix, 1));
byte_buffer buf;
buf.insert(buf.end(), prefix_bytes.begin(), prefix_bytes.end());
auto bytes = as_bytes(make_span(msg));
buf.insert(buf.end(), bytes.begin(), bytes.end());
return buf;
}
void write(std::string_view msg) {
auto buf = encode(msg);
if (net::write(fd(), buf) < 0)
FAIL("failed to write: " << net::last_socket_error_as_string());
}
private:
net::socket_guard<net::stream_socket> sg_;
};
class app_t : public net::message_oriented::upper_layer {
public:
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
}
static auto make(resource_type output) {
return std::make_unique<app_t>(std::move(output));
}
error init(net::socket_manager* mgr,
net::message_oriented::lower_layer* down_ptr,
const settings&) override {
down = down_ptr;
down->request_messages();
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->shutdown(); });
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 prepare_send() override {
// nop
}
bool done_sending() override {
return true;
}
void abort(const error& reason) override {
if (reason == caf::sec::socket_disconnected || reason == caf::sec::disposed)
adapter_->close();
else
adapter_->abort(reason);
}
ptrdiff_t consume(byte_span buf) override {
auto val = int32_t{0};
auto str = std::string_view{reinterpret_cast<char*>(buf.data()),
buf.size()};
if (auto err = detail::parse(str, val))
FAIL("unable to parse input: " << err);
++received_messages;
if (auto capacity_left = adapter_->push(val); capacity_left == 0) {
down->suspend_reading();
}
return static_cast<ptrdiff_t>(buf.size());
}
net::message_oriented::lower_layer* down = nullptr;
size_t received_messages = 0;
adapter_ptr adapter_;
resource_type output_;
};
struct fixture : test_coordinator_fixture<> {
fixture() : mm(sys) {
if (auto err = mm.mpx().init())
CAF_FAIL("mpx.init() failed: " << err);
}
bool handle_io_event() override {
mm.mpx().apply_updates();
return mm.mpx().poll_once(false);
}
auto mpx() {
return mm.mpx_ptr();
}
net::middleman mm;
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("publisher adapters suspend reads if the buffer becomes full") {
GIVEN("an actor reading from a buffer resource") {
static constexpr size_t num_items = 13;
std::vector<int32_t> outputs;
auto [rd, wr] = async::make_spsc_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 writer_thread = std::thread{[fd1{fd1}] {
writer out{fd1};
for (size_t i = 0; i < num_items; ++i)
out.write(std::to_string(i));
}};
if (auto err = nonblocking(fd2, true))
FAIL("nonblocking(fd2) returned an error: " << err);
auto app = app_t::make(std::move(wr));
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));
if (auto err = mgr->init(content(cfg)))
FAIL("mgr->init() failed: " << err);
THEN("the actor receives all items from the writer (socket)") {
while (outputs.size() < num_items)
run();
auto ls = [](auto... xs) { return std::vector<int32_t>{xs...}; };
CHECK_EQ(outputs, ls(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12));
}
writer_thread.join();
}
}
}
END_FIXTURE_SCOPE()
......@@ -59,7 +59,7 @@ public:
*done_ = true;
}
error init(socket_manager*, net::stream_oriented::lower_layer* down,
error init(net::stream_oriented::lower_layer* down,
const settings&) override {
MESSAGE("initialize dummy app");
down_ = down;
......
......@@ -75,7 +75,7 @@ public:
net::stream_oriented::lower_layer* down;
error init(net::socket_manager*, net::stream_oriented::lower_layer* down_ptr,
error init(net::stream_oriented::lower_layer* down_ptr,
const settings&) override {
down = down_ptr;
down->configure_read(net::receive_policy::exactly(hello_manager.size()));
......
......@@ -11,6 +11,7 @@
#include "caf/byte.hpp"
#include "caf/byte_span.hpp"
#include "caf/callback.hpp"
#include "caf/net/make_actor_shell.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/socket_manager.hpp"
......@@ -29,24 +30,20 @@ using string_consumer = typed_actor<result<void>(std::string)>;
class app_t : public net::stream_oriented::upper_layer {
public:
app_t() = default;
explicit app_t(actor hdl) : worker(std::move(hdl)) {
// nop
explicit app_t(actor_system& sys, async::execution_context_ptr loop,
actor hdl = {})
: worker(std::move(hdl)) {
self = net::make_actor_shell<string_consumer>(sys, loop);
}
static auto make() {
return std::make_unique<app_t>();
static auto make(actor_system& sys, async::execution_context_ptr loop,
actor hdl = {}) {
return std::make_unique<app_t>(sys, std::move(loop), std::move(hdl));
}
static auto make(actor hdl) {
return std::make_unique<app_t>(std::move(hdl));
}
error init(net::socket_manager* mgr, net::stream_oriented::lower_layer* down,
error init(net::stream_oriented::lower_layer* down,
const settings&) override {
this->down = down;
self = mgr->make_actor_shell<string_consumer>();
self->set_behavior([this](std::string& line) {
CAF_MESSAGE("received an asynchronous message: " << line);
lines.emplace_back(std::move(line));
......@@ -60,12 +57,11 @@ public:
}
void prepare_send() override {
while (!self->terminated() && self->consume_message())
; // repeat
// nop
}
bool done_sending() override {
return self->try_block_mailbox();
return true;
}
void abort(const error&) override {
......@@ -133,7 +129,7 @@ public:
std::vector<std::string> lines;
// Actor shell representing this app.
net::typed_actor_shell_ptr_t<string_consumer> self;
net::actor_shell_ptr_t<string_consumer> self;
// Counts how many bytes we've consumed in total.
size_t consumed_bytes = 0;
......@@ -197,7 +193,7 @@ CAF_TEST_FIXTURE_SCOPE(actor_shell_tests, fixture)
CAF_TEST(actor shells expose their mailbox to their owners) {
auto fd = testee_socket_guard.release();
auto app_uptr = app_t::make();
auto app_uptr = app_t::make(sys, mpx);
auto app = app_uptr.get();
auto transport = net::stream_transport::make(fd, std::move(app_uptr));
auto mgr = net::socket_manager::make(mpx.get(), fd, std::move(transport));
......@@ -209,6 +205,8 @@ CAF_TEST(actor shells expose their mailbox to their owners) {
anon_send(hdl, "line 3");
run_while([&] { return app->lines.size() != 3; });
CAF_CHECK_EQUAL(app->lines, svec({"line 1", "line 2", "line 3"}));
self_socket_guard.reset();
run_while([&] { return mpx->num_socket_managers() > 1; });
}
CAF_TEST(actor shells can send requests and receive responses) {
......@@ -218,7 +216,7 @@ CAF_TEST(actor shells can send requests and receive responses) {
};
});
auto fd = testee_socket_guard.release();
auto app_uptr = app_t::make(worker);
auto app_uptr = app_t::make(sys, mpx, worker);
auto app = app_uptr.get();
auto transport = net::stream_transport::make(fd, std::move(app_uptr));
auto mgr = net::socket_manager::make(mpx.get(), fd, std::move(transport));
......@@ -232,6 +230,8 @@ CAF_TEST(actor shells can send requests and receive responses) {
std::string_view received_response{reinterpret_cast<char*>(recv_buf.data()),
recv_buf.size()};
CAF_CHECK_EQUAL(received_response, expected_response);
self_socket_guard.reset();
run_while([&] { return mpx->num_socket_managers() > 1; });
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -26,8 +26,7 @@ public:
settings cfg;
error init(net::socket_manager*, net::web_socket::lower_layer*,
const settings& init_cfg) override {
error init(net::web_socket::lower_layer*, const settings& init_cfg) override {
cfg = init_cfg;
return none;
}
......
......@@ -33,8 +33,7 @@ public:
return std::make_unique<app_t>();
}
error init(net::socket_manager*, net::web_socket::lower_layer*,
const settings& init_cfg) override {
error init(net::web_socket::lower_layer*, const settings& init_cfg) override {
cfg = init_cfg;
return none;
}
......
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