Commit d3769f0c authored by Dominik Charousset's avatar Dominik Charousset

Drop Broker examples and update the Manual

The new caf-net module offers superior alternatives to the Brokers from
the I/O module. Eventually, we want to fade out the Brokers API
altogether. Hence, we remove the Broker examples and encourage
developers to use the new alternatives instead.
parent d81e4e93
......@@ -68,24 +68,6 @@ if(TARGET CAF::io)
add_io_example(remoting remote_spawn)
add_io_example(remoting distributed_calculator)
# basic I/O with brokers
add_io_example(broker simple_broker)
add_io_example(broker simple_http_broker)
if(CAF_ENABLE_PROTOBUF_EXAMPLES)
find_package(Protobuf REQUIRED)
if(NOT PROTOBUF_PROTOC_EXECUTABLE)
message(FATAL_ERROR "CMake was unable to set PROTOBUF_PROTOC_EXECUTABLE")
endif()
protobuf_generate_cpp(ProtoSources ProtoHeaders "${CMAKE_CURRENT_SOURCE_DIR}/remoting/pingpong.proto")
include_directories(${PROTOBUF_INCLUDE_DIR})
include_directories(${CMAKE_CURRENT_BINARY_DIR})
add_executable(protobuf_broker broker/protobuf_broker.cpp ${ProtoSources})
target_link_libraries(protobuf_broker
PRIVATE ${PROTOBUF_LIBRARIES} CAF::internal CAF::io)
add_dependencies(protobuf_broker all_examples)
endif()
if(CAF_ENABLE_CURL_EXAMPLES)
find_package(CURL REQUIRED)
add_executable(curl_fuse curl/curl_fuse.cpp)
......
#include <iostream>
#include <limits>
#include <memory>
#include <string>
#include <vector>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
#ifdef CAF_WINDOWS
# include <winsock2.h>
#else
# include <arpa/inet.h>
#endif
CAF_PUSH_WARNINGS
#include "pingpong.pb.h"
CAF_POP_WARNINGS
CAF_BEGIN_TYPE_ID_BLOCK(protobuf_example, first_custom_type_id)
CAF_ADD_ATOM(protobuf_example, kickoff_atom)
CAF_END_TYPE_ID_BLOCK(protobuf_example)
namespace {
using namespace caf;
using namespace caf::io;
// utility function to print an exit message with custom name
void print_on_exit(scheduled_actor* self, const std::string& name) {
self->attach_functor([=](const error& reason) {
aout(self) << name << " exited: " << to_string(reason) << std::endl;
});
}
struct ping_state {
size_t count = 0;
};
behavior ping(stateful_actor<ping_state>* self, size_t num_pings) {
print_on_exit(self, "ping");
return {
[=](kickoff_atom, const actor& pong) {
self->send(pong, ping_atom_v, 1);
self->become([=](pong_atom, int value) -> result<ping_atom, int> {
if (++(self->state.count) >= num_pings)
self->quit();
return {ping_atom_v, value + 1};
});
},
};
}
behavior pong(event_based_actor* self) {
print_on_exit(self, "pong");
return {
[=](ping_atom, int value) { return make_message(pong_atom_v, value); },
};
}
void protobuf_io(broker* self, connection_handle hdl, const actor& buddy) {
print_on_exit(self, "protobuf_io");
aout(self) << "protobuf broker started" << std::endl;
self->monitor(buddy);
self->set_down_handler([=](const down_msg& dm) {
if (dm.source == buddy) {
aout(self) << "our buddy is down" << std::endl;
self->quit(dm.reason);
}
});
auto write = [=](const org::caf::PingOrPong& p) {
std::string buf = p.SerializeAsString();
auto s = htonl(static_cast<uint32_t>(buf.size()));
self->write(hdl, sizeof(uint32_t), &s);
self->write(hdl, buf.size(), buf.data());
self->flush(hdl);
};
auto default_callbacks = message_handler{
[=](const connection_closed_msg&) {
aout(self) << "connection closed" << std::endl;
self->send_exit(buddy, exit_reason::remote_link_unreachable);
self->quit(exit_reason::remote_link_unreachable);
},
[=](ping_atom, int i) {
aout(self) << "'ping' " << i << std::endl;
org::caf::PingOrPong p;
p.mutable_ping()->set_id(i);
write(p);
},
[=](pong_atom, int i) {
aout(self) << "'pong' " << i << std::endl;
org::caf::PingOrPong p;
p.mutable_pong()->set_id(i);
write(p);
},
};
auto await_protobuf_data = message_handler {
[=](const new_data_msg& msg) {
org::caf::PingOrPong p;
p.ParseFromArray(msg.buf.data(), static_cast<int>(msg.buf.size()));
if (p.has_ping()) {
self->send(buddy, ping_atom_v, p.ping().id());
}
else if (p.has_pong()) {
self->send(buddy, pong_atom_v, p.pong().id());
}
else {
self->quit(exit_reason::user_shutdown);
std::cerr << "neither Ping nor Pong!" << std::endl;
}
// receive next length prefix
self->configure_read(hdl, receive_policy::exactly(sizeof(uint32_t)));
self->unbecome();
},
}.or_else(default_callbacks);
auto await_length_prefix = message_handler {
[=](const new_data_msg& msg) {
uint32_t num_bytes;
memcpy(&num_bytes, msg.buf.data(), sizeof(uint32_t));
num_bytes = htonl(num_bytes);
if (num_bytes > (1024 * 1024)) {
aout(self) << "someone is trying something nasty" << std::endl;
self->quit(exit_reason::user_shutdown);
return;
}
// receive protobuf data
auto nb = static_cast<size_t>(num_bytes);
self->configure_read(hdl, receive_policy::exactly(nb));
self->become(keep_behavior, await_protobuf_data);
},
}.or_else(default_callbacks);
// initial setup
self->configure_read(hdl, receive_policy::exactly(sizeof(uint32_t)));
self->become(await_length_prefix);
}
behavior server(broker* self, const actor& buddy) {
print_on_exit(self, "server");
aout(self) << "server is running" << std::endl;
return {
[=](const new_connection_msg& msg) {
aout(self) << "server accepted new connection" << std::endl;
auto io_actor = self->fork(protobuf_io, msg.handle, buddy);
// only accept 1 connection in our example
self->quit();
},
};
}
class config : public actor_system_config {
public:
uint16_t port = 0;
std::string host = "localhost";
bool server_mode = false;
config() {
opt_group{custom_options_, "global"}
.add(port, "port,p", "set port")
.add(host, "host,H", "set host (ignored in server mode)")
.add(server_mode, "server-mode,s", "enable server mode");
}
};
void run_server(actor_system& system, const config& cfg) {
std::cout << "run in server mode" << std::endl;
auto pong_actor = system.spawn(pong);
auto server_actor = system.middleman().spawn_server(server, cfg.port,
pong_actor);
if (!server_actor)
std::cerr << "unable to spawn server: " << to_string(server_actor.error())
<< std::endl;
}
void run_client(actor_system& system, const config& cfg) {
std::cout << "run in client mode" << std::endl;
auto ping_actor = system.spawn(ping, 20u);
auto io_actor = system.middleman().spawn_client(protobuf_io, cfg.host,
cfg.port, ping_actor);
if (!io_actor) {
std::cout << "cannot connect to " << cfg.host << " at port " << cfg.port
<< ": " << to_string(io_actor.error()) << std::endl;
return;
}
send_as(*io_actor, ping_actor, kickoff_atom_v, *io_actor);
}
void caf_main(actor_system& system, const config& cfg) {
auto f = cfg.server_mode ? run_server : run_client;
f(system, cfg);
}
} // namespace
CAF_MAIN(id_block::protobuf_example, io::middleman)
/******************************************************************************\
* This example program showcases how to manually manage socket IO using *
* a broker. Server and client exchange integers in a 'ping-pong protocol'. *
* *
* Minimal setup: *
* - simple_broker -s 4242 *
* - simple_broker -c localhost 4242 *
\******************************************************************************/
#include "caf/config.hpp"
#ifdef CAF_WINDOWS
# define _WIN32_WINNT 0x0600
# include <winsock2.h>
#else
# include <arpa/inet.h> // htonl
#endif
#include <cassert>
#include <cstdint>
#include <iostream>
#include <limits>
#include <memory>
#include <string>
#include <vector>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
using std::cerr;
using std::cout;
using std::endl;
using namespace caf;
using namespace caf::io;
namespace {
// --(rst-attach-begin)--
// Utility function to print an exit message with custom name.
void print_on_exit(const actor& hdl, const std::string& name) {
hdl->attach_functor([=](const error& reason) {
cout << name << " exited: " << to_string(reason) << endl;
});
}
// --(rst-attach-end)--
enum class op : uint8_t {
ping,
pong,
};
behavior ping(event_based_actor* self, size_t num_pings) {
auto count = std::make_shared<size_t>(0);
return {
[=](ok_atom, const actor& pong) {
self->send(pong, ping_atom_v, int32_t{1});
self->become([=](pong_atom, int32_t value) -> result<ping_atom, int32_t> {
if (++*count >= num_pings)
self->quit();
return {ping_atom_v, value + 1};
});
},
};
}
behavior pong() {
return {
[](ping_atom, int32_t value) -> result<pong_atom, int32_t> {
return {pong_atom_v, value};
},
};
}
// Utility function for sending an integer type.
template <class T>
void write_int(broker* self, connection_handle hdl, T value) {
using unsigned_type = typename std::make_unsigned<T>::type;
if constexpr (sizeof(T) > 1) {
auto cpy = static_cast<T>(htonl(static_cast<unsigned_type>(value)));
self->write(hdl, sizeof(T), &cpy);
} else {
self->write(hdl, sizeof(T), &value);
}
}
// Utility function for reading an integer from incoming data.
template <class T>
void read_int(const void* data, T& storage) {
using unsigned_type = typename std::make_unsigned<T>::type;
memcpy(&storage, data, sizeof(T));
if constexpr (sizeof(T) > 1)
storage = static_cast<T>(ntohl(static_cast<unsigned_type>(storage)));
}
// Implementation of our broker.
behavior broker_impl(broker* self, connection_handle hdl, const actor& buddy) {
// We assume io_fsm manages a broker with exactly one connection,
// i.e., the connection ponted to by `hdl`.
assert(self->num_connections() == 1);
// Monitor buddy to quit broker if buddy is done.
self->monitor(buddy);
self->set_down_handler([=](down_msg& dm) {
if (dm.source == buddy) {
aout(self) << "our buddy is down" << endl;
// Quit for same reason.
self->quit(dm.reason);
}
});
// Setup: we are exchanging only messages consisting of an operation
// (as uint8_t) and an integer value (int32_t).
self->configure_read(hdl, receive_policy::exactly(sizeof(uint8_t)
+ sizeof(int32_t)));
// Our message handlers.
return {
[=](const connection_closed_msg& msg) {
// Brokers can multiplex any number of connections, however
// this example assumes io_fsm to manage a broker with
// exactly one connection.
if (msg.handle == hdl) {
aout(self) << "connection closed" << endl;
// force buddy to quit
self->send_exit(buddy, exit_reason::remote_link_unreachable);
self->quit(exit_reason::remote_link_unreachable);
}
},
[=](ping_atom, int32_t i) {
aout(self) << "send {ping, " << i << "}" << endl;
write_int(self, hdl, static_cast<uint8_t>(op::ping));
write_int(self, hdl, i);
self->flush(hdl);
},
[=](pong_atom, int32_t i) {
aout(self) << "send {pong, " << i << "}" << endl;
write_int(self, hdl, static_cast<uint8_t>(op::pong));
write_int(self, hdl, i);
self->flush(hdl);
},
[=](const new_data_msg& msg) {
// Keeps track of our position in the buffer.
auto rd_pos = msg.buf.data();
// Read the operation value as uint8_t from the buffer.
auto op_val = uint8_t{0};
read_int(rd_pos, op_val);
++rd_pos;
// Read the integer value from the buffer.
auto ival = int32_t{0};
read_int(rd_pos, ival);
// Show some output.
// Send composed message to our buddy.
switch (static_cast<op>(op_val)) {
case op::ping:
aout(self) << "received {ping, " << ival << "}" << endl;
self->send(buddy, ping_atom_v, ival);
break;
case op::pong:
aout(self) << "received {pong, " << ival << "}" << endl;
self->send(buddy, pong_atom_v, ival);
break;
default:
aout(self) << "invalid value for op_val, stop" << endl;
self->quit(sec::invalid_argument);
}
},
};
}
behavior server(broker* self, const actor& buddy) {
aout(self) << "server is running" << endl;
return {
[=](const new_connection_msg& msg) {
aout(self) << "server accepted new connection" << endl;
// By forking into a new broker, we are no longer
// responsible for the connection.
auto impl = self->fork(broker_impl, msg.handle, buddy);
print_on_exit(impl, "broker_impl");
aout(self) << "quit server (only accept 1 connection)" << endl;
self->quit();
},
};
}
class config : public actor_system_config {
public:
uint16_t port = 0;
std::string host = "localhost";
bool server_mode = false;
config() {
opt_group{custom_options_, "global"}
.add(port, "port,p", "set port")
.add(host, "host,H", "set host (ignored in server mode)")
.add(server_mode, "server-mode,s", "enable server mode");
}
};
void run_server(actor_system& system, const config& cfg) {
cout << "run in server mode" << endl;
auto pong_actor = system.spawn(pong);
auto server_actor = system.middleman().spawn_server(server, cfg.port,
pong_actor);
if (!server_actor) {
std::cerr << "failed to spawn server: " << to_string(server_actor.error())
<< endl;
return;
}
print_on_exit(*server_actor, "server");
print_on_exit(pong_actor, "pong");
}
void run_client(actor_system& system, const config& cfg) {
auto ping_actor = system.spawn(ping, size_t{20});
auto io_actor = system.middleman().spawn_client(broker_impl, cfg.host,
cfg.port, ping_actor);
if (!io_actor) {
std::cerr << "failed to spawn client: " << to_string(io_actor.error())
<< endl;
return;
}
print_on_exit(ping_actor, "ping actor");
print_on_exit(*io_actor, "broker");
send_as(*io_actor, ping_actor, ok_atom_v, *io_actor);
}
void caf_main(actor_system& system, const config& cfg) {
auto f = cfg.server_mode ? run_server : run_client;
f(system, cfg);
}
} // namespace
CAF_MAIN(io::middleman)
#include <chrono>
#include <iostream>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
using std::cerr;
using std::cout;
using std::endl;
using namespace caf;
using namespace caf::io;
namespace {
constexpr const char http_ok[] = R"__(HTTP/1.1 200 OK
Content-Type: text/plain
Connection: keep-alive
Transfer-Encoding: chunked
d
Hi there! :)
0
)__";
template <size_t Size>
constexpr size_t cstr_size(const char (&)[Size]) {
return Size;
}
behavior connection_worker(broker* self, connection_handle hdl) {
self->configure_read(hdl, receive_policy::at_most(1024));
return {
[=](const new_data_msg& msg) {
self->write(msg.handle, cstr_size(http_ok), http_ok);
self->quit();
},
[=](const connection_closed_msg&) { self->quit(); },
};
}
behavior server(broker* self) {
auto counter = std::make_shared<int>(0);
self->set_down_handler([=](down_msg&) { ++*counter; });
self->delayed_send(self, std::chrono::seconds(1), tick_atom_v);
return {
[=](const new_connection_msg& ncm) {
auto worker = self->fork(connection_worker, ncm.handle);
self->monitor(worker);
self->link_to(worker);
},
[=](tick_atom) {
aout(self) << "Finished " << *counter << " requests per second." << endl;
*counter = 0;
self->delayed_send(self, std::chrono::seconds(1), tick_atom_v);
},
};
}
class config : public actor_system_config {
public:
uint16_t port = 0;
config() {
opt_group{custom_options_, "global"}.add(port, "port,p", "set port");
}
};
void caf_main(actor_system& system, const config& cfg) {
auto server_actor = system.middleman().spawn_server(server, cfg.port);
if (!server_actor) {
cerr << "*** cannot spawn server: " << to_string(server_actor.error())
<< endl;
return;
}
cout << "*** listening on port " << cfg.port << endl;
cout << "*** to quit the program, simply press <enter>" << endl;
// wait for any input
std::string dummy;
std::getline(std::cin, dummy);
// kill server
anon_send_exit(*server_actor, exit_reason::user_shutdown);
}
} // namespace
CAF_MAIN(io::middleman)
// Non-interactive example to illustrate how to connect flows over an
// asynchronous SPSC (Single Producer Single Consumer) buffer.
// asynchronous SPSC (Single Producer Single Consumer) buffer manually. Usually,
// CAF generates the SPSC buffers implicitly.
#include "caf/actor_system.hpp"
#include "caf/async/spsc_buffer.hpp"
......
......@@ -416,10 +416,14 @@ the actor has already exited. Otherwise, the actor will execute it as part of
its termination. The following example attaches a function object to actors for
printing a custom string on exit.
.. literalinclude:: /examples/broker/simple_broker.cpp
:language: C++
:start-after: --(rst-attach-begin)--
:end-before: --(rst-attach-end)--
.. code-block:: C++
// Utility function to print an exit message with custom name.
void print_on_exit(const actor& hdl, const std::string& name) {
hdl->attach_functor([=](const error& reason) {
cout << name << " exited: " << to_string(reason) << endl;
});
}
It is possible to attach code to remote actors. However, the cleanup code will
run on the local machine.
......
......@@ -3,6 +3,14 @@
Network I/O with Brokers
========================
.. note::
We no longer recommend the Brokers from the I/O module for new code. Please
consider using Networking Module instead: :ref:`net_overview`. The new caf-net
module offers higher-level abstractions that interface with flows (see
:ref:`data_flows`). Message-based communication is also available with an
:ref:`actor_shell`.
When communicating to other services in the network, sometimes low-level socket
I/O is inevitable. For this reason, CAF provides *brokers*. A broker is
an event-based actor running in the middleman that multiplexes socket I/O. It
......
......@@ -135,7 +135,7 @@ adds three options to the ``global`` category.
.. literalinclude:: /examples/remoting/distributed_calculator.cpp
:language: C++
:begin-after: --(rst-config-begin)--
:start-after: --(rst-config-begin)--
:end-before: --(rst-config-end)--
We create a new ``global`` category in ``custom_options_``. Each following call
......
.. _worker-groups:
Managing Groups of Workers :sup:`experimental`
===============================================
Managing Groups of Workers :sup:`experimental`
==============================================
When managing a set of workers, a central actor often dispatches requests to a
set of workers. For this purpose, the class ``actor_pool`` implements a
......
......@@ -43,8 +43,9 @@ Copy on Write
-------------
CAF allows multiple actors to implicitly share message contents, as long as no
actor performs writes. This allows groups (see :ref:`groups`) to send the same
content to all subscribed actors without any copying overhead.
actor performs writes. This allows sending the same message to multiple
receivers without copying overhead, as long as all receivers only read the
content of the message.
Actors copy message contents whenever other actors hold references to it and if
one or more arguments of a message handler take a mutable reference.
......
.. _utility:
Utility
=======
CAF includes a few utility classes that are likely to be part of C++
eventually (or already are in newer versions of the standard). However, until
these classes are part of the standard library on all supported compilers, we
unfortunately have to maintain our own implementations.
.. _optional:
Class ``optional``
------------------
Represents a value that may or may not exist.
+-----------------------------+---------------------------------------------+
| **Constructors** | |
+-----------------------------+---------------------------------------------+
| ``(T value)`` | Constructs an object with a value. |
+-----------------------------+---------------------------------------------+
| ``(none_t = none)`` | Constructs an object without a value. |
+-----------------------------+---------------------------------------------+
| | |
+-----------------------------+---------------------------------------------+
| **Observers** | |
+-----------------------------+---------------------------------------------+
| ``explicit operator bool()``| Checks whether the object contains a value. |
+-----------------------------+---------------------------------------------+
| ``T* operator->()`` | Accesses the contained value. |
+-----------------------------+---------------------------------------------+
| ``T& operator*()`` | Accesses the contained value. |
+-----------------------------+---------------------------------------------+
Class ``expected``
------------------
Represents the result of a computation that *should* return a value. If no value
could be produced, the ``expected<T>`` contains an ``error`` (see :ref:`error`).
+-----------------------------+---------------------------------------------+
| **Constructors** | |
+-----------------------------+---------------------------------------------+
| ``(T value)`` | Constructs an object with a value. |
+-----------------------------+---------------------------------------------+
| ``(error err)`` | Constructs an object with an error. |
+-----------------------------+---------------------------------------------+
| | |
+-----------------------------+---------------------------------------------+
| **Observers** | |
+-----------------------------+---------------------------------------------+
| ``explicit operator bool()``| Checks whether the object contains a value. |
+-----------------------------+---------------------------------------------+
| ``T* operator->()`` | Accesses the contained value. |
+-----------------------------+---------------------------------------------+
| ``T& operator*()`` | Accesses the contained value. |
+-----------------------------+---------------------------------------------+
| ``error& error()`` | Accesses the contained error. |
+-----------------------------+---------------------------------------------+
Constant ``unit``
-----------------
The constant ``unit`` of type ``unit_t`` is the equivalent of
``void`` and can be used to initialize ``optional<void>`` and
``expected<void>``.
Constant ``none``
-----------------
The constant ``none`` of type ``none_t`` can be used to
initialize an ``optional<T>`` to represent "nothing".
......@@ -110,7 +110,7 @@ rst_epilog = """
#
# This is also used if you do content translation via gettext catalogs.
# Usually you set "language" from the command line for these cases.
language = None
language = "en"
# There are two options for replacing |today|: either, you set today to some
# non-false value, then it is used:
......
......@@ -39,14 +39,15 @@ Contents
:caption: Networking Library
net/Overview
net/Prometheus
net/ActorShell
net/HTTP
net/LengthPrefixFraming
net/Prometheus
.. toctree::
:maxdepth: 2
:caption: Appendix
FAQ
Utility
CommonPitfalls
UsingAout
.. _actor_shell:
Actor Shell
===========
TODO
.. _net_http:
HTTP :sup:`experimental`
========================
TODO
.. _net_prometheus:
.. _length_prefix_framing:
Length-prefix Framing
=====================
Length-prefix Framing :sup:`experimental`
=========================================
Length-prefix framing is a simple protocol for encoding variable-length messages
on the network. Each message is preceded by a fixed-length value (32 bit) that
......
......@@ -3,12 +3,15 @@
Overview
========
The networking module offers high-level APIs for individual protocols as well as
low-level building blocks for building implementing new protocols and assembling
protocol stacks.
The networking module offers a high-level, declarative DSL for individual
protocols as well as low-level building blocks for implementing new protocols
and assembling protocol stacks.
High-level APIs
===============
When using caf-net for applications, we generally recommend sticking to the
declarative API.
Declarative High-level DSL :sup:`experimental`
----------------------------------------------
The high-level APIs follow a factory pattern that configures each layer from the
bottom up, usually starting at the actor system. For example:
......@@ -82,4 +85,4 @@ for the handshake such as protocols or extensions fields. These options are
listed at the documentation for the individual protocols.
Layering
========
--------
.. _net_prometheus:
Prometheus
==========
Prometheus :sup:`experimental`
==============================
CAF ships a Prometheus server implementation that allows a scraper to collect
metrics from the actor system. The Prometheus server sits on top of an HTTP
......
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