Commit ba247d65 authored by Dominik Charousset's avatar Dominik Charousset

updated documentation, updated ChangeLog

parent db667857
Version 0.9.0
-------------
__2014_XX_XX__
__2014_04_05__
- Redesigned large parts of the library related to type-safe actors:
* No more thread-local `self`, because it can only return a type-erased handle
......@@ -12,6 +12,8 @@ __2014_XX_XX__
* New header `system_messages.hpp` for message types used by the runtime
- Announce properly handles empty & POD types as well as enums
- Brokers now use proper message types rather than 'IO_*' atom prefixed tuples
- Brokers can be spawned 'empty' and add connections/acceptors later on
- New work-stealing scheduler
Version 0.8.2
-------------
......
......@@ -630,15 +630,19 @@ typed_remote_actor(const std::string& host, std::uint16_t port) {
}
/**
* @brief Spawns an IO actor of type @p Impl.
* @param args Constructor arguments.
* @tparam Impl Subtype of {@link io::broker}.
* @brief Spawns a new, function-based IO actor.
* @param fun A functor implementing the actor's behavior.
* @param in The actor's input stream.
* @param out The actor's output stream.
* @param args Optional arguments for @p fun.
* @tparam Os Optional flags to modify <tt>spawn</tt>'s behavior.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
* @returns A {@link actor handle} to the spawned actor.
*/
template<class Impl, spawn_options Os = no_spawn_options, typename... Ts>
actor spawn_io(Ts&&... args) {
auto ptr = make_counted<Impl>(std::forward<Ts>(args)...);
template<spawn_options Os = no_spawn_options,
typename F = std::function<void (io::broker*)>,
typename... Ts>
actor spawn_io(F fun, Ts&&... args) {
auto ptr = io::broker::from(std::move(fun), std::forward<Ts>(args)...);
return {io::init_and_launch(std::move(ptr))};
}
......@@ -654,10 +658,10 @@ actor spawn_io(Ts&&... args) {
template<spawn_options Os = no_spawn_options,
typename F = std::function<void (io::broker*)>,
typename... Ts>
actor spawn_io(F fun,
io::input_stream_ptr in,
io::output_stream_ptr out,
Ts&&... args) {
actor spawn_io_client(F fun,
io::input_stream_ptr in,
io::output_stream_ptr out,
Ts&&... args) {
auto ptr = io::broker::from(std::move(fun), std::move(in), std::move(out),
std::forward<Ts>(args)...);
return {io::init_and_launch(std::move(ptr))};
......@@ -666,9 +670,9 @@ actor spawn_io(F fun,
template<spawn_options Os = no_spawn_options,
typename F = std::function<void (io::broker*)>,
typename... Ts>
actor spawn_io(F fun, const std::string& host, uint16_t port, Ts&&... args) {
actor spawn_io_client(F fun, const std::string& host, uint16_t port, Ts&&... args) {
auto ptr = io::tcp_io_stream::connect_to(host.c_str(), port);
return spawn_io(std::move(fun), ptr, ptr, std::forward<Ts>(args)...);
return spawn_io_client(std::move(fun), ptr, ptr, std::forward<Ts>(args)...);
}
template<spawn_options Os = no_spawn_options,
......
......@@ -85,54 +85,43 @@ class broker : public extend<local_actor>::
friend broker_ptr init_and_launch(broker_ptr);
broker() = delete;
public:
~broker();
/**
* @brief Used to configure {@link receive_policy()}.
*/
enum policy_flag { at_least, at_most, exactly };
void enqueue(msg_hdr_cref, any_tuple, execution_unit*) override;
bool initialized() const;
/**
* @brief Modifies the receive policy for this broker.
* @param hdl Identifies the affected connection.
* @param policy Sets the policy for given buffer size.
* @param buffer_size Sets the minimal, maximum, or exact number of bytes
* the middleman should read on this connection
* before sending the next {@link new_data_msg}.
*/
void receive_policy(const connection_handle& hdl,
broker::policy_flag policy,
size_t buffer_size);
/**
* @brief Sends data.
*/
void write(const connection_handle& hdl, size_t num_bytes, const void* buf);
/**
* @brief Sends data.
*/
void write(const connection_handle& hdl, const util::buffer& buf);
/**
* @brief Sends data.
*/
void write(const connection_handle& hdl, util::buffer&& buf);
template<typename F, typename... Ts>
static broker_ptr from(F fun,
input_stream_ptr in,
output_stream_ptr out,
Ts&&... args) {
auto hdl = connection_handle::from_int(in->read_handle());
return from_impl(std::bind(std::move(fun),
std::placeholders::_1,
hdl,
std::forward<Ts>(args)...),
std::move(in),
std::move(out));
}
static broker_ptr from(std::function<void (broker*)> fun, acceptor_uptr in);
static broker_ptr from(std::function<behavior (broker*)> fun, acceptor_uptr in);
template<typename F, typename T0, typename... Ts>
static broker_ptr from(F fun, acceptor_uptr in, T0&& arg0, Ts&&... args) {
return from(std::bind(std::move(fun),
std::placeholders::_1,
std::forward<T0>(arg0),
std::forward<Ts>(args)...),
std::move(in));
}
/** @cond PRIVATE */
template<typename F, typename... Ts>
actor fork(F fun, connection_handle hdl, Ts&&... args) {
......@@ -143,11 +132,6 @@ class broker : public extend<local_actor>::
hdl);
}
template<typename F>
inline void for_each_connection(F fun) const {
for (auto& kvp : m_io) fun(kvp.first);
}
inline size_t num_connections() const {
return m_io.size();
}
......@@ -168,12 +152,49 @@ class broker : public extend<local_actor>::
return add_acceptor(tcp_acceptor::from_sockfd(tcp_sockfd));
}
void enqueue(msg_hdr_cref, any_tuple, execution_unit*) override;
bool initialized() const;
template<typename F, typename... Ts>
static broker_ptr from(F fun,
input_stream_ptr in,
output_stream_ptr out,
Ts&&... args) {
auto hdl = connection_handle::from_int(in->read_handle());
return from_impl(std::bind(std::move(fun),
std::placeholders::_1,
hdl,
std::forward<Ts>(args)...),
std::move(in),
std::move(out));
}
static broker_ptr from(std::function<void (broker*)> fun);
static broker_ptr from(std::function<behavior (broker*)> fun);
static broker_ptr from(std::function<void (broker*)> fun, acceptor_uptr in);
static broker_ptr from(std::function<behavior (broker*)> fun, acceptor_uptr in);
template<typename F, typename T0, typename... Ts>
static broker_ptr from(F fun, acceptor_uptr in, T0&& arg0, Ts&&... args) {
return from(std::bind(std::move(fun),
std::placeholders::_1,
std::forward<T0>(arg0),
std::forward<Ts>(args)...),
std::move(in));
}
protected:
broker(input_stream_ptr in, output_stream_ptr out);
broker(acceptor_uptr in);
broker();
void cleanup(std::uint32_t reason) override;
typedef std::unique_ptr<broker::scribe> scribe_pointer;
......@@ -184,6 +205,8 @@ class broker : public extend<local_actor>::
virtual behavior make_behavior() = 0;
/** @endcond */
private:
actor fork_impl(std::function<void (broker*)> fun,
......@@ -218,26 +241,6 @@ class broker : public extend<local_actor>::
};
class default_broker : public broker {
public:
typedef std::function<behavior (broker*)> function_type;
default_broker(function_type f, input_stream_ptr in, output_stream_ptr out);
default_broker(function_type f, scribe_pointer ptr);
default_broker(function_type f, acceptor_uptr ptr);
behavior make_behavior() override;
private:
function_type m_fun;
};
} // namespace io
} // namespace cppa
......
......@@ -190,7 +190,7 @@ int main(int argc, char** argv) {
},
on("-c", val<string>, as_u16) >> [&](const string& host, uint16_t port) {
auto ping_actor = spawn(ping, 20);
auto io_actor = spawn_io(protobuf_io, host, port, ping_actor);
auto io_actor = spawn_io_client(protobuf_io, host, port, ping_actor);
print_on_exit(io_actor, "protobuf_io");
print_on_exit(ping_actor, "ping");
send_as(io_actor, ping_actor, atom("kickoff"), io_actor);
......
......@@ -159,7 +159,7 @@ int main(int argc, char** argv) {
},
on("-c", val<string>, as_u16) >> [&](const string& host, uint16_t port) {
auto ping_actor = spawn(ping, 20);
auto io_actor = spawn_io(broker_impl, host, port, ping_actor);
auto io_actor = spawn_io_client(broker_impl, host, port, ping_actor);
print_on_exit(io_actor, "protobuf_io");
print_on_exit(ping_actor, "ping");
send_as(io_actor, ping_actor, atom("kickoff"), io_actor);
......
......@@ -21,7 +21,7 @@ Represents an optional value.
\lstinline^type^ & \lstinline^T^ \\
\hline
\\
\multicolumn{2}{l}{\large{\textbf{Member functions}}\vspace{3pt}} \\
\multicolumn{2}{l}{\large{\textbf{Member Functions}}\vspace{3pt}} \\
\hline
\lstinline^option()^ & Constructs an empty option \\
\hline
......
......@@ -47,8 +47,9 @@ Please submit a bug report that includes (a) your compiler version, (b) your OS,
\item Linux
\item Mac OS X
\item \textit{Note for MS Windows}:
\libcppa relies on C++11 features such as variadic templates.
\libcppa relies on C++11 features such as unrestricted unions.
We will support this platform as soon as Microsoft's compiler implements all required C++11 features.
In the meantime, libcppa can be used with MinGW.
\end{itemize*}
\clearpage
......
......@@ -26,7 +26,22 @@ The \lstinline^"local"^ group module creates groups for in-process communication
For example, a group for GUI related events could be identified by \lstinline^group::get("local", "GUI events")^.
The group ID \lstinline^"GUI events"^ uniquely identifies a singleton group instance of the module \lstinline^"local"^.
\subsection{Spawn Actors in Groups}
\subsection{Remote Groups}
\label{Sec::Group::RemoteGroups}
To deploy groups in a network, one host can act as group server by publishing its local groups at any given port:
\begin{lstlisting}
void publish_local_groups(std::uint16_t port, const char* addr)
\end{lstlisting}
By calling \lstinline^group::get("remote", "<group>@<host>:<port>")^, other hosts are now able to connect to a remotely running group.
Please note that the group communication is no longer available once the server disconnects.
This implementation uses N-times unicast underneath.
It is worth mentioning that user-implemented groups can be build on top of IP multicast or overlay technologies such as Scribe to achieve better performance or reliability.
\clearpage
\subsection{Spawning Actors in Groups}
\label{Sec::Group::Spawn}
The function \lstinline^spawn_in_group^ can be used to create actors as members of a group.
......
\section{Network IO}
\label{Sec::NetworkIO}
When communicating to other services in the network, sometimes low-level socket IO is inevitable.
For this reason, \libcppa provides \emph{brokers}.
A broker is an event-based actor running in the middleman that multiplexes socket IO.
It can maintain any number of acceptors and connections.
Since the broker runs in the middleman, implementations should be careful to consume as little time as possible in message handlers.
Any considerable amount work should outsourced by spawning new actors (or maintaining worker actors).
\subsection{Spawning Brokers}
Brokers are spawned using the function \lstinline^spawn_io^ and always use functor-based implementations capturing the self pointer of type \lstinline^broker*^.
For convenience, \lstinline^spawn_io_server^ can be used to spawn a new broker listening to a local port and \lstinline^spawn_io_client^ can be used to spawn a new broker that connects to given host and port or uses existing IO streams.
\begin{lstlisting}
template<spawn_options Os = no_spawn_options,
typename F = std::function<behavior (io::broker*)>,
typename... Ts>
actor spawn_io(F fun, Ts&&... args);
template<spawn_options Os = no_spawn_options,
typename F = std::function<behavior (io::broker*)>,
typename... Ts>
actor spawn_io_client(F fun,
io::input_stream_ptr in,
io::output_stream_ptr out,
Ts&&... args);
template<spawn_options Os = no_spawn_options,
typename F = std::function<behavior (io::broker*)>,
typename... Ts>
actor spawn_io_client(F fun, string host, uint16_t port, Ts&&... args);
template<spawn_options Os = no_spawn_options,
typename F = std::function<behavior (io::broker*)>,
typename... Ts>
actor spawn_io_server(F fun, uint16_t port, Ts&&... args);
\end{lstlisting}
\clearpage
\subsection{Broker Interface}
\begin{lstlisting}
class io::broker;
\end{lstlisting}
{\small
\begin{tabular*}{\textwidth}{m{0.51\textwidth}m{0.44\textwidth}}
\multicolumn{2}{m{\linewidth}}{\large{\textbf{Data Types}}\vspace{3pt}} \\
\\
\hline
\multicolumn{2}{m{\linewidth}}{\texttt{enum policy\_flag \{ at\_least, at\_most, exactly \}}} \\
\hline
\\
\multicolumn{2}{m{\linewidth}}{\large{\textbf{Member Functions}}\vspace{3pt}} \\
\\
\hline
\lstinline^void receive_policy(^ \lstinline^connection_handle hdl,^ \lstinline^policy_flag policy,^ \lstinline^size_t buffer_size)^ & Modifies the receive policy for the connection identified by \lstinline^hdl^. This will cause the middleman to enqueue the next \lstinline^new_data_msg^ after \emph{at least}, \emph{at most}, or \emph{exactly} \lstinline^buffer_size^ bytes have been read \\
\hline
\lstinline^void write(connection_handle hdl,^ \lstinline^size_t num_bytes,^ \lstinline^const void* buf)^ & Sends data \\
\hline
\lstinline^void write(connection_handle hdl,^ \lstinline^const util::buffer& buf)^ & Sends data \\
\hline
\lstinline^void write(connection_handle hdl,^ \lstinline^util::buffer&& buf)^ & Sends data \\
\hline
\lstinline^template<typename F, typename... Ts>^ \lstinline^actor fork(F fun,^ \lstinline^connection_handle hdl, Ts&&... args)^ & Spawns a new broker that takes ownership of given connection \\
\hline
\lstinline^size_t num_connections()^ & Returns the number of open connections \\
\hline
\lstinline^connection_handle add_connection(^ \lstinline^input_stream_ptr in,^ \lstinline^output_stream_ptr out)^ & Adds a new connection from input and output stream \\
\hline
\lstinline^connection_handle add_connection(^ \lstinline^stream_ptr sptr)^ & Adds a new connection from an IO stream \\
\hline
\lstinline^connection_handle add_tcp_connection(^ \lstinline^native_socket_type tcp_sockfd)^ & Adds a new connection from a native TCP socket descriptor \\
\hline
\lstinline^accept_handle add_acceptor(^ \lstinline^acceptor_uptr ptr)^ & Adds a new acceptor \\
\hline
\lstinline^accept_handle add_tcp_acceptor(^ \lstinline^native_socket_type tcp_sockfd)^ & Adds a new acceptor from a native TCP socket descriptor \\
\hline
\end{tabular*}
}
\clearpage
\subsection{Broker-related Message Types}
Brokers, just like any other dynamically typed actor, can receive messages of any type.
However, it also receives system messages from the middleman:
\begin{lstlisting}
struct new_connection_msg {
io::accept_handle source;
io::connection_handle handle;
};
\end{lstlisting}
A \lstinline^new_connection_msg^ is received whenever a new incoming connection (identified by the \lstinline^handle^ field) has been accepted for one of the broker's accept handles (identified by the \lstinline^source^ field).
\begin{lstlisting}
struct new_data_msg {
io::connection_handle handle;
util::buffer buf;
};
\end{lstlisting}
A \lstinline^new_data_msg^ is received whenever data on a connection is ready.
The data can be accessed as buffer object (see \ref{Appendix::Buffer}).
The amount of data, i.e., how often this message is received, can be configured using the brokers receive policy (see \ref{Sec::NetworkIO::ReceivePolicy}).
It is worth mentioning that the buffer is re-used whenever possible.
This means, as long as the broker does not create any new references to the message by copying it, the middleman will always use only a single buffer per connection.
\begin{lstlisting}
struct connection_closed_msg {
io::connection_handle handle;
};
\end{lstlisting}
A \lstinline^connection_closed_msg^ informs the broker that one of its connections has been closed.
\begin{lstlisting}
struct acceptor_closed_msg {
io::accept_handle handle;
};
\end{lstlisting}
A \lstinline^ acceptor_closed_msg^ informs the broker that of its acceptors has been closed.
......@@ -33,7 +33,7 @@ Actors can also use the result of a \lstinline^sync_send^ to answer to a request
\begin{lstlisting}
void client(event_based_actor* self, const actor& master) {
become (
on("foo", arg_match) >> [=](const string& request) -> string {
on("foo", arg_match) >> [=](const string& request) {
return self->sync_send(master, atom("bar"), request).then(
on_arg_match >> [=](const std::string& response) {
return response;
......
......@@ -45,7 +45,7 @@
%BEGIN LATEX
\texttt{\huge{\textbf{libcppa}}}\\~\\A C++ library for actor programming\\~\\~\\~\\%
%END LATEX
User Manual\\\normalsize{\texttt{libcppa} version 0.9.0} PRERELEASE\vfill}
User Manual\\\normalsize{\texttt{libcppa} version 0.9.0}\vfill}
\author{Dominik Charousset}
......@@ -124,6 +124,7 @@ User Manual\\\normalsize{\texttt{libcppa} version 0.9.0} PRERELEASE\vfill}
\include{SpawningActors}
\include{MessagePriorities}
\include{NetworkTransparency}
\include{NetworkIO}
\include{GroupCommunication}
\include{ActorCompanions}
\include{TypeSystem}
......
......@@ -58,31 +58,45 @@ constexpr size_t default_max_buffer_size = 65535;
} // namespace <anonymous>
default_broker::default_broker(function_type f,
input_stream_ptr in,
output_stream_ptr out)
: broker(std::move(in), std::move(out)), m_fun(std::move(f)) { }
default_broker::default_broker(function_type f, scribe_pointer ptr)
: broker(std::move(ptr)), m_fun(std::move(f)) { }
class default_broker : public broker {
default_broker::default_broker(function_type f, acceptor_uptr ptr)
: broker(std::move(ptr)), m_fun(std::move(f)) { }
public:
behavior default_broker::make_behavior() {
CPPA_PUSH_AID(id());
CPPA_LOG_TRACE("");
enqueue({invalid_actor_addr, channel{this}},
make_any_tuple(atom("INITMSG")),
nullptr);
return (
on(atom("INITMSG")) >> [=] {
unbecome();
auto bhvr = m_fun(this);
if (bhvr) become(std::move(bhvr));
}
);
}
typedef std::function<behavior (broker*)> function_type;
default_broker(function_type f) : m_fun(std::move(f)) { }
default_broker(function_type f, input_stream_ptr in, output_stream_ptr out)
: broker(std::move(in), std::move(out)), m_fun(std::move(f)) { }
default_broker(function_type f, scribe_pointer ptr)
: broker(std::move(ptr)), m_fun(std::move(f)) { }
default_broker(function_type f, acceptor_uptr ptr)
: broker(std::move(ptr)), m_fun(std::move(f)) { }
behavior make_behavior() override {
CPPA_PUSH_AID(id());
CPPA_LOG_TRACE("");
enqueue({invalid_actor_addr, channel{this}},
make_any_tuple(atom("INITMSG")),
nullptr);
return (
on(atom("INITMSG")) >> [=] {
unbecome();
auto bhvr = m_fun(this);
if (bhvr) become(std::move(bhvr));
}
);
}
private:
function_type m_fun;
};
class broker::continuation {
......@@ -125,11 +139,6 @@ class broker::servant : public continuable {
void dispose() override {
auto ptr = m_broker;
ptr->erase_io(read_handle());
if (ptr->m_io.empty() && ptr->m_accept.empty()) {
// release implicit reference count held by middleman
// in caes no reader/writer is left for this broker
ptr->deref();
}
}
void set_broker(broker_ptr new_broker) {
......@@ -364,11 +373,15 @@ void broker::invoke_message(msg_hdr_cref hdr, any_tuple msg) {
// cleanup if needed
if (planned_exit_reason() != exit_reason::not_exited) {
cleanup(planned_exit_reason());
// release implicit reference count held by MM
deref();
}
else if (bhvr_stack().empty()) {
CPPA_LOG_DEBUG("bhvr_stack().empty(), quit for normal exit reason");
quit(exit_reason::normal);
cleanup(planned_exit_reason());
// release implicit reference count held by MM
deref();
}
}
......@@ -405,6 +418,10 @@ void broker::init_broker() {
get_actor_registry()->inc_running();
}
broker::broker() {
init_broker();
}
broker::broker(input_stream_ptr in, output_stream_ptr out) {
using namespace std;
init_broker();
......@@ -483,6 +500,17 @@ broker_ptr broker::from_impl(std::function<void (broker*)> fun,
return make_counted<default_broker>(f, move(in), move(out));
}
broker_ptr broker::from(std::function<behavior (broker*)> fun) {
return make_counted<default_broker>(fun);
}
broker_ptr broker::from(std::function<void (broker*)> fun) {
return from([=](broker* self) -> behavior {
fun(self);
return {};
});
}
broker_ptr broker::from(std::function<behavior (broker*)> fun, acceptor_uptr in) {
return make_counted<default_broker>(move(fun), move(in));
}
......
......@@ -143,7 +143,7 @@ int main(int argc, char** argv) {
CPPA_CHECKPOINT();
auto p = spawn(ping, 10);
CPPA_CHECKPOINT();
auto cl = spawn_io(peer, "localhost", port, p);
auto cl = spawn_io_client(peer, "localhost", port, p);
CPPA_CHECKPOINT();
anon_send(p, atom("kickoff"), cl);
CPPA_CHECKPOINT();
......
......@@ -171,22 +171,6 @@ void testee1(event_based_actor* self) {
});
}
void testee2(event_based_actor* self, actor other) {
self->link_to(other);
self->send(other, uint32_t(1));
self->become (
on<uint32_t>() >> [=](uint32_t sleep_time) {
// "sleep" for sleep_time milliseconds
self->become (
keep_behavior,
after(chrono::milliseconds(sleep_time)) >> [=] {
self->unbecome();
}
);
}
);
}
template<class Testee>
string behavior_test(scoped_actor& self, actor et) {
string testee_name = detail::to_uniform_name(typeid(Testee));
......
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