Commit 823cfbea authored by Dominik Charousset's avatar Dominik Charousset

Iterate on the caf-net design, add Prometheus

parent 1588071c
......@@ -139,5 +139,9 @@ else()
endfunction()
endif()
add_net_example(web_socket quote-server)
add_net_example(http time-server)
add_net_example(web_socket echo)
add_net_example(web_socket hello-client)
add_net_example(web_socket quote-server)
add_net_example(web_socket stock-ticker)
// Simple WebSocket server that sends everything it receives back to the sender.
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/caf_main.hpp"
#include "caf/deep_to_string.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/http/serve.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <iostream>
#include <utility>
static constexpr uint16_t default_port = 8080;
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<uint16_t>("port,p", "port to listen for incoming connections");
}
};
int caf_main(caf::actor_system& sys, const config& cfg) {
using namespace caf;
// Open up a TCP port for incoming connections.
auto port = caf::get_or(cfg, "port", default_port);
net::tcp_accept_socket fd;
if (auto maybe_fd = net::make_tcp_accept_socket({ipv4_address{}, port},
true)) {
std::cout << "*** started listening for incoming connections on port "
<< port << '\n';
fd = std::move(*maybe_fd);
} else {
std::cerr << "*** unable to open port " << port << ": "
<< to_string(maybe_fd.error()) << '\n';
return EXIT_FAILURE;
}
// Create buffers to signal events from the WebSocket server to the worker.
auto [worker_pull, server_push] = net::http::make_request_resource();
// Spin up the HTTP server.
net::http::serve(sys, fd, std::move(server_push));
// Spin up a worker to handle the HTTP requests.
auto worker = sys.spawn([wres = worker_pull](event_based_actor* self) {
// For each incoming request ...
wres
.observe_on(self) //
.for_each([](const net::http::request& req) {
// ... we simply return the current time as string.
// Note: we cannot respond more than once to a request.
auto str = caf::deep_to_string(make_timestamp());
req.respond(net::http::status::ok, "text/plain", str);
});
});
sys.await_all_actors_done();
return EXIT_SUCCESS;
}
CAF_MAIN(caf::net::middleman)
......@@ -50,8 +50,7 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
.for_each([self](const event_t& event) {
// ... that simply pushes data back to the sender.
auto [pull, push] = event.data();
self->make_observable()
.from_resource(pull)
pull.observe_on(self)
.do_on_next([](const ws::frame& x) {
if (x.is_binary()) {
std::cout << "*** received a binary WebSocket frame of size "
......
// Simple WebSocket server that sends everything it receives back to the sender.
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/net/web_socket/connect.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <iostream>
#include <utility>
static constexpr uint16_t default_port = 8080;
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<std::string>("host", "TCP endpoint to connect to")
.add<uint16_t>("port,p", "port for connecting to the host")
.add<std::string>("endpoint", "sets the Request-URI field")
.add<std::string>("protocols", "sets the Sec-WebSocket-Protocol field")
.add<size_t>("max,m", "maximum number of message to receive");
}
};
int caf_main(caf::actor_system& sys, const config& cfg) {
namespace cn = caf::net;
namespace ws = cn::web_socket;
// Sanity checking.
auto host = caf::get_or(cfg, "host", "");
if (host.empty()) {
std::cerr << "*** missing mandatory argument: host\n";
return EXIT_FAILURE;
}
// Ask user for the hello message.
std::string hello;
std::cout << "Please enter a hello message for the server: " << std::flush;
std::getline(std::cin, hello);
// Open up the TCP connection.
auto port = caf::get_or(cfg, "port", default_port);
cn::tcp_stream_socket fd;
if (auto maybe_fd = cn::make_connected_tcp_stream_socket(host, port)) {
std::cout << "*** connected to " << host << ':' << port << '\n';
fd = std::move(*maybe_fd);
} else {
std::cerr << "*** unable to connect to " << host << ':' << port << ": "
<< to_string(maybe_fd.error()) << '\n';
return EXIT_FAILURE;
}
// Spin up the WebSocket.
ws::handshake hs;
hs.host(host);
hs.endpoint(caf::get_or(cfg, "endpoint", "/"));
if (auto str = caf::get_as<std::string>(cfg, "protocols");
str && !str->empty())
hs.protocols(std::move(*str));
ws::connect(sys, fd, std::move(hs), [&](const ws::connect_event_t& conn) {
sys.spawn([conn, hello](caf::event_based_actor* self) {
auto [pull, push] = conn.data();
// Print everything from the server to stdout.
pull.observe_on(self)
.do_on_error([](const caf::error& what) {
std::cerr << "*** error while reading from the WebSocket: "
<< to_string(what) << '\n';
})
.compose([self](auto in) {
if (auto limit = caf::get_as<size_t>(self->config(), "max")) {
return std::move(in).take(*limit).as_observable();
} else {
return std::move(in).as_observable();
}
})
.for_each([](const ws::frame& msg) {
if (msg.is_text()) {
std::cout << "Server: " << msg.as_text() << '\n';
} else if (msg.is_binary()) {
std::cout << "Server: [binary message of size "
<< msg.as_binary().size() << "]\n";
}
});
// Send our hello message and wait until the server closes the socket.
self->make_observable()
.just(ws::frame{hello})
.concat(self->make_observable().never<ws::frame>())
.subscribe(push);
});
});
return EXIT_SUCCESS;
}
CAF_MAIN(caf::net::middleman)
......@@ -102,7 +102,8 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
// Open up a TCP port for incoming connections.
auto port = caf::get_or(cfg, "port", default_port);
cn::tcp_accept_socket fd;
if (auto maybe_fd = cn::make_tcp_accept_socket({caf::ipv4_address{}, port})) {
if (auto maybe_fd = cn::make_tcp_accept_socket({caf::ipv4_address{}, port},
true)) {
std::cout << "*** started listening for incoming connections on port "
<< port << '\n';
fd = std::move(*maybe_fd);
......@@ -127,7 +128,8 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
auto quote = quotes.empty() ? "Try /epictetus, /seneca or /plato."
: f(quotes);
self->make_observable().just(ws::frame{quote}).subscribe(push);
// Note: we simply drop `pull` here, which will close the buffer.
// We ignore whatever the client may send to us.
pull.observe_on(self).subscribe(std::ignore);
});
});
// Callback for incoming WebSocket requests.
......
// Pseudo "Stock Ticker" that publishes random updates once per second via
// WebSocket feed.
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/caf_main.hpp"
#include "caf/cow_string.hpp"
#include "caf/cow_tuple.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/json_writer.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/net/web_socket/accept.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include "caf/span.hpp"
#include <cassert>
#include <chrono>
#include <cstdint>
#include <iostream>
#include <utility>
using namespace std::literals;
static constexpr uint16_t default_port = 8080;
namespace stock {
struct info {
std::string symbol;
std::string currency;
double current;
double open;
double high;
double low;
};
template <class Inspector>
bool inspect(Inspector& f, info& x) {
return f.object(x).fields(f.field("symbol", x.symbol),
f.field("currency", x.currency),
f.field("open", x.open), f.field("high", x.high),
f.field("low", x.low));
}
} // namespace stock
struct random_feed_state {
using frame = caf::net::web_socket::frame;
using accept_event = caf::net::web_socket::accept_event_t<>;
random_feed_state(caf::event_based_actor* selfptr,
caf::async::consumer_resource<accept_event> events,
caf::timespan update_interval)
: self(selfptr), val_dist(0, 100000), index_dist(0, 19) {
// Init random number generator.
std::random_device rd;
rng.seed(rd());
// Fill vector with some stuff.
for (size_t i = 0; i < 20; ++i) {
std::uniform_int_distribution<int> char_dist{'A', 'Z'};
std::string symbol;
for (size_t j = 0; j < 5; ++j)
symbol += static_cast<char>(char_dist(rng));
auto val = next_value();
infos.emplace_back(
stock::info{std::move(symbol), "USD", val, val, val, val});
}
// Create the feed to push updates once per second.
writer.skip_object_type_annotation(true);
feed = self->make_observable()
.interval(update_interval)
.map([this](int64_t) {
writer.reset();
auto& x = update();
if (!writer.apply(x)) {
std::cerr << "*** failed to generate JSON: "
<< to_string(writer.get_error()) << '\n';
return frame{};
}
return frame{writer.str()};
})
.filter([](const frame& x) {
// Just in case: drop frames that failed to generate JSON.
return x.is_text();
})
.share();
// Subscribe once to start the feed immediately and to keep it running.
feed.for_each([n = 1](const frame&) mutable {
std::cout << "*** tick " << n++ << "\n";
});
// Add each incoming WebSocket listener to the feed.
auto n = std::make_shared<int>(0);
events
.observe_on(self) //
.for_each([this, n](const accept_event& ev) {
std::cout << "*** added listener (n = " << ++*n << ")\n";
auto [pull, push] = ev.data();
pull.observe_on(self)
.do_finally([n] { //
std::cout << "*** removed listener (n = " << --*n << ")\n";
})
.subscribe(std::ignore);
feed.subscribe(push);
});
}
// Picks a random stock, assigns a new value to it, and returns it.
stock::info& update() {
auto& x = infos[index_dist(rng)];
auto val = next_value();
x.current = val;
x.high = std::max(x.high, val);
x.low = std::min(x.high, val);
return x;
}
double next_value() {
return val_dist(rng) / 100.0;
}
caf::event_based_actor* self;
caf::flow::observable<frame> feed;
caf::json_writer writer;
std::vector<stock::info> infos;
std::minstd_rand rng;
std::uniform_int_distribution<int> val_dist;
std::uniform_int_distribution<size_t> index_dist;
};
using random_feed_impl = caf::stateful_actor<random_feed_state>;
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<uint16_t>("port,p", "port to listen for incoming connections")
.add<caf::timespan>("interval,i", "update interval");
}
};
int caf_main(caf::actor_system& sys, const config& cfg) {
namespace cn = caf::net;
namespace ws = cn::web_socket;
// Open up a TCP port for incoming connections.
auto port = caf::get_or(cfg, "port", default_port);
cn::tcp_accept_socket fd;
if (auto maybe_fd = cn::make_tcp_accept_socket({caf::ipv4_address{}, port},
true)) {
std::cout << "*** started listening for incoming connections on port "
<< port << '\n';
fd = std::move(*maybe_fd);
} else {
std::cerr << "*** unable to open port " << port << ": "
<< to_string(maybe_fd.error()) << '\n';
return EXIT_FAILURE;
}
// Create buffers to signal events from the WebSocket server to the worker.
auto [wres, sres] = ws::make_accept_event_resources();
// Spin up a worker to handle the events.
auto interval = caf::get_or(cfg, "interval", caf::timespan{1s});
auto worker = sys.spawn<random_feed_impl>(std::move(wres), interval);
// Callback for incoming WebSocket requests.
auto on_request = [](const caf::settings&, auto& req) { req.accept(); };
// Set everything in motion.
ws::accept(sys, fd, std::move(sres), on_request);
return EXIT_SUCCESS;
}
CAF_MAIN(caf::net::middleman)
......@@ -224,6 +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
......
......@@ -329,7 +329,7 @@ struct resource_ctrl : ref_counted {
~resource_ctrl() {
if (buf) {
if constexpr (IsProducer) {
auto err = make_error(sec::invalid_upstream,
auto err = make_error(sec::disposed,
"producer_resource destroyed without opening it");
buf->abort(err);
} else {
......@@ -397,11 +397,19 @@ public:
}
}
/// Convenience function for calling
/// `ctx->make_observable().from_resource(*this)`.
template <class Coordinator>
auto observe_on(Coordinator* ctx) const {
return ctx->make_observable().from_resource(*this);
}
/// Calls `try_open` and on success immediately calls `cancel` on the buffer.
void cancel() {
if (auto buf = try_open())
buf->cancel();
}
explicit operator bool() const noexcept {
return ctrl_ != nullptr;
}
......@@ -454,6 +462,12 @@ public:
}
}
/// Calls `try_open` and on success immediately calls `close` on the buffer.
void close() {
if (auto buf = try_open())
buf->close();
}
explicit operator bool() const noexcept {
return ctrl_ != nullptr;
}
......
......@@ -17,6 +17,23 @@
// -- hard-coded default values for various CAF options ------------------------
namespace caf::defaults {
/// Stores the name of a parameter along with the fallback value.
template <class T>
struct parameter {
std::string_view name;
T fallback;
};
/// @relates parameter
template <class T>
constexpr parameter<T> make_parameter(std::string_view name, T fallback) {
return {name, fallback};
}
} // namespace caf::defaults
namespace caf::defaults::stream {
constexpr auto max_batch_delay = timespan{1'000'000};
......@@ -108,12 +125,12 @@ constexpr auto format = std::string_view{"[%c:%p] %d %m"};
namespace caf::defaults::middleman {
constexpr auto app_identifier = std::string_view{"generic-caf-app"};
constexpr auto network_backend = std::string_view{"default"};
constexpr auto max_consecutive_reads = size_t{50};
constexpr auto heartbeat_interval = timespan{10'000'000'000};
constexpr auto connection_timeout = timespan{30'000'000'000};
constexpr auto cached_udp_buffers = size_t{10};
constexpr auto connection_timeout = timespan{30'000'000'000};
constexpr auto heartbeat_interval = timespan{10'000'000'000};
constexpr auto max_consecutive_reads = size_t{50};
constexpr auto max_pending_msgs = size_t{10};
constexpr auto network_backend = std::string_view{"default"};
} // namespace caf::defaults::middleman
......@@ -124,3 +141,12 @@ constexpr auto batch_size = size_t{32};
constexpr auto buffer_size = size_t{128};
} // namespace caf::defaults::flow
namespace caf::defaults::net {
/// Configures how many concurrent connections an acceptor allows. When reaching
/// this limit, the connector stops accepting additional connections until a
/// previous connection has been closed.
constexpr auto max_connections = make_parameter("max-connections", size_t{64});
} // namespace caf::defaults::net
......@@ -131,6 +131,12 @@ public:
return pimpl_.compare(other.pimpl_);
}
// -- utility ----------------------------------------------------------------
/// Erases each `x` from `xs` where `x.disposed()`.
/// @returns The number of erased elements.
static size_t erase_disposed(std::vector<disposable>& xs);
private:
intrusive_ptr<impl> pimpl_;
};
......
......@@ -203,14 +203,13 @@ public:
bool value(span<std::byte> x) override;
/// @private
std::string current_field_name();
private:
[[nodiscard]] position pos() const noexcept;
void append_current_field_name(std::string& str);
std::string current_field_name();
std::string mandatory_field_missing_str(std::string_view name);
template <bool PopOrAdvanceOnSuccess, class F>
......
......@@ -7,6 +7,7 @@
#include <string_view>
#include "caf/config_value.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/move_if_not_ptr.hpp"
#include "caf/dictionary.hpp"
......@@ -68,6 +69,12 @@ auto get_or(const settings& xs, std::string_view name, Fallback&& fallback) {
}
}
/// Convenience overload for calling `get_or(xs, param.name, param.fallback)`.
template <class T>
auto get_or(const settings& xs, defaults::parameter<T> param) {
return get_or(xs, param.name, param.fallback);
}
/// Tries to retrieve the value associated to `name` from `xs` as an instance of
/// type `T`.
/// @relates actor_system_config
......
......@@ -31,6 +31,7 @@ private:
telemetry::int_gauge* rss_ = nullptr;
telemetry::int_gauge* vms_ = nullptr;
telemetry::dbl_gauge* cpu_ = nullptr;
telemetry::int_gauge* fds_ = nullptr;
};
} // namespace caf::telemetry::importer
......@@ -238,100 +238,27 @@ std::ostream& operator<<(std::ostream& out, indentation indent) {
return out;
}
// Fakes a buffer interface but really prints to std::cout.
struct out_buf {
int end() {
return 0;
}
void push_back(char c) {
std::cout << c;
}
template <class Iterator, class Sentinel>
void insert(int, Iterator i, Sentinel e) {
while (i != e)
std::cout << *i++;
}
};
void print(const config_value::dictionary& xs, indentation indent);
void print_val(const config_value& val, indentation indent) {
out_buf out;
using std::cout;
switch (val.get_data().index()) {
default: // none
// omit
break;
case 1: // integer
detail::print(out, get<config_value::integer>(val));
break;
case 2: // boolean
detail::print(out, get<config_value::boolean>(val));
break;
case 3: // real
detail::print(out, get<config_value::real>(val));
break;
case 4: // timespan
detail::print(out, get<timespan>(val));
break;
case 5: // uri
cout << '<' << get<uri>(val).str() << '>';
break;
case 6: // string
detail::print_escaped(out, get<std::string>(val));
break;
case 7: { // list
auto& xs = get<config_value::list>(val);
if (xs.empty()) {
cout << "[]";
} else {
auto list_indent = indent + 2;
cout << "[\n";
for (auto& x : xs) {
cout << list_indent;
print_val(x, list_indent);
cout << ",\n";
}
cout << indent << ']';
}
break;
}
case 8: { // dictionary
print(get<config_value::dictionary>(val), indent + 2);
}
}
}
bool needs_quotes(const std::string& key) {
auto is_alnum_or_dash = [](char x) {
return isalnum(x) || x == '-' || x == '_';
};
return key.empty() || !std::all_of(key.begin(), key.end(), is_alnum_or_dash);
}
void print(const config_value::dictionary& xs, indentation indent) {
out_buf out;
using std::cout;
bool top_level = indent.size == 0;
for (const auto& [key, val] : xs) {
if (!top_level || (key != "dump-config" && key != "config-file")) {
cout << indent;
if (!needs_quotes(key)) {
cout << key;
} else {
detail::print_escaped(out, key);
}
if (!holds_alternative<config_value::dictionary>(val)) {
cout << " = ";
print_val(val, indent);
cout << '\n';
} else if (auto xs = get<config_value::dictionary>(val); xs.empty()) {
cout << "{}\n";
for (const auto& kvp : xs) {
if (kvp.first == "dump-config")
continue;
if (auto submap = get_if<config_value::dictionary>(&kvp.second)) {
cout << indent << kvp.first << " {\n";
print(*submap, indent + 2);
cout << indent << "}\n";
} else if (auto lst = get_if<config_value::list>(&kvp.second)) {
if (lst->empty()) {
cout << indent << kvp.first << " = []\n";
} else {
cout << " {\n";
print(get<config_value::dictionary>(val), indent + 2);
cout << indent << "}\n";
cout << indent << kvp.first << " = [\n";
auto list_indent = indent + 2;
for (auto& x : *lst)
cout << list_indent << to_string(x) << ",\n";
cout << indent << "]\n";
}
} else {
cout << indent << kvp.first << " = " << to_string(kvp.second) << '\n';
}
}
}
......
......@@ -10,6 +10,18 @@
namespace caf {
// -- member types -------------------------------------------------------------
disposable::impl::~impl() {
// nop
}
disposable disposable::impl::as_disposable() noexcept {
return disposable{intrusive_ptr<impl>{this}};
}
// -- factories ----------------------------------------------------------------
namespace {
class composite_impl : public ref_counted, public disposable::impl {
......@@ -52,14 +64,6 @@ private:
} // namespace
disposable::impl::~impl() {
// nop
}
disposable disposable::impl::as_disposable() noexcept {
return disposable{intrusive_ptr<impl>{this}};
}
disposable disposable::make_composite(std::vector<disposable> entries) {
if (entries.empty())
return {};
......@@ -67,4 +71,18 @@ disposable disposable::make_composite(std::vector<disposable> entries) {
return disposable{make_counted<composite_impl>(std::move(entries))};
}
// -- utility ------------------------------------------------------------------
size_t disposable::erase_disposed(std::vector<disposable>& xs) {
auto is_disposed = [](auto& hdl) { return hdl.disposed(); };
auto xs_end = xs.end();
if (auto e = std::remove_if(xs.begin(), xs_end, is_disposed); e != xs_end) {
auto res = std::distance(e, xs_end);
xs.erase(e, xs_end);
return static_cast<size_t>(res);
} else {
return 0;
}
}
} // namespace caf
......@@ -634,15 +634,10 @@ json_reader::position json_reader::pos() const noexcept {
}
void json_reader::append_current_field_name(std::string& result) {
if (field_.empty()) {
result += "null";
} else {
auto i = field_.begin();
result += *i++;
while (i != field_.end()) {
result += '.';
result += *i++;
}
result += "ROOT";
for (auto& key : field_) {
result += '.';
result.insert(result.end(), key.begin(), key.end());
}
}
......
......@@ -752,12 +752,9 @@ void scheduled_actor::run_actions() {
void scheduled_actor::update_watched_disposables() {
CAF_LOG_TRACE("");
auto disposed = [](auto& hdl) { return hdl.disposed(); };
auto& xs = watched_disposables_;
if (auto e = std::remove_if(xs.begin(), xs.end(), disposed); e != xs.end()) {
xs.erase(e, xs.end());
CAF_LOG_DEBUG("now watching" << xs.size() << "disposables");
}
[[maybe_unused]] auto n = disposable::erase_disposed(watched_disposables_);
CAF_LOG_DEBUG_IF(n > 0, "now watching" << watched_disposables_.size()
<< "disposables");
}
} // namespace caf
......@@ -21,18 +21,19 @@
namespace {
struct sys_stats {
int64_t rss;
int64_t vms;
double cpu_time;
int64_t rss = 0;
int64_t vms = 0;
double cpu_time = 0.0;
int64_t fds = 0;
};
sys_stats read_sys_stats();
static constexpr bool platform_supported_v = true;
constexpr bool platform_supported_v = true;
template <class CpuPtr, class RssPtr, class VmsPtr>
template <class CpuPtr, class RssPtr, class VmsPtr, class FdsPtr>
void sys_stats_init(caf::telemetry::metric_registry& reg, RssPtr& rss,
VmsPtr& vms, CpuPtr& cpu) {
VmsPtr& vms, CpuPtr& cpu, FdsPtr& fds) {
rss = reg.gauge_singleton("process", "resident_memory",
"Resident memory size.", "bytes");
vms = reg.gauge_singleton("process", "virtual_memory", "Virtual memory size.",
......@@ -40,15 +41,19 @@ void sys_stats_init(caf::telemetry::metric_registry& reg, RssPtr& rss,
cpu = reg.gauge_singleton<double>("process", "cpu",
"Total user and system CPU time spent.",
"seconds", true);
fds = reg.gauge_singleton("process", "open_fds",
"Number of open file descriptors.");
}
void update_impl(caf::telemetry::int_gauge* rss_gauge,
caf::telemetry::int_gauge* vms_gauge,
caf::telemetry::dbl_gauge* cpu_gauge) {
auto [rss, vmsize, cpu] = read_sys_stats();
caf::telemetry::dbl_gauge* cpu_gauge,
caf::telemetry::int_gauge* fds_gauge) {
auto [rss, vmsize, cpu, fds] = read_sys_stats();
rss_gauge->value(rss);
vms_gauge->value(vmsize);
cpu_gauge->value(cpu);
fds_gauge->value(fds);
}
} // namespace
......@@ -57,7 +62,7 @@ void update_impl(caf::telemetry::int_gauge* rss_gauge,
namespace {
static constexpr bool platform_supported_v = false;
constexpr bool platform_supported_v = false;
template <class... Ts>
void sys_stats_init(Ts&&...) {
......@@ -77,14 +82,16 @@ void update_impl(Ts&&...) {
#ifdef CAF_MACOS
# include <libproc.h>
# include <mach/mach.h>
# include <mach/task.h>
# include <sys/resource.h>
# include <unistd.h>
namespace {
sys_stats read_sys_stats() {
sys_stats result{0, 0, 0};
sys_stats result;
// Fetch memory usage.
{
mach_task_basic_info info;
......@@ -110,6 +117,20 @@ sys_stats read_sys_stats() {
result.cpu_time += ceil(info.system_time.microseconds / 1000.0) / 1000.0;
}
}
// Fetch open file handles.
{
// proc_pidinfo is undocumented, but this is what lsof also uses.
auto suggested_buf_size = proc_pidinfo(getpid(), PROC_PIDLISTFDS, 0,
nullptr, 0);
if (suggested_buf_size > 0) {
auto buf_size = suggested_buf_size;
auto buf = malloc(buf_size); // TODO: could be thread-local
auto res = proc_pidinfo(getpid(), PROC_PIDLISTFDS, 0, buf, buf_size);
free(buf);
if (res > 0)
result.fds = res / sizeof(proc_fdinfo);
}
}
return result;
}
......@@ -121,8 +142,22 @@ sys_stats read_sys_stats() {
#if defined(CAF_LINUX) || defined(CAF_NET_BSD)
# include <dirent.h>
# include <unistd.h>
int64_t count_entries_in_directory(const char* path) {
int64_t result = 0;
if (auto dptr = opendir(path); dptr != nullptr) {
for (auto entry = readdir(dptr); entry != nullptr; entry = readdir(dptr)) {
auto fname = entry->d_name;
if (strcmp(".", fname) != 0 && strcmp("..", fname) != 0)
++result;
}
closedir(dptr);
}
return result;
}
/// Caches the result from a `sysconf` call in a cache variable to avoid
/// frequent syscalls. Sets `cache_var` to -1 in case of an error. Initially,
/// `cache_var` must be 0 and we assume a successful syscall would always return
......@@ -160,6 +195,7 @@ bool load_system_setting(std::atomic<long>& cache_var, long& var, int name,
#ifdef CAF_LINUX
# include <cstdio>
# include <dirent.h>
namespace {
......@@ -168,7 +204,7 @@ std::atomic<long> global_ticks_per_second;
std::atomic<long> global_page_size;
sys_stats read_sys_stats() {
sys_stats result{0, 0, 0};
sys_stats result;
long ticks_per_second = 0;
long page_size = 0;
if (!TRY_LOAD(ticks_per_second, _SC_CLK_TCK)
......@@ -218,6 +254,7 @@ sys_stats read_sys_stats() {
result.cpu_time += stime_ticks;
result.cpu_time /= ticks_per_second;
}
result.fds = count_entries_in_directory("/proc/self/fd");
return result;
}
......@@ -234,7 +271,7 @@ namespace {
std::atomic<long> global_page_size;
sys_stats read_sys_stats() {
auto result = sys_stats{0, 0, 0};
auto result = sys_stats;
auto kip2 = kinfo_proc2{};
auto kip2_size = sizeof(kip2);
int mib[6] = {
......@@ -263,7 +300,7 @@ sys_stats read_sys_stats() {
namespace caf::telemetry::importer {
process::process(metric_registry& reg) {
sys_stats_init(reg, rss_, vms_, cpu_);
sys_stats_init(reg, rss_, vms_, cpu_, fds_);
}
bool process::platform_supported() noexcept {
......@@ -271,7 +308,7 @@ bool process::platform_supported() noexcept {
}
void process::update() {
update_impl(rss_, vms_, cpu_);
update_impl(rss_, vms_, cpu_, fds_);
}
} // namespace caf::telemetry::importer
// 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.blocking_producer
#include "caf/async/blocking_producer.hpp"
#include "core-test.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>;
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");
}
async::blocking_producer out{std::move(buf)};
for (auto i = begin; i != end; ++i)
out.push(i);
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");
}
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("blocking producers allow threads to generate data") {
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>();
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()
// 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()
// 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.publishing_queue
#include "caf/async/publishing_queue.hpp"
#include "core-test.hpp"
#include "caf/scheduled_actor/flow.hpp"
using namespace caf;
namespace {
struct fixture {
actor_system_config cfg;
actor_system sys;
fixture() : sys(cfg.set("caf.scheduler.max-threads", 2)) {
// nop
}
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("publishing queues connect asynchronous producers to observers") {
GIVEN("a producer and a consumer, living in separate threads") {
WHEN("connecting producer and consumer via a publishing queue") {
THEN("the consumer receives all produced values in order") {
auto [queue, src] = async::make_publishing_queue<int>(sys, 100);
auto producer_thread = std::thread{[q{std::move(queue)}] {
for (int i = 0; i < 5000; ++i)
q->push(i);
}};
std::vector<int> values;
auto consumer_thread = std::thread{[&values, src{src}] {
src.blocking_for_each([&](int x) { values.emplace_back(x); });
}};
producer_thread.join();
consumer_thread.join();
std::vector<int> expected_values;
expected_values.resize(5000);
std::iota(expected_values.begin(), expected_values.end(), 0);
CHECK_EQ(values, expected_values);
}
}
}
}
END_FIXTURE_SCOPE()
#include "caf/cow_vector.hpp"
#include "caf/fwd.hpp"
#include "caf/result.hpp"
#include "caf/test/bdd_dsl.hpp"
......@@ -418,6 +419,7 @@ bool inspect(Inspector& f, phone_book& x) {
CAF_BEGIN_TYPE_ID_BLOCK(core_test, caf::first_custom_type_id)
ADD_TYPE_ID((caf::cow_vector<int>) )
ADD_TYPE_ID((circle))
ADD_TYPE_ID((dummy_enum))
ADD_TYPE_ID((dummy_enum_class))
......
......@@ -124,64 +124,4 @@ CAF_TEST(json baselines) {
}
}
constexpr std::string_view foo_json = R"_({
"@first-type": "int32_t",
"first": 42,
"top": {
"mid": {
"bottom": null
}
}
})_";
SCENARIO("the json reader keeps track of the current field") {
GIVEN("a sequence of member functions calls for type inspection") {
WHEN("encountering new fields") {
THEN("the json reader updates the current field name") {
auto dummy = int32_t{0};
auto first_present = false;
auto types_vec = std::vector{type_id_v<int32_t>, type_id_v<double>};
auto types = make_span(types_vec);
auto first_index = size_t{0};
auto top_present = false;
auto bottom_present = false;
auto bottom_index = size_t{0};
json_reader uut;
CHECK(uut.load(foo_json));
CHECK_EQ(uut.current_field_name(), "null");
// clang-format off
uut.begin_object(invalid_type_id, "foo");
// begin_field overload #4
uut.begin_field("first", first_present, types, first_index);
CHECK_EQ(first_present, true);
CHECK_EQ(first_index, 0u);
CHECK_EQ(uut.current_field_name(), "first");
uut.value(dummy);
CHECK_EQ(dummy, 42);
uut.end_field();
// begin_field overload #2
uut.begin_field("top", top_present);
CHECK_EQ(top_present, true);
CHECK_EQ(uut.current_field_name(), "top");
uut.begin_object(invalid_type_id, "top");
// begin_field overload #1
uut.begin_field("mid");
CHECK_EQ(uut.current_field_name(), "top.mid");
uut.begin_object(invalid_type_id, "bottom");
// begin_field overload #3
uut.begin_field("bottom", bottom_present, types, bottom_index);
CHECK_EQ(bottom_present, false);
CHECK_EQ(uut.current_field_name(), "top.mid.bottom");
uut.end_field();
uut.end_object();
uut.end_field();
uut.end_object();
uut.end_field();
uut.end_object();
// clang-format on
}
}
}
}
END_FIXTURE_SCOPE()
......@@ -421,14 +421,17 @@ strong_actor_ptr middleman::remote_lookup(std::string name,
void middleman::start() {
CAF_LOG_TRACE("");
// Launch background tasks.
if (auto prom = get_if<config_value::dictionary>(
&system().config(), "caf.middleman.prometheus-http")) {
auto ptr = std::make_unique<prometheus_scraping>(system());
if (auto port = ptr->start(*prom)) {
CAF_ASSERT(*port != 0);
prometheus_scraping_port_ = *port;
background_tasks_.emplace_back(std::move(ptr));
// Launch background tasks unless caf-net is also available. In that case, the
// net::middleman takes care of these.
if (!system().has_network_manager()) {
if (auto prom = get_if<config_value::dictionary>(
&system().config(), "caf.middleman.prometheus-http")) {
auto ptr = std::make_unique<prometheus_scraping>(system());
if (auto port = ptr->start(*prom)) {
CAF_ASSERT(*port != 0);
prometheus_scraping_port_ = *port;
background_tasks_.emplace_back(std::move(ptr));
}
}
}
// Launch backend.
......
......@@ -33,24 +33,31 @@ caf_add_component(
src/detail/rfc6455.cpp
src/net/abstract_actor_shell.cpp
src/net/actor_shell.cpp
src/net/binary/default_trait.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
src/net/http/header.cpp
src/net/http/lower_layer.cpp
src/net/http/method.cpp
src/net/http/request.cpp
src/net/http/response.cpp
src/net/http/serve.cpp
src/net/http/server.cpp
src/net/http/status.cpp
src/net/http/upper_layer.cpp
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
src/net/pipe_socket.cpp
src/net/pollset_updater.cpp
src/net/prometheus/server.cpp
src/net/socket.cpp
src/net/socket_event_layer.cpp
src/net/socket_manager.cpp
......@@ -84,7 +91,6 @@ caf_add_component(
detail.rfc6455
net.accept_socket
net.actor_shell
net.consumer_adapter
net.datagram_socket
net.http.server
net.ip
......@@ -93,7 +99,7 @@ caf_add_component(
net.network_socket
net.operation
net.pipe_socket
net.producer_adapter
net.prometheus.server
net.socket
net.socket_guard
net.ssl.transport
......
......@@ -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/byte_span.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/net/binary/fwd.hpp"
#include <string_view>
#include <vector>
namespace caf::net::binary {
class CAF_NET_EXPORT default_trait {
public:
/// The input type of the application, i.e., what that flows from the
/// WebSocket to the application layer.
using input_type = frame;
/// The output type of the application, i.e., what flows from the application
/// layer to the WebSocket.
using output_type = frame;
bool convert(const output_type& x, byte_buffer& bytes);
bool convert(const_byte_span bytes, input_type& x);
error last_error();
};
} // 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_adapter.hpp"
#include "caf/async/producer_adapter.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/fwd.hpp"
#include "caf/net/binary/lower_layer.hpp"
#include "caf/net/binary/upper_layer.hpp"
#include "caf/net/flow_connector.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 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 connector_pointer = flow_connector_ptr<Trait>;
explicit flow_bridge(async::execution_context_ptr loop,
connector_pointer conn)
: loop_(std::move(loop)), conn_(std::move(conn)) {
// nop
}
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) {
down_->begin_message();
auto& bytes = down_->message_buffer();
return trait_.convert(item, bytes) && down_->end_message();
}
bool running() const noexcept {
return in_ || out_;
}
void self_ref(disposable ref) {
self_ref_ = std::move(ref);
}
// -- implementation of binary::lower_layer ----------------------------------
error start(binary::lower_layer* down, const settings& cfg) override {
CAF_ASSERT(down != 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(), loop_, std::move(do_wakeup));
out_ = producer_type::make(push.try_open(), loop_, std::move(do_resume),
std::move(do_cancel));
conn_ = nullptr;
if (in_ && out_) {
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();
self_ref_ = nullptr;
}
ptrdiff_t consume(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());
}
private:
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_;
/// Our event loop.
async::execution_context_ptr loop_;
/// Initializes the bridge. Disposed (set to null) after initializing.
connector_pointer conn_;
/// Type-erased handle to the @ref socket_manager. This reference is important
/// to keep the bridge alive while the manager is not registered for writing
/// or reading.
disposable self_ref_;
};
} // 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
......@@ -4,9 +4,10 @@
#pragma once
namespace caf::net::http {
namespace caf::net::binary {
/// Stores context information for a request. For HTTP/2, this is the stream ID.
class context {};
class frame;
class lower_layer;
class upper_layer;
} // namespace caf::net::http
} // 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();
......@@ -47,6 +21,9 @@ public:
/// Pulls messages from the transport until calling `suspend_reading`.
virtual void request_messages() = 0;
/// Stops reading messages until calling `request_messages`.
virtual void suspend_reading() = 0;
/// Prepares the layer for an outgoing message, e.g., by allocating an
/// output buffer as necessary.
virtual void begin_message() = 0;
......@@ -62,17 +39,6 @@ public:
/// @note When returning `false`, clients must also call
/// `down.set_read_error(...)` with an appropriate error code.
virtual bool end_message() = 0;
/// Inform the remote endpoint that no more messages will arrive.
/// @note Not every protocol has a dedicated close message. Some
/// implementation may simply do nothing.
virtual void send_close_message() = 0;
/// Inform the remote endpoint that no more messages will arrive because of
/// an error.
/// @note Not every protocol has a dedicated close message. Some
/// implementation may simply do nothing.
virtual void send_close_message(const error& reason) = 0;
};
} // namespace caf::net::message_oriented
} // 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 start(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
......@@ -4,6 +4,7 @@
#pragma once
#include "caf/net/connection_factory.hpp"
#include "caf/net/socket_event_layer.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/settings.hpp"
......@@ -12,99 +13,116 @@ namespace caf::net {
/// A connection_acceptor accepts connections from an accept socket and creates
/// socket managers to handle them via its factory.
template <class Socket, class Factory>
template <class Socket>
class connection_acceptor : public socket_event_layer {
public:
// -- member types -----------------------------------------------------------
using socket_type = Socket;
using factory_type = Factory;
using connected_socket_type = typename socket_type::connected_socket_type;
using read_result = typename socket_event_layer::read_result;
using write_result = typename socket_event_layer::write_result;
using factory_type = connection_factory<connected_socket_type>;
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
explicit connection_acceptor(Socket fd, size_t limit, Ts&&... xs)
: fd_(fd), limit_(limit), factory_(std::forward<Ts>(xs)...) {
// nop
template <class FactoryPtr, class... Ts>
connection_acceptor(Socket fd, FactoryPtr fptr, size_t max_connections)
: fd_(fd),
factory_(factory_type::decorate(std::move(fptr))),
max_connections_(max_connections) {
CAF_ASSERT(max_connections_ > 0);
}
~connection_acceptor() {
on_conn_close_.dispose();
if (fd_.id != invalid_socket_id)
close(fd_);
}
// -- factories --------------------------------------------------------------
template <class... Ts>
template <class FactoryPtr, class... Ts>
static std::unique_ptr<connection_acceptor>
make(Socket fd, size_t limit, Ts&&... xs) {
return std::make_unique<connection_acceptor>(fd, limit,
std::forward<Ts>(xs)...);
make(Socket fd, FactoryPtr fptr, size_t max_connections) {
return std::make_unique<connection_acceptor>(fd, std::move(fptr),
max_connections);
}
// -- implementation of socket_event_layer -----------------------------------
error init(socket_manager* owner, const settings& cfg) override {
error start(socket_manager* owner, const settings& cfg) override {
CAF_LOG_TRACE("");
owner_ = owner;
cfg_ = cfg;
if (auto err = factory_.init(owner, cfg))
if (auto err = factory_->start(owner, cfg)) {
CAF_LOG_DEBUG("factory_->start failed:" << err);
return err;
}
on_conn_close_ = make_action([this] { connection_closed(); });
owner->register_reading();
return none;
}
read_result handle_read_event() override {
socket handle() const override {
return fd_;
}
void handle_read_event() override {
CAF_LOG_TRACE("");
if (auto x = accept(fd_)) {
socket_manager_ptr child = factory_.make(owner_->mpx_ptr(), *x);
CAF_ASSERT(owner_ != nullptr);
if (open_connections_ == max_connections_) {
owner_->deregister_reading();
} else if (auto x = accept(fd_)) {
socket_manager_ptr child = factory_->make(owner_->mpx_ptr(), *x);
if (!child) {
CAF_LOG_ERROR("factory failed to create a new child");
return read_result::abort;
}
if (auto err = child->init(cfg_)) {
CAF_LOG_ERROR("failed to initialize new child:" << err);
return read_result::abort;
}
if (limit_ == 0) {
return read_result::again;
} else {
return ++accepted_ < limit_ ? read_result::again : read_result::stop;
on_conn_close_.dispose();
owner_->shutdown();
return;
}
if (++open_connections_ == max_connections_)
owner_->deregister_reading();
child->add_cleanup_listener(on_conn_close_);
std::ignore = child->start(cfg_);
} else if (x.error() == sec::unavailable_or_would_block) {
// Encountered a "soft" error: simply try again later.
CAF_LOG_DEBUG("accept failed:" << x.error());
} else {
CAF_LOG_ERROR("accept failed:" << x.error());
return read_result::stop;
// Encountered a "hard" error: stop.
abort(x.error());
owner_->deregister_reading();
}
}
read_result handle_buffered_data() override {
return read_result::again;
}
read_result handle_continue_reading() override {
return read_result::again;
}
write_result handle_write_event() override {
void handle_write_event() override {
CAF_LOG_ERROR("connection_acceptor received write event");
return write_result::stop;
owner_->deregister_writing();
}
void abort(const error& reason) override {
CAF_LOG_ERROR("connection_acceptor aborts due to an error: " << reason);
factory_.abort(reason);
CAF_LOG_ERROR("connection_acceptor aborts due to an error:" << reason);
factory_->abort(reason);
on_conn_close_.dispose();
}
private:
void connection_closed() {
if (open_connections_-- == max_connections_)
owner_->register_reading();
}
Socket fd_;
size_t limit_;
connection_factory_ptr<connected_socket_type> factory_;
size_t max_connections_;
factory_type factory_;
size_t open_connections_ = 0;
socket_manager* owner_ = nullptr;
size_t accepted_ = 0;
action on_conn_close_;
settings cfg_;
};
......
// 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/fwd.hpp"
#include "caf/net/fwd.hpp"
#include <memory>
namespace caf::net {
/// Creates new socket managers for a @ref socket_acceptor.
template <class Socket>
class connection_factory {
public:
using socket_type = Socket;
virtual ~connection_factory() {
// nop
}
virtual error start(net::socket_manager*, const settings&) {
return none;
}
virtual void abort(const error&) {
// nop
}
virtual socket_manager_ptr make(multiplexer*, Socket) = 0;
template <class Factory>
static std::unique_ptr<connection_factory>
decorate(std::unique_ptr<Factory> ptr);
};
template <class Socket>
using connection_factory_ptr = std::unique_ptr<connection_factory<Socket>>;
template <class Socket, class DecoratedSocket>
class connection_factory_decorator : public connection_factory<Socket> {
public:
using decorated_ptr = connection_factory_ptr<DecoratedSocket>;
explicit connection_factory_decorator(decorated_ptr decorated)
: decorated_(std::move(decorated)) {
// nop
}
error start(net::socket_manager* mgr, const settings& cfg) override {
return decorated_->start(mgr, cfg);
}
void abort(const error& reason) override {
decorated_->abort(reason);
}
socket_manager_ptr make(multiplexer* mpx, Socket fd) override {
return decorated_->make(mpx, fd);
}
private:
decorated_ptr decorated_;
};
/// Lifts a factory from a subtype of `Socket` to a factory for `Socket`.
template <class Socket>
template <class Factory>
std::unique_ptr<connection_factory<Socket>>
connection_factory<Socket>::decorate(std::unique_ptr<Factory> ptr) {
using socket_sub_type = typename Factory::socket_type;
if constexpr (std::is_same_v<Socket, socket_sub_type>) {
return ptr;
} else {
static_assert(std::is_assignable_v<socket_sub_type, Socket>);
using decorator_t = connection_factory_decorator<Socket, socket_sub_type>;
return std::make_unique<decorator_t>(std::move(ptr));
}
}
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/async/consumer.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/ref_counted.hpp"
namespace caf::net {
/// Connects a socket manager to an asynchronous consumer resource. Whenever new
/// data becomes ready, the adapter registers the socket manager for writing.
template <class Buffer>
class consumer_adapter final : public ref_counted, public async::consumer {
public:
using buf_ptr = intrusive_ptr<Buffer>;
void on_producer_ready() override {
// nop
}
void on_producer_wakeup() override {
mgr_->mpx().schedule_fn([adapter = strong_this()] { //
adapter->on_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;
}
bool has_data() const noexcept {
return buf_->has_data();
}
/// Tries to open the resource for reading.
/// @returns a connected adapter that reads from the resource on success,
/// `nullptr` otherwise.
template <class Resource>
static intrusive_ptr<consumer_adapter>
try_open(socket_manager* owner, Resource src) {
CAF_ASSERT(owner != nullptr);
if (auto buf = src.try_open()) {
using ptr_type = intrusive_ptr<consumer_adapter>;
auto adapter = ptr_type{new consumer_adapter(owner, buf), false};
buf->set_consumer(adapter);
return adapter;
} else {
return nullptr;
}
}
friend void intrusive_ptr_add_ref(const consumer_adapter* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const consumer_adapter* ptr) noexcept {
ptr->deref();
}
private:
consumer_adapter(socket_manager* owner, buf_ptr buf)
: mgr_(owner), buf_(std::move(buf)) {
// nop
}
auto strong_this() {
return intrusive_ptr{this};
}
void on_wakeup() {
if (buf_ && buf_->has_consumer_event()) {
mgr_->mpx().register_writing(mgr_);
}
}
intrusive_ptr<socket_manager> mgr_;
intrusive_ptr<Buffer> buf_;
};
template <class T>
using consumer_adapter_ptr = intrusive_ptr<consumer_adapter<T>>;
} // namespace caf::net
......@@ -56,9 +56,9 @@ public:
// -- public member functions ------------------------------------------------
error init(endpoint_manager& manager) override {
error start(endpoint_manager& manager) override {
CAF_LOG_TRACE("");
if (auto err = super::init(manager))
if (auto err = super::start(manager))
return err;
prepare_next_read();
return none;
......
// 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/blocking_producer.hpp"
#include "caf/error.hpp"
#include "caf/net/web_socket/request.hpp"
#include "caf/sec.hpp"
#include <tuple>
namespace caf::net {
/// Connects a flow bridge to input and output buffers.
template <class Trait>
class flow_connector {
public:
using input_type = typename Trait::input_type;
using output_type = typename Trait::output_type;
using pull_t = async::consumer_resource<input_type>;
using push_t = async::producer_resource<output_type>;
using result_type = std::tuple<error, pull_t, push_t>;
virtual ~flow_connector() {
// nop
}
virtual result_type on_request(const settings& cfg) = 0;
/// Returns a trivial implementation that simply returns @p pull and @p push
/// from @p on_request.
static flow_connector_ptr<Trait> make_trivial(pull_t pull, push_t push);
};
/// Trivial flow connector that passes its constructor arguments to the
/// @ref flow_bridge.
template <class Trait>
class flow_connector_trivial_impl : public flow_connector<Trait> {
public:
using super = flow_connector<Trait>;
using input_type = typename Trait::input_type;
using output_type = typename Trait::output_type;
using pull_t = async::consumer_resource<input_type>;
using push_t = async::producer_resource<output_type>;
using result_type = typename super::result_type;
flow_connector_trivial_impl(pull_t pull, push_t push)
: pull_(std::move(pull)), push_(std::move(push)) {
// nop
}
result_type on_request(const settings&) override {
return {{}, std::move(pull_), std::move(push_)};
}
private:
pull_t pull_;
push_t push_;
};
template <class Trait>
flow_connector_ptr<Trait>
flow_connector<Trait>::make_trivial(pull_t pull, push_t push) {
using impl_t = flow_connector_trivial_impl<Trait>;
return std::make_shared<impl_t>(std::move(pull), std::move(push));
}
} // 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"
......@@ -25,6 +27,9 @@ class typed_actor_shell;
template <class... Sigs>
class typed_actor_shell_ptr;
template <class Trait>
class flow_connector;
// -- classes ------------------------------------------------------------------
class actor_shell;
......@@ -48,12 +53,36 @@ struct udp_datagram_socket;
// -- smart pointer aliases ----------------------------------------------------
using multiplexer_ptr = std::shared_ptr<multiplexer>;
using multiplexer_ptr = intrusive_ptr<multiplexer>;
using socket_manager_ptr = intrusive_ptr<socket_manager>;
using weak_multiplexer_ptr = std::weak_ptr<multiplexer>;
template <class Trait>
using flow_connector_ptr = std::shared_ptr<flow_connector<Trait>>;
// -- miscellaneous aliases ----------------------------------------------------
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
......@@ -5,6 +5,7 @@
#pragma once
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
namespace caf::net {
......@@ -17,12 +18,22 @@ public:
/// Queries whether the output device can accept more data straight away.
[[nodiscard]] virtual bool can_send_more() const noexcept = 0;
/// Halt receiving of additional bytes or messages.
virtual void suspend_reading() = 0;
/// Queries whether the lower layer is currently reading from its input
/// device.
[[nodiscard]] virtual bool is_reading() const noexcept = 0;
/// Queries whether the lower layer is currently configured to halt receiving
/// of additional bytes or messages.
[[nodiscard]] virtual bool stopped_reading() const noexcept = 0;
/// 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;
/// Shuts down any connection or session du to an error. Any pending data gets
/// flushed before closing the socket. Protocols with a dedicated closing
/// handshake such as WebSocket may send the close reason to the peer.
virtual void shutdown(const error& reason);
};
} // namespace caf::net
......@@ -15,21 +15,16 @@ class CAF_NET_EXPORT generic_upper_layer {
public:
virtual ~generic_upper_layer();
/// Prepares any pending data for sending.
/// @returns `false` in case of an error to cause the lower layer to stop,
/// `true` otherwise.
[[nodiscard]] virtual bool prepare_send() = 0;
/// Gives the upper layer an opportunity to add additional data to the output
/// buffer.
virtual void prepare_send() = 0;
/// Queries whether all pending data has been sent. The lower calls this
/// function to decide whether it has to wait for write events on the socket.
[[nodiscard]] virtual bool done_sending() = 0;
/// Called by the lower layer after some event triggered re-registering the
/// socket manager for read operations after it has been stopped previously
/// by the read policy. May restart consumption of bytes or messages.
virtual void continue_reading() = 0;
/// Called by the lower layer for cleaning up any state in case of an error.
/// Called by the lower layer for cleaning up any state in case of an error or
/// when disposed.
virtual void abort(const error& reason) = 0;
};
......
......@@ -10,7 +10,6 @@
namespace caf::net::http {
class context;
class header;
class lower_layer;
class server;
......
......@@ -11,6 +11,8 @@
#include "caf/net/generic_lower_layer.hpp"
#include "caf/net/http/fwd.hpp"
#include <string_view>
namespace caf::net::http {
/// Parses HTTP requests and passes them to the upper layer.
......@@ -18,28 +20,40 @@ class CAF_NET_EXPORT lower_layer : public generic_lower_layer {
public:
virtual ~lower_layer();
/// Sends the next header to the client.
virtual bool
send_header(context ctx, status code, const header_fields_map& fields)
= 0;
/// Start or re-start reading data from the client.
virtual void request_messages() = 0;
/// Stops reading messages until calling `request_messages`.
virtual void suspend_reading() = 0;
/// Starts writing an HTTP header.
virtual void begin_header(status code) = 0;
/// Adds a header field. Users may only call this function between
/// `begin_header` and `end_header`.
virtual void add_header_field(std::string_view key, std::string_view val) = 0;
/// Seals the header and transports it to the client.
virtual bool end_header() = 0;
/// Sends the payload after the header.
virtual bool send_payload(context, const_byte_span bytes) = 0;
virtual bool send_payload(const_byte_span bytes) = 0;
/// Sends a chunk of data if the full payload is unknown when starting to
/// send.
virtual bool send_chunk(context, const_byte_span bytes) = 0;
virtual bool send_chunk(const_byte_span bytes) = 0;
/// Sends the last chunk, completing a chunked payload.
virtual bool send_end_of_chunks() = 0;
/// Terminates an HTTP context.
virtual void fin(context) = 0;
/// Convenience function for sending header and payload. Automatically sets
/// the header fields `Content-Type` and `Content-Length`.
bool send_response(context ctx, status codek, std::string_view content_type,
bool send_response(status code, std::string_view content_type,
const_byte_span content);
/// @copydoc send_response
bool send_response(status code, std::string_view content_type,
std::string_view content);
};
} // namespace caf::net::http
// 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/promise.hpp"
#include "caf/byte_span.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/net/http/fwd.hpp"
#include "caf/net/http/header.hpp"
#include "caf/net/http/response.hpp"
#include <cstdint>
#include <memory>
#include <string>
#include <string_view>
#include <type_traits>
namespace caf::net::http {
/// Handle type (implicitly shared) that represents an HTTP client request.
class CAF_NET_EXPORT request {
public:
struct impl {
header hdr;
std::vector<std::byte> body;
async::promise<response> prom;
};
request() = default;
request(request&&) = default;
request(const request&) = default;
request& operator=(request&&) = default;
request& operator=(const request&) = default;
/// @private
explicit request(std::shared_ptr<impl> pimpl) : pimpl_(std::move(pimpl)) {
// nop
}
/// Returns the HTTP header.
/// @pre `valid()`
const header& hdr() const {
return pimpl_->hdr;
}
/// Returns the HTTP body (payload).
/// @pre `valid()`
const_byte_span body() const {
return make_span(pimpl_->body);
}
/// Sends an HTTP response message to the client. Automatically sets the
/// `Content-Type` and `Content-Length` header fields.
/// @pre `valid()`
void respond(status code, std::string_view content_type,
const_byte_span content) const;
/// Sends an HTTP response message to the client. Automatically sets the
/// `Content-Type` and `Content-Length` header fields.
/// @pre `valid()`
void respond(status code, std::string_view content_type,
std::string_view content) const {
return respond(code, content_type, as_bytes(make_span(content)));
}
private:
std::shared_ptr<impl> pimpl_;
};
} // namespace caf::net::http
// 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/promise.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/net/http/fwd.hpp"
#include "caf/span.hpp"
#include "caf/unordered_flat_map.hpp"
#include <cstdint>
#include <memory>
#include <string>
#include <type_traits>
namespace caf::net::http {
/// Handle type (implicitly shared) that represents an HTTP server response.
class CAF_NET_EXPORT response {
public:
using fields_map = unordered_flat_map<std::string, std::string>;
struct impl {
status code;
fields_map fields;
byte_buffer body;
};
response(status code, fields_map fields, byte_buffer body);
/// Returns the HTTP status code.
status code() const {
return pimpl_->code;
}
/// Returns the HTTP header fields.
span<const std::pair<std::string, std::string>> header_fields() const {
return pimpl_->fields.container();
}
/// Returns the HTTP body (payload).
const_byte_span body() const {
return make_span(pimpl_->body);
}
private:
std::shared_ptr<impl> pimpl_;
};
} // namespace caf::net::http
// 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/actor_system.hpp"
#include "caf/async/blocking_producer.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/atomic_ref_counted.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/net/http/request.hpp"
#include "caf/net/http/server.hpp"
#include "caf/net/middleman.hpp"
#include <memory>
#include <type_traits>
namespace caf::detail {
class CAF_NET_EXPORT http_request_producer : public atomic_ref_counted,
public async::producer {
public:
using buffer_ptr = async::spsc_buffer_ptr<net::http::request>;
http_request_producer(buffer_ptr buf) : buf_(std::move(buf)) {
// nop
}
static auto make(buffer_ptr buf) {
auto ptr = make_counted<http_request_producer>(buf);
buf->set_producer(ptr);
return ptr;
}
void on_consumer_ready() override;
void on_consumer_cancel() override;
void on_consumer_demand(size_t) override;
void ref_producer() const noexcept override;
void deref_producer() const noexcept override;
bool push(const net::http::request& item);
private:
buffer_ptr buf_;
};
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(async::execution_context_ptr loop,
http_request_producer_ptr ptr)
: loop_(std::move(loop)), producer_(std::move(ptr)) {
// nop
}
void prepare_send() override;
bool done_sending() override;
void abort(const error& reason) override;
error start(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(async::execution_context_ptr loop,
http_request_producer_ptr ptr) {
return std::make_unique<http_flow_adapter>(loop, ptr);
}
private:
async::execution_context_ptr loop_;
net::http::lower_layer* down_ = nullptr;
std::vector<disposable> pending_;
http_request_producer_ptr producer_;
};
template <class Transport>
class http_acceptor_factory
: public net::connection_factory<typename Transport::socket_type> {
public:
using socket_type = typename Transport::socket_type;
explicit http_acceptor_factory(http_request_producer_ptr producer)
: producer_(std::move(producer)) {
// nop
}
net::socket_manager_ptr make(net::multiplexer* mpx, socket_type fd) override {
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, std::move(transport));
mpx->watch(res->as_disposable());
return res;
}
private:
http_request_producer_ptr producer_;
};
} // namespace caf::detail
namespace caf::net::http {
/// Convenience function for creating async resources for connecting the HTTP
/// server to a worker.
inline auto make_request_resource() {
return async::make_spsc_buffer_resource<request>();
}
/// Listens for incoming HTTP requests on @p fd.
/// @param sys The host system.
/// @param fd An accept socket in listening mode. For a TCP socket, this socket
/// must already listen to a port.
/// @param out A buffer resource that connects the server to a listener that
/// processes the requests.
/// @param cfg Configuration parameters for the acceptor.
template <class Transport = stream_transport, class Socket>
disposable serve(actor_system& sys, Socket fd,
async::producer_resource<request> out,
const settings& cfg = {}) {
using factory_t = detail::http_acceptor_factory<Transport>;
using impl_t = connection_acceptor<Socket>;
auto max_connections = get_or(cfg, defaults::net::max_connections);
if (auto buf = out.try_open()) {
auto& mpx = sys.network_manager().mpx();
auto producer = detail::http_request_producer::make(std::move(buf));
auto factory = std::make_unique<factory_t>(std::move(producer));
auto impl = impl_t::make(fd, std::move(factory), max_connections);
auto ptr = socket_manager::make(&mpx, std::move(impl));
mpx.start(ptr);
return ptr->as_disposable();
} else {
return disposable{};
}
}
} // namespace caf::net::http
......@@ -6,11 +6,11 @@
#include "caf/byte_span.hpp"
#include "caf/detail/append_hex.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp"
#include "caf/logger.hpp"
#include "caf/net/connection_acceptor.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/http/context.hpp"
#include "caf/net/http/header.hpp"
#include "caf/net/http/header_fields_map.hpp"
#include "caf/net/http/lower_layer.hpp"
......@@ -31,7 +31,8 @@
namespace caf::net::http {
/// Implements the server part for the HTTP Protocol as defined in RFC 7231.
class server : public stream_oriented::upper_layer, public http::lower_layer {
class CAF_NET_EXPORT server : public stream_oriented::upper_layer,
public http::lower_layer {
public:
// -- member types -----------------------------------------------------------
......@@ -39,8 +40,6 @@ public:
using status_code_type = status;
using context_type = context;
using header_type = header;
enum class mode {
......@@ -80,34 +79,39 @@ public:
bool can_send_more() const noexcept override;
bool is_reading() const noexcept override;
void write_later() override;
void shutdown() override;
void request_messages() override;
void suspend_reading() override;
bool stopped_reading() const noexcept override;
void begin_header(status code) override;
bool send_header(context, status code,
const header_fields_map& fields) override;
void add_header_field(std::string_view key, std::string_view val) override;
bool send_payload(context, const_byte_span bytes) override;
bool end_header() override;
bool send_chunk(context, const_byte_span bytes) override;
bool send_payload(const_byte_span bytes) override;
bool send_end_of_chunks() override;
bool send_chunk(const_byte_span bytes) override;
void fin(context) override;
bool send_end_of_chunks() override;
// -- stream_oriented::upper_layer implementation ----------------------------
error init(socket_manager* owner, stream_oriented::lower_layer* down,
const settings& config) override;
error start(stream_oriented::lower_layer* down,
const settings& config) override;
void abort(const error& reason) override;
bool prepare_send() override;
void prepare_send() override;
bool done_sending() override;
void continue_reading() override;
ptrdiff_t consume(byte_span input, byte_span) override;
private:
......
......@@ -18,28 +18,17 @@ 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 start(lower_layer* down, const settings& config) = 0;
/// Consumes an HTTP message.
/// @param ctx Identifies this request. The response message must pass this
/// back to the lower layer. Allows clients to send multiple
/// requests in "parallel" (i.e., send multiple requests before
/// receiving the response on the first one).
/// @param hdr The header fields for the received message.
/// @param payload The 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.
virtual ptrdiff_t
consume(context ctx, const header& hdr, const_byte_span payload)
= 0;
virtual ptrdiff_t consume(const header& hdr, const_byte_span payload) = 0;
};
} // namespace caf::net::http
......@@ -20,18 +20,27 @@ namespace caf::net::http::v1 {
CAF_NET_EXPORT std::pair<std::string_view, byte_span>
split_header(byte_span bytes);
/// Writes an HTTP header to the buffer.
/// Writes an HTTP header to @p buf.
CAF_NET_EXPORT void write_header(status code, const header_fields_map& fields,
byte_buffer& buf);
/// Writes a complete HTTP response to the buffer. Automatically sets
/// Content-Type and Content-Length header fields.
/// Write the status code for an HTTP header to @p buf.
CAF_NET_EXPORT void begin_header(status code, byte_buffer& buf);
/// Write a header field to @p buf.
CAF_NET_EXPORT void add_header_field(std::string_view key, std::string_view val,
byte_buffer& buf);
/// Write the status code for an HTTP header to @buf.
CAF_NET_EXPORT bool end_header(byte_buffer& buf);
/// Writes a complete HTTP response to @p buf. Automatically sets Content-Type
/// and Content-Length header fields.
CAF_NET_EXPORT void write_response(status code, std::string_view content_type,
std::string_view content, byte_buffer& buf);
/// Writes a complete HTTP response to the buffer. Automatically sets
/// Content-Type and Content-Length header fields followed by the user-defined
/// @p fields.
/// Writes a complete HTTP response to @p buf. Automatically sets Content-Type
/// and Content-Length header fields followed by the user-defined @p fields.
CAF_NET_EXPORT void write_response(status code, std::string_view content_type,
std::string_view content,
const header_fields_map& fields,
......
......@@ -4,9 +4,17 @@
#pragma once
#include "caf/actor_system.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/byte_span.hpp"
#include "caf/cow_tuple.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/net/message_oriented.hpp"
#include "caf/disposable.hpp"
#include "caf/net/binary/default_trait.hpp"
#include "caf/net/binary/flow_bridge.hpp"
#include "caf/net/binary/lower_layer.hpp"
#include "caf/net/binary/upper_layer.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_oriented.hpp"
#include <cstdint>
......@@ -23,11 +31,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,49 +53,70 @@ 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);
// }
/// Describes the one-time connection event.
using connect_event_t
= cow_tuple<async::consumer_resource<binary::frame>, // Socket to App.
async::producer_resource<binary::frame>>; // App to Socket.
/// Runs length-prefix framing on given connection.
/// @param sys The host system.
/// @param conn A connected stream socket or SSL connection, depending on the
/// `Transport`.
/// @param init Function object for setting up the created flows.
template <class Transport = stream_transport, class Connection, class Init>
static disposable run(actor_system& sys, Connection conn, Init init) {
using trait_t = binary::default_trait;
using frame_t = binary::frame;
static_assert(std::is_invocable_v<Init, connect_event_t&&>,
"invalid signature found for init");
auto [fd_pull, app_push] = async::make_spsc_buffer_resource<frame_t>();
auto [app_pull, fd_push] = async::make_spsc_buffer_resource<frame_t>();
auto mpx = sys.network_manager().mpx_ptr();
auto fc = flow_connector<trait_t>::make_trivial(std::move(fd_pull),
std::move(fd_push));
auto bridge = binary::flow_bridge<trait_t>::make(mpx, std::move(fc));
auto bridge_ptr = bridge.get();
auto impl = length_prefix_framing::make(std::move(bridge));
auto transport = Transport::make(std::move(conn), std::move(impl));
auto ptr = socket_manager::make(mpx, std::move(transport));
bridge_ptr->self_ref(ptr->as_disposable());
mpx->start(ptr);
init(connect_event_t{app_pull, app_push});
return disposable{std::move(ptr)};
}
/// The default number of concurrently open connections when using `accept`.
static constexpr size_t default_max_connections = 128;
/// A producer resource for the acceptor. Any accepted connection is
/// represented by two buffers: one for input and one for output.
using acceptor_resource_t = async::producer_resource<connect_event_t>;
/// Listens for incoming connection on @p fd.
/// @param sys The host system.
/// @param fd An accept socket in listening mode. For a TCP socket, this
/// socket must already listen to a port.
/// @param cfg Configures the acceptor. Currently, the only supported
/// configuration parameter is `max-connections`.
template <class Transport = stream_transport, class Socket, class OnConnect>
static disposable accept(actor_system& sys, Socket fd,
acceptor_resource_t out, const settings& cfg = {}) {
}
// -- implementation of stream_oriented::upper_layer -------------------------
error init(socket_manager* owner, stream_oriented::lower_layer* down,
const settings& config) override;
error start(stream_oriented::lower_layer* down,
const settings& config) override;
void abort(const error& reason) override;
ptrdiff_t consume(byte_span buffer, byte_span delta) override;
void continue_reading() override;
bool prepare_send() override;
void prepare_send() override;
bool done_sending() override;
// -- implementation of message_oriented::lower_layer ------------------------
// -- implementation of binary::lower_layer ----------------------------------
bool can_send_more() const noexcept override;
......@@ -95,7 +124,9 @@ public:
void suspend_reading() override;
bool stopped_reading() const noexcept override;
bool is_reading() const noexcept override;
void write_later() override;
void begin_message() override;
......@@ -103,9 +134,7 @@ public:
bool end_message() override;
void send_close_message() override;
void send_close_message(const error& reason) override;
void shutdown() override;
// -- utility functions ------------------------------------------------------
......
#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,
const settings&) {
error start(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,63 +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;
}
};
bool prepare_send() override {
write_helper helper{this};
while (down_->can_send_more() && in_) {
auto [again, consumed] = in_->pull(async::delay_errors, 1, helper);
if (!again) {
if (helper.err) {
down_->send_close_message(helper.err);
} else {
down_->send_close_message();
}
in_ = nullptr;
} else if (helper.aborted) {
in_->cancel();
in_ = nullptr;
return false;
} else if (consumed == 0) {
return true;
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;
}
}
return true;
}
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();
......@@ -163,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_;
......
......@@ -69,19 +69,15 @@ public:
}
multiplexer& mpx() noexcept {
return mpx_;
return *mpx_;
}
const multiplexer& mpx() const noexcept {
return mpx_;
return *mpx_;
}
multiplexer* mpx_ptr() noexcept {
return &mpx_;
}
const multiplexer* mpx_ptr() const noexcept {
return &mpx_;
multiplexer* mpx_ptr() const noexcept {
return mpx_.get();
}
private:
......@@ -91,7 +87,7 @@ private:
actor_system& sys_;
/// Stores the global socket I/O multiplexer.
multiplexer mpx_;
multiplexer_ptr mpx_;
/// Runs the multiplexer's event loop
std::thread mpx_thread_;
......
......@@ -9,6 +9,8 @@
#include <thread>
#include "caf/action.hpp"
#include "caf/async/execution_context.hpp"
#include "caf/detail/atomic_ref_counted.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/operation.hpp"
......@@ -28,7 +30,8 @@ namespace caf::net {
class pollset_updater;
/// Multiplexes any number of ::socket_manager objects with a ::socket.
class CAF_NET_EXPORT multiplexer {
class CAF_NET_EXPORT multiplexer : public detail::atomic_ref_counted,
public async::execution_context {
public:
// -- member types -----------------------------------------------------------
......@@ -55,13 +58,18 @@ public:
// -- constructors, destructors, and assignment operators --------------------
~multiplexer();
// -- factories --------------------------------------------------------------
/// @param parent Points to the owning middleman instance. May be `nullptr`
/// only for the purpose of unit testing if no @ref
/// socket_manager requires access to the @ref middleman or the
/// @ref actor_system.
explicit multiplexer(middleman* parent);
~multiplexer();
static multiplexer_ptr make(middleman* parent) {
auto ptr = new multiplexer(parent);
return multiplexer_ptr{ptr, false};
}
// -- initialization ---------------------------------------------------------
......@@ -73,10 +81,10 @@ public:
size_t num_socket_managers() const noexcept;
/// Returns the index of `mgr` in the pollset or `-1`.
ptrdiff_t index_of(const socket_manager_ptr& mgr);
ptrdiff_t index_of(const socket_manager_ptr& mgr) const noexcept;
/// Returns the index of `fd` in the pollset or `-1`.
ptrdiff_t index_of(socket fd);
ptrdiff_t index_of(socket fd) const noexcept;
/// Returns the owning @ref middleman instance.
middleman& owner();
......@@ -87,54 +95,48 @@ public:
/// Computes the current mask for the manager. Mostly useful for testing.
operation mask_of(const socket_manager_ptr& mgr);
// -- thread-safe signaling --------------------------------------------------
// -- implementation of execution_context ------------------------------------
/// Registers `mgr` for read events.
/// @thread-safe
void register_reading(const socket_manager_ptr& mgr);
void ref_execution_context() const noexcept override;
/// Registers `mgr` for write events.
/// @thread-safe
void register_writing(const socket_manager_ptr& mgr);
void deref_execution_context() const noexcept override;
/// Triggers a continue reading event for `mgr`.
/// @thread-safe
void continue_reading(const socket_manager_ptr& mgr);
void schedule(action what) override;
/// Triggers a continue writing event for `mgr`.
/// @thread-safe
void continue_writing(const socket_manager_ptr& mgr);
void watch(disposable what) override;
/// Schedules a call to `mgr->handle_error(sec::discarded)`.
/// @thread-safe
void discard(const socket_manager_ptr& mgr);
// -- thread-safe signaling --------------------------------------------------
/// Stops further reading by `mgr`.
/// Registers `mgr` for initialization in the multiplexer's thread.
/// @thread-safe
void shutdown_reading(const socket_manager_ptr& mgr);
void start(socket_manager_ptr mgr);
/// Stops further writing by `mgr`.
/// Signals the multiplexer to initiate shutdown.
/// @thread-safe
void shutdown_writing(const socket_manager_ptr& mgr);
void shutdown();
/// Schedules an action for execution on this multiplexer.
/// @thread-safe
void schedule(const action& what);
// -- callbacks for socket managers ------------------------------------------
/// Schedules an action for execution on this multiplexer.
/// @thread-safe
template <class F>
void schedule_fn(F f) {
schedule(make_action(std::move(f)));
}
/// Registers `mgr` for read events.
void register_reading(socket_manager* mgr);
/// Registers `mgr` for initialization in the multiplexer's thread.
/// @thread-safe
void init(const socket_manager_ptr& mgr);
/// Registers `mgr` for write events.
void register_writing(socket_manager* mgr);
/// Signals the multiplexer to initiate shutdown.
/// @thread-safe
void shutdown();
/// Deregisters `mgr` from read events.
void deregister_reading(socket_manager* mgr);
/// Deregisters `mgr` from write events.
void deregister_writing(socket_manager* mgr);
/// Deregisters @p mgr from read and write events.
void deregister(socket_manager* mgr);
/// Queries whether `mgr` is currently registered for reading.
bool is_reading(const socket_manager* mgr) const noexcept;
/// Queries whether `mgr` is currently registered for writing.
bool is_writing(const socket_manager* mgr) const noexcept;
// -- control flow -----------------------------------------------------------
......@@ -160,18 +162,16 @@ protected:
/// Handles an I/O event on given manager.
void handle(const socket_manager_ptr& mgr, short events, short revents);
/// Transfers socket ownership from one manager to another.
void do_handover(const socket_manager_ptr& mgr);
/// Returns a change entry for the socket at given index. Lazily creates a new
/// entry before returning if necessary.
poll_update& update_for(ptrdiff_t index);
/// Returns a change entry for the socket of the manager.
poll_update& update_for(const socket_manager_ptr& mgr);
poll_update& update_for(socket_manager* mgr);
/// Writes `opcode` and pointer to `mgr` the the pipe for handling an event
/// later via the pollset updater.
/// @warning assumes ownership of @p ptr.
template <class T>
void write_to_pipe(uint8_t opcode, T* ptr);
......@@ -182,13 +182,7 @@ protected:
}
/// Queries the currently active event bitmask for `mgr`.
short active_mask_of(const socket_manager_ptr& mgr);
/// Queries whether `mgr` is currently registered for reading.
bool is_reading(const socket_manager_ptr& mgr);
/// Queries whether `mgr` is currently registered for writing.
bool is_writing(const socket_manager_ptr& mgr);
short active_mask_of(const socket_manager* mgr) const noexcept;
// -- member variables -------------------------------------------------------
......@@ -219,26 +213,21 @@ protected:
/// Signals whether shutdown has been requested.
bool shutting_down_ = false;
/// Keeps track of watched disposables.
std::vector<disposable> watched_;
private:
// -- constructors, destructors, and assignment operators --------------------
explicit multiplexer(middleman* parent);
// -- internal callbacks the pollset updater ---------------------------------
void do_shutdown();
void do_register_reading(const socket_manager_ptr& mgr);
void do_register_writing(const socket_manager_ptr& mgr);
void do_continue_reading(const socket_manager_ptr& mgr);
void do_continue_writing(const socket_manager_ptr& mgr);
void do_discard(const socket_manager_ptr& mgr);
void do_shutdown_reading(const socket_manager_ptr& mgr);
void do_shutdown_writing(const socket_manager_ptr& mgr);
void do_init(const socket_manager_ptr& mgr);
void do_start(const socket_manager_ptr& mgr);
};
} // namespace caf::net
......@@ -25,12 +25,7 @@ public:
using msg_buf = std::array<std::byte, sizeof(intptr_t) + 1>;
enum class code : uint8_t {
register_reading,
continue_reading,
register_writing,
continue_writing,
init_manager,
discard_manager,
start_manager,
shutdown_reading,
shutdown_writing,
run_action,
......@@ -47,20 +42,19 @@ public:
// -- implementation of socket_event_layer -----------------------------------
error init(socket_manager* owner, const settings& cfg) override;
error start(socket_manager* owner, const settings& cfg) override;
read_result handle_read_event() override;
socket handle() const override;
read_result handle_buffered_data() override;
void handle_read_event() override;
read_result handle_continue_reading() override;
write_result handle_write_event() override;
void handle_write_event() override;
void abort(const error& reason) override;
private:
pipe_socket fd_;
socket_manager* owner_ = nullptr;
multiplexer* mpx_ = nullptr;
msg_buf buf_;
size_t buf_size_ = 0;
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <memory>
#include <new>
#include "caf/async/producer.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/subscription.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/ref_counted.hpp"
namespace caf::net {
/// Connects a socket manager to an asynchronous producer resource.
template <class Buffer>
class producer_adapter final : public ref_counted, public async::producer {
public:
using atomic_count = std::atomic<size_t>;
using buf_ptr = intrusive_ptr<Buffer>;
using value_type = typename Buffer::value_type;
void on_consumer_ready() override {
// nop
}
void on_consumer_cancel() override {
mgr_->mpx().schedule_fn([adapter = strong_this()] { //
adapter->on_cancel();
});
}
void on_consumer_demand(size_t) override {
mgr_->continue_reading();
}
void ref_producer() const noexcept override {
this->ref();
}
void deref_producer() const noexcept override {
this->deref();
}
/// Tries to open the resource for writing.
/// @returns a connected adapter that writes to the resource on success,
/// `nullptr` otherwise.
template <class Resource>
static intrusive_ptr<producer_adapter>
try_open(socket_manager* owner, Resource src) {
CAF_ASSERT(owner != nullptr);
if (auto buf = src.try_open()) {
using ptr_type = intrusive_ptr<producer_adapter>;
auto adapter = ptr_type{new producer_adapter(owner, buf), false};
buf->set_producer(adapter);
return adapter;
} else {
return nullptr;
}
}
/// 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();
buf_ = nullptr;
mgr_ = nullptr;
}
}
void abort(error reason) {
if (buf_) {
buf_->abort(std::move(reason));
buf_ = nullptr;
mgr_ = nullptr;
}
}
friend void intrusive_ptr_add_ref(const producer_adapter* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const producer_adapter* ptr) noexcept {
ptr->deref();
}
private:
producer_adapter(socket_manager* owner, buf_ptr buf)
: mgr_(owner), buf_(std::move(buf)) {
// nop
}
void on_cancel() {
if (buf_) {
mgr_->mpx().shutdown_reading(mgr_);
buf_ = nullptr;
mgr_ = nullptr;
}
}
auto strong_this() {
return intrusive_ptr{this};
}
intrusive_ptr<socket_manager> mgr_;
intrusive_ptr<Buffer> buf_;
};
template <class T>
using producer_adapter_ptr = intrusive_ptr<producer_adapter<T>>;
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/actor_system.hpp"
#include "caf/fwd.hpp"
#include "caf/net/http/server.hpp"
#include "caf/net/prometheus/server.hpp"
#include "caf/net/stream_transport.hpp"
namespace caf::detail {
template <class Transport>
class prometheus_acceptor_factory
: public net::connection_factory<typename Transport::socket_type> {
public:
using state_ptr = net::prometheus::server::scrape_state_ptr;
using socket_type = typename Transport::socket_type;
explicit prometheus_acceptor_factory(state_ptr ptr) : ptr_(std::move(ptr)) {
// nop
}
net::socket_manager_ptr make(net::multiplexer* mpx, socket_type fd) override {
auto prom_serv = net::prometheus::server::make(ptr_);
auto http_serv = net::http::server::make(std::move(prom_serv));
auto transport = Transport::make(fd, std::move(http_serv));
return net::socket_manager::make(mpx, std::move(transport));
}
private:
state_ptr ptr_;
};
} // namespace caf::detail
namespace caf::net::prometheus {
/// Listens for incoming WebSocket connection on @p fd.
/// @param sys The host system.
/// @param fd An accept socket in listening mode. For a TCP socket, this socket
/// must already listen to a port.
template <class Transport = stream_transport, class Socket>
disposable serve(actor_system& sys, Socket fd, const settings& cfg = {}) {
using factory_t = detail::prometheus_acceptor_factory<Transport>;
using impl_t = connection_acceptor<Socket>;
auto mpx = &sys.network_manager().mpx();
auto registry = &sys.metrics();
auto state = prometheus::server::scrape_state::make(registry);
auto factory = std::make_unique<factory_t>(std::move(state));
auto max_connections = get_or(cfg, defaults::net::max_connections);
auto impl = impl_t::make(fd, std::move(factory), max_connections);
auto mgr = socket_manager::make(mpx, std::move(impl));
mpx->start(mgr);
return mgr->as_disposable();
}
} // namespace caf::net::prometheus
// 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/fwd.hpp"
#include "caf/net/http/fwd.hpp"
#include "caf/net/http/upper_layer.hpp"
#include "caf/telemetry/collector/prometheus.hpp"
#include "caf/telemetry/importer/process.hpp"
#include <chrono>
#include <memory>
#include <string_view>
namespace caf::net::prometheus {
/// Makes metrics available to clients via the Prometheus exposition format.
class server : public http::upper_layer {
public:
// -- member types -----------------------------------------------------------
/// State for scraping Metrics data. Shared between all server instances.
class scrape_state {
public:
using clock_type = std::chrono::steady_clock;
using time_point = clock_type::time_point;
using duration = clock_type::duration;
std::string_view scrape();
explicit scrape_state(telemetry::metric_registry* ptr)
: registry(ptr), last_scrape(duration{0}), proc_importer(*ptr) {
// nop
}
static std::shared_ptr<scrape_state> make(telemetry::metric_registry* ptr) {
return std::make_shared<scrape_state>(ptr);
}
telemetry::metric_registry* registry;
std::chrono::steady_clock::time_point last_scrape;
telemetry::importer::process proc_importer;
telemetry::collector::prometheus collector;
};
using scrape_state_ptr = std::shared_ptr<scrape_state>;
// -- factories --------------------------------------------------------------
static std::unique_ptr<server> make(scrape_state_ptr state) {
return std::unique_ptr<server>{new server(std::move(state))};
}
// -- implementation of http::upper_layer ------------------------------------
void prepare_send() override;
bool done_sending() override;
void abort(const error& reason) override;
error start(http::lower_layer* down, const settings& config) override;
ptrdiff_t consume(const http::header& hdr, const_byte_span payload) override;
private:
explicit server(scrape_state_ptr state) : state_(std::move(state)) {
// nop
}
scrape_state_ptr state_;
http::lower_layer* down_ = nullptr;
};
} // namespace caf::net::prometheus
......@@ -37,6 +37,14 @@ struct CAF_NET_EXPORT socket : detail::comparable<socket> {
return static_cast<signed_socket_id>(id)
- static_cast<signed_socket_id>(other.id);
}
constexpr explicit operator bool() const noexcept {
return id != invalid_socket_id;
}
constexpr bool operator!() const noexcept {
return id == invalid_socket_id;
}
};
/// @relates socket
......
......@@ -16,64 +16,27 @@ class CAF_NET_EXPORT socket_event_layer {
public:
virtual ~socket_event_layer();
/// Encodes how to proceed after a read operation.
enum class read_result {
/// Indicates that a manager wants to read again later.
again,
/// Indicates that a manager wants to stop reading until explicitly resumed.
stop,
/// Indicates that a manager wants to write to the socket instead of reading
/// from the socket.
want_write,
/// Indicates that the manager no longer reads from the socket.
close,
/// Indicates that the manager encountered a fatal error and stops both
/// reading and writing.
abort,
/// Indicates that a manager is done with the socket and hands ownership to
/// another manager.
handover,
};
/// Starts processing on this layer.
virtual error start(socket_manager* owner, const settings& cfg) = 0;
/// Encodes how to proceed after a write operation.
enum class write_result {
/// Indicates that a manager wants to read again later.
again,
/// Indicates that a manager wants to stop reading until explicitly resumed.
stop,
/// Indicates that a manager wants to read from the socket instead of
/// writing to the socket.
want_read,
/// Indicates that the manager no longer writes to the socket.
close,
/// Indicates that the manager encountered a fatal error and stops both
/// reading and writing.
abort,
/// Indicates that a manager is done with the socket and hands ownership to
/// another manager.
handover,
};
/// Initializes the layer.
virtual error init(socket_manager* owner, const settings& cfg) = 0;
/// Returns the handle for the managed socket.
virtual socket handle() const = 0;
/// Handles a read event on the managed socket.
virtual read_result handle_read_event() = 0;
/// Handles internally buffered data.
virtual read_result handle_buffered_data() = 0;
/// Handles a request to continue reading on the socket.
virtual read_result handle_continue_reading() = 0;
virtual void handle_read_event() = 0;
/// Handles a write event on the managed socket.
virtual write_result handle_write_event() = 0;
virtual void handle_write_event() = 0;
/// Called after returning `handover` from a read or write handler.
virtual bool do_handover(std::unique_ptr<socket_event_layer>& next);
/// Called on hard errors on the managed socket.
/// Called on socket errors or when the manager gets disposed.
virtual void abort(const error& reason) = 0;
/// Queries whether the object can be safely discarded after calling
/// @ref abort on it, e.g., that pending data has been written.
virtual bool finalized() const noexcept;
};
} // namespace caf::net
......@@ -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;
......
......@@ -4,11 +4,14 @@
#pragma once
#include "caf/action.hpp"
#include "caf/actor.hpp"
#include "caf/actor_system.hpp"
#include "caf/callback.hpp"
#include "caf/detail/atomic_ref_counted.hpp"
#include "caf/detail/infer_actor_shell_ptr_type.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/disposable.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/actor_shell.hpp"
......@@ -16,7 +19,6 @@
#include "caf/net/socket.hpp"
#include "caf/net/socket_event_layer.hpp"
#include "caf/net/typed_actor_shell.hpp"
#include "caf/ref_counted.hpp"
#include "caf/sec.hpp"
#include <type_traits>
......@@ -24,27 +26,18 @@
namespace caf::net {
/// Manages the lifetime of a single socket and handles any I/O events on it.
class CAF_NET_EXPORT socket_manager : public ref_counted {
class CAF_NET_EXPORT socket_manager : public detail::atomic_ref_counted,
public disposable_impl {
public:
// -- member types -----------------------------------------------------------
using read_result = socket_event_layer::read_result;
using write_result = socket_event_layer::write_result;
using event_handler_ptr = std::unique_ptr<socket_event_layer>;
/// Stores manager-related flags in a single block.
struct flags_t {
bool read_closed : 1;
bool write_closed : 1;
};
// -- constructors, destructors, and assignment operators --------------------
/// @pre `handle != invalid_socket`
/// @pre `mpx!= nullptr`
socket_manager(multiplexer* mpx, socket handle, event_handler_ptr handler);
socket_manager(multiplexer* mpx, event_handler_ptr handler);
~socket_manager() override;
......@@ -54,28 +47,7 @@ public:
// -- factories --------------------------------------------------------------
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));
}
static socket_manager_ptr make(multiplexer* mpx, event_handler_ptr handler);
// -- properties -------------------------------------------------------------
......@@ -112,92 +84,112 @@ public:
return mpx_;
}
/// Returns whether the manager closed read operations on the socket.
[[nodiscard]] bool read_closed() const noexcept {
return flags_.read_closed;
}
/// Queries whether the manager is registered for reading.
bool is_reading() const noexcept;
/// Returns whether the manager closed write operations on the socket.
[[nodiscard]] bool write_closed() const noexcept {
return flags_.write_closed;
}
/// Queries whether the manager is registered for writing.
bool is_writing() const noexcept;
// -- event loop management --------------------------------------------------
/// Registers the manager for read operations on the @ref multiplexer.
/// @thread-safe
/// Registers the manager for read operations.
void register_reading();
/// Registers the manager for write operations on the @ref multiplexer.
/// @thread-safe
/// Registers the manager for write operations.
void register_writing();
/// Schedules a call to `handle_continue_reading` on the @ref multiplexer.
/// This mechanism allows users to signal changes in the environment to the
/// manager that allow it to make progress, e.g., new demand in asynchronous
/// buffer that allow the manager to push available data downstream. The event
/// is a no-op if the manager is already registered for reading.
/// @thread-safe
void continue_reading();
/// Schedules a call to `handle_continue_reading` on the @ref multiplexer.
/// This mechanism allows users to signal changes in the environment to the
/// manager that allow it to make progress, e.g., new data for writing in an
/// asynchronous buffer. The event is a no-op if the manager is already
/// registered for writing.
/// @thread-safe
void continue_writing();
/// Deregisters the manager from read operations.
void deregister_reading();
// -- callbacks for the multiplexer ------------------------------------------
/// Deregisters the manager from write operations.
void deregister_writing();
/// Closes the read channel of the socket.
void close_read() noexcept;
/// Deregisters the manager from both read and write operations.
void deregister();
/// Closes the write channel of the socket.
void close_write() noexcept;
/// Schedules a call to `fn` on the multiplexer when this socket manager
/// cleans up its state.
/// @note Must be called before `start`.
void add_cleanup_listener(action fn) {
cleanup_listeners_.push_back(std::move(fn));
}
/// Initializes the manager and its all of its sub-components.
error init(const settings& cfg);
// -- callbacks for the handler ----------------------------------------------
/// Called whenever the socket received new data.
read_result handle_read_event();
/// Schedules a call to `do_handover` on the handler.
void schedule_handover();
/// Called after handovers to allow the manager to process any data that is
/// already buffered at the transport policy and thus would not trigger a read
/// event on the socket.
read_result handle_buffered_data();
/// Schedules @p what to run later.
void schedule(action what);
/// Restarts a socket manager that suspended reads. Calling this member
/// function on active managers is a no-op. This function also should read any
/// data buffered outside of the socket.
read_result handle_continue_reading();
/// Schedules @p what to run later.
template <class F>
void schedule_fn(F&& what) {
schedule(make_action(std::forward<F>(what)));
}
/// Shuts down this socket manager.
void shutdown();
// -- callbacks for the multiplexer ------------------------------------------
/// Starts the manager and its all of its processing layers.
error start(const settings& cfg);
/// Called whenever the socket received new data.
void handle_read_event();
/// Called whenever the socket is allowed to send data.
write_result handle_write_event();
void handle_write_event();
/// Called when the remote side becomes unreachable due to an error.
/// @param code The error code as reported by the operating system.
/// Called when the remote side becomes unreachable due to an error or after
/// calling @ref dispose.
/// @param code The error code as reported by the operating system or
/// @ref sec::disposed.
void handle_error(sec code);
/// Performs a handover to another transport after `handle_read_event` or
/// `handle_read_event` returned `handover`.
[[nodiscard]] bool do_handover();
// -- implementation of disposable_impl --------------------------------------
void dispose() override;
bool disposed() const noexcept override;
void ref_disposable() const noexcept override;
void deref_disposable() const noexcept override;
private:
// -- utility functions ------------------------------------------------------
void cleanup();
socket_manager_ptr strong_this();
// -- member variables -------------------------------------------------------
/// Stores the socket file descriptor. The socket manager automatically closes
/// the socket in its destructor.
socket fd_;
/// Points to the multiplexer that owns this manager.
/// Points to the multiplexer that executes this manager. Note: we do not need
/// to increase the reference count for the multiplexer, because the
/// multiplexer owns all managers in the sense that calling any member
/// function on a socket manager may not occur if the actor system has shut
/// down (and the multiplexer is part of the actor system).
multiplexer* mpx_;
/// Stores flags for the socket file descriptor.
flags_t flags_;
/// Stores the event handler that operators on the socket file descriptor.
event_handler_ptr handler_;
/// Stores whether `shutdown` has been called.
bool shutting_down_ = false;
/// Stores whether the manager has been either explicitly disposed or shut
/// down by demand of the application.
std::atomic<bool> disposed_;
/// Callbacks to run when calling `cleanup`.
std::vector<action> cleanup_listeners_;
};
/// @relates socket_manager
......
......@@ -13,6 +13,10 @@
namespace caf::net::ssl {
struct close_on_shutdown_t {};
constexpr close_on_shutdown_t close_on_shutdown = close_on_shutdown_t{};
/// SSL state, shared by multiple connections.
class CAF_NET_EXPORT context {
public:
......@@ -71,8 +75,15 @@ public:
// -- connections ------------------------------------------------------------
/// Creates a new SSL connection on `fd`. The connection does not take
/// ownership of the socket, i.e., does not close the socket when the SSL
/// session ends.
expected<connection> new_connection(stream_socket fd);
/// Creates a new SSL connection on `fd`. The connection takes ownership of
/// the socket, i.e., closes the socket when the SSL session ends.
expected<connection> new_connection(stream_socket fd, close_on_shutdown_t);
// -- certificates and keys --------------------------------------------------
/// Configure the context to use the default locations for loading CA
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
......@@ -32,6 +32,8 @@ public:
bool convert(const_byte_span bytes, input_type& x);
bool convert(std::string_view text, input_type& x);
error last_error();
};
} // namespace caf::net::web_socket
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
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