Commit 930d232a authored by Jakob Otto's avatar Jakob Otto

Merge branch master

parents df629f20 b156634c
......@@ -5,23 +5,43 @@ project(caf_net C CXX)
# e.g., for creating proper Xcode projects
file(GLOB_RECURSE LIBCAF_NET_HDRS "caf/*.hpp")
enum_to_string("caf/net/basp/connection_state.hpp" "basp_conn_strings.cpp")
enum_to_string("caf/net/basp/ec.hpp" "basp_ec_strings.cpp")
enum_to_string("caf/net/basp/message_type.hpp" "basp_message_type_strings.cpp")
enum_to_string("caf/net/operation.hpp" "operation_strings.cpp")
# list cpp files excluding platform-dependent files
set(LIBCAF_NET_SRCS
"${CMAKE_CURRENT_BINARY_DIR}/basp_conn_strings.cpp"
"${CMAKE_CURRENT_BINARY_DIR}/basp_ec_strings.cpp"
"${CMAKE_CURRENT_BINARY_DIR}/basp_message_type_strings.cpp"
"${CMAKE_CURRENT_BINARY_DIR}/operation_strings.cpp"
src/actor_proxy_impl.cpp
src/application.cpp
src/convert_ip_endpoint.cpp
src/datagram_socket.cpp
src/defaults.cpp
src/ec.cpp
src/endpoint_manager.cpp
src/header.cpp
src/host.cpp
src/ip.cpp
src/multiplexer.cpp
src/net/backend/test.cpp
src/net/endpoint_manager_queue.cpp
src/net/middleman.cpp
src/net/middleman_backend.cpp
src/net/packet_writer.cpp
src/network_socket.cpp
src/pipe_socket.cpp
src/pollset_updater.cpp
src/socket.cpp
src/socket_manager.cpp
src/stream_socket.cpp
src/scribe.cpp
src/tcp_accept_socket.cpp
src/tcp_stream_socket.cpp
src/udp_datagram_socket.cpp
src/defaults.cpp
src/message_queue.cpp
src/worker.cpp
)
add_custom_target(libcaf_net)
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/fwd.hpp"
namespace caf::detail {
void convert(const ip_endpoint& src, sockaddr_storage& dst);
error convert(const sockaddr_storage& src, ip_endpoint& dst);
} // namespace caf::detail
......@@ -20,8 +20,7 @@
#include "caf/detail/socket_sys_includes.hpp"
namespace caf {
namespace detail {
namespace caf::detail {
inline auto addr_of(sockaddr_in& what) -> decltype(what.sin_addr)& {
return what.sin_addr;
......@@ -47,5 +46,4 @@ inline auto port_of(sockaddr_in6& what) -> decltype(what.sin6_port)& {
return what.sin6_port;
}
} // namespace detail
} // namespace caf
} // namespace caf::detail
......@@ -20,8 +20,7 @@
#include "caf/config.hpp"
namespace caf {
namespace net {
namespace caf::net {
#ifdef CAF_WINDOWS
......@@ -41,5 +40,4 @@ using socket_size_type = unsigned;
#endif // CAF_WINDOWS
} // namespace net
} // namespace caf
} // namespace caf::net
......@@ -52,5 +52,6 @@
# include <netinet/tcp.h>
# include <sys/socket.h>
# include <unistd.h>
# include <sys/uio.h>
#endif
// clang-format on
......@@ -21,8 +21,7 @@
#include "caf/actor_proxy.hpp"
#include "caf/net/endpoint_manager.hpp"
namespace caf {
namespace net {
namespace caf::net {
/// Implements a simple proxy forwarding all operations to a manager.
class actor_proxy_impl : public actor_proxy {
......@@ -35,10 +34,6 @@ public:
void enqueue(mailbox_element_ptr what, execution_unit* context) override;
bool add_backlink(abstract_actor* x) override;
bool remove_backlink(abstract_actor* x) override;
void kill_proxy(execution_unit* ctx, error rsn) override;
private:
......@@ -46,5 +41,4 @@ private:
endpoint_manager_ptr dst_;
};
} // namespace net
} // namespace caf
} // namespace caf::net
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/datagram_socket.hpp"
#include "caf/net/datagram_transport.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/endpoint_manager_impl.hpp"
#include "caf/net/endpoint_manager_queue.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/host.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/network_socket.hpp"
#include "caf/net/operation.hpp"
#include "caf/net/packet_writer.hpp"
#include "caf/net/packet_writer_decorator.hpp"
#include "caf/net/pipe_socket.hpp"
#include "caf/net/pollset_updater.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/socket_id.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/net/transport_base.hpp"
#include "caf/net/transport_worker.hpp"
#include "caf/net/transport_worker_dispatcher.hpp"
#include "caf/net/udp_datagram_socket.hpp"
/******************************************************************************_opt
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include <map>
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/node_id.hpp"
namespace caf::net::backend {
/// Minimal backend for unit testing.
/// @warning this backend is *not* thread safe.
class test : public middleman_backend {
public:
// -- member types -----------------------------------------------------------
using peer_entry = std::pair<stream_socket, endpoint_manager_ptr>;
// -- constructors, destructors, and assignment operators --------------------
test(middleman& mm);
~test() override;
// -- interface functions ----------------------------------------------------
error init() override;
endpoint_manager_ptr peer(const node_id& id) override;
void resolve(const uri& locator, const actor& listener) override;
strong_actor_ptr make_proxy(node_id nid, actor_id aid) override;
void set_last_hop(node_id*) override;
// -- properties -------------------------------------------------------------
stream_socket socket(const node_id& peer_id) {
return get_peer(peer_id).first;
}
peer_entry& emplace(const node_id& peer_id, stream_socket first,
stream_socket second);
private:
peer_entry& get_peer(const node_id& id);
middleman& mm_;
std::map<node_id, peer_entry> peers_;
proxy_registry proxies_;
};
} // namespace caf::net::backend
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include <cstdint>
#include <memory>
#include <type_traits>
#include <unordered_map>
#include <unordered_set>
#include <vector>
#include "caf/actor_addr.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/byte.hpp"
#include "caf/callback.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/worker_hub.hpp"
#include "caf/error.hpp"
#include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/header.hpp"
#include "caf/net/basp/message_queue.hpp"
#include "caf/net/basp/message_type.hpp"
#include "caf/net/basp/worker.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/packet_writer.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/node_id.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/response_promise.hpp"
#include "caf/scoped_execution_unit.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/span.hpp"
#include "caf/unit.hpp"
namespace caf::net::basp {
/// An implementation of BASP as an application layer protocol.
class application {
public:
// -- member types -----------------------------------------------------------
using buffer_type = std::vector<byte>;
using byte_span = span<const byte>;
using hub_type = detail::worker_hub<worker>;
struct test_tag {};
// -- constructors, destructors, and assignment operators --------------------
explicit application(proxy_registry& proxies);
// -- interface functions ----------------------------------------------------
template <class Parent>
error init(Parent& parent) {
// Initialize member variables.
system_ = &parent.system();
executor_.system_ptr(system_);
executor_.proxy_registry_ptr(&proxies_);
// TODO: use `if constexpr` when switching to C++17.
// Allow unit tests to run the application without endpoint manager.
if (!std::is_base_of<test_tag, Parent>::value)
manager_ = &parent.manager();
auto workers = get_or(system_->config(), "middleman.workers",
defaults::middleman::workers);
for (size_t i = 0; i < workers; ++i)
hub_->add_new_worker(*queue_, proxies_);
// Write handshake.
auto hdr = parent.next_header_buffer();
auto payload = parent.next_payload_buffer();
if (auto err = generate_handshake(payload))
return err;
to_bytes(header{message_type::handshake,
static_cast<uint32_t>(payload.size()), version},
hdr);
parent.write_packet(hdr, payload);
parent.transport().configure_read(receive_policy::exactly(header_size));
return none;
}
error write_message(packet_writer& writer,
std::unique_ptr<endpoint_manager_queue::message> ptr);
template <class Parent>
error handle_data(Parent& parent, byte_span bytes) {
static_assert(std::is_base_of<packet_writer, Parent>::value,
"parent must implement packet_writer");
size_t next_read_size = header_size;
if (auto err = handle(next_read_size, parent, bytes))
return err;
parent.transport().configure_read(receive_policy::exactly(next_read_size));
return none;
}
void resolve(packet_writer& writer, string_view path, const actor& listener);
static void new_proxy(packet_writer& writer, actor_id id);
void local_actor_down(packet_writer& writer, actor_id id, error reason);
template <class Parent>
void timeout(Parent&, atom_value, uint64_t) {
// nop
}
void handle_error(sec) {
// nop
}
static expected<buffer_type> serialize(actor_system& sys,
const type_erased_tuple& x);
// -- utility functions ------------------------------------------------------
strong_actor_ptr resolve_local_path(string_view path);
// -- properties -------------------------------------------------------------
connection_state state() const noexcept {
return state_;
}
actor_system& system() const noexcept {
return *system_;
}
private:
// -- handling of incoming messages ------------------------------------------
error handle(size_t& next_read_size, packet_writer& writer, byte_span bytes);
error handle(packet_writer& writer, header hdr, byte_span payload);
error handle_handshake(packet_writer& writer, header hdr, byte_span payload);
error handle_actor_message(packet_writer& writer, header hdr,
byte_span payload);
error handle_resolve_request(packet_writer& writer, header rec_hdr,
byte_span received);
error handle_resolve_response(packet_writer& writer, header received_hdr,
byte_span received);
error handle_monitor_message(packet_writer& writer, header received_hdr,
byte_span received);
error handle_down_message(packet_writer& writer, header received_hdr,
byte_span received);
/// Writes the handshake payload to `buf_`.
error generate_handshake(buffer_type& buf);
// -- member variables -------------------------------------------------------
/// Stores a pointer to the parent actor system.
actor_system* system_ = nullptr;
/// Stores the expected type of the next incoming message.
connection_state state_ = connection_state::await_handshake_header;
/// Caches the last header while waiting for the matching payload.
header hdr_;
/// Stores the ID of our peer.
node_id peer_id_;
/// Tracks which local actors our peer monitors.
std::unordered_set<actor_addr> monitored_actors_; // TODO: this is unused
/// Caches actor handles obtained via `resolve`.
std::unordered_map<uint64_t, actor> pending_resolves_;
/// Ascending ID generator for requests to our peer.
uint64_t next_request_id_ = 1;
/// Points to the factory object for generating proxies.
proxy_registry& proxies_;
/// Points to the endpoint manager that owns this applications.
endpoint_manager* manager_ = nullptr;
/// Provides pointers to the actor system as well as the registry,
/// serializers and deserializer.
scoped_execution_unit executor_;
std::unique_ptr<message_queue> queue_;
std::unique_ptr<hub_type> hub_;
};
} // namespace caf::net::basp
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include <string>
namespace caf::net::basp {
/// @addtogroup BASP
/// Stores the state of a connection in a `basp::application`.
enum class connection_state {
/// Initial state for any connection to wait for the peer's handshake.
await_handshake_header,
/// Indicates that the header for the peer's handshake arrived and BASP
/// requires the payload next.
await_handshake_payload,
/// Indicates that a connection is established and this node is waiting for
/// the next BASP header.
await_header,
/// Indicates that this node has received a header with non-zero payload and
/// is waiting for the data.
await_payload,
/// Indicates that the connection is about to shut down.
shutdown,
};
/// @relates connection_state
std::string to_string(connection_state x);
/// @}
} // namespace caf::net::basp
......@@ -19,51 +19,19 @@
#pragma once
#include <cstddef>
#include <limits>
#include <cstdint>
#include "caf/net/socket_id.hpp"
namespace caf::net::basp {
namespace caf {
namespace net {
/// @addtogroup BASP
template <class Derived>
struct abstract_socket {
socket_id id;
/// The current BASP version.
/// @note BASP is not backwards compatible.
constexpr uint64_t version = 1;
constexpr abstract_socket() : id(invalid_socket_id) {
// nop
}
/// Size of a BASP header in serialized form.
constexpr size_t header_size = 13;
constexpr abstract_socket(socket_id id) : id(id) {
// nop
}
/// @}
constexpr abstract_socket(const Derived& other) : id(other.id) {
// nop
}
abstract_socket& operator=(const Derived& other) {
id = other.id;
return *this;
}
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f, Derived& x) {
return f(x.id);
}
friend constexpr bool operator==(Derived x, Derived y) {
return x.id == y.id;
}
friend constexpr bool operator!=(Derived x, Derived y) {
return x.id != y.id;
}
friend constexpr bool operator<(Derived x, Derived y) {
return x.id < y.id;
}
};
} // namespace net
} // namespace caf
} // namespace caf::net::basp
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include <cstdint>
#include <string>
#include "caf/fwd.hpp"
namespace caf::net::basp {
/// BASP-specific error codes.
enum class ec : uint8_t {
invalid_magic_number = 1,
unexpected_number_of_bytes,
unexpected_payload,
missing_payload,
illegal_state,
invalid_handshake,
missing_handshake,
unexpected_handshake,
version_mismatch,
unimplemented = 10,
app_identifiers_mismatch,
invalid_payload,
invalid_scheme,
invalid_locator,
};
/// @relates ec
std::string to_string(ec x);
/// @relates ec
error make_error(ec x);
} // namespace caf::net::basp
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include <array>
#include <cstdint>
#include "caf/detail/comparable.hpp"
#include "caf/fwd.hpp"
#include "caf/meta/hex_formatted.hpp"
#include "caf/meta/type_name.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/message_type.hpp"
namespace caf::net::basp {
/// @addtogroup BASP
/// The header of a Binary Actor System Protocol (BASP) message.
struct header : detail::comparable<header> {
// -- constructors, destructors, and assignment operators --------------------
constexpr header() noexcept
: type(message_type::handshake), payload_len(0), operation_data(0) {
// nop
}
constexpr header(message_type type, uint32_t payload_len,
uint64_t operation_data) noexcept
: type(type), payload_len(payload_len), operation_data(operation_data) {
// nop
}
header(const header&) noexcept = default;
header& operator=(const header&) noexcept = default;
// -- factory functions ------------------------------------------------------
/// @pre `bytes.size() == header_size`
static header from_bytes(span<const byte> bytes);
// -- comparison -------------------------------------------------------------
int compare(header other) const noexcept;
// -- member variables -------------------------------------------------------
/// Denotes the BASP operation and how `operation_data` gets interpreted.
message_type type;
/// Stores the size in bytes for the payload that follows this header.
uint32_t payload_len;
/// Stores type-specific information such as the BASP version in handshakes.
uint64_t operation_data;
};
/// Serializes a header to a byte representation.
/// @relates header
std::array<byte, header_size> to_bytes(header x);
/// Serializes a header to a byte representation.
/// @relates header
void to_bytes(header x, std::vector<byte>& buf);
/// @relates header
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, header& x) {
return f(meta::type_name("basp::header"), x.type, x.payload_len,
x.operation_data);
}
/// @}
} // namespace caf::net::basp
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include <cstdint>
#include <mutex>
#include <vector>
#include "caf/actor_control_block.hpp"
#include "caf/fwd.hpp"
#include "caf/mailbox_element.hpp"
namespace caf::net::basp {
/// Enforces strict order of message delivery, i.e., deliver messages in the
/// same order as if they were deserialized by a single thread.
class message_queue {
public:
// -- member types -----------------------------------------------------------
/// Request for sending a message to an actor at a later time.
struct actor_msg {
uint64_t id;
strong_actor_ptr receiver;
mailbox_element_ptr content;
};
// -- constructors, destructors, and assignment operators --------------------
message_queue();
// -- mutators ---------------------------------------------------------------
/// Adds a new message to the queue or deliver it immediately if possible.
void push(execution_unit* ctx, uint64_t id, strong_actor_ptr receiver,
mailbox_element_ptr content);
/// Marks given ID as dropped, effectively skipping it without effect.
void drop(execution_unit* ctx, uint64_t id);
/// Returns the next ascending ID.
uint64_t new_id();
// -- member variables -------------------------------------------------------
/// Protects all other properties.
std::mutex lock;
/// The next available ascending ID. The counter is large enough to overflow
/// after roughly 600 years if we dispatch a message every microsecond.
uint64_t next_id;
/// The next ID that we can ship.
uint64_t next_undelivered;
/// Keeps messages in sorted order in case a message other than
/// `next_undelivered` gets ready first.
std::vector<actor_msg> pending;
};
} // namespace caf::net::basp
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include <cstdint>
#include <string>
namespace caf::net::basp {
/// @addtogroup BASP
/// Describes the first header field of a BASP message and determines the
/// interpretation of the other header fields.
enum class message_type : uint8_t {
/// Sends supported BASP version and node information to the server.
///
/// ![](client_handshake.png)
handshake = 0,
/// Transmits an actor-to-actor messages.
///
/// ![](direct_message.png)
actor_message = 1,
/// Tries to resolve a path on the receiving node.
///
/// ![](resolve_request.png)
resolve_request = 2,
/// Transmits the result of a path lookup.
///
/// ![](resolve_response.png)
resolve_response = 3,
/// Informs the receiving node that the sending node has created a proxy
/// instance for one of its actors. Causes the receiving node to attach a
/// functor to the actor that triggers a down_message on termination.
///
/// ![](monitor_message.png)
monitor_message = 4,
/// Informs the receiving node that it has a proxy for an actor that has been
/// terminated.
///
/// ![](down_message.png)
down_message = 5,
/// Used to generate periodic traffic between two nodes in order to detect
/// disconnects.
///
/// ![](heartbeat.png)
heartbeat = 6,
};
/// @relates message_type
std::string to_string(message_type);
/// @}
} // namespace caf::net::basp
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include <vector>
#include "caf/actor_control_block.hpp"
#include "caf/actor_proxy.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/config.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/execution_unit.hpp"
#include "caf/logger.hpp"
#include "caf/message.hpp"
#include "caf/message_id.hpp"
#include "caf/net/basp/header.hpp"
#include "caf/node_id.hpp"
namespace caf::net::basp {
template <class Subtype>
class remote_message_handler {
public:
void handle_remote_message(execution_unit* ctx) {
// Local variables.
auto& dref = static_cast<Subtype&>(*this);
auto& payload = dref.payload_;
auto& hdr = dref.hdr_;
auto& registry = dref.system_->registry();
auto& proxies = *dref.proxies_;
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
// Deserialize payload.
actor_id src_id = 0;
node_id src_node;
actor_id dst_id = 0;
std::vector<strong_actor_ptr> fwd_stack;
message content;
binary_deserializer source{ctx, payload};
if (auto err = source(src_node, src_id, dst_id, fwd_stack, content)) {
CAF_LOG_ERROR("could not deserialize payload: " << CAF_ARG(err));
return;
}
// Sanity checks.
if (dst_id == 0)
return;
// Try to fetch the receiver.
auto dst_hdl = registry.get(dst_id);
if (dst_hdl == nullptr) {
CAF_LOG_DEBUG("no actor found for given ID, drop message");
return;
}
// Try to fetch the sender.
strong_actor_ptr src_hdl;
if (src_node != none && src_id != 0)
src_hdl = proxies.get_or_put(src_node, src_id);
// Ship the message.
auto ptr = make_mailbox_element(std::move(src_hdl),
make_message_id(hdr.operation_data),
std::move(fwd_stack), std::move(content));
dst_hdl->get()->enqueue(std::move(ptr), nullptr);
}
};
} // namespace caf::net::basp
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include <atomic>
#include <cstdint>
#include <vector>
#include "caf/config.hpp"
#include "caf/detail/abstract_worker.hpp"
#include "caf/detail/worker_hub.hpp"
#include "caf/net/basp/header.hpp"
#include "caf/net/basp/message_queue.hpp"
#include "caf/net/basp/remote_message_handler.hpp"
#include "caf/net/fwd.hpp"
#include "caf/node_id.hpp"
#include "caf/resumable.hpp"
namespace caf::net::basp {
/// Deserializes payloads for BASP messages asynchronously.
class worker : public detail::abstract_worker,
public remote_message_handler<worker> {
public:
// -- friends ----------------------------------------------------------------
friend remote_message_handler<worker>;
// -- member types -----------------------------------------------------------
using super = detail::abstract_worker;
using scheduler_type = scheduler::abstract_coordinator;
using buffer_type = std::vector<byte>;
using hub_type = detail::worker_hub<worker>;
// -- constructors, destructors, and assignment operators --------------------
/// Only the ::worker_hub has access to the construtor.
worker(hub_type& hub, message_queue& queue, proxy_registry& proxies);
~worker() override;
// -- management -------------------------------------------------------------
void launch(const node_id& last_hop, const basp::header& hdr,
span<const byte> payload);
// -- implementation of resumable --------------------------------------------
resume_result resume(execution_unit* ctx, size_t) override;
private:
// -- constants and assertions -----------------------------------------------
/// Stores how many bytes the "first half" of this object requires.
static constexpr size_t pointer_members_size = sizeof(hub_type*)
+ sizeof(message_queue*)
+ sizeof(proxy_registry*)
+ sizeof(actor_system*);
static_assert(CAF_CACHE_LINE_SIZE > pointer_members_size,
"invalid cache line size");
// -- member variables -------------------------------------------------------
/// Points to our home hub.
hub_type* hub_;
/// Points to the queue for establishing strict ordering.
message_queue* queue_;
/// Points to our proxy registry / factory.
proxy_registry* proxies_;
/// Points to the parent system.
actor_system* system_;
/// Prevents false sharing when writing to `next`.
char pad_[CAF_CACHE_LINE_SIZE - pointer_members_size];
/// ID for local ordering.
uint64_t msg_id_;
/// Identifies the node that sent us `hdr_` and `payload_`.
node_id last_hop_;
/// The header for the next message. Either a direct_message or a
/// routed_message.
header hdr_;
/// Contains whatever this worker deserializes next.
buffer_type payload_;
};
} // namespace caf::net::basp
......@@ -21,22 +21,13 @@
#include "caf/net/network_socket.hpp"
#include "caf/variant.hpp"
namespace caf {
namespace net {
namespace caf::net {
/// A datagram-oriented network communication endpoint.
struct datagram_socket : abstract_socket<datagram_socket> {
using super = abstract_socket<datagram_socket>;
struct datagram_socket : network_socket {
using super = network_socket;
using super::super;
constexpr operator socket() const noexcept {
return socket{id};
}
constexpr operator network_socket() const noexcept {
return network_socket{id};
}
};
/// Enables or disables `SIO_UDP_CONNRESET` error on `x`.
......@@ -49,5 +40,4 @@ error allow_connreset(datagram_socket x, bool new_value);
variant<size_t, sec>
check_datagram_socket_io_res(std::make_signed<size_t>::type res);
} // namespace net
} // namespace caf
} // namespace caf::net
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include <deque>
#include <vector>
#include "caf/fwd.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/logger.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/transport_base.hpp"
#include "caf/net/transport_worker_dispatcher.hpp"
#include "caf/net/udp_datagram_socket.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
namespace caf::net {
template <class Factory>
using datagram_transport_base = transport_base<
datagram_transport<Factory>,
transport_worker_dispatcher<Factory, ip_endpoint>, udp_datagram_socket,
Factory, ip_endpoint>;
/// Implements a udp_transport policy that manages a datagram socket.
template <class Factory>
class datagram_transport : public datagram_transport_base<Factory> {
public:
// Maximal UDP-packet size
static constexpr size_t max_datagram_size = std::numeric_limits<
uint16_t>::max();
// -- member types -----------------------------------------------------------
using factory_type = Factory;
using id_type = ip_endpoint;
using application_type = typename factory_type::application_type;
using super = datagram_transport_base<factory_type>;
using buffer_type = typename super::buffer_type;
using buffer_cache_type = typename super::buffer_cache_type;
// -- constructors, destructors, and assignment operators --------------------
datagram_transport(udp_datagram_socket handle, factory_type factory)
: super(handle, std::move(factory)) {
// nop
}
// -- public member functions ------------------------------------------------
error init(endpoint_manager& manager) override {
CAF_LOG_TRACE("");
if (auto err = super::init(manager))
return err;
prepare_next_read();
return none;
}
bool handle_read_event(endpoint_manager&) override {
CAF_LOG_TRACE(CAF_ARG(this->handle_.id));
auto ret = read(this->handle_, make_span(this->read_buf_));
if (auto res = get_if<std::pair<size_t, ip_endpoint>>(&ret)) {
auto num_bytes = res->first;
CAF_LOG_DEBUG("received " << num_bytes << " bytes");
auto ep = res->second;
this->read_buf_.resize(num_bytes);
this->next_layer_.handle_data(*this, make_span(this->read_buf_),
std::move(ep));
prepare_next_read();
} else {
auto err = get<sec>(ret);
CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
return false;
}
return true;
}
bool handle_write_event(endpoint_manager& manager) override {
CAF_LOG_TRACE(CAF_ARG(this->handle_.id)
<< CAF_ARG2("queue-size", packet_queue_.size()));
// Try to write leftover data.
write_some();
// Get new data from parent.
for (auto msg = manager.next_message(); msg != nullptr;
msg = manager.next_message()) {
this->next_layer_.write_message(*this, std::move(msg));
}
// Write prepared data.
return write_some();
}
// TODO: remove this function. `resolve` should add workers when needed.
error add_new_worker(node_id node, id_type id) {
auto worker = this->next_layer_.add_new_worker(*this, node, id);
if (!worker)
return worker.error();
return none;
}
void write_packet(id_type id, span<buffer_type*> buffers) override {
CAF_LOG_TRACE("");
CAF_ASSERT(!buffers.empty());
if (packet_queue_.empty())
this->manager().register_writing();
// By convention, the first buffer is a header buffer. Every other buffer is
// a payload buffer.
packet_queue_.emplace_back(id, buffers);
}
/// Helper struct for managing outgoing packets
struct packet {
id_type id;
buffer_cache_type bytes;
size_t size;
packet(id_type id, span<buffer_type*> bufs) : id(id) {
size = 0;
for (auto buf : bufs) {
size += buf->size();
bytes.emplace_back(std::move(*buf));
}
}
std::vector<std::vector<byte>*> get_buffer_ptrs() {
std::vector<std::vector<byte>*> ptrs;
for (auto& buf : bytes)
ptrs.emplace_back(&buf);
return ptrs;
}
};
private:
// -- utility functions ------------------------------------------------------
void prepare_next_read() {
this->read_buf_.resize(max_datagram_size);
}
bool write_some() {
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle_.id));
// Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() {
auto& front = packet_queue_.front();
auto& bufs = front.bytes;
auto it = bufs.begin();
if (this->header_bufs_.size() < this->header_bufs_.capacity()) {
it->clear();
this->header_bufs_.emplace_back(std::move(*it++));
}
for (; it != bufs.end()
&& this->payload_bufs_.size() < this->payload_bufs_.capacity();
++it) {
it->clear();
this->payload_bufs_.emplace_back(std::move(*it));
}
packet_queue_.pop_front();
};
// Write as many bytes as possible.
while (!packet_queue_.empty()) {
auto& packet = packet_queue_.front();
auto ptrs = packet.get_buffer_ptrs();
auto write_ret = write(this->handle_, make_span(ptrs), packet.id);
if (auto num_bytes = get_if<size_t>(&write_ret)) {
CAF_LOG_DEBUG(CAF_ARG(this->handle_.id) << CAF_ARG(*num_bytes));
CAF_LOG_WARNING_IF(*num_bytes < packet.size,
"packet was not sent completely");
recycle();
} else {
auto err = get<sec>(write_ret);
if (err != sec::unavailable_or_would_block) {
CAF_LOG_ERROR("write failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
return false;
}
CAF_LOG_DEBUG("write returned `unavailable_or_would_block`");
return true;
}
}
return false;
}
std::deque<packet> packet_queue_;
};
} // namespace caf::net
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#pragma once
#include <cstddef>
// -- hard-coded default values for various CAF options ------------------------
namespace caf::defaults::middleman {
/// Maximum number of cached buffers for sending payloads.
extern const size_t max_payload_buffers;
/// Maximum number of cached buffers for sending headers.
extern const size_t max_header_buffers;
} // namespace caf::defaults::middleman
......@@ -25,13 +25,13 @@
#include "caf/net/stream_socket.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/policy/scribe.hpp"
#include "caf/send.hpp"
#include "caf/stream_transport.hpp"
namespace caf {
namespace policy {
/// A doorman accepts TCP connections and creates scribes to handle them.
/// A doorman accepts TCP connections and creates stream_transports to handle
/// them.
class doorman {
public:
doorman(net::tcp_accept_socket acceptor) : acceptor_(acceptor) {
......@@ -96,5 +96,4 @@ public:
}
};
} // namespace policy
} // namespace caf
......@@ -23,17 +23,18 @@
#include <memory>
#include "caf/actor.hpp"
#include "caf/actor_clock.hpp"
#include "caf/byte.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/net/endpoint_manager_queue.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/variant.hpp"
namespace caf {
namespace net {
namespace caf::net {
/// Manages a communication endpoint.
class endpoint_manager : public socket_manager {
......@@ -49,71 +50,6 @@ public:
using serialize_fun_type = maybe_buffer (*)(actor_system&,
const type_erased_tuple&);
struct event : intrusive::singly_linked<event> {
struct resolve_request {
std::string path;
actor listener;
};
struct timeout {
atom_value type;
uint64_t id;
};
event(std::string path, actor listener);
event(atom_value type, uint64_t id);
/// Either contains a string for `resolve` requests or an `atom_value`
variant<resolve_request, timeout> value;
};
struct event_policy {
using deficit_type = size_t;
using task_size_type = size_t;
using mapped_type = event;
using unique_pointer = std::unique_ptr<event>;
using queue_type = intrusive::drr_queue<event_policy>;
task_size_type task_size(const event&) const noexcept {
return 1;
}
};
using event_queue_type = intrusive::fifo_inbox<event_policy>;
struct message : intrusive::singly_linked<message> {
/// Original message to a remote actor.
mailbox_element_ptr msg;
/// Serialized representation of of `msg->content()`.
std::vector<byte> payload;
message(mailbox_element_ptr msg, std::vector<byte> payload);
};
struct message_policy {
using deficit_type = size_t;
using task_size_type = size_t;
using mapped_type = message;
using unique_pointer = std::unique_ptr<message>;
using queue_type = intrusive::drr_queue<message_policy>;
task_size_type task_size(const message& x) const noexcept {
return x.payload.size();
}
};
using message_queue_type = intrusive::fifo_inbox<message_policy>;
// -- constructors, destructors, and assignment operators --------------------
endpoint_manager(socket handle, const multiplexer_ptr& parent,
......@@ -127,15 +63,22 @@ public:
return sys_;
}
std::unique_ptr<message> next_message();
endpoint_manager_queue::message_ptr next_message();
// -- event management -------------------------------------------------------
/// Resolves a path to a remote actor.
void resolve(std::string path, actor listener);
void resolve(uri locator, actor listener);
/// Enqueues a message to the endpoint.
void enqueue(mailbox_element_ptr msg, std::vector<byte> payload);
void enqueue(mailbox_element_ptr msg, strong_actor_ptr receiver,
std::vector<byte> payload);
/// Enqueues an event to the endpoint.
template <class... Ts>
void enqueue_event(Ts&&... xs) {
enqueue(new endpoint_manager_queue::event(std::forward<Ts>(xs)...));
}
// -- pure virtual member functions ------------------------------------------
......@@ -146,17 +89,18 @@ public:
virtual serialize_fun_type serialize_fun() const noexcept = 0;
protected:
bool enqueue(endpoint_manager_queue::element* ptr);
/// Points to the hosting actor system.
actor_system& sys_;
/// Stores control events.
event_queue_type events_;
/// Stores control events and outbound messages.
endpoint_manager_queue::type queue_;
/// Stores outbound messages.
message_queue_type messages_;
/// Stores a proxy for interacting with the actor clock.
actor timeout_proxy_;
};
using endpoint_manager_ptr = intrusive_ptr<endpoint_manager>;
} // namespace net
} // namespace caf
} // namespace caf::net
......@@ -18,12 +18,16 @@
#pragma once
#include "caf/abstract_actor.hpp"
#include "caf/actor_cast.hpp"
#include "caf/actor_system.hpp"
#include "caf/atom.hpp"
#include "caf/detail/overload.hpp"
#include "caf/net/endpoint_manager.hpp"
namespace caf {
namespace net {
namespace caf::net {
template <class Transport, class Application>
template <class Transport>
class endpoint_manager_impl : public endpoint_manager {
public:
// -- member types -----------------------------------------------------------
......@@ -32,15 +36,13 @@ public:
using transport_type = Transport;
using application_type = Application;
using application_type = typename transport_type::application_type;
// -- constructors, destructors, and assignment operators --------------------
endpoint_manager_impl(const multiplexer_ptr& parent, actor_system& sys,
Transport trans, Application app)
: super(trans.handle(), parent, sys),
transport_(std::move(trans)),
application_(std::move(app)) {
Transport trans)
: super(trans.handle(), parent, sys), transport_(std::move(trans)) {
// nop
}
......@@ -54,13 +56,26 @@ public:
return transport_;
}
application_type& application() {
return application_;
endpoint_manager_impl& manager() {
return *this;
}
// -- timeout management -----------------------------------------------------
template <class... Ts>
uint64_t set_timeout(actor_clock::time_point tp, atom_value type,
Ts&&... xs) {
auto act = actor_cast<abstract_actor*>(timeout_proxy_);
CAF_ASSERT(act != nullptr);
sys_.clock().set_multi_timeout(tp, act, type, next_timeout_id_);
transport_.set_timeout(next_timeout_id_, std::forward<Ts>(xs)...);
return next_timeout_id_++;
}
// -- interface functions ----------------------------------------------------
error init() override {
this->register_reading();
return transport_.init(*this);
}
......@@ -69,29 +84,40 @@ public:
}
bool handle_write_event() override {
if (!this->events_.blocked() && !this->events_.empty()) {
if (!this->queue_.blocked()) {
this->queue_.fetch_more();
auto& q = std::get<0>(this->queue_.queue().queues());
do {
this->events_.fetch_more();
auto& q = this->events_.queue();
q.inc_deficit(q.total_task_size());
for (auto ptr = q.next(); ptr != nullptr; ptr = q.next()) {
using timeout = endpoint_manager::event::timeout;
using resolve_request = endpoint_manager::event::resolve_request;
if (auto rr = get_if<resolve_request>(&ptr->value)) {
transport_.resolve(*this, std::move(rr->path),
std::move(rr->listener));
} else {
auto& t = get<timeout>(ptr->value);
transport_.timeout(*this, t.type, t.id);
}
auto f = detail::make_overload(
[&](endpoint_manager_queue::event::resolve_request& x) {
transport_.resolve(*this, x.locator, x.listener);
},
[&](endpoint_manager_queue::event::new_proxy& x) {
transport_.new_proxy(*this, x.peer, x.id);
},
[&](endpoint_manager_queue::event::local_actor_down& x) {
transport_.local_actor_down(*this, x.observing_peer, x.id,
std::move(x.reason));
},
[&](endpoint_manager_queue::event::timeout& x) {
transport_.timeout(*this, x.type, x.id);
});
visit(f, ptr->value);
}
} while (!this->events_.try_block());
} while (!q.empty());
}
return transport_.handle_write_event(*this);
if (!transport_.handle_write_event(*this)) {
if (this->queue_.blocked())
return false;
return !(this->queue_.empty() && this->queue_.try_block());
}
return true;
}
void handle_error(sec code) override {
transport_.handle_error(application_, code);
transport_.handle_error(code);
}
serialize_fun_type serialize_fun() const noexcept override {
......@@ -100,8 +126,9 @@ public:
private:
transport_type transport_;
application_type application_;
/// Stores the id for the next timeout.
uint64_t next_timeout_id_;
};
} // namespace net
} // namespace caf
} // namespace caf::net
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include "caf/actor.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "caf/intrusive/wdrr_fixed_multiplexed_queue.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/unit.hpp"
#include "caf/variant.hpp"
namespace caf::net {
class endpoint_manager_queue {
public:
enum class element_type { event, message };
class element : public intrusive::singly_linked<element> {
public:
explicit element(element_type tag) : tag_(tag) {
// nop
}
virtual ~element();
virtual size_t task_size() const noexcept = 0;
element_type tag() const noexcept {
return tag_;
}
private:
element_type tag_;
};
using element_ptr = std::unique_ptr<element>;
class event final : public element {
public:
struct resolve_request {
uri locator;
actor listener;
};
struct new_proxy {
node_id peer;
actor_id id;
};
struct local_actor_down {
node_id observing_peer;
actor_id id;
error reason;
};
struct timeout {
atom_value type;
uint64_t id;
};
event(uri locator, actor listener);
event(node_id peer, actor_id proxy_id);
event(node_id observing_peer, actor_id local_actor_id, error reason);
event(atom_value type, uint64_t id);
~event() override;
size_t task_size() const noexcept override;
/// Holds the event data.
variant<resolve_request, new_proxy, local_actor_down, timeout> value;
};
using event_ptr = std::unique_ptr<event>;
struct event_policy {
using deficit_type = size_t;
using task_size_type = size_t;
using mapped_type = event;
using unique_pointer = event_ptr;
using queue_type = intrusive::drr_queue<event_policy>;
constexpr event_policy(unit_t) {
// nop
}
task_size_type task_size(const event&) const noexcept {
return 1;
}
};
class message : public element {
public:
/// Original message to a remote actor.
mailbox_element_ptr msg;
/// ID of the receiving actor.
strong_actor_ptr receiver;
/// Serialized representation of of `msg->content()`.
std::vector<byte> payload;
message(mailbox_element_ptr msg, strong_actor_ptr receiver,
std::vector<byte> payload);
~message() override;
size_t task_size() const noexcept override;
};
using message_ptr = std::unique_ptr<message>;
struct message_policy {
using deficit_type = size_t;
using task_size_type = size_t;
using mapped_type = message;
using unique_pointer = message_ptr;
using queue_type = intrusive::drr_queue<message_policy>;
constexpr message_policy(unit_t) {
// nop
}
static task_size_type task_size(const message& x) noexcept {
// Return at least 1 if the payload is empty.
return x.payload.size() + static_cast<task_size_type>(x.payload.empty());
}
};
struct categorized {
using deficit_type = size_t;
using task_size_type = size_t;
using mapped_type = element;
using unique_pointer = element_ptr;
constexpr categorized(unit_t) {
// nop
}
template <class Queue>
deficit_type quantum(const Queue&, deficit_type x) const noexcept {
return x;
}
size_t id_of(const element& x) const noexcept {
return static_cast<size_t>(x.tag());
}
};
struct policy {
using deficit_type = size_t;
using task_size_type = size_t;
using mapped_type = element;
using unique_pointer = std::unique_ptr<element>;
using queue_type = intrusive::wdrr_fixed_multiplexed_queue<
categorized, event_policy::queue_type, message_policy::queue_type>;
task_size_type task_size(const message& x) const noexcept {
return x.task_size();
}
};
using type = intrusive::fifo_inbox<policy>;
};
} // namespace caf::net
......@@ -22,22 +22,47 @@
#include "caf/intrusive_ptr.hpp"
namespace caf {
namespace net {
namespace caf::net {
// -- templates ----------------------------------------------------------------
template <class Application>
class stream_transport;
template <class Factory>
class datagram_transport;
template <class Application, class IdType = unit_t>
class transport_worker;
template <class Transport, class IdType = unit_t>
class transport_worker_dispatcher;
// -- classes ------------------------------------------------------------------
class endpoint_manager;
class middleman;
class middleman_backend;
class multiplexer;
class socket_manager;
// -- structs ------------------------------------------------------------------
struct network_socket;
struct pipe_socket;
struct socket;
struct stream_socket;
struct tcp_accept_socket;
struct tcp_stream_socket;
struct datagram_socket;
struct udp_datagram_socket;
using socket_manager_ptr = intrusive_ptr<socket_manager>;
// -- smart pointers -----------------------------------------------------------
using endpoint_manager_ptr = intrusive_ptr<endpoint_manager>;
using middleman_backend_ptr = std::unique_ptr<middleman_backend>;
using multiplexer_ptr = std::shared_ptr<multiplexer>;
using socket_manager_ptr = intrusive_ptr<socket_manager>;
using weak_multiplexer_ptr = std::weak_ptr<multiplexer>;
} // namespace net
} // namespace caf
} // namespace caf::net
......@@ -20,8 +20,7 @@
#include "caf/fwd.hpp"
namespace caf {
namespace net {
namespace caf::net {
struct this_host {
/// Initializes the network subsystem.
......@@ -31,5 +30,4 @@ struct this_host {
static void cleanup();
};
} // namespace net
} // namespace caf
} // namespace caf::net
......@@ -5,7 +5,7 @@
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* Copyright 2011-2019 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 *
......@@ -23,16 +23,22 @@
#include "caf/fwd.hpp"
namespace caf {
namespace net {
namespace ip {
namespace caf::net::ip {
/// Returns all IP addresses of to `host` (if any).
/// Returns all IP addresses of `host` (if any).
std::vector<ip_address> resolve(string_view host);
/// Returns all IP addresses of `host` (if any).
std::vector<ip_address> resolve(ip_address host);
/// Returns the IP addresses for a local endpoint, which is either an address,
/// an interface name, or the string "localhost".
std::vector<ip_address> local_addresses(string_view host);
/// Returns the IP addresses for a local endpoint address.
std::vector<ip_address> local_addresses(ip_address host);
/// Returns the hostname of this device.
std::string hostname();
} // namespace ip
} // namespace net
} // namespace caf
} // namespace caf::net::ip
......@@ -22,16 +22,13 @@
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/endpoint_manager_impl.hpp"
namespace caf {
namespace net {
namespace caf::net {
template <class Transport, class Application>
template <class Transport>
endpoint_manager_ptr make_endpoint_manager(const multiplexer_ptr& mpx,
actor_system& sys, Transport trans,
Application app) {
using impl = endpoint_manager_impl<Transport, Application>;
return make_counted<impl>(mpx, sys, std::move(trans), std::move(app));
actor_system& sys, Transport trans) {
using impl = endpoint_manager_impl<Transport>;
return make_counted<impl>(mpx, sys, std::move(trans));
}
} // namespace net
} // namespace caf
} // namespace caf::net
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include <thread>
#include "caf/actor_system.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
namespace caf::net {
class middleman : public actor_system::module {
public:
// -- member types -----------------------------------------------------------
using module = actor_system::module;
using module_ptr = actor_system::module_ptr;
using middleman_backend_list = std::vector<middleman_backend_ptr>;
// -- constructors, destructors, and assignment operators --------------------
~middleman() override;
// -- interface functions ----------------------------------------------------
void start() override;
void stop() override;
void init(actor_system_config&) override;
id_t id() const override;
void* subtype_ptr() override;
// -- factory functions ------------------------------------------------------
template <class... Ts>
static module* make(actor_system& sys, detail::type_list<Ts...> token) {
std::unique_ptr<middleman> result{new middleman(sys)};
if (sizeof...(Ts) > 0) {
result->backends_.reserve(sizeof...(Ts));
create_backends(*result, token);
}
return result.release();
}
// -- remoting ---------------------------------------------------------------
/// Resolves a path to a remote actor.
void resolve(const uri& locator, const actor& listener);
// -- properties -------------------------------------------------------------
actor_system& system() {
return sys_;
}
const actor_system_config& config() const noexcept {
return sys_.config();
}
const multiplexer_ptr& mpx() const noexcept {
return mpx_;
}
middleman_backend* backend(string_view scheme) const noexcept;
private:
// -- constructors, destructors, and assignment operators --------------------
explicit middleman(actor_system& sys);
// -- utility functions ------------------------------------------------------
static void create_backends(middleman&, detail::type_list<>) {
// End of recursion.
}
template <class T, class... Ts>
static void create_backends(middleman& mm, detail::type_list<T, Ts...>) {
mm.backends_.emplace_back(new T(mm));
create_backends(mm, detail::type_list<Ts...>{});
}
// -- member variables -------------------------------------------------------
/// Points to the parent system.
actor_system& sys_;
/// Stores the global socket I/O multiplexer.
multiplexer_ptr mpx_;
/// Stores all available backends for managing peers.
middleman_backend_list backends_;
/// Runs the multiplexer's event loop
std::thread mpx_thread_;
};
} // namespace caf::net
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include <string>
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/proxy_registry.hpp"
namespace caf::net {
/// Technology-specific backend for connecting to and managing peer
/// connections.
/// @relates middleman
class middleman_backend : public proxy_registry::backend {
public:
// -- constructors, destructors, and assignment operators --------------------
middleman_backend(std::string id);
virtual ~middleman_backend();
// -- interface functions ----------------------------------------------------
/// Initializes the backend.
virtual error init() = 0;
/// @returns The endpoint manager for `peer` on success, `nullptr` otherwise.
virtual endpoint_manager_ptr peer(const node_id& id) = 0;
/// Resolves a path to a remote actor.
virtual void resolve(const uri& locator, const actor& listener) = 0;
// -- properties -------------------------------------------------------------
const std::string& id() const noexcept {
return id_;
}
private:
/// Stores the technology-specific identifier.
std::string id_;
};
/// @relates middleman_backend
using middleman_backend_ptr = std::unique_ptr<middleman_backend>;
} // namespace caf::net
......@@ -34,8 +34,7 @@ struct pollfd;
} // extern "C"
namespace caf {
namespace net {
namespace caf::net {
/// Multiplexes any number of ::socket_manager objects with a ::socket.
class multiplexer : public std::enable_shared_from_this<multiplexer> {
......@@ -64,9 +63,13 @@ public:
// -- thread-safe signaling --------------------------------------------------
/// Causes the multiplexer to update its event bitmask for `mgr`.
/// Registers `mgr` for read events.
/// @thread-safe
void update(const socket_manager_ptr& mgr);
void register_reading(const socket_manager_ptr& mgr);
/// Registers `mgr` for write events.
/// @thread-safe
void register_writing(const socket_manager_ptr& mgr);
/// Closes the pipe for signaling updates to the multiplexer. After closing
/// the pipe, calls to `update` no longer have any effect.
......@@ -79,21 +82,25 @@ public:
/// ready as a result.
bool poll_once(bool blocking);
/// Sets the thread ID to `std::this_thread::id()`.
void set_thread_id();
/// Polls until no socket event handler remains.
void run();
/// Processes all updates on the socket managers.
void handle_updates();
protected:
// -- utility functions ------------------------------------------------------
/// Handles an I/O event on given manager.
void handle(const socket_manager_ptr& mgr, int mask);
short handle(const socket_manager_ptr& mgr, short events, short revents);
/// Adds a new socket manager to the pollset.
void add(socket_manager_ptr mgr);
/// Writes `opcode` and pointer to `mgr` the the pipe for handling an event
/// later via the pollset updater.
void write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr);
// -- member variables -------------------------------------------------------
/// Bookkeeping data for managed sockets.
......@@ -103,9 +110,6 @@ protected:
/// order as their sockets appear in `pollset_`.
manager_list managers_;
/// Managers that updated their event mask and need updating.
manager_list dirty_managers_;
/// Stores the ID of the thread this multiplexer is running in. Set when
/// calling `init()`.
std::thread::id tid_;
......@@ -123,5 +127,4 @@ using multiplexer_ptr = std::shared_ptr<multiplexer>;
/// @relates multiplexer
using weak_multiplexer_ptr = std::weak_ptr<multiplexer>;
} // namespace net
} // namespace caf
} // namespace caf::net
......@@ -25,22 +25,16 @@
#include "caf/config.hpp"
#include "caf/fwd.hpp"
#include "caf/net/abstract_socket.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/socket_id.hpp"
namespace caf {
namespace net {
namespace caf::net {
/// A bidirectional network communication endpoint.
struct network_socket : abstract_socket<network_socket> {
using super = abstract_socket<network_socket>;
struct network_socket : socket {
using super = socket;
using super::super;
constexpr operator socket() const noexcept {
return socket{id};
}
};
/// Enables or disables `SIGPIPE` events from `x`.
......@@ -88,5 +82,4 @@ void shutdown_write(network_socket x);
/// @relates network_socket
void shutdown(network_socket x);
} // namespace net
} // namespace caf
} // namespace caf::net
......@@ -18,8 +18,9 @@
#pragma once
namespace caf {
namespace net {
#include <string>
namespace caf::net {
/// Values for representing bitmask of I/O operations.
enum class operation {
......@@ -45,5 +46,6 @@ constexpr operation operator~(operation x) {
return static_cast<operation>(~static_cast<int>(x));
}
} // namespace net
} // namespace caf
std::string to_string(operation x);
} // namespace caf::net
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include <vector>
#include "caf/byte.hpp"
#include "caf/net/fwd.hpp"
#include "caf/span.hpp"
namespace caf::net {
/// Implements an interface for packet writing in application-layers.
class packet_writer {
public:
using buffer_type = std::vector<byte>;
virtual ~packet_writer();
/// Returns a buffer for writing header information.
virtual buffer_type next_header_buffer() = 0;
/// Returns a buffer for writing payload content.
virtual buffer_type next_payload_buffer() = 0;
/// Convenience function to write a packet consisting of multiple buffers.
/// @param buffers all buffers for the packet. The first buffer is a header
/// buffer, the other buffers are payload buffer.
/// @warning this function takes ownership of `buffers`.
template <class... Ts>
void write_packet(Ts&... buffers) {
buffer_type* bufs[] = {&buffers...};
write_impl(make_span(bufs, sizeof...(Ts)));
}
protected:
/// Implementing function for `write_packet`.
/// @param buffers a `span` containing all buffers of a packet.
virtual void write_impl(span<buffer_type*> buffers) = 0;
};
} // namespace caf::net
......@@ -18,42 +18,67 @@
#pragma once
namespace caf {
namespace net {
#include "caf/net/packet_writer.hpp"
#include "caf/byte.hpp"
#include "caf/span.hpp"
namespace caf::net {
/// Implements the interface for transport and application policies and
/// dispatches member functions either to `decorator` or `parent`.
/// dispatches member functions either to `object` or `parent`.
template <class Object, class Parent>
class write_packet_decorator {
class packet_writer_decorator final : public packet_writer {
public:
dispatcher(Object& object, Parent& parent)
// -- member types -----------------------------------------------------------
using transport_type = typename Parent::transport_type;
using application_type = typename Parent::application_type;
// -- constructors, destructors, and assignment operators --------------------
packet_writer_decorator(Object& object, Parent& parent)
: object_(object), parent_(parent) {
// nop
}
template <class Header>
void write_packet(const Header& header, span<const byte> payload) {
object_.write_packet(parent_, header, payload);
}
// -- properties -------------------------------------------------------------
actor_system& system() {
parent_.system();
return parent_.system();
}
void cancel_timeout(atom_value type, uint64_t id) {
parent_.cancel_timeout(type, id);
transport_type& transport() {
return parent_.transport();
}
void set_timeout(timestamp tout, atom_value type, uint64_t id) {
parent_.set_timeout(tout, type, id);
endpoint_manager& manager() {
return parent_.manager();
}
buffer_type next_header_buffer() override {
return transport().next_header_buffer();
}
buffer_type next_payload_buffer() override {
return transport().next_payload_buffer();
}
// -- member functions -------------------------------------------------------
void cancel_timeout(atom_value type, uint64_t id) {
parent_.cancel_timeout(type, id);
}
typename parent::transport_type& transport() {
return parent_.transport_;
template <class... Ts>
uint64_t set_timeout(timestamp tout, atom_value type, Ts&&... xs) {
return parent_.set_timeout(tout, type, std::forward<Ts>(xs)...);
}
typename parent::application_type& application() {
return parent_.application_;
protected:
void write_impl(span<buffer_type*> buffers) override {
parent_.write_packet(object_.id(), buffers);
}
private:
......@@ -62,10 +87,9 @@ private:
};
template <class Object, class Parent>
write_packet_decorator<Object, Parent>
make_write_packet_decorator(Object& object, Parent& parent) {
packet_writer_decorator<Object, Parent>
make_packet_writer_decorator(Object& object, Parent& parent) {
return {object, parent};
}
} // namespace net
} // namespace caf
} // namespace caf::net
......@@ -23,22 +23,16 @@
#include <utility>
#include "caf/fwd.hpp"
#include "caf/net/abstract_socket.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/socket_id.hpp"
namespace caf {
namespace net {
namespace caf::net {
/// A unidirectional communication endpoint for inter-process communication.
struct pipe_socket : abstract_socket<pipe_socket> {
using super = abstract_socket<pipe_socket>;
struct pipe_socket : socket {
using super = socket;
using super::super;
constexpr operator socket() const noexcept {
return socket{id};
}
};
/// Creates two connected sockets. The first socket is the read handle and the
......@@ -68,5 +62,4 @@ variant<size_t, sec> read(pipe_socket x, span<byte>);
variant<size_t, sec>
check_pipe_socket_io_res(std::make_signed<size_t>::type res);
} // namespace net
} // namespace caf
} // namespace caf::net
......@@ -20,13 +20,13 @@
#include <array>
#include <cstdint>
#include <mutex>
#include "caf/byte.hpp"
#include "caf/net/pipe_socket.hpp"
#include "caf/net/socket_manager.hpp"
namespace caf {
namespace net {
namespace caf::net {
class pollset_updater : public socket_manager {
public:
......@@ -34,9 +34,11 @@ public:
using super = socket_manager;
using msg_buf = std::array<byte, sizeof(intptr_t) + 1>;
// -- constructors, destructors, and assignment operators --------------------
pollset_updater(pipe_socket read_handle, multiplexer_ptr parent);
pollset_updater(pipe_socket read_handle, const multiplexer_ptr& parent);
~pollset_updater() override;
......@@ -56,9 +58,8 @@ public:
void handle_error(sec code) override;
private:
std::array<byte, sizeof(intptr_t)> buf_;
msg_buf buf_;
size_t buf_size_;
};
} // namespace net
} // namespace caf
} // namespace caf::net
......@@ -24,8 +24,7 @@
#include "caf/config.hpp"
namespace caf {
namespace net {
namespace caf::net {
enum class receive_policy_flag : unsigned { at_least, at_most, exactly };
......@@ -57,5 +56,4 @@ public:
}
};
} // namespace net
} // namespace caf
} // namespace caf::net
......@@ -23,21 +23,41 @@
#include <type_traits>
#include "caf/config.hpp"
#include "caf/detail/comparable.hpp"
#include "caf/fwd.hpp"
#include "caf/net/abstract_socket.hpp"
#include "caf/net/socket_id.hpp"
namespace caf {
namespace net {
namespace caf::net {
/// An internal endpoint for sending or receiving data. Can be either a
/// ::network_socket, ::pipe_socket, ::stream_socket, or ::datagram_socket.
struct socket : abstract_socket<socket> {
using super = abstract_socket<socket>;
struct socket : detail::comparable<socket> {
socket_id id;
using super::super;
constexpr socket() noexcept : id(invalid_socket_id) {
// nop
}
constexpr explicit socket(socket_id id) noexcept : id(id) {
// nop
}
constexpr socket(const socket& other) noexcept = default;
socket& operator=(const socket& other) noexcept = default;
constexpr signed_socket_id compare(socket other) const noexcept {
return static_cast<signed_socket_id>(id)
- static_cast<signed_socket_id>(other.id);
}
};
/// @relates socket
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, socket& x) {
return f(x.id);
}
/// Denotes the invalid socket.
constexpr auto invalid_socket = socket{invalid_socket_id};
......@@ -68,5 +88,4 @@ error child_process_inherit(socket x, bool new_value);
/// @relates socket
error nonblocking(socket x, bool new_value);
} // namespace net
} // namespace caf
} // namespace caf::net
......@@ -18,32 +18,56 @@
#pragma once
#include "caf/net/socket.hpp"
#include "caf/net/socket_id.hpp"
namespace caf {
namespace net {
namespace caf::net {
/// Closes the guarded socket when destroyed.
template <class Socket>
class socket_guard {
public:
explicit socket_guard(Socket sock) : sock_(sock) {
socket_guard() noexcept : sock_(invalid_socket_id) {
// nop
}
explicit socket_guard(Socket sock) noexcept : sock_(sock) {
// nop
}
socket_guard(socket_guard&& other) noexcept : sock_(other.release()) {
// nop
}
socket_guard(const socket_guard&) = delete;
socket_guard& operator=(socket_guard&& other) noexcept {
reset(other.release());
return *this;
}
socket_guard& operator=(const socket_guard&) = delete;
~socket_guard() {
if (sock_.id != invalid_socket_id) {
net::close(sock_);
sock_.id = invalid_socket_id;
}
if (sock_.id != invalid_socket_id)
close(sock_);
}
Socket release() {
void reset(Socket x) noexcept {
if (sock_.id != invalid_socket_id)
close(sock_);
sock_ = x;
}
Socket release() noexcept {
auto sock = sock_;
sock_.id = invalid_socket_id;
return sock;
}
Socket socket() const noexcept {
return sock_;
}
private:
Socket sock_;
};
......@@ -53,5 +77,4 @@ socket_guard<Socket> make_socket_guard(Socket sock) {
return socket_guard<Socket>{sock};
}
} // namespace net
} // namespace caf
} // namespace caf::net
......@@ -20,11 +20,11 @@
#include <cstddef>
#include <limits>
#include <type_traits>
#include "caf/config.hpp"
namespace caf {
namespace net {
namespace caf::net {
#ifdef CAF_WINDOWS
......@@ -46,5 +46,8 @@ constexpr socket_id invalid_socket_id = -1;
#endif // CAF_WINDOWS
} // namespace net
} // namespace caf
/// Signed counterpart of `socket_id`.
/// @relates socket
using signed_socket_id = std::make_signed<socket_id>::type;
} // namespace caf::net
......@@ -18,8 +18,6 @@
#pragma once
#include <atomic>
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
......@@ -27,8 +25,7 @@
#include "caf/net/socket.hpp"
#include "caf/ref_counted.hpp"
namespace caf {
namespace net {
namespace caf::net {
/// Manages the lifetime of a single socket and handles any I/O events on it.
class socket_manager : public ref_counted {
......@@ -59,21 +56,26 @@ public:
}
/// Returns registered operations (read, write, or both).
operation mask() const noexcept;
operation mask() const noexcept {
return mask_;
}
/// Adds given flag(s) to the event mask and updates the parent on success.
/// Adds given flag(s) to the event mask.
/// @returns `false` if `mask() | flag == mask()`, `true` otherwise.
/// @pre `has_parent()`
/// @pre `flag != operation::none`
bool mask_add(operation flag) noexcept;
/// Deletes given flag(s) from the event mask and updates the parent on
/// success.
/// Tries to clear given flag(s) from the event mask.
/// @returns `false` if `mask() & ~flag == mask()`, `true` otherwise.
/// @pre `has_parent()`
/// @pre `flag != operation::none`
bool mask_del(operation flag) noexcept;
// -- event loop management --------------------------------------------------
void register_reading();
void register_writing();
// -- pure virtual member functions ------------------------------------------
/// Called whenever the socket received new data.
......@@ -91,12 +93,11 @@ protected:
socket handle_;
std::atomic<operation> mask_;
operation mask_;
weak_multiplexer_ptr parent_;
};
using socket_manager_ptr = intrusive_ptr<socket_manager>;
} // namespace net
} // namespace caf
} // namespace caf::net
......@@ -21,23 +21,14 @@
#include "caf/fwd.hpp"
#include "caf/net/network_socket.hpp"
namespace caf {
namespace net {
namespace caf::net {
/// A connection-oriented network communication endpoint for bidirectional byte
/// streams.
struct stream_socket : abstract_socket<stream_socket> {
using super = abstract_socket<stream_socket>;
struct stream_socket : network_socket {
using super = network_socket;
using super::super;
constexpr operator socket() const noexcept {
return socket{id};
}
constexpr operator network_socket() const noexcept {
return network_socket{id};
}
};
/// Creates two connected sockets to mimic network communication (usually for
......@@ -56,7 +47,6 @@ error nodelay(stream_socket x, bool new_value);
/// Receives data from `x`.
/// @param x Connected endpoint.
/// @param buf Points to destination buffer.
/// @param buf_size Specifies the maximum size of the buffer in bytes.
/// @returns The number of received bytes on success, an error code otherwise.
/// @relates stream_socket
/// @post either the result is a `sec` or a positive (non-zero) integer
......@@ -65,17 +55,26 @@ variant<size_t, sec> read(stream_socket x, span<byte> buf);
/// Transmits data from `x` to its peer.
/// @param x Connected endpoint.
/// @param buf Points to the message to send.
/// @param buf_size Specifies the size of the buffer in bytes.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates stream_socket
/// @post either the result is a `sec` or a positive (non-zero) integer
variant<size_t, sec> write(stream_socket x, span<const byte> buf);
/// Transmits data from `x` to its peer.
/// @param x Connected endpoint.
/// @param bufs Points to the message to send, scattered across up to 10
/// buffers.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates stream_socket
/// @post either the result is a `sec` or a positive (non-zero) integer
/// @pre `bufs.size() < 10`
variant<size_t, sec> write(stream_socket x,
std::initializer_list<span<const byte>> bufs);
/// Converts the result from I/O operation on a ::stream_socket to either an
/// error code or a non-zero positive integer.
/// @relates stream_socket
variant<size_t, sec>
check_stream_socket_io_res(std::make_signed<size_t>::type res);
} // namespace net
} // namespace caf
} // namespace caf::net
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include <deque>
#include <vector>
#include "caf/fwd.hpp"
#include "caf/logger.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/transport_base.hpp"
#include "caf/net/transport_worker.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
namespace caf::net {
template <class Application>
using stream_transport_base = transport_base<
stream_transport<Application>, transport_worker<Application>, stream_socket,
Application, unit_t>;
/// Implements a stream_transport that manages a stream socket.
template <class Application>
class stream_transport : public stream_transport_base<Application> {
public:
// -- member types -----------------------------------------------------------
using application_type = Application;
using worker_type = transport_worker<application_type>;
using super = stream_transport_base<application_type>;
using id_type = typename super::id_type;
using buffer_type = typename super::buffer_type;
using write_queue_type = std::deque<std::pair<bool, buffer_type>>;
// -- constructors, destructors, and assignment operators --------------------
stream_transport(stream_socket handle, application_type application)
: super(handle, std::move(application)),
written_(0),
read_threshold_(1024),
collected_(0),
max_(1024),
rd_flag_(net::receive_policy_flag::exactly) {
// nop
}
// -- member functions -------------------------------------------------------
bool handle_read_event(endpoint_manager&) override {
auto buf = this->read_buf_.data() + this->collected_;
size_t len = this->read_threshold_ - this->collected_;
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle().id)
<< CAF_ARG2("missing", len));
auto ret = read(this->handle_, make_span(buf, len));
// Update state.
if (auto num_bytes = get_if<size_t>(&ret)) {
CAF_LOG_DEBUG(CAF_ARG(len)
<< CAF_ARG(this->handle_.id) << CAF_ARG(*num_bytes));
this->collected_ += *num_bytes;
if (this->collected_ >= this->read_threshold_) {
if (auto err = this->next_layer_.handle_data(*this, this->read_buf_)) {
CAF_LOG_ERROR("handle_data failed: " << CAF_ARG(err));
return false;
}
this->prepare_next_read();
}
} else {
auto err = get<sec>(ret);
if (err != sec::unavailable_or_would_block) {
CAF_LOG_DEBUG("read failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
return false;
}
}
return true;
}
bool handle_write_event(endpoint_manager& parent) override {
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle().id));
// Try to write leftover data.
write_some();
// Get new data from parent.
// TODO: dont read all messages at once - get one by one.
for (auto msg = parent.next_message(); msg != nullptr;
msg = parent.next_message()) {
this->next_layer_.write_message(*this, std::move(msg));
}
// Write prepared data.
return write_some();
}
void write_packet(id_type, span<buffer_type*> buffers) override {
CAF_LOG_TRACE("");
CAF_ASSERT(!buffers.empty());
if (this->write_queue_.empty())
this->manager().register_writing();
// By convention, the first buffer is a header buffer. Every other buffer is
// a payload buffer.
auto i = buffers.begin();
this->write_queue_.emplace_back(true, std::move(*(*i++)));
while (i != buffers.end())
this->write_queue_.emplace_back(false, std::move(*(*i++)));
}
void configure_read(receive_policy::config cfg) override {
rd_flag_ = cfg.first;
max_ = cfg.second;
prepare_next_read();
}
private:
// -- utility functions ------------------------------------------------------
void prepare_next_read() {
collected_ = 0;
switch (rd_flag_) {
case net::receive_policy_flag::exactly:
if (this->read_buf_.size() != max_)
this->read_buf_.resize(max_);
read_threshold_ = max_;
break;
case net::receive_policy_flag::at_most:
if (this->read_buf_.size() != max_)
this->read_buf_.resize(max_);
read_threshold_ = 1;
break;
case net::receive_policy_flag::at_least: {
// read up to 10% more, but at least allow 100 bytes more
auto max_size = max_ + std::max<size_t>(100, max_ / 10);
if (this->read_buf_.size() != max_size)
this->read_buf_.resize(max_size);
read_threshold_ = max_;
break;
}
}
}
bool write_some() {
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle_.id));
// Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() {
auto& front = this->write_queue_.front();
auto& is_header = front.first;
auto& buf = front.second;
written_ = 0;
buf.clear();
if (is_header) {
if (this->header_bufs_.size() < this->header_bufs_.capacity())
this->header_bufs_.emplace_back(std::move(buf));
} else if (this->payload_bufs_.size() < this->payload_bufs_.capacity()) {
this->payload_bufs_.emplace_back(std::move(buf));
}
write_queue_.pop_front();
};
// Write buffers from the write_queue_ for as long as possible.
while (!write_queue_.empty()) {
auto& buf = write_queue_.front().second;
CAF_ASSERT(!buf.empty());
auto data = buf.data() + written_;
auto len = buf.size() - written_;
auto write_ret = write(this->handle(), make_span(data, len));
if (auto num_bytes = get_if<size_t>(&write_ret)) {
CAF_LOG_DEBUG(CAF_ARG(this->handle_.id) << CAF_ARG(*num_bytes));
written_ += *num_bytes;
if (written_ >= buf.size()) {
recycle();
written_ = 0;
}
} else {
auto err = get<sec>(write_ret);
if (err != sec::unavailable_or_would_block) {
CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
return false;
}
return true;
}
}
return false;
}
write_queue_type write_queue_;
size_t written_;
size_t read_threshold_;
size_t collected_;
size_t max_;
receive_policy_flag rd_flag_;
// TODO implement retries using this member!
// size_t max_consecutive_reads_;
};
} // namespace caf::net
......@@ -18,35 +18,24 @@
#pragma once
#include "caf/ip_endpoint.hpp"
#include "caf/net/abstract_socket.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/network_socket.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/uri.hpp"
namespace caf {
namespace net {
namespace caf::net {
/// Represents a TCP acceptr in listening mode.
struct tcp_accept_socket : abstract_socket<tcp_accept_socket> {
using super = abstract_socket<tcp_accept_socket>;
/// Represents a TCP acceptor in listening mode.
struct tcp_accept_socket : network_socket {
using super = network_socket;
using super::super;
constexpr operator socket() const noexcept {
return socket{id};
}
constexpr operator network_socket() const noexcept {
return network_socket{id};
}
};
/// Creates a new TCP socket to accept connections on a given port.
/// @param node The endpoint to listen on and the filter for incoming addresses.
/// Passing the address `0.0.0.0` will accept incoming connection from any host.
/// Passing port 0 lets the OS choose the port.
/// @relates tcp_accept_socket
expected<tcp_accept_socket> make_tcp_accept_socket(ip_endpoint node,
bool reuse_addr = false);
......@@ -54,6 +43,7 @@ expected<tcp_accept_socket> make_tcp_accept_socket(ip_endpoint node,
/// Creates a new TCP socket to accept connections on a given port.
/// @param node The endpoint to listen on and the filter for incoming addresses.
/// Passing the address `0.0.0.0` will accept incoming connection from any host.
/// Passing port 0 lets the OS choose the port.
/// @param reuse_addr Optionally sets the SO_REUSEADDR option on the socket.
/// @relates tcp_accept_socket
expected<tcp_accept_socket>
......@@ -67,5 +57,4 @@ make_tcp_accept_socket(const uri::authority_type& node,
/// @relates tcp_accept_socket
expected<tcp_stream_socket> accept(tcp_accept_socket x);
} // namespace net
} // namespace caf
} // namespace caf::net
......@@ -19,46 +19,30 @@
#pragma once
#include "caf/ip_endpoint.hpp"
#include "caf/net/abstract_socket.hpp"
#include "caf/net/network_socket.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/uri.hpp"
namespace caf {
namespace net {
namespace caf::net {
/// Represents a TCP connection.
struct tcp_stream_socket : abstract_socket<tcp_stream_socket> {
using super = abstract_socket<tcp_stream_socket>;
struct tcp_stream_socket : stream_socket {
using super = stream_socket;
using super::super;
constexpr operator socket() const noexcept {
return socket{id};
}
constexpr operator network_socket() const noexcept {
return network_socket{id};
}
constexpr operator stream_socket() const noexcept {
return stream_socket{id};
}
};
/// Creates a `tcp_stream_socket` connected to given remote node.
/// @param node Host and port of the remote node
/// @param node Host and port of the remote node.
/// @returns The connected socket or an error.
/// @relates tcp_stream_socket
expected<tcp_stream_socket> make_connected_tcp_stream_socket(ip_endpoint node);
/// Create a `tcp_stream_socket` connected to `auth`.
/// @param node Host and port of the remote node
/// @param node Host and port of the remote node.
/// @returns The connected socket or an error.
/// @relates tcp_stream_socket
expected<tcp_stream_socket>
make_connected_tcp_stream_socket(const uri::authority_type& node);
} // namespace net
} // namespace caf
} // namespace caf::net
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include "caf/detail/overload.hpp"
#include "caf/fwd.hpp"
#include "caf/logger.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/receive_policy.hpp"
namespace caf::net {
/// Implements a base class for transports.
/// @tparam Transport The derived type of the transport implementation.
/// @tparam NextLayer The Following Layer. Either `transport_worker` or
/// `transport_worker_dispatcher`
/// @tparam Handle The type of the related socket_handle.
/// @tparam Application The type of the application used in this stack.
/// @tparam IdType The id type of the derived transport, must match the IdType
/// of `NextLayer`.
template <class Transport, class NextLayer, class Handle, class Application,
class IdType>
class transport_base {
public:
// -- member types -----------------------------------------------------------
using next_layer_type = NextLayer;
using handle_type = Handle;
using transport_type = Transport;
using application_type = Application;
using id_type = IdType;
using buffer_type = std::vector<byte>;
using buffer_cache_type = std::vector<buffer_type>;
// -- constructors, destructors, and assignment operators --------------------
transport_base(handle_type handle, application_type application)
: next_layer_(std::move(application)), handle_(handle), manager_(nullptr) {
// nop
}
// -- properties -------------------------------------------------------------
/// Returns the socket_handle of this transport.
handle_type handle() const noexcept {
return handle_;
}
/// Returns a reference to the `actor_system` of this transport.
/// @pre `init` must be called before calling this getter.
actor_system& system() {
return manager().system();
}
/// Returns a reference to the application of this transport.
application_type& application() {
return next_layer_.application();
}
/// Returns a reference to this transport.
transport_type& transport() {
return *reinterpret_cast<transport_type*>(this);
}
/// Returns a reference to the `endpoint_manager` of this transport.
/// @pre `init` must be called before calling this getter.
endpoint_manager& manager() {
CAF_ASSERT(manager_);
return *manager_;
}
// -- transport member functions ---------------------------------------------
/// Initializes this transport.
/// @param parent The `endpoint_manager` that manages this transport.
/// @returns `error` on failure, none on success.
virtual error init(endpoint_manager& parent) {
CAF_LOG_TRACE("");
manager_ = &parent;
auto& cfg = system().config();
auto max_header_bufs = get_or(cfg, "middleman.max-header-buffers",
defaults::middleman::max_header_buffers);
header_bufs_.reserve(max_header_bufs);
auto max_payload_bufs = get_or(cfg, "middleman.max-payload-buffers",
defaults::middleman::max_payload_buffers);
payload_bufs_.reserve(max_payload_bufs);
if (auto err = next_layer_.init(*this))
return err;
return none;
}
/// Resolves a remote actor using `locator` and sends the resolved actor to
/// listener on success - an `error` otherwise.
/// @param locator The `uri` of the remote actor.
/// @param listener The `actor_handle` which the result should be sent to.
auto resolve(endpoint_manager&, const uri& locator, const actor& listener) {
CAF_LOG_TRACE(CAF_ARG(locator) << CAF_ARG(listener));
auto f = detail::make_overload(
[&](auto& layer) -> decltype(layer.resolve(*this, locator, listener)) {
return layer.resolve(*this, locator, listener);
},
[&](auto& layer) -> decltype(
layer.resolve(*this, locator.path(), listener)) {
return layer.resolve(*this, locator.path(), listener);
});
f(next_layer_);
}
/// Gets called by an actor proxy after creation.
/// @param peer The `node_id` of the remote node.
/// @param id The id of the remote actor.
void new_proxy(endpoint_manager&, const node_id& peer, actor_id id) {
next_layer_.new_proxy(*this, peer, id);
}
/// Notifies the remote endpoint that the local actor is down.
/// @param peer The `node_id` of the remote endpoint.
/// @param id The `actor_id` of the remote actor.
/// @param reason The reason why the local actor has shut down.
void local_actor_down(endpoint_manager&, const node_id& peer, actor_id id,
error reason) {
next_layer_.local_actor_down(*this, peer, id, std::move(reason));
}
/// Notifies the transport that the timeout identified by `value` plus `id`
/// was triggered.
/// @param value The `atom_value` of the timeout.
/// @param id The timeout id of the timeout.
void timeout(endpoint_manager&, atom_value value, uint64_t id) {
next_layer_.timeout(*this, value, id);
}
/// Callback for setting a timeout. Will be called after setting a timeout to
/// get the timeout id for local use.
/// @param timeout_id The id of the previously set timeout.
/// @param ts Any further information that was passed when setting the
/// timeout.
template <class... Ts>
void set_timeout(uint64_t timeout_id, Ts&&... ts) {
next_layer_.set_timeout(timeout_id, std::forward<Ts>(ts)...);
}
/// Callback for when an error occurs.
/// @param code The error code to handle.
void handle_error(sec code) {
next_layer_.handle_error(code);
}
// -- (pure) virtual functions -----------------------------------------------
/// Configures this transport for the next read event.
virtual void configure_read(receive_policy::config) {
// nop
}
/// Called by the endpoint manager when the transport can read data from its
/// socket.
virtual bool handle_read_event(endpoint_manager&) = 0;
/// Called by the endpoint manager when the transport can write data to its
/// socket.
virtual bool handle_write_event(endpoint_manager& parent) = 0;
/// Queues a packet scattered across multiple buffers to be sent via this
/// transport.
/// @param id The id of the destination endpoint.
/// @param buffers Pointers to the buffers that make up the packet content.
virtual void write_packet(id_type id, span<buffer_type*> buffers) = 0;
// -- buffer management ------------------------------------------------------
/// Returns the next cached header buffer or creates a new one if no buffers
/// are cached.
buffer_type next_header_buffer() {
return next_buffer_impl(header_bufs_);
}
/// Returns the next cached payload buffer or creates a new one if no buffers
/// are cached.
buffer_type next_payload_buffer() {
return next_buffer_impl(payload_bufs_);
}
private:
// -- utility functions ------------------------------------------------------
static buffer_type next_buffer_impl(buffer_cache_type cache) {
if (cache.empty())
return {};
auto buf = std::move(cache.back());
cache.pop_back();
return buf;
}
protected:
next_layer_type next_layer_;
handle_type handle_;
buffer_cache_type header_bufs_;
buffer_cache_type payload_bufs_;
buffer_type read_buf_;
endpoint_manager* manager_;
};
} // namespace caf::net
......@@ -5,7 +5,7 @@
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* Copyright 2011-2019 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 *
......@@ -18,135 +18,102 @@
#pragma once
#include "caf/byte.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/logger.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp"
namespace caf {
namespace policy {
/// Implements a scribe policy that manages a stream socket.
class scribe {
#include "caf/net/endpoint_manager_queue.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/packet_writer_decorator.hpp"
#include "caf/unit.hpp"
namespace caf::net {
/// Implements a worker for transport protocols.
template <class Application, class IdType>
class transport_worker {
public:
explicit scribe(net::stream_socket handle);
// -- member types -----------------------------------------------------------
using id_type = IdType;
using application_type = Application;
// -- constructors, destructors, and assignment operators --------------------
explicit transport_worker(application_type application,
id_type id = id_type{})
: application_(std::move(application)), id_(std::move(id)) {
// nop
}
// -- properties -------------------------------------------------------------
application_type& application() noexcept {
return application_;
}
const application_type& application() const noexcept {
return application_;
}
net::stream_socket handle() const noexcept {
return handle_;
const id_type& id() const noexcept {
return id_;
}
// -- member functions -------------------------------------------------------
template <class Parent>
error init(Parent& parent) {
if (auto err = parent.application().init(parent))
return err;
parent.mask_add(net::operation::read);
return none;
auto writer = make_packet_writer_decorator(*this, parent);
return application_.init(writer);
}
template <class Parent>
bool handle_read_event(Parent& parent) {
auto buf = read_buf_.data() + collected_;
size_t len = read_threshold_ - collected_;
CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(len));
auto ret = read(handle_, make_span(buf, len));
// Update state.
if (auto num_bytes = get_if<size_t>(&ret)) {
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
collected_ += *num_bytes;
if (collected_ >= read_threshold_) {
parent.application().handle_data(*this, read_buf_);
prepare_next_read();
}
return true;
} else {
auto err = get<sec>(ret);
CAF_LOG_DEBUG("receive failed" << CAF_ARG(err));
parent.application().handle_error(err);
return false;
}
error handle_data(Parent& parent, span<const byte> data) {
auto writer = make_packet_writer_decorator(*this, parent);
return application_.handle_data(writer, data);
}
template <class Parent>
bool handle_write_event(Parent& parent) {
// Try to write leftover data.
write_some(parent);
// Get new data from parent.
// TODO: dont read all messages at once - get one by one.
for (auto msg = parent.next_message(); msg != nullptr;
msg = parent.next_message()) {
parent.application().write_message(*this, std::move(msg));
}
// Write prepared data.
return write_some(parent);
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) {
auto writer = make_packet_writer_decorator(*this, parent);
application_.write_message(writer, std::move(msg));
}
template <class Parent>
bool write_some(Parent& parent) {
if (write_buf_.empty())
return false;
auto len = write_buf_.size() - written_;
auto buf = write_buf_.data() + written_;
CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(len));
auto ret = net::write(handle_, make_span(buf, len));
if (auto num_bytes = get_if<size_t>(&ret)) {
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
// Update state.
written_ += *num_bytes;
if (written_ >= write_buf_.size()) {
written_ = 0;
write_buf_.clear();
return false;
}
} else {
auto err = get<sec>(ret);
CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
parent.application().handle_error(err);
return false;
}
return true;
void resolve(Parent& parent, string_view path, const actor& listener) {
auto writer = make_packet_writer_decorator(*this, parent);
application_.resolve(writer, path, listener);
}
template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) {
parent.application().resolve(parent, path, listener);
void new_proxy(Parent& parent, const node_id&, actor_id id) {
auto writer = make_packet_writer_decorator(*this, parent);
application_.new_proxy(writer, id);
}
template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) {
parent.application().timeout(*this, value, id);
void local_actor_down(Parent& parent, const node_id&, actor_id id,
error reason) {
auto writer = make_packet_writer_decorator(*this, parent);
application_.local_actor_down(writer, id, std::move(reason));
}
template <class Application>
void handle_error(Application& application, sec code) {
application.handle_error(code);
template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) {
auto writer = make_packet_writer_decorator(*this, parent);
application_.timeout(writer, value, id);
}
void prepare_next_read();
void configure_read(net::receive_policy::config cfg);
void write_packet(span<const byte> buf);
void handle_error(sec error) {
application_.handle_error(error);
}
private:
net::stream_socket handle_;
std::vector<byte> read_buf_;
std::vector<byte> write_buf_;
size_t max_consecutive_reads_; // TODO use this field!
size_t read_threshold_;
size_t collected_;
size_t max_;
net::receive_policy_flag rd_flag_;
size_t written_;
application_type application_;
id_type id_;
};
} // namespace policy
} // namespace caf
template <class Application, class IdType = unit_t>
using transport_worker_ptr = std::shared_ptr<
transport_worker<Application, IdType>>;
} // namespace caf::net
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include <unordered_map>
#include "caf/logger.hpp"
#include "caf/net/endpoint_manager_queue.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/packet_writer_decorator.hpp"
#include "caf/net/transport_worker.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"
namespace caf::net {
/// Implements a dispatcher that dispatches between transport and workers.
template <class Factory, class IdType>
class transport_worker_dispatcher {
public:
// -- member types -----------------------------------------------------------
using id_type = IdType;
using factory_type = Factory;
using application_type = typename factory_type::application_type;
using worker_type = transport_worker<application_type, id_type>;
using worker_ptr = transport_worker_ptr<application_type, id_type>;
// -- constructors, destructors, and assignment operators --------------------
explicit transport_worker_dispatcher(factory_type factory)
: factory_(std::move(factory)) {
// nop
}
// -- member functions -------------------------------------------------------
template <class Parent>
error init(Parent&) {
CAF_ASSERT(workers_by_id_.empty());
return none;
}
template <class Parent>
error handle_data(Parent& parent, span<byte> data, id_type id) {
if (auto worker = find_worker(id))
return worker->handle_data(parent, data);
// TODO: Where to get node_id from here?
auto worker = add_new_worker(parent, node_id{}, id);
if (worker)
return (*worker)->handle_data(parent, data);
else
return std::move(worker.error());
}
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) {
auto receiver = msg->receiver;
if (!receiver)
return;
auto nid = receiver->node();
if (auto worker = find_worker(nid)) {
worker->write_message(parent, std::move(msg));
return;
}
// TODO: where to get id_type from here?
if (auto worker = add_new_worker(parent, nid, id_type{}))
(*worker)->write_message(parent, std::move(msg));
}
template <class Parent>
void resolve(Parent& parent, const uri& locator, const actor& listener) {
if (auto worker = find_worker(make_node_id(locator)))
worker->resolve(parent, locator.path(), listener);
else
anon_send(listener,
make_error(sec::runtime_error, "could not resolve node"));
}
template <class Parent>
void new_proxy(Parent& parent, const node_id& nid, actor_id id) {
if (auto worker = find_worker(nid))
worker->new_proxy(parent, nid, id);
}
template <class Parent>
void local_actor_down(Parent& parent, const node_id& nid, actor_id id,
error reason) {
if (auto worker = find_worker(nid))
worker->local_actor_down(parent, nid, id, std::move(reason));
}
template <class... Ts>
void set_timeout(uint64_t timeout_id, id_type id, Ts&&...) {
workers_by_timeout_id_.emplace(timeout_id, workers_by_id_.at(id));
}
template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) {
if (auto worker = workers_by_timeout_id_.at(id)) {
worker->timeout(parent, value, id);
workers_by_timeout_id_.erase(id);
}
}
void handle_error(sec error) {
for (const auto& p : workers_by_id_) {
auto worker = p.second;
worker->handle_error(error);
}
}
template <class Parent>
expected<worker_ptr> add_new_worker(Parent& parent, node_id node,
id_type id) {
CAF_LOG_TRACE(CAF_ARG(node) << CAF_ARG(id));
auto application = factory_.make();
auto worker = std::make_shared<worker_type>(std::move(application), id);
if (auto err = worker->init(parent))
return err;
workers_by_id_.emplace(std::move(id), worker);
workers_by_node_.emplace(std::move(node), worker);
return worker;
}
private:
worker_ptr find_worker(const node_id& nid) {
return find_worker_impl(workers_by_node_, nid);
}
worker_ptr find_worker(const id_type& id) {
return find_worker_impl(workers_by_id_, id);
}
template <class Key>
worker_ptr find_worker_impl(const std::unordered_map<Key, worker_ptr>& map,
const Key& key) {
if (map.count(key) == 0) {
CAF_LOG_DEBUG("could not find worker: " << CAF_ARG(key));
return nullptr;
}
return map.at(key);
}
// -- worker lookups ---------------------------------------------------------
std::unordered_map<id_type, worker_ptr> workers_by_id_;
std::unordered_map<node_id, worker_ptr> workers_by_node_;
std::unordered_map<uint64_t, worker_ptr> workers_by_timeout_id_;
factory_type factory_;
};
} // namespace caf::net
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include "caf/fwd.hpp"
#include "caf/net/network_socket.hpp"
namespace caf::net {
/// A datagram-oriented network communication endpoint for bidirectional
/// byte transmission.
struct udp_datagram_socket : network_socket {
using super = network_socket;
using super::super;
};
/// Creates a `udp_datagram_socket` bound to given port.
/// @param node ip_endpoint that contains the port to bind to. Pass port '0' to
/// bind to any unused port - The endpoint will be updated with the specific
/// port that was bound.
/// @returns The connected socket or an error.
/// @relates udp_datagram_socket
expected<std::pair<udp_datagram_socket, uint16_t>>
make_udp_datagram_socket(ip_endpoint ep, bool reuse_addr = false);
/// Enables or disables `SIO_UDP_CONNRESET` error on `x`.
/// @relates udp_datagram_socket
error allow_connreset(udp_datagram_socket x, bool new_value);
/// Receives the next datagram on socket `x`.
/// @param x The UDP socket for receiving datagrams.
/// @param buf Writable output buffer.
/// @returns The number of received bytes and the sender as `ip_endpoint` on
/// success, an error code otherwise.
/// @relates udp_datagram_socket
/// @post buf was modified and the resulting integer represents the length of
/// the received datagram, even if it did not fit into the given buffer.
variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
span<byte> buf);
/// Sends the content of `bufs` as a datagram to the endpoint `ep` on socket
/// `x`.
/// @param x The UDP socket for sending datagrams.
/// @param bufs Points to the datagram to send, scattered across up to 10
/// buffers.
/// @param ep The enpoint to send the datagram to.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates udp_datagram_socket
/// @pre `bufs.size() < 10`
variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>*> bufs,
ip_endpoint ep);
/// Sends the content of `buf` as a datagram to the endpoint `ep` on socket `x`.
/// @param x The UDP socket for sending datagrams.
/// @param buf The buffer to send.
/// @param ep The enpoint to send the datagram to.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates udp_datagram_socket
variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf,
ip_endpoint ep);
/// Converts the result from I/O operation on a ::udp_datagram_socket to either
/// an error code or a non-zero positive integer.
/// @relates udp_datagram_socket
variant<size_t, sec>
check_udp_datagram_socket_io_res(std::make_signed<size_t>::type res);
} // namespace caf::net
......@@ -22,51 +22,31 @@
#include "caf/expected.hpp"
#include "caf/logger.hpp"
namespace caf {
namespace net {
namespace caf::net {
actor_proxy_impl::actor_proxy_impl(actor_config& cfg, endpoint_manager_ptr dst)
: super(cfg), sf_(dst->serialize_fun()), dst_(std::move(dst)) {
// anon_send(broker_, monitor_atom::value, ctrl());
CAF_ASSERT(dst_ != nullptr);
dst_->enqueue_event(node(), id());
}
actor_proxy_impl::~actor_proxy_impl() {
// anon_send(broker_, make_message(delete_atom::value, node(), id()));
// nop
}
void actor_proxy_impl::enqueue(mailbox_element_ptr what, execution_unit*) {
CAF_PUSH_AID(0);
CAF_ASSERT(what != nullptr);
CAF_LOG_SEND_EVENT(what);
if (auto payload = sf_(home_system(), what->content()))
dst_->enqueue(std::move(what), std::move(*payload));
dst_->enqueue(std::move(what), ctrl(), std::move(*payload));
else
CAF_LOG_ERROR(
"unable to serialize payload: " << home_system().render(payload.error()));
}
bool actor_proxy_impl::add_backlink(abstract_actor* x) {
if (monitorable_actor::add_backlink(x)) {
enqueue(make_mailbox_element(ctrl(), make_message_id(), {},
link_atom::value, x->ctrl()),
nullptr);
return true;
}
return false;
}
bool actor_proxy_impl::remove_backlink(abstract_actor* x) {
if (monitorable_actor::remove_backlink(x)) {
enqueue(make_mailbox_element(ctrl(), make_message_id(), {},
unlink_atom::value, x->ctrl()),
nullptr);
return true;
}
return false;
}
void actor_proxy_impl::kill_proxy(execution_unit* ctx, error rsn) {
cleanup(std::move(rsn), ctx);
}
} // namespace net
} // namespace caf
} // namespace caf::net
This diff is collapsed.
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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/detail/convert_ip_endpoint.hpp"
#include "caf/error.hpp"
#include "caf/ipv4_endpoint.hpp"
#include "caf/ipv6_endpoint.hpp"
#include "caf/sec.hpp"
namespace caf::detail {
void convert(const ip_endpoint& src, sockaddr_storage& dst) {
memset(&dst, 0, sizeof(sockaddr_storage));
if (src.address().embeds_v4()) {
auto sockaddr4 = reinterpret_cast<sockaddr_in*>(&dst);
sockaddr4->sin_family = AF_INET;
sockaddr4->sin_port = ntohs(src.port());
sockaddr4->sin_addr.s_addr = src.address().embedded_v4().bits();
} else {
auto sockaddr6 = reinterpret_cast<sockaddr_in6*>(&dst);
sockaddr6->sin6_family = AF_INET6;
sockaddr6->sin6_port = ntohs(src.port());
memcpy(&sockaddr6->sin6_addr, src.address().bytes().data(),
src.address().bytes().size());
}
}
error convert(const sockaddr_storage& src, ip_endpoint& dst) {
if (src.ss_family == AF_INET) {
auto sockaddr4 = reinterpret_cast<const sockaddr_in&>(src);
ipv4_address ipv4_addr;
memcpy(ipv4_addr.data().data(), &sockaddr4.sin_addr, ipv4_addr.size());
dst = ip_endpoint{ipv4_addr, htons(sockaddr4.sin_port)};
} else if (src.ss_family == AF_INET6) {
auto sockaddr6 = reinterpret_cast<const sockaddr_in6&>(src);
ipv6_address ipv6_addr;
memcpy(ipv6_addr.bytes().data(), &sockaddr6.sin6_addr,
ipv6_addr.bytes().size());
dst = ip_endpoint{ipv6_addr, htons(sockaddr6.sin6_port)};
} else {
return sec::invalid_argument;
}
return none;
}
} // namespace caf::detail
......@@ -22,8 +22,7 @@
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/logger.hpp"
namespace caf {
namespace net {
namespace caf::net {
#ifdef CAF_WINDOWS
......@@ -60,5 +59,4 @@ check_datagram_socket_io_res(std::make_signed<size_t>::type res) {
return static_cast<size_t>(res);
}
} // namespace net
} // namespace caf
} // namespace caf::net
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/net/defaults.hpp"
namespace caf::defaults::middleman {
const size_t max_payload_buffers = 100;
const size_t max_header_buffers = 10;
} // namespace caf::defaults::middleman
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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/net/basp/ec.hpp"
#include "caf/atom.hpp"
#include "caf/error.hpp"
#include "caf/string_view.hpp"
namespace caf::net::basp {
error make_error(ec x) {
return {static_cast<uint8_t>(x), atom("basp")};
}
} // namespace caf::net::basp
......@@ -20,78 +20,70 @@
#include "caf/byte.hpp"
#include "caf/intrusive/inbox_result.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"
namespace caf {
namespace net {
endpoint_manager::event::event(std::string path, actor listener)
: value(resolve_request{std::move(path), std::move(listener)}) {
// nop
}
endpoint_manager::event::event(atom_value type, uint64_t id)
: value(timeout{type, id}) {
// nop
}
endpoint_manager::message::message(mailbox_element_ptr msg,
std::vector<byte> payload)
: msg(std::move(msg)), payload(std::move(payload)) {
// nop
}
namespace caf::net {
endpoint_manager::endpoint_manager(socket handle, const multiplexer_ptr& parent,
actor_system& sys)
: super(handle, parent),
sys_(sys),
events_(event_policy{}),
messages_(message_policy{}) {
events_.try_block();
messages_.try_block();
: super(handle, parent), sys_(sys), queue_(unit, unit, unit) {
queue_.try_block();
}
endpoint_manager::~endpoint_manager() {
// nop
}
std::unique_ptr<endpoint_manager::message> endpoint_manager::next_message() {
if (messages_.blocked())
endpoint_manager_queue::message_ptr endpoint_manager::next_message() {
if (queue_.blocked())
return nullptr;
messages_.fetch_more();
auto& q = messages_.queue();
queue_.fetch_more();
auto& q = std::get<1>(queue_.queue().queues());
auto ts = q.next_task_size();
if (ts == 0)
return nullptr;
q.inc_deficit(ts);
auto result = q.next();
if (q.empty())
messages_.try_block();
if (queue_.empty())
queue_.try_block();
return result;
}
void endpoint_manager::resolve(std::string path, actor listener) {
void endpoint_manager::resolve(uri locator, actor listener) {
using intrusive::inbox_result;
auto ptr = new event(std::move(path), std::move(listener));
switch (events_.push_back(ptr)) {
default:
break;
case inbox_result::unblocked_reader:
mask_add(operation::write);
break;
case inbox_result::queue_closed:
anon_send(listener, resolve_atom::value,
make_error(sec::request_receiver_down));
}
using event_type = endpoint_manager_queue::event;
auto ptr = new event_type(std::move(locator), listener);
if (!enqueue(ptr))
anon_send(listener, resolve_atom::value,
make_error(sec::request_receiver_down));
}
void endpoint_manager::enqueue(mailbox_element_ptr msg,
strong_actor_ptr receiver,
std::vector<byte> payload) {
auto ptr = new message(std::move(msg), std::move(payload));
if (messages_.push_back(ptr) == intrusive::inbox_result::unblocked_reader)
mask_add(operation::write);
using message_type = endpoint_manager_queue::message;
auto ptr = new message_type(std::move(msg), std::move(receiver),
std::move(payload));
enqueue(ptr);
}
bool endpoint_manager::enqueue(endpoint_manager_queue::element* ptr) {
switch (queue_.push_back(ptr)) {
case intrusive::inbox_result::success:
return true;
case intrusive::inbox_result::unblocked_reader: {
auto mpx = parent_.lock();
if (mpx) {
mpx->register_writing(this);
return true;
}
return false;
}
default:
return false;
}
}
} // namespace net
} // namespace caf
} // namespace caf::net
......@@ -16,64 +16,55 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/policy/scribe.hpp"
#include "caf/net/basp/header.hpp"
#include <system_error>
#include <cstring>
#include "caf/byte.hpp"
#include "caf/config.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/span.hpp"
namespace caf {
namespace policy {
namespace caf::net::basp {
scribe::scribe(caf::net::stream_socket handle)
: handle_(handle),
max_consecutive_reads_(0),
read_threshold_(1024),
collected_(0),
max_(1024),
rd_flag_(net::receive_policy_flag::exactly),
written_(0) {
// nop
namespace {
void to_bytes_impl(const header& x, byte* ptr) {
*ptr = static_cast<byte>(x.type);
auto payload_len = detail::to_network_order(x.payload_len);
memcpy(ptr + 1, &payload_len, sizeof(payload_len));
auto operation_data = detail::to_network_order(x.operation_data);
memcpy(ptr + 5, &operation_data, sizeof(operation_data));
}
} // namespace
int header::compare(header other) const noexcept {
auto x = to_bytes(*this);
auto y = to_bytes(other);
return memcmp(x.data(), y.data(), header_size);
}
void scribe::prepare_next_read() {
read_buf_.clear();
collected_ = 0;
// This cast does nothing, but prevents a weird compiler error on GCC <= 4.9.
// TODO: remove cast when dropping support for GCC 4.9.
switch (static_cast<net::receive_policy_flag>(rd_flag_)) {
case net::receive_policy_flag::exactly:
if (read_buf_.size() != max_)
read_buf_.resize(max_);
read_threshold_ = max_;
break;
case net::receive_policy_flag::at_most:
if (read_buf_.size() != max_)
read_buf_.resize(max_);
read_threshold_ = 1;
break;
case net::receive_policy_flag::at_least: {
// read up to 10% more, but at least allow 100 bytes more
auto max_size = max_ + std::max<size_t>(100, max_ / 10);
if (read_buf_.size() != max_size)
read_buf_.resize(max_size);
read_threshold_ = max_;
break;
}
}
header header::from_bytes(span<const byte> bytes) {
CAF_ASSERT(bytes.size() >= header_size);
header result;
auto ptr = bytes.data();
result.type = *reinterpret_cast<const message_type*>(ptr);
auto payload_len = *reinterpret_cast<const uint32_t*>(ptr + 1);
result.payload_len = detail::from_network_order(payload_len);
auto operation_data = *reinterpret_cast<const uint64_t*>(ptr + 5);
result.operation_data = detail::from_network_order(operation_data);
return result;
}
void scribe::configure_read(net::receive_policy::config cfg) {
rd_flag_ = cfg.first;
max_ = cfg.second;
prepare_next_read();
std::array<byte, header_size> to_bytes(header x) {
std::array<byte, header_size> result{};
to_bytes_impl(x, result.data());
return result;
}
void scribe::write_packet(span<const byte> buf) {
write_buf_.insert(write_buf_.end(), buf.begin(), buf.end());
void to_bytes(header x, std::vector<byte>& buf) {
buf.resize(header_size);
to_bytes_impl(x, buf.data());
}
} // namespace policy
} // namespace caf
} // namespace caf::net::basp
......@@ -25,8 +25,7 @@
#include "caf/net/socket.hpp"
#include "caf/none.hpp"
namespace caf {
namespace net {
namespace caf::net {
#ifdef CAF_WINDOWS
......@@ -53,5 +52,4 @@ void this_host::cleanup() {
#endif // CAF_WINDOWS
} // namespace net
} // namespace caf
} // namespace caf::net
......@@ -5,7 +5,7 @@
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* Copyright 2011-2019 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 *
......@@ -16,6 +16,8 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/net/ip.hpp"
#include <cstddef>
#include <string>
#include <utility>
......@@ -25,23 +27,24 @@
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/error.hpp"
#include "caf/ip_address.hpp"
#include "caf/ip_subnet.hpp"
#include "caf/ipv4_address.hpp"
#include "caf/logger.hpp"
#include "caf/string_algorithms.hpp"
#include "caf/string_view.hpp"
// clang-format off
#ifdef CAF_WINDOWS
# ifndef _WIN32_WINNT
# define _WIN32_WINNT 0x0600
# endif
# include <iphlpapi.h>
# include <winsock.h>
#else
# include <ifaddrs.h>
# include <net/if.h>
# include <netdb.h>
# include <ifaddrs.h>
# include <sys/ioctl.h>
# include <sys/types.h>
#endif
// clang-format on
#ifndef HOST_NAME_MAX
# define HOST_NAME_MAX 255
......@@ -51,31 +54,111 @@ using std::pair;
using std::string;
using std::vector;
namespace caf {
namespace net {
namespace ip {
namespace caf::net::ip {
namespace {
// Dummy port to resolve empty string with getaddrinfo.
constexpr string_view dummy_port = "42";
constexpr string_view localhost = "localhost";
void* fetch_in_addr(int family, sockaddr* addr) {
if (family == AF_INET)
return &reinterpret_cast<sockaddr_in*>(addr)->sin_addr;
return &reinterpret_cast<sockaddr_in6*>(addr)->sin6_addr;
}
// TODO: Use getnameinfo instead?
int fetch_addr_str(bool get_ipv4, bool get_ipv6, char (&buf)[INET6_ADDRSTRLEN],
sockaddr* addr) {
int fetch_addr_str(char (&buf)[INET6_ADDRSTRLEN], sockaddr* addr) {
if (addr == nullptr)
return AF_UNSPEC;
auto family = addr->sa_family;
auto in_addr = fetch_in_addr(family, addr);
return ((family == AF_INET && get_ipv4) || (family == AF_INET6 && get_ipv6))
return (family == AF_INET || family == AF_INET6)
&& inet_ntop(family, in_addr, buf, INET6_ADDRSTRLEN) == buf
? family
: AF_UNSPEC;
}
#ifdef CAF_WINDOWS
template <class F>
void for_each_adapter(F f, bool is_link_local = false) {
using adapters_ptr = std::unique_ptr<IP_ADAPTER_ADDRESSES, void (*)(void*)>;
ULONG len = 0;
if (GetAdaptersAddresses(AF_UNSPEC, GAA_FLAG_INCLUDE_PREFIX, nullptr, nullptr,
&len)
!= ERROR_BUFFER_OVERFLOW) {
CAF_LOG_ERROR("failed to get adapter addresses buffer length");
return;
}
auto adapters = adapters_ptr{reinterpret_cast<IP_ADAPTER_ADDRESSES*>(
::malloc(len)),
free};
if (!adapters) {
CAF_LOG_ERROR("malloc failed");
return;
}
// TODO: The Microsoft WIN32 API example propopses to try three times, other
// examples online just perform the call once. If we notice the call to be
// unreliable, we might adapt that behavior.
if (GetAdaptersAddresses(AF_UNSPEC, GAA_FLAG_INCLUDE_PREFIX, nullptr,
adapters.get(), &len)
!= ERROR_SUCCESS) {
CAF_LOG_ERROR("failed to get adapter addresses");
return;
}
char ip_buf[INET6_ADDRSTRLEN];
char name_buf[HOST_NAME_MAX];
std::vector<std::pair<std::string, ip_address>> interfaces;
for (auto* it = adapters.get(); it != nullptr; it = it->Next) {
memset(name_buf, 0, HOST_NAME_MAX);
WideCharToMultiByte(CP_ACP, 0, it->FriendlyName, wcslen(it->FriendlyName),
name_buf, HOST_NAME_MAX, nullptr, nullptr);
std::string name{name_buf};
for (auto* addr = it->FirstUnicastAddress; addr != nullptr;
addr = addr->Next) {
memset(ip_buf, 0, INET6_ADDRSTRLEN);
getnameinfo(addr->Address.lpSockaddr, addr->Address.iSockaddrLength,
ip_buf, sizeof(ip_buf), nullptr, 0, NI_NUMERICHOST);
ip_address ip;
if (!is_link_local && starts_with(ip_buf, "fe80:")) {
CAF_LOG_DEBUG("skipping link-local address: " << ip_buf);
continue;
} else if (auto err = parse(ip_buf, ip))
continue;
f(name_buf, ip);
}
}
}
#else // CAF_WINDOWS
template <class F>
void for_each_adapter(F f, bool is_link_local = false) {
ifaddrs* tmp = nullptr;
if (getifaddrs(&tmp) != 0)
return;
std::unique_ptr<ifaddrs, decltype(freeifaddrs)*> addrs{tmp, freeifaddrs};
char buffer[INET6_ADDRSTRLEN];
std::vector<std::pair<std::string, ip_address>> interfaces;
for (auto i = addrs.get(); i != nullptr; i = i->ifa_next) {
auto family = fetch_addr_str(buffer, i->ifa_addr);
if (family != AF_UNSPEC) {
ip_address ip;
if (!is_link_local && starts_with(buffer, "fe80:")) {
CAF_LOG_DEBUG("skipping link-local address: " << buffer);
continue;
} else if (auto err = parse(buffer, ip)) {
CAF_LOG_ERROR("could not parse into ip address " << buffer);
continue;
}
f({i->ifa_name, strlen(i->ifa_name)}, ip);
}
}
}
#endif // CAF_WINDOWS
} // namespace
std::vector<ip_address> resolve(string_view host) {
......@@ -87,13 +170,15 @@ std::vector<ip_address> resolve(string_view host) {
hint.ai_flags = AI_PASSIVE;
addrinfo* tmp = nullptr;
std::string host_str{host.begin(), host.end()};
if (getaddrinfo(host_str.c_str(), nullptr, &hint, &tmp) != 0)
if (getaddrinfo(host.empty() ? nullptr : host_str.c_str(),
host.empty() ? dummy_port.data() : nullptr, &hint, &tmp)
!= 0)
return {};
std::unique_ptr<addrinfo, decltype(freeaddrinfo)*> addrs{tmp, freeaddrinfo};
char buffer[INET6_ADDRSTRLEN];
std::vector<ip_address> results;
for (auto i = addrs.get(); i != nullptr; i = i->ai_next) {
auto family = fetch_addr_str(true, true, buffer, i->ai_addr);
auto family = fetch_addr_str(buffer, i->ai_addr);
if (family != AF_UNSPEC) {
ip_address ip;
if (auto err = parse(buffer, ip))
......@@ -110,27 +195,58 @@ std::vector<ip_address> resolve(string_view host) {
return results;
}
#ifdef CAF_WINDOWS
std::vector<ip_address> resolve(ip_address host) {
return resolve(to_string(host));
}
std::string hostname() {
TCHAR buf[MAX_COMPUTERNAME_LENGTH + 1];
DWORD size = MAX_COMPUTERNAME_LENGTH;
GetComputerName(buf, &size);
return buf;
std::vector<ip_address> local_addresses(string_view host) {
ip_address host_ip;
std::vector<ip_address> results;
if (host.empty()) {
for_each_adapter(
[&](string_view, ip_address ip) { results.push_back(ip); });
} else if (host == localhost) {
auto v6_local = ip_address{{0}, {0x1}};
auto v4_local = ip_address{make_ipv4_address(127, 0, 0, 1)};
for_each_adapter([&](string_view, ip_address ip) {
if (ip == v4_local || ip == v6_local)
results.push_back(ip);
});
} else if (auto err = parse(host, host_ip)) {
for_each_adapter([&](string_view iface, ip_address ip) {
if (iface == host)
results.push_back(ip);
});
} else {
return local_addresses(host_ip);
}
return results;
}
#else // CAF_WINDOWS
std::vector<ip_address> local_addresses(ip_address host) {
static auto v6_any = ip_address{{0}, {0}};
static auto v4_any = ip_address{make_ipv4_address(0, 0, 0, 0)};
if (host == v4_any || host == v6_any)
return resolve("");
auto link_local = ip_address({0xfe, 0x8, 0x0, 0x0}, {0x0, 0x0, 0x0, 0x0});
auto ll_prefix = ip_subnet(link_local, 10);
// Unless explicitly specified we are going to skip link-local addresses.
auto is_link_local = ll_prefix.contains(host);
std::vector<ip_address> results;
for_each_adapter(
[&](string_view, ip_address ip) {
if (host == ip)
results.push_back(ip);
},
is_link_local);
return results;
}
std::string hostname() {
char buf[HOST_NAME_MAX + 1];
buf[HOST_NAME_MAX] = '\0';
gethostname(buf, HOST_NAME_MAX);
gethostbyname(buf);
return buf;
}
#endif // CAF_WINDOWS
} // namespace ip
} // namespace net
} // namespace caf
} // namespace caf::net::ip
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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/net/basp/message_queue.hpp"
#include <iterator>
namespace caf::net::basp {
message_queue::message_queue() : next_id(0), next_undelivered(0) {
// nop
}
void message_queue::push(execution_unit* ctx, uint64_t id,
strong_actor_ptr receiver,
mailbox_element_ptr content) {
std::unique_lock<std::mutex> guard{lock};
CAF_ASSERT(id >= next_undelivered);
CAF_ASSERT(id < next_id);
auto first = pending.begin();
auto last = pending.end();
if (id == next_undelivered) {
// Dispatch current head.
if (receiver != nullptr)
receiver->enqueue(std::move(content), ctx);
auto next = id + 1;
// Check whether we can deliver more.
if (first == last || first->id != next) {
next_undelivered = next;
CAF_ASSERT(next_undelivered <= next_id);
return;
}
// Deliver everything until reaching a non-consecutive ID or the end.
auto i = first;
for (; i != last && i->id == next; ++i, ++next)
if (i->receiver != nullptr)
i->receiver->enqueue(std::move(i->content), ctx);
next_undelivered = next;
pending.erase(first, i);
CAF_ASSERT(next_undelivered <= next_id);
return;
}
// Get the insertion point.
auto pred = [&](const actor_msg& x) { return x.id >= id; };
pending.emplace(std::find_if(first, last, pred),
actor_msg{id, std::move(receiver), std::move(content)});
}
void message_queue::drop(execution_unit* ctx, uint64_t id) {
push(ctx, id, nullptr, nullptr);
}
uint64_t message_queue::new_id() {
std::unique_lock<std::mutex> guard{lock};
return next_id++;
}
} // namespace caf::net::basp
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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/net/middleman_backend.hpp"
namespace caf::net {
middleman_backend::middleman_backend(std::string id) : id_(std::move(id)) {
// nop
}
middleman_backend::~middleman_backend() {
// nop
}
} // namespace caf::net
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.
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.
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.
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