Commit fac1cc62 authored by Dominik Charousset's avatar Dominik Charousset

Remove broken UDP option for BASP

parent 14c187b8
...@@ -18,13 +18,14 @@ ...@@ -18,13 +18,14 @@
#pragma once #pragma once
#include "caf/io/basp/buffer_type.hpp"
#include "caf/io/basp/connection_state.hpp"
#include "caf/io/basp/endpoint_context.hpp"
#include "caf/io/basp/header.hpp" #include "caf/io/basp/header.hpp"
#include "caf/io/basp/version.hpp"
#include "caf/io/basp/instance.hpp" #include "caf/io/basp/instance.hpp"
#include "caf/io/basp/buffer_type.hpp"
#include "caf/io/basp/message_type.hpp" #include "caf/io/basp/message_type.hpp"
#include "caf/io/basp/routing_table.hpp" #include "caf/io/basp/routing_table.hpp"
#include "caf/io/basp/connection_state.hpp" #include "caf/io/basp/version.hpp"
/// @defgroup BASP Binary Actor Sytem Protocol /// @defgroup BASP Binary Actor Sytem Protocol
/// ///
......
...@@ -35,14 +35,12 @@ namespace basp { ...@@ -35,14 +35,12 @@ namespace basp {
// stores meta information for active endpoints // stores meta information for active endpoints
struct endpoint_context { struct endpoint_context {
using pending_map = std::map<sequence_type, std::pair<basp::header,
std::vector<char>>>;
// denotes what message we expect from the remote node next // denotes what message we expect from the remote node next
basp::connection_state cstate; basp::connection_state cstate;
// our currently processed BASP header // our currently processed BASP header
basp::header hdr; basp::header hdr;
// the handle for I/O operations // the handle for I/O operations
variant<connection_handle, datagram_handle> hdl; connection_handle hdl;
// network-agnostic node identifier // network-agnostic node identifier
node_id id; node_id id;
// ports // ports
...@@ -50,15 +48,6 @@ struct endpoint_context { ...@@ -50,15 +48,6 @@ struct endpoint_context {
uint16_t local_port; uint16_t local_port;
// pending operations to be performed after handshake completed // pending operations to be performed after handshake completed
optional<response_promise> callback; 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
sequence_type seq_incoming;
sequence_type seq_outgoing;
// pending messages due to ordering
pending_map pending;
// track if a timeout to deliver pending messages is set
bool did_set_timeout;
}; };
} // namespace basp } // namespace basp
......
...@@ -35,9 +35,6 @@ namespace basp { ...@@ -35,9 +35,6 @@ 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
...@@ -53,38 +50,18 @@ struct header { ...@@ -53,38 +50,18 @@ 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,
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)
: 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(0) {
// nop
}
inline header(message_type m_operation, uint8_t m_flags, header(message_type m_operation, uint8_t m_flags, uint32_t m_payload_len,
uint32_t m_payload_len, uint64_t m_operation_data, uint64_t m_operation_data, node_id m_source_node, node_id m_dest_node,
node_id m_source_node, node_id m_dest_node, actor_id m_source_actor, actor_id m_dest_actor)
actor_id m_source_actor, actor_id m_dest_actor, : operation(m_operation),
sequence_type m_sequence_number) flags(m_flags),
: operation(m_operation), payload_len(m_payload_len),
flags(m_flags), operation_data(m_operation_data),
payload_len(m_payload_len), source_node(std::move(m_source_node)),
operation_data(m_operation_data), dest_node(std::move(m_dest_node)),
source_node(std::move(m_source_node)), source_actor(m_source_actor),
dest_node(std::move(m_dest_node)), dest_actor(m_dest_actor) {
source_actor(m_source_actor),
dest_actor(m_dest_actor),
sequence_number(m_sequence_number) {
// nop // nop
} }
...@@ -94,7 +71,7 @@ struct header { ...@@ -94,7 +71,7 @@ struct header {
static const uint8_t named_receiver_flag = 0x01; static const uint8_t named_receiver_flag = 0x01;
/// Queries whether this header has the given flag. /// Queries whether this header has the given flag.
inline bool has(uint8_t flag) const { bool has(uint8_t flag) const {
return (flags & flag) != 0; return (flags & flag) != 0;
} }
}; };
...@@ -109,8 +86,7 @@ typename Inspector::result_type inspect(Inspector& f, header& hdr) { ...@@ -109,8 +86,7 @@ typename Inspector::result_type inspect(Inspector& f, header& hdr) {
meta::omittable(), pad, meta::omittable(), pad,
hdr.flags, hdr.payload_len, hdr.operation_data, hdr.flags, hdr.payload_len, hdr.operation_data,
hdr.source_node, hdr.dest_node, hdr.source_node, hdr.dest_node,
hdr.source_actor, hdr.dest_actor, hdr.source_actor, hdr.dest_actor);
hdr.sequence_number);
} }
/// @relates header /// @relates header
...@@ -140,8 +116,7 @@ bool valid(const header& hdr); ...@@ -140,8 +116,7 @@ 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.
...@@ -21,25 +21,10 @@ ...@@ -21,25 +21,10 @@
#include <unordered_map> #include <unordered_map>
#include <unordered_set> #include <unordered_set>
#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"
#include "caf/node_id.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 {
...@@ -51,7 +36,6 @@ namespace basp { ...@@ -51,7 +36,6 @@ 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);
...@@ -60,7 +44,7 @@ public: ...@@ -60,7 +44,7 @@ public:
/// Describes a routing path to a node. /// Describes a routing path to a node.
struct route { struct route {
const node_id& next_hop; const node_id& next_hop;
endpoint_handle hdl; connection_handle hdl;
}; };
/// Describes a function object for erase operations that /// Describes a function object for erase operations that
...@@ -72,11 +56,11 @@ public: ...@@ -72,11 +56,11 @@ 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 endpoint_handle& hdl) const; node_id lookup_direct(const connection_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.
optional<endpoint_handle> lookup_direct(const node_id& nid) const; optional<connection_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`.
...@@ -84,7 +68,7 @@ public: ...@@ -84,7 +68,7 @@ public:
/// 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 endpoint_handle& hdl, const node_id& nid); void add_direct(const connection_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);
...@@ -95,7 +79,7 @@ public: ...@@ -95,7 +79,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 endpoint_handle& hdl, erase_callback& cb); void erase_direct(const connection_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`.
...@@ -131,8 +115,8 @@ public: ...@@ -131,8 +115,8 @@ public:
node_id_set>; // hop node_id_set>; // hop
abstract_broker* parent_; abstract_broker* parent_;
std::unordered_map<endpoint_handle, node_id> direct_by_hdl_; std::unordered_map<connection_handle, node_id> direct_by_hdl_;
std::unordered_map<node_id, endpoint_handle> direct_by_nid_; std::unordered_map<node_id, connection_handle> direct_by_nid_;
indirect_entries indirect_; indirect_entries indirect_;
indirect_entries blacklist_; indirect_entries blacklist_;
}; };
......
...@@ -35,11 +35,8 @@ ...@@ -35,11 +35,8 @@
#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 {
...@@ -89,42 +86,9 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -89,42 +86,9 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// inherited from basp::instance::callee // 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(execution_unit* ctx, basp::endpoint_context& ep,
uint16_t seq, basp::header hdr,
std::vector<char> payload) override;
// inherited from basp::instance::callee
bool deliver_pending(execution_unit* ctx, basp::endpoint_context& ep,
bool force) override;
// inherited from basp::instance::callee
void drop_pending(basp::endpoint_context& ep, uint16_t seq) 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 // inherited from basp::instance::callee
buffer_type& get_buffer(connection_handle hdl) override; 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 // inherited from basp::instance::callee
void flush(connection_handle hdl) override; void flush(connection_handle hdl) override;
...@@ -134,11 +98,9 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -134,11 +98,9 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
/// 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;
...@@ -146,14 +108,10 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -146,14 +108,10 @@ 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_tcp_map = std::unordered_map<connection_handle, using ctx_map = std::unordered_map<connection_handle, basp::endpoint_context>;
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_tcp_map ctx_tcp; ctx_map ctx;
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`
basp::endpoint_context* this_context = nullptr; basp::endpoint_context* this_context = nullptr;
...@@ -167,19 +125,6 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -167,19 +125,6 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
/// routing paths by forming a mesh between all nodes. /// routing paths by forming a mesh between all nodes.
bool automatic_connections = false; bool automatic_connections = false;
/// Configures whether BASP allows TCP connections.
bool allow_tcp = true;
/// Configures whether BASP allows UDP connections.
bool allow_udp = false;
// reusable send buffers for UDP communication
const size_t max_buffers;
std::stack<buffer_type> cached_buffers;
// maximum queue size for pending messages of endpoints with ordering
const size_t max_pending_messages;
// timeout for delivery of pending messages of endpoints with ordering // timeout for delivery of pending messages of endpoints with ordering
const std::chrono::milliseconds pending_to = std::chrono::milliseconds(100); const std::chrono::milliseconds pending_to = std::chrono::milliseconds(100);
......
...@@ -76,22 +76,6 @@ public: ...@@ -76,22 +76,6 @@ 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 UDP 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`
...@@ -108,14 +92,6 @@ public: ...@@ -108,14 +92,6 @@ 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.
...@@ -131,21 +107,6 @@ public: ...@@ -131,21 +107,6 @@ 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 UDP 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);
...@@ -372,20 +333,11 @@ private: ...@@ -372,20 +333,11 @@ 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
......
...@@ -68,12 +68,6 @@ namespace io { ...@@ -68,12 +68,6 @@ 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.
...@@ -98,23 +92,14 @@ using middleman_actor = ...@@ -98,23 +92,14 @@ using middleman_actor =
::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>>
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* *
* 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. *
******************************************************************************/
#pragma once
#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
...@@ -88,9 +88,11 @@ void abstract_broker::ack_writes(connection_handle hdl, bool enable) { ...@@ -88,9 +88,11 @@ void abstract_broker::ack_writes(connection_handle hdl, bool enable) {
} }
std::vector<char>& abstract_broker::wr_buf(connection_handle hdl) { std::vector<char>& abstract_broker::wr_buf(connection_handle hdl) {
CAF_ASSERT(hdl != invalid_connection_handle);
auto x = by_id(hdl); auto x = by_id(hdl);
if (!x) { if (!x) {
CAF_LOG_ERROR("tried to access wr_buf() of an unknown connection_handle"); CAF_LOG_ERROR("tried to access wr_buf() of an unknown connection_handle:"
<< CAF_ARG(hdl));
return dummy_wr_buf_; return dummy_wr_buf_;
} }
return x->wr_buf(); return x->wr_buf();
......
This diff is collapsed.
...@@ -43,8 +43,7 @@ std::string to_string(const header &hdr) { ...@@ -43,8 +43,7 @@ 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,8 +56,7 @@ bool operator==(const header& lhs, const header& rhs) { ...@@ -57,8 +56,7 @@ 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 {
......
This diff is collapsed.
...@@ -144,21 +144,6 @@ expected<uint16_t> middleman::publish(const strong_actor_ptr& whom, ...@@ -144,21 +144,6 @@ expected<uint16_t> middleman::publish(const strong_actor_ptr& whom,
return f(publish_atom::value, port, std::move(whom), std::move(sigs), in, ru); return f(publish_atom::value, port, std::move(whom), std::move(sigs), in, ru);
} }
expected<uint16_t> middleman::publish_udp(const strong_actor_ptr& whom,
std::set<std::string> sigs,
uint16_t port, const char* cstr,
bool ru) {
CAF_LOG_TRACE(CAF_ARG(whom) << CAF_ARG(sigs) << CAF_ARG(port));
if (!whom)
return sec::cannot_publish_invalid_actor;
std::string in;
if (cstr != nullptr)
in = cstr;
auto f = make_function_view(actor_handle());
return f(publish_udp_atom::value, port, std::move(whom),
std::move(sigs), in, ru);
}
expected<uint16_t> middleman::publish_local_groups(uint16_t port, expected<uint16_t> middleman::publish_local_groups(uint16_t port,
const char* in, bool reuse) { const char* in, bool reuse) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(in)); CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(in));
...@@ -185,12 +170,6 @@ expected<void> middleman::unpublish(const actor_addr& whom, uint16_t port) { ...@@ -185,12 +170,6 @@ expected<void> middleman::unpublish(const actor_addr& whom, uint16_t port) {
return f(unpublish_atom::value, whom, port); return f(unpublish_atom::value, whom, port);
} }
expected<void> middleman::unpublish_udp(const actor_addr& whom, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(whom) << CAF_ARG(port));
auto f = make_function_view(actor_handle());
return f(unpublish_udp_atom::value, whom, port);
}
expected<strong_actor_ptr> middleman::remote_actor(std::set<std::string> ifs, expected<strong_actor_ptr> middleman::remote_actor(std::set<std::string> ifs,
std::string host, std::string host,
uint16_t port) { uint16_t port) {
...@@ -208,23 +187,6 @@ expected<strong_actor_ptr> middleman::remote_actor(std::set<std::string> ifs, ...@@ -208,23 +187,6 @@ expected<strong_actor_ptr> middleman::remote_actor(std::set<std::string> ifs,
return ptr; return ptr;
} }
expected<strong_actor_ptr>
middleman::remote_actor_udp(std::set<std::string> ifs, std::string host,
uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(ifs) << CAF_ARG(host) << CAF_ARG(port));
auto f = make_function_view(actor_handle());
auto res = f(contact_atom::value, std::move(host), port);
if (!res)
return std::move(res.error());
strong_actor_ptr ptr = std::move(std::get<1>(*res));
if (!ptr)
return make_error(sec::no_actor_published_at_port, port);
if (!system().assignable(std::get<2>(*res), ifs))
return make_error(sec::unexpected_actor_messaging_interface, std::move(ifs),
std::move(std::get<2>(*res)));
return ptr;
}
expected<group> middleman::remote_group(const std::string& group_uri) { expected<group> middleman::remote_group(const std::string& group_uri) {
CAF_LOG_TRACE(CAF_ARG(group_uri)); CAF_LOG_TRACE(CAF_ARG(group_uri));
// format of group_identifier is group@host:port // format of group_identifier is group@host:port
......
...@@ -53,14 +53,6 @@ middleman_actor_impl::middleman_actor_impl(actor_config& cfg, ...@@ -53,14 +53,6 @@ middleman_actor_impl::middleman_actor_impl(actor_config& cfg,
else else
++i; ++i;
} }
i = cached_udp_.begin();
e = cached_udp_.end();
while (i != e) {
if (get<1>(i->second) == dm.source)
i = cached_udp_.erase(i);
else
++i;
}
}); });
set_exit_handler([=](exit_msg&) { set_exit_handler([=](exit_msg&) {
// ignored, the MM links group nameservers // ignored, the MM links group nameservers
...@@ -72,7 +64,6 @@ void middleman_actor_impl::on_exit() { ...@@ -72,7 +64,6 @@ void middleman_actor_impl::on_exit() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
broker_ = nullptr; broker_ = nullptr;
cached_tcp_.clear(); cached_tcp_.clear();
cached_udp_.clear();
for (auto& kvp : pending_) for (auto& kvp : pending_)
for (auto& promise : kvp.second) for (auto& promise : kvp.second)
promise.deliver(make_error(sec::cannot_connect_to_node)); promise.deliver(make_error(sec::cannot_connect_to_node));
...@@ -85,32 +76,20 @@ const char* middleman_actor_impl::name() const { ...@@ -85,32 +76,20 @@ const char* middleman_actor_impl::name() const {
auto middleman_actor_impl::make_behavior() -> behavior_type { auto middleman_actor_impl::make_behavior() -> behavior_type {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
auto tcp_disabled = [=] {
return get_or(config(), "middleman.disable-tcp", false);
};
auto udp_disabled = [=] {
return !get_or(config(), "middleman.enable-udp", false);
};
return { return {
[=](publish_atom, uint16_t port, strong_actor_ptr& whom, mpi_set& sigs, [=](publish_atom, uint16_t port, strong_actor_ptr& whom, mpi_set& sigs,
std::string& addr, bool reuse) -> put_res { std::string& addr, bool reuse) -> put_res {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
if (tcp_disabled())
return make_error(sec::feature_disabled);
return put(port, whom, sigs, addr.c_str(), reuse); return put(port, whom, sigs, addr.c_str(), reuse);
}, },
[=](open_atom, uint16_t port, std::string& addr, bool reuse) -> put_res { [=](open_atom, uint16_t port, std::string& addr, bool reuse) -> put_res {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
if (tcp_disabled())
return make_error(sec::feature_disabled);
strong_actor_ptr whom; strong_actor_ptr whom;
mpi_set sigs; mpi_set sigs;
return put(port, whom, sigs, addr.c_str(), reuse); return put(port, whom, sigs, addr.c_str(), reuse);
}, },
[=](connect_atom, std::string& hostname, uint16_t port) -> get_res { [=](connect_atom, std::string& hostname, uint16_t port) -> get_res {
CAF_LOG_TRACE(CAF_ARG(hostname) << CAF_ARG(port)); CAF_LOG_TRACE(CAF_ARG(hostname) << CAF_ARG(port));
if (tcp_disabled())
return make_error(sec::feature_disabled);
auto rp = make_response_promise(); auto rp = make_response_promise();
endpoint key{std::move(hostname), port}; endpoint key{std::move(hostname), port};
// respond immediately if endpoint is cached // respond immediately if endpoint is cached
...@@ -162,78 +141,11 @@ auto middleman_actor_impl::make_behavior() -> behavior_type { ...@@ -162,78 +141,11 @@ auto middleman_actor_impl::make_behavior() -> behavior_type {
}); });
return get_delegated{}; return get_delegated{};
}, },
[=](publish_udp_atom, uint16_t port, strong_actor_ptr& whom,
mpi_set& sigs, std::string& addr, bool reuse) -> put_res {
CAF_LOG_TRACE("");
if (udp_disabled())
return make_error(sec::feature_disabled);
return put_udp(port, whom, sigs, addr.c_str(), reuse);
},
[=](contact_atom, std::string& hostname, uint16_t port) -> get_res {
CAF_LOG_TRACE(CAF_ARG(hostname) << CAF_ARG(port));
if (udp_disabled())
return make_error(sec::feature_disabled);
auto rp = make_response_promise();
endpoint key{std::move(hostname), port};
// respond immediately if endpoint is cached
auto x = cached_udp(key);
if (x) {
CAF_LOG_DEBUG("found cached entry" << CAF_ARG(*x));
rp.deliver(get<0>(*x), get<1>(*x), get<2>(*x));
return get_delegated{};
}
// attach this promise to a pending request if possible
auto rps = pending(key);
if (rps) {
CAF_LOG_DEBUG("attach to pending request");
rps->emplace_back(std::move(rp));
return get_delegated{};
}
// connect to endpoint and initiate handshake etc.
auto r = contact(key.first, port);
if (!r) {
rp.deliver(std::move(r.error()));
return get_delegated{};
}
auto& ptr = *r;
std::vector<response_promise> tmp{std::move(rp)};
pending_.emplace(key, std::move(tmp));
request(broker_, infinite, contact_atom::value, std::move(ptr), port)
.then(
[=](node_id& nid, strong_actor_ptr& addr, mpi_set& sigs) {
auto i = pending_.find(key);
if (i == pending_.end())
return;
if (nid && addr) {
monitor(addr);
cached_udp_.emplace(key, std::make_tuple(nid, addr, sigs));
}
auto res = make_message(std::move(nid), std::move(addr),
std::move(sigs));
for (auto& promise : i->second)
promise.deliver(res);
pending_.erase(i);
},
[=](error& err) {
auto i = pending_.find(key);
if (i == pending_.end())
return;
for (auto& promise : i->second)
promise.deliver(err);
pending_.erase(i);
});
return get_delegated{};
},
[=](unpublish_atom atm, actor_addr addr, uint16_t p) -> del_res { [=](unpublish_atom atm, actor_addr addr, uint16_t p) -> del_res {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
delegate(broker_, atm, std::move(addr), p); delegate(broker_, atm, std::move(addr), p);
return {}; return {};
}, },
[=](unpublish_udp_atom atm, actor_addr addr, uint16_t p) -> del_res {
CAF_LOG_TRACE("");
delegate(broker_, atm, std::move(addr), p);
return {};
},
[=](close_atom atm, uint16_t p) -> del_res { [=](close_atom atm, uint16_t p) -> del_res {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
delegate(broker_, atm, p); delegate(broker_, atm, p);
......
...@@ -52,11 +52,11 @@ optional<routing_table::route> routing_table::lookup(const node_id& target) { ...@@ -52,11 +52,11 @@ optional<routing_table::route> routing_table::lookup(const node_id& target) {
return none; return none;
} }
node_id routing_table::lookup_direct(const endpoint_handle& hdl) const { node_id routing_table::lookup_direct(const connection_handle& hdl) const {
return get_opt(direct_by_hdl_, hdl, none); return get_opt(direct_by_hdl_, hdl, none);
} }
optional<routing_table::endpoint_handle> optional<connection_handle>
routing_table::lookup_direct(const node_id& nid) const { routing_table::lookup_direct(const node_id& nid) const {
auto i = direct_by_nid_.find(nid); auto i = direct_by_nid_.find(nid);
if (i != direct_by_nid_.end()) if (i != direct_by_nid_.end())
...@@ -83,7 +83,7 @@ void routing_table::blacklist(const node_id& hop, const node_id& dest) { ...@@ -83,7 +83,7 @@ void routing_table::blacklist(const node_id& hop, const node_id& dest) {
indirect_.erase(i); indirect_.erase(i);
} }
void routing_table::erase_direct(const endpoint_handle& hdl, void routing_table::erase_direct(const connection_handle& hdl,
erase_callback& cb) { erase_callback& cb) {
auto i = direct_by_hdl_.find(hdl); auto i = direct_by_hdl_.find(hdl);
if (i == direct_by_hdl_.end()) if (i == direct_by_hdl_.end())
...@@ -105,7 +105,7 @@ bool routing_table::erase_indirect(const node_id& dest) { ...@@ -105,7 +105,7 @@ bool routing_table::erase_indirect(const node_id& dest) {
return true; return true;
} }
void routing_table::add_direct(const endpoint_handle& hdl, void routing_table::add_direct(const connection_handle& hdl,
const node_id& nid) { const node_id& nid) {
CAF_ASSERT(direct_by_hdl_.count(hdl) == 0); CAF_ASSERT(direct_by_hdl_.count(hdl) == 0);
CAF_ASSERT(direct_by_nid_.count(nid) == 0); CAF_ASSERT(direct_by_nid_.count(nid) == 0);
......
...@@ -18,7 +18,7 @@ ...@@ -18,7 +18,7 @@
#include "caf/config.hpp" #include "caf/config.hpp"
#define CAF_SUITE io_basp_tcp #define CAF_SUITE io_basp
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
#include <array> #include <array>
...@@ -771,10 +771,8 @@ CAF_TEST(automatic_connection) { ...@@ -771,10 +771,8 @@ CAF_TEST(automatic_connection) {
// (this node receives a message from jupiter via mars and responds via mars, // (this node receives a message from jupiter via mars and responds via mars,
// but then also establishes a connection to jupiter directly) // but then also establishes a connection to jupiter directly)
auto check_node_in_tbl = [&](node& n) { auto check_node_in_tbl = [&](node& n) {
io::id_visitor id_vis;
auto hdl = tbl().lookup_direct(n.id); auto hdl = tbl().lookup_direct(n.id);
CAF_REQUIRE(hdl); CAF_REQUIRE(hdl);
CAF_CHECK_EQUAL(visit(id_vis, *hdl), n.connection.id());
}; };
mpx()->provide_scribe("jupiter", 8080, jupiter().connection); mpx()->provide_scribe("jupiter", 8080, jupiter().connection);
CAF_CHECK(mpx()->has_pending_scribe("jupiter", 8080)); CAF_CHECK(mpx()->has_pending_scribe("jupiter", 8080));
......
This diff is collapsed.
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* 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/config.hpp"
#define CAF_SUITE io_dynamic_remote_actor_udp
#include "caf/test/dsl.hpp"
#include <vector>
#include <sstream>
#include <utility>
#include <algorithm>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
using namespace caf;
namespace {
constexpr char local_host[] = "127.0.0.1";
class config : public actor_system_config {
public:
config() {
load<io::middleman>();
set("middleman.enable-udp", true);
add_message_type<std::vector<int>>("std::vector<int>");
if (auto err = parse(test::engine::argc(), test::engine::argv()))
CAF_FAIL("failed to parse config: " << to_string(err));
}
};
struct fixture {
// State for the server.
config server_side_config;
actor_system server_side;
io::middleman& server_side_mm;
// State for the client.
config client_side_config;
actor_system client_side;
io::middleman& client_side_mm;
fixture()
: server_side(server_side_config),
server_side_mm(server_side.middleman()),
client_side(client_side_config),
client_side_mm(client_side.middleman()) {
// nop
}
};
behavior make_pong_behavior() {
return {
[](int val) -> int {
++val;
CAF_MESSAGE("pong with " << val);
return val;
}
};
}
behavior make_ping_behavior(event_based_actor* self, const actor& pong) {
CAF_MESSAGE("ping with " << 0);
self->send(pong, 0);
return {
[=](int val) -> int {
if (val == 3) {
CAF_MESSAGE("ping with exit");
self->send_exit(self->current_sender(),
exit_reason::user_shutdown);
CAF_MESSAGE("ping quits");
self->quit();
}
CAF_MESSAGE("ping with " << val);
return val;
}
};
}
behavior make_sort_behavior() {
return {
[](std::vector<int>& vec) -> std::vector<int> {
CAF_MESSAGE("sorter received: " << deep_to_string(vec));
std::sort(vec.begin(), vec.end());
CAF_MESSAGE("sorter sent: " << deep_to_string(vec));
return std::move(vec);
}
};
}
behavior make_sort_requester_behavior(event_based_actor* self,
const actor& sorter) {
self->send(sorter, std::vector<int>{5, 4, 3, 2, 1});
return {
[=](const std::vector<int>& vec) {
CAF_MESSAGE("sort requester received: " << deep_to_string(vec));
std::vector<int> expected_vec{1, 2, 3, 4, 5};
CAF_CHECK_EQUAL(vec, expected_vec);
self->send_exit(sorter, exit_reason::user_shutdown);
self->quit();
}
};
}
behavior fragile_mirror(event_based_actor* self) {
return {
[=](int i) {
self->quit(exit_reason::user_shutdown);
return i;
}
};
}
behavior linking_actor(event_based_actor* self, const actor& buddy) {
CAF_MESSAGE("link to mirror and send dummy message");
self->link_to(buddy);
self->send(buddy, 42);
return {
[](int i) {
CAF_CHECK_EQUAL(i, 42);
}
};
}
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(dynamic_remote_actor_tests_udp, fixture)
CAF_TEST(identity_semantics_udp) {
// server side
auto server = server_side.spawn(make_pong_behavior);
auto port1 = unbox(server_side_mm.publish_udp(server, 0, local_host));
auto port2 = unbox(server_side_mm.publish_udp(server, 0, local_host));
CAF_REQUIRE_NOT_EQUAL(port1, port2);
auto same_server = unbox(server_side_mm.remote_actor_udp(local_host, port2));
CAF_REQUIRE_EQUAL(same_server, server);
CAF_CHECK_EQUAL(same_server->node(), server_side.node());
auto server1 = unbox(client_side_mm.remote_actor_udp(local_host, port1));
auto server2 = unbox(client_side_mm.remote_actor_udp(local_host, port2));
CAF_CHECK_EQUAL(server1, client_side_mm.remote_actor_udp(local_host, port1));
CAF_CHECK_EQUAL(server2, client_side_mm.remote_actor_udp(local_host, port2));
anon_send_exit(server, exit_reason::user_shutdown);
}
CAF_TEST(ping_pong_udp) {
// server side
auto port = unbox(server_side_mm.publish_udp(
server_side.spawn(make_pong_behavior), 0, local_host));
// client side
auto pong = unbox(client_side_mm.remote_actor_udp(local_host, port));
client_side.spawn(make_ping_behavior, pong);
}
CAF_TEST(custom_message_type_udp) {
// server side
auto port = unbox(server_side_mm.publish_udp(
server_side.spawn(make_sort_behavior), 0, local_host));
// client side
auto sorter = unbox(client_side_mm.remote_actor_udp(local_host, port));
client_side.spawn(make_sort_requester_behavior, sorter);
}
CAF_TEST(remote_link_udp) {
// server side
auto port = unbox(server_side_mm.publish_udp(
server_side.spawn(fragile_mirror), 0, local_host));
// client side
auto mirror = unbox(client_side_mm.remote_actor_udp(local_host, port));
auto linker = client_side.spawn(linking_actor, mirror);
scoped_actor self{client_side};
self->wait_for(linker);
CAF_MESSAGE("linker exited");
self->wait_for(mirror);
CAF_MESSAGE("mirror exited");
}
CAF_TEST_DISABLED(multiple_endpoints_udp) {
config cfg;
// Setup server.
CAF_MESSAGE("creating server");
actor_system server_sys{cfg};
auto mirror = server_sys.spawn([]() -> behavior {
return {
[] (std::string str) {
std::reverse(begin(str), end(str));
return str;
}
};
});
auto port = server_sys.middleman().publish_udp(mirror, 0);
CAF_REQUIRE(port);
CAF_MESSAGE("server running on port " << port);
auto client_fun = [](event_based_actor* self) -> behavior {
return {
[=](actor s) {
self->send(s, "hellow, world");
},
[=](const std::string& str) {
CAF_CHECK_EQUAL(str, "dlrow ,wolleh");
self->quit();
CAF_MESSAGE("done");
}
};
};
// Setup a client.
CAF_MESSAGE("creating first client");
config client_cfg;
actor_system client_sys{client_cfg};
auto client = client_sys.spawn(client_fun);
// Acquire remote actor from the server.
auto client_srv = client_sys.middleman().remote_actor_udp("localhost", *port);
CAF_REQUIRE(client_srv);
// Setup other clients.
for (int i = 0; i < 5; ++i) {
config other_cfg;
actor_system other_sys{other_cfg};
CAF_MESSAGE("creating new client");
auto other = other_sys.spawn(client_fun);
// Acquire remote actor from the new server.
auto other_srv = other_sys.middleman().remote_actor_udp("localhost", *port);
CAF_REQUIRE(other_srv);
// Establish communication and exit.
CAF_MESSAGE("client contacts server and exits");
anon_send(other, *other_srv);
other_sys.await_all_actors_done();
}
// Start communicate from the first actor.
CAF_MESSAGE("first client contacts server and exits");
anon_send(client, *client_srv);
client_sys.await_all_actors_done();
anon_send_exit(mirror, exit_reason::user_shutdown);
}
CAF_TEST_FIXTURE_SCOPE_END()
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