Commit 692e4119 authored by Dominik Charousset's avatar Dominik Charousset

moved all network IO classes to cppa::network, modularized middleman even...

moved all network IO classes to cppa::network, modularized middleman even further to pave the way for user-defined implementations of peers and network connections, and got rid of network_manager by promoting middleman to be directly managed by singleton_manager
parent 1c0e4c2d
......@@ -34,7 +34,7 @@ endif ()
# check for g++ >= 4.7 or clang++ > = 3.2
try_run(ProgramResult
CompilationSucceeded
${CMAKE_BINARY_DIR} ${CMAKE_SOURCE_DIR}/src/get_compiler_version.cpp
${CMAKE_BINARY_DIR} ${CMAKE_SOURCE_DIR}/get_compiler_version.cpp
RUN_OUTPUT_VARIABLE CompilerVersion)
if (NOT CompilationSucceeded OR NOT ProgramResult EQUAL 0)
message(FATAL_ERROR "Cannot determine compiler version")
......@@ -75,64 +75,7 @@ if (DISABLE_CONTEXT_SWITCHING)
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DCPPA_DISABLE_CONTEXT_SWITCHING")
endif ()
set(LIBCPPA_SRC
src/abstract_tuple.cpp
src/actor.cpp
src/actor_count.cpp
src/actor_proxy.cpp
src/actor_proxy_cache.cpp
src/actor_registry.cpp
src/addressed_message.cpp
src/any_tuple.cpp
src/atom.cpp
src/attachable.cpp
src/behavior_stack.cpp
src/binary_deserializer.cpp
src/binary_serializer.cpp
src/buffer.cpp
src/channel.cpp
src/thread_mapped_actor.cpp
src/demangle.cpp
src/deserializer.cpp
src/duration.cpp
src/empty_tuple.cpp
src/event_based_actor.cpp
src/exception.cpp
src/factory.cpp
src/fd_util.cpp
src/fiber.cpp
src/group.cpp
src/group_manager.cpp
src/ipv4_acceptor.cpp
src/ipv4_io_stream.cpp
src/local_actor.cpp
src/middleman.cpp
src/network_manager.cpp
src/object.cpp
src/object_array.cpp
src/opt.cpp
src/partial_function.cpp
src/pattern.cpp
src/primitive_variant.cpp
src/process_information.cpp
src/receive.cpp
src/ripemd_160.cpp
src/scheduled_actor.cpp
src/scheduled_actor_dummy.cpp
src/scheduler.cpp
src/self.cpp
src/serializer.cpp
src/shared_spinlock.cpp
src/singleton_manager.cpp
src/string_serialization.cpp
src/thread_pool_scheduler.cpp
src/to_uniform_name.cpp
src/unicast_network.cpp
src/uniform_type_info.cpp
src/yield_interface.cpp
src/context_switching_actor.cpp
src/decorated_names_map.cpp
)
file(GLOB LIBCPPA_SRC "src/*.cpp")
set(boost_context third_party/boost_context)
......
......@@ -40,7 +40,7 @@ cppa/detail/abstract_tuple.hpp
cppa/detail/actor_count.hpp
cppa/detail/actor_proxy_cache.hpp
cppa/detail/actor_registry.hpp
cppa/detail/addressed_message.hpp
cppa/network/addressed_message.hpp
cppa/detail/atom_val.hpp
cppa/detail/boxed.hpp
cppa/detail/channel.hpp
......@@ -55,7 +55,6 @@ cppa/detail/get_behavior.hpp
cppa/detail/group_manager.hpp
cppa/detail/implicit_conversions.hpp
cppa/detail/matches.hpp
cppa/detail/network_manager.hpp
cppa/detail/object_array.hpp
cppa/detail/object_impl.hpp
cppa/detail/pair_member.hpp
......@@ -195,7 +194,6 @@ src/fiber.cpp
src/group.cpp
src/group_manager.cpp
src/local_actor.cpp
src/network_manager.cpp
src/object.cpp
src/object_array.cpp
src/partial_function.cpp
......@@ -249,15 +247,15 @@ cppa/message_id.hpp
cppa/cppa_fwd.hpp
cppa/timeout_definition.hpp
cppa/detail/behavior_impl.hpp
cppa/util/output_stream.hpp
cppa/util/input_stream.hpp
cppa/util/io_stream.hpp
cppa/util/acceptor.hpp
cppa/detail/ipv4_io_stream.hpp
cppa/network/output_stream.hpp
cppa/network/input_stream.hpp
cppa/network/io_stream.hpp
cppa/network/acceptor.hpp
cppa/network/ipv4_io_stream.hpp
src/ipv4_io_stream.cpp
cppa/detail/ipv4_acceptor.hpp
cppa/network/ipv4_acceptor.hpp
src/ipv4_acceptor.cpp
cppa/detail/middleman.hpp
cppa/network/middleman.hpp
src/middleman.cpp
src/buffer.cpp
cppa/message_future.hpp
......@@ -272,3 +270,13 @@ cppa/detail/opt_impls.hpp
examples/type_plugins.hpp
cppa/detail/decorated_names_map.hpp
src/decorated_names_map.cpp
cppa/network/peer_acceptor.hpp
src/peer_acceptor.cpp
cppa/network/peer.hpp
src/peer.cpp
cppa/network/continuable_reader.hpp
src/continuable_reader.cpp
cppa/network/default_peer_impl.hpp
src/default_peer_impl.cpp
cppa/network/default_peer_acceptor_impl.hpp
src/default_peer_acceptor_impl.cpp
......@@ -61,7 +61,7 @@
#include "cppa/event_based_actor.hpp"
#include "cppa/util/rm_ref.hpp"
#include "cppa/util/acceptor.hpp"
#include "cppa/network/acceptor.hpp"
#include "cppa/detail/actor_count.hpp"
#include "cppa/detail/get_behavior.hpp"
......@@ -764,7 +764,7 @@ void publish(actor_ptr whom, std::uint16_t port, const char* addr = nullptr);
* @param whom Actor that should be published at @p port.
* @param acceptor Network technology-specific acceptor implementation.
*/
void publish(actor_ptr whom, std::unique_ptr<util::acceptor> acceptor);
void publish(actor_ptr whom, std::unique_ptr<network::acceptor> acceptor);
/**
* @brief Establish a new connection to the actor at @p host on given @p port.
......@@ -789,7 +789,7 @@ inline actor_ptr remote_actor(const std::string& host, std::uint16_t port) {
* @returns An {@link actor_ptr} to the proxy instance
* representing a remote actor.
*/
actor_ptr remote_actor(util::io_stream_ptr_pair connection);
actor_ptr remote_actor(network::io_stream_ptr_pair connection);
/**
* @brief Destroys all singletons, disconnects all peers and stops the
......
......@@ -38,19 +38,17 @@ namespace cppa { namespace detail { namespace fd_util {
#if defined(CPPA_MACOS) || defined(CPPA_LINUX)
typedef int file_descriptor;
// throws ios_base::failure and adds errno failure if @p add_errno_failure
void throw_io_failure(std::string&& what, bool add_errno_failure = true);
// returns true if fd is nonblocking
// throws @p ios_base::failure on error
bool nonblocking(file_descriptor fd);
bool nonblocking(native_socket_type fd);
// sets fd to nonblocking if <tt>set_nonblocking == true</tt>
// or to blocking if <tt>set_nonblocking == false</tt>
// throws @p ios_base::failure on error
void nonblocking(file_descriptor fd, bool new_value);
void nonblocking(native_socket_type fd, bool new_value);
// returns true if fd is nodelay socket
// throws @p ios_base::failure on error
......
......@@ -31,6 +31,8 @@
#ifndef CPPA_SINGLETON_MANAGER_HPP
#define CPPA_SINGLETON_MANAGER_HPP
#include <atomic>
namespace cppa {
class scheduler;
......@@ -38,13 +40,14 @@ class msg_content;
} // namespace cppa
namespace cppa { namespace network { class middleman; } }
namespace cppa { namespace detail {
class empty_tuple;
class group_manager;
class abstract_tuple;
class actor_registry;
class network_manager;
class decorated_names_map;
class uniform_type_info_map;
......@@ -54,6 +57,8 @@ class singleton_manager {
public:
static void shutdown();
static scheduler* get_scheduler();
static bool set_scheduler(scheduler*);
......@@ -62,8 +67,8 @@ class singleton_manager {
static actor_registry* get_actor_registry();
// created on-the-fly on a successfull cast to set_scheduler()
static network_manager* get_network_manager();
// created on-the-fly on a successfull call to set_scheduler()
static network::middleman* get_middleman();
static uniform_type_info_map* get_uniform_type_info_map();
......@@ -73,6 +78,24 @@ class singleton_manager {
static decorated_names_map* get_decorated_names_map();
private:
template<typename T>
static void stop_and_kill(std::atomic<T*>& ptr) {
for (;;) {
auto p = ptr.load();
if (p == nullptr) {
return;
}
else if (ptr.compare_exchange_weak(p, nullptr)) {
p->stop();
delete p;
ptr = nullptr;
return;
}
}
}
};
} } // namespace cppa::detail
......
......@@ -37,6 +37,7 @@
#include <type_traits>
#include "cppa/any_tuple.hpp"
#include "cppa/match_expr.hpp"
#include "cppa/partial_function.hpp"
#include "cppa/util/tbind.hpp"
......
......@@ -31,18 +31,20 @@
#ifndef CPPA_ACCEPTOR_HPP
#define CPPA_ACCEPTOR_HPP
#include <memory>
#include "cppa/config.hpp"
#include "cppa/option.hpp"
#include "cppa/util/input_stream.hpp"
#include "cppa/util/output_stream.hpp"
#include "cppa/network/input_stream.hpp"
#include "cppa/network/output_stream.hpp"
namespace cppa { namespace util {
namespace cppa { namespace network {
/**
* @brief A pair of input and output stream pointers.
*/
typedef std::pair<input_stream_ptr, output_stream_ptr> io_stream_ptr_pair;
typedef std::pair<input_stream_ptr,output_stream_ptr> io_stream_ptr_pair;
/**
* @brief Accepts connections from client processes.
......@@ -57,7 +59,7 @@ class acceptor {
* @brief Returns the internal file descriptor. This descriptor is needed
* for socket multiplexing using select().
*/
virtual native_socket_type acceptor_file_handle() const = 0;
virtual native_socket_type file_handle() const = 0;
/**
* @brief Accepts a new connection and returns an input/output stream pair.
......@@ -73,6 +75,9 @@ class acceptor {
};
typedef std::unique_ptr<acceptor> acceptor_uptr;
} } // namespace cppa::util
#endif // CPPA_ACCEPTOR_HPP
......@@ -39,7 +39,7 @@
#include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp"
namespace cppa { namespace detail {
namespace cppa { namespace network {
class addressed_message {
......@@ -106,6 +106,6 @@ inline bool operator!=(const addressed_message& lhs,
return !(lhs == rhs);
}
} } // namespace cppa::detail
} } // namespace cppa::network
#endif // CPPA_ADDRESSED_MESSAGE_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* 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 3 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/>. *
\******************************************************************************/
#ifndef CONTINUABLE_READER_HPP
#define CONTINUABLE_READER_HPP
#include "cppa/actor.hpp"
#include "cppa/config.hpp"
#include "cppa/ref_counted.hpp"
namespace cppa { namespace network {
class middleman;
enum continue_reading_result {
read_failure,
read_closed,
read_continue_later
};
class continuable_reader : public ref_counted {
public:
/**
* @brief Returns the file descriptor for incoming data.
*/
inline native_socket_type read_handle() const {
return m_read_handle;
}
/**
* @brief Reads from {@link read_handle()}.
*/
virtual continue_reading_result continue_reading() = 0;
/**
* @brief Returns @p true if @p this is a {@link peer_acceptor} that
* is assigned to the published actor @p whom.
*/
virtual bool is_acceptor_of(const actor_ptr& whom) const;
/**
* @brief Returns true if this is a subtype of {@link peer}, i.e.,
* if @p static_cast<peer*>(this) is well-defined.
*/
inline bool is_peer() const {
return m_is_peer;
}
protected:
continuable_reader(middleman* parent, native_socket_type rd, bool is_peer);
inline middleman* parent() {
return m_parent;
}
inline const middleman* parent() const {
return m_parent;
}
private:
bool m_is_peer;
middleman* m_parent;
native_socket_type m_read_handle;
};
typedef intrusive_ptr<continuable_reader> continuable_reader_ptr;
} } // namespace cppa::network
#endif // CONTINUABLE_READER_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* 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 3 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/>. *
\******************************************************************************/
#ifndef IPV4_PEER_ACCEPTOR_HPP
#define IPV4_PEER_ACCEPTOR_HPP
#include "cppa/actor.hpp"
#include "cppa/network/ipv4_acceptor.hpp"
#include "cppa/network/peer_acceptor.hpp"
namespace cppa { namespace network {
class default_peer_acceptor_impl : public peer_acceptor {
typedef peer_acceptor super;
public:
continue_reading_result continue_reading();
default_peer_acceptor_impl(middleman* parent,
acceptor_uptr ptr,
const actor_ptr& published_actor);
private:
acceptor_uptr m_ptr;
};
} } // namespace cppa::detail
#endif // IPV4_PEER_ACCEPTOR_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* 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 3 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/>. *
\******************************************************************************/
#ifndef IPV4_PEER_HPP
#define IPV4_PEER_HPP
#include "cppa/process_information.hpp"
#include "cppa/util/buffer.hpp"
#include "cppa/network/peer.hpp"
#include "cppa/network/input_stream.hpp"
#include "cppa/network/output_stream.hpp"
namespace cppa { namespace network {
class default_peer_impl : public peer {
typedef peer super;
public:
default_peer_impl(middleman* parent,
const input_stream_ptr& in,
const output_stream_ptr& out,
process_information_ptr peer_ptr = nullptr);
continue_reading_result continue_reading();
continue_writing_result continue_writing();
bool enqueue(const addressed_message& msg);
protected:
~default_peer_impl();
private:
enum read_state {
// connection just established; waiting for process information
wait_for_process_info,
// wait for the size of the next message
wait_for_msg_size,
// currently reading a message
read_message
};
input_stream_ptr m_in;
output_stream_ptr m_out;
read_state m_state;
process_information_ptr m_peer;
const uniform_type_info* m_meta_msg;
bool m_has_unwritten_data;
util::buffer m_rd_buf;
util::buffer m_wr_buf;
};
} } // namespace cppa::network
#endif // IPV4_PEER_HPP
......@@ -35,7 +35,7 @@
#include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp"
namespace cppa { namespace util {
namespace cppa { namespace network {
/**
* @brief An abstract input stream interface.
......@@ -48,7 +48,7 @@ class input_stream : public virtual ref_counted {
* @brief Returns the internal file descriptor. This descriptor is needed
* for socket multiplexing using select().
*/
virtual native_socket_type read_file_handle() const = 0;
virtual native_socket_type read_handle() const = 0;
/**
* @brief Reads exactly @p num_bytes from the data source and blocks the
......
......@@ -31,10 +31,10 @@
#ifndef CPPA_IO_STREAM_HPP
#define CPPA_IO_STREAM_HPP
#include "cppa/util/input_stream.hpp"
#include "cppa/util/output_stream.hpp"
#include "cppa/network/input_stream.hpp"
#include "cppa/network/output_stream.hpp"
namespace cppa { namespace util {
namespace cppa { namespace network {
/**
* @brief A stream capable of both reading and writing.
......
......@@ -33,25 +33,26 @@
#include <memory>
#include <cstdint>
#include "cppa/config.hpp"
#include "cppa/util/acceptor.hpp"
#include "cppa/network/acceptor.hpp"
namespace cppa { namespace detail {
namespace cppa { namespace network {
class ipv4_acceptor : public util::acceptor {
class ipv4_acceptor : public acceptor {
public:
static std::unique_ptr<util::acceptor> create(std::uint16_t port,
const char* addr);
static std::unique_ptr<acceptor> create(std::uint16_t port,
const char* addr);
~ipv4_acceptor();
native_socket_type acceptor_file_handle() const;
native_socket_type file_handle() const;
util::io_stream_ptr_pair accept_connection();
io_stream_ptr_pair accept_connection();
option<util::io_stream_ptr_pair> try_accept_connection();
option<io_stream_ptr_pair> try_accept_connection();
private:
......
......@@ -32,21 +32,21 @@
#define CPPA_IPV4_IO_STREAM_HPP
#include "cppa/config.hpp"
#include "cppa/util/io_stream.hpp"
#include "cppa/network/io_stream.hpp"
namespace cppa { namespace detail {
namespace cppa { namespace network {
class ipv4_io_stream : public util::io_stream {
class ipv4_io_stream : public io_stream {
public:
static util::io_stream_ptr connect_to(const char* host, std::uint16_t port);
static io_stream_ptr connect_to(const char* host, std::uint16_t port);
static util::io_stream_ptr from_native_socket(native_socket_type fd);
static io_stream_ptr from_native_socket(native_socket_type fd);
native_socket_type read_file_handle() const;
native_socket_type read_handle() const;
native_socket_type write_file_handle() const;
native_socket_type write_handle() const;
void read(void* buf, size_t len);
......
......@@ -31,88 +31,105 @@
#ifndef MIDDLEMAN_HPP
#define MIDDLEMAN_HPP
#include <map>
#include <vector>
#include <memory>
#include "cppa/actor.hpp"
#include "cppa/process_information.hpp"
#include "cppa/network/peer.hpp"
#include "cppa/network/acceptor.hpp"
#include "cppa/network/peer_acceptor.hpp"
#include "cppa/network/continuable_reader.hpp"
#include "cppa/util/acceptor.hpp"
namespace cppa { namespace detail { class singleton_manager; } }
#include "cppa/intrusive/single_reader_queue.hpp"
namespace cppa { namespace network {
#include "cppa/detail/network_manager.hpp"
#include "cppa/detail/addressed_message.hpp"
#include "cppa/detail/singleton_manager.hpp"
class middleman_impl;
class middleman_overseer;
class middleman_event_handler;
namespace cppa { namespace detail {
void middleman_loop(middleman_impl*);
enum class middleman_message_type {
add_peer,
publish,
unpublish,
outgoing_message,
shutdown
};
class middleman {
friend class peer;
friend class peer_acceptor;
friend class singleton_manager;
friend class middleman_overseer;
friend class middleman_event_handler;
friend class detail::singleton_manager;
friend void middleman_loop(middleman_impl*);
public:
virtual ~middleman();
virtual void publish(std::unique_ptr<acceptor> server,
const actor_ptr& published_actor) = 0;
virtual void add_peer(const io_stream_ptr_pair& io,
const process_information_ptr& node_info) = 0;
virtual void unpublish(const actor_ptr& whom) = 0;
struct middleman_message {
middleman_message* next;
const middleman_message_type type;
union {
std::pair<util::io_stream_ptr_pair, process_information_ptr> new_peer;
std::pair<std::unique_ptr<util::acceptor>, actor_ptr> new_published_actor;
actor_ptr published_actor;
std::pair<process_information_ptr, addressed_message> out_msg;
};
middleman_message();
middleman_message(util::io_stream_ptr_pair, process_information_ptr);
middleman_message(std::unique_ptr<util::acceptor>, actor_ptr);
middleman_message(process_information_ptr, addressed_message);
middleman_message(actor_ptr);
~middleman_message();
template<typename... Args>
static inline std::unique_ptr<middleman_message> create(Args&&... args) {
return std::unique_ptr<middleman_message>(new middleman_message(std::forward<Args>(args)...));
virtual void enqueue(const process_information_ptr& receiving_node,
const addressed_message& message) = 0;
inline void enqueue(const process_information_ptr& receiving_node,
actor_ptr sender,
channel_ptr receiver,
any_tuple msg,
message_id_t id = message_id_t()) {
enqueue(receiving_node,
addressed_message(std::move(sender),
std::move(receiver),
std::move(msg),
id));
}
};
typedef intrusive::single_reader_queue<middleman_message> middleman_queue;
void middleman_loop(int pipe_rd, middleman_queue& queue);
template<typename... Args>
inline void send2mm(Args&&... args) {
auto nm = singleton_manager::get_network_manager();
nm->send_to_middleman(middleman_message::create(std::forward<Args>(args)...));
}
inline void middleman_add_peer(util::io_stream_ptr_pair peer_streams,
process_information_ptr peer_ptr ) {
send2mm(std::move(peer_streams), std::move(peer_ptr));
}
inline void middleman_publish(std::unique_ptr<util::acceptor> server,
actor_ptr published_actor ) {
send2mm(std::move(server), std::move(published_actor));
}
inline void middleman_unpublish(actor_ptr whom) {
send2mm(std::move(whom));
}
inline void middleman_enqueue(process_information_ptr peer,
addressed_message outgoing_message) {
send2mm(std::move(peer), std::move(outgoing_message));
}
inline void middleman_enqueue(process_information_ptr peer,
actor_ptr sender,
channel_ptr receiver,
any_tuple&& msg,
message_id_t id = message_id_t()) {
addressed_message amsg(std::move(sender), std::move(receiver),
std::move(msg), id);
send2mm(std::move(peer), std::move(amsg));
}
protected:
middleman();
virtual void stop() = 0;
virtual void start() = 0;
private:
// to be called from singleton_manager
static middleman* create_singleton();
// to be called from peer
void continue_writing_later(const peer_ptr& ptr);
void register_peer(const process_information& node, const peer_ptr& ptr);
// to be called form peer_acceptor
void add(const continuable_reader_ptr& what);
// to be called from m_handler or middleman_overseer
inline void quit() { m_done = true; }
inline bool done() const { return m_done; }
void erase(const continuable_reader_ptr& what);
continuable_reader_ptr acceptor_of(const actor_ptr& whom);
peer_ptr get_peer(const process_information& node);
// member variables
bool m_done;
std::vector<continuable_reader_ptr> m_readers;
std::map<process_information,peer_ptr> m_peers;
std::unique_ptr<middleman_event_handler> m_handler;
};
} } // namespace cppa::detail
......
......@@ -35,7 +35,7 @@
#include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp"
namespace cppa { namespace util {
namespace cppa { namespace network {
/**
* @brief An abstract output stream interface.
......@@ -48,7 +48,7 @@ class output_stream : public virtual ref_counted {
* @brief Returns the internal file descriptor. This descriptor is needed
* for socket multiplexing using select().
*/
virtual native_socket_type write_file_handle() const = 0;
virtual native_socket_type write_handle() const = 0;
/**
* @brief Writes @p num_bytes bytes from @p buf to the data sink.
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* 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 3 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/>. *
\******************************************************************************/
#ifndef PEER_HPP
#define PEER_HPP
#include "cppa/network/addressed_message.hpp"
#include "cppa/network/continuable_reader.hpp"
enum continue_writing_result {
write_failure,
write_closed,
write_continue_later,
write_done
};
namespace cppa { namespace network {
class middleman;
/**
* @brief Represents a bidirectional connection to a peer.
*/
class peer : public continuable_reader {
typedef continuable_reader super;
public:
/**
* @brief Returns the file descriptor for outgoing data.
*/
native_socket_type write_handle() const {
return m_write_handle;
}
/**
* @brief Writes to {@link write_handle()}.
*/
virtual continue_writing_result continue_writing() = 0;
/**
* @brief Enqueues @p msg to the list of outgoing messages.
* @returns @p true on success, @p false otherwise.
* @note Implementation should call {@link begin_writing()} and perform IO
* only in its implementation of {@link continue_writing()}.
* @note Returning @p false from this function is interpreted as error
* and causes the middleman to remove this peer.
*/
virtual bool enqueue(const addressed_message& msg) = 0;
protected:
/**
* @brief Tells the middleman to add write_handle() to the list of
* observed sockets and to call continue_writing() if
* write_handle() is ready to write.
* @note Does nothing if write_handle() is already registered for the
* event loop.
*/
void begin_writing();
void register_peer(const process_information& pinfo);
peer(middleman* parent, native_socket_type rd, native_socket_type wr);
private:
native_socket_type m_write_handle;
};
typedef intrusive_ptr<peer> peer_ptr;
} } // namespace cppa::network
#endif // PEER_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* 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 3 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/>. *
\******************************************************************************/
#ifndef PEER_ACCEPTOR_HPP
#define PEER_ACCEPTOR_HPP
#include "cppa/network/peer.hpp"
#include "cppa/network/continuable_reader.hpp"
namespace cppa { namespace network {
class peer_acceptor : public continuable_reader {
typedef continuable_reader super;
public:
bool is_acceptor_of(const actor_ptr& whom) const;
protected:
peer_acceptor(middleman* parent,
native_socket_type fd,
const actor_ptr& published_actor);
void add_peer(const peer_ptr& ptr);
inline const actor_ptr& published_actor() const {
return m_published_actor;
}
private:
actor_ptr m_published_actor;
};
typedef intrusive_ptr<peer_acceptor> peer_acceptor_ptr;
} } // namespace cppa::network
#endif // PEER_ACCEPTOR_HPP
......@@ -57,7 +57,7 @@ template<>
struct has_cppa_to_string<any_tuple> : std::true_type { };
template<>
struct has_cppa_to_string<addressed_message> : std::true_type { };
struct has_cppa_to_string<network::addressed_message> : std::true_type { };
template<>
struct has_cppa_to_string<actor_ptr> : std::true_type { };
......
......@@ -33,6 +33,8 @@
#include <cstddef> // size_t
namespace cppa { namespace network { class input_stream; } }
namespace cppa { namespace util {
class input_stream;
......@@ -119,7 +121,7 @@ class buffer {
/**
* @brief Appends up to @p remaining() bytes from @p istream to the buffer.
*/
void append_from(input_stream* istream);
void append_from(network::input_stream* istream);
private:
......
......@@ -43,7 +43,7 @@
#include "cppa/process_information.hpp"
namespace cppa { class any_tuple; }
namespace cppa { namespace detail { class addressed_message; } }
namespace cppa { namespace network { class addressed_message; } }
namespace cppa { namespace util {
......@@ -71,7 +71,7 @@ template<>
struct is_builtin<any_tuple> : std::true_type { };
template<>
struct is_builtin<detail::addressed_message> : std::true_type { };
struct is_builtin<network::addressed_message> : std::true_type { };
template<>
struct is_builtin<actor_ptr> : std::true_type { };
......
......@@ -37,6 +37,7 @@
#include "cppa/detail/yield_interface.hpp"
#include "cppa/detail/abstract_scheduled_actor.hpp"
/*
namespace cppa { namespace detail {
namespace {
......@@ -226,3 +227,4 @@ bool scheduled_actor_dummy::attach(attachable*) {
}
} } // namespace cppa::detail
*/
......@@ -28,22 +28,32 @@
\******************************************************************************/
#include <utility>
#include <iostream>
#include "cppa/atom.hpp"
#include "cppa/to_string.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/actor_proxy.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/detail/middleman.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/detail/types_array.hpp"
#include "cppa/detail/network_manager.hpp"
#include "cppa/detail/singleton_manager.hpp"
#include <iostream>
using namespace std;
namespace cppa {
using detail::middleman_enqueue;
namespace {
template<typename... Args>
void middleman_enqueue(Args&&... args) {
detail::singleton_manager::get_middleman()->enqueue(forward<Args>(args)...);
}
} // namespace <anonymous>
actor_proxy::actor_proxy(std::uint32_t mid, const process_information_ptr& pptr)
: super(mid, pptr) { }
......
......@@ -37,8 +37,7 @@
#include "cppa/util/shared_lock_guard.hpp"
#include "cppa/util/upgrade_lock_guard.hpp"
#include "cppa/detail/middleman.hpp"
#include "cppa/detail/network_manager.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/detail/actor_proxy_cache.hpp"
#include "cppa/detail/singleton_manager.hpp"
......@@ -89,12 +88,11 @@ actor_proxy_ptr actor_proxy_cache::get_impl(const key_tuple& key, bool do_put) {
get_actor_proxy_cache().erase(result);
});
auto pself = process_information::get();
middleman_enqueue(peer,
nullptr,
nullptr,
make_any_tuple(atom("MONITOR"),
pself,
std::get<2>(key)));
singleton_manager::get_middleman()->enqueue(
peer,
nullptr,
nullptr,
make_any_tuple(atom("MONITOR"), pself, std::get<2>(key)));
return result;
}
......
......@@ -28,10 +28,10 @@
\******************************************************************************/
#include "cppa/detail/addressed_message.hpp"
#include "cppa/network/addressed_message.hpp"
#include "cppa/detail/singleton_manager.hpp"
namespace cppa { namespace detail {
namespace cppa { namespace network {
addressed_message::addressed_message(actor_ptr from,
channel_ptr to,
......
......@@ -33,7 +33,7 @@
#include <utility>
#include "cppa/util/buffer.hpp"
#include "cppa/util/input_stream.hpp"
#include "cppa/network/input_stream.hpp"
namespace cppa { namespace util {
......@@ -150,7 +150,7 @@ void buffer::write(size_t num_bytes, const void* data, buffer_write_policy wp) {
inc_size(num_bytes);
}
void buffer::append_from(input_stream* istream) {
void buffer::append_from(network::input_stream* istream) {
CPPA_REQUIRE(remaining() > 0);
auto num_bytes = istream->read_some(wr_ptr(), remaining());
if (num_bytes > 0) {
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* 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 3 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 "cppa/network/continuable_reader.hpp"
namespace cppa { namespace network {
continuable_reader::continuable_reader(middleman* parent,
native_socket_type rd,
bool is_peer)
: m_is_peer(is_peer), m_parent(parent), m_read_handle(rd) { }
bool continuable_reader::is_acceptor_of(const actor_ptr&) const {
return false;
}
} } // namespace cppa::network
......@@ -102,7 +102,7 @@ decorated_names_map::decorated_names_map() {
{ demangled<char32_t>(), mapped_int_name<char32_t>() },
{ "cppa::atom_value", "@atom" },
{ "cppa::any_tuple", "@<>" },
{ "cppa::detail::addressed_message", "@msg" },
{ "cppa::network::addressed_message", "@msg" },
{ "cppa::intrusive_ptr<cppa::actor>", "@actor" },
{ "cppa::intrusive_ptr<cppa::group>" ,"@group" },
{ "cppa::intrusive_ptr<cppa::channel>", "@channel" },
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* 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 3 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 <iostream>
#include <exception>
#include "cppa/process_information.hpp"
#include "cppa/network/default_peer_impl.hpp"
#include "cppa/network/default_peer_acceptor_impl.hpp"
#include "cppa/detail/demangle.hpp"
using namespace std;
namespace cppa { namespace network {
default_peer_acceptor_impl::default_peer_acceptor_impl(middleman* mm,
acceptor_uptr aur,
const actor_ptr& pa)
: super(mm, aur->file_handle(), pa), m_ptr(std::move(aur)) { }
continue_reading_result default_peer_acceptor_impl::continue_reading() {
for (;;) {
option<io_stream_ptr_pair> opt;
try { opt = m_ptr->try_accept_connection(); }
catch (...) { return read_failure; }
if (opt) {
auto& pair = *opt;
auto& pself = process_information::get();
uint32_t process_id = pself->process_id();
try {
actor_id aid = published_actor()->id();
pair.second->write(&aid, sizeof(actor_id));
pair.second->write(&process_id, sizeof(uint32_t));
pair.second->write(pself->node_id().data(),
pself->node_id().size());
add_peer(new default_peer_impl(parent(), pair.first, pair.second));
}
catch (exception& e) {
cerr << "*** exception while sending actor and process id; "
<< detail::demangle(typeid(e))
<< ", what(): " << e.what()
<< endl;
}
}
else return read_continue_later;
}
}
} } // namespace cppa::network
This diff is collapsed.
......@@ -59,7 +59,7 @@ void throw_io_failure(std::string&& what, bool add_errno_failure) {
throw std::ios_base::failure(std::move(what));
}
int rd_flags(file_descriptor fd) {
int rd_flags(native_socket_type fd) {
auto flags = fcntl(fd, F_GETFL, 0);
if (flags == -1) {
throw_io_failure("unable to read socket flags");
......@@ -67,11 +67,11 @@ int rd_flags(file_descriptor fd) {
return flags;
}
bool nonblocking(file_descriptor fd) {
bool nonblocking(native_socket_type fd) {
return (rd_flags(fd) & O_NONBLOCK) != 0;
}
void nonblocking(file_descriptor fd, bool new_value) {
void nonblocking(native_socket_type fd, bool new_value) {
auto rf = rd_flags(fd);
auto wf = new_value ? (rf | O_NONBLOCK) : (rf & (~(O_NONBLOCK)));
if (fcntl(fd, F_SETFL, wf) < 0) {
......
#include <iostream>
using namespace std;
int main() {
# ifdef __clang__
cout << __clang_major__
<< "."
<< __clang_minor__
<< endl;
# elif defined(__GNUC__)
cout << __GNUC__
<< "."
<< __GNUC_MINOR__
<< endl;
# else
cout << "0.0" << endl;
# endif
return 0;
}
......@@ -41,10 +41,10 @@
#include "cppa/deserializer.hpp"
#include "cppa/event_based_actor.hpp"
#include "cppa/detail/middleman.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/detail/types_array.hpp"
#include "cppa/detail/group_manager.hpp"
#include "cppa/detail/addressed_message.hpp"
#include "cppa/network/addressed_message.hpp"
#include "cppa/util/shared_spinlock.hpp"
#include "cppa/util/shared_lock_guard.hpp"
......
......@@ -33,10 +33,12 @@
#include <errno.h>
#include <iostream>
#include "cppa/exception.hpp"
#include "cppa/util/io_stream.hpp"
#include "cppa/network/io_stream.hpp"
#include "cppa/network/ipv4_acceptor.hpp"
#include "cppa/network/ipv4_io_stream.hpp"
#include "cppa/detail/fd_util.hpp"
#include "cppa/detail/ipv4_acceptor.hpp"
#include "cppa/detail/ipv4_io_stream.hpp"
#ifdef CPPA_WINDOWS
#else
......@@ -49,9 +51,9 @@
# include <netinet/tcp.h>
#endif
namespace cppa { namespace detail {
namespace cppa { namespace network {
using namespace fd_util;
using namespace ::cppa::detail::fd_util;
namespace {
......@@ -74,7 +76,9 @@ struct socket_guard {
};
bool accept_impl(util::io_stream_ptr_pair& result, native_socket_type fd, bool nonblocking) {
bool accept_impl(io_stream_ptr_pair& result,
native_socket_type fd,
bool nonblocking) {
sockaddr addr;
socklen_t addrlen;
memset(&addr, 0, sizeof(addr));
......@@ -87,7 +91,7 @@ bool accept_impl(util::io_stream_ptr_pair& result, native_socket_type fd, bool n
}
throw_io_failure("accept failed");
}
util::io_stream_ptr ptr(ipv4_io_stream::from_native_socket(sfd));
io_stream_ptr ptr(ipv4_io_stream::from_native_socket(sfd));
result.first = ptr;
result.second = ptr;
return true;
......@@ -98,8 +102,8 @@ bool accept_impl(util::io_stream_ptr_pair& result, native_socket_type fd, bool n
ipv4_acceptor::ipv4_acceptor(native_socket_type fd, bool nonblocking)
: m_fd(fd), m_is_nonblocking(nonblocking) { }
std::unique_ptr<util::acceptor> ipv4_acceptor::create(std::uint16_t port,
const char* addr) {
std::unique_ptr<acceptor> ipv4_acceptor::create(std::uint16_t port,
const char* addr) {
native_socket_type sockfd;
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd == invalid_socket) {
......@@ -109,7 +113,7 @@ std::unique_ptr<util::acceptor> ipv4_acceptor::create(std::uint16_t port,
socket_guard sguard(sockfd);
int on = 1;
if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)) < 0) {
detail::throw_io_failure("unable to set SO_REUSEADDR");
throw_io_failure("unable to set SO_REUSEADDR");
}
struct sockaddr_in serv_addr;
memset((char*) &serv_addr, 0, sizeof(serv_addr));
......@@ -140,26 +144,26 @@ ipv4_acceptor::~ipv4_acceptor() {
closesocket(m_fd);
}
native_socket_type ipv4_acceptor::acceptor_file_handle() const {
native_socket_type ipv4_acceptor::file_handle() const {
return m_fd;
}
util::io_stream_ptr_pair ipv4_acceptor::accept_connection() {
io_stream_ptr_pair ipv4_acceptor::accept_connection() {
if (m_is_nonblocking) {
nonblocking(m_fd, false);
m_is_nonblocking = false;
}
util::io_stream_ptr_pair result;
io_stream_ptr_pair result;
accept_impl(result, m_fd, m_is_nonblocking);
return result;
}
option<util::io_stream_ptr_pair> ipv4_acceptor::try_accept_connection() {
option<io_stream_ptr_pair> ipv4_acceptor::try_accept_connection() {
if (!m_is_nonblocking) {
nonblocking(m_fd, true);
m_is_nonblocking = true;
}
util::io_stream_ptr_pair result;
io_stream_ptr_pair result;
if (accept_impl(result, m_fd, m_is_nonblocking)) {
return result;
}
......
......@@ -35,7 +35,7 @@
#include "cppa/exception.hpp"
#include "cppa/detail/fd_util.hpp"
#include "cppa/detail/ipv4_io_stream.hpp"
#include "cppa/network/ipv4_io_stream.hpp"
#ifdef CPPA_WINDOWS
#else
......@@ -47,17 +47,17 @@
# include <netinet/tcp.h>
#endif
namespace cppa { namespace detail {
namespace cppa { namespace network {
using namespace fd_util;
using namespace ::cppa::detail::fd_util;
ipv4_io_stream::ipv4_io_stream(native_socket_type fd) : m_fd(fd) { }
native_socket_type ipv4_io_stream::read_file_handle() const {
native_socket_type ipv4_io_stream::read_handle() const {
return m_fd;
}
native_socket_type ipv4_io_stream::write_file_handle() const {
native_socket_type ipv4_io_stream::write_handle() const {
return m_fd;
}
......@@ -114,13 +114,13 @@ size_t ipv4_io_stream::write_some(const void* buf, size_t len) {
return static_cast<size_t>(send_result);
}
util::io_stream_ptr ipv4_io_stream::from_native_socket(native_socket_type fd) {
network::io_stream_ptr ipv4_io_stream::from_native_socket(native_socket_type fd) {
tcp_nodelay(fd, true);
nonblocking(fd, true);
return new ipv4_io_stream(fd);
}
util::io_stream_ptr ipv4_io_stream::connect_to(const char* host,
network::io_stream_ptr ipv4_io_stream::connect_to(const char* host,
std::uint16_t port) {
struct sockaddr_in serv_addr;
struct hostent* server;
......
This diff is collapsed.
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* 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 3 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 <cstdio>
#include <thread>
#include <cstdint>
#include <cstring> // strerror
#include <unistd.h>
#include <iostream>
#include <sys/time.h>
#include <sys/types.h>
#include "cppa/self.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/thread_mapped_actor.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
#include "cppa/detail/fd_util.hpp"
#include "cppa/detail/middleman.hpp"
#include "cppa/detail/network_manager.hpp"
namespace {
using namespace cppa;
using namespace cppa::detail;
struct network_manager_impl : network_manager {
middleman_queue m_middleman_queue;
std::thread m_middleman_thread;
int pipe_fd[2];
void start() { // override
if (pipe(pipe_fd) != 0) {
CPPA_CRITICAL("cannot create pipe");
}
// store pipe read handle in local variables for lambda expression
int pipe_fd0 = pipe_fd[0];
fd_util::nonblocking(pipe_fd0, true);
// start threads
m_middleman_thread = std::thread([this, pipe_fd0] {
middleman_loop(pipe_fd0, this->m_middleman_queue);
});
}
void stop() { // override
//m_mailman->enqueue(nullptr, make_any_tuple(atom("DONE")));
send_to_middleman(middleman_message::create());
m_middleman_thread.join();
close(pipe_fd[0]);
close(pipe_fd[1]);
}
void send_to_middleman(std::unique_ptr<middleman_message> msg) {
m_middleman_queue._push_back(msg.release());
std::atomic_thread_fence(std::memory_order_seq_cst);
std::uint8_t dummy = 0;
if (write(pipe_fd[1], &dummy, sizeof(dummy)) != sizeof(dummy)) {
CPPA_CRITICAL("cannot write to pipe");
}
}
};
} // namespace <anonymous>
namespace cppa { namespace detail {
network_manager::~network_manager() { }
network_manager* network_manager::create_singleton() {
return new network_manager_impl;
}
} } // namespace cppa::detail
......@@ -28,33 +28,20 @@
\******************************************************************************/
#ifndef CPPA_NETWORK_MANAGER_HPP
#define CPPA_NETWORK_MANAGER_HPP
#include "cppa/network/peer.hpp"
#include "cppa/network/middleman.hpp"
#include <memory>
namespace cppa { namespace network {
//#include "cppa/detail/post_office.hpp"
peer::peer(middleman* parent, native_socket_type rd, native_socket_type wr)
: super(parent, rd, true), m_write_handle(wr) { }
namespace cppa { namespace detail {
void peer::begin_writing() {
parent()->continue_writing_later(this);
}
struct middleman_message;
void peer::register_peer(const process_information& pinfo) {
parent()->register_peer(pinfo, this);
}
class network_manager {
public:
virtual ~network_manager();
virtual void start() = 0;
virtual void stop() = 0;
virtual void send_to_middleman(std::unique_ptr<middleman_message> msg) = 0;
static network_manager* create_singleton();
};
} } // namespace cppa::detail
#endif // CPPA_NETWORK_MANAGER_HPP
} } // namespace cppa::network
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* 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 3 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 "cppa/network/middleman.hpp"
#include "cppa/network/peer_acceptor.hpp"
namespace cppa { namespace network {
peer_acceptor::peer_acceptor(middleman* mm, native_socket_type fd, const actor_ptr& pa)
: super(mm, fd, false), m_published_actor(pa) { }
bool peer_acceptor::is_acceptor_of(const actor_ptr& whom) const {
return m_published_actor == whom;
}
void peer_acceptor::add_peer(const peer_ptr& ptr) {
parent()->add(ptr);
}
} } // namespace cppa::network
......@@ -37,11 +37,12 @@
#include "cppa/local_actor.hpp"
#include "cppa/thread_mapped_actor.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/detail/actor_count.hpp"
#include "cppa/detail/empty_tuple.hpp"
#include "cppa/detail/group_manager.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/network_manager.hpp"
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/detail/decorated_names_map.hpp"
#include "cppa/detail/thread_pool_scheduler.hpp"
......@@ -57,30 +58,15 @@ using namespace cppa::detail;
//volatile uniform_type_info_map* s_uniform_type_info_map = 0;
std::atomic<uniform_type_info_map*> s_uniform_type_info_map;
std::atomic<decorated_names_map*> s_decorated_names_map;
std::atomic<network_manager*> s_network_manager;
std::atomic<network::middleman*> s_middleman;
std::atomic<actor_registry*> s_actor_registry;
std::atomic<group_manager*> s_group_manager;
std::atomic<empty_tuple*> s_empty_tuple;
std::atomic<scheduler*> s_scheduler;
template<typename T>
void stop_and_kill(std::atomic<T*>& ptr) {
for (;;) {
auto p = ptr.load();
if (p == nullptr) {
return;
}
else if (ptr.compare_exchange_weak(p, nullptr)) {
p->stop();
delete p;
ptr = nullptr;
return;
}
}
}
template<typename T>
T* lazy_get(std::atomic<T*>& ptr) {
T* result = ptr.load(std::memory_order_seq_cst);
......@@ -99,9 +85,11 @@ T* lazy_get(std::atomic<T*>& ptr) {
} // namespace <anonymous>
namespace cppa {
namespace cppa { void shutdown() { singleton_manager::shutdown(); } }
void shutdown() {
namespace cppa { namespace detail {
void singleton_manager::shutdown() {
if (self.unchecked() != nullptr) {
try { self.unchecked()->quit(exit_reason::normal); }
catch (actor_exited&) { }
......@@ -110,7 +98,7 @@ void shutdown() {
if (rptr) {
rptr->await_running_count_equal(0);
}
stop_and_kill(s_network_manager);
stop_and_kill(s_middleman);
stop_and_kill(s_scheduler);
std::atomic_thread_fence(std::memory_order_seq_cst);
// it's safe now to delete all other singletons now
......@@ -127,10 +115,6 @@ void shutdown() {
s_decorated_names_map = nullptr;
}
} // namespace cppa
namespace cppa { namespace detail {
actor_registry* singleton_manager::get_actor_registry() {
return lazy_get(s_actor_registry);
}
......@@ -155,14 +139,12 @@ bool singleton_manager::set_scheduler(scheduler* ptr) {
scheduler* expected = nullptr;
if (s_scheduler.compare_exchange_weak(expected, ptr)) {
ptr->start();
auto nm = network_manager::create_singleton();
network_manager* nm_expected = nullptr;
if (s_network_manager.compare_exchange_weak(nm_expected, nm)) {
nm->start();
}
else {
delete nm;
network::middleman* mm = network::middleman::create_singleton();
network::middleman* mm_expected = nullptr;
if (s_middleman.compare_exchange_weak(mm_expected, mm)) {
mm->start();
}
else delete mm;
return true;
}
else {
......@@ -171,13 +153,13 @@ bool singleton_manager::set_scheduler(scheduler* ptr) {
}
}
network_manager* singleton_manager::get_network_manager() {
network_manager* result = s_network_manager.load();
network::middleman* singleton_manager::get_middleman() {
auto result = s_middleman.load();
if (result == nullptr) {
scheduler* s = new thread_pool_scheduler;
// set_scheduler sets s_network_manager
set_scheduler(s);
return get_network_manager();
return get_middleman();
}
return result;
}
......
......@@ -47,7 +47,7 @@
#include "cppa/detail/demangle.hpp"
#include "cppa/detail/to_uniform_name.hpp"
#include "cppa/detail/addressed_message.hpp"
#include "cppa/network/addressed_message.hpp"
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/detail/decorated_names_map.hpp"
......
......@@ -49,23 +49,26 @@
#include "cppa/intrusive/single_reader_queue.hpp"
#include "cppa/detail/middleman.hpp"
#include "cppa/detail/ipv4_acceptor.hpp"
#include "cppa/detail/ipv4_io_stream.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/ipv4_acceptor.hpp"
#include "cppa/network/ipv4_io_stream.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/network_manager.hpp"
#include "cppa/detail/actor_proxy_cache.hpp"
#include "cppa/detail/singleton_manager.hpp"
namespace cppa {
void publish(actor_ptr whom, std::unique_ptr<util::acceptor> acceptor) {
using namespace detail;
using namespace network;
void publish(actor_ptr whom, std::unique_ptr<acceptor> acceptor) {
if (!whom && !acceptor) return;
detail::singleton_manager::get_actor_registry()->put(whom->id(), whom);
detail::middleman_publish(std::move(acceptor), whom);
singleton_manager::get_actor_registry()->put(whom->id(), whom);
singleton_manager::get_middleman()->publish(std::move(acceptor), whom);
}
actor_ptr remote_actor(util::io_stream_ptr_pair peer) {
actor_ptr remote_actor(io_stream_ptr_pair peer) {
auto pinf = process_information::get();
std::uint32_t process_id = pinf->process_id();
// throws on error
......@@ -84,23 +87,23 @@ actor_ptr remote_actor(util::io_stream_ptr_pair peer) {
std::cerr << "*** warning: remote_actor() called to access a local actor\n"
<< std::flush;
# endif
return detail::singleton_manager::get_actor_registry()->get(remote_actor_id);
return singleton_manager::get_actor_registry()->get(remote_actor_id);
}
//auto key = std::make_tuple(remote_actor_id, pinfptr->process_id(), pinfptr->node_id());
detail::middleman_add_peer(peer, pinfptr);
return detail::get_actor_proxy_cache().get_or_put(remote_actor_id,
pinfptr->process_id(),
pinfptr->node_id());
singleton_manager::get_middleman()->add_peer(peer, pinfptr);
return get_actor_proxy_cache().get_or_put(remote_actor_id,
pinfptr->process_id(),
pinfptr->node_id());
}
void publish(actor_ptr whom, std::uint16_t port, const char* addr) {
if (whom) publish(whom, detail::ipv4_acceptor::create(port, addr));
if (whom) publish(whom, ipv4_acceptor::create(port, addr));
}
actor_ptr remote_actor(const char* host, std::uint16_t port) {
// throws on error
util::io_stream_ptr peer = detail::ipv4_io_stream::connect_to(host, port);
util::io_stream_ptr_pair ptrpair(peer, peer);
io_stream_ptr peer = ipv4_io_stream::connect_to(host, port);
io_stream_ptr_pair ptrpair(peer, peer);
return remote_actor(ptrpair);
}
......
......@@ -56,12 +56,13 @@
#include "cppa/detail/object_array.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/to_uniform_name.hpp"
#include "cppa/detail/addressed_message.hpp"
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/detail/actor_proxy_cache.hpp"
#include "cppa/detail/uniform_type_info_map.hpp"
#include "cppa/detail/default_uniform_type_info_impl.hpp"
#include "cppa/network/addressed_message.hpp"
namespace cppa { namespace detail {
namespace {
......@@ -431,7 +432,7 @@ class any_tuple_tinfo : public util::abstract_uniform_type_info<any_tuple> {
};
class addr_msg_tinfo : public util::abstract_uniform_type_info<addressed_message> {
class addr_msg_tinfo : public util::abstract_uniform_type_info<network::addressed_message> {
string any_tuple_name;
string actor_ptr_name;
......@@ -441,7 +442,7 @@ class addr_msg_tinfo : public util::abstract_uniform_type_info<addressed_message
public:
virtual void serialize(const void* instance, serializer* sink) const {
auto& msg = *reinterpret_cast<const addressed_message*>(instance);
auto& msg = *reinterpret_cast<const network::addressed_message*>(instance);
auto& data = msg.content();
sink->begin_object(name());
actor_ptr_tinfo::s_serialize(msg.sender(), sink, actor_ptr_name);
......@@ -458,7 +459,7 @@ class addr_msg_tinfo : public util::abstract_uniform_type_info<addressed_message
virtual void deserialize(void* instance, deserializer* source) const {
assert_type_name(source);
source->begin_object(name());
auto& msg = *reinterpret_cast<addressed_message*>(instance);
auto& msg = *reinterpret_cast<network::addressed_message*>(instance);
actor_ptr_tinfo::s_deserialize(msg.sender(), source, actor_ptr_name);
channel_ptr_tinfo::s_deserialize(msg.receiver(),
source,
......@@ -690,7 +691,7 @@ uniform_type_info_map::uniform_type_info_map() {
insert({raw_name<group_ptr>()}, new group_ptr_tinfo);
insert({raw_name<channel_ptr>()}, new channel_ptr_tinfo);
insert({raw_name<atom_value>()}, new atom_value_tinfo);
insert({raw_name<detail::addressed_message>()}, new addr_msg_tinfo);
insert({raw_name<network::addressed_message>()}, new addr_msg_tinfo);
insert({raw_name<util::void_type>()}, new void_type_tinfo);
insert({raw_name<process_information_ptr>()}, new process_info_ptr_tinfo);
insert({raw_name<map<string,string>>()}, new default_uniform_type_info_impl<map<string,string>>);
......
......@@ -21,7 +21,7 @@
#include "cppa/deserializer.hpp"
#include "cppa/uniform_type_info.hpp"
#include "cppa/detail/types_array.hpp"
#include "cppa/detail/addressed_message.hpp"
#include "cppa/network/addressed_message.hpp"
#include "cppa/util/callable_trait.hpp"
......@@ -119,7 +119,7 @@ int main() {
std::uint8_t, std::uint16_t, std::uint32_t, std::uint64_t,
std::string, std::u16string, std::u32string,
float, double,
atom_value, any_tuple, detail::addressed_message,
atom_value, any_tuple, network::addressed_message,
actor_ptr, group_ptr,
channel_ptr, intrusive_ptr<process_information>
>::arr;
......@@ -140,7 +140,7 @@ int main() {
uniform_typeid<double>(),
uniform_typeid<atom_value>(),
uniform_typeid<any_tuple>(),
uniform_typeid<detail::addressed_message>(),
uniform_typeid<network::addressed_message>(),
uniform_typeid<actor_ptr>(),
uniform_typeid<group_ptr>(),
uniform_typeid<channel_ptr>(),
......
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