Unverified Commit 866dc9b3 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #86

Revise layering API, add actor_shell abstraction, add WebSocket layer
parents 18776a84 c582e2db
...@@ -31,6 +31,7 @@ option(CAF_INC_ENABLE_STANDALONE_BUILD ...@@ -31,6 +31,7 @@ option(CAF_INC_ENABLE_STANDALONE_BUILD
option(CAF_INC_ENABLE_TESTING "Build unit test suites" ON) option(CAF_INC_ENABLE_TESTING "Build unit test suites" ON)
option(CAF_INC_ENABLE_NET_MODULE "Build networking module" ON) option(CAF_INC_ENABLE_NET_MODULE "Build networking module" ON)
option(CAF_INC_ENABLE_BB_MODULE "Build building blocks module" ON) option(CAF_INC_ENABLE_BB_MODULE "Build building blocks module" ON)
option(CAF_INC_ENABLE_EXAMPLES "Build small programs" ON)
# -- incubator options with non-boolean values --------------------------------- # -- incubator options with non-boolean values ---------------------------------
...@@ -49,7 +50,7 @@ if(CAF_INC_ENABLE_STANDALONE_BUILD) ...@@ -49,7 +50,7 @@ if(CAF_INC_ENABLE_STANDALONE_BUILD)
FetchContent_Declare( FetchContent_Declare(
actor_framework actor_framework
GIT_REPOSITORY https://github.com/actor-framework/actor-framework.git GIT_REPOSITORY https://github.com/actor-framework/actor-framework.git
GIT_TAG 9caa83be4 GIT_TAG 4f8609b
) )
FetchContent_Populate(actor_framework) FetchContent_Populate(actor_framework)
set(CAF_ENABLE_EXAMPLES OFF CACHE BOOL "" FORCE) set(CAF_ENABLE_EXAMPLES OFF CACHE BOOL "" FORCE)
...@@ -208,3 +209,7 @@ endif() ...@@ -208,3 +209,7 @@ endif()
if(CAF_INC_ENABLE_NET_MODULE) if(CAF_INC_ENABLE_NET_MODULE)
add_subdirectory(libcaf_bb) add_subdirectory(libcaf_bb)
endif() endif()
if(CAF_INC_ENABLE_EXAMPLES)
add_subdirectory(examples)
endif()
...@@ -36,6 +36,7 @@ Convenience options: ...@@ -36,6 +36,7 @@ Convenience options:
--build-type=Debug --build-type=Debug
--sanitizers=address,undefined --sanitizers=address,undefined
--enable-utility-targets --enable-utility-targets
--enable-export-compile-commands
Flags (use --enable-<name> to activate and --disable-<name> to deactivate): Flags (use --enable-<name> to activate and --disable-<name> to deactivate):
...@@ -143,6 +144,7 @@ while [ $# -ne 0 ]; do ...@@ -143,6 +144,7 @@ while [ $# -ne 0 ]; do
CMakeBuildType='Debug' CMakeBuildType='Debug'
append_cache_entry CAF_INC_SANITIZERS STRING 'address,undefined' append_cache_entry CAF_INC_SANITIZERS STRING 'address,undefined'
set_build_flag utility-targets ON set_build_flag utility-targets ON
set_build_flag export-compile-commands ON
;; ;;
--enable-*) --enable-*)
set_build_flag $optarg ON set_build_flag $optarg ON
......
add_custom_target(all_examples)
function(add_example folder name)
add_executable(${name} ${folder}/${name}.cpp ${ARGN})
install(FILES ${folder}/${name}.cpp DESTINATION ${CMAKE_INSTALL_DATADIR}/caf/examples/${folder})
add_dependencies(${name} all_examples)
endfunction()
# -- examples for CAF::net -----------------------------------------------------
if(TARGET CAF::net)
function(add_net_example name)
add_example("net" ${name} ${ARGN})
target_link_libraries(${name} CAF::net CAF::core)
endfunction()
add_net_example(web-socket-calculator)
add_net_example(web-socket-feed)
endif()
// This example application wraps a simple calculator actor and allows clients
// to communicate to this worker via JSON-ish messages over a WebSocket
// connection.
//
// To run the server at port 4242 (defaults to 8080):
//
// ~~~
// web-socket-calculator -p 4242
// ~~~
//
// Once started, the application waits for incoming WebSocket connections that
// send text frames. A simple WebSocket client written in Python could look as
// follows:
//
// ~~~(py)
// #!/usr/bin/env python
//
// import asyncio
// import websockets
//
// line1 = '{ values = [ { "@type" = "task::addition", x = 17, y = 8 } ] }\n'
// line2 = '{ values = [ { "@type" = "task::subtraction", x = 17, y = 8 } ] }\n'
//
// async def hello():
// uri = "ws://localhost:8080"
// async with websockets.connect(uri) as websocket:
// await websocket.send(line1)
// print(f"> {line1}")
// response = await websocket.recv()
// print(f"< {response}")
// await websocket.send(line2)
// print(f"> {line2}")
// response = await websocket.recv()
// print(f"< {response}")
//
// asyncio.get_event_loop().run_until_complete(hello())
// ~~~
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/byte_span.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/exec_main.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/net/actor_shell.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/web_socket_server.hpp"
#include "caf/tag/mixed_message_oriented.hpp"
#include <cstdint>
// -- custom message types -----------------------------------------------------
// Usually, we prefer atoms to prefix certain operations. However, using custom
// message types provides a nicer interface for the text-based WebSocket
// communication.
namespace task {
struct addition {
int32_t x;
int32_t y;
};
template <class Inspector>
bool inspect(Inspector& f, addition& x) {
return f.object(x).fields(f.field("x", x.x), f.field("y", x.y));
}
struct subtraction {
int32_t x;
int32_t y;
};
template <class Inspector>
bool inspect(Inspector& f, subtraction& x) {
return f.object(x).fields(f.field("x", x.x), f.field("y", x.y));
}
} // namespace task
CAF_BEGIN_TYPE_ID_BLOCK(web_socket_calculator, caf::first_custom_type_id)
CAF_ADD_TYPE_ID(web_socket_calculator, (task::addition))
CAF_ADD_TYPE_ID(web_socket_calculator, (task::subtraction))
CAF_END_TYPE_ID_BLOCK(web_socket_calculator)
// -- implementation of the calculator actor -----------------------------------
caf::behavior calculator() {
return {
[](task::addition input) { return input.x + input.y; },
[](task::subtraction input) { return input.x - input.y; },
};
}
// -- implementation of the WebSocket application ------------------------------
class app {
public:
// -- member types -----------------------------------------------------------
// We expect a stream-oriented interface to the lower communication layers.
using input_tag = caf::tag::mixed_message_oriented;
// -- constants --------------------------------------------------------------
// Restricts our buffer to a maximum of 1MB.
static constexpr size_t max_buf_size = 1024 * 1024;
// -- constructors, destructors, and assignment operators --------------------
explicit app(caf::actor worker) : worker_(std::move(worker)) {
// nop
}
template <class LowerLayerPtr>
caf::error init(caf::net::socket_manager* mgr, LowerLayerPtr down,
const caf::settings&) {
buf_.reserve(max_buf_size);
// Create the actor-shell wrapper for this application.
self_ = mgr->make_actor_shell(down);
std::cout << "*** established new connection on socket "
<< down->handle().id << "\n";
return caf::none;
}
// -- mixed_message_oriented interface functions -----------------------------
template <class LowerLayerPtr>
bool prepare_send(LowerLayerPtr down) {
// The lower level calls this function whenever a send event occurs, but
// before performing any socket I/O operation. We use this as a hook for
// constantly checking our mailbox. Returning `false` here aborts the
// application and closes the socket.
while (self_->consume_message()) {
// We set abort_reason in our response handlers in case of an error.
if (down->abort_reason())
return false;
// else: repeat.
}
return true;
}
template <class LowerLayerPtr>
bool done_sending(LowerLayerPtr) {
// The lower level calls this function to check whether we are ready to
// unregister our socket from send events. We must make sure to put our
// mailbox in to the blocking state in order to get re-registered once new
// messages arrive.
return self_->try_block_mailbox();
}
template <class LowerLayerPtr>
void abort(LowerLayerPtr down, const caf::error& reason) {
std::cout << "*** lost connection on socket " << down->handle().id << ": "
<< to_string(reason) << "\n";
}
template <class LowerLayerPtr>
ptrdiff_t consume_text(LowerLayerPtr down, caf::string_view text) {
// The other functions in this class provide mostly boilerplate code. Here
// comes our main logic. In this function, we receive a text data frame from
// the WebSocket. We don't assume that the client sends one line per
// frame, so we buffer all incoming text until finding a newline character.
if (buf_.size() + text.size() > max_buf_size) {
auto err = caf::make_error(caf::sec::runtime_error,
"exceeded max text buffer size");
down->abort_reason(std::move(err));
return -1;
}
buf_.insert(buf_.end(), text.begin(), text.end());
auto nl_pos = [this] { return std::find(buf_.begin(), buf_.end(), '\n'); };
for (auto i = nl_pos(); i != buf_.end(); i = nl_pos()) {
// Skip empty lines.
if (i == buf_.begin()) {
buf_.erase(buf_.begin(), buf_.begin() + 1);
continue;
}
// Deserialize config value / message from received line.
auto num_bytes = std::distance(buf_.begin(), i) + 1;
caf::string_view line{buf_.data(), static_cast<size_t>(num_bytes) - 1};
std::cout << "*** [socket " << down->handle().id << "] INPUT: " << line
<< "\n";
caf::config_value val;
if (auto parsed_val = caf::config_value::parse(line)) {
val = std::move(*parsed_val);
} else {
down->abort_reason(std::move(parsed_val.error()));
return -1;
}
caf::config_value_reader reader{&val};
caf::message msg;
if (!reader.apply_object(msg)) {
down->abort_reason(reader.get_error());
return -1;
}
// Dispatch message to worker.
self_->request(worker_, std::chrono::seconds{1}, std::move(msg))
.then(
[this, down](int32_t value) {
// Simply respond with the value as string, wrapped into a WebSocket
// text message frame.
auto str_response = std::to_string(value);
std::cout << "*** [socket " << down->handle().id
<< "] OUTPUT: " << str_response << "\n";
str_response += '\n';
down->begin_text_message();
auto& buf = down->text_message_buffer();
buf.insert(buf.end(), str_response.begin(), str_response.end());
down->end_text_message();
},
[down](caf::error& err) { down->abort_reason(std::move(err)); });
// Erase consumed data from the buffer.
buf_.erase(buf_.begin(), i + 1);
}
return static_cast<ptrdiff_t>(text.size());
}
template <class LowerLayerPtr>
ptrdiff_t consume_binary(LowerLayerPtr down, caf::byte_span) {
// Reject binary input.
auto err = caf::make_error(caf::sec::runtime_error,
"received binary WebSocket frame (unsupported)");
down->abort_reason(std::move(err));
return -1;
}
private:
// Caches incoming text data until finding a newline character.
std::vector<char> buf_;
// Stores a handle to our worker.
caf::actor worker_;
// Enables the application to send and receive actor messages.
caf::net::actor_shell_ptr self_;
};
// -- main ---------------------------------------------------------------------
static constexpr uint16_t default_port = 8080;
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<uint16_t>("port,p", "port to listen for incoming connections");
}
};
int caf_main(caf::actor_system& sys, const config& cfg) {
using namespace caf;
using namespace caf::net;
// Open up a TCP port for incoming connections.
auto port = get_or(cfg, "port", default_port);
tcp_accept_socket sock;
if (auto sock_res = make_tcp_accept_socket({ipv4_address{}, port})) {
std::cout << "*** started listening for incoming connections on port "
<< port << '\n';
sock = std::move(*sock_res);
} else {
std::cerr << "*** unable to open port " << port << ": "
<< to_string(sock_res.error()) << '\n';
return EXIT_FAILURE;
}
// Spawn our worker actor and initiate the protocol stack.
auto worker = sys.spawn(calculator);
auto add_conn = [worker](tcp_stream_socket sock, multiplexer* mpx) {
return make_socket_manager<app, web_socket_server, stream_transport>(
sock, mpx, worker);
};
sys.network_manager().make_acceptor(sock, add_conn);
return EXIT_SUCCESS;
}
CAF_MAIN(caf::id_block::web_socket_calculator, caf::net::middleman)
This diff is collapsed.
...@@ -34,8 +34,12 @@ endfunction() ...@@ -34,8 +34,12 @@ endfunction()
# -- add library targets ------------------------------------------------------- # -- add library targets -------------------------------------------------------
add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS} add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS}
src/actor_proxy_impl.cpp #src/actor_proxy_impl.cpp
src/basp/application.cpp #src/basp/application.cpp
#src/endpoint_manager.cpp
#src/net/backend/tcp.cpp
#src/net/backend/test.cpp
#src/net/endpoint_manager_queue.cpp
src/basp/connection_state_strings.cpp src/basp/connection_state_strings.cpp
src/basp/ec_strings.cpp src/basp/ec_strings.cpp
src/basp/message_type_strings.cpp src/basp/message_type_strings.cpp
...@@ -43,16 +47,13 @@ add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS} ...@@ -43,16 +47,13 @@ add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS}
src/convert_ip_endpoint.cpp src/convert_ip_endpoint.cpp
src/datagram_socket.cpp src/datagram_socket.cpp
src/defaults.cpp src/defaults.cpp
src/defaults.cpp src/detail/rfc6455.cpp
src/endpoint_manager.cpp
src/header.cpp src/header.cpp
src/host.cpp src/host.cpp
src/ip.cpp src/ip.cpp
src/message_queue.cpp src/message_queue.cpp
src/multiplexer.cpp src/multiplexer.cpp
src/net/backend/test.cpp src/net/actor_shell.cpp
src/net/backend/tcp.cpp
src/net/endpoint_manager_queue.cpp
src/net/middleman.cpp src/net/middleman.cpp
src/net/middleman_backend.cpp src/net/middleman_backend.cpp
src/net/packet_writer.cpp src/net/packet_writer.cpp
...@@ -80,7 +81,8 @@ set_property(TARGET libcaf_net_obj PROPERTY POSITION_INDEPENDENT_CODE ON) ...@@ -80,7 +81,8 @@ set_property(TARGET libcaf_net_obj PROPERTY POSITION_INDEPENDENT_CODE ON)
caf_net_set_default_properties(libcaf_net_obj libcaf_net) caf_net_set_default_properties(libcaf_net_obj libcaf_net)
target_include_directories(libcaf_net INTERFACE target_include_directories(libcaf_net INTERFACE
$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}>) $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}>
$<BUILD_INTERFACE:${CMAKE_BINARY_DIR}>)
add_library(CAF::net ALIAS libcaf_net) add_library(CAF::net ALIAS libcaf_net)
...@@ -123,30 +125,34 @@ target_include_directories(caf-net-test PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/tes ...@@ -123,30 +125,34 @@ target_include_directories(caf-net-test PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/tes
target_link_libraries(caf-net-test PRIVATE CAF::test) target_link_libraries(caf-net-test PRIVATE CAF::test)
caf_incubator_add_test_suites(caf-net-test caf_incubator_add_test_suites(caf-net-test
net.basp.message_queue
net.basp.ping_pong
net.basp.worker
accept_socket accept_socket
#application
convert_ip_endpoint
datagram_socket
detail.rfc6455
#datagram_transport
#doorman
#endpoint_manager
header
ip
multiplexer
net.actor_shell
#net.backend.tcp
#net.basp.message_queue
#net.basp.ping_pong
#net.basp.worker
net.length_prefix_framing
net.web_socket_server
network_socket
pipe_socket pipe_socket
application
socket socket
convert_ip_endpoint
socket_guard socket_guard
datagram_socket #stream_application
stream_application
datagram_transport
stream_socket stream_socket
doorman
stream_transport stream_transport
endpoint_manager #string_application
string_application
header
tcp_sockets tcp_sockets
ip #transport_worker
transport_worker #transport_worker_dispatcher
multiplexer
transport_worker_dispatcher
udp_datagram_socket udp_datagram_socket
network_socket
net.backend.tcp
) )
This diff is collapsed.
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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/byte.hpp"
#include "caf/detail/net_export.hpp"
#include <cstdint>
#include <vector>
namespace caf::detail {
struct CAF_NET_EXPORT rfc6455 {
// -- member types -----------------------------------------------------------
using binary_buffer = std::vector<byte>;
struct header {
bool fin;
uint8_t opcode;
uint32_t mask_key;
uint64_t payload_len;
};
// -- constants --------------------------------------------------------------
static constexpr uint8_t continuation_frame = 0x00;
static constexpr uint8_t text_frame = 0x01;
static constexpr uint8_t binary_frame = 0x02;
static constexpr uint8_t connection_close = 0x08;
static constexpr uint8_t ping = 0x09;
static constexpr uint8_t pong = 0x0A;
// -- utility functions ------------------------------------------------------
static void mask_data(uint32_t key, span<char> data);
static void mask_data(uint32_t key, span<byte> data);
static void assemble_frame(uint32_t mask_key, span<const char> data,
binary_buffer& out);
static void assemble_frame(uint32_t mask_key, span<const byte> data,
binary_buffer& out);
static void assemble_frame(uint8_t opcode, uint32_t mask_key,
span<const byte> data, binary_buffer& out);
static ptrdiff_t decode_header(span<const byte> data, header& hdr);
};
} // namespace caf::detail
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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_traits.hpp"
#include "caf/callback.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/extend.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/local_actor.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/net/fwd.hpp"
#include "caf/none.hpp"
#include "caf/policy/normal_messages.hpp"
namespace caf::net {
///
class CAF_NET_EXPORT actor_shell
: public extend<local_actor, actor_shell>::with<mixin::sender,
mixin::requester>,
public dynamically_typed_actor_base,
public non_blocking_actor_base {
public:
// -- friends ----------------------------------------------------------------
friend class actor_shell_ptr;
// -- member types -----------------------------------------------------------
using super
= extend<local_actor, actor_shell>::with<mixin::sender, mixin::requester>;
using signatures = none_t;
using behavior_type = behavior;
struct mailbox_policy {
using queue_type = intrusive::drr_queue<policy::normal_messages>;
using deficit_type = policy::normal_messages::deficit_type;
using mapped_type = policy::normal_messages::mapped_type;
using unique_pointer = policy::normal_messages::unique_pointer;
};
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>;
using fallback_handler = unique_callback_ptr<result<message>(message&)>;
// -- constructors, destructors, and assignment operators --------------------
actor_shell(actor_config& cfg, socket_manager* owner);
~actor_shell() override;
// -- state modifiers --------------------------------------------------------
/// Detaches the shell from its owner and closes the mailbox.
void quit(error reason);
/// Overrides the callbacks for incoming messages.
template <class... Fs>
void set_behavior(Fs... fs) {
bhvr_ = behavior{std::move(fs)...};
}
/// Overrides the default handler for unexpected messages.
template <class F>
void set_fallback(F f) {
fallback_ = make_type_erased_callback(std::move(f));
}
// -- mailbox access ---------------------------------------------------------
auto& mailbox() noexcept {
return mailbox_;
}
/// Dequeues and returns the next message from the mailbox or returns
/// `nullptr` if the mailbox is empty.
mailbox_element_ptr next_message();
/// Tries to put the mailbox into the `blocked` state, causing the next
/// enqueue to register the owning socket manager for write events.
bool try_block_mailbox();
// -- message processing -----------------------------------------------------
/// Dequeues and processes the next message from the mailbox.
/// @returns `true` if a message was dequeued and process, `false` if the
/// mailbox was empty.
bool consume_message();
/// Adds a callback for a multiplexed response.
void add_multiplexed_response_handler(message_id response_id, behavior bhvr);
// -- overridden functions of abstract_actor ---------------------------------
using abstract_actor::enqueue;
void enqueue(mailbox_element_ptr ptr, execution_unit* eu) override;
mailbox_element* peek_at_next_mailbox_element() override;
// -- overridden functions of local_actor ------------------------------------
const char* name() const override;
void launch(execution_unit* eu, bool lazy, bool hide) override;
bool cleanup(error&& fail_state, execution_unit* host) override;
private:
// Stores incoming actor messages.
mailbox_type mailbox_;
// Guards access to owner_.
std::mutex owner_mtx_;
// Points to the owning manager (nullptr after quit was called).
socket_manager* owner_;
// Handler for consuming messages from the mailbox.
behavior bhvr_;
// Handler for unexpected messages.
fallback_handler fallback_;
// Stores callbacks for multiplexed responses.
detail::unordered_flat_map<message_id, behavior> multiplexed_responses_;
};
/// An "owning" pointer to an actor shell in the sense that it calls `quit()` on
/// the shell when going out of scope.
class CAF_NET_EXPORT actor_shell_ptr {
public:
friend class socket_manager;
constexpr actor_shell_ptr() noexcept {
// nop
}
constexpr actor_shell_ptr(std::nullptr_t) noexcept {
// nop
}
actor_shell_ptr(actor_shell_ptr&& other) noexcept = default;
actor_shell_ptr& operator=(actor_shell_ptr&& other) noexcept = default;
actor_shell_ptr(const actor_shell_ptr& other) = delete;
actor_shell_ptr& operator=(const actor_shell_ptr& other) = delete;
~actor_shell_ptr();
/// Returns an actor handle to the managed actor shell.
actor as_actor() const noexcept;
void detach(error reason);
actor_shell* get() const noexcept;
actor_shell* operator->() const noexcept {
return get();
}
actor_shell& operator*() const noexcept {
return *get();
}
bool operator!() const noexcept {
return !ptr_;
}
explicit operator bool() const noexcept {
return static_cast<bool>(ptr_);
}
private:
/// @pre `ptr != nullptr`
explicit actor_shell_ptr(strong_actor_ptr ptr) noexcept;
strong_actor_ptr ptr_;
};
} // namespace caf::net
...@@ -23,7 +23,6 @@ ...@@ -23,7 +23,6 @@
#include "caf/net/datagram_transport.hpp" #include "caf/net/datagram_transport.hpp"
#include "caf/net/defaults.hpp" #include "caf/net/defaults.hpp"
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
#include "caf/net/endpoint_manager_impl.hpp"
#include "caf/net/endpoint_manager_queue.hpp" #include "caf/net/endpoint_manager_queue.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/host.hpp" #include "caf/net/host.hpp"
......
...@@ -29,6 +29,7 @@ ...@@ -29,6 +29,7 @@
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/byte_span.hpp"
#include "caf/callback.hpp" #include "caf/callback.hpp"
#include "caf/defaults.hpp" #include "caf/defaults.hpp"
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
...@@ -57,8 +58,6 @@ class CAF_NET_EXPORT application { ...@@ -57,8 +58,6 @@ class CAF_NET_EXPORT application {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using byte_span = span<const byte>;
using hub_type = detail::worker_hub<worker>; using hub_type = detail::worker_hub<worker>;
struct test_tag {}; struct test_tag {};
......
...@@ -83,9 +83,10 @@ CAF_NET_EXPORT void to_bytes(header x, byte_buffer& buf); ...@@ -83,9 +83,10 @@ CAF_NET_EXPORT void to_bytes(header x, byte_buffer& buf);
/// @relates header /// @relates header
template <class Inspector> template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, header& x) { bool inspect(Inspector& f, header& x) {
return f(meta::type_name("basp::header"), x.type, x.payload_len, return f.object(x).fields(f.field("type", x.type),
x.operation_data); f.field("payload_len", x.payload_len),
f.field("operation_data", x.operation_data));
} }
/// @} /// @}
......
...@@ -53,8 +53,9 @@ public: ...@@ -53,8 +53,9 @@ public:
std::vector<strong_actor_ptr> fwd_stack; std::vector<strong_actor_ptr> fwd_stack;
message content; message content;
binary_deserializer source{ctx, payload}; binary_deserializer source{ctx, payload};
if (auto err = source(src_node, src_id, dst_id, fwd_stack, content)) { if (!source.apply_objects(src_node, src_id, dst_id, fwd_stack, content)) {
CAF_LOG_ERROR("could not deserialize payload: " << CAF_ARG(err)); CAF_LOG_ERROR(
"failed to deserialize payload:" << CAF_ARG(source.get_error()));
return; return;
} }
// Sanity checks. // Sanity checks.
......
...@@ -29,102 +29,105 @@ ...@@ -29,102 +29,105 @@
namespace caf::net { namespace caf::net {
/// A doorman accepts TCP connections and creates stream_transports to handle /// A connection_acceptor accepts connections from an accept socket and creates
/// them. /// socket managers to handle them via its factory.
template <class Factory> template <class Socket, class Factory>
class doorman { class connection_acceptor {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using factory_type = Factory; using input_tag = tag::io_event_oriented;
using socket_type = Socket;
using application_type = typename Factory::application_type; using factory_type = Factory;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
explicit doorman(net::tcp_accept_socket acceptor, factory_type factory) template <class... Ts>
: acceptor_(acceptor), factory_(std::move(factory)) { explicit connection_acceptor(Ts&&... xs) : factory_(std::forward<Ts>(xs)...) {
// nop // nop
} }
// -- properties -------------------------------------------------------------
net::tcp_accept_socket handle() {
return acceptor_;
}
// -- member functions ------------------------------------------------------- // -- member functions -------------------------------------------------------
template <class Parent> template <class ParentPtr>
error init(Parent& parent) { error init(socket_manager* owner, ParentPtr parent, const settings& config) {
// TODO: is initializing application factory nessecary? CAF_LOG_TRACE("");
if (auto err = factory_.init(parent)) owner_ = owner;
cfg_ = config;
if (auto err = factory_.init(owner, config))
return err; return err;
parent->register_reading();
return none; return none;
} }
template <class Parent> template <class ParentPtr>
bool handle_read_event(Parent& parent) { bool handle_read_event(ParentPtr parent) {
auto x = net::accept(acceptor_); CAF_LOG_TRACE("");
if (!x) { if (auto x = accept(parent->handle())) {
socket_manager_ptr child = factory_.make(*x, owner_->mpx_ptr());
CAF_ASSERT(child != nullptr);
if (auto err = child->init(cfg_)) {
CAF_LOG_ERROR("failed to initialize new child:" << err);
parent->abort_reason(std::move(err));
return false;
}
return true;
} else {
CAF_LOG_ERROR("accept failed:" << x.error()); CAF_LOG_ERROR("accept failed:" << x.error());
return false; return false;
} }
auto mpx = parent.multiplexer();
if (!mpx) {
CAF_LOG_DEBUG("unable to get multiplexer from parent");
return false;
}
auto child = make_endpoint_manager(
mpx, parent.system(),
stream_transport<application_type>{*x, factory_.make()});
if (auto err = child->init())
return false;
return true;
} }
template <class Parent> template <class ParentPtr>
bool handle_write_event(Parent&) { bool handle_write_event(ParentPtr) {
CAF_LOG_ERROR("doorman received write event"); CAF_LOG_ERROR("connection_acceptor received write event");
return false; return false;
} }
template <class Parent> template <class ParentPtr>
void void abort(ParentPtr, const error& reason) {
resolve(Parent&, [[maybe_unused]] const uri& locator, const actor& listener) { CAF_LOG_ERROR("connection_acceptor aborts due to an error: " << reason);
CAF_LOG_ERROR("doorman called to resolve" << CAF_ARG(locator)); factory_.abort(reason);
anon_send(listener, resolve_atom_v, "doormen cannot resolve paths");
} }
void new_proxy(endpoint_manager&, const node_id& peer, actor_id id) { private:
CAF_LOG_ERROR("doorman received new_proxy" << CAF_ARG(peer) << CAF_ARG(id)); factory_type factory_;
CAF_IGNORE_UNUSED(peer);
CAF_IGNORE_UNUSED(id); socket_manager* owner_;
settings cfg_;
};
/// Converts a function object into a factory object for a
/// @ref connection_acceptor.
template <class FunctionObject>
class connection_acceptor_factory_adapter {
public:
explicit connection_acceptor_factory_adapter(FunctionObject f)
: f_(std::move(f)) {
// nop
} }
void local_actor_down(endpoint_manager&, const node_id& peer, actor_id id, connection_acceptor_factory_adapter(connection_acceptor_factory_adapter&&)
error reason) { = default;
CAF_LOG_ERROR("doorman received local_actor_down"
<< CAF_ARG(peer) << CAF_ARG(id) << CAF_ARG(reason)); error init(socket_manager*, const settings&) {
CAF_IGNORE_UNUSED(peer); return none;
CAF_IGNORE_UNUSED(id);
CAF_IGNORE_UNUSED(reason);
} }
template <class Parent> template <class Socket>
void timeout(Parent&, [[maybe_unused]] const std::string& tag, socket_manager_ptr make(Socket connected_socket, multiplexer* mpx) {
[[maybe_unused]] uint64_t id) { return f_(std::move(connected_socket), mpx);
CAF_LOG_ERROR("doorman received timeout" << CAF_ARG(tag) << CAF_ARG(id));
} }
void handle_error([[maybe_unused]] sec err) { void abort(const error&) {
CAF_LOG_ERROR("doorman encounterd error: " << err); // nop
} }
private: private:
net::tcp_accept_socket acceptor_; FunctionObject f_;
factory_type factory_;
}; };
} // namespace caf::net } // namespace caf::net
...@@ -36,4 +36,9 @@ CAF_NET_EXPORT extern const size_t max_header_buffers; ...@@ -36,4 +36,9 @@ CAF_NET_EXPORT extern const size_t max_header_buffers;
/// Port to listen on for tcp. /// Port to listen on for tcp.
CAF_NET_EXPORT extern const uint16_t tcp_port; CAF_NET_EXPORT extern const uint16_t tcp_port;
/// Caps how much Bytes a stream transport pushes to its write buffer before
/// stopping to read from its message queue. Default TCP send buffer is 16kB (at
/// least on Linux).
constexpr auto stream_output_buf_cap = size_t{32768};
} // namespace caf::defaults::middleman } // namespace caf::defaults::middleman
...@@ -52,10 +52,16 @@ public: ...@@ -52,10 +52,16 @@ public:
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
actor_system& system() { actor_system& system() noexcept {
return sys_; return sys_;
} }
const actor_system_config& config() const noexcept;
// -- queue access -----------------------------------------------------------
bool at_end_of_message_queue();
endpoint_manager_queue::message_ptr next_message(); endpoint_manager_queue::message_ptr next_message();
// -- event management ------------------------------------------------------- // -- event management -------------------------------------------------------
...@@ -75,7 +81,7 @@ public: ...@@ -75,7 +81,7 @@ public:
// -- pure virtual member functions ------------------------------------------ // -- pure virtual member functions ------------------------------------------
/// Initializes the manager before adding it to the multiplexer's event loop. /// Initializes the manager before adding it to the multiplexer's event loop.
virtual error init() = 0; // virtual error init() = 0;
protected: protected:
bool enqueue(endpoint_manager_queue::element* ptr); bool enqueue(endpoint_manager_queue::element* ptr);
......
...@@ -40,8 +40,8 @@ public: ...@@ -40,8 +40,8 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
endpoint_manager_impl(const multiplexer_ptr& parent, actor_system& sys, endpoint_manager_impl(const multiplexer_ptr& parent, actor_system& sys,
Transport trans) socket handle, Transport trans)
: super(trans.handle(), parent, sys), transport_(std::move(trans)) { : super(handle, parent, sys), transport_(std::move(trans)) {
// nop // nop
} }
...@@ -73,7 +73,7 @@ public: ...@@ -73,7 +73,7 @@ public:
// -- interface functions ---------------------------------------------------- // -- interface functions ----------------------------------------------------
error init() override { error init() /*override*/ {
this->register_reading(); this->register_reading();
return transport_.init(*this); return transport_.init(*this);
} }
...@@ -124,6 +124,8 @@ private: ...@@ -124,6 +124,8 @@ private:
/// Stores the id for the next timeout. /// Stores the id for the next timeout.
uint64_t next_timeout_id_; uint64_t next_timeout_id_;
error err_;
}; };
} // namespace caf::net } // namespace caf::net
...@@ -42,6 +42,8 @@ enum class ec : uint8_t; ...@@ -42,6 +42,8 @@ enum class ec : uint8_t;
// -- classes ------------------------------------------------------------------ // -- classes ------------------------------------------------------------------
class actor_shell;
class actor_shell_ptr;
class endpoint_manager; class endpoint_manager;
class middleman; class middleman;
class middleman_backend; class middleman_backend;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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 <cstring>
#include <memory>
#include "caf/byte.hpp"
#include "caf/byte_span.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/error.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/tag/message_oriented.hpp"
#include "caf/tag/stream_oriented.hpp"
namespace caf::net {
/// Length-prefixed message framing for discretizing a Byte stream into messages
/// of varying size. The framing uses 4 Bytes for the length prefix, but
/// messages (including the 4 Bytes for the length prefix) are limited to a
/// maximum size of INT32_MAX. This limitation comes from the POSIX API (recv)
/// on 32-bit platforms.
template <class UpperLayer>
class length_prefix_framing {
public:
using input_tag = tag::stream_oriented;
using output_tag = tag::message_oriented;
using length_prefix_type = uint32_t;
static constexpr size_t max_message_length = INT32_MAX;
// -- interface for the upper layer ------------------------------------------
template <class LowerLayer>
class access {
public:
access(LowerLayer* lower_layer, length_prefix_framing* this_layer)
: lower_layer_(lower_layer), this_layer_(this_layer) {
// nop
}
void begin_message() {
lower_layer_->begin_output();
auto& buf = message_buffer();
message_offset_ = buf.size();
buf.insert(buf.end(), 4, byte{0});
}
byte_buffer& message_buffer() {
return lower_layer_->output_buffer();
}
bool end_message() {
using detail::to_network_order;
auto& buf = message_buffer();
auto msg_begin = buf.begin() + message_offset_;
auto msg_size = std::distance(msg_begin + 4, buf.end());
if (msg_size > 0 && msg_size < max_message_length) {
auto u32_size = to_network_order(static_cast<uint32_t>(msg_size));
memcpy(std::addressof(*msg_begin), &u32_size, 4);
return true;
} else {
abort_reason(make_error(
sec::runtime_error, msg_size == 0 ? "logic error: message of size 0"
: "maximum message size exceeded"));
return false;
}
}
bool can_send_more() const noexcept {
return lower_layer_->can_send_more();
}
void abort_reason(error reason) {
return lower_layer_->abort_reason(std::move(reason));
}
void configure_read(receive_policy policy) {
lower_layer_->configure_read(policy);
}
private:
LowerLayer* lower_layer_;
length_prefix_framing* this_layer_;
size_t message_offset_ = 0;
};
// -- properties -------------------------------------------------------------
auto& upper_layer() noexcept {
return upper_layer_;
}
const auto& upper_layer() const noexcept {
return upper_layer_;
}
// -- role: upper layer ------------------------------------------------------
template <class LowerLayer>
bool prepare_send(LowerLayer& down) {
access<LowerLayer> this_layer{&down, this};
return upper_layer_.prepare_send(this_layer);
}
template <class LowerLayer>
bool done_sending(LowerLayer& down) {
access<LowerLayer> this_layer{&down, this};
return upper_layer_.done_sending(this_layer);
}
template <class LowerLayer>
void abort(LowerLayer& down, const error& reason) {
access<LowerLayer> this_layer{&down, this};
return upper_layer_.abort(this_layer, reason);
}
template <class LowerLayer>
ptrdiff_t consume(LowerLayer& down, byte_span buffer, byte_span) {
using detail::from_network_order;
if (buffer.size() < 4)
return 0;
uint32_t u32_size = 0;
memcpy(&u32_size, buffer.data(), 4);
auto msg_size = static_cast<size_t>(from_network_order(u32_size));
if (buffer.size() < msg_size + 4)
return 0;
upper_layer_.consume(down, make_span(buffer.data() + 4, msg_size));
return msg_size + 4;
}
private:
UpperLayer upper_layer_;
};
} // namespace caf::net
...@@ -26,8 +26,8 @@ ...@@ -26,8 +26,8 @@
namespace caf::net { namespace caf::net {
template <class Transport> template <class Transport>
endpoint_manager_ptr CAF_NET_EXPORT make_endpoint_manager( endpoint_manager_ptr make_endpoint_manager(const multiplexer_ptr& mpx,
const multiplexer_ptr& mpx, actor_system& sys, Transport trans) { actor_system& sys, Transport trans) {
using impl = endpoint_manager_impl<Transport>; using impl = endpoint_manager_impl<Transport>;
return make_counted<impl>(mpx, sys, std::move(trans)); return make_counted<impl>(mpx, sys, std::move(trans));
} }
......
...@@ -27,8 +27,11 @@ ...@@ -27,8 +27,11 @@
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
#include "caf/detail/type_list.hpp" #include "caf/detail/type_list.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/connection_acceptor.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/middleman_backend.hpp" #include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/scoped_actor.hpp" #include "caf/scoped_actor.hpp"
namespace caf::net { namespace caf::net {
...@@ -49,8 +52,32 @@ public: ...@@ -49,8 +52,32 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
explicit middleman(actor_system& sys);
~middleman() override; ~middleman() override;
// -- socket manager functions -----------------------------------------------
///
/// @param sock An accept socket in listening mode. For a TCP socket, this
/// socket must already listen to an address plus port.
/// @param factory
template <class Socket, class Factory>
auto make_acceptor(Socket sock, Factory factory) {
using connected_socket_type = typename Socket::connected_socket_type;
if constexpr (detail::is_callable_with<Factory, connected_socket_type,
multiplexer*>::value) {
connection_acceptor_factory_adapter<Factory> adapter{std::move(factory)};
return make_acceptor(std::move(sock), std::move(adapter));
} else {
using impl = connection_acceptor<Socket, Factory>;
auto ptr = make_socket_manager<impl>(std::move(sock), &mpx_,
std::move(factory));
mpx_.init(ptr);
return ptr;
}
}
// -- interface functions ---------------------------------------------------- // -- interface functions ----------------------------------------------------
void start() override; void start() override;
...@@ -128,7 +155,11 @@ public: ...@@ -128,7 +155,11 @@ public:
return sys_.config(); return sys_.config();
} }
const multiplexer_ptr& mpx() const noexcept { multiplexer& mpx() noexcept {
return mpx_;
}
const multiplexer& mpx() const noexcept {
return mpx_; return mpx_;
} }
...@@ -137,10 +168,6 @@ public: ...@@ -137,10 +168,6 @@ public:
expected<uint16_t> port(string_view scheme) const; expected<uint16_t> port(string_view scheme) const;
private: private:
// -- constructors, destructors, and assignment operators --------------------
explicit middleman(actor_system& sys);
// -- utility functions ------------------------------------------------------ // -- utility functions ------------------------------------------------------
static void create_backends(middleman&, detail::type_list<>) { static void create_backends(middleman&, detail::type_list<>) {
...@@ -159,7 +186,7 @@ private: ...@@ -159,7 +186,7 @@ private:
actor_system& sys_; actor_system& sys_;
/// Stores the global socket I/O multiplexer. /// Stores the global socket I/O multiplexer.
multiplexer_ptr mpx_; multiplexer mpx_;
/// Stores all available backends for managing peers. /// Stores all available backends for managing peers.
middleman_backend_list backends_; middleman_backend_list backends_;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp"
namespace caf::net {
/// Wraps a pointer to a mixed-message-oriented layer with a pointer to its
/// lower layer. Both pointers are then used to implement the interface required
/// for a mixed-message-oriented layer when calling into its upper layer.
template <class Layer, class LowerLayerPtr>
class mixed_message_oriented_layer_ptr {
public:
class access {
public:
access(Layer* layer, LowerLayerPtr down) : lptr_(layer), llptr_(down) {
// nop
}
bool can_send_more() const noexcept {
return lptr_->can_send_more(llptr_);
}
auto handle() const noexcept {
return lptr_->handle(llptr_);
}
void begin_binary_message() {
lptr_->begin_binary_message(llptr_);
}
auto& binary_message_buffer() {
return lptr_->binary_message_buffer(llptr_);
}
void end_binary_message() {
lptr_->end_binary_message(llptr_);
}
void begin_text_message() {
lptr_->begin_text_message(llptr_);
}
auto& text_message_buffer() {
return lptr_->text_message_buffer(llptr_);
}
void end_text_message() {
lptr_->end_text_message(llptr_);
}
void abort_reason(error reason) {
return lptr_->abort_reason(llptr_, std::move(reason));
}
const error& abort_reason() {
return lptr_->abort_reason(llptr_);
}
private:
Layer* lptr_;
LowerLayerPtr llptr_;
};
mixed_message_oriented_layer_ptr(Layer* layer, LowerLayerPtr down)
: access_(layer, down) {
// nop
}
mixed_message_oriented_layer_ptr(const mixed_message_oriented_layer_ptr&)
= default;
explicit operator bool() const noexcept {
return true;
}
access* operator->() const noexcept {
return &access_;
}
access& operator*() const noexcept {
return access_;
}
private:
mutable access access_;
};
template <class Layer, class LowerLayerPtr>
auto make_mixed_message_oriented_layer_ptr(Layer* this_layer,
LowerLayerPtr down) {
using result_t = mixed_message_oriented_layer_ptr<Layer, LowerLayerPtr>;
return result_t{this_layer, down};
}
} // namespace caf::net
...@@ -38,8 +38,7 @@ struct pollfd; ...@@ -38,8 +38,7 @@ struct pollfd;
namespace caf::net { namespace caf::net {
/// Multiplexes any number of ::socket_manager objects with a ::socket. /// Multiplexes any number of ::socket_manager objects with a ::socket.
class CAF_NET_EXPORT multiplexer class CAF_NET_EXPORT multiplexer {
: public std::enable_shared_from_this<multiplexer> {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -49,10 +48,16 @@ public: ...@@ -49,10 +48,16 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
multiplexer(); /// @param parent Points to the owning middleman instance. May be `nullptr`
/// only for the purpose of unit testing if no @ref
/// socket_manager requires access to the @ref middleman or the
/// @ref actor_system.
explicit multiplexer(middleman* parent);
~multiplexer(); ~multiplexer();
// -- initialization ---------------------------------------------------------
error init(); error init();
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
...@@ -63,6 +68,12 @@ public: ...@@ -63,6 +68,12 @@ public:
/// Returns the index of `mgr` in the pollset or `-1`. /// Returns the index of `mgr` in the pollset or `-1`.
ptrdiff_t index_of(const socket_manager_ptr& mgr); ptrdiff_t index_of(const socket_manager_ptr& mgr);
/// Returns the owning @ref middleman instance.
middleman& owner();
/// Returns the enclosing @ref actor_system.
actor_system& system();
// -- thread-safe signaling -------------------------------------------------- // -- thread-safe signaling --------------------------------------------------
/// Registers `mgr` for read events. /// Registers `mgr` for read events.
...@@ -73,6 +84,10 @@ public: ...@@ -73,6 +84,10 @@ public:
/// @thread-safe /// @thread-safe
void register_writing(const socket_manager_ptr& mgr); void register_writing(const socket_manager_ptr& mgr);
/// Registers `mgr` for initialization in the multiplexer's thread.
/// @thread-safe
void init(const socket_manager_ptr& mgr);
/// Closes the pipe for signaling updates to the multiplexer. After closing /// Closes the pipe for signaling updates to the multiplexer. After closing
/// the pipe, calls to `update` no longer have any effect. /// the pipe, calls to `update` no longer have any effect.
/// @thread-safe /// @thread-safe
...@@ -123,20 +138,17 @@ protected: ...@@ -123,20 +138,17 @@ protected:
/// calling `init()`. /// calling `init()`.
std::thread::id tid_; std::thread::id tid_;
/// Used for pushing updates to the multiplexer's thread.
pipe_socket write_handle_;
/// Guards `write_handle_`. /// Guards `write_handle_`.
std::mutex write_lock_; std::mutex write_lock_;
/// Signals shutdown has been requested. /// Used for pushing updates to the multiplexer's thread.
bool shutting_down_; pipe_socket write_handle_;
};
/// @relates multiplexer /// Points to the owning middleman.
using multiplexer_ptr = std::shared_ptr<multiplexer>; middleman* owner_;
/// @relates multiplexer /// Signals whether shutdown has been requested.
using weak_multiplexer_ptr = std::weak_ptr<multiplexer>; bool shutting_down_ = false;
};
} // namespace caf::net } // namespace caf::net
...@@ -36,9 +36,19 @@ public: ...@@ -36,9 +36,19 @@ public:
using msg_buf = std::array<byte, sizeof(intptr_t) + 1>; using msg_buf = std::array<byte, sizeof(intptr_t) + 1>;
// -- constants --------------------------------------------------------------
static constexpr uint8_t register_reading_code = 0x00;
static constexpr uint8_t register_writing_code = 0x01;
static constexpr uint8_t init_manager_code = 0x02;
static constexpr uint8_t shutdown_code = 0x04;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
pollset_updater(pipe_socket read_handle, const multiplexer_ptr& parent); pollset_updater(pipe_socket read_handle, multiplexer* parent);
~pollset_updater() override; ~pollset_updater() override;
...@@ -51,6 +61,8 @@ public: ...@@ -51,6 +61,8 @@ public:
// -- interface functions ---------------------------------------------------- // -- interface functions ----------------------------------------------------
error init(const settings& config) override;
bool handle_read_event() override; bool handle_read_event() override;
bool handle_write_event() override; bool handle_write_event() override;
...@@ -59,7 +71,7 @@ public: ...@@ -59,7 +71,7 @@ public:
private: private:
msg_buf buf_; msg_buf buf_;
size_t buf_size_; size_t buf_size_ = 0;
}; };
} // namespace caf::net } // namespace caf::net
...@@ -18,45 +18,32 @@ ...@@ -18,45 +18,32 @@
#pragma once #pragma once
#include <cstddef> #include <cstdint>
#include <string>
#include <utility>
#include "caf/config.hpp"
namespace caf::net { namespace caf::net {
enum class CAF_NET_EXPORT receive_policy_flag : unsigned { struct receive_policy {
at_least, uint32_t min_size;
at_most, uint32_t max_size;
exactly
};
inline std::string to_string(receive_policy_flag x) {
return x == receive_policy_flag::at_least
? "at_least"
: (x == receive_policy_flag::at_most ? "at_most" : "exactly");
}
class CAF_NET_EXPORT receive_policy { /// @pre `min_size > 0`
public: /// @pre `min_size <= max_size`
receive_policy() = delete; static constexpr receive_policy between(uint32_t min_size,
uint32_t max_size) {
using config = std::pair<receive_policy_flag, size_t>; return {min_size, max_size};
}
static config at_least(size_t num_bytes) { /// @pre `size > 0`
CAF_ASSERT(num_bytes > 0); static constexpr receive_policy exactly(uint32_t size) {
return {receive_policy_flag::at_least, num_bytes}; return {size, size};
} }
static config at_most(size_t num_bytes) { static constexpr receive_policy up_to(uint32_t max_size) {
CAF_ASSERT(num_bytes > 0); return {1, max_size};
return {receive_policy_flag::at_most, num_bytes};
} }
static config exactly(size_t num_bytes) { static constexpr receive_policy stop() {
CAF_ASSERT(num_bytes > 0); return {0, 0};
return {receive_policy_flag::exactly, num_bytes};
} }
}; };
......
...@@ -55,9 +55,8 @@ struct CAF_NET_EXPORT socket : detail::comparable<socket> { ...@@ -55,9 +55,8 @@ struct CAF_NET_EXPORT socket : detail::comparable<socket> {
/// @relates socket /// @relates socket
template <class Inspector> template <class Inspector>
typename Inspector::result_type CAF_NET_EXPORT inspect(Inspector& f, bool inspect(Inspector& f, socket& x) {
socket& x) { return f.object(x).fields(f.field("id", x.id));
return f(x.id);
} }
/// Denotes the invalid socket. /// Denotes the invalid socket.
...@@ -65,7 +64,7 @@ constexpr auto invalid_socket = socket{invalid_socket_id}; ...@@ -65,7 +64,7 @@ constexpr auto invalid_socket = socket{invalid_socket_id};
/// Converts between different socket types. /// Converts between different socket types.
template <class To, class From> template <class To, class From>
To CAF_NET_EXPORT socket_cast(From x) { To socket_cast(From x) {
return To{x.id}; return To{x.id};
} }
......
...@@ -18,24 +18,32 @@ ...@@ -18,24 +18,32 @@
#pragma once #pragma once
#include "caf/callback.hpp"
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/make_counted.hpp"
#include "caf/net/actor_shell.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/operation.hpp" #include "caf/net/operation.hpp"
#include "caf/net/socket.hpp" #include "caf/net/socket.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/tag/io_event_oriented.hpp"
namespace caf::net { namespace caf::net {
/// Manages the lifetime of a single socket and handles any I/O events on it. /// Manages the lifetime of a single socket and handles any I/O events on it.
class CAF_NET_EXPORT socket_manager : public ref_counted { class CAF_NET_EXPORT socket_manager : public ref_counted {
public: public:
// -- member types -----------------------------------------------------------
using fallback_handler = unique_callback_ptr<result<message>(message&)>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
/// @pre `parent != nullptr`
/// @pre `handle != invalid_socket` /// @pre `handle != invalid_socket`
socket_manager(socket handle, const multiplexer_ptr& parent); /// @pre `parent != nullptr`
socket_manager(socket handle, multiplexer* parent);
~socket_manager() override; ~socket_manager() override;
...@@ -46,13 +54,28 @@ public: ...@@ -46,13 +54,28 @@ public:
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
/// Returns the managed socket. /// Returns the managed socket.
socket handle() const noexcept { socket handle() const {
return handle_; return handle_;
} }
/// Returns a pointer to the multiplexer running this `socket_manager`. /// Returns the owning @ref multiplexer instance.
multiplexer_ptr multiplexer() const { multiplexer& mpx() noexcept {
return parent_.lock(); return *parent_;
}
/// Returns the owning @ref multiplexer instance.
const multiplexer& mpx() const noexcept {
return *parent_;
}
/// Returns a pointer to the owning @ref multiplexer instance.
multiplexer* mpx_ptr() noexcept {
return parent_;
}
/// Returns a pointer to the owning @ref multiplexer instance.
const multiplexer* mpx_ptr() const noexcept {
return parent_;
} }
/// Returns registered operations (read, write, or both). /// Returns registered operations (read, write, or both).
...@@ -70,6 +93,38 @@ public: ...@@ -70,6 +93,38 @@ public:
/// @pre `flag != operation::none` /// @pre `flag != operation::none`
bool mask_del(operation flag) noexcept; bool mask_del(operation flag) noexcept;
const error& abort_reason() const noexcept {
return abort_reason_;
}
void abort_reason(error reason) noexcept {
abort_reason_ = std::move(reason);
}
template <class... Ts>
const error& abort_reason_or(Ts&&... xs) {
if (!abort_reason_)
abort_reason_ = make_error(std::forward<Ts>(xs)...);
return abort_reason_;
}
// -- factories --------------------------------------------------------------
template <class LowerLayerPtr, class FallbackHandler>
actor_shell_ptr make_actor_shell(LowerLayerPtr, FallbackHandler f) {
auto ptr = make_actor_shell_impl();
ptr->set_fallback(std::move(f));
return ptr;
}
template <class LowerLayerPtr>
actor_shell_ptr make_actor_shell(LowerLayerPtr down) {
return make_actor_shell(down, [down](message& msg) -> result<message> {
down->abort_reason(make_error(sec::unexpected_message, std::move(msg)));
return make_error(sec::unexpected_message);
});
}
// -- event loop management -------------------------------------------------- // -- event loop management --------------------------------------------------
void register_reading(); void register_reading();
...@@ -78,6 +133,8 @@ public: ...@@ -78,6 +133,8 @@ public:
// -- pure virtual member functions ------------------------------------------ // -- pure virtual member functions ------------------------------------------
virtual error init(const settings& config) = 0;
/// Called whenever the socket received new data. /// Called whenever the socket received new data.
virtual bool handle_read_event() = 0; virtual bool handle_read_event() = 0;
...@@ -95,9 +152,140 @@ protected: ...@@ -95,9 +152,140 @@ protected:
operation mask_; operation mask_;
weak_multiplexer_ptr parent_; multiplexer* parent_;
error abort_reason_;
private:
actor_shell_ptr make_actor_shell_impl();
}; };
template <class Protocol>
class socket_manager_impl : public socket_manager {
public:
// -- member types -----------------------------------------------------------
using output_tag = tag::io_event_oriented;
using socket_type = typename Protocol::socket_type;
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
socket_manager_impl(socket_type handle, multiplexer* mpx, Ts&&... xs)
: socket_manager(handle, mpx), protocol_(std::forward<Ts>(xs)...) {
// nop
}
// -- initialization ---------------------------------------------------------
error init(const settings& config) override {
CAF_LOG_TRACE("");
if (auto err = nonblocking(handle(), true)) {
CAF_LOG_ERROR("failed to set nonblocking flag in socket:" << err);
return err;
}
return protocol_.init(static_cast<socket_manager*>(this), this, config);
}
// -- properties -------------------------------------------------------------
/// Returns the managed socket.
socket_type handle() const {
return socket_cast<socket_type>(this->handle_);
}
// -- event callbacks --------------------------------------------------------
bool handle_read_event() override {
return protocol_.handle_read_event(this);
}
bool handle_write_event() override {
return protocol_.handle_write_event(this);
}
void handle_error(sec code) override {
abort_reason_ = code;
return protocol_.abort(this, abort_reason_);
}
auto& protocol() noexcept {
return protocol_;
}
const auto& protocol() const noexcept {
return protocol_;
}
auto& top_layer() noexcept {
return climb(protocol_);
}
const auto& top_layer() const noexcept {
return climb(protocol_);
}
private:
template <class FinalLayer>
static FinalLayer& climb(FinalLayer& layer) {
return layer;
}
template <class FinalLayer>
static const FinalLayer& climb(const FinalLayer& layer) {
return layer;
}
template <template <class> class Layer, class Next>
static auto& climb(Layer<Next>& layer) {
return climb(layer.upper_layer());
}
template <template <class> class Layer, class Next>
static const auto& climb(const Layer<Next>& layer) {
return climb(layer.upper_layer());
}
Protocol protocol_;
};
/// @relates socket_manager
using socket_manager_ptr = intrusive_ptr<socket_manager>; using socket_manager_ptr = intrusive_ptr<socket_manager>;
template <class B, template <class> class... Layers>
struct make_socket_manager_helper;
template <class B>
struct make_socket_manager_helper<B> {
using type = B;
};
template <class B, template <class> class Layer,
template <class> class... Layers>
struct make_socket_manager_helper<B, Layer, Layers...>
: make_socket_manager_helper<Layer<B>, Layers...> {
using upper_input = typename B::input_tag;
using lower_output = typename Layer<B>::output_tag;
static_assert(std::is_same<upper_input, lower_output>::value);
};
template <class B, template <class> class... Layers>
using make_socket_manager_helper_t =
typename make_socket_manager_helper<B, Layers...>::type;
template <class B, template <class> class... Layers>
using socket_type_t =
typename make_socket_manager_helper_t<B, Layers...,
socket_manager_impl>::socket_type;
template <class App, template <class> class... Layers, class... Ts>
auto make_socket_manager(socket_type_t<App, Layers...> handle, multiplexer* mpx,
Ts&&... xs) {
using impl
= make_socket_manager_helper_t<App, Layers..., socket_manager_impl>;
static_assert(std::is_base_of<socket, typename impl::socket_type>::value);
return make_counted<impl>(std::move(handle), mpx, std::forward<Ts>(xs)...);
}
} // namespace caf::net } // namespace caf::net
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp"
namespace caf::net {
/// Wraps a pointer to a stream-oriented layer with a pointer to its lower
/// layer. Both pointers are then used to implement the interface required for a
/// stream-oriented layer when calling into its upper layer.
template <class Layer, class LowerLayerPtr>
class stream_oriented_layer_ptr {
public:
class access {
public:
access(Layer* layer, LowerLayerPtr down) : lptr_(layer), llptr_(down) {
// nop
}
bool can_send_more() const noexcept {
return lptr_->can_send_more(llptr_);
}
auto handle() const noexcept {
return lptr_->handle(llptr_);
}
void begin_output() {
lptr_->begin_output(llptr_);
}
byte_buffer& output_buffer() {
return lptr_->output_buffer(llptr_);
}
void end_output() {
lptr_->end_output(llptr_);
}
void abort_reason(error reason) {
return lptr_->abort_reason(llptr_, std::move(reason));
}
const error& abort_reason() {
return lptr_->abort_reason(llptr_);
}
void configure_read(receive_policy policy) {
lptr_->configure_read(llptr_, policy);
}
private:
Layer* lptr_;
LowerLayerPtr llptr_;
};
stream_oriented_layer_ptr(Layer* layer, LowerLayerPtr down)
: access_(layer, down) {
// nop
}
stream_oriented_layer_ptr(const stream_oriented_layer_ptr&) = default;
explicit operator bool() const noexcept {
return true;
}
access* operator->() const noexcept {
return &access_;
}
access& operator*() const noexcept {
return access_;
}
private:
mutable access access_;
};
template <class Layer, class LowerLayerPtr>
auto make_stream_oriented_layer_ptr(Layer* this_layer, LowerLayerPtr down) {
return stream_oriented_layer_ptr<Layer, LowerLayerPtr>{this_layer, down};
}
} // namespace caf::net
This diff is collapsed.
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/network_socket.hpp" #include "caf/net/network_socket.hpp"
#include "caf/uri.hpp" #include "caf/uri.hpp"
...@@ -31,6 +32,8 @@ struct CAF_NET_EXPORT tcp_accept_socket : network_socket { ...@@ -31,6 +32,8 @@ struct CAF_NET_EXPORT tcp_accept_socket : network_socket {
using super = network_socket; using super = network_socket;
using super::super; using super::super;
using connected_socket_type = tcp_stream_socket;
}; };
/// Creates a new TCP socket to accept connections on a given port. /// Creates a new TCP socket to accept connections on a given port.
......
...@@ -173,11 +173,6 @@ public: ...@@ -173,11 +173,6 @@ public:
// -- (pure) virtual functions ----------------------------------------------- // -- (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 /// Called by the endpoint manager when the transport can read data from its
/// socket. /// socket.
virtual bool handle_read_event(endpoint_manager&) = 0; virtual bool handle_read_event(endpoint_manager&) = 0;
......
This diff is collapsed.
This diff is collapsed.
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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
namespace caf::tag {
struct io_event_oriented {};
} // namespace caf::tag
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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
namespace caf::tag {
struct message_oriented {};
} // namespace caf::tag
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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
namespace caf::tag {
struct mixed_message_oriented {};
} // namespace caf::tag
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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
namespace caf::tag {
struct stream_oriented {};
} // namespace caf::tag
...@@ -62,14 +62,15 @@ error application::write_message( ...@@ -62,14 +62,15 @@ error application::write_message(
if (src != nullptr) { if (src != nullptr) {
auto src_id = src->id(); auto src_id = src->id();
system().registry().put(src_id, src); system().registry().put(src_id, src);
if (auto err = sink(src->node(), src_id, dst->id(), ptr->msg->stages)) if (!sink.apply_objects(src->node(), src_id, dst->id(), ptr->msg->stages))
return err; return sink.get_error();
} else { } else {
if (auto err = sink(node_id{}, actor_id{0}, dst->id(), ptr->msg->stages)) if (!sink.apply_objects(node_id{}, actor_id{0}, dst->id(),
return err; ptr->msg->stages))
return sink.get_error();
} }
if (auto err = sink(ptr->msg->content())) if (!sink.apply_objects(ptr->msg->content()))
return err; return sink.get_error();
auto hdr = writer.next_header_buffer(); auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::actor_message, to_bytes(header{message_type::actor_message,
static_cast<uint32_t>(payload_buf.size()), static_cast<uint32_t>(payload_buf.size()),
...@@ -84,8 +85,8 @@ void application::resolve(packet_writer& writer, string_view path, ...@@ -84,8 +85,8 @@ void application::resolve(packet_writer& writer, string_view path,
CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(listener)); CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(listener));
auto payload = writer.next_payload_buffer(); auto payload = writer.next_payload_buffer();
binary_serializer sink{&executor_, payload}; binary_serializer sink{&executor_, payload};
if (auto err = sink(path)) { if (!sink.apply_objects(path)) {
CAF_LOG_ERROR("unable to serialize path" << CAF_ARG(err)); CAF_LOG_ERROR("unable to serialize path:" << sink.get_error());
return; return;
} }
auto req_id = next_request_id_++; auto req_id = next_request_id_++;
...@@ -108,7 +109,7 @@ void application::local_actor_down(packet_writer& writer, actor_id id, ...@@ -108,7 +109,7 @@ void application::local_actor_down(packet_writer& writer, actor_id id,
error reason) { error reason) {
auto payload = writer.next_payload_buffer(); auto payload = writer.next_payload_buffer();
binary_serializer sink{system(), payload}; binary_serializer sink{system(), payload};
if (auto err = sink(reason)) if (!sink.apply_objects(reason))
CAF_RAISE_ERROR("unable to serialize an error"); CAF_RAISE_ERROR("unable to serialize an error");
auto hdr = writer.next_header_buffer(); auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::down_message, to_bytes(header{message_type::down_message,
...@@ -218,8 +219,8 @@ error application::handle_handshake(packet_writer&, header hdr, ...@@ -218,8 +219,8 @@ error application::handle_handshake(packet_writer&, header hdr,
node_id peer_id; node_id peer_id;
std::vector<std::string> app_ids; std::vector<std::string> app_ids;
binary_deserializer source{&executor_, payload}; binary_deserializer source{&executor_, payload};
if (auto err = source(peer_id, app_ids)) if (!source.apply_objects(peer_id, app_ids))
return err; return source.get_error();
if (!peer_id || app_ids.empty()) if (!peer_id || app_ids.empty())
return ec::invalid_handshake; return ec::invalid_handshake;
auto ids = get_or(system().config(), "caf.middleman.app-identifiers", auto ids = get_or(system().config(), "caf.middleman.app-identifiers",
...@@ -277,8 +278,8 @@ error application::handle_resolve_request(packet_writer& writer, header rec_hdr, ...@@ -277,8 +278,8 @@ error application::handle_resolve_request(packet_writer& writer, header rec_hdr,
CAF_ASSERT(rec_hdr.type == message_type::resolve_request); CAF_ASSERT(rec_hdr.type == message_type::resolve_request);
size_t path_size = 0; size_t path_size = 0;
binary_deserializer source{&executor_, received}; binary_deserializer source{&executor_, received};
if (auto err = source.begin_sequence(path_size)) if (!source.begin_sequence(path_size))
return err; return source.get_error();
// We expect the received buffer to contain the path only. // We expect the received buffer to contain the path only.
if (path_size != source.remaining()) if (path_size != source.remaining())
return ec::invalid_payload; return ec::invalid_payload;
...@@ -298,8 +299,8 @@ error application::handle_resolve_request(packet_writer& writer, header rec_hdr, ...@@ -298,8 +299,8 @@ error application::handle_resolve_request(packet_writer& writer, header rec_hdr,
// TODO: figure out how to obtain messaging interface. // TODO: figure out how to obtain messaging interface.
auto payload = writer.next_payload_buffer(); auto payload = writer.next_payload_buffer();
binary_serializer sink{&executor_, payload}; binary_serializer sink{&executor_, payload};
if (auto err = sink(aid, ifs)) if (!sink.apply_objects(aid, ifs))
return err; return sink.get_error();
auto hdr = writer.next_header_buffer(); auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::resolve_response, to_bytes(header{message_type::resolve_response,
static_cast<uint32_t>(payload.size()), static_cast<uint32_t>(payload.size()),
...@@ -323,9 +324,9 @@ error application::handle_resolve_response(packet_writer&, header received_hdr, ...@@ -323,9 +324,9 @@ error application::handle_resolve_response(packet_writer&, header received_hdr,
actor_id aid; actor_id aid;
std::set<std::string> ifs; std::set<std::string> ifs;
binary_deserializer source{&executor_, received}; binary_deserializer source{&executor_, received};
if (auto err = source(aid, ifs)) { if (!source.apply_objects(aid, ifs)) {
anon_send(i->second, sec::remote_lookup_failed); anon_send(i->second, sec::remote_lookup_failed);
return err; return source.get_error();
} }
if (aid == 0) { if (aid == 0) {
anon_send(i->second, strong_actor_ptr{nullptr}, std::move(ifs)); anon_send(i->second, strong_actor_ptr{nullptr}, std::move(ifs));
...@@ -354,8 +355,8 @@ error application::handle_monitor_message(packet_writer& writer, ...@@ -354,8 +355,8 @@ error application::handle_monitor_message(packet_writer& writer,
error reason = exit_reason::unknown; error reason = exit_reason::unknown;
auto payload = writer.next_payload_buffer(); auto payload = writer.next_payload_buffer();
binary_serializer sink{&executor_, payload}; binary_serializer sink{&executor_, payload};
if (auto err = sink(reason)) if (!sink.apply_objects(reason))
return err; return sink.get_error();
auto hdr = writer.next_header_buffer(); auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::down_message, to_bytes(header{message_type::down_message,
static_cast<uint32_t>(payload.size()), static_cast<uint32_t>(payload.size()),
...@@ -372,17 +373,20 @@ error application::handle_down_message(packet_writer&, header received_hdr, ...@@ -372,17 +373,20 @@ error application::handle_down_message(packet_writer&, header received_hdr,
<< CAF_ARG2("received.size", received.size())); << CAF_ARG2("received.size", received.size()));
error reason; error reason;
binary_deserializer source{&executor_, received}; binary_deserializer source{&executor_, received};
if (auto err = source(reason)) if (!source.apply_objects(reason))
return err; return source.get_error();
proxies_.erase(peer_id_, received_hdr.operation_data, std::move(reason)); proxies_.erase(peer_id_, received_hdr.operation_data, std::move(reason));
return none; return none;
} }
error application::generate_handshake(byte_buffer& buf) { error application::generate_handshake(byte_buffer& buf) {
binary_serializer sink{&executor_, buf}; binary_serializer sink{&executor_, buf};
return sink(system().node(), if (!sink.apply_objects(system().node(),
get_or(system().config(), "caf.middleman.app-identifiers", get_or(system().config(),
application::default_app_ids())); "caf.middleman.app-identifiers",
application::default_app_ids())))
return sink.get_error();
return none;
} }
} // namespace caf::net::basp } // namespace caf::net::basp
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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/rfc6455.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/span.hpp"
#include <cstring>
namespace caf::detail {
void rfc6455::mask_data(uint32_t key, span<char> data) {
mask_data(key, as_writable_bytes(data));
}
void rfc6455::mask_data(uint32_t key, span<byte> data) {
auto no_key = to_network_order(key);
byte arr[4];
memcpy(arr, &no_key, 4);
size_t i = 0;
for (auto& x : data) {
x = x ^ arr[i];
i = (i + 1) % 4;
}
}
void rfc6455::assemble_frame(uint32_t mask_key, span<const char> data,
binary_buffer& out) {
assemble_frame(text_frame, mask_key, as_bytes(data), out);
}
void rfc6455::assemble_frame(uint32_t mask_key, span<const byte> data,
binary_buffer& out) {
assemble_frame(binary_frame, mask_key, data, out);
}
void rfc6455::assemble_frame(uint8_t opcode, uint32_t mask_key,
span<const byte> data, binary_buffer& out) {
// First 8 bits: FIN flag + opcode (we never fragment frames).
out.push_back(byte{static_cast<uint8_t>(0x80 | opcode)});
// Mask flag + payload length (7 bits, 7+16 bits, or 7+64 bits)
auto mask_bit = byte{static_cast<uint8_t>(mask_key == 0 ? 0x00 : 0x80)};
if (data.size() < 126) {
auto len = static_cast<uint8_t>(data.size());
out.push_back(mask_bit | byte{len});
} else if (data.size() < std::numeric_limits<uint16_t>::max()) {
auto len = static_cast<uint16_t>(data.size());
auto no_len = to_network_order(len);
byte len_data[2];
memcpy(len_data, &no_len, 2);
out.push_back(mask_bit | byte{126});
for (auto x : len_data)
out.push_back(x);
} else {
auto len = static_cast<uint64_t>(data.size());
auto no_len = to_network_order(len);
byte len_data[8];
memcpy(len_data, &no_len, 8);
out.push_back(mask_bit | byte{127});
out.insert(out.end(), len_data, len_data + 8);
}
// Masking key: 0 or 4 bytes.
if (mask_key != 0) {
auto no_key = to_network_order(mask_key);
byte key_data[4];
memcpy(key_data, &no_key, 4);
out.insert(out.end(), key_data, key_data + 4);
}
// Application data.
out.insert(out.end(), data.begin(), data.end());
}
ptrdiff_t rfc6455::decode_header(span<const byte> data, header& hdr) {
if (data.size() < 2)
return 0;
auto byte1 = to_integer<uint8_t>(data[0]);
auto byte2 = to_integer<uint8_t>(data[1]);
// Fetch FIN flag and opcode.
hdr.fin = (byte1 & 0x80) != 0;
hdr.opcode = byte1 & 0x0F;
// Decode mask bit and payload length field.
bool masked = (byte2 & 0x80) != 0;
auto len_field = byte2 & 0x7F;
size_t header_length;
if (len_field < 126) {
header_length = 2 + (masked ? 4 : 0);
hdr.payload_len = len_field;
} else if (len_field == 126) {
header_length = 4 + (masked ? 4 : 0);
} else {
header_length = 10 + (masked ? 4 : 0);
}
// Make sure we can read all the data we need.
if (data.size() < header_length)
return 0;
// Start decoding remaining header bytes.
const byte* p = data.data() + 2;
// Fetch payload size
if (len_field == 126) {
uint16_t no_len;
memcpy(&no_len, p, 2);
hdr.payload_len = from_network_order(no_len);
p += 2;
} else if (len_field == 127) {
uint64_t no_len;
memcpy(&no_len, p, 8);
hdr.payload_len = from_network_order(no_len);
p += 8;
}
// Fetch mask key.
if (masked) {
uint32_t no_key;
memcpy(&no_key, p, 4);
hdr.mask_key = from_network_order(no_key);
p += 4;
} else {
hdr.mask_key = 0;
}
// No extension bits allowed.
if (byte1 & 0x70)
return -1;
// Verify opcode and return number of consumed bytes.
switch (hdr.opcode) {
case continuation_frame:
case text_frame:
case binary_frame:
case connection_close:
case ping:
case pong:
return static_cast<ptrdiff_t>(header_length);
default:
return -1;
}
}
} // namespace caf::detail
...@@ -25,6 +25,8 @@ ...@@ -25,6 +25,8 @@
namespace caf::net { namespace caf::net {
// -- constructors, destructors, and assignment operators ----------------------
endpoint_manager::endpoint_manager(socket handle, const multiplexer_ptr& parent, endpoint_manager::endpoint_manager(socket handle, const multiplexer_ptr& parent,
actor_system& sys) actor_system& sys)
: super(handle, parent), sys_(sys), queue_(unit, unit, unit) { : super(handle, parent), sys_(sys), queue_(unit, unit, unit) {
...@@ -35,6 +37,18 @@ endpoint_manager::~endpoint_manager() { ...@@ -35,6 +37,18 @@ endpoint_manager::~endpoint_manager() {
// nop // nop
} }
// -- properties ---------------------------------------------------------------
const actor_system_config& endpoint_manager::config() const noexcept {
return sys_.config();
}
// -- queue access -------------------------------------------------------------
bool endpoint_manager::at_end_of_message_queue() {
return queue_.empty() && queue_.try_block();
}
endpoint_manager_queue::message_ptr endpoint_manager::next_message() { endpoint_manager_queue::message_ptr endpoint_manager::next_message() {
if (queue_.blocked()) if (queue_.blocked())
return nullptr; return nullptr;
...@@ -50,6 +64,8 @@ endpoint_manager_queue::message_ptr endpoint_manager::next_message() { ...@@ -50,6 +64,8 @@ endpoint_manager_queue::message_ptr endpoint_manager::next_message() {
return result; return result;
} }
// -- event management ---------------------------------------------------------
void endpoint_manager::resolve(uri locator, actor listener) { void endpoint_manager::resolve(uri locator, actor listener) {
using intrusive::inbox_result; using intrusive::inbox_result;
using event_type = endpoint_manager_queue::event; using event_type = endpoint_manager_queue::event;
......
...@@ -26,6 +26,7 @@ ...@@ -26,6 +26,7 @@
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/operation.hpp" #include "caf/net/operation.hpp"
#include "caf/net/pollset_updater.hpp" #include "caf/net/pollset_updater.hpp"
#include "caf/net/socket_manager.hpp" #include "caf/net/socket_manager.hpp"
...@@ -78,7 +79,9 @@ short to_bitmask(operation op) { ...@@ -78,7 +79,9 @@ short to_bitmask(operation op) {
} // namespace } // namespace
multiplexer::multiplexer() : shutting_down_(false) { // -- constructors, destructors, and assignment operators ----------------------
multiplexer::multiplexer(middleman* owner) : owner_(owner) {
// nop // nop
} }
...@@ -86,15 +89,23 @@ multiplexer::~multiplexer() { ...@@ -86,15 +89,23 @@ multiplexer::~multiplexer() {
// nop // nop
} }
// -- initialization -----------------------------------------------------------
error multiplexer::init() { error multiplexer::init() {
auto pipe_handles = make_pipe(); auto pipe_handles = make_pipe();
if (!pipe_handles) if (!pipe_handles)
return std::move(pipe_handles.error()); return std::move(pipe_handles.error());
add(make_counted<pollset_updater>(pipe_handles->first, shared_from_this())); auto updater = make_counted<pollset_updater>(pipe_handles->first, this);
settings dummy;
if (auto err = updater->init(dummy))
return err;
add(std::move(updater));
write_handle_ = pipe_handles->second; write_handle_ = pipe_handles->second;
return none; return none;
} }
// -- properties ---------------------------------------------------------------
size_t multiplexer::num_socket_managers() const noexcept { size_t multiplexer::num_socket_managers() const noexcept {
return managers_.size(); return managers_.size();
} }
...@@ -106,7 +117,19 @@ ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) { ...@@ -106,7 +117,19 @@ ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) {
return i == last ? -1 : std::distance(first, i); return i == last ? -1 : std::distance(first, i);
} }
middleman& multiplexer::owner() {
CAF_ASSERT(owner_ != nullptr);
return *owner_;
}
actor_system& multiplexer::system() {
return owner().system();
}
// -- thread-safe signaling ----------------------------------------------------
void multiplexer::register_reading(const socket_manager_ptr& mgr) { void multiplexer::register_reading(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) { if (std::this_thread::get_id() == tid_) {
if (shutting_down_) { if (shutting_down_) {
// discard // discard
...@@ -120,11 +143,12 @@ void multiplexer::register_reading(const socket_manager_ptr& mgr) { ...@@ -120,11 +143,12 @@ void multiplexer::register_reading(const socket_manager_ptr& mgr) {
add(mgr); add(mgr);
} }
} else { } else {
write_to_pipe(0, mgr); write_to_pipe(pollset_updater::register_reading_code, mgr);
} }
} }
void multiplexer::register_writing(const socket_manager_ptr& mgr) { void multiplexer::register_writing(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) { if (std::this_thread::get_id() == tid_) {
if (shutting_down_) { if (shutting_down_) {
// discard // discard
...@@ -138,11 +162,30 @@ void multiplexer::register_writing(const socket_manager_ptr& mgr) { ...@@ -138,11 +162,30 @@ void multiplexer::register_writing(const socket_manager_ptr& mgr) {
add(mgr); add(mgr);
} }
} else { } else {
write_to_pipe(1, mgr); write_to_pipe(pollset_updater::register_writing_code, mgr);
}
}
void multiplexer::init(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) {
if (shutting_down_) {
// discard
} else {
if (auto err = mgr->init(content(system().config()))) {
CAF_LOG_ERROR("mgr->init failed: " << err);
// The socket manager should not register itself for any events if
// initialization fails. So there's probably nothing we could do
// here other than discarding the manager.
}
}
} else {
write_to_pipe(pollset_updater::init_manager_code, mgr);
} }
} }
void multiplexer::close_pipe() { void multiplexer::close_pipe() {
CAF_LOG_TRACE("");
std::lock_guard<std::mutex> guard{write_lock_}; std::lock_guard<std::mutex> guard{write_lock_};
if (write_handle_ != invalid_socket) { if (write_handle_ != invalid_socket) {
close(write_handle_); close(write_handle_);
...@@ -150,7 +193,10 @@ void multiplexer::close_pipe() { ...@@ -150,7 +193,10 @@ void multiplexer::close_pipe() {
} }
} }
// -- control flow -------------------------------------------------------------
bool multiplexer::poll_once(bool blocking) { bool multiplexer::poll_once(bool blocking) {
CAF_LOG_TRACE(CAF_ARG(blocking));
if (pollset_.empty()) if (pollset_.empty())
return false; return false;
// We'll call poll() until poll() succeeds or fails. // We'll call poll() until poll() succeeds or fails.
...@@ -216,6 +262,7 @@ bool multiplexer::poll_once(bool blocking) { ...@@ -216,6 +262,7 @@ bool multiplexer::poll_once(bool blocking) {
} }
void multiplexer::set_thread_id() { void multiplexer::set_thread_id() {
CAF_LOG_TRACE("");
tid_ = std::this_thread::get_id(); tid_ = std::this_thread::get_id();
} }
...@@ -226,7 +273,9 @@ void multiplexer::run() { ...@@ -226,7 +273,9 @@ void multiplexer::run() {
} }
void multiplexer::shutdown() { void multiplexer::shutdown() {
CAF_LOG_TRACE("");
if (std::this_thread::get_id() == tid_) { if (std::this_thread::get_id() == tid_) {
CAF_LOG_DEBUG("initiate shutdown");
shutting_down_ = true; shutting_down_ = true;
// First manager is the pollset_updater. Skip it and delete later. // First manager is the pollset_updater. Skip it and delete later.
for (size_t i = 1; i < managers_.size();) { for (size_t i = 1; i < managers_.size();) {
...@@ -242,10 +291,13 @@ void multiplexer::shutdown() { ...@@ -242,10 +291,13 @@ void multiplexer::shutdown() {
} }
close_pipe(); close_pipe();
} else { } else {
CAF_LOG_DEBUG("push shutdown event to pipe");
write_to_pipe(4, nullptr); write_to_pipe(4, nullptr);
} }
} }
// -- utility functions --------------------------------------------------------
short multiplexer::handle(const socket_manager_ptr& mgr, short events, short multiplexer::handle(const socket_manager_ptr& mgr, short events,
short revents) { short revents) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle())); CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle()));
...@@ -279,6 +331,7 @@ short multiplexer::handle(const socket_manager_ptr& mgr, short events, ...@@ -279,6 +331,7 @@ short multiplexer::handle(const socket_manager_ptr& mgr, short events,
} }
void multiplexer::add(socket_manager_ptr mgr) { void multiplexer::add(socket_manager_ptr mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle()));
CAF_ASSERT(index_of(mgr) == -1); CAF_ASSERT(index_of(mgr) == -1);
pollfd new_entry{socket_cast<socket_id>(mgr->handle()), pollfd new_entry{socket_cast<socket_id>(mgr->handle()),
to_bitmask(mgr->mask()), 0}; to_bitmask(mgr->mask()), 0};
...@@ -293,10 +346,13 @@ void multiplexer::del(ptrdiff_t index) { ...@@ -293,10 +346,13 @@ void multiplexer::del(ptrdiff_t index) {
} }
void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) { void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) {
CAF_ASSERT(opcode == 0 || opcode == 1 || opcode == 4); CAF_ASSERT(opcode == pollset_updater::register_reading_code
CAF_ASSERT(mgr != nullptr || opcode == 4); || opcode == pollset_updater::register_writing_code
|| opcode == pollset_updater::init_manager_code
|| opcode == pollset_updater::shutdown_code);
CAF_ASSERT(mgr != nullptr || opcode == pollset_updater::shutdown_code);
pollset_updater::msg_buf buf; pollset_updater::msg_buf buf;
if (opcode != 4) if (opcode != pollset_updater::shutdown_code)
mgr->ref(); mgr->ref();
buf[0] = static_cast<byte>(opcode); buf[0] = static_cast<byte>(opcode);
auto value = reinterpret_cast<intptr_t>(mgr.get()); auto value = reinterpret_cast<intptr_t>(mgr.get());
...@@ -307,7 +363,7 @@ void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) { ...@@ -307,7 +363,7 @@ void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) {
if (write_handle_ != invalid_socket) if (write_handle_ != invalid_socket)
res = write(write_handle_, buf); res = write(write_handle_, buf);
} }
if (res <= 0 && opcode != 4) if (res <= 0 && opcode != pollset_updater::shutdown_code)
mgr->deref(); mgr->deref();
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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/actor_shell.hpp"
#include "caf/callback.hpp"
#include "caf/config.hpp"
#include "caf/detail/default_invoke_result_visitor.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/invoke_message_result.hpp"
#include "caf/logger.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.hpp"
namespace caf::net {
// -- constructors, destructors, and assignment operators ----------------------
actor_shell::actor_shell(actor_config& cfg, socket_manager* owner)
: super(cfg), mailbox_(policy::normal_messages{}), owner_(owner) {
mailbox_.try_block();
}
actor_shell::~actor_shell() {
// nop
}
// -- state modifiers ----------------------------------------------------------
void actor_shell::quit(error reason) {
cleanup(std::move(reason), nullptr);
}
// -- mailbox access -----------------------------------------------------------
mailbox_element_ptr actor_shell::next_message() {
if (!mailbox_.blocked()) {
mailbox_.fetch_more();
auto& q = mailbox_.queue();
if (q.total_task_size() > 0) {
q.inc_deficit(1);
return q.next();
}
}
return nullptr;
}
bool actor_shell::try_block_mailbox() {
return mailbox_.try_block();
}
// -- message processing -------------------------------------------------------
bool actor_shell::consume_message() {
CAF_LOG_TRACE("");
if (auto msg = next_message()) {
current_element_ = msg.get();
CAF_LOG_RECEIVE_EVENT(current_element_);
CAF_BEFORE_PROCESSING(this, *msg);
auto mid = msg->mid;
if (!mid.is_response()) {
detail::default_invoke_result_visitor<actor_shell> visitor{this};
if (auto result = bhvr_(msg->payload)) {
visitor(*result);
} else {
auto fallback_result = (*fallback_)(msg->payload);
visit(visitor, fallback_result);
}
} else if (auto i = multiplexed_responses_.find(mid);
i != multiplexed_responses_.end()) {
auto bhvr = std::move(i->second);
multiplexed_responses_.erase(i);
auto res = bhvr(msg->payload);
if (!res) {
CAF_LOG_DEBUG("got unexpected_response");
auto err_msg = make_message(
make_error(sec::unexpected_response, std::move(msg->payload)));
bhvr(err_msg);
}
}
CAF_AFTER_PROCESSING(this, invoke_message_result::consumed);
CAF_LOG_SKIP_OR_FINALIZE_EVENT(invoke_message_result::consumed);
return true;
}
return false;
}
void actor_shell::add_multiplexed_response_handler(message_id response_id,
behavior bhvr) {
if (bhvr.timeout() != infinite)
request_response_timeout(bhvr.timeout(), response_id);
multiplexed_responses_.emplace(response_id, std::move(bhvr));
}
// -- overridden functions of abstract_actor -----------------------------------
void actor_shell::enqueue(mailbox_element_ptr ptr, execution_unit*) {
CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(!getf(is_blocking_flag));
CAF_LOG_TRACE(CAF_ARG(*ptr));
CAF_LOG_SEND_EVENT(ptr);
auto mid = ptr->mid;
auto sender = ptr->sender;
auto collects_metrics = getf(abstract_actor::collects_metrics_flag);
if (collects_metrics) {
ptr->set_enqueue_time();
metrics_.mailbox_size->inc();
}
switch (mailbox().push_back(std::move(ptr))) {
case intrusive::inbox_result::unblocked_reader: {
CAF_LOG_ACCEPT_EVENT(true);
std::unique_lock<std::mutex> guard{owner_mtx_};
// The owner can only be null if this enqueue succeeds, then we close the
// mailbox and reset owner_ in cleanup() before acquiring the mutex here.
// Hence, the mailbox element has already been disposed and we can simply
// skip any further processing.
if (owner_)
owner_->mpx().register_writing(owner_);
break;
}
case intrusive::inbox_result::queue_closed: {
CAF_LOG_REJECT_EVENT();
home_system().base_metrics().rejected_messages->inc();
if (collects_metrics)
metrics_.mailbox_size->dec();
if (mid.is_request()) {
detail::sync_request_bouncer f{exit_reason()};
f(sender, mid);
}
break;
}
case intrusive::inbox_result::success:
// Enqueued to a running actors' mailbox: nothing to do.
CAF_LOG_ACCEPT_EVENT(false);
break;
}
}
mailbox_element* actor_shell::peek_at_next_mailbox_element() {
return mailbox().closed() || mailbox().blocked() ? nullptr : mailbox().peek();
}
const char* actor_shell::name() const {
return "caf.net.actor-shell";
}
void actor_shell::launch(execution_unit*, bool, bool hide) {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE(CAF_ARG(hide));
CAF_ASSERT(!getf(is_blocking_flag));
if (!hide)
register_at_system();
}
bool actor_shell::cleanup(error&& fail_state, execution_unit* host) {
CAF_LOG_TRACE(CAF_ARG(fail_state));
// Clear mailbox.
if (!mailbox_.closed()) {
mailbox_.close();
detail::sync_request_bouncer bounce{fail_state};
auto dropped = mailbox_.queue().new_round(1000, bounce).consumed_items;
while (dropped > 0) {
if (getf(abstract_actor::collects_metrics_flag)) {
auto val = static_cast<int64_t>(dropped);
metrics_.mailbox_size->dec(val);
}
dropped = mailbox_.queue().new_round(1000, bounce).consumed_items;
}
}
// Detach from owner.
{
std::unique_lock<std::mutex> guard{owner_mtx_};
owner_ = nullptr;
}
// Dispatch to parent's `cleanup` function.
return super::cleanup(std::move(fail_state), host);
}
actor_shell_ptr::actor_shell_ptr(strong_actor_ptr ptr) noexcept
: ptr_(std::move(ptr)) {
// nop
}
actor actor_shell_ptr::as_actor() const noexcept {
return actor_cast<actor>(ptr_);
}
void actor_shell_ptr::detach(error reason) {
if (auto ptr = get()) {
ptr->quit(std::move(reason));
ptr_.release();
}
}
actor_shell_ptr::~actor_shell_ptr() {
if (auto ptr = get())
ptr->quit(exit_reason::normal);
}
actor_shell* actor_shell_ptr::get() const noexcept {
if (ptr_) {
auto ptr = actor_cast<abstract_actor*>(ptr_);
return static_cast<actor_shell*>(ptr);
}
return nullptr;
}
} // namespace caf::net
...@@ -25,6 +25,7 @@ ...@@ -25,6 +25,7 @@
#include "caf/net/basp/application.hpp" #include "caf/net/basp/application.hpp"
#include "caf/net/basp/application_factory.hpp" #include "caf/net/basp/application_factory.hpp"
#include "caf/net/basp/ec.hpp" #include "caf/net/basp/ec.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/doorman.hpp" #include "caf/net/doorman.hpp"
#include "caf/net/ip.hpp" #include "caf/net/ip.hpp"
#include "caf/net/make_endpoint_manager.hpp" #include "caf/net/make_endpoint_manager.hpp"
......
...@@ -25,7 +25,6 @@ ...@@ -25,7 +25,6 @@
#include "caf/net/basp/ec.hpp" #include "caf/net/basp/ec.hpp"
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
#include "caf/net/middleman_backend.hpp" #include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/raise_error.hpp" #include "caf/raise_error.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
...@@ -37,8 +36,8 @@ void middleman::init_global_meta_objects() { ...@@ -37,8 +36,8 @@ void middleman::init_global_meta_objects() {
caf::init_global_meta_objects<id_block::net_module>(); caf::init_global_meta_objects<id_block::net_module>();
} }
middleman::middleman(actor_system& sys) : sys_(sys) { middleman::middleman(actor_system& sys) : sys_(sys), mpx_(this) {
mpx_ = std::make_shared<multiplexer>(); // nop
} }
middleman::~middleman() { middleman::~middleman() {
...@@ -47,15 +46,13 @@ middleman::~middleman() { ...@@ -47,15 +46,13 @@ middleman::~middleman() {
void middleman::start() { void middleman::start() {
if (!get_or(config(), "caf.middleman.manual-multiplexing", false)) { if (!get_or(config(), "caf.middleman.manual-multiplexing", false)) {
auto mpx = mpx_; mpx_thread_ = std::thread{[this] {
auto sys_ptr = &system(); CAF_SET_LOGGER_SYS(&sys_);
mpx_thread_ = std::thread{[mpx, sys_ptr] { detail::set_thread_name("caf.net.mpx");
CAF_SET_LOGGER_SYS(sys_ptr); sys_.thread_started();
detail::set_thread_name("caf.multiplexer"); mpx_.set_thread_id();
sys_ptr->thread_started(); mpx_.run();
mpx->set_thread_id(); sys_.thread_terminates();
mpx->run();
sys_ptr->thread_terminates();
}}; }};
} }
} }
...@@ -63,23 +60,23 @@ void middleman::start() { ...@@ -63,23 +60,23 @@ void middleman::start() {
void middleman::stop() { void middleman::stop() {
for (const auto& backend : backends_) for (const auto& backend : backends_)
backend->stop(); backend->stop();
mpx_->shutdown(); mpx_.shutdown();
if (mpx_thread_.joinable()) if (mpx_thread_.joinable())
mpx_thread_.join(); mpx_thread_.join();
else else
mpx_->run(); mpx_.run();
} }
void middleman::init(actor_system_config& cfg) { void middleman::init(actor_system_config& cfg) {
if (auto err = mpx_->init()) { if (auto err = mpx_.init()) {
CAF_LOG_ERROR("mgr->init() failed: " << err); CAF_LOG_ERROR("mpx_.init() failed: " << err);
CAF_RAISE_ERROR("mpx->init() failed"); CAF_RAISE_ERROR("mpx_.init() failed");
} }
if (auto node_uri = get_if<uri>(&cfg, "caf.middleman.this-node")) { if (auto node_uri = get_if<uri>(&cfg, "caf.middleman.this-node")) {
auto this_node = make_node_id(std::move(*node_uri)); auto this_node = make_node_id(std::move(*node_uri));
sys_.node_.swap(this_node); sys_.node_.swap(this_node);
} else { } else {
CAF_RAISE_ERROR("no valid entry for caf.middleman.this-node found"); // CAF_RAISE_ERROR("no valid entry for caf.middleman.this-node found");
} }
for (auto& backend : backends_) for (auto& backend : backends_)
if (auto err = backend->init()) { if (auto err = backend->init()) {
......
...@@ -26,6 +26,7 @@ ...@@ -26,6 +26,7 @@
#include "caf/detail/socket_sys_aliases.hpp" #include "caf/detail/socket_sys_aliases.hpp"
#include "caf/detail/socket_sys_includes.hpp" #include "caf/detail/socket_sys_includes.hpp"
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/logger.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
...@@ -81,11 +82,13 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() { ...@@ -81,11 +82,13 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() {
} }
ptrdiff_t write(pipe_socket x, span<const byte> buf) { ptrdiff_t write(pipe_socket x, span<const byte> buf) {
CAF_LOG_TRACE(CAF_ARG2("socket", x.id) << CAF_ARG2("bytes", buf.size()));
return ::write(x.id, reinterpret_cast<socket_send_ptr>(buf.data()), return ::write(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size()); buf.size());
} }
ptrdiff_t read(pipe_socket x, span<byte> buf) { ptrdiff_t read(pipe_socket x, span<byte> buf) {
CAF_LOG_TRACE(CAF_ARG2("socket", x.id) << CAF_ARG2("bytes", buf.size()));
return ::read(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()), return ::read(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()),
buf.size()); buf.size());
} }
......
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#include <cstring> #include <cstring>
#include "caf/actor_system.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
...@@ -28,20 +29,24 @@ ...@@ -28,20 +29,24 @@
namespace caf::net { namespace caf::net {
pollset_updater::pollset_updater(pipe_socket read_handle, pollset_updater::pollset_updater(pipe_socket read_handle, multiplexer* parent)
const multiplexer_ptr& parent) : super(read_handle, parent) {
: super(read_handle, parent), buf_size_(0) {
mask_ = operation::read; mask_ = operation::read;
if (auto err = nonblocking(read_handle, true))
CAF_LOG_ERROR("nonblocking failed: " << err);
} }
pollset_updater::~pollset_updater() { pollset_updater::~pollset_updater() {
// nop // nop
} }
error pollset_updater::init(const settings&) {
CAF_LOG_TRACE("");
return nonblocking(handle(), true);
}
bool pollset_updater::handle_read_event() { bool pollset_updater::handle_read_event() {
CAF_LOG_TRACE("");
for (;;) { for (;;) {
CAF_ASSERT((buf_.size() - buf_size_) > 0);
auto num_bytes = read(handle(), make_span(buf_.data() + buf_size_, auto num_bytes = read(handle(), make_span(buf_.data() + buf_size_,
buf_.size() - buf_size_)); buf_.size() - buf_size_));
if (num_bytes > 0) { if (num_bytes > 0) {
...@@ -52,25 +57,29 @@ bool pollset_updater::handle_read_event() { ...@@ -52,25 +57,29 @@ bool pollset_updater::handle_read_event() {
intptr_t value; intptr_t value;
memcpy(&value, buf_.data() + 1, sizeof(intptr_t)); memcpy(&value, buf_.data() + 1, sizeof(intptr_t));
socket_manager_ptr mgr{reinterpret_cast<socket_manager*>(value), false}; socket_manager_ptr mgr{reinterpret_cast<socket_manager*>(value), false};
if (auto ptr = parent_.lock()) { switch (opcode) {
switch (opcode) { case register_reading_code:
case 0: parent_->register_reading(mgr);
ptr->register_reading(mgr); break;
break; case register_writing_code:
case 1: parent_->register_writing(mgr);
ptr->register_writing(mgr); break;
break; case init_manager_code:
case 4: parent_->init(mgr);
ptr->shutdown(); break;
break; case shutdown_code:
default: parent_->shutdown();
CAF_LOG_DEBUG("opcode not recognized: " << CAF_ARG(opcode)); break;
break; default:
} CAF_LOG_ERROR("opcode not recognized: " << CAF_ARG(opcode));
break;
} }
} }
} else if (num_bytes == 0) {
CAF_LOG_DEBUG("pipe closed, assume shutdown");
return false;
} else { } else {
return num_bytes < 0 && last_socket_error_is_temporary(); return last_socket_error_is_temporary();
} }
} }
} }
......
...@@ -19,14 +19,15 @@ ...@@ -19,14 +19,15 @@
#include "caf/net/socket_manager.hpp" #include "caf/net/socket_manager.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/net/actor_shell.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
namespace caf::net { namespace caf::net {
socket_manager::socket_manager(socket handle, const multiplexer_ptr& parent) socket_manager::socket_manager(socket handle, multiplexer* parent)
: handle_(handle), mask_(operation::none), parent_(parent) { : handle_(handle), mask_(operation::none), parent_(parent) {
CAF_ASSERT(parent != nullptr);
CAF_ASSERT(handle_ != invalid_socket); CAF_ASSERT(handle_ != invalid_socket);
CAF_ASSERT(parent != nullptr);
} }
socket_manager::~socket_manager() { socket_manager::~socket_manager() {
...@@ -54,17 +55,19 @@ bool socket_manager::mask_del(operation flag) noexcept { ...@@ -54,17 +55,19 @@ bool socket_manager::mask_del(operation flag) noexcept {
void socket_manager::register_reading() { void socket_manager::register_reading() {
if ((mask() & operation::read) == operation::read) if ((mask() & operation::read) == operation::read)
return; return;
auto ptr = parent_.lock(); parent_->register_reading(this);
if (ptr != nullptr)
ptr->register_reading(this);
} }
void socket_manager::register_writing() { void socket_manager::register_writing() {
if ((mask() & operation::write) == operation::write) if ((mask() & operation::write) == operation::write)
return; return;
auto ptr = parent_.lock(); parent_->register_writing(this);
if (ptr != nullptr) }
ptr->register_writing(this);
actor_shell_ptr socket_manager::make_actor_shell_impl() {
CAF_ASSERT(parent_ != nullptr);
auto hdl = parent_->system().spawn<actor_shell>(this);
return actor_shell_ptr{actor_cast<strong_actor_ptr>(std::move(hdl))};
} }
} // namespace caf::net } // namespace caf::net
...@@ -169,11 +169,13 @@ error nodelay(stream_socket x, bool new_value) { ...@@ -169,11 +169,13 @@ error nodelay(stream_socket x, bool new_value) {
} }
ptrdiff_t read(stream_socket x, span<byte> buf) { ptrdiff_t read(stream_socket x, span<byte> buf) {
CAF_LOG_TRACE(CAF_ARG2("socket", x.id) << CAF_ARG2("bytes", buf.size()));
return ::recv(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()), buf.size(), return ::recv(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()), buf.size(),
no_sigpipe_io_flag); no_sigpipe_io_flag);
} }
ptrdiff_t write(stream_socket x, span<const byte> buf) { ptrdiff_t write(stream_socket x, span<const byte> buf) {
CAF_LOG_TRACE(CAF_ARG2("socket", x.id) << CAF_ARG2("bytes", buf.size()));
return ::send(x.id, reinterpret_cast<socket_send_ptr>(buf.data()), buf.size(), return ::send(x.id, reinterpret_cast<socket_send_ptr>(buf.data()), buf.size(),
no_sigpipe_io_flag); no_sigpipe_io_flag);
} }
......
...@@ -98,21 +98,23 @@ expected<tcp_accept_socket> make_tcp_accept_socket(ip_endpoint node, ...@@ -98,21 +98,23 @@ expected<tcp_accept_socket> make_tcp_accept_socket(ip_endpoint node,
bool reuse_addr) { bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(node)); CAF_LOG_TRACE(CAF_ARG(node));
auto addr = to_string(node.address()); auto addr = to_string(node.address());
auto make_acceptor = node.address().embeds_v4() bool is_v4 = node.address().embeds_v4();
? new_tcp_acceptor_impl<AF_INET> bool is_zero = is_v4 ? node.address().embedded_v4().bits() == 0
: new_tcp_acceptor_impl<AF_INET6>; : node.address().zero();
auto p = make_acceptor(node.port(), addr.c_str(), reuse_addr, auto make_acceptor = is_v4 ? new_tcp_acceptor_impl<AF_INET>
node.address().zero()); : new_tcp_acceptor_impl<AF_INET6>;
if (!p) { if (auto p = make_acceptor(node.port(), addr.c_str(), reuse_addr, is_zero)) {
CAF_LOG_WARNING("could not create tcp socket for: " << to_string(node)); auto sock = socket_cast<tcp_accept_socket>(*p);
auto sguard = make_socket_guard(sock);
CAF_NET_SYSCALL("listen", tmp, !=, 0, listen(sock.id, SOMAXCONN));
CAF_LOG_DEBUG(CAF_ARG(sock.id));
return sguard.release();
} else {
CAF_LOG_WARNING("could not create tcp socket:" << CAF_ARG(node)
<< CAF_ARG(p.error()));
return make_error(sec::cannot_open_port, "tcp socket creation failed", return make_error(sec::cannot_open_port, "tcp socket creation failed",
to_string(node)); to_string(node), std::move(p.error()));
} }
auto sock = socket_cast<tcp_accept_socket>(*p);
auto sguard = make_socket_guard(sock);
CAF_NET_SYSCALL("listen", tmp, !=, 0, listen(sock.id, SOMAXCONN));
CAF_LOG_DEBUG(CAF_ARG(sock.id));
return sguard.release();
} }
expected<tcp_accept_socket> expected<tcp_accept_socket>
......
...@@ -114,8 +114,8 @@ variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf, ...@@ -114,8 +114,8 @@ variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf,
ip_endpoint ep) { ip_endpoint ep) {
sockaddr_storage addr = {}; sockaddr_storage addr = {};
detail::convert(ep, addr); detail::convert(ep, addr);
auto len = ep.address().embeds_v4() ? sizeof(sockaddr_in) auto len = static_cast<socklen_t>(
: sizeof(sockaddr_in6); ep.address().embeds_v4() ? sizeof(sockaddr_in) : sizeof(sockaddr_in6));
auto res = ::sendto(x.id, reinterpret_cast<socket_send_ptr>(buf.data()), auto res = ::sendto(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size(), 0, reinterpret_cast<sockaddr*>(&addr), len); buf.size(), 0, reinterpret_cast<sockaddr*>(&addr), len);
auto ret = check_udp_datagram_socket_io_res(res); auto ret = check_udp_datagram_socket_io_res(res);
...@@ -172,7 +172,7 @@ variant<size_t, sec> write(udp_datagram_socket x, span<byte_buffer*> bufs, ...@@ -172,7 +172,7 @@ variant<size_t, sec> write(udp_datagram_socket x, span<byte_buffer*> bufs,
message.msg_namelen = ep.address().embeds_v4() ? sizeof(sockaddr_in) message.msg_namelen = ep.address().embeds_v4() ? sizeof(sockaddr_in)
: sizeof(sockaddr_in6); : sizeof(sockaddr_in6);
message.msg_iov = buf_array; message.msg_iov = buf_array;
message.msg_iovlen = bufs.size(); message.msg_iovlen = static_cast<int>(bufs.size());
auto res = sendmsg(x.id, &message, 0); auto res = sendmsg(x.id, &message, 0);
return check_udp_datagram_socket_io_res(res); return check_udp_datagram_socket_io_res(res);
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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. *
******************************************************************************/
#define CAF_SUITE detail.rfc6455
#include "caf/detail/rfc6455.hpp"
#include "caf/test/dsl.hpp"
#include "caf/byte.hpp"
#include "caf/span.hpp"
#include "caf/string_view.hpp"
#include <cstdint>
#include <initializer_list>
#include <vector>
using namespace caf;
namespace {
struct fixture {
using impl = detail::rfc6455;
auto bytes(std::initializer_list<uint8_t> xs) {
std::vector<byte> result;
for (auto x : xs)
result.emplace_back(static_cast<byte>(x));
return result;
}
template <class T>
auto take(const T& xs, size_t num_bytes) {
auto n = std::min(xs.size(), num_bytes);
return std::vector<typename T::value_type>{xs.begin(), xs.begin() + n};
}
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(rfc6455_tests, fixture)
CAF_TEST(masking) {
auto key = uint32_t{0xDEADC0DE};
auto data = bytes({0x12, 0x34, 0x45, 0x67, 0x89, 0x9A});
auto masked_data = data;
CAF_MESSAGE("masking XORs the repeated key to data");
impl::mask_data(key, masked_data);
CAF_CHECK_EQUAL(masked_data, bytes({
0x12 ^ 0xDE,
0x34 ^ 0xAD,
0x45 ^ 0xC0,
0x67 ^ 0xDE,
0x89 ^ 0xDE,
0x9A ^ 0xAD,
}));
CAF_MESSAGE("masking masked data again gives the original data");
impl::mask_data(key, masked_data);
CAF_CHECK_EQUAL(masked_data, data);
}
CAF_TEST(no mask key plus small data) {
std::vector<uint8_t> data{0x12, 0x34, 0x45, 0x67};
std::vector<byte> out;
impl::assemble_frame(impl::binary_frame, 0, as_bytes(make_span(data)), out);
CAF_CHECK_EQUAL(out, bytes({
0x82, // FIN + binary frame opcode
0x04, // data size = 4
0x12, 0x34, 0x45, 0x67, // masked data
}));
impl::header hdr;
CAF_CHECK_EQUAL(impl::decode_header(out, hdr), 2);
CAF_CHECK_EQUAL(hdr.fin, true);
CAF_CHECK_EQUAL(hdr.mask_key, 0u);
CAF_CHECK_EQUAL(hdr.opcode, impl::binary_frame);
CAF_CHECK_EQUAL(hdr.payload_len, data.size());
}
CAF_TEST(valid mask key plus small data) {
std::vector<uint8_t> data{0x12, 0x34, 0x45, 0x67};
std::vector<byte> out;
impl::assemble_frame(impl::binary_frame, 0xDEADC0DE,
as_bytes(make_span(data)), out);
CAF_CHECK_EQUAL(out, bytes({
0x82, // FIN + binary frame opcode
0x84, // MASKED + data size = 4
0xDE, 0xAD, 0xC0, 0xDE, // mask key,
0x12, 0x34, 0x45, 0x67, // masked data
}));
impl::header hdr;
CAF_CHECK_EQUAL(impl::decode_header(out, hdr), 6);
CAF_CHECK_EQUAL(hdr.fin, true);
CAF_CHECK_EQUAL(hdr.mask_key, 0xDEADC0DEu);
CAF_CHECK_EQUAL(hdr.opcode, impl::binary_frame);
CAF_CHECK_EQUAL(hdr.payload_len, data.size());
}
CAF_TEST(no mask key plus medium data) {
std::vector<uint8_t> data;
data.insert(data.end(), 126, 0xFF);
std::vector<byte> out;
impl::assemble_frame(impl::binary_frame, 0, as_bytes(make_span(data)), out);
CAF_CHECK_EQUAL(take(out, 8),
bytes({
0x82, // FIN + binary frame opcode
0x7E, // 126 -> uint16 size
0x00, 0x7E, // data size = 126
0xFF, 0xFF, 0xFF, 0xFF, // first 4 masked bytes
}));
impl::header hdr;
CAF_CHECK_EQUAL(impl::decode_header(out, hdr), 4);
CAF_CHECK_EQUAL(hdr.fin, true);
CAF_CHECK_EQUAL(hdr.mask_key, 0u);
CAF_CHECK_EQUAL(hdr.opcode, impl::binary_frame);
CAF_CHECK_EQUAL(hdr.payload_len, data.size());
}
CAF_TEST(valid mask key plus medium data) {
std::vector<uint8_t> data;
data.insert(data.end(), 126, 0xFF);
std::vector<byte> out;
impl::assemble_frame(impl::binary_frame, 0xDEADC0DE,
as_bytes(make_span(data)), out);
CAF_CHECK_EQUAL(take(out, 12),
bytes({
0x82, // FIN + binary frame opcode
0xFE, // MASKED + 126 -> uint16 size
0x00, 0x7E, // data size = 126
0xDE, 0xAD, 0xC0, 0xDE, // mask key,
0xFF, 0xFF, 0xFF, 0xFF, // first 4 masked bytes
}));
impl::header hdr;
CAF_CHECK_EQUAL(impl::decode_header(out, hdr), 8);
CAF_CHECK_EQUAL(hdr.fin, true);
CAF_CHECK_EQUAL(hdr.mask_key, 0xDEADC0DEu);
CAF_CHECK_EQUAL(hdr.opcode, impl::binary_frame);
CAF_CHECK_EQUAL(hdr.payload_len, data.size());
}
CAF_TEST(no mask key plus large data) {
std::vector<uint8_t> data;
data.insert(data.end(), 65536, 0xFF);
std::vector<byte> out;
impl::assemble_frame(impl::binary_frame, 0, as_bytes(make_span(data)), out);
CAF_CHECK_EQUAL(take(out, 14),
bytes({
0x82, // FIN + binary frame opcode
0x7F, // 127 -> uint64 size
0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, // 65536
0xFF, 0xFF, 0xFF, 0xFF, // first 4 masked bytes
}));
impl::header hdr;
CAF_CHECK_EQUAL(impl::decode_header(out, hdr), 10);
CAF_CHECK_EQUAL(hdr.fin, true);
CAF_CHECK_EQUAL(hdr.mask_key, 0u);
CAF_CHECK_EQUAL(hdr.opcode, impl::binary_frame);
CAF_CHECK_EQUAL(hdr.payload_len, data.size());
}
CAF_TEST(valid mask key plus large data) {
std::vector<uint8_t> data;
data.insert(data.end(), 65536, 0xFF);
std::vector<byte> out;
impl::assemble_frame(impl::binary_frame, 0xDEADC0DE,
as_bytes(make_span(data)), out);
CAF_CHECK_EQUAL(take(out, 18),
bytes({
0x82, // FIN + binary frame opcode
0xFF, // MASKED + 127 -> uint64 size
0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, // 65536
0xDE, 0xAD, 0xC0, 0xDE, // mask key,
0xFF, 0xFF, 0xFF, 0xFF, // first 4 masked bytes
}));
impl::header hdr;
CAF_CHECK_EQUAL(impl::decode_header(out, hdr), 14);
CAF_CHECK_EQUAL(hdr.fin, true);
CAF_CHECK_EQUAL(hdr.mask_key, 0xDEADC0DEu);
CAF_CHECK_EQUAL(hdr.opcode, impl::binary_frame);
CAF_CHECK_EQUAL(hdr.payload_len, data.size());
}
CAF_TEST_FIXTURE_SCOPE_END()
...@@ -25,6 +25,7 @@ ...@@ -25,6 +25,7 @@
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp" #include "caf/binary_serializer.hpp"
#include "caf/byte_buffer.hpp" #include "caf/byte_buffer.hpp"
#include "caf/deep_to_string.hpp"
using namespace caf; using namespace caf;
using namespace caf::net; using namespace caf::net;
...@@ -34,7 +35,7 @@ CAF_TEST(serialization) { ...@@ -34,7 +35,7 @@ CAF_TEST(serialization) {
byte_buffer buf; byte_buffer buf;
{ {
binary_serializer sink{nullptr, buf}; binary_serializer sink{nullptr, buf};
CAF_CHECK_EQUAL(sink(x), none); CAF_CHECK(sink.apply_object(x));
} }
CAF_CHECK_EQUAL(buf.size(), basp::header_size); CAF_CHECK_EQUAL(buf.size(), basp::header_size);
auto buf2 = to_bytes(x); auto buf2 = to_bytes(x);
...@@ -43,7 +44,7 @@ CAF_TEST(serialization) { ...@@ -43,7 +44,7 @@ CAF_TEST(serialization) {
basp::header y; basp::header y;
{ {
binary_deserializer source{nullptr, buf}; binary_deserializer source{nullptr, buf};
CAF_CHECK_EQUAL(source(y), none); CAF_CHECK(source.apply_object(y));
} }
CAF_CHECK_EQUAL(x, y); CAF_CHECK_EQUAL(x, y);
auto z = basp::header::from_bytes(buf); auto z = basp::header::from_bytes(buf);
...@@ -53,5 +54,5 @@ CAF_TEST(serialization) { ...@@ -53,5 +54,5 @@ CAF_TEST(serialization) {
CAF_TEST(to_string) { CAF_TEST(to_string) {
basp::header x{basp::message_type::handshake, 42, 4}; basp::header x{basp::message_type::handshake, 42, 4};
CAF_CHECK_EQUAL(to_string(x), "basp::header(handshake, 42, 4)"); CAF_CHECK_EQUAL(deep_to_string(x), "basp::header(handshake, 42, 4)");
} }
This diff is collapsed.
#pragma once
#include "caf/error.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/settings.hpp"
#include "caf/span.hpp"
#include "caf/string_view.hpp"
#include "caf/tag/stream_oriented.hpp"
#include "caf/test/dsl.hpp"
template <class UpperLayer>
class mock_stream_transport {
public:
// -- member types -----------------------------------------------------------
using output_tag = caf::tag::stream_oriented;
// -- interface for the upper layer ------------------------------------------
void begin_output() {
// nop
}
auto& output_buffer() {
return output;
}
constexpr void end_output() {
// nop
}
constexpr caf::net::socket handle() noexcept {
return caf::net::invalid_socket;
}
bool can_send_more() const noexcept {
return true;
}
const caf::error& abort_reason() {
return abort_reason_;
}
void abort_reason(caf::error reason) {
abort_reason_ = std::move(reason);
}
void configure_read(caf::net::receive_policy policy) {
min_read_size = policy.min_size;
max_read_size = policy.max_size;
}
// -- initialization ---------------------------------------------------------
caf::error init(const caf::settings& config) {
return upper_layer.init(nullptr, this, config);
}
caf::error init() {
caf::settings config;
return init(config);
}
// -- buffer management ------------------------------------------------------
void push(caf::span<const caf::byte> bytes) {
input.insert(input.begin(), bytes.begin(), bytes.end());
}
void push(caf::string_view str) {
push(caf::as_bytes(caf::make_span(str)));
}
size_t unconsumed() const noexcept {
return read_buf_.size();
}
caf::string_view output_as_str() const noexcept {
return {reinterpret_cast<const char*>(output.data()), output.size()};
}
// -- event callbacks --------------------------------------------------------
ptrdiff_t handle_input() {
ptrdiff_t result = 0;
while (max_read_size > 0) {
CAF_ASSERT(max_read_size > static_cast<size_t>(read_size_));
size_t len = max_read_size - static_cast<size_t>(read_size_);
CAF_LOG_DEBUG(CAF_ARG2("available capacity:", len));
auto num_bytes = std::min(input.size(), len);
if (num_bytes == 0)
return result;
auto delta_offset = static_cast<ptrdiff_t>(read_buf_.size());
read_buf_.insert(read_buf_.end(), input.begin(),
input.begin() + num_bytes);
input.erase(input.begin(), input.begin() + num_bytes);
read_size_ += static_cast<ptrdiff_t>(num_bytes);
if (static_cast<size_t>(read_size_) < min_read_size)
return result;
auto delta = make_span(read_buf_.data() + delta_offset,
read_size_ - delta_offset);
auto consumed = upper_layer.consume(this, caf::make_span(read_buf_),
delta);
if (consumed > 0) {
result += static_cast<ptrdiff_t>(consumed);
read_buf_.erase(read_buf_.begin(), read_buf_.begin() + consumed);
read_size_ -= consumed;
} else if (consumed < 0) {
if (!abort_reason_)
abort_reason_ = caf::sec::runtime_error;
upper_layer.abort(this, abort_reason_);
return -1;
}
}
return result;
}
// -- member variables -------------------------------------------------------
UpperLayer upper_layer;
std::vector<caf::byte> output;
std::vector<caf::byte> input;
uint32_t min_read_size = 0;
uint32_t max_read_size = 0;
private:
std::vector<caf::byte> read_buf_;
ptrdiff_t read_size_ = 0;
caf::error abort_reason_;
};
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