Commit af56c6ee authored by Dominik Charousset's avatar Dominik Charousset

streamlined broker implementation

this patch changes the broker implementation, i.e., provides a single
broker class for both accept and I/O brokers
parent 9735fe76
...@@ -109,7 +109,6 @@ set(LIBCPPA_SRC ...@@ -109,7 +109,6 @@ set(LIBCPPA_SRC
src/binary_serializer.cpp src/binary_serializer.cpp
src/broker.cpp src/broker.cpp
src/buffer.cpp src/buffer.cpp
src/buffered_writer.cpp
src/channel.cpp src/channel.cpp
src/context_switching_actor.cpp src/context_switching_actor.cpp
src/continuable.cpp src/continuable.cpp
...@@ -133,7 +132,6 @@ set(LIBCPPA_SRC ...@@ -133,7 +132,6 @@ set(LIBCPPA_SRC
src/get_mac_addresses.cpp src/get_mac_addresses.cpp
src/group.cpp src/group.cpp
src/group_manager.cpp src/group_manager.cpp
src/io_service.cpp
src/ipv4_acceptor.cpp src/ipv4_acceptor.cpp
src/ipv4_io_stream.cpp src/ipv4_io_stream.cpp
src/local_actor.cpp src/local_actor.cpp
......
...@@ -81,7 +81,6 @@ cppa/intrusive/single_reader_queue.hpp ...@@ -81,7 +81,6 @@ cppa/intrusive/single_reader_queue.hpp
cppa/intrusive_ptr.hpp cppa/intrusive_ptr.hpp
cppa/io/acceptor.hpp cppa/io/acceptor.hpp
cppa/io/broker.hpp cppa/io/broker.hpp
cppa/io/buffered_writer.hpp
cppa/io/continuable.hpp cppa/io/continuable.hpp
cppa/io/default_actor_addressing.hpp cppa/io/default_actor_addressing.hpp
cppa/io/default_actor_proxy.hpp cppa/io/default_actor_proxy.hpp
...@@ -90,7 +89,7 @@ cppa/io/default_peer.hpp ...@@ -90,7 +89,7 @@ cppa/io/default_peer.hpp
cppa/io/default_peer_acceptor.hpp cppa/io/default_peer_acceptor.hpp
cppa/io/default_protocol.hpp cppa/io/default_protocol.hpp
cppa/io/input_stream.hpp cppa/io/input_stream.hpp
cppa/io/io_handle.hpp cppa/io/connection_handle.hpp
cppa/io/ipv4_acceptor.hpp cppa/io/ipv4_acceptor.hpp
cppa/io/ipv4_io_stream.hpp cppa/io/ipv4_io_stream.hpp
cppa/io/middleman.hpp cppa/io/middleman.hpp
...@@ -218,7 +217,6 @@ src/binary_deserializer.cpp ...@@ -218,7 +217,6 @@ src/binary_deserializer.cpp
src/binary_serializer.cpp src/binary_serializer.cpp
src/broker.cpp src/broker.cpp
src/buffer.cpp src/buffer.cpp
src/buffered_writer.cpp
src/channel.cpp src/channel.cpp
src/context_switching_actor.cpp src/context_switching_actor.cpp
src/continuable.cpp src/continuable.cpp
...@@ -242,7 +240,6 @@ src/get_mac_addresses.cpp ...@@ -242,7 +240,6 @@ src/get_mac_addresses.cpp
src/get_root_uuid.cpp src/get_root_uuid.cpp
src/group.cpp src/group.cpp
src/group_manager.cpp src/group_manager.cpp
src/io_service.cpp
src/ipv4_acceptor.cpp src/ipv4_acceptor.cpp
src/ipv4_io_stream.cpp src/ipv4_io_stream.cpp
src/local_actor.cpp src/local_actor.cpp
...@@ -309,3 +306,7 @@ unit_testing/test_sync_send.cpp ...@@ -309,3 +306,7 @@ unit_testing/test_sync_send.cpp
unit_testing/test_tuple.cpp unit_testing/test_tuple.cpp
unit_testing/test_uniform_type.cpp unit_testing/test_uniform_type.cpp
unit_testing/test_yield_interface.cpp unit_testing/test_yield_interface.cpp
cppa/io/accept_handle.hpp
unit_testing/test_broker.cpp
cppa/io/buffered_writing.hpp
cppa/detail/handle.hpp
...@@ -67,12 +67,16 @@ ...@@ -67,12 +67,16 @@
#include "cppa/event_based_actor.hpp" #include "cppa/event_based_actor.hpp"
#include "cppa/util/type_traits.hpp" #include "cppa/util/type_traits.hpp"
#include "cppa/io/acceptor.hpp"
#include "cppa/io/broker.hpp" #include "cppa/io/broker.hpp"
#include "cppa/io/acceptor.hpp"
#include "cppa/io/middleman.hpp" #include "cppa/io/middleman.hpp"
#include "cppa/io/io_handle.hpp"
#include "cppa/io/input_stream.hpp" #include "cppa/io/input_stream.hpp"
#include "cppa/io/output_stream.hpp" #include "cppa/io/output_stream.hpp"
#include "cppa/io/accept_handle.hpp"
#include "cppa/io/ipv4_acceptor.hpp"
#include "cppa/io/ipv4_io_stream.hpp"
#include "cppa/io/connection_handle.hpp"
#include "cppa/detail/memory.hpp" #include "cppa/detail/memory.hpp"
#include "cppa/detail/get_behavior.hpp" #include "cppa/detail/get_behavior.hpp"
...@@ -623,14 +627,9 @@ actor_ptr spawn_io(io::input_stream_ptr in, ...@@ -623,14 +627,9 @@ actor_ptr spawn_io(io::input_stream_ptr in,
io::output_stream_ptr out, io::output_stream_ptr out,
Ts&&... args) { Ts&&... args) {
using namespace io; using namespace io;
auto mm = get_middleman(); using namespace std;
auto ptr = make_counted<Impl>(std::move(in), std::move(out), std::forward<Ts>(args)...); auto ptr = make_counted<Impl>(move(in), move(out), forward<Ts>(args)...);
{ return eval_sopts(Options, io::init_and_launch(move(ptr)));
scoped_self_setter sss{ptr.get()};
ptr->init();
}
if (ptr->has_behavior()) mm->run_later([=] { mm->continue_reader(ptr.get()); });
return eval_sopts(Options, std::move(ptr));
} }
/** /**
...@@ -640,21 +639,40 @@ actor_ptr spawn_io(io::input_stream_ptr in, ...@@ -640,21 +639,40 @@ actor_ptr spawn_io(io::input_stream_ptr in,
* @returns An {@link actor_ptr} to the spawned {@link actor}. * @returns An {@link actor_ptr} to the spawned {@link actor}.
*/ */
template<spawn_options Options = no_spawn_options, template<spawn_options Options = no_spawn_options,
typename F = std::function<void (io::io_handle*)>, typename F = std::function<void (io::broker*)>,
typename... Ts> typename... Ts>
actor_ptr spawn_io(F fun, actor_ptr spawn_io(F fun,
io::input_stream_ptr in, io::input_stream_ptr in,
io::output_stream_ptr out, io::output_stream_ptr out,
Ts&&... args) { Ts&&... args) {
using namespace io; using namespace std;
auto mm = get_middleman(); auto ptr = io::broker::from(move(fun), move(in), move(out),
auto ptr = broker::from(std::move(fun), std::move(in), std::move(out), std::forward<Ts>(args)...); forward<Ts>(args)...);
{ return eval_sopts(Options, io::init_and_launch(move(ptr)));
scoped_self_setter sss{ptr.get()}; }
ptr->init();
} template<class Impl, spawn_options Options = no_spawn_options, typename... Ts>
// default_broker_impl will call continue_reader() actor_ptr spawn_io(const char* host, uint16_t port, Ts&&... args) {
return eval_sopts(Options, std::move(ptr)); auto ptr = io::ipv4_io_stream(host, port);
return spawn_io<Impl>(ptr, ptr, std::forward<Ts>(args)...);
}
template<spawn_options Options = no_spawn_options,
typename F = std::function<void (io::broker*)>,
typename... Ts>
actor_ptr spawn_io(F fun, const char* host, uint16_t port, Ts&&... args) {
auto ptr = io::ipv4_io_stream::connect_to(host, port);
return spawn_io(std::move(fun), ptr, ptr, std::forward<Ts>(args)...);
}
template<spawn_options Options = no_spawn_options,
typename F = std::function<void (io::broker*)>,
typename... Ts>
actor_ptr spawn_io_server(F fun, uint16_t port, Ts&&... args) {
using namespace std;
auto ptr = io::broker::from(move(fun), io::ipv4_acceptor::create(port),
forward<Ts>(args)...);
return eval_sopts(Options, io::init_and_launch(move(ptr)));
} }
/** /**
......
...@@ -80,6 +80,11 @@ class behavior_stack ...@@ -80,6 +80,11 @@ class behavior_stack
return m_elements.back().first; return m_elements.back().first;
} }
inline message_id back_id() {
CPPA_REQUIRE(!empty());
return m_elements.back().second;
}
inline void push_back(behavior&& what, inline void push_back(behavior&& what,
message_id response_id = message_id::invalid) { message_id response_id = message_id::invalid) {
m_elements.emplace_back(std::move(what), response_id); m_elements.emplace_back(std::move(what), response_id);
......
...@@ -28,53 +28,56 @@ ...@@ -28,53 +28,56 @@
\******************************************************************************/ \******************************************************************************/
#ifndef BUFFERED_WRITER_HPP #ifndef CPPA_DETAIL_HANDLE_HPP
#define BUFFERED_WRITER_HPP #define CPPA_DETAIL_HANDLE_HPP
#include "cppa/util/buffer.hpp" #include "cppa/util/comparable.hpp"
#include "cppa/io/output_stream.hpp" namespace cppa { namespace detail {
#include "cppa/io/continuable.hpp"
namespace cppa { namespace io { template<typename Subtype>
class handle : util::comparable<Subtype> {
class middleman; public:
class buffered_writer : public continuable { inline handle() : m_id{-1} { }
typedef continuable super; handle(const Subtype& other) {
m_id = other.id();
}
public: Subtype& operator=(const handle& other) {
m_id = other.id();
return *static_cast<Subtype*>(this);
}
buffered_writer(middleman* parent, inline int id() const {
native_socket_type read_fd, return m_id;
output_stream_ptr out); }
continue_writing_result continue_writing() override; inline int compare(const Subtype& other) const {
return m_id - other.id();
}
inline bool has_unwritten_data() const { inline bool invalid() const {
return m_has_unwritten_data; return m_id == -1;
} }
protected: static inline Subtype from_int(int id) {
return {id};
}
void write(size_t num_bytes, const void* data);
void register_for_writing(); protected:
inline util::buffer& write_buffer() { inline handle(int handle_id) : m_id{handle_id} { }
return m_buf;
}
private: private:
middleman* m_middleman; int m_id;
output_stream_ptr m_out;
bool m_has_unwritten_data;
util::buffer m_buf;
}; };
} } // namespace cppa::network } } // namespace cppa::detail
#endif // BUFFERED_WRITER_HPP #endif // CPPA_DETAIL_HANDLE_HPP
...@@ -204,11 +204,11 @@ class receive_policy { ...@@ -204,11 +204,11 @@ class receive_policy {
return client->await_message(); return client->await_message();
} }
private:
typedef typename rp_flag<rp_nestable>::type nestable; typedef typename rp_flag<rp_nestable>::type nestable;
typedef typename rp_flag<rp_sequential>::type sequential; typedef typename rp_flag<rp_sequential>::type sequential;
private:
std::list<std::unique_ptr<mailbox_element, disposer> > m_cache; std::list<std::unique_ptr<mailbox_element, disposer> > m_cache;
template<class Client> template<class Client>
...@@ -342,6 +342,7 @@ class receive_policy { ...@@ -342,6 +342,7 @@ class receive_policy {
client->m_current_node = previous; client->m_current_node = previous;
} }
public:
// workflow 'template' // workflow 'template'
......
...@@ -36,7 +36,6 @@ ...@@ -36,7 +36,6 @@
#include "cppa/detail/matches.hpp" #include "cppa/detail/matches.hpp"
#include "cppa/detail/tuple_vals.hpp" #include "cppa/detail/tuple_vals.hpp"
#include "cppa/detail/types_array.hpp" #include "cppa/detail/types_array.hpp"
#include "cppa/detail/object_array.hpp"
#include "cppa/detail/abstract_tuple.hpp" #include "cppa/detail/abstract_tuple.hpp"
namespace cppa { namespace detail { namespace cppa { namespace detail {
......
...@@ -48,6 +48,9 @@ ...@@ -48,6 +48,9 @@
#include "cppa/detail/singleton_mixin.hpp" #include "cppa/detail/singleton_mixin.hpp"
#include "cppa/io/accept_handle.hpp"
#include "cppa/io/connection_handle.hpp"
namespace cppa { class uniform_type_info; } namespace cppa { class uniform_type_info; }
namespace cppa { namespace detail { namespace cppa { namespace detail {
...@@ -61,6 +64,8 @@ using mapped_type_list = util::type_list< ...@@ -61,6 +64,8 @@ using mapped_type_list = util::type_list<
channel_ptr, channel_ptr,
group_ptr, group_ptr,
process_information_ptr, process_information_ptr,
io::accept_handle,
io::connection_handle,
message_header, message_header,
std::nullptr_t, std::nullptr_t,
util::buffer, util::buffer,
......
...@@ -28,43 +28,31 @@ ...@@ -28,43 +28,31 @@
\******************************************************************************/ \******************************************************************************/
#ifndef IO_SERVICE_HPP #ifndef CPPA_IO_ACCEPT_HANDLE_HPP
#define IO_SERVICE_HPP #define CPPA_IO_ACCEPT_HANDLE_HPP
#include <cstddef> #include "cppa/detail/handle.hpp"
namespace cppa { namespace io { namespace cppa { namespace io {
class io_handle { class broker;
public: class accept_handle : public detail::handle<accept_handle> {
friend class detail::handle<accept_handle>;
virtual ~io_handle(); typedef detail::handle<accept_handle> super;
/** public:
* @brief Denotes when an actor will receive a read buffer.
*/
enum policy_flag { at_least, at_most, exactly };
/** accept_handle() = default;
* @brief Closes the network connection.
*/
virtual void close() = 0;
/** private:
* @brief Asynchronously sends @p size bytes of @p data.
*/
virtual void write(size_t size, const void* data) = 0;
/** inline accept_handle(int handle_id) : super{handle_id} { }
* @brief Adjusts the rule receiving 'IO_receive' messages.
* The default settings are <tt>policy = io_handle::at_least</tt>
* and <tt>buffer_size = 0</tt>.
*/
virtual void receive_policy(policy_flag policy, size_t buffer_size) = 0;
}; };
} } // namespace cppa::network } } // namespace cppa::io
#endif // IO_SERVICE_HPP #endif // CPPA_IO_ACCEPT_HANDLE_HPP
...@@ -31,44 +31,57 @@ ...@@ -31,44 +31,57 @@
#ifndef CPPA_BROKER_HPP #ifndef CPPA_BROKER_HPP
#define CPPA_BROKER_HPP #define CPPA_BROKER_HPP
#include <functional> #include <map>
#include "cppa/stackless.hpp" #include "cppa/stackless.hpp"
#include "cppa/threadless.hpp" #include "cppa/threadless.hpp"
#include "cppa/local_actor.hpp" #include "cppa/local_actor.hpp"
#include "cppa/mailbox_element.hpp"
#include "cppa/io/buffered_writer.hpp" #include "cppa/util/buffer.hpp"
#include "cppa/io/acceptor.hpp"
#include "cppa/io/input_stream.hpp"
#include "cppa/io/output_stream.hpp"
#include "cppa/io/accept_handle.hpp"
#include "cppa/io/connection_handle.hpp"
#include "cppa/detail/fwd.hpp" #include "cppa/detail/fwd.hpp"
namespace cppa { namespace io { namespace cppa { namespace io {
class broker_continuation; class broker;
class broker : public extend<local_actor>::with<threadless, stackless> typedef intrusive_ptr<broker> broker_ptr;
, public buffered_writer {
typedef combined_type super1; local_actor_ptr init_and_launch(broker_ptr);
typedef buffered_writer super2;
friend class broker_continuation; /**
* @brief A broker mediates between a libcppa-based actor system
* and other components in the network.
* @extends local_actor
*/
class broker : public extend<local_actor>::with<threadless, stackless> {
public: typedef combined_type super;
enum policy_flag { at_least, at_most, exactly };
broker(input_stream_ptr in, output_stream_ptr out); // implementation relies on several helper classes ...
class scribe;
class servant;
class doorman;
class continuation;
void io_failed() override; // ... and some helpers need friendship
friend class scribe;
friend class doorman;
friend class continuation;
void dispose() override; friend local_actor_ptr init_and_launch(broker_ptr);
void receive_policy(policy_flag policy, size_t buffer_size); broker() = delete;
continue_reading_result continue_reading() override; public:
void write(size_t num_bytes, const void* data); enum policy_flag { at_least, at_most, exactly };
void enqueue(const message_header& hdr, any_tuple msg); void enqueue(const message_header& hdr, any_tuple msg);
...@@ -76,16 +89,23 @@ class broker : public extend<local_actor>::with<threadless, stackless> ...@@ -76,16 +89,23 @@ class broker : public extend<local_actor>::with<threadless, stackless>
void quit(std::uint32_t reason); void quit(std::uint32_t reason);
static intrusive_ptr<broker> from(std::function<void (broker*)> fun, void receive_policy(const connection_handle& hdl,
input_stream_ptr in, broker::policy_flag policy,
output_stream_ptr out); size_t buffer_size);
void write(const connection_handle& hdl, size_t num_bytes, const void* buf);
void write(const connection_handle& hdl, const util::buffer& buf);
void write(const connection_handle& hdl, util::buffer&& buf);
static broker_ptr from(std::function<void (broker*)> fun,
input_stream_ptr in,
output_stream_ptr out);
template<typename F, typename T0, typename... Ts> template<typename F, typename T0, typename... Ts>
static intrusive_ptr<broker> from(F fun, static broker_ptr from(F fun, input_stream_ptr in, output_stream_ptr out,
input_stream_ptr in, T0&& arg0, Ts&&... args) {
output_stream_ptr out,
T0&& arg0,
Ts&&... args) {
return from(std::bind(std::move(fun), return from(std::bind(std::move(fun),
std::placeholders::_1, std::placeholders::_1,
detail::fwd<T0>(arg0), detail::fwd<T0>(arg0),
...@@ -94,29 +114,75 @@ class broker : public extend<local_actor>::with<threadless, stackless> ...@@ -94,29 +114,75 @@ class broker : public extend<local_actor>::with<threadless, stackless>
std::move(out)); std::move(out));
} }
static broker_ptr from(std::function<void (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,
detail::fwd<T0>(arg0),
detail::fwd<Ts>(args)...),
std::move(in));
}
actor_ptr fork(std::function<void (broker*)> fun,
const connection_handle& hdl);
template<typename F, typename T0, typename... Ts>
actor_ptr fork(F fun,
const connection_handle& hdl,
T0&& arg0,
Ts&&... args) {
return this->fork(std::bind(std::move(fun),
std::placeholders::_1,
detail::fwd<T0>(arg0),
detail::fwd<Ts>(args)...),
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();
}
protected: protected:
void cleanup(std::uint32_t reason); broker(input_stream_ptr in, output_stream_ptr out);
broker(acceptor_uptr in);
void cleanup(std::uint32_t reason) override;
typedef std::unique_ptr<broker::scribe> scribe_pointer;
typedef std::unique_ptr<broker::doorman> doorman_pointer;
explicit broker(scribe_pointer);
private: private:
void invoke_message(const message_header& hdr, any_tuple msg); void invoke_message(const message_header& hdr, any_tuple msg);
void disconnect(); void erase_io(int id);
static constexpr size_t default_max_buffer_size = 65535; void erase_acceptor(int id);
bool m_is_continue_reading; void init_broker();
bool m_disconnected;
bool m_dirty; connection_handle add_scribe(input_stream_ptr in, output_stream_ptr out);
policy_flag m_policy;
size_t m_policy_buffer_size; accept_handle add_doorman(acceptor_uptr ptr);
input_stream_ptr m_in;
cow_tuple<atom_value, uint32_t, util::buffer> m_read; std::map<accept_handle, doorman_pointer> m_accept;
std::map<connection_handle, scribe_pointer> m_io;
}; };
typedef intrusive_ptr<broker> broker_ptr; //typedef intrusive_ptr<broker> broker_ptr;
} } // namespace cppa::network } } // namespace cppa::network
......
...@@ -28,56 +28,104 @@ ...@@ -28,56 +28,104 @@
\******************************************************************************/ \******************************************************************************/
#include "cppa/logging.hpp" #ifndef CPPA_IO_BUFFERED_WRITING_HPP
#include "cppa/to_string.hpp" #define CPPA_IO_BUFFERED_WRITING_HPP
#include "cppa/singletons.hpp"
#include <utility>
#include "cppa/util/buffer.hpp"
#include "cppa/io/middleman.hpp" #include "cppa/io/middleman.hpp"
#include "cppa/io/buffered_writer.hpp" #include "cppa/io/continuable.hpp"
#include "cppa/io/output_stream.hpp"
namespace cppa { namespace io { namespace cppa { namespace io {
buffered_writer::buffered_writer(middleman* pptr, native_socket_type rfd, output_stream_ptr out) template<class Base, class Subtype>
: super(rfd, out->write_handle()), m_middleman(pptr) class buffered_writing : public Base {
, m_out(out), m_has_unwritten_data(false) { }
typedef Base super;
continue_writing_result buffered_writer::continue_writing() {
CPPA_LOG_TRACE(""); public:
CPPA_LOG_DEBUG_IF(!m_has_unwritten_data, "nothing to write (done)");
while (m_has_unwritten_data) { template<typename... Ts>
size_t written; buffered_writing(middleman* mm, output_stream_ptr out, Ts&&... args)
try { written = m_out->write_some(m_buf.data(), m_buf.size()); } : super{std::forward<Ts>(args)...}, m_middleman{mm}, m_out{out}
catch (std::exception& e) { , m_has_unwritten_data{false} { }
CPPA_LOG_ERROR(to_verbose_string(e));
static_cast<void>(e); // keep compiler happy continue_writing_result continue_writing() override {
return write_failure; CPPA_LOG_TRACE("");
} CPPA_LOG_DEBUG_IF(!m_has_unwritten_data, "nothing to write (done)");
if (written != m_buf.size()) { while (m_has_unwritten_data) {
CPPA_LOGMF(CPPA_DEBUG, self, "tried to write " << m_buf.size() << "bytes, " size_t written;
<< "only " << written << " bytes written"); try { written = m_out->write_some(m_buf.data(), m_buf.size()); }
m_buf.erase_leading(written); catch (std::exception& e) {
return write_continue_later; CPPA_LOG_ERROR(to_verbose_string(e));
static_cast<void>(e); // keep compiler happy
return write_failure;
}
if (written != m_buf.size()) {
CPPA_LOG_DEBUG("tried to write " << m_buf.size() << "bytes, "
<< "only " << written << " bytes written");
m_buf.erase_leading(written);
return write_continue_later;
}
else {
m_buf.clear();
m_has_unwritten_data = false;
CPPA_LOG_DEBUG("write done, " << written << " bytes written");
}
} }
return write_done;
}
inline bool has_unwritten_data() const {
return m_has_unwritten_data;
}
void write(size_t num_bytes, const void* data) {
m_buf.write(num_bytes, data);
register_for_writing();
}
void write(const util::buffer& buf) {
write(buf.size(), buf.data());
}
void write(util::buffer&& buf) {
if (m_buf.empty()) m_buf = std::move(buf);
else { else {
m_buf.clear(); m_buf.write(buf.size(), buf.data());
m_has_unwritten_data = false; buf.clear();
CPPA_LOGMF(CPPA_DEBUG, self, "write done, " << written << "bytes written"); }
register_for_writing();
}
void register_for_writing() {
if (!m_has_unwritten_data) {
CPPA_LOG_DEBUG("register for writing");
m_has_unwritten_data = true;
m_middleman->continue_writer(this);
} }
} }
return write_done;
} inline util::buffer& write_buffer() {
return m_buf;
void buffered_writer::write(size_t num_bytes, const void* data) {
m_buf.write(num_bytes, data);
register_for_writing();
}
void buffered_writer::register_for_writing() {
if (!m_has_unwritten_data) {
CPPA_LOGMF(CPPA_DEBUG, self, "register for writing");
m_has_unwritten_data = true;
m_middleman->continue_writer(this);
} }
}
} } // namespace cppa::network protected:
typedef buffered_writing combined_type;
private:
middleman* m_middleman;
output_stream_ptr m_out;
bool m_has_unwritten_data;
util::buffer m_buf;
};
} } // namespace cppa::io
#endif // CPPA_IO_BUFFERED_WRITING_HPP
...@@ -28,10 +28,31 @@ ...@@ -28,10 +28,31 @@
\******************************************************************************/ \******************************************************************************/
#include "cppa/io/io_handle.hpp" #ifndef IO_SERVICE_HPP
#define IO_SERVICE_HPP
#include "cppa/detail/handle.hpp"
namespace cppa { namespace io { namespace cppa { namespace io {
io_handle::~io_handle() { } class broker;
class connection_handle : public detail::handle<connection_handle> {
friend class detail::handle<connection_handle>;
typedef detail::handle<connection_handle> super;
public:
connection_handle() = default;
private:
inline connection_handle(int handle_id) : super{handle_id} { }
};
} } // namespace cppa::network } } // namespace cppa::network
#endif // IO_SERVICE_HPP
...@@ -34,6 +34,7 @@ ...@@ -34,6 +34,7 @@
#include <map> #include <map>
#include <cstdint> #include <cstdint>
#include "cppa/extend.hpp"
#include "cppa/actor_proxy.hpp" #include "cppa/actor_proxy.hpp"
#include "cppa/partial_function.hpp" #include "cppa/partial_function.hpp"
#include "cppa/weak_intrusive_ptr.hpp" #include "cppa/weak_intrusive_ptr.hpp"
...@@ -43,16 +44,16 @@ ...@@ -43,16 +44,16 @@
#include "cppa/io/input_stream.hpp" #include "cppa/io/input_stream.hpp"
#include "cppa/io/output_stream.hpp" #include "cppa/io/output_stream.hpp"
#include "cppa/io/buffered_writer.hpp" #include "cppa/io/buffered_writing.hpp"
#include "cppa/io/default_message_queue.hpp" #include "cppa/io/default_message_queue.hpp"
namespace cppa { namespace io { namespace cppa { namespace io {
class default_protocol; class default_protocol;
class default_peer : public buffered_writer { class default_peer : public extend<continuable>::with<buffered_writing> {
typedef buffered_writer super; typedef combined_type super;
friend class default_protocol; friend class default_protocol;
......
...@@ -46,7 +46,7 @@ class default_peer_acceptor : public continuable { ...@@ -46,7 +46,7 @@ class default_peer_acceptor : public continuable {
public: public:
continue_reading_result continue_reading(); continue_reading_result continue_reading() override;
default_peer_acceptor(default_protocol* parent, default_peer_acceptor(default_protocol* parent,
acceptor_uptr ptr, acceptor_uptr ptr,
...@@ -56,7 +56,7 @@ class default_peer_acceptor : public continuable { ...@@ -56,7 +56,7 @@ class default_peer_acceptor : public continuable {
void dispose() override; void dispose() override;
void io_failed(); void io_failed() override;
private: private:
......
This diff is collapsed.
...@@ -60,7 +60,7 @@ default_peer::default_peer(default_protocol* parent, ...@@ -60,7 +60,7 @@ default_peer::default_peer(default_protocol* parent,
const input_stream_ptr& in, const input_stream_ptr& in,
const output_stream_ptr& out, const output_stream_ptr& out,
process_information_ptr peer_ptr) process_information_ptr peer_ptr)
: super(parent->parent(), in->read_handle(), out) : super(parent->parent(), out, in->read_handle(), out->write_handle())
, m_parent(parent), m_in(in) , m_parent(parent), m_in(in)
, m_state((peer_ptr) ? wait_for_msg_size : wait_for_process_info) , m_state((peer_ptr) ? wait_for_msg_size : wait_for_process_info)
, m_node(peer_ptr) { , m_node(peer_ptr) {
......
...@@ -55,7 +55,6 @@ ...@@ -55,7 +55,6 @@
#include "cppa/util/void_type.hpp" #include "cppa/util/void_type.hpp"
#include "cppa/detail/demangle.hpp" #include "cppa/detail/demangle.hpp"
#include "cppa/detail/object_array.hpp"
#include "cppa/detail/actor_registry.hpp" #include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/to_uniform_name.hpp" #include "cppa/detail/to_uniform_name.hpp"
#include "cppa/detail/singleton_manager.hpp" #include "cppa/detail/singleton_manager.hpp"
......
...@@ -33,6 +33,7 @@ ...@@ -33,6 +33,7 @@
#include <vector> #include <vector>
#include <cstring> #include <cstring>
#include <algorithm> #include <algorithm>
#include <type_traits>
#include "cppa/any_tuple.hpp" #include "cppa/any_tuple.hpp"
#include "cppa/message_header.hpp" #include "cppa/message_header.hpp"
...@@ -41,6 +42,7 @@ ...@@ -41,6 +42,7 @@
#include "cppa/util/duration.hpp" #include "cppa/util/duration.hpp"
#include "cppa/util/limited_vector.hpp" #include "cppa/util/limited_vector.hpp"
#include "cppa/detail/object_array.hpp"
#include "cppa/detail/uniform_type_info_map.hpp" #include "cppa/detail/uniform_type_info_map.hpp"
#include "cppa/detail/default_uniform_type_info_impl.hpp" #include "cppa/detail/default_uniform_type_info_impl.hpp"
...@@ -49,27 +51,41 @@ namespace cppa { namespace detail { ...@@ -49,27 +51,41 @@ namespace cppa { namespace detail {
// maps demangled names to libcppa names // maps demangled names to libcppa names
// WARNING: this map is sorted, insert new elements *in sorted order* as well! // WARNING: this map is sorted, insert new elements *in sorted order* as well!
/* extern */ const char* mapped_type_names[][2] = { /* extern */ const char* mapped_type_names[][2] = {
{ "bool", "bool" }, { "bool", "bool" },
{ "cppa::any_tuple", "@tuple" }, { "cppa::any_tuple", "@tuple" },
{ "cppa::atom_value", "@atom" }, { "cppa::atom_value", "@atom" },
{ "cppa::intrusive_ptr<cppa::actor>", "@actor" }, { "cppa::intrusive_ptr<cppa::actor>", "@actor" },
{ "cppa::intrusive_ptr<cppa::channel>", "@channel" }, { "cppa::intrusive_ptr<cppa::channel>", "@channel" },
{ "cppa::intrusive_ptr<cppa::group>", "@group" }, { "cppa::intrusive_ptr<cppa::group>", "@group" },
{ "cppa::intrusive_ptr<cppa::process_information>", "@proc"}, { "cppa::intrusive_ptr<cppa::process_information>", "@proc"},
{ "cppa::message_header", "@header" }, { "cppa::io::accept_handle", "@ac_hdl" },
{ "cppa::nullptr_t", "@null" }, { "cppa::io::connection_handle", "@cn_hdl" },
{ "cppa::util::buffer", "@buffer" }, { "cppa::message_header", "@header" },
{ "cppa::util::duration", "@duration" }, { "cppa::nullptr_t", "@null" },
{ "cppa::util::void_type", "@0" }, { "cppa::util::buffer", "@buffer" },
{ "double", "double" }, { "cppa::util::duration", "@duration" },
{ "float", "float" }, { "cppa::util::void_type", "@0" },
{ "long double", "@ldouble" }, { "double", "double" },
{ "std::basic_string<@i16,std::char_traits<@i16>,std::allocator<@i16>>", "@u16str" }, { "float", "float" },
{ "std::basic_string<@i32,std::char_traits<@i32>,std::allocator<@i32>>", "@u32str" }, { "long double", "@ldouble" },
{ "std::basic_string<@i8,std::char_traits<@i8>,std::allocator<@i8>>", "@str" }, // std::u16string
{ "std::basic_string<@u16,std::char_traits<@u16>,std::allocator<@u16>>", "@u16str" }, { "std::basic_string<@i16,std::char_traits<@i16>,std::allocator<@i16>>",
{ "std::basic_string<@u32,std::char_traits<@u32>,std::allocator<@u32>>", "@u32str" }, "@u16str" },
{ "std::map<@str,@str,std::less<@str>,std::allocator<std::pair<const @str,@str>>>", "@strmap" } // std::u32string
{ "std::basic_string<@i32,std::char_traits<@i32>,std::allocator<@i32>>",
"@u32str" },
// std::string
{ "std::basic_string<@i8,std::char_traits<@i8>,std::allocator<@i8>>",
"@str" },
// std::u16string (again, using unsigned char type)
{ "std::basic_string<@u16,std::char_traits<@u16>,std::allocator<@u16>>",
"@u16str" },
// std::u32string (again, using unsigned char type)
{ "std::basic_string<@u32,std::char_traits<@u32>,std::allocator<@u32>>",
"@u32str" },
// std::map<std::string,std::string>
{ "std::map<@str,@str,std::less<@str>,std::allocator<std::pair<const @str,@str>>>",
"@strmap" }
}; };
// maps sizeof(T) => {unsigned name, signed name} // maps sizeof(T) => {unsigned name, signed name}
...@@ -133,20 +149,38 @@ void assert_type_name(const char* expected_name, deserializer* source) { ...@@ -133,20 +149,38 @@ void assert_type_name(const char* expected_name, deserializer* source) {
} }
template<typename T> template<typename T>
void serialize_impl(const T& val, serializer* sink) { typename std::enable_if<util::is_primitive<T>::value>::type
serialize_impl(const T& val, serializer* sink) {
sink->begin_object(mapped_name<T>()); sink->begin_object(mapped_name<T>());
sink->write_value(val); sink->write_value(val);
sink->end_object(); sink->end_object();
} }
template<typename T> template<typename T>
void deserialize_impl(T& val, deserializer* source) { typename std::enable_if<util::is_primitive<T>::value>::type
deserialize_impl(T& val, deserializer* source) {
assert_type_name(mapped_name<T>(), source); assert_type_name(mapped_name<T>(), source);
source->begin_object(mapped_name<T>()); source->begin_object(mapped_name<T>());
val = source->read<T>(); val = source->read<T>();
source->end_object(); source->end_object();
} }
template<typename T>
void serialize_impl(const detail::handle<T>& hdl, serializer* sink) {
sink->begin_object(mapped_name<T>());
sink->write_value(static_cast<int32_t>(hdl.id()));
sink->end_object();
}
template<typename T>
void deserialize_impl(detail::handle<T>& hdl, deserializer* source) {
auto tname = mapped_name<T>();
assert_type_name(tname, source);
source->begin_object(tname);
hdl = T::from_int(source->read<int32_t>());
source->end_object();
}
void serialize_impl(const util::void_type&, serializer* sink) { void serialize_impl(const util::void_type&, serializer* sink) {
sink->begin_object(mapped_name<util::void_type>()); sink->begin_object(mapped_name<util::void_type>());
sink->end_object(); sink->end_object();
...@@ -639,10 +673,12 @@ class utim_impl : public uniform_type_info_map { ...@@ -639,10 +673,12 @@ class utim_impl : public uniform_type_info_map {
// fill builtin types *in sorted order* (by uniform name) // fill builtin types *in sorted order* (by uniform name)
size_t i = 0; size_t i = 0;
m_builtin_types[i++] = &m_type_void; // @0 m_builtin_types[i++] = &m_type_void; // @0
m_builtin_types[i++] = &m_ac_hdl; // @ac_hdl
m_builtin_types[i++] = &m_type_actor; // @actor m_builtin_types[i++] = &m_type_actor; // @actor
m_builtin_types[i++] = &m_type_atom; // @atom m_builtin_types[i++] = &m_type_atom; // @atom
m_builtin_types[i++] = &m_type_buffer; // @buffer m_builtin_types[i++] = &m_type_buffer; // @buffer
m_builtin_types[i++] = &m_type_channel; // @channel m_builtin_types[i++] = &m_type_channel; // @channel
m_builtin_types[i++] = &m_cn_hdl; // @cn_hdl
m_builtin_types[i++] = &m_type_duration; // @duration m_builtin_types[i++] = &m_type_duration; // @duration
m_builtin_types[i++] = &m_type_group; // @group m_builtin_types[i++] = &m_type_group; // @group
m_builtin_types[i++] = &m_type_header; // @header m_builtin_types[i++] = &m_type_header; // @header
...@@ -735,6 +771,8 @@ class utim_impl : public uniform_type_info_map { ...@@ -735,6 +771,8 @@ class utim_impl : public uniform_type_info_map {
typedef std::map<std::string, std::string> strmap; typedef std::map<std::string, std::string> strmap;
uti_impl<process_information_ptr> m_type_proc; uti_impl<process_information_ptr> m_type_proc;
uti_impl<io::accept_handle> m_ac_hdl;
uti_impl<io::connection_handle> m_cn_hdl;
uti_impl<channel_ptr> m_type_channel; uti_impl<channel_ptr> m_type_channel;
buffer_type_info_impl m_type_buffer; buffer_type_info_impl m_type_buffer;
uti_impl<actor_ptr> m_type_actor; uti_impl<actor_ptr> m_type_actor;
...@@ -762,7 +800,7 @@ class utim_impl : public uniform_type_info_map { ...@@ -762,7 +800,7 @@ class utim_impl : public uniform_type_info_map {
int_tinfo<std::uint64_t> m_type_u64; int_tinfo<std::uint64_t> m_type_u64;
// both containers are sorted by uniform name // both containers are sorted by uniform name
std::array<pointer, 26> m_builtin_types; std::array<pointer, 28> m_builtin_types;
std::vector<uniform_type_info*> m_user_types; std::vector<uniform_type_info*> m_user_types;
template<typename Container> template<typename Container>
......
...@@ -26,6 +26,7 @@ add_unit_test(spawn ping_pong.cpp) ...@@ -26,6 +26,7 @@ add_unit_test(spawn ping_pong.cpp)
add_unit_test(local_group) add_unit_test(local_group)
add_unit_test(sync_send) add_unit_test(sync_send)
add_unit_test(remote_actor ping_pong.cpp) add_unit_test(remote_actor ping_pong.cpp)
add_unit_test(broker)
if (ENABLE_OPENCL) if (ENABLE_OPENCL)
add_unit_test(opencl) add_unit_test(opencl)
......
...@@ -3,9 +3,10 @@ ...@@ -3,9 +3,10 @@
#include "test.hpp" #include "test.hpp"
#include "cppa/cppa.hpp" #include "cppa/cppa.hpp"
using namespace std;
using namespace cppa; using namespace cppa;
namespace { std::atomic<size_t> s_error_count{0}; } namespace { atomic<size_t> s_error_count{0}; }
size_t cppa_error_count() { size_t cppa_error_count() {
return s_error_count; return s_error_count;
...@@ -15,10 +16,10 @@ void cppa_inc_error_count() { ...@@ -15,10 +16,10 @@ void cppa_inc_error_count() {
++s_error_count; ++s_error_count;
} }
std::string cppa_fill4(int value) { string cppa_fill4(int value) {
std::string result = std::to_string(value); string result = to_string(value);
while (result.size() < 4) result.insert(result.begin(), '0'); while (result.size() < 4) result.insert(result.begin(), '0');
return std::move(result); return move(result);
} }
const char* cppa_strip_path(const char* fname) { const char* cppa_strip_path(const char* fname) {
...@@ -41,13 +42,46 @@ void cppa_unexpected_timeout(const char* fname, size_t line_num) { ...@@ -41,13 +42,46 @@ void cppa_unexpected_timeout(const char* fname, size_t line_num) {
CPPA_PRINTERRC(fname, line_num, "unexpected timeout"); CPPA_PRINTERRC(fname, line_num, "unexpected timeout");
} }
std::vector<std::string> split(const std::string& str, char delim, bool keep_empties) { vector<string> split(const string& str, char delim, bool keep_empties) {
using namespace std; using namespace std;
vector<string> result; vector<string> result;
stringstream strs{str}; stringstream strs{str};
string tmp; string tmp;
while (getline(strs, tmp, delim)) { while (getline(strs, tmp, delim)) {
if (!tmp.empty() || keep_empties) result.push_back(std::move(tmp)); if (!tmp.empty() || keep_empties) result.push_back(move(tmp));
}
return result;
}
void verbose_terminate() {
try { if (uncaught_exception()) throw; }
catch (exception& e) {
CPPA_PRINTERR("terminate called after throwing "
<< to_verbose_string(e));
}
catch (...) {
CPPA_PRINTERR("terminate called after throwing an unknown exception");
}
abort();
}
void set_default_test_settings() {
CPPA_SET_DEBUG_NAME("main");
set_terminate(verbose_terminate);
cout.unsetf(ios_base::unitbuf);
}
map<string, string> get_kv_pairs(int argc, char** argv, int begin) {
map<string, string> result;
for (int i = begin; i < argc; ++i) {
auto vec = split(argv[i], '=');
if (vec.size() != 2) {
CPPA_PRINTERR("\"" << argv[i] << "\" is not a key-value pair");
}
else if (result.count(vec[0]) != 0) {
CPPA_PRINTERR("key \"" << vec[0] << "\" is already defined");
}
else result[vec[0]] = vec[1];
} }
return result; return result;
} }
...@@ -9,11 +9,14 @@ ...@@ -9,11 +9,14 @@
#include <iostream> #include <iostream>
#include <type_traits> #include <type_traits>
#include "cppa/cppa.hpp"
#include "cppa/actor.hpp" #include "cppa/actor.hpp"
#include "cppa/logging.hpp" #include "cppa/logging.hpp"
#include "cppa/to_string.hpp" #include "cppa/to_string.hpp"
#include "cppa/util/scope_guard.hpp" #include "cppa/util/scope_guard.hpp"
void set_default_test_settings();
size_t cppa_error_count(); size_t cppa_error_count();
void cppa_inc_error_count(); void cppa_inc_error_count();
std::string cppa_fill4(int value); std::string cppa_fill4(int value);
...@@ -114,6 +117,7 @@ inline void cppa_check_value(V1 v1, ...@@ -114,6 +117,7 @@ inline void cppa_check_value(V1 v1,
auto cppa_test_scope_guard = ::cppa::util::make_scope_guard([] { \ auto cppa_test_scope_guard = ::cppa::util::make_scope_guard([] { \
std::cout << cppa_error_count() << " error(s) detected" << std::endl; \ std::cout << cppa_error_count() << " error(s) detected" << std::endl; \
}); \ }); \
set_default_test_settings(); \
CPPA_LOGF_INFO("run unit test " << #testname) CPPA_LOGF_INFO("run unit test " << #testname)
#define CPPA_TEST_RESULT() ((cppa_error_count() == 0) ? 0 : -1) #define CPPA_TEST_RESULT() ((cppa_error_count() == 0) ? 0 : -1)
...@@ -162,4 +166,20 @@ inline void cppa_check_value(V1 v1, ...@@ -162,4 +166,20 @@ inline void cppa_check_value(V1 v1,
std::vector<std::string> split(const std::string& str, char delim = ' ', bool keep_empties = true); std::vector<std::string> split(const std::string& str, char delim = ' ', bool keep_empties = true);
std::map<std::string, std::string> get_kv_pairs(int argc, char** argv, int begin = 1);
template<typename F>
void run_client_part(const std::map<std::string, std::string>& args, F fun) {
CPPA_LOGF_INFO("run in client mode");
auto i = args.find("port");
if (i == args.end()) {
CPPA_LOGF_ERROR("no port specified");
throw std::logic_error("no port specified");
}
auto port = static_cast<std::uint16_t>(stoi(i->second));
fun(port);
cppa::await_all_others_done();
cppa::shutdown();
}
#endif // TEST_HPP #endif // TEST_HPP
...@@ -29,7 +29,7 @@ void foo() { ...@@ -29,7 +29,7 @@ void foo() {
int main() { int main() {
bool matched_pattern[3] = { false, false, false }; bool matched_pattern[3] = { false, false, false };
CPPA_TEST(test_atom); CPPA_TEST(test_atom);
// check if there are leading bits that distinguish "zzz" and "000 " // check if there are leading bits that distinguish "zzz" and "000 "
CPPA_CHECK_NOT_EQUAL(atom("zzz"), atom("000 ")); CPPA_CHECK_NOT_EQUAL(atom("zzz"), atom("000 "));
// check if there are leading bits that distinguish "abc" and " abc" // check if there are leading bits that distinguish "abc" and " abc"
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#include <memory>
#include<iostream>
#include "test.hpp"
#include "cppa/cppa.hpp"
using namespace std;
using namespace cppa;
namespace { constexpr size_t message_size = sizeof(atom_value) + sizeof(int); }
void ping(size_t num_pings) {
auto count = std::make_shared<size_t>(0);
become (
on(atom("kickoff"), arg_match) >> [=](const actor_ptr& pong) {
send(pong, atom("ping"), 1);
become (
on(atom("pong"), arg_match) >> [=](int value) {
if (++*count >= num_pings) self->quit();
else reply(atom("ping"), value + 1);
}
);
}
);
}
void pong() {
become (
on(atom("ping"), arg_match) >> [](int value) {
self->monitor(self->last_sender());
reply(atom("pong"), value);
become (
on(atom("ping"), arg_match) >> [](int value) {
reply(atom("pong"), value);
},
on(atom("DOWN"), arg_match) >> [=](uint32_t reason) {
self->quit(reason);
}
);
}
);
}
void peer(io::broker* thisptr, const actor_ptr& buddy) {
self->monitor(buddy);
if (thisptr->num_connections() == 0) {
cout << "num_connections() != 1" << endl;
throw std::logic_error("num_connections() != 1");
}
thisptr->for_each_connection([=](io::connection_handle hdl) {
thisptr->receive_policy(hdl, io::broker::exactly, message_size);
});
auto write = [=](atom_value type, int value) {
thisptr->for_each_connection([=](io::connection_handle hdl) {
thisptr->write(hdl, sizeof(type), &type);
thisptr->write(hdl, sizeof(value), &value);
});
};
become (
on(atom("IO_closed"), arg_match) >> [=](io::connection_handle) {
self->quit();
},
on(atom("IO_read"), arg_match) >> [=](io::connection_handle, const util::buffer& buf) {
atom_value type;
int value;
memcpy(&type, buf.data(), sizeof(atom_value));
memcpy(&value, buf.offset_data(sizeof(atom_value)), sizeof(int));
send(buddy, type, value);
},
on(atom("ping"), arg_match) >> [=](int value) {
write(atom("ping"), value);
},
on(atom("pong"), arg_match) >> [=](int value) {
write(atom("pong"), value);
},
on(atom("DOWN"), arg_match) >> [=](uint32_t reason) {
if (thisptr->last_sender() == buddy) self->quit(reason);
},
others() >> [] {
cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
}
);
}
void peer_acceptor(io::broker* thisptr, const actor_ptr& buddy) {
become (
on(atom("IO_accept"), arg_match) >> [=](io::accept_handle, io::connection_handle hdl) {
thisptr->fork(peer, hdl, buddy);
self->quit();
},
others() >> [] {
cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
}
);
}
int main(int argc, char** argv) {
CPPA_TEST(test_broker);
string app_path = argv[0];
if (argc == 3) {
if (strcmp(argv[1], "mode=client") == 0) {
CPPA_CHECKPOINT();
run_client_part(get_kv_pairs(argc, argv), [](uint16_t port) {
CPPA_CHECKPOINT();
auto p = spawn(ping, 10);
CPPA_CHECKPOINT();
auto cl = spawn_io(peer, "localhost", port, p);
CPPA_CHECKPOINT();
send_as(nullptr, p, atom("kickoff"), cl);
CPPA_CHECKPOINT();
});
CPPA_CHECKPOINT();
return CPPA_TEST_RESULT();
}
return CPPA_TEST_RESULT();
}
else if (argc > 1) {
cerr << "usage: " << app_path << " [mode=client port={PORT}]" << endl;
return -1;
}
CPPA_CHECKPOINT();
auto p = spawn(pong);
uint16_t port = 4242;
for (;;) {
try {
spawn_io_server(peer_acceptor, port, p);
CPPA_CHECKPOINT();
ostringstream oss;
oss << app_path << " mode=client port=" << port << " &>/dev/null";
thread child{[&oss] {
CPPA_LOGC_TRACE("NONE", "main$thread_launcher", "");
auto cmdstr = oss.str();
if (system(cmdstr.c_str()) != 0) {
CPPA_PRINTERR("FATAL: command failed: " << cmdstr);
abort();
}
}};
CPPA_CHECKPOINT();
child.join();
CPPA_CHECKPOINT();
await_all_others_done();
CPPA_CHECKPOINT();
shutdown();
return CPPA_TEST_RESULT();
}
catch (bind_failure&) {
// try next port
++port;
}
}
}
...@@ -20,24 +20,6 @@ typedef std::pair<std::string, std::string> string_pair; ...@@ -20,24 +20,6 @@ typedef std::pair<std::string, std::string> string_pair;
typedef vector<actor_ptr> actor_vector; typedef vector<actor_ptr> actor_vector;
vector<string_pair> get_kv_pairs(int argc, char** argv, int begin = 1) {
vector<string_pair> result;
for (int i = begin; i < argc; ++i) {
auto vec = split(argv[i], '=');
auto eq_fun = [&](const string_pair& p) { return p.first == vec[0]; };
if (vec.size() != 2) {
CPPA_PRINTERR("\"" << argv[i] << "\" is not a key-value pair");
}
else if (any_of(result.begin(), result.end(), eq_fun)) {
CPPA_PRINTERR("key \"" << vec[0] << "\" is already defined");
}
else {
result.emplace_back(vec[0], vec[1]);
}
}
return result;
}
void reflector() { void reflector() {
CPPA_SET_DEBUG_NAME("reflector" << self->id()); CPPA_SET_DEBUG_NAME("reflector" << self->id());
become ( become (
...@@ -150,19 +132,6 @@ void spawn5_client() { ...@@ -150,19 +132,6 @@ void spawn5_client() {
} // namespace <anonymous> } // namespace <anonymous>
void verbose_terminate() {
try { if (std::uncaught_exception()) throw; }
catch (std::exception& e) {
CPPA_PRINTERR("terminate called after throwing "
<< to_verbose_string(e));
}
catch (...) {
CPPA_PRINTERR("terminate called after throwing an unknown exception");
}
abort();
}
template<typename T> template<typename T>
void await_down(actor_ptr ptr, T continuation) { void await_down(actor_ptr ptr, T continuation) {
become ( become (
...@@ -340,37 +309,8 @@ class server : public event_based_actor { ...@@ -340,37 +309,8 @@ class server : public event_based_actor {
}; };
void run_client_part(const vector<string_pair>& args) {
CPPA_LOGF_INFO("run in client mode");
CPPA_TEST(test_remote_actor_client_part);
auto i = find_if(args.begin(), args.end(),
[](const string_pair& p) { return p.first == "port"; });
if (i == args.end()) {
CPPA_LOGF_ERROR("no port specified");
throw std::logic_error("no port specified");
}
auto port = static_cast<uint16_t>(stoi(i->second));
auto serv = remote_actor("localhost", port);
// remote_actor is supposed to return the same server when connecting to
// the same host again
{
auto server2 = remote_actor("localhost", port);
CPPA_CHECK(serv == server2);
}
auto c = spawn<client, monitored>(serv);
receive (
on(atom("DOWN"), arg_match) >> [=](uint32_t rsn) {
CPPA_CHECK_EQUAL(self->last_sender(), c);
CPPA_CHECK_EQUAL(rsn, exit_reason::normal);
}
);
}
int main(int argc, char** argv) { int main(int argc, char** argv) {
set_terminate(verbose_terminate);
announce<actor_vector>(); announce<actor_vector>();
CPPA_SET_DEBUG_NAME("main");
cout.unsetf(ios_base::unitbuf);
string app_path = argv[0]; string app_path = argv[0];
bool run_remote_actor = true; bool run_remote_actor = true;
if (argc > 1) { if (argc > 1) {
...@@ -379,9 +319,22 @@ int main(int argc, char** argv) { ...@@ -379,9 +319,22 @@ int main(int argc, char** argv) {
run_remote_actor = false; run_remote_actor = false;
} }
else { else {
run_client_part(get_kv_pairs(argc, argv)); run_client_part(get_kv_pairs(argc, argv), [](uint16_t port) {
await_all_others_done(); auto serv = remote_actor("localhost", port);
shutdown(); // remote_actor is supposed to return the same server
// when connecting to the same host again
{
auto server2 = remote_actor("localhost", port);
CPPA_CHECK(serv == server2);
}
auto c = spawn<client, monitored>(serv);
receive (
on(atom("DOWN"), arg_match) >> [=](uint32_t rsn) {
CPPA_CHECK_EQUAL(self->last_sender(), c);
CPPA_CHECK_EQUAL(rsn, exit_reason::normal);
}
);
});
return CPPA_TEST_RESULT(); return CPPA_TEST_RESULT();
} }
} }
......
...@@ -315,6 +315,7 @@ int main() { ...@@ -315,6 +315,7 @@ int main() {
CPPA_TEST(test_spawn); CPPA_TEST(test_spawn);
cout << "sizeof(event_based_actor) = " << sizeof(event_based_actor) << endl; cout << "sizeof(event_based_actor) = " << sizeof(event_based_actor) << endl;
cout << "sizeof(broker) = " << sizeof(io::broker) << endl;
CPPA_PRINT("test send()"); CPPA_PRINT("test send()");
send(self, 1, 2, 3, true); send(self, 1, 2, 3, true);
......
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