Commit 42dcb781 authored by Joseph Noir's avatar Joseph Noir

Add basic UDP support

This commit adds basic support for UDP to CAF. It includes a new
datagram servant for the multiplexer, associated handles, message types
and tests. CAF messages sent via UDP are delivered in order but
unreliably.

The receive buffer has a size of 65k which should enable CAF to receive
all regular datagrams. Currently, messages that are bigger than the MTU
will be fragmented by IP.

Message slicing and optional reliability are planned for the future.
parent 6c0f3d70
...@@ -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); 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);
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 boost { namespace boost {
...@@ -73,6 +74,26 @@ public: ...@@ -73,6 +74,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.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
...@@ -116,7 +116,7 @@ CAF_TEST(unpublishing) { ...@@ -116,7 +116,7 @@ CAF_TEST(unpublishing) {
CAF_MESSAGE("check whether testee is still available via cache"); CAF_MESSAGE("check whether testee is still available via cache");
auto x1 = remote_actor("127.0.0.1", port); auto x1 = remote_actor("127.0.0.1", port);
CAF_CHECK_EQUAL(x1, testee); CAF_CHECK_EQUAL(x1, testee);
CAF_MESSAGE("fake dead of testee and check if testee becomes unavailable"); CAF_MESSAGE("fake death of testee and check if testee becomes unavailable");
anon_send(actor_cast<actor>(system.middleman().actor_handle()), anon_send(actor_cast<actor>(system.middleman().actor_handle()),
down_msg{testee.address(), exit_reason::normal}); down_msg{testee.address(), exit_reason::normal});
// must fail now // must fail now
......
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