Unverified Commit bec8b3fe authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #626

Add basic UDP support
parents 078d2db1 42dcb781
...@@ -53,6 +53,10 @@ detach-utility-actors=true ...@@ -53,6 +53,10 @@ detach-utility-actors=true
; setting this to false allows fully deterministic execution in unit test and ; setting this to false allows fully deterministic execution in unit test and
; requires the user to trigger I/O manually ; requires the user to trigger I/O manually
detach-multiplexer=true detach-multiplexer=true
; enable or disable communication via the TCP transport protocol
middleman_enable_tcp=true
; enable or disable communication via the UDP transport protocol
middleman_enable_udp=false
; when compiling with logging enabled ; when compiling with logging enabled
[logger] [logger]
......
...@@ -257,6 +257,7 @@ public: ...@@ -257,6 +257,7 @@ public:
actor_system_config& set(const char* cn, config_value cv); actor_system_config& set(const char* cn, config_value cv);
// -- config parameters of the scheduler ------------------------------------- // -- config parameters of the scheduler -------------------------------------
atom_value scheduler_policy; atom_value scheduler_policy;
size_t scheduler_max_threads; size_t scheduler_max_threads;
size_t scheduler_max_throughput; size_t scheduler_max_throughput;
...@@ -298,6 +299,9 @@ public: ...@@ -298,6 +299,9 @@ public:
size_t middleman_heartbeat_interval; size_t middleman_heartbeat_interval;
bool middleman_detach_utility_actors; bool middleman_detach_utility_actors;
bool middleman_detach_multiplexer; bool middleman_detach_multiplexer;
bool middleman_enable_tcp;
bool middleman_enable_udp;
size_t middleman_cached_udp_buffers;
// -- config parameters of the OpenCL module --------------------------------- // -- config parameters of the OpenCL module ---------------------------------
......
...@@ -141,9 +141,15 @@ using unlink_atom = atom_constant<atom("unlink")>; ...@@ -141,9 +141,15 @@ using unlink_atom = atom_constant<atom("unlink")>;
/// Used for publishing actors at a given port. /// Used for publishing actors at a given port.
using publish_atom = atom_constant<atom("publish")>; using publish_atom = atom_constant<atom("publish")>;
/// Used for publishing actors at a given port.
using publish_udp_atom = atom_constant<atom("pub_udp")>;
/// Used for removing an actor/port mapping. /// Used for removing an actor/port mapping.
using unpublish_atom = atom_constant<atom("unpublish")>; using unpublish_atom = atom_constant<atom("unpublish")>;
/// Used for removing an actor/port mapping.
using unpublish_udp_atom = atom_constant<atom("unpub_udp")>;
/// Used for signalizing group membership. /// Used for signalizing group membership.
using subscribe_atom = atom_constant<atom("subscribe")>; using subscribe_atom = atom_constant<atom("subscribe")>;
...@@ -153,6 +159,9 @@ using unsubscribe_atom = atom_constant<atom("unsubscrib")>; ...@@ -153,6 +159,9 @@ using unsubscribe_atom = atom_constant<atom("unsubscrib")>;
/// Used for establishing network connections. /// Used for establishing network connections.
using connect_atom = atom_constant<atom("connect")>; using connect_atom = atom_constant<atom("connect")>;
/// Used for contacting a remote UDP endpoint
using contact_atom = atom_constant<atom("contact")>;
/// Used for opening ports or files. /// Used for opening ports or files.
using open_atom = atom_constant<atom("open")>; using open_atom = atom_constant<atom("open")>;
...@@ -168,6 +177,9 @@ using migrate_atom = atom_constant<atom("migrate")>; ...@@ -168,6 +177,9 @@ using migrate_atom = atom_constant<atom("migrate")>;
/// Used for triggering periodic operations. /// Used for triggering periodic operations.
using tick_atom = atom_constant<atom("tick")>; using tick_atom = atom_constant<atom("tick")>;
/// Used for pending out of order messages.
using pending_atom = atom_constant<atom("pending")>;
} // namespace caf } // namespace caf
namespace std { namespace std {
......
...@@ -111,7 +111,9 @@ enum class sec : uint8_t { ...@@ -111,7 +111,9 @@ enum class sec : uint8_t {
/// Stream aborted due to unexpected error. /// Stream aborted due to unexpected error.
unhandled_stream_error, unhandled_stream_error,
/// A function view was called without assigning an actor first. /// A function view was called without assigning an actor first.
bad_function_call = 40 bad_function_call = 40,
/// Feature is disabled in the actor system config.
feature_disabled,
}; };
/// @relates sec /// @relates sec
......
...@@ -137,6 +137,9 @@ actor_system_config::actor_system_config() ...@@ -137,6 +137,9 @@ actor_system_config::actor_system_config()
middleman_heartbeat_interval = 0; middleman_heartbeat_interval = 0;
middleman_detach_utility_actors = true; middleman_detach_utility_actors = true;
middleman_detach_multiplexer = true; middleman_detach_multiplexer = true;
middleman_enable_tcp = true;
middleman_enable_udp = false;
middleman_cached_udp_buffers = 10;
// fill our options vector for creating INI and CLI parsers // fill our options vector for creating INI and CLI parsers
opt_group{options_, "scheduler"} opt_group{options_, "scheduler"}
.add(scheduler_policy, "policy", .add(scheduler_policy, "policy",
...@@ -199,7 +202,14 @@ actor_system_config::actor_system_config() ...@@ -199,7 +202,14 @@ actor_system_config::actor_system_config()
.add(middleman_detach_utility_actors, "detach-utility-actors", .add(middleman_detach_utility_actors, "detach-utility-actors",
"enables or disables detaching of utility actors") "enables or disables detaching of utility actors")
.add(middleman_detach_multiplexer, "detach-multiplexer", .add(middleman_detach_multiplexer, "detach-multiplexer",
"enables or disables background activity of the multiplexer"); "enables or disables background activity of the multiplexer")
.add(middleman_enable_tcp, "enable-tcp",
"enable communication via TCP (on by default)")
.add(middleman_enable_udp, "enable-udp",
"enable communication via UDP (off by default)")
.add(middleman_cached_udp_buffers, "cached-udp-buffers",
"sets the max number of UDP send buffers that will be cached for reuse "
"(default: 10)");
opt_group(options_, "opencl") opt_group(options_, "opencl")
.add(opencl_device_ids, "device-ids", .add(opencl_device_ids, "device-ids",
"restricts which OpenCL devices are accessed by CAF"); "restricts which OpenCL devices are accessed by CAF");
......
...@@ -65,7 +65,8 @@ const char* sec_strings[] = { ...@@ -65,7 +65,8 @@ const char* sec_strings[] = {
"no_downstream_stages_defined", "no_downstream_stages_defined",
"stream_init_failed", "stream_init_failed",
"invalid_stream_state", "invalid_stream_state",
"bad_function_call" "bad_function_call",
"feature_disabled",
}; };
} // namespace <anonymous> } // namespace <anonymous>
......
...@@ -24,6 +24,13 @@ set (LIBCAF_IO_SRCS ...@@ -24,6 +24,13 @@ set (LIBCAF_IO_SRCS
src/scribe.cpp src/scribe.cpp
src/stream_manager.cpp src/stream_manager.cpp
src/test_multiplexer.cpp src/test_multiplexer.cpp
src/acceptor_manager.cpp
src/multiplexer.cpp
src/datagram_servant.cpp
src/datagram_manager.cpp
src/ip_endpoint.cpp
src/connection_helper.cpp
src/receive_buffer.cpp
# BASP files # BASP files
src/header.cpp src/header.cpp
src/message_type.cpp src/message_type.cpp
......
...@@ -30,12 +30,15 @@ ...@@ -30,12 +30,15 @@
#include "caf/io/fwd.hpp" #include "caf/io/fwd.hpp"
#include "caf/io/accept_handle.hpp" #include "caf/io/accept_handle.hpp"
#include "caf/io/receive_policy.hpp" #include "caf/io/receive_policy.hpp"
#include "caf/io/datagram_handle.hpp"
#include "caf/io/system_messages.hpp" #include "caf/io/system_messages.hpp"
#include "caf/io/connection_handle.hpp" #include "caf/io/connection_handle.hpp"
#include "caf/io/network/ip_endpoint.hpp"
#include "caf/io/network/native_socket.hpp" #include "caf/io/network/native_socket.hpp"
#include "caf/io/network/stream_manager.hpp" #include "caf/io/network/stream_manager.hpp"
#include "caf/io/network/acceptor_manager.hpp" #include "caf/io/network/acceptor_manager.hpp"
#include "caf/io/network/datagram_manager.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
...@@ -62,7 +65,7 @@ class middleman; ...@@ -62,7 +65,7 @@ class middleman;
/// sending its content via the network. Instead of actively receiving data, /// sending its content via the network. Instead of actively receiving data,
/// brokers configure a scribe to asynchronously receive data, e.g., /// brokers configure a scribe to asynchronously receive data, e.g.,
/// `self->configure_read(hdl, receive_policy::exactly(1024))` would /// `self->configure_read(hdl, receive_policy::exactly(1024))` would
/// configure the scribe associated to `hdl` to receive *exactly* 1024 bytes /// configure the scribe associated with `hdl` to receive *exactly* 1024 bytes
/// and generate a `new_data_msg` message for the broker once the /// and generate a `new_data_msg` message for the broker once the
/// data is available. The buffer in this message will be re-used by the /// data is available. The buffer in this message will be re-used by the
/// scribe to minimize memory usage and heap allocations. /// scribe to minimize memory usage and heap allocations.
...@@ -82,6 +85,7 @@ public: ...@@ -82,6 +85,7 @@ public:
// even brokers need friends // even brokers need friends
friend class scribe; friend class scribe;
friend class doorman; friend class doorman;
friend class datagram_servant;
// -- overridden modifiers of abstract_actor --------------------------------- // -- overridden modifiers of abstract_actor ---------------------------------
...@@ -136,23 +140,38 @@ public: ...@@ -136,23 +140,38 @@ public:
} }
} }
/// Modifies the receive policy for given connection. /// Modifies the receive policy for a given connection.
/// @param hdl Identifies the affected connection. /// @param hdl Identifies the affected connection.
/// @param cfg Contains the new receive policy. /// @param cfg Contains the new receive policy.
void configure_read(connection_handle hdl, receive_policy::config cfg); void configure_read(connection_handle hdl, receive_policy::config cfg);
/// Enables or disables write notifications for given connection. /// Enables or disables write notifications for a given connection.
void ack_writes(connection_handle hdl, bool enable); void ack_writes(connection_handle hdl, bool enable);
/// Returns the write buffer for given connection. /// Returns the write buffer for a given connection.
std::vector<char>& wr_buf(connection_handle hdl); std::vector<char>& wr_buf(connection_handle hdl);
/// Writes `data` into the buffer for given connection. /// Writes `data` into the buffer for a given connection.
void write(connection_handle hdl, size_t bs, const void* buf); void write(connection_handle hdl, size_t bs, const void* buf);
/// Sends the content of the buffer for given connection. /// Sends the content of the buffer for a given connection.
void flush(connection_handle hdl); void flush(connection_handle hdl);
/// Enables or disables write notifications for a given datagram socket.
void ack_writes(datagram_handle hdl, bool enable);
/// Returns the write buffer for a given sink.
std::vector<char>& wr_buf(datagram_handle hdl);
/// Enqueue a buffer to be sent as a datagram via a given endpoint.
void enqueue_datagram(datagram_handle, std::vector<char>);
/// Writes `data` into the buffer of a given sink.
void write(datagram_handle hdl, size_t data_size, const void* data);
/// Sends the content of the buffer to a UDP endpoint.
void flush(datagram_handle hdl);
/// Returns the middleman instance this broker belongs to. /// Returns the middleman instance this broker belongs to.
inline middleman& parent() { inline middleman& parent() {
return system().middleman(); return system().middleman();
...@@ -191,24 +210,75 @@ public: ...@@ -191,24 +210,75 @@ public:
add_tcp_doorman(uint16_t port = 0, const char* in = nullptr, add_tcp_doorman(uint16_t port = 0, const char* in = nullptr,
bool reuse_addr = false); bool reuse_addr = false);
/// Returns the remote address associated to `hdl` /// Adds a `datagram_servant` to this broker.
void add_datagram_servant(datagram_servant_ptr ptr);
/// Adds the `datagram_servant` under an additional `hdl`.
void add_hdl_for_datagram_servant(datagram_servant_ptr ptr,
datagram_handle hdl);
/// Creates and assigns a new `datagram_servant` from a given socket `fd`.
datagram_handle add_datagram_servant(network::native_socket fd);
/// Creates and assigns a new `datagram_servant` from a given socket `fd`
/// for the remote endpoint `ep`.
datagram_handle
add_datagram_servant_for_endpoint(network::native_socket fd,
const network::ip_endpoint& ep);
/// Creates a new `datagram_servant` for the remote endpoint `host` and `port`.
/// @returns The handle to the new `datagram_servant`.
expected<datagram_handle>
add_udp_datagram_servant(const std::string& host, uint16_t port);
/// Tries to open a local port and creates a `datagram_servant` managing it on
/// success. If `port == 0`, then the broker will ask the operating system to
/// pick a random port.
/// @returns The handle of the new `datagram_servant` and the assigned port.
expected<std::pair<datagram_handle, uint16_t>>
add_udp_datagram_servant(uint16_t port = 0, const char* in = nullptr,
bool reuse_addr = false);
/// Moves an initialized `datagram_servant` instance `ptr` from another broker
/// to this one.
void move_datagram_servant(datagram_servant_ptr ptr);
/// Returns the remote address associated with `hdl`
/// or empty string if `hdl` is invalid. /// or empty string if `hdl` is invalid.
std::string remote_addr(connection_handle hdl); std::string remote_addr(connection_handle hdl);
/// Returns the remote port associated to `hdl` /// Returns the remote port associated with `hdl`
/// or `0` if `hdl` is invalid. /// or `0` if `hdl` is invalid.
uint16_t remote_port(connection_handle hdl); uint16_t remote_port(connection_handle hdl);
/// Returns the local address associated to `hdl` /// Returns the local address associated with `hdl`
/// or empty string if `hdl` is invalid. /// or empty string if `hdl` is invalid.
std::string local_addr(accept_handle hdl); std::string local_addr(accept_handle hdl);
/// Returns the local port associated to `hdl` or `0` if `hdl` is invalid. /// Returns the local port associated with `hdl` or `0` if `hdl` is invalid.
uint16_t local_port(accept_handle hdl); uint16_t local_port(accept_handle hdl);
/// Returns the handle associated to given local `port` or `none`. /// Returns the handle associated with given local `port` or `none`.
accept_handle hdl_by_port(uint16_t port); accept_handle hdl_by_port(uint16_t port);
/// Returns the dgram handle associated with given local `port` or `none`.
datagram_handle datagram_hdl_by_port(uint16_t port);
/// Returns the remote address associated with `hdl`
/// or an empty string if `hdl` is invalid.
std::string remote_addr(datagram_handle hdl);
/// Returns the remote port associated with `hdl`
/// or `0` if `hdl` is invalid.
uint16_t remote_port(datagram_handle hdl);
/// Returns the remote port associated with `hdl`
/// or `0` if `hdl` is invalid.
uint16_t local_port(datagram_handle hdl);
/// Remove the endpoint `hdl` from the broker.
bool remove_endpoint(datagram_handle hdl);
/// Closes all connections and acceptors. /// Closes all connections and acceptors.
void close_all(); void close_all();
...@@ -237,6 +307,30 @@ public: ...@@ -237,6 +307,30 @@ public:
if (i != elements.end()) if (i != elements.end())
elements.erase(i); elements.erase(i);
} }
// meta programming utility (not implemented)
static intrusive_ptr<doorman> ptr_of(accept_handle);
// meta programming utility (not implemented)
static intrusive_ptr<scribe> ptr_of(connection_handle);
// meta programming utility (not implemented)
static intrusive_ptr<datagram_servant> ptr_of(datagram_handle);
/// Returns an intrusive pointer to a `scribe` or `doorman`
/// identified by `hdl` and remove it from this broker.
template <class Handle>
auto take(Handle hdl) -> decltype(ptr_of(hdl)) {
using std::swap;
auto& elements = get_map(hdl);
decltype(ptr_of(hdl)) result;
auto i = elements.find(hdl);
if (i == elements.end())
return nullptr;
swap(result, i->second);
elements.erase(i);
return result;
}
/// @endcond /// @endcond
// -- overridden observers of abstract_actor --------------------------------- // -- overridden observers of abstract_actor ---------------------------------
...@@ -270,6 +364,8 @@ protected: ...@@ -270,6 +364,8 @@ protected:
using scribe_map = std::unordered_map<connection_handle, using scribe_map = std::unordered_map<connection_handle,
intrusive_ptr<scribe>>; intrusive_ptr<scribe>>;
using datagram_servant_map
= std::unordered_map<datagram_handle, intrusive_ptr<datagram_servant>>;
/// @cond PRIVATE /// @cond PRIVATE
// meta programming utility // meta programming utility
...@@ -282,12 +378,9 @@ protected: ...@@ -282,12 +378,9 @@ protected:
return scribes_; return scribes_;
} }
// meta programming utility (not implemented) inline datagram_servant_map& get_map(datagram_handle) {
static intrusive_ptr<doorman> ptr_of(accept_handle); return datagram_servants_;
}
// meta programming utility (not implemented)
static intrusive_ptr<scribe> ptr_of(connection_handle);
/// @endcond /// @endcond
/// Returns a `scribe` or `doorman` identified by `hdl`. /// Returns a `scribe` or `doorman` identified by `hdl`.
...@@ -300,21 +393,6 @@ protected: ...@@ -300,21 +393,6 @@ protected:
return *(i->second); return *(i->second);
} }
/// Returns an intrusive pointer to a `scribe` or `doorman`
/// identified by `hdl` and remove it from this broker.
template <class Handle>
auto take(Handle hdl) -> decltype(ptr_of(hdl)) {
using std::swap;
auto& elements = get_map(hdl);
decltype(ptr_of(hdl)) result;
auto i = elements.find(hdl);
if (i == elements.end())
return nullptr;
swap(result, i->second);
elements.erase(i);
return result;
}
private: private:
inline void launch_servant(scribe_ptr&) { inline void launch_servant(scribe_ptr&) {
// nop // nop
...@@ -322,6 +400,8 @@ private: ...@@ -322,6 +400,8 @@ private:
void launch_servant(doorman_ptr& ptr); void launch_servant(doorman_ptr& ptr);
void launch_servant(datagram_servant_ptr& ptr);
template <class T> template <class T>
typename T::handle_type add_servant(intrusive_ptr<T>&& ptr) { typename T::handle_type add_servant(intrusive_ptr<T>&& ptr) {
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
...@@ -345,6 +425,7 @@ private: ...@@ -345,6 +425,7 @@ private:
scribe_map scribes_; scribe_map scribes_;
doorman_map doormen_; doorman_map doormen_;
datagram_servant_map datagram_servants_;
detail::intrusive_partitioned_list<mailbox_element, detail::disposer> cache_; detail::intrusive_partitioned_list<mailbox_element, detail::disposer> cache_;
std::vector<char> dummy_wr_buf_; std::vector<char> dummy_wr_buf_;
}; };
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_BASP_ENDPOINT_CONTEXT_HPP
#define CAF_IO_BASP_ENDPOINT_CONTEXT_HPP
#include <unordered_map>
#include "caf/variant.hpp"
#include "caf/response_promise.hpp"
#include "caf/io/datagram_handle.hpp"
#include "caf/io/connection_handle.hpp"
#include "caf/io/basp/header.hpp"
#include "caf/io/basp/connection_state.hpp"
namespace caf {
namespace io {
namespace basp {
// stores meta information for active endpoints
struct endpoint_context {
using pending_map = std::unordered_map<uint16_t,
std::pair<basp::header,
std::vector<char>>>;
// denotes what message we expect from the remote node next
basp::connection_state cstate;
// our currently processed BASP header
basp::header hdr;
// the handle for I/O operations
variant<connection_handle, datagram_handle> hdl;
// network-agnostic node identifier
node_id id;
// ports
uint16_t remote_port;
uint16_t local_port;
// pending operations to be performed after handshake completed
optional<response_promise> callback;
// protocols that do not implement ordering are ordered by CAF
bool requires_ordering;
// sequence numbers and a buffer to establish order
uint16_t seq_incoming;
uint16_t seq_outgoing;
pending_map pending;
};
} // namespace basp
} // namespace io
} // namespace caf
#endif // CAF_IO_BASP_ENDPOINT_CONTEXT_HPP
...@@ -37,6 +37,9 @@ namespace basp { ...@@ -37,6 +37,9 @@ namespace basp {
/// @addtogroup BASP /// @addtogroup BASP
/// Sequence number type for BASP headers.
using sequence_type = uint16_t;
/// The header of a Binary Actor System Protocol (BASP) message. /// The header of a Binary Actor System Protocol (BASP) message.
/// A BASP header consists of a routing part, i.e., source and /// A BASP header consists of a routing part, i.e., source and
/// destination, as well as an operation and operation data. Several /// destination, as well as an operation and operation data. Several
...@@ -52,6 +55,7 @@ struct header { ...@@ -52,6 +55,7 @@ struct header {
node_id dest_node; node_id dest_node;
actor_id source_actor; actor_id source_actor;
actor_id dest_actor; actor_id dest_actor;
sequence_type sequence_number;
inline header(message_type m_operation, uint8_t m_flags, inline header(message_type m_operation, uint8_t m_flags,
uint32_t m_payload_len, uint64_t m_operation_data, uint32_t m_payload_len, uint64_t m_operation_data,
...@@ -64,7 +68,25 @@ struct header { ...@@ -64,7 +68,25 @@ struct header {
source_node(std::move(m_source_node)), source_node(std::move(m_source_node)),
dest_node(std::move(m_dest_node)), dest_node(std::move(m_dest_node)),
source_actor(m_source_actor), source_actor(m_source_actor),
dest_actor(m_dest_actor) { dest_actor(m_dest_actor),
sequence_number(0) {
// nop
}
inline header(message_type m_operation, uint8_t m_flags,
uint32_t m_payload_len, uint64_t m_operation_data,
node_id m_source_node, node_id m_dest_node,
actor_id m_source_actor, actor_id m_dest_actor,
sequence_type m_sequence_number)
: operation(m_operation),
flags(m_flags),
payload_len(m_payload_len),
operation_data(m_operation_data),
source_node(std::move(m_source_node)),
dest_node(std::move(m_dest_node)),
source_actor(m_source_actor),
dest_actor(m_dest_actor),
sequence_number(m_sequence_number) {
// nop // nop
} }
...@@ -87,8 +109,10 @@ typename Inspector::result_type inspect(Inspector& f, header& hdr) { ...@@ -87,8 +109,10 @@ typename Inspector::result_type inspect(Inspector& f, header& hdr) {
hdr.operation, hdr.operation,
meta::omittable(), pad, meta::omittable(), pad,
meta::omittable(), pad, meta::omittable(), pad,
hdr.flags, hdr.payload_len, hdr.operation_data, hdr.source_node, hdr.flags, hdr.payload_len, hdr.operation_data,
hdr.dest_node, hdr.source_actor, hdr.dest_actor); hdr.source_node, hdr.dest_node,
hdr.source_actor, hdr.dest_actor,
hdr.sequence_number);
} }
/// @relates header /// @relates header
...@@ -118,7 +142,8 @@ bool valid(const header& hdr); ...@@ -118,7 +142,8 @@ bool valid(const header& hdr);
constexpr size_t header_size = node_id::serialized_size * 2 constexpr size_t header_size = node_id::serialized_size * 2
+ sizeof(actor_id) * 2 + sizeof(actor_id) * 2
+ sizeof(uint32_t) * 2 + sizeof(uint32_t) * 2
+ sizeof(uint64_t); + sizeof(uint64_t)
+ sizeof(sequence_type);
/// @} /// @}
......
This diff is collapsed.
...@@ -26,10 +26,23 @@ ...@@ -26,10 +26,23 @@
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
#include "caf/callback.hpp" #include "caf/callback.hpp"
#include "caf/io/visitors.hpp"
#include "caf/io/abstract_broker.hpp" #include "caf/io/abstract_broker.hpp"
#include "caf/io/basp/buffer_type.hpp" #include "caf/io/basp/buffer_type.hpp"
namespace std {
template<>
struct hash<caf::variant<caf::io::connection_handle,caf::io::datagram_handle>> {
size_t operator()(const caf::variant<caf::io::connection_handle,
caf::io::datagram_handle>& hdl) const {
return caf::visit(caf::io::hash_visitor{}, hdl);
}
};
} // namespace std
namespace caf { namespace caf {
namespace io { namespace io {
namespace basp { namespace basp {
...@@ -40,15 +53,16 @@ namespace basp { ...@@ -40,15 +53,16 @@ namespace basp {
/// BASP peer and provides both direct and indirect paths. /// BASP peer and provides both direct and indirect paths.
class routing_table { class routing_table {
public: public:
using endpoint_handle = variant<connection_handle, datagram_handle>;
explicit routing_table(abstract_broker* parent); explicit routing_table(abstract_broker* parent);
virtual ~routing_table(); virtual ~routing_table();
/// Describes a routing path to a node. /// Describes a routing path to a node.
struct route { struct route {
buffer_type& wr_buf;
const node_id& next_hop; const node_id& next_hop;
connection_handle hdl; endpoint_handle hdl;
}; };
/// Describes a function object for erase operations that /// Describes a function object for erase operations that
...@@ -60,22 +74,19 @@ public: ...@@ -60,22 +74,19 @@ public:
/// Returns the ID of the peer connected via `hdl` or /// Returns the ID of the peer connected via `hdl` or
/// `none` if `hdl` is unknown. /// `none` if `hdl` is unknown.
node_id lookup_direct(const connection_handle& hdl) const; node_id lookup_direct(const endpoint_handle& hdl) const;
/// Returns the handle offering a direct connection to `nid` or /// Returns the handle offering a direct connection to `nid` or
/// `invalid_connection_handle` if no direct connection to `nid` exists. /// `invalid_connection_handle` if no direct connection to `nid` exists.
connection_handle lookup_direct(const node_id& nid) const; optional<endpoint_handle> lookup_direct(const node_id& nid) const;
/// Returns the next hop that would be chosen for `nid` /// Returns the next hop that would be chosen for `nid`
/// or `none` if there's no indirect route to `nid`. /// or `none` if there's no indirect route to `nid`.
node_id lookup_indirect(const node_id& nid) const; node_id lookup_indirect(const node_id& nid) const;
/// Flush output buffer for `r`.
void flush(const route& r);
/// Adds a new direct route to the table. /// Adds a new direct route to the table.
/// @pre `hdl != invalid_connection_handle && nid != none` /// @pre `hdl != invalid_connection_handle && nid != none`
void add_direct(const connection_handle& hdl, const node_id& nid); void add_direct(const endpoint_handle& hdl, const node_id& nid);
/// Adds a new indirect route to the table. /// Adds a new indirect route to the table.
bool add_indirect(const node_id& hop, const node_id& dest); bool add_indirect(const node_id& hop, const node_id& dest);
...@@ -86,7 +97,7 @@ public: ...@@ -86,7 +97,7 @@ public:
/// Removes a direct connection and calls `cb` for any node /// Removes a direct connection and calls `cb` for any node
/// that became unreachable as a result of this operation, /// that became unreachable as a result of this operation,
/// including the node that is assigned as direct path for `hdl`. /// including the node that is assigned as direct path for `hdl`.
void erase_direct(const connection_handle& hdl, erase_callback& cb); void erase_direct(const endpoint_handle& hdl, erase_callback& cb);
/// Removes any entry for indirect connection to `dest` and returns /// Removes any entry for indirect connection to `dest` and returns
/// `true` if `dest` had an indirect route, otherwise `false`. /// `true` if `dest` had an indirect route, otherwise `false`.
...@@ -101,6 +112,11 @@ public: ...@@ -101,6 +112,11 @@ public:
/// @returns the number of removed routes (direct and indirect) /// @returns the number of removed routes (direct and indirect)
size_t erase(const node_id& dest, erase_callback& cb); size_t erase(const node_id& dest, erase_callback& cb);
/// Returns the parent broker.
inline abstract_broker* parent() {
return parent_;
}
public: public:
template <class Map, class Fallback> template <class Map, class Fallback>
typename Map::mapped_type typename Map::mapped_type
...@@ -117,8 +133,8 @@ public: ...@@ -117,8 +133,8 @@ public:
node_id_set>; // hop node_id_set>; // hop
abstract_broker* parent_; abstract_broker* parent_;
std::unordered_map<connection_handle, node_id> direct_by_hdl_; std::unordered_map<endpoint_handle, node_id> direct_by_hdl_;
std::unordered_map<node_id, connection_handle> direct_by_nid_; std::unordered_map<node_id, endpoint_handle> direct_by_nid_;
indirect_entries indirect_; indirect_entries indirect_;
indirect_entries blacklist_; indirect_entries blacklist_;
}; };
...@@ -130,4 +146,3 @@ public: ...@@ -130,4 +146,3 @@ public:
} // namespace caf } // namespace caf
#endif // CAF_IO_BASP_ROUTING_TABLE_HPP #endif // CAF_IO_BASP_ROUTING_TABLE_HPP
...@@ -22,6 +22,7 @@ ...@@ -22,6 +22,7 @@
#include <map> #include <map>
#include <set> #include <set>
#include <stack>
#include <string> #include <string>
#include <future> #include <future>
#include <vector> #include <vector>
...@@ -36,8 +37,11 @@ ...@@ -36,8 +37,11 @@
#include "caf/io/basp/all.hpp" #include "caf/io/basp/all.hpp"
#include "caf/io/broker.hpp" #include "caf/io/broker.hpp"
#include "caf/io/visitors.hpp"
#include "caf/io/typed_broker.hpp" #include "caf/io/typed_broker.hpp"
#include "caf/io/basp/endpoint_context.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
...@@ -52,22 +56,22 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -52,22 +56,22 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// inherited from proxy_registry::backend // inherited from proxy_registry::backend
execution_unit* registry_context() override; execution_unit* registry_context() override;
// inherited from basp::instance::listener // inherited from basp::instance::callee
void finalize_handshake(const node_id& nid, actor_id aid, void finalize_handshake(const node_id& nid, actor_id aid,
std::set<std::string>& sigs) override; std::set<std::string>& sigs) override;
// inherited from basp::instance::listener // inherited from basp::instance::callee
void purge_state(const node_id& nid) override; void purge_state(const node_id& nid) override;
// inherited from basp::instance::listener // inherited from basp::instance::callee
void proxy_announced(const node_id& nid, actor_id aid) override; void proxy_announced(const node_id& nid, actor_id aid) override;
// inherited from basp::instance::listener // inherited from basp::instance::callee
void deliver(const node_id& src_nid, actor_id src_aid, void deliver(const node_id& src_nid, actor_id src_aid,
actor_id dest_aid, message_id mid, actor_id dest_aid, message_id mid,
std::vector<strong_actor_ptr>& stages, message& msg) override; std::vector<strong_actor_ptr>& stages, message& msg) override;
// inherited from basp::instance::listener // inherited from basp::instance::callee
void deliver(const node_id& src_nid, actor_id src_aid, void deliver(const node_id& src_nid, actor_id src_aid,
atom_value dest_name, message_id mid, atom_value dest_name, message_id mid,
std::vector<strong_actor_ptr>& stages, message& msg) override; std::vector<strong_actor_ptr>& stages, message& msg) override;
...@@ -80,38 +84,62 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -80,38 +84,62 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// performs bookkeeping such as managing `spawn_servers` // performs bookkeeping such as managing `spawn_servers`
void learned_new_node(const node_id& nid); void learned_new_node(const node_id& nid);
// inherited from basp::instance::listener // inherited from basp::instance::callee
void learned_new_node_directly(const node_id& nid, void learned_new_node_directly(const node_id& nid,
bool was_indirectly_before) override; bool was_indirectly_before) override;
// inherited from basp::instance::listener // inherited from basp::instance::callee
void learned_new_node_indirectly(const node_id& nid) override; void learned_new_node_indirectly(const node_id& nid) override;
// inherited from basp::instance::callee
uint16_t next_sequence_number(connection_handle hdl) override;
// inherited from basp::instance::callee
uint16_t next_sequence_number(datagram_handle hdl) override;
// inherited from basp::instance::callee
void add_pending(uint16_t seq, basp::endpoint_context& ep, basp::header hdr,
std::vector<char> payload) override;
// inherited from basp::instance::callee
bool deliver_pending(execution_unit* ctx,
basp::endpoint_context& ep) override;
// inherited from basp::instance::callee
void drop_pending(uint16_t seq, basp::endpoint_context& ep) override;
// inherited from basp::instance::callee
buffer_type& get_buffer(endpoint_handle hdl) override;
// inherited from basp::instance::callee
buffer_type& get_buffer(datagram_handle hdl) override;
// inherited from basp::instance::callee
buffer_type& get_buffer(connection_handle hdl) override;
// inherited from basp::instance::callee
buffer_type pop_datagram_buffer(datagram_handle hdl) override;
// inherited from basp::instance::callee
void flush(endpoint_handle hdl) override;
// inherited from basp::instance::callee
void flush(datagram_handle hdl) override;
// inherited from basp::instance::callee
void flush(connection_handle hdl) override;
void handle_heartbeat(const node_id&) override { void handle_heartbeat(const node_id&) override {
// nop // nop
} }
// stores meta information for open connections
struct connection_context {
// denotes what message we expect from the remote node next
basp::connection_state cstate;
// our currently processed BASP header
basp::header hdr;
// the connection handle for I/O operations
connection_handle hdl;
// network-agnostic node identifier
node_id id;
// connected port
uint16_t remote_port;
// pending operations to be performed after handhsake completed
optional<response_promise> callback;
};
/// Sets `this_context` by either creating or accessing state for `hdl`. /// Sets `this_context` by either creating or accessing state for `hdl`.
void set_context(connection_handle hdl); void set_context(connection_handle hdl);
void set_context(datagram_handle hdl);
/// Cleans up any state for `hdl`. /// Cleans up any state for `hdl`.
void cleanup(connection_handle hdl); void cleanup(connection_handle hdl);
void cleanup(datagram_handle hdl);
// pointer to ourselves // pointer to ourselves
broker* self; broker* self;
...@@ -119,13 +147,17 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -119,13 +147,17 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// protocol instance of BASP // protocol instance of BASP
basp::instance instance; basp::instance instance;
using ctx_map = std::unordered_map<connection_handle, connection_context>; using ctx_tcp_map = std::unordered_map<connection_handle,
basp::endpoint_context>;
using ctx_udp_map = std::unordered_map<datagram_handle,
basp::endpoint_context>;
// keeps context information for all open connections // keeps context information for all open connections
ctx_map ctx; ctx_tcp_map ctx_tcp;
ctx_udp_map ctx_udp;
// points to the current context for callbacks such as `make_proxy` // points to the current context for callbacks such as `make_proxy`
connection_context* this_context = nullptr; basp::endpoint_context* this_context = nullptr;
// stores handles to spawn servers for other nodes; these servers // stores handles to spawn servers for other nodes; these servers
// are spawned whenever the broker learns a new node ID and try to // are spawned whenever the broker learns a new node ID and try to
...@@ -136,6 +168,12 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -136,6 +168,12 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// to establish new connections at runtime to optimize // to establish new connections at runtime to optimize
// routing paths by forming a mesh between all nodes // routing paths by forming a mesh between all nodes
bool enable_automatic_connections = false; bool enable_automatic_connections = false;
bool enable_tcp = true;
bool enable_udp = false;
// reusable send buffers for UDP communication
const size_t max_buffers;
std::stack<buffer_type> cached_buffers;
// returns the node identifier of the underlying BASP instance // returns the node identifier of the underlying BASP instance
const node_id& this_node() const { const node_id& this_node() const {
......
...@@ -31,6 +31,7 @@ ...@@ -31,6 +31,7 @@
#include "caf/io/scribe.hpp" #include "caf/io/scribe.hpp"
#include "caf/io/doorman.hpp" #include "caf/io/doorman.hpp"
#include "caf/io/abstract_broker.hpp" #include "caf/io/abstract_broker.hpp"
#include "caf/io/datagram_servant.hpp"
#include "caf/mixin/sender.hpp" #include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.hpp" #include "caf/mixin/requester.hpp"
......
...@@ -102,7 +102,11 @@ protected: ...@@ -102,7 +102,11 @@ protected:
typename std::conditional< typename std::conditional<
std::is_same<handle_type, connection_handle>::value, std::is_same<handle_type, connection_handle>::value,
connection_passivated_msg, connection_passivated_msg,
acceptor_passivated_msg typename std::conditional<
std::is_same<handle_type, accept_handle>::value,
acceptor_passivated_msg,
datagram_servant_passivated_msg
>::type
>::type; >::type;
using tmp_t = mailbox_element_vals<passiv_t>; using tmp_t = mailbox_element_vals<passiv_t>;
tmp_t tmp{strong_actor_ptr{}, message_id::make(), tmp_t tmp{strong_actor_ptr{}, message_id::make(),
......
...@@ -71,7 +71,7 @@ public: ...@@ -71,7 +71,7 @@ public:
} // namespace io } // namespace io
} // namespace caf } // namespace caf
namespace std{ namespace std {
template<> template<>
struct hash<caf::io::connection_handle> { struct hash<caf::io::connection_handle> {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_CONNECTION_HELPER_HPP
#define CAF_IO_CONNECTION_HELPER_HPP
#include <chrono>
#include "caf/stateful_actor.hpp"
#include "caf/io/network/interfaces.hpp"
#include "caf/after.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/basp_broker.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/datagram_handle.hpp"
#include "caf/io/basp/all.hpp"
#include "caf/io/network/datagram_manager.hpp"
#include "caf/io/network/default_multiplexer.hpp"
namespace caf {
namespace io {
struct connection_helper_state {
static const char* name;
};
behavior datagram_connection_broker(broker* self,
uint16_t port,
network::address_listing addresses,
actor system_broker);
behavior connection_helper(stateful_actor<connection_helper_state>* self,
actor b);
} // namespace io
} // namespace caf
#endif // CAF_IO_CONNECTION_HELPER_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_DATAGRAM_HANDLE_HPP
#define CAF_IO_DATAGRAM_HANDLE_HPP
#include <functional>
#include "caf/error.hpp"
#include "caf/io/handle.hpp"
#include "caf/meta/type_name.hpp"
namespace caf {
namespace io {
struct invalid_datagram_handle_t {
constexpr invalid_datagram_handle_t() {
// nop
}
};
constexpr invalid_datagram_handle_t invalid_datagram_handle
= invalid_datagram_handle_t{};
/// Generic handle type for identifying datagram endpoints
class datagram_handle : public handle<datagram_handle,
invalid_datagram_handle_t> {
public:
friend class handle<datagram_handle, invalid_datagram_handle_t>;
using super = handle<datagram_handle, invalid_datagram_handle_t>;
constexpr datagram_handle() {
// nop
}
constexpr datagram_handle(const invalid_datagram_handle_t&) {
// nop
}
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f,
datagram_handle& x) {
return f(meta::type_name("datagram_handle"), x.id_);
}
private:
inline datagram_handle(int64_t handle_id) : super{handle_id} {
// nop
}
};
} // namespace io
} // namespace caf
namespace std {
template<>
struct hash<caf::io::datagram_handle> {
size_t operator()(const caf::io::datagram_handle& hdl) const {
return std::hash<int64_t>{}(hdl.id());
}
};
} // namespace std
#endif // CAF_IO_DATAGRAM_HANDLE_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_DATAGRAM_SERVANT_HPP
#define CAF_IO_DATAGRAM_SERVANT_HPP
#include <vector>
#include "caf/message.hpp"
#include "caf/io/datagram_handle.hpp"
#include "caf/io/broker_servant.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/network/ip_endpoint.hpp"
#include "caf/io/network/datagram_manager.hpp"
#include "caf/io/network/receive_buffer.hpp"
namespace caf {
namespace io {
using datagram_servant_base = broker_servant<network::datagram_manager,
datagram_handle, new_datagram_msg>;
/// Manages writing to a datagram sink.
/// @ingroup Broker
class datagram_servant : public datagram_servant_base {
public:
datagram_servant(datagram_handle hdl);
~datagram_servant() override;
/// Enables or disables write notifications.
virtual void ack_writes(bool enable) = 0;
/// Returns a new output buffer.
virtual std::vector<char>& wr_buf(datagram_handle) = 0;
/// Enqueue a buffer to be sent as a datagram.
virtual void enqueue_datagram(datagram_handle, std::vector<char>) = 0;
/// Returns the current input buffer.
virtual network::receive_buffer& rd_buf() = 0;
/// Flushes the output buffer, i.e., sends the
/// content of the buffer via the network.
virtual void flush() = 0;
/// Returns the local port of associated socket.
virtual uint16_t local_port() const = 0;
/// Returns all the handles associated with this servant
virtual std::vector<datagram_handle> hdls() const = 0;
/// Adds a new remote endpoint identified by the `ip_endpoint` to
/// the related manager.
virtual void add_endpoint(const network::ip_endpoint& ep,
datagram_handle hdl) = 0;
virtual void remove_endpoint(datagram_handle hdl) = 0;
void io_failure(execution_unit* ctx, network::operation op) override;
bool consume(execution_unit*, datagram_handle hdl,
network::receive_buffer& buf) override;
void datagram_sent(execution_unit*, datagram_handle hdl, size_t,
std::vector<char> buffer) override;
virtual void detach_handles() = 0;
using datagram_servant_base::new_endpoint;
virtual void launch() = 0;
protected:
message detach_message() override;
};
using datagram_servant_ptr = intrusive_ptr<datagram_servant>;
} // namespace io
} // namespace caf
// Allows the `middleman_actor` to create an `datagram_servant` and then send it
// to the BASP broker.
CAF_ALLOW_UNSAFE_MESSAGE_TYPE(caf::io::datagram_servant_ptr)
#endif // CAF_IO_DATAGRAM_SERVANT_HPP
...@@ -42,11 +42,13 @@ class middleman; ...@@ -42,11 +42,13 @@ class middleman;
class basp_broker; class basp_broker;
class receive_policy; class receive_policy;
class abstract_broker; class abstract_broker;
class datagram_servant;
// -- aliases ------------------------------------------------------------------ // -- aliases ------------------------------------------------------------------
using scribe_ptr = intrusive_ptr<scribe>; using scribe_ptr = intrusive_ptr<scribe>;
using doorman_ptr = intrusive_ptr<doorman>; using doorman_ptr = intrusive_ptr<doorman>;
using datagram_servant_ptr = intrusive_ptr<datagram_servant>;
// -- nested namespaces -------------------------------------------------------- // -- nested namespaces --------------------------------------------------------
......
...@@ -77,6 +77,22 @@ public: ...@@ -77,6 +77,22 @@ public:
system().message_types(tk), port, in, reuse); system().message_types(tk), port, in, reuse);
} }
/// Tries to publish `whom` at `port` and returns either an
/// `error` or the bound port.
/// @param whom Actor that should be published at `port`.
/// @param port Unused TCP port.
/// @param in The IP address to listen to or `INADDR_ANY` if `in == nullptr`.
/// @param reuse Create socket using `SO_REUSEADDR`.
/// @returns The actual port the OS uses after `bind()`. If `port == 0`
/// the OS chooses a random high-level port.
template <class Handle>
expected<uint16_t> publish_udp(Handle&& whom, uint16_t port,
const char* in = nullptr, bool reuse = false) {
detail::type_list<typename std::decay<Handle>::type> tk;
return publish_udp(actor_cast<strong_actor_ptr>(std::forward<Handle>(whom)),
system().message_types(tk), port, in, reuse);
}
/// Makes *all* local groups accessible via network /// Makes *all* local groups accessible via network
/// on address `addr` and `port`. /// on address `addr` and `port`.
/// @returns The actual port the OS uses after `bind()`. If `port == 0` /// @returns The actual port the OS uses after `bind()`. If `port == 0`
...@@ -93,6 +109,14 @@ public: ...@@ -93,6 +109,14 @@ public:
return unpublish(whom.address(), port); return unpublish(whom.address(), port);
} }
/// Unpublishes `whom` by closing `port` or all assigned ports if `port == 0`.
/// @param whom Actor that should be unpublished at `port`.
/// @param port UDP port.
template <class Handle>
expected<void> unpublish_udp(const Handle& whom, uint16_t port = 0) {
return unpublish_udp(whom.address(), port);
}
/// Establish a new connection to the actor at `host` on given `port`. /// Establish a new connection to the actor at `host` on given `port`.
/// @param host Valid hostname or IP address. /// @param host Valid hostname or IP address.
/// @param port TCP port. /// @param port TCP port.
...@@ -108,6 +132,21 @@ public: ...@@ -108,6 +132,21 @@ public:
return actor_cast<ActorHandle>(std::move(*x)); return actor_cast<ActorHandle>(std::move(*x));
} }
/// Contacts the actor at `host` on given `port`.
/// @param host Valid hostname or IP address.
/// @param port TCP port.
/// @returns An `actor` to the proxy instance representing
/// a remote actor or an `error`.
template <class ActorHandle = actor>
expected<ActorHandle> remote_actor_udp(std::string host, uint16_t port) {
detail::type_list<ActorHandle> tk;
auto x = remote_actor_udp(system().message_types(tk), std::move(host), port);
if (!x)
return x.error();
CAF_ASSERT(x && *x);
return actor_cast<ActorHandle>(std::move(*x));
}
/// <group-name>@<host>:<port> /// <group-name>@<host>:<port>
expected<group> remote_group(const std::string& group_uri); expected<group> remote_group(const std::string& group_uri);
...@@ -323,12 +362,20 @@ private: ...@@ -323,12 +362,20 @@ private:
std::set<std::string> sigs, std::set<std::string> sigs,
uint16_t port, const char* cstr, bool ru); uint16_t port, const char* cstr, bool ru);
expected<uint16_t> publish_udp(const strong_actor_ptr& whom,
std::set<std::string> sigs,
uint16_t port, const char* cstr, bool ru);
expected<void> unpublish(const actor_addr& whom, uint16_t port); expected<void> unpublish(const actor_addr& whom, uint16_t port);
expected<void> unpublish_udp(const actor_addr& whom, uint16_t port);
expected<strong_actor_ptr> remote_actor(std::set<std::string> ifs, expected<strong_actor_ptr> remote_actor(std::set<std::string> ifs,
std::string host, uint16_t port); std::string host, uint16_t port);
expected<strong_actor_ptr> remote_actor_udp(std::set<std::string> ifs,
std::string host, uint16_t port);
static int exec_slave_mode(actor_system&, const actor_system_config&); static int exec_slave_mode(actor_system&, const actor_system_config&);
// environment // environment
......
...@@ -70,6 +70,12 @@ namespace io { ...@@ -70,6 +70,12 @@ namespace io {
/// (unpublish_atom, strong_actor_ptr whom, uint16_t port) /// (unpublish_atom, strong_actor_ptr whom, uint16_t port)
/// -> void /// -> void
/// ///
/// // Closes `port` if it is mapped to `whom`.
/// // whom: A published actor.
/// // port: Used UDP port.
/// (unpublish_udp_atom, strong_actor_ptr whom, uint16_t port)
/// -> void
///
/// // Unconditionally closes `port`, removing any actor /// // Unconditionally closes `port`, removing any actor
/// // published at this port. /// // published at this port.
/// // port: Used TCP port. /// // port: Used TCP port.
...@@ -93,14 +99,24 @@ using middleman_actor = ...@@ -93,14 +99,24 @@ using middleman_actor =
std::set<std::string>, std::string, bool> std::set<std::string>, std::string, bool>
::with<uint16_t>, ::with<uint16_t>,
replies_to<publish_udp_atom, uint16_t, strong_actor_ptr,
std::set<std::string>, std::string, bool>
::with<uint16_t>,
replies_to<open_atom, uint16_t, std::string, bool> replies_to<open_atom, uint16_t, std::string, bool>
::with<uint16_t>, ::with<uint16_t>,
replies_to<connect_atom, std::string, uint16_t> replies_to<connect_atom, std::string, uint16_t>
::with<node_id, strong_actor_ptr, std::set<std::string>>, ::with<node_id, strong_actor_ptr, std::set<std::string>>,
replies_to<contact_atom, std::string, uint16_t>
::with<node_id, strong_actor_ptr, std::set<std::string>>,
reacts_to<unpublish_atom, actor_addr, uint16_t>, reacts_to<unpublish_atom, actor_addr, uint16_t>,
reacts_to<unpublish_udp_atom, actor_addr, uint16_t>,
reacts_to<close_atom, uint16_t>, reacts_to<close_atom, uint16_t>,
replies_to<spawn_atom, node_id, std::string, message, std::set<std::string>> replies_to<spawn_atom, node_id, std::string, message, std::set<std::string>>
......
...@@ -38,9 +38,11 @@ public: ...@@ -38,9 +38,11 @@ public:
using mpi_set = std::set<std::string>; using mpi_set = std::set<std::string>;
using get_res = delegated<node_id, strong_actor_ptr, mpi_set>; using get_res = result<node_id, strong_actor_ptr, mpi_set>;
using del_res = delegated<void>; using get_delegated = delegated<node_id, strong_actor_ptr, mpi_set>;
using del_res = result<void>;
using endpoint_data = std::tuple<node_id, strong_actor_ptr, mpi_set>; using endpoint_data = std::tuple<node_id, strong_actor_ptr, mpi_set>;
...@@ -59,21 +61,36 @@ protected: ...@@ -59,21 +61,36 @@ protected:
/// calls `system().middleman().backend().new_tcp_scribe(host, port)`. /// calls `system().middleman().backend().new_tcp_scribe(host, port)`.
virtual expected<scribe_ptr> connect(const std::string& host, uint16_t port); virtual expected<scribe_ptr> connect(const std::string& host, uint16_t port);
/// Tries to connect to given `host` and `port`. The default implementation
/// calls `system().middleman().backend().new_udp`.
virtual expected<datagram_servant_ptr> contact(const std::string& host,
uint16_t port);
/// Tries to open a local port. The default implementation calls /// Tries to open a local port. The default implementation calls
/// `system().middleman().backend().new_tcp_doorman(port, addr, reuse)`. /// `system().middleman().backend().new_tcp_doorman(port, addr, reuse)`.
virtual expected<doorman_ptr> open(uint16_t port, const char* addr, virtual expected<doorman_ptr> open(uint16_t port, const char* addr,
bool reuse); bool reuse);
/// Tries to open a local port. The default implementation calls
/// `system().middleman().backend().new_tcp_doorman(port, addr, reuse)`.
virtual expected<datagram_servant_ptr> open_udp(uint16_t port,
const char* addr, bool reuse);
private: private:
result<uint16_t> put(uint16_t port, strong_actor_ptr& whom, mpi_set& sigs, put_res put(uint16_t port, strong_actor_ptr& whom, mpi_set& sigs,
const char* in = nullptr, bool reuse_addr = false);
put_res put_udp(uint16_t port, strong_actor_ptr& whom, mpi_set& sigs,
const char* in = nullptr, bool reuse_addr = false); const char* in = nullptr, bool reuse_addr = false);
optional<endpoint_data&> cached(const endpoint& ep); optional<endpoint_data&> cached_tcp(const endpoint& ep);
optional<endpoint_data&> cached_udp(const endpoint& ep);
optional<std::vector<response_promise>&> pending(const endpoint& ep); optional<std::vector<response_promise>&> pending(const endpoint& ep);
actor broker_; actor broker_;
std::map<endpoint, endpoint_data> cached_; std::map<endpoint, endpoint_data> cached_tcp_;
std::map<endpoint, endpoint_data> cached_udp_;
std::map<endpoint, std::vector<response_promise>> pending_; std::map<endpoint, std::vector<response_promise>> pending_;
}; };
......
...@@ -37,6 +37,9 @@ public: ...@@ -37,6 +37,9 @@ public:
/// @returns `true` if the manager accepts further connections, /// @returns `true` if the manager accepts further connections,
/// otherwise `false`. /// otherwise `false`.
virtual bool new_connection() = 0; virtual bool new_connection() = 0;
/// Get the port of the underlying I/O device.
virtual uint16_t port() const = 0;
}; };
} // namespace network } // namespace network
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_NETWORK_DATAGRAM_MANAGER_HPP
#define CAF_IO_NETWORK_DATAGRAM_MANAGER_HPP
#include "caf/io/datagram_handle.hpp"
#include "caf/io/network/manager.hpp"
#include "caf/io/network/receive_buffer.hpp"
namespace caf {
namespace io {
namespace network {
/// A datagram manager provides callbacks for outgoing
/// datagrams as well as for error handling.
class datagram_manager : public manager {
public:
~datagram_manager() override;
/// Called by the underlying I/O device whenever it received data.
/// @returns `true` if the manager accepts further reads, otherwise `false`.
virtual bool consume(execution_unit*, datagram_handle hdl,
receive_buffer& buf) = 0;
/// Called by the underlying I/O device whenever it sent data.
virtual void datagram_sent(execution_unit*, datagram_handle hdl, size_t,
std::vector<char> buffer) = 0;
/// Called by the underlying I/O device to indicate that a new remote
/// endpoint has been detected, passing in the received datagram.
/// @returns `true` if the manager accepts further enpoints,
/// otherwise `false`.
virtual bool new_endpoint(receive_buffer& buf) = 0;
/// Get the port of the underlying I/O device.
virtual uint16_t port(datagram_handle) const = 0;
};
} // namespace network
} // namespace io
} // namespace caf
#endif // CAF_IO_NETWORK_DATAGRAM_MANAGER_HPP
...@@ -30,6 +30,7 @@ ...@@ -30,6 +30,7 @@
#include "caf/optional.hpp" #include "caf/optional.hpp"
#include "caf/io/network/protocol.hpp" #include "caf/io/network/protocol.hpp"
#include "caf/io/network/ip_endpoint.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
...@@ -78,6 +79,11 @@ public: ...@@ -78,6 +79,11 @@ public:
static std::vector<std::pair<std::string, protocol::network>> static std::vector<std::pair<std::string, protocol::network>>
server_address(uint16_t port, const char* host, server_address(uint16_t port, const char* host,
optional<protocol::network> preferred = none); optional<protocol::network> preferred = none);
/// Writes datagram endpoint info for `host`:`port` into ep
static bool
get_endpoint(const std::string& host, uint16_t port, ip_endpoint& ep,
optional<protocol::network> preferred = none);
}; };
} // namespace network } // namespace network
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_IP_ENDPOINT_HPP
#define CAF_IO_IP_ENDPOINT_HPP
#include <deque>
#include <vector>
#include <string>
#include <functional>
#include "caf/error.hpp"
#include "caf/meta/type_name.hpp"
#include "caf/meta/save_callback.hpp"
#include "caf/meta/load_callback.hpp"
struct sockaddr;
struct sockaddr_storage;
struct sockaddr_in;
struct sockaddr_in6;
namespace caf {
namespace io {
namespace network {
// hash for char*, see:
// - https://en.wikipedia.org/wiki/Fowler%E2%80%93Noll%E2%80%93Vo_hash_function
// - http://www.isthe.com/chongo/tech/comp/fnv/index.html
// Always hash 128 bit address, for v4 we use the embedded addr.
class ep_hash {
public:
ep_hash();
size_t operator()(const sockaddr& sa) const noexcept;
size_t hash(const sockaddr_in* sa) const noexcept;
size_t hash(const sockaddr_in6* sa) const noexcept;
};
/// A hashable wrapper for a sockaddr storage.
struct ip_endpoint {
public:
/// Default constructor for sockaddr storage which reserves memory for the
/// internal data structure on creation.
ip_endpoint();
/// Move constructor.
ip_endpoint(ip_endpoint&&) = default;
/// Copy constructor.
ip_endpoint(const ip_endpoint&);
/// Destructor
~ip_endpoint() = default;
/// Copy assignment operator.
ip_endpoint& operator=(const ip_endpoint&);
/// Move assignment operator.
ip_endpoint& operator=(ip_endpoint&&) = default;
/// Returns a pointer to the internal address storage.
sockaddr* address();
/// Returns a constant pointer to the internal address storage.
const sockaddr* caddress() const;
/// Returns the length of the stored address.
size_t* length();
/// Returns the length of the stored address.
const size_t* clength() const;
/// Null internal storage and length.
void clear();
private:
struct impl;
struct impl_deleter { void operator()(impl*) const; };
std::unique_ptr<impl,impl_deleter> ptr_;
};
bool operator==(const ip_endpoint& lhs, const ip_endpoint& rhs);
std::string to_string(const ip_endpoint& ep);
std::string host(const ip_endpoint& ep);
uint16_t port(const ip_endpoint& ep);
uint32_t family(const ip_endpoint& ep);
error load_endpoint(ip_endpoint& ep, uint32_t& f, std::string& h,
uint16_t& p, size_t& l);
error save_endpoint(ip_endpoint& ep, uint32_t& f, std::string& h,
uint16_t& p, size_t& l);
template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun, ip_endpoint& ep) {
uint32_t f;
std::string h;
uint16_t p;
size_t l;
if (*ep.length() > 0) {
f = family(ep);
h = host(ep);
p = port(ep);
l = *ep.length();
}
auto load = [&] { return load_endpoint(ep, f, h, p, l); };
auto save = [&] { return save_endpoint(ep, f, h, p, l); };
return fun(meta::type_name("ip_endpoint"), f, h, p, l,
meta::load_callback(load), meta::save_callback(save));
}
} // namespace network
} // namespace io
} // namespace caf
namespace std {
template <>
struct hash<caf::io::network::ip_endpoint> {
using argument_type = caf::io::network::ip_endpoint;
using result_type = size_t;
result_type operator()(const argument_type& ep) const {
auto ptr = ep.caddress();
return caf::io::network::ep_hash{}(*ptr);
}
};
} // namespace std
#endif // CAF_IO_IP_ENDPOINT_HPP
...@@ -71,9 +71,6 @@ public: ...@@ -71,9 +71,6 @@ public:
/// Get the address of the underlying I/O device. /// Get the address of the underlying I/O device.
virtual std::string addr() const = 0; virtual std::string addr() const = 0;
/// Get the port of the underlying I/O device.
virtual uint16_t port() const = 0;
protected: protected:
/// Creates a message signalizing a disconnect to the parent. /// Creates a message signalizing a disconnect to the parent.
virtual message detach_message() = 0; virtual message detach_message() = 0;
......
...@@ -35,6 +35,7 @@ ...@@ -35,6 +35,7 @@
#include "caf/io/connection_handle.hpp" #include "caf/io/connection_handle.hpp"
#include "caf/io/network/protocol.hpp" #include "caf/io/network/protocol.hpp"
#include "caf/io/network/ip_endpoint.hpp"
#include "caf/io/network/native_socket.hpp" #include "caf/io/network/native_socket.hpp"
namespace caf { namespace caf {
...@@ -69,6 +70,26 @@ public: ...@@ -69,6 +70,26 @@ public:
const char* in = nullptr, const char* in = nullptr,
bool reuse_addr = false) = 0; bool reuse_addr = false) = 0;
/// Creates a new `datagram_servant` from a native socket handle.
/// @threadsafe
virtual datagram_servant_ptr new_datagram_servant(native_socket fd) = 0;
virtual datagram_servant_ptr
new_datagram_servant_for_endpoint(native_socket fd, const ip_endpoint& ep) = 0;
/// Create a new `datagram_servant` to contact a remote endpoint `host` and
/// `port`.
/// @warning Do not call from outside the multiplexer's event loop.
virtual expected<datagram_servant_ptr>
new_remote_udp_endpoint(const std::string& host, uint16_t port) = 0;
/// Create a new `datagram_servant` that receives datagrams on the local
/// `port`, optionally only accepting connections from IP address `in`.
/// @warning Do not call from outside the multiplexer's event loop.
virtual expected<datagram_servant_ptr>
new_local_udp_endpoint(uint16_t port, const char* in = nullptr,
bool reuse_addr = false) = 0;
/// Simple wrapper for runnables /// Simple wrapper for runnables
class runnable : public resumable, public ref_counted { class runnable : public resumable, public ref_counted {
public: public:
......
...@@ -20,6 +20,8 @@ ...@@ -20,6 +20,8 @@
#ifndef CAF_IO_NETWORK_OPERATION_HPP #ifndef CAF_IO_NETWORK_OPERATION_HPP
#define CAF_IO_NETWORK_OPERATION_HPP #define CAF_IO_NETWORK_OPERATION_HPP
#include <string>
namespace caf { namespace caf {
namespace io { namespace io {
namespace network { namespace network {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_NETWORK_RECEIVE_BUFFER_HPP
#define CAF_IO_NETWORK_RECEIVE_BUFFER_HPP
#include <memory>
#include <cstddef>
#include "caf/allowed_unsafe_message_type.hpp"
namespace caf {
namespace io {
namespace network {
/// A container that does not call constructors and destructors for its values.
class receive_buffer {
public:
using value_type = char;
using size_type = size_t;
using difference_type = std::ptrdiff_t;
using reference = value_type&;
using const_reference = const value_type&;
using pointer = value_type*;
using const_pointer = std::pointer_traits<pointer>::rebind<const value_type>;
using iterator = pointer;
using const_iterator = const_pointer;
using reverse_iterator = std::reverse_iterator<iterator>;
using const_reverse_iterator = std::reverse_iterator<const_iterator>;
using buffer_ptr = std::unique_ptr<value_type[],
std::default_delete<value_type[]>>;
/// Create an empty container.
receive_buffer();
/// Create an empty container with `size` storage. Data in the storage is
/// not initialized.
receive_buffer(size_t size);
/// Move constructor.
receive_buffer(receive_buffer&& other) noexcept;
/// Copy constructor.
receive_buffer(const receive_buffer& other);
/// Move assignment operator.
receive_buffer& operator=(receive_buffer&& other) noexcept;
/// Copy assignment operator.
receive_buffer& operator=(const receive_buffer& other);
/// Returns a pointer to the underlying buffer.
inline pointer data() noexcept {
return buffer_.get();
}
/// Returns a const pointer to the data.
inline const_pointer data() const noexcept {
return buffer_.get();
}
/// Returns the number of stored elements.
size_type size() const noexcept {
return size_;
}
/// Returns the number of elements that the container has allocated space for.
size_type capacity() const noexcept {
return capacity_;
}
/// Returns the maximum possible number of elements the container
/// could theoretically hold.
size_type max_size() const noexcept {
return std::numeric_limits<size_t>::max();
}
/// Resize the container to `new_size`. While this may increase its storage,
/// no storage will be released.
void resize(size_type new_size);
/// Set the size of the storage to `new_size`. If `new_size` is smaller than
/// the current capacity nothing happens. If `new_size` is larger than the
/// current capacity all iterators are invalidated.
void reserve(size_type new_size);
/// Shrink the container to its current size.
void shrink_to_fit();
/// Check if the container is empty.
inline bool empty() const noexcept {
return size_ == 0;
}
/// Clears the content of the container and releases the allocated storage.
void clear();
/// Swap contents with `other` receive buffer.
void swap(receive_buffer& other) noexcept;
/// Returns an iterator to the beginning.
inline iterator begin() noexcept {
return buffer_.get();
}
/// Returns an iterator to the end.
inline iterator end() noexcept {
return buffer_.get() + size_;
}
/// Returns an iterator to the beginning.
inline const_iterator begin() const noexcept {
return buffer_.get();
}
/// Returns an iterator to the end.
inline const_iterator end() const noexcept {
return buffer_.get() + size_;
}
/// Returns an iterator to the beginning.
inline const_iterator cbegin() const noexcept {
return buffer_.get();
}
/// Returns an iterator to the end.
inline const_iterator cend() const noexcept {
return buffer_.get() + size_;
}
/// Returns an iterator to the reverse beginning.
inline reverse_iterator rbegin() noexcept {
return reverse_iterator{buffer_.get() + size_};
}
/// Returns an iterator to the reverse end of the data.
inline reverse_iterator rend() noexcept {
return reverse_iterator{buffer_.get()};
}
/// Returns an iterator to the reverse beginning.
inline const_reverse_iterator rbegin() const noexcept {
return const_reverse_iterator{buffer_.get() + size_};
}
/// Returns an iterator to the reverse end of the data.
inline const_reverse_iterator rend() const noexcept {
return const_reverse_iterator{buffer_.get()};
}
/// Returns an iterator to the reverse beginning.
inline const_reverse_iterator crbegin() const noexcept {
return const_reverse_iterator{buffer_.get() + size_};
}
/// Returns an iterator to the reverse end of the data.
inline const_reverse_iterator crend() const noexcept {
return const_reverse_iterator{buffer_.get()};
}
/// Insert `value` before `pos`.
iterator insert(iterator pos, value_type value);
/// Append `value`.
void push_back(value_type value);
private:
// Increse the buffer capacity, maintaining its data. May invalidate iterators.
void increase_by(size_t bytes);
// Reduce the buffer capacity, maintaining its data. May invalidate iterators.
void shrink_by(size_t bytes);
buffer_ptr buffer_;
size_type capacity_;
size_type size_;
};
} // namepsace network
} // namespace io
} // namespace caf
#endif // CAF_IO_NETWORK_RECEIVE_BUFFER_HPP
...@@ -41,6 +41,9 @@ public: ...@@ -41,6 +41,9 @@ public:
/// Called by the underlying I/O device whenever it sent data. /// Called by the underlying I/O device whenever it sent data.
virtual void data_transferred(execution_unit* ctx, size_t num_bytes, virtual void data_transferred(execution_unit* ctx, size_t num_bytes,
size_t remaining_bytes) = 0; size_t remaining_bytes) = 0;
/// Get the port of the underlying I/O device.
virtual uint16_t port() const = 0;
}; };
} // namespace network } // namespace network
......
...@@ -26,12 +26,16 @@ ...@@ -26,12 +26,16 @@
#include "caf/io/abstract_broker.hpp" #include "caf/io/abstract_broker.hpp"
#include "caf/io/network/multiplexer.hpp" #include "caf/io/network/multiplexer.hpp"
#include "caf/io/network/receive_buffer.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
namespace network { namespace network {
class test_multiplexer : public multiplexer { class test_multiplexer : public multiplexer {
private:
struct datagram_data;
public: public:
explicit test_multiplexer(actor_system* sys); explicit test_multiplexer(actor_system* sys);
...@@ -47,6 +51,19 @@ public: ...@@ -47,6 +51,19 @@ public:
expected<doorman_ptr> new_tcp_doorman(uint16_t prt, const char* in, expected<doorman_ptr> new_tcp_doorman(uint16_t prt, const char* in,
bool reuse_addr) override; bool reuse_addr) override;
datagram_servant_ptr new_datagram_servant(native_socket fd) override;
datagram_servant_ptr
new_datagram_servant_for_endpoint(native_socket fd,
const ip_endpoint& ep) override;
expected<datagram_servant_ptr>
new_remote_udp_endpoint(const std::string& host, uint16_t port) override;
expected<datagram_servant_ptr>
new_local_udp_endpoint(uint16_t port, const char* in = nullptr,
bool reuse_addr = false) override;
/// Checks whether `x` is assigned to any known doorman or is user-provided /// Checks whether `x` is assigned to any known doorman or is user-provided
/// for future assignment. /// for future assignment.
bool is_known_port(uint16_t x) const; bool is_known_port(uint16_t x) const;
...@@ -55,6 +72,8 @@ public: ...@@ -55,6 +72,8 @@ public:
/// for future assignment. /// for future assignment.
bool is_known_handle(accept_handle x) const; bool is_known_handle(accept_handle x) const;
bool is_known_handle(datagram_handle x) const;
supervisor_ptr make_supervisor() override; supervisor_ptr make_supervisor() override;
bool try_run_once() override; bool try_run_once() override;
...@@ -67,43 +86,101 @@ public: ...@@ -67,43 +86,101 @@ public:
doorman_ptr new_doorman(accept_handle, uint16_t port); doorman_ptr new_doorman(accept_handle, uint16_t port);
public:
datagram_servant_ptr new_datagram_servant(datagram_handle, uint16_t port);
datagram_servant_ptr new_datagram_servant(datagram_handle,
const std::string& host,
uint16_t port);
void provide_scribe(std::string host, uint16_t desired_port, void provide_scribe(std::string host, uint16_t desired_port,
connection_handle hdl); connection_handle hdl);
void provide_acceptor(uint16_t desired_port, accept_handle hdl); void provide_acceptor(uint16_t desired_port, accept_handle hdl);
/// A buffer storing bytes. void provide_datagram_servant(uint16_t desired_port, datagram_handle hdl);
void provide_datagram_servant(std::string host, uint16_t desired_port,
datagram_handle hdl);
/// Generate an id for a new servant.
int64_t next_endpoint_id();
/// A buffer storing bytes used for TCP related components.
using buffer_type = std::vector<char>; using buffer_type = std::vector<char>;
/// Buffers storing bytes for UDP related components.
using read_buffer_type = network::receive_buffer;
using write_buffer_type = buffer_type;
using read_job_type = std::pair<datagram_handle, read_buffer_type>;
using write_job_type = std::pair<datagram_handle, write_buffer_type>;
using write_job_queue_type = std::deque<write_job_type>;
using shared_buffer_type = std::shared_ptr<buffer_type>; using shared_buffer_type = std::shared_ptr<buffer_type>;
using shared_job_queue_type = std::shared_ptr<write_job_queue_type>;
/// Models pending data on the network, i.e., the network /// Models pending data on the network, i.e., the network
/// input buffer usually managed by the operating system. /// input buffer usually managed by the operating system.
buffer_type& virtual_network_buffer(connection_handle hdl); buffer_type& virtual_network_buffer(connection_handle hdl);
/// Models pending data on the network, i.e., the network
/// input buffer usually managed by the operating system.
write_job_queue_type& virtual_network_buffer(datagram_handle hdl);
/// Returns the output buffer of the scribe identified by `hdl`. /// Returns the output buffer of the scribe identified by `hdl`.
buffer_type& output_buffer(connection_handle hdl); buffer_type& output_buffer(connection_handle hdl);
/// Returns the input buffer of the scribe identified by `hdl`. /// Returns the input buffer of the scribe identified by `hdl`.
buffer_type& input_buffer(connection_handle hdl); buffer_type& input_buffer(connection_handle hdl);
/// Returns the output buffer of the dgram servant identified by `hdl`.
write_job_type& output_buffer(datagram_handle hdl);
/// Returns the queue with all outgoing datagrams for the dgram servant
/// identified by `hdl`.
write_job_queue_type& output_queue(datagram_handle hdl);
/// Returns the input buffer of the dgram servant identified by `hdl`.
read_job_type& input_buffer(datagram_handle hdl);
/// Returns the configured read policy of the scribe identified by `hdl`. /// Returns the configured read policy of the scribe identified by `hdl`.
receive_policy::config& read_config(connection_handle hdl); receive_policy::config& read_config(connection_handle hdl);
/// Returns whether the scribe identified by `hdl` receives write ACKs. /// Returns whether the scribe identified by `hdl` receives write ACKs.
bool& ack_writes(connection_handle hdl); bool& ack_writes(connection_handle hdl);
/// Returns whether the dgram servant identified by `hdl` receives write ACKs.
bool& ack_writes(datagram_handle hdl);
/// Returns `true` if this handle has been closed /// Returns `true` if this handle has been closed
/// for reading, `false` otherwise. /// for reading, `false` otherwise.
bool& stopped_reading(connection_handle hdl); bool& stopped_reading(connection_handle hdl);
/// Returns `true` if this handle has been closed
/// for reading, `false` otherwise.
bool& stopped_reading(datagram_handle hdl);
/// Returns `true` if this handle is inactive, otherwise `false`. /// Returns `true` if this handle is inactive, otherwise `false`.
bool& passive_mode(connection_handle hdl); bool& passive_mode(connection_handle hdl);
/// Returns `true` if this handle is inactive, otherwise `false`.
bool& passive_mode(datagram_handle hdl);
scribe_ptr& impl_ptr(connection_handle hdl); scribe_ptr& impl_ptr(connection_handle hdl);
uint16_t& port(accept_handle hdl); uint16_t& port(accept_handle hdl);
uint16_t& port(datagram_handle hdl);
uint16_t& local_port(datagram_handle hdl);
size_t& datagram_size(datagram_handle hdl);
datagram_servant_ptr& impl_ptr(datagram_handle hdl);
/// Returns a map with all servants related to the servant `hdl`.
std::set<datagram_handle>& servants(datagram_handle hdl);
/// Returns `true` if this handle has been closed /// Returns `true` if this handle has been closed
/// for reading, `false` otherwise. /// for reading, `false` otherwise.
bool& stopped_reading(accept_handle hdl); bool& stopped_reading(accept_handle hdl);
...@@ -123,18 +200,33 @@ public: ...@@ -123,18 +200,33 @@ public:
test_multiplexer& peer, std::string host, test_multiplexer& peer, std::string host,
uint16_t port, connection_handle peer_hdl); uint16_t port, connection_handle peer_hdl);
/// Stores `hdl` as a pending endpoint for `src`.
void add_pending_endpoint(datagram_handle src, datagram_handle hdl);
using pending_connects_map = std::unordered_multimap<accept_handle, using pending_connects_map = std::unordered_multimap<accept_handle,
connection_handle>; connection_handle>;
pending_connects_map& pending_connects(); pending_connects_map& pending_connects();
using pending_endpoints_map = std::unordered_map<int64_t, datagram_handle>;
pending_endpoints_map& pending_endpoints();
using pending_scribes_map = std::map<std::pair<std::string, uint16_t>, using pending_scribes_map = std::map<std::pair<std::string, uint16_t>,
connection_handle>; connection_handle>;
using pending_doorman_map = std::unordered_map<uint16_t, accept_handle>; using pending_doorman_map = std::unordered_map<uint16_t, accept_handle>;
using pending_local_datagram_endpoints_map = std::map<uint16_t,
datagram_handle>;
using pending_remote_datagram_endpoints_map
= std::map<std::pair<std::string, uint16_t>, datagram_handle>;
bool has_pending_scribe(std::string x, uint16_t y); bool has_pending_scribe(std::string x, uint16_t y);
bool has_pending_remote_endpoint(std::string x, uint16_t y);
/// Accepts a pending connect on `hdl`. /// Accepts a pending connect on `hdl`.
void accept_connection(accept_handle hdl); void accept_connection(accept_handle hdl);
...@@ -154,10 +246,18 @@ public: ...@@ -154,10 +246,18 @@ public:
/// the configured read policy no longer allows receiving. /// the configured read policy no longer allows receiving.
bool read_data(connection_handle hdl); bool read_data(connection_handle hdl);
/// Reads the next datagram from the external input buffer.
bool read_data(datagram_handle hdl);
/// Appends `buf` to the virtual network buffer of `hdl` /// Appends `buf` to the virtual network buffer of `hdl`
/// and calls `read_data(hdl)` afterwards. /// and calls `read_data(hdl)` afterwards.
void virtual_send(connection_handle hdl, const buffer_type& buf); void virtual_send(connection_handle hdl, const buffer_type& buf);
/// Appends `buf` to the virtual network buffer of `hdl`
/// and calls `read_data(hdl)` afterwards.
void virtual_send(datagram_handle src, datagram_handle ep,
const buffer_type&);
/// Waits until a `runnable` is available and executes it. /// Waits until a `runnable` is available and executes it.
void exec_runnable(); void exec_runnable();
...@@ -197,6 +297,8 @@ private: ...@@ -197,6 +297,8 @@ private:
using guard_type = std::unique_lock<std::mutex>; using guard_type = std::unique_lock<std::mutex>;
std::shared_ptr<datagram_data> data_for_hdl(datagram_handle hdl);
struct scribe_data { struct scribe_data {
shared_buffer_type vn_buf_ptr; shared_buffer_type vn_buf_ptr;
shared_buffer_type wr_buf_ptr; shared_buffer_type wr_buf_ptr;
...@@ -223,19 +325,49 @@ private: ...@@ -223,19 +325,49 @@ private:
doorman_data(); doorman_data();
}; };
struct datagram_data {
shared_job_queue_type vn_buf_ptr;
shared_job_queue_type wr_buf_ptr;
write_job_queue_type& vn_buf;
write_job_queue_type& wr_buf;
read_job_type rd_buf;
datagram_servant_ptr ptr;
bool stopped_reading;
bool passive_mode;
bool ack_writes;
uint16_t port;
uint16_t local_port;
std::set<datagram_handle> servants;
size_t datagram_size;
// Allows creating an entangled scribes where the input of this scribe is
// the output of another scribe and vice versa.
datagram_data(
shared_job_queue_type input = std::make_shared<write_job_queue_type>(),
shared_job_queue_type output = std::make_shared<write_job_queue_type>()
);
};
using scribe_data_map = std::unordered_map<connection_handle, scribe_data>; using scribe_data_map = std::unordered_map<connection_handle, scribe_data>;
using doorman_data_map = std::unordered_map<accept_handle, doorman_data>; using doorman_data_map = std::unordered_map<accept_handle, doorman_data>;
using datagram_data_map = std::unordered_map<datagram_handle,
std::shared_ptr<datagram_data>>;
// guards resumables_ and scribes_ // guards resumables_ and scribes_
std::mutex mx_; std::mutex mx_;
std::condition_variable cv_; std::condition_variable cv_;
std::list<resumable_ptr> resumables_; std::list<resumable_ptr> resumables_;
pending_scribes_map scribes_; pending_scribes_map scribes_;
std::unordered_map<uint16_t, accept_handle> doormen_; pending_doorman_map doormen_;
scribe_data_map scribe_data_; scribe_data_map scribe_data_;
doorman_data_map doorman_data_; doorman_data_map doorman_data_;
pending_local_datagram_endpoints_map local_endpoints_;
pending_remote_datagram_endpoints_map remote_endpoints_;
pending_connects_map pending_connects_; pending_connects_map pending_connects_;
pending_endpoints_map pending_endpoints_;
datagram_data_map datagram_data_;
// extra state for making sure the test multiplexer is not used in a // extra state for making sure the test multiplexer is not used in a
// multithreaded setup // multithreaded setup
...@@ -246,6 +378,8 @@ private: ...@@ -246,6 +378,8 @@ private:
// Configures a one-shot handler for the next inlined runnable. // Configures a one-shot handler for the next inlined runnable.
std::function<void()> inline_runnable_callback_; std::function<void()> inline_runnable_callback_;
int64_t servant_ids_;
}; };
} // namespace network } // namespace network
......
...@@ -30,7 +30,9 @@ ...@@ -30,7 +30,9 @@
#include "caf/io/handle.hpp" #include "caf/io/handle.hpp"
#include "caf/io/accept_handle.hpp" #include "caf/io/accept_handle.hpp"
#include "caf/io/datagram_handle.hpp"
#include "caf/io/connection_handle.hpp" #include "caf/io/connection_handle.hpp"
#include "caf/io/network/receive_buffer.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
...@@ -128,6 +130,60 @@ inspect(Inspector& f, acceptor_passivated_msg& x) { ...@@ -128,6 +130,60 @@ inspect(Inspector& f, acceptor_passivated_msg& x) {
return f(meta::type_name("acceptor_passivated_msg"), x.handle); return f(meta::type_name("acceptor_passivated_msg"), x.handle);
} }
/// Signalizes that a datagram with a certain size has been sent.
struct new_datagram_msg {
// Handle to the endpoint used.
datagram_handle handle;
// Buffer containing received data.
network::receive_buffer buf;
};
/// @relates new_datagram_msg
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, new_datagram_msg& x) {
return f(meta::type_name("new_datagram_msg"), x.handle, x.buf);
}
/// Signalizes that a datagram with a certain size has been sent.
struct datagram_sent_msg {
// Handle to the endpoint used.
datagram_handle handle;
// Number of bytes written.
uint64_t written;
// Buffer of the sent datagram, for reuse.
std::vector<char> buf;
};
/// @relates datagram_sent_msg
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, datagram_sent_msg& x) {
return f(meta::type_name("datagram_sent_msg"), x.handle, x.written, x.buf);
}
/// Signalizes that a datagram sink has entered passive mode.
struct datagram_servant_passivated_msg {
datagram_handle handle;
};
/// @relates datagram_servant_passivated_msg
template <class Inspector>
typename Inspector::result_type
inspect(Inspector& f, datagram_servant_passivated_msg& x) {
return f(meta::type_name("datagram_servant_passivated_msg"), x.handle);
}
/// Signalizes that a datagram endpoint has entered passive mode.
struct datagram_servant_closed_msg {
std::vector<datagram_handle> handles;
};
/// @relates datagram_servant_closed_msg
template <class Inspector>
typename Inspector::result_type
inspect(Inspector& f, datagram_servant_closed_msg& x) {
return f(meta::type_name("datagram_servant_closed_msg"), x.handles);
}
} // namespace io } // namespace io
} // namespace caf } // namespace caf
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_VISITORS_HPP
#define CAF_IO_VISITORS_HPP
#include "caf/io/abstract_broker.hpp"
namespace caf {
namespace io {
struct addr_visitor {
using result_type = std::string;
addr_visitor(abstract_broker* ptr) : ptr_(ptr) { }
template <class Handle>
result_type operator()(const Handle& hdl) { return ptr_->remote_addr(hdl); }
abstract_broker* ptr_;
};
struct port_visitor {
using result_type = uint16_t;
port_visitor(abstract_broker* ptr) : ptr_(ptr) { }
template <class Handle>
result_type operator()(const Handle& hdl) { return ptr_->remote_port(hdl); }
abstract_broker* ptr_;
};
struct id_visitor {
using result_type = int64_t;
template <class Handle>
result_type operator()(const Handle& hdl) { return hdl.id(); }
};
struct hash_visitor {
using result_type = size_t;
template <class Handle>
result_type operator()(const Handle& hdl) const {
std::hash<Handle> f;
return f(hdl);
}
};
} // namespace io
} // namespace caf
#endif // CAF_IO_VISITORS_HPP
...@@ -64,6 +64,7 @@ bool abstract_broker::cleanup(error&& reason, execution_unit* host) { ...@@ -64,6 +64,7 @@ bool abstract_broker::cleanup(error&& reason, execution_unit* host) {
close_all(); close_all();
CAF_ASSERT(doormen_.empty()); CAF_ASSERT(doormen_.empty());
CAF_ASSERT(scribes_.empty()); CAF_ASSERT(scribes_.empty());
CAF_ASSERT(datagram_servants_.empty());
cache_.clear(); cache_.clear();
return local_actor::cleanup(std::move(reason), host); return local_actor::cleanup(std::move(reason), host);
} }
...@@ -109,6 +110,46 @@ void abstract_broker::flush(connection_handle hdl) { ...@@ -109,6 +110,46 @@ void abstract_broker::flush(connection_handle hdl) {
x->flush(); x->flush();
} }
void abstract_broker::ack_writes(datagram_handle hdl, bool enable) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(enable));
auto x = by_id(hdl);
if (x)
x->ack_writes(enable);
}
std::vector<char>& abstract_broker::wr_buf(datagram_handle hdl) {
auto x = by_id(hdl);
if (!x) {
CAF_LOG_ERROR("tried to access wr_buf() of an unknown"
"datagram_handle");
return dummy_wr_buf_;
}
return x->wr_buf(hdl);
}
void abstract_broker::enqueue_datagram(datagram_handle hdl,
std::vector<char> buf) {
auto x = by_id(hdl);
if (!x)
CAF_LOG_ERROR("tried to access datagram_buffer() of an unknown"
"datagram_handle");
x->enqueue_datagram(hdl, std::move(buf));
}
void abstract_broker::write(datagram_handle hdl, size_t bs,
const void* buf) {
auto& out = wr_buf(hdl);
auto first = reinterpret_cast<const char*>(buf);
auto last = first + bs;
out.insert(out.end(), first, last);
}
void abstract_broker::flush(datagram_handle hdl) {
auto x = by_id(hdl);
if (x)
x->flush();
}
std::vector<connection_handle> abstract_broker::connections() const { std::vector<connection_handle> abstract_broker::connections() const {
std::vector<connection_handle> result; std::vector<connection_handle> result;
result.reserve(scribes_.size()); result.reserve(scribes_.size());
...@@ -164,6 +205,83 @@ abstract_broker::add_tcp_doorman(uint16_t port, const char* in, ...@@ -164,6 +205,83 @@ abstract_broker::add_tcp_doorman(uint16_t port, const char* in,
return std::move(eptr.error()); return std::move(eptr.error());
} }
void abstract_broker::add_datagram_servant(datagram_servant_ptr ptr) {
CAF_LOG_TRACE(CAF_ARG(ptr));
CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(ptr->parent() == nullptr);
ptr->set_parent(this);
auto hdls = ptr->hdls();
launch_servant(ptr);
for (auto& hdl : hdls)
add_hdl_for_datagram_servant(ptr, hdl);
}
void abstract_broker::add_hdl_for_datagram_servant(datagram_servant_ptr ptr,
datagram_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(hdl));
CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(ptr->parent() == this);
get_map(hdl).emplace(hdl, std::move(ptr));
}
datagram_handle abstract_broker::add_datagram_servant(network::native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
auto ptr = backend().new_datagram_servant(fd);
auto hdl = ptr->hdl();
add_datagram_servant(std::move(ptr));
return hdl;
}
datagram_handle
abstract_broker::add_datagram_servant_for_endpoint(network::native_socket fd,
const network::ip_endpoint& ep) {
CAF_LOG_TRACE(CAF_ARG(fd));
auto ptr = backend().new_datagram_servant_for_endpoint(fd, ep);
auto hdl = ptr->hdl();
add_datagram_servant(std::move(ptr));
return hdl;
}
expected<datagram_handle>
abstract_broker::add_udp_datagram_servant(const std::string& host,
uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(host) << CAF_ARG(port));
auto eptr = backend().new_remote_udp_endpoint(host, port);
if (eptr) {
auto ptr = std::move(*eptr);
auto hdl = ptr->hdl();
add_datagram_servant(std::move(ptr));
return hdl;
}
return std::move(eptr.error());
}
expected<std::pair<datagram_handle, uint16_t>>
abstract_broker::add_udp_datagram_servant(uint16_t port, const char* in,
bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(in) << CAF_ARG(reuse_addr));
auto eptr = backend().new_local_udp_endpoint(port, in, reuse_addr);
if (eptr) {
auto ptr = std::move(*eptr);
auto p = ptr->local_port();
auto hdl = ptr->hdl();
add_datagram_servant(std::move(ptr));
return std::make_pair(hdl, p);
}
return std::move(eptr.error());
}
void abstract_broker::move_datagram_servant(datagram_servant_ptr ptr) {
CAF_LOG_TRACE(CAF_ARG(ptr));
CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(ptr->parent() != nullptr && ptr->parent() != this);
ptr->set_parent(this);
CAF_ASSERT(ptr->parent() == this);
auto hdls = ptr->hdls();
for (auto& hdl : hdls)
add_hdl_for_datagram_servant(ptr, hdl);
}
std::string abstract_broker::remote_addr(connection_handle hdl) { std::string abstract_broker::remote_addr(connection_handle hdl) {
auto i = scribes_.find(hdl); auto i = scribes_.find(hdl);
return i != scribes_.end() ? i->second->addr() : std::string{}; return i != scribes_.end() ? i->second->addr() : std::string{};
...@@ -191,6 +309,36 @@ accept_handle abstract_broker::hdl_by_port(uint16_t port) { ...@@ -191,6 +309,36 @@ accept_handle abstract_broker::hdl_by_port(uint16_t port) {
return invalid_accept_handle; return invalid_accept_handle;
} }
datagram_handle abstract_broker::datagram_hdl_by_port(uint16_t port) {
for (auto& kvp : datagram_servants_)
if (kvp.second->port(kvp.first) == port)
return kvp.first;
return invalid_datagram_handle;
}
std::string abstract_broker::remote_addr(datagram_handle hdl) {
auto i = datagram_servants_.find(hdl);
return i != datagram_servants_.end() ? i->second->addr() : std::string{};
}
uint16_t abstract_broker::remote_port(datagram_handle hdl) {
auto i = datagram_servants_.find(hdl);
return i != datagram_servants_.end() ? i->second->port(hdl) : 0;
}
uint16_t abstract_broker::local_port(datagram_handle hdl) {
auto i = datagram_servants_.find(hdl);
return i != datagram_servants_.end() ? i->second->local_port() : 0;
}
bool abstract_broker::remove_endpoint(datagram_handle hdl) {
auto x = by_id(hdl);
if (!x)
return false;
x->remove_endpoint(hdl);
return true;
}
void abstract_broker::close_all() { void abstract_broker::close_all() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
while (!doormen_.empty()) { while (!doormen_.empty()) {
...@@ -201,6 +349,10 @@ void abstract_broker::close_all() { ...@@ -201,6 +349,10 @@ void abstract_broker::close_all() {
// stop_reading will remove the scribe from scribes_ // stop_reading will remove the scribe from scribes_
scribes_.begin()->second->stop_reading(); scribes_.begin()->second->stop_reading();
} }
while (!datagram_servants_.empty()) {
// stop reading will remove dgram servants from datagram_servants_
datagram_servants_.begin()->second->stop_reading();
}
} }
resumable::subtype_t abstract_broker::subtype() const { resumable::subtype_t abstract_broker::subtype() const {
...@@ -225,7 +377,6 @@ void abstract_broker::init_broker() { ...@@ -225,7 +377,6 @@ void abstract_broker::init_broker() {
// might call functions like add_connection // might call functions like add_connection
for (auto& kvp : doormen_) for (auto& kvp : doormen_)
kvp.second->launch(); kvp.second->launch();
} }
abstract_broker::abstract_broker(actor_config& cfg) : scheduled_actor(cfg) { abstract_broker::abstract_broker(actor_config& cfg) : scheduled_actor(cfg) {
...@@ -243,5 +394,10 @@ void abstract_broker::launch_servant(doorman_ptr& ptr) { ...@@ -243,5 +394,10 @@ void abstract_broker::launch_servant(doorman_ptr& ptr) {
ptr->launch(); ptr->launch();
} }
void abstract_broker::launch_servant(datagram_servant_ptr& ptr) {
if (getf(is_initialized_flag))
ptr->launch();
}
} // namespace io } // namespace io
} // namespace caf } // namespace caf
This diff is collapsed.
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include <chrono>
#include "caf/io/basp/instance.hpp"
#include "caf/io/connection_helper.hpp"
namespace caf {
namespace io {
namespace {
auto autoconnect_timeout = std::chrono::minutes(10);
} // namespace <anonymous>
const char* connection_helper_state::name = "connection_helper";
behavior datagram_connection_broker(broker* self, uint16_t port,
network::address_listing addresses,
actor system_broker) {
auto& mx = self->system().middleman().backend();
auto& this_node = self->system().node();
auto& app_id = self->system().config().middleman_app_identifier;
for (auto& kvp : addresses) {
for (auto& addr : kvp.second) {
auto eptr = mx.new_remote_udp_endpoint(addr, port);
if (eptr) {
auto hdl = (*eptr)->hdl();
self->add_datagram_servant(std::move(*eptr));
basp::instance::write_client_handshake(self->context(),
self->wr_buf(hdl),
none, this_node,
app_id);
}
}
}
return {
[=](new_datagram_msg& msg) {
auto hdl = msg.handle;
self->send(system_broker, std::move(msg), self->take(hdl), port);
self->quit();
},
after(autoconnect_timeout) >> [=]() {
CAF_LOG_TRACE(CAF_ARG(""));
// nothing heard in about 10 minutes... just a call it a day, then
CAF_LOG_INFO("aborted direct connection attempt after 10min:"
<< CAF_ARG(nid));
self->quit(exit_reason::user_shutdown);
}
};
}
behavior connection_helper(stateful_actor<connection_helper_state>* self,
actor b) {
CAF_LOG_TRACE(CAF_ARG(s));
self->monitor(b);
self->set_down_handler([=](down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm));
self->quit(std::move(dm.reason));
});
return {
// this config is send from the remote `ConfigServ`
[=](const std::string& item, message& msg) {
CAF_LOG_TRACE(CAF_ARG(item) << CAF_ARG(msg));
CAF_LOG_DEBUG("received requested config:" << CAF_ARG(msg));
// whatever happens, we are done afterwards
self->quit();
msg.apply({
[&](uint16_t port, network::address_listing& addresses) {
if (item == "basp.default-connectivity-tcp") {
auto& mx = self->system().middleman().backend();
for (auto& kvp : addresses) {
for (auto& addr : kvp.second) {
auto hdl = mx.new_tcp_scribe(addr, port);
if (hdl) {
// gotcha! send scribe to our BASP broker
// to initiate handshake etc.
CAF_LOG_INFO("connected directly:" << CAF_ARG(addr));
self->send(b, connect_atom::value, *hdl, port);
return;
}
}
}
CAF_LOG_INFO("could not connect to node directly:" << CAF_ARG(nid));
} else if (item == "basp.default-connectivity-udp") {
// create new broker to try addresses for communication via UDP
if (self->system().config().middleman_detach_utility_actors) {
self->system().middleman().spawn_broker<detached + hidden>(
datagram_connection_broker, port, std::move(addresses), b
);
} else {
self->system().middleman().spawn_broker<hidden>(
datagram_connection_broker, port, std::move(addresses), b
);
}
} else {
CAF_LOG_INFO("aborted direct connection attempt, unknown item: "
<< CAF_ARG(item));
}
}
});
},
after(autoconnect_timeout) >> [=] {
CAF_LOG_TRACE(CAF_ARG(""));
// nothing heard in about 10 minutes... just a call it a day, then
CAF_LOG_INFO("aborted direct connection attempt after 10min:"
<< CAF_ARG(nid));
self->quit(exit_reason::user_shutdown);
}
};
}
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/io/network/datagram_manager.hpp"
namespace caf {
namespace io {
namespace network {
datagram_manager::~datagram_manager() {
// nop
}
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/io/datagram_servant.hpp"
#include "caf/logger.hpp"
namespace caf {
namespace io {
datagram_servant::datagram_servant(datagram_handle hdl)
: datagram_servant_base(hdl) {
// nop
}
datagram_servant::~datagram_servant() {
// nop
}
message datagram_servant::detach_message() {
return make_message(datagram_servant_closed_msg{hdls()});
}
bool datagram_servant::consume(execution_unit* ctx, datagram_handle hdl,
network::receive_buffer& buf) {
CAF_ASSERT(ctx != nullptr);
CAF_LOG_TRACE(CAF_ARG(buf.size()));
if (detached()) {
// we are already disconnected from the broker while the multiplexer
// did not yet remove the socket, this can happen if an I/O event causes
// the broker to call close_all() while the pollset contained
// further activities for the broker
return false;
}
// keep a strong reference to our parent until we leave scope
// to avoid UB when becoming detached during invocation
auto guard = parent_;
msg().handle = hdl;
auto& msg_buf = msg().buf;
msg_buf.swap(buf);
auto result = invoke_mailbox_element(ctx);
// swap buffer back to stream and implicitly flush wr_buf()
msg_buf.swap(buf);
flush();
return result;
}
void datagram_servant::datagram_sent(execution_unit* ctx, datagram_handle hdl,
size_t written, std::vector<char> buffer) {
CAF_LOG_TRACE(CAF_ARG(written));
if (detached())
return;
using sent_t = datagram_sent_msg;
using tmp_t = mailbox_element_vals<datagram_sent_msg>;
tmp_t tmp{strong_actor_ptr{}, message_id::make(),
mailbox_element::forwarding_stack{},
sent_t{hdl, written, std::move(buffer)}};
invoke_mailbox_element_impl(ctx, tmp);
}
void datagram_servant::io_failure(execution_unit* ctx, network::operation op) {
CAF_LOG_TRACE(CAF_ARG(hdl()) << CAF_ARG(op));
// keep compiler happy when compiling w/o logging
static_cast<void>(op);
detach(ctx, true);
}
} // namespace io
} // namespace caf
This diff is collapsed.
...@@ -44,7 +44,8 @@ std::string to_string(const header &hdr) { ...@@ -44,7 +44,8 @@ std::string to_string(const header &hdr) {
<< to_string(hdr.source_node) << ", " << to_string(hdr.source_node) << ", "
<< to_string(hdr.dest_node) << ", " << to_string(hdr.dest_node) << ", "
<< hdr.source_actor << ", " << hdr.source_actor << ", "
<< hdr.dest_actor << hdr.dest_actor << ", "
<< hdr.sequence_number
<< "}"; << "}";
return oss.str(); return oss.str();
} }
...@@ -57,7 +58,8 @@ bool operator==(const header& lhs, const header& rhs) { ...@@ -57,7 +58,8 @@ bool operator==(const header& lhs, const header& rhs) {
&& lhs.source_node == rhs.source_node && lhs.source_node == rhs.source_node
&& lhs.dest_node == rhs.dest_node && lhs.dest_node == rhs.dest_node
&& lhs.source_actor == rhs.source_actor && lhs.source_actor == rhs.source_actor
&& lhs.dest_actor == rhs.dest_actor; && lhs.dest_actor == rhs.dest_actor
&& lhs.sequence_number == rhs.sequence_number;
} }
namespace { namespace {
...@@ -73,18 +75,15 @@ bool zero(T val) { ...@@ -73,18 +75,15 @@ bool zero(T val) {
bool server_handshake_valid(const header& hdr) { bool server_handshake_valid(const header& hdr) {
return valid(hdr.source_node) return valid(hdr.source_node)
&& !valid(hdr.dest_node)
&& zero(hdr.dest_actor) && zero(hdr.dest_actor)
&& !zero(hdr.operation_data); && !zero(hdr.operation_data);
} }
bool client_handshake_valid(const header& hdr) { bool client_handshake_valid(const header& hdr) {
return valid(hdr.source_node) return valid(hdr.source_node)
&& valid(hdr.dest_node)
&& hdr.source_node != hdr.dest_node && hdr.source_node != hdr.dest_node
&& zero(hdr.source_actor) && zero(hdr.source_actor)
&& zero(hdr.dest_actor) && zero(hdr.dest_actor);
&& zero(hdr.operation_data);
} }
bool dispatch_message_valid(const header& hdr) { bool dispatch_message_valid(const header& hdr) {
......
This diff is collapsed.
...@@ -50,6 +50,8 @@ ...@@ -50,6 +50,8 @@
#include "caf/detail/get_mac_addresses.hpp" #include "caf/detail/get_mac_addresses.hpp"
#include "caf/io/network/ip_endpoint.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
namespace network { namespace network {
...@@ -276,6 +278,35 @@ interfaces::server_address(uint16_t port, const char* host, ...@@ -276,6 +278,35 @@ interfaces::server_address(uint16_t port, const char* host,
return results; return results;
} }
bool interfaces::get_endpoint(const std::string& host, uint16_t port,
ip_endpoint& ep,
optional<protocol::network> preferred) {
static_assert(sizeof(uint16_t) == sizeof(unsigned short int),
"uint16_t cannot be printed with %hu in snprintf");
addrinfo hint;
// max port is 2^16 which needs 5 characters plus null terminator
char port_hint[6];
sprintf(port_hint, "%hu", port);
memset(&hint, 0, sizeof(hint));
hint.ai_socktype = SOCK_DGRAM;
if (preferred)
hint.ai_family = *preferred == protocol::network::ipv4 ? AF_INET : AF_INET6;
if (hint.ai_family == AF_INET6)
hint.ai_flags = AI_V4MAPPED;
addrinfo* tmp = nullptr;
if (getaddrinfo(host.c_str(), port_hint, &hint, &tmp) != 0)
return false;
std::unique_ptr<addrinfo, decltype(freeaddrinfo)*> addrs{tmp, freeaddrinfo};
for (auto i = addrs.get(); i != nullptr; i = i->ai_next) {
if (i->ai_family != AF_UNSPEC) {
memcpy(ep.address(), i->ai_addr, i->ai_addrlen);
*ep.length() = i->ai_addrlen;
return true;
}
}
return false;
}
} // namespace network } // namespace network
} // namespace io } // namespace io
} // namespace caf } // namespace caf
This diff is collapsed.
...@@ -150,6 +150,21 @@ expected<uint16_t> middleman::publish(const strong_actor_ptr& whom, ...@@ -150,6 +150,21 @@ expected<uint16_t> middleman::publish(const strong_actor_ptr& whom,
return f(publish_atom::value, port, std::move(whom), std::move(sigs), in, ru); return f(publish_atom::value, port, std::move(whom), std::move(sigs), in, ru);
} }
expected<uint16_t> middleman::publish_udp(const strong_actor_ptr& whom,
std::set<std::string> sigs,
uint16_t port, const char* cstr,
bool ru) {
CAF_LOG_TRACE(CAF_ARG(whom) << CAF_ARG(sigs) << CAF_ARG(port));
if (!whom)
return sec::cannot_publish_invalid_actor;
std::string in;
if (cstr != nullptr)
in = cstr;
auto f = make_function_view(actor_handle());
return f(publish_udp_atom::value, port, std::move(whom),
std::move(sigs), in, ru);
}
expected<uint16_t> middleman::publish_local_groups(uint16_t port, expected<uint16_t> middleman::publish_local_groups(uint16_t port,
const char* in, bool reuse) { const char* in, bool reuse) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(in)); CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(in));
...@@ -176,6 +191,12 @@ expected<void> middleman::unpublish(const actor_addr& whom, uint16_t port) { ...@@ -176,6 +191,12 @@ expected<void> middleman::unpublish(const actor_addr& whom, uint16_t port) {
return f(unpublish_atom::value, whom, port); return f(unpublish_atom::value, whom, port);
} }
expected<void> middleman::unpublish_udp(const actor_addr& whom, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(whom) << CAF_ARG(port));
auto f = make_function_view(actor_handle());
return f(unpublish_udp_atom::value, whom, port);
}
expected<strong_actor_ptr> middleman::remote_actor(std::set<std::string> ifs, expected<strong_actor_ptr> middleman::remote_actor(std::set<std::string> ifs,
std::string host, std::string host,
uint16_t port) { uint16_t port) {
...@@ -193,6 +214,23 @@ expected<strong_actor_ptr> middleman::remote_actor(std::set<std::string> ifs, ...@@ -193,6 +214,23 @@ expected<strong_actor_ptr> middleman::remote_actor(std::set<std::string> ifs,
return ptr; return ptr;
} }
expected<strong_actor_ptr>
middleman::remote_actor_udp(std::set<std::string> ifs, std::string host,
uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(ifs) << CAF_ARG(host) << CAF_ARG(port));
auto f = make_function_view(actor_handle());
auto res = f(contact_atom::value, std::move(host), port);
if (!res)
return std::move(res.error());
strong_actor_ptr ptr = std::move(std::get<1>(*res));
if (!ptr)
return make_error(sec::no_actor_published_at_port, port);
if (!system().assignable(std::get<2>(*res), ifs))
return make_error(sec::unexpected_actor_messaging_interface, std::move(ifs),
std::move(std::get<2>(*res)));
return ptr;
}
expected<group> middleman::remote_group(const std::string& group_uri) { expected<group> middleman::remote_group(const std::string& group_uri) {
CAF_LOG_TRACE(CAF_ARG(group_uri)); CAF_LOG_TRACE(CAF_ARG(group_uri));
// format of group_identifier is group@host:port // format of group_identifier is group@host:port
...@@ -359,6 +397,7 @@ void middleman::init(actor_system_config& cfg) { ...@@ -359,6 +397,7 @@ void middleman::init(actor_system_config& cfg) {
// add I/O-related types to config // add I/O-related types to config
cfg.add_message_type<network::protocol>("@protocol") cfg.add_message_type<network::protocol>("@protocol")
.add_message_type<network::address_listing>("@address_listing") .add_message_type<network::address_listing>("@address_listing")
.add_message_type<network::receive_buffer>("@receive_buffer")
.add_message_type<new_data_msg>("@new_data_msg") .add_message_type<new_data_msg>("@new_data_msg")
.add_message_type<new_connection_msg>("@new_connection_msg") .add_message_type<new_connection_msg>("@new_connection_msg")
.add_message_type<acceptor_closed_msg>("@acceptor_closed_msg") .add_message_type<acceptor_closed_msg>("@acceptor_closed_msg")
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
...@@ -19,7 +19,7 @@ ...@@ -19,7 +19,7 @@
#include "caf/config.hpp" #include "caf/config.hpp"
#define CAF_SUITE io_basp #define CAF_SUITE io_basp_tcp
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
#include <array> #include <array>
...@@ -482,7 +482,7 @@ CAF_TEST_FIXTURE_SCOPE(basp_tests, fixture) ...@@ -482,7 +482,7 @@ CAF_TEST_FIXTURE_SCOPE(basp_tests, fixture)
CAF_TEST(empty_server_handshake) { CAF_TEST(empty_server_handshake) {
// test whether basp instance correctly sends a // test whether basp instance correctly sends a
// server handshake whene there's no actor published // server handshake when there's no actor published
buffer buf; buffer buf;
instance().write_server_handshake(mpx(), buf, none); instance().write_server_handshake(mpx(), buf, none);
basp::header hdr; basp::header hdr;
...@@ -786,6 +786,12 @@ CAF_TEST(automatic_connection) { ...@@ -786,6 +786,12 @@ CAF_TEST(automatic_connection) {
// jupiter [remote hdl 0] -> mars [remote hdl 1] -> earth [this_node] // jupiter [remote hdl 0] -> mars [remote hdl 1] -> earth [this_node]
// (this node receives a message from jupiter via mars and responds via mars, // (this node receives a message from jupiter via mars and responds via mars,
// but then also establishes a connection to jupiter directly) // but then also establishes a connection to jupiter directly)
auto check_node_in_tbl = [&](node& n) {
io::id_visitor id_vis;
auto hdl = tbl().lookup_direct(n.id);
CAF_REQUIRE(hdl);
CAF_CHECK_EQUAL(visit(id_vis, *hdl), n.connection.id());
};
mpx()->provide_scribe("jupiter", 8080, jupiter().connection); mpx()->provide_scribe("jupiter", 8080, jupiter().connection);
CAF_CHECK(mpx()->has_pending_scribe("jupiter", 8080)); CAF_CHECK(mpx()->has_pending_scribe("jupiter", 8080));
CAF_MESSAGE("self: " << to_string(self()->address())); CAF_MESSAGE("self: " << to_string(self()->address()));
...@@ -795,7 +801,8 @@ CAF_TEST(automatic_connection) { ...@@ -795,7 +801,8 @@ CAF_TEST(automatic_connection) {
mpx()->flush_runnables(); // process publish message in basp_broker mpx()->flush_runnables(); // process publish message in basp_broker
CAF_MESSAGE("connect to mars"); CAF_MESSAGE("connect to mars");
connect_node(mars(), ax, self()->id()); connect_node(mars(), ax, self()->id());
CAF_CHECK_EQUAL(tbl().lookup_direct(mars().id).id(), mars().connection.id()); //CAF_CHECK_EQUAL(tbl().lookup_direct(mars().id).id(), mars().connection.id());
check_node_in_tbl(mars());
CAF_MESSAGE("simulate that an actor from jupiter " CAF_MESSAGE("simulate that an actor from jupiter "
"sends a message to us via mars"); "sends a message to us via mars");
mock(mars().connection, mock(mars().connection,
...@@ -819,7 +826,7 @@ CAF_TEST(automatic_connection) { ...@@ -819,7 +826,7 @@ CAF_TEST(automatic_connection) {
invalid_actor_id, invalid_actor_id,
config_serv_atom, config_serv_atom,
std::vector<actor_id>{}, std::vector<actor_id>{},
make_message(get_atom::value, "basp.default-connectivity")) make_message(get_atom::value, "basp.default-connectivity-tcp"))
.receive(mars().connection, .receive(mars().connection,
basp::message_type::announce_proxy, no_flags, no_payload, basp::message_type::announce_proxy, no_flags, no_payload,
no_operation_data, this_node(), jupiter().id, no_operation_data, this_node(), jupiter().id,
...@@ -827,7 +834,7 @@ CAF_TEST(automatic_connection) { ...@@ -827,7 +834,7 @@ CAF_TEST(automatic_connection) {
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u); CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
CAF_CHECK_EQUAL(tbl().lookup_indirect(jupiter().id), mars().id); CAF_CHECK_EQUAL(tbl().lookup_indirect(jupiter().id), mars().id);
CAF_CHECK_EQUAL(tbl().lookup_indirect(mars().id), none); CAF_CHECK_EQUAL(tbl().lookup_indirect(mars().id), none);
auto connection_helper = sys.latest_actor_id(); auto connection_helper_actor = sys.latest_actor_id();
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u); CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
// create a dummy config server and respond to the name lookup // create a dummy config server and respond to the name lookup
CAF_MESSAGE("receive ConfigServ of jupiter"); CAF_MESSAGE("receive ConfigServ of jupiter");
...@@ -836,9 +843,9 @@ CAF_TEST(automatic_connection) { ...@@ -836,9 +843,9 @@ CAF_TEST(automatic_connection) {
mock(mars().connection, mock(mars().connection,
{basp::message_type::dispatch_message, 0, 0, 0, {basp::message_type::dispatch_message, 0, 0, 0,
this_node(), this_node(), this_node(), this_node(),
invalid_actor_id, connection_helper}, invalid_actor_id, connection_helper_actor},
std::vector<actor_id>{}, std::vector<actor_id>{},
make_message("basp.default-connectivity", make_message("basp.default-connectivity-tcp",
make_message(uint16_t{8080}, std::move(res)))); make_message(uint16_t{8080}, std::move(res))));
// our connection helper should now connect to jupiter and // our connection helper should now connect to jupiter and
// send the scribe handle over to the BASP broker // send the scribe handle over to the BASP broker
...@@ -861,9 +868,8 @@ CAF_TEST(automatic_connection) { ...@@ -861,9 +868,8 @@ CAF_TEST(automatic_connection) {
invalid_actor_id, invalid_actor_id, std::string{}); invalid_actor_id, invalid_actor_id, std::string{});
CAF_CHECK_EQUAL(tbl().lookup_indirect(jupiter().id), none); CAF_CHECK_EQUAL(tbl().lookup_indirect(jupiter().id), none);
CAF_CHECK_EQUAL(tbl().lookup_indirect(mars().id), none); CAF_CHECK_EQUAL(tbl().lookup_indirect(mars().id), none);
CAF_CHECK_EQUAL(tbl().lookup_direct(jupiter().id).id(), check_node_in_tbl(jupiter());
jupiter().connection.id()); check_node_in_tbl(mars());
CAF_CHECK_EQUAL(tbl().lookup_direct(mars().id).id(), mars().connection.id());
CAF_MESSAGE("receive message from jupiter"); CAF_MESSAGE("receive message from jupiter");
self()->receive( self()->receive(
[](const std::string& str) -> std::string { [](const std::string& str) -> std::string {
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment