Commit 1f25e71b authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'master' into topic/remove-atom

parents 34f3758b d8081bee
...@@ -187,14 +187,41 @@ function(pretty_yes var) ...@@ -187,14 +187,41 @@ function(pretty_yes var)
endif() endif()
endfunction(pretty_yes) endfunction(pretty_yes)
add_executable(caf-generate-enum-strings cmake/caf-generate-enum-strings.cpp) add_executable(caf-generate-enum-strings
EXCLUDE_FROM_ALL
function(enum_to_string relative_input_file relative_output_file) cmake/caf-generate-enum-strings.cpp)
set(input "${CMAKE_CURRENT_SOURCE_DIR}/${relative_input_file}")
set(output "${CMAKE_BINARY_DIR}/${relative_output_file}") add_custom_target(consistency-check)
add_custom_target(update-enum-strings)
# adds a consistency check that verifies that `cpp_file` is still valid by
# re-generating the file and comparing it to the existing file
function(add_enum_consistency_check hpp_file cpp_file)
set(input "${CMAKE_CURRENT_SOURCE_DIR}/${hpp_file}")
set(file_under_test "${CMAKE_CURRENT_SOURCE_DIR}/${cpp_file}")
set(output "${CMAKE_CURRENT_BINARY_DIR}/check/${cpp_file}")
get_filename_component(output_dir "${output}" DIRECTORY)
file(MAKE_DIRECTORY "${output_dir}")
add_custom_command(OUTPUT "${output}" add_custom_command(OUTPUT "${output}"
COMMAND caf-generate-enum-strings "${input}" "${output}" COMMAND caf-generate-enum-strings "${input}" "${output}"
DEPENDS caf-generate-enum-strings "${input}") DEPENDS caf-generate-enum-strings "${input}")
get_filename_component(target_name "${input}" NAME_WE)
add_custom_target("${target_name}"
COMMAND
"${CMAKE_COMMAND}"
"-Dfile_under_test=${file_under_test}"
"-Dgenerated_file=${output}"
-P "${PROJECT_SOURCE_DIR}/cmake/check-consistency.cmake"
DEPENDS "${output}")
add_dependencies(consistency-check "${target_name}")
add_custom_target("${target_name}-update"
COMMAND
caf-generate-enum-strings
"${input}"
"${file_under_test}"
DEPENDS caf-generate-enum-strings "${input}")
add_dependencies(update-enum-strings "${target_name}-update")
endfunction() endfunction()
################################################################################ ################################################################################
......
...@@ -141,6 +141,24 @@ pipeline { ...@@ -141,6 +141,24 @@ pipeline {
runClangFormat(config) runClangFormat(config)
} }
} }
stage('Check Consistency') {
agent { label 'unix' }
steps {
deleteDir()
unstash('sources')
dir('sources') {
cmakeBuild([
buildDir: 'build',
installation: 'cmake in search path',
sourceDir: '.',
steps: [[
args: '--target consistency-check',
withCmake: true,
]],
])
}
}
}
stage('Build') { stage('Build') {
steps { steps {
buildParallel(config, PrettyJobBaseName) buildParallel(config, PrettyJobBaseName)
......
...@@ -47,7 +47,7 @@ void keep_alnum(string& str) { ...@@ -47,7 +47,7 @@ void keep_alnum(string& str) {
int main(int argc, char** argv) { int main(int argc, char** argv) {
if (argc != 3) { if (argc != 3) {
cerr << "wrong number of arguments.\n" cerr << "wrong number of arguments.\n"
<< "usage: " << argv[0] << "input-file output-file\n"; << "usage: " << argv[0] << " input-file output-file\n";
return EXIT_FAILURE; return EXIT_FAILURE;
} }
std::ifstream in{argv[1]}; std::ifstream in{argv[1]};
...@@ -96,11 +96,15 @@ int main(int argc, char** argv) { ...@@ -96,11 +96,15 @@ int main(int argc, char** argv) {
} }
std::ofstream out{argv[2]}; std::ofstream out{argv[2]};
if (!out) { if (!out) {
cerr << "unable to open output file: " << argv[1] << '\n'; cerr << "unable to open output file: " << argv[2] << '\n';
return EXIT_FAILURE; return EXIT_FAILURE;
} }
// Print file header. // Print file header.
out << "#include \"" << namespaces[0]; out << "// clang-format off\n"
<< "// DO NOT EDIT: "
"this file is auto-generated by caf-generate-enum-strings.\n"
"// Run the target update-enum-strings if this file is out of sync.\n"
<< "#include \"" << namespaces[0];
for (size_t i = 1; i < namespaces.size(); ++i) for (size_t i = 1; i < namespaces.size(); ++i)
out << '/' << namespaces[i]; out << '/' << namespaces[i];
out << '/' << enum_name << ".hpp\"\n\n" out << '/' << enum_name << ".hpp\"\n\n"
......
execute_process(COMMAND ${CMAKE_COMMAND} -E compare_files
"${file_under_test}" "${generated_file}"
RESULT_VARIABLE result)
if(result EQUAL 0)
# files still in sync
else()
message(SEND_ERROR "${file_under_test} is out of sync! Run target "
"'update-enum-strings' to update automatically")
endif()
...@@ -192,7 +192,7 @@ Sending the same message to a group of workers is a common work flow in actor ...@@ -192,7 +192,7 @@ Sending the same message to a group of workers is a common work flow in actor
applications. Usually, a manager maintains a set of workers. On request, the applications. Usually, a manager maintains a set of workers. On request, the
manager fans-out the request to all of its workers and then collects the manager fans-out the request to all of its workers and then collects the
results. The function \lstinline`fan_out_request` combined with the merge policy results. The function \lstinline`fan_out_request` combined with the merge policy
\lstinline`fan_in_responses` streamlines this exact use case. \lstinline`select_all` streamlines this exact use case.
In the following snippet, we have a matrix actor (\lstinline`self`) that stores In the following snippet, we have a matrix actor (\lstinline`self`) that stores
worker actors for each cell (each simply storing an integer). For computing the worker actors for each cell (each simply storing an integer). For computing the
...@@ -204,6 +204,9 @@ results. ...@@ -204,6 +204,9 @@ results.
\cppexample[86-98]{message_passing/fan_out_request} \cppexample[86-98]{message_passing/fan_out_request}
The policy \lstinline`select_any` models a second common use case: sending a
request to multiple receivers but only caring for the first arriving response.
\clearpage \clearpage
\subsubsection{Error Handling in Requests} \subsubsection{Error Handling in Requests}
\label{error-response} \label{error-response}
......
...@@ -14,7 +14,7 @@ ...@@ -14,7 +14,7 @@
#include "caf/event_based_actor.hpp" #include "caf/event_based_actor.hpp"
#include "caf/exec_main.hpp" #include "caf/exec_main.hpp"
#include "caf/function_view.hpp" #include "caf/function_view.hpp"
#include "caf/policy/fan_in_responses.hpp" #include "caf/policy/select_all.hpp"
#include "caf/scoped_actor.hpp" #include "caf/scoped_actor.hpp"
#include "caf/stateful_actor.hpp" #include "caf/stateful_actor.hpp"
#include "caf/typed_actor.hpp" #include "caf/typed_actor.hpp"
...@@ -87,7 +87,7 @@ matrix::behavior_type matrix_actor(matrix::stateful_pointer<matrix_state> self, ...@@ -87,7 +87,7 @@ matrix::behavior_type matrix_actor(matrix::stateful_pointer<matrix_state> self,
assert(row < rows); assert(row < rows);
auto rp = self->make_response_promise<double>(); auto rp = self->make_response_promise<double>();
auto& row_vec = self->state.rows[row]; auto& row_vec = self->state.rows[row];
self->fan_out_request<policy::fan_in_responses>(row_vec, infinite, get) self->fan_out_request<policy::select_all>(row_vec, infinite, get)
.then( .then(
[=](std::vector<int> xs) mutable { [=](std::vector<int> xs) mutable {
assert(xs.size() == static_cast<size_t>(columns)); assert(xs.size() == static_cast<size_t>(columns));
...@@ -104,7 +104,7 @@ matrix::behavior_type matrix_actor(matrix::stateful_pointer<matrix_state> self, ...@@ -104,7 +104,7 @@ matrix::behavior_type matrix_actor(matrix::stateful_pointer<matrix_state> self,
for (int row = 0; row < rows; ++row) for (int row = 0; row < rows; ++row)
columns.emplace_back(rows_vec[row][column]); columns.emplace_back(rows_vec[row][column]);
auto rp = self->make_response_promise<double>(); auto rp = self->make_response_promise<double>();
self->fan_out_request<policy::fan_in_responses>(columns, infinite, get) self->fan_out_request<policy::select_all>(columns, infinite, get)
.then( .then(
[=](std::vector<int> xs) mutable { [=](std::vector<int> xs) mutable {
assert(xs.size() == static_cast<size_t>(rows)); assert(xs.size() == static_cast<size_t>(rows));
......
...@@ -2,28 +2,26 @@ ...@@ -2,28 +2,26 @@
file(GLOB_RECURSE CAF_CORE_HEADERS "caf/*.hpp") file(GLOB_RECURSE CAF_CORE_HEADERS "caf/*.hpp")
# -- auto generate to_string for enum types ------------------------------------ # -- add consistency checks for enum to_string implementations -----------------
enum_to_string("caf/exit_reason.hpp" "exit_reason_strings.cpp") add_enum_consistency_check("caf/sec.hpp" "src/sec_strings.cpp")
enum_to_string("caf/intrusive/inbox_result.hpp" "inbox_result_strings.cpp") add_enum_consistency_check("caf/pec.hpp" "src/pec_strings.cpp")
enum_to_string("caf/intrusive/task_result.hpp" "task_result_strings.cpp") add_enum_consistency_check("caf/stream_priority.hpp"
enum_to_string("caf/invoke_message_result.hpp" "invoke_msg_result_strings.cpp") "src/stream_priority_strings.cpp")
enum_to_string("caf/message_priority.hpp" "message_priority_strings.cpp") add_enum_consistency_check("caf/exit_reason.hpp"
enum_to_string("caf/pec.hpp" "pec_strings.cpp") "src/exit_reason_strings.cpp")
enum_to_string("caf/sec.hpp" "sec_strings.cpp") add_enum_consistency_check("caf/invoke_message_result.hpp"
enum_to_string("caf/stream_priority.hpp" "stream_priority_strings.cpp") "src/invoke_msg_result_strings.cpp")
add_enum_consistency_check("caf/message_priority.hpp"
"src/message_priority_strings.cpp")
add_enum_consistency_check("caf/intrusive/inbox_result.hpp"
"src/intrusive/inbox_result_strings.cpp")
add_enum_consistency_check("caf/intrusive/task_result.hpp"
"src/intrusive/task_result_strings.cpp")
# -- list cpp files for libcaf_core -------------------------------------------- # -- list cpp files for libcaf_core --------------------------------------------
set(CAF_CORE_SOURCES set(CAF_CORE_SOURCES
"${CMAKE_BINARY_DIR}/exit_reason_strings.cpp"
"${CMAKE_BINARY_DIR}/inbox_result_strings.cpp"
"${CMAKE_BINARY_DIR}/invoke_msg_result_strings.cpp"
"${CMAKE_BINARY_DIR}/message_priority_strings.cpp"
"${CMAKE_BINARY_DIR}/pec_strings.cpp"
"${CMAKE_BINARY_DIR}/sec_strings.cpp"
"${CMAKE_BINARY_DIR}/stream_priority_strings.cpp"
"${CMAKE_BINARY_DIR}/task_result_strings.cpp"
src/abstract_actor.cpp src/abstract_actor.cpp
src/abstract_channel.cpp src/abstract_channel.cpp
src/abstract_composable_behavior.cpp src/abstract_composable_behavior.cpp
...@@ -91,11 +89,15 @@ set(CAF_CORE_SOURCES ...@@ -91,11 +89,15 @@ set(CAF_CORE_SOURCES
src/error.cpp src/error.cpp
src/event_based_actor.cpp src/event_based_actor.cpp
src/execution_unit.cpp src/execution_unit.cpp
src/exit_reason_strings.cpp
src/forwarding_actor_proxy.cpp src/forwarding_actor_proxy.cpp
src/group.cpp src/group.cpp
src/group_manager.cpp src/group_manager.cpp
src/group_module.cpp src/group_module.cpp
src/inbound_path.cpp src/inbound_path.cpp
src/intrusive/inbox_result_strings.cpp
src/intrusive/task_result_strings.cpp
src/invoke_msg_result_strings.cpp
src/ipv4_address.cpp src/ipv4_address.cpp
src/ipv4_endpoint.cpp src/ipv4_endpoint.cpp
src/ipv4_subnet.cpp src/ipv4_subnet.cpp
...@@ -111,10 +113,12 @@ set(CAF_CORE_SOURCES ...@@ -111,10 +113,12 @@ set(CAF_CORE_SOURCES
src/message.cpp src/message.cpp
src/message_builder.cpp src/message_builder.cpp
src/message_handler.cpp src/message_handler.cpp
src/message_priority_strings.cpp
src/message_view.cpp src/message_view.cpp
src/monitorable_actor.cpp src/monitorable_actor.cpp
src/node_id.cpp src/node_id.cpp
src/outbound_path.cpp src/outbound_path.cpp
src/pec_strings.cpp
src/policy/downstream_messages.cpp src/policy/downstream_messages.cpp
src/policy/unprofiled.cpp src/policy/unprofiled.cpp
src/policy/work_sharing.cpp src/policy/work_sharing.cpp
...@@ -132,12 +136,14 @@ set(CAF_CORE_SOURCES ...@@ -132,12 +136,14 @@ set(CAF_CORE_SOURCES
src/scheduler/test_coordinator.cpp src/scheduler/test_coordinator.cpp
src/scoped_actor.cpp src/scoped_actor.cpp
src/scoped_execution_unit.cpp src/scoped_execution_unit.cpp
src/sec_strings.cpp
src/serializer.cpp src/serializer.cpp
src/settings.cpp src/settings.cpp
src/size_based_credit_controller.cpp src/size_based_credit_controller.cpp
src/skip.cpp src/skip.cpp
src/stream_aborter.cpp src/stream_aborter.cpp
src/stream_manager.cpp src/stream_manager.cpp
src/stream_priority_strings.cpp
src/string_algorithms.cpp src/string_algorithms.cpp
src/string_view.cpp src/string_view.cpp
src/term.cpp src/term.cpp
...@@ -243,7 +249,8 @@ set(CAF_CORE_TEST_SOURCES ...@@ -243,7 +249,8 @@ set(CAF_CORE_TEST_SOURCES
test/or_else.cpp test/or_else.cpp
test/pipeline_streaming.cpp test/pipeline_streaming.cpp
test/policy/categorized.cpp test/policy/categorized.cpp
test/policy/fan_in_responses.cpp test/policy/select_all.cpp
test/policy/select_any.cpp
test/request_timeout.cpp test/request_timeout.cpp
test/result.cpp test/result.cpp
test/rtti_pair.cpp test/rtti_pair.cpp
......
...@@ -67,7 +67,8 @@ void read_uri_percent_encoded(State& ps, std::string& str) { ...@@ -67,7 +67,8 @@ void read_uri_percent_encoded(State& ps, std::string& str) {
} }
inline bool uri_unprotected_char(char c) { inline bool uri_unprotected_char(char c) {
return in_whitelist(alphanumeric_chars, c) || in_whitelist("-._~", c); // Consider valid characters not explicitly stated as reserved as unreserved.
return isprint(c) && !in_whitelist(":/?#[]@!$&'()*+,;=<>", c);
} }
// clang-format off // clang-format off
...@@ -127,7 +128,9 @@ void read_uri(State& ps, Consumer&& consumer) { ...@@ -127,7 +128,9 @@ void read_uri(State& ps, Consumer&& consumer) {
return res; return res;
}; };
// Allowed character sets. // Allowed character sets.
auto path_char = [](char c) { return uri_unprotected_char(c) || c == '/'; }; auto path_char = [](char c) {
return uri_unprotected_char(c) || c == '/' || c == ':';
};
// Utility setters for avoiding code duplication. // Utility setters for avoiding code duplication.
auto set_path = [&] { consumer.path(take_str()); }; auto set_path = [&] { consumer.path(take_str()); };
auto set_host = [&] { consumer.host(take_str()); }; auto set_host = [&] { consumer.host(take_str()); };
...@@ -159,6 +162,8 @@ void read_uri(State& ps, Consumer&& consumer) { ...@@ -159,6 +162,8 @@ void read_uri(State& ps, Consumer&& consumer) {
epsilon(read_path, any_char, str += '/') epsilon(read_path, any_char, str += '/')
} }
state(start_authority) { state(start_authority) {
// A third '/' skips the authority, e.g., "file:///".
transition(read_path, '/', str += '/')
read_next_char(read_authority, str) read_next_char(read_authority, str)
fsm_transition(read_ipv6_address(ps, ip_consumer), await_end_of_ipv6, '[') fsm_transition(read_ipv6_address(ps, ip_consumer), await_end_of_ipv6, '[')
} }
......
...@@ -36,7 +36,7 @@ ...@@ -36,7 +36,7 @@
namespace caf::detail { namespace caf::detail {
template <class F, class T> template <class F, class T>
struct fan_in_responses_helper { struct select_all_helper {
std::vector<T> results; std::vector<T> results;
std::shared_ptr<size_t> pending; std::shared_ptr<size_t> pending;
F f; F f;
...@@ -51,7 +51,7 @@ struct fan_in_responses_helper { ...@@ -51,7 +51,7 @@ struct fan_in_responses_helper {
} }
template <class Fun> template <class Fun>
fan_in_responses_helper(size_t pending, Fun&& f) select_all_helper(size_t pending, Fun&& f)
: pending(std::make_shared<size_t>(pending)), f(std::forward<Fun>(f)) { : pending(std::make_shared<size_t>(pending)), f(std::forward<Fun>(f)) {
results.reserve(pending); results.reserve(pending);
} }
...@@ -62,7 +62,7 @@ struct fan_in_responses_helper { ...@@ -62,7 +62,7 @@ struct fan_in_responses_helper {
}; };
template <class F, class... Ts> template <class F, class... Ts>
struct fan_in_responses_tuple_helper { struct select_all_tuple_helper {
using value_type = std::tuple<Ts...>; using value_type = std::tuple<Ts...>;
std::vector<value_type> results; std::vector<value_type> results;
std::shared_ptr<size_t> pending; std::shared_ptr<size_t> pending;
...@@ -78,7 +78,7 @@ struct fan_in_responses_tuple_helper { ...@@ -78,7 +78,7 @@ struct fan_in_responses_tuple_helper {
} }
template <class Fun> template <class Fun>
fan_in_responses_tuple_helper(size_t pending, Fun&& f) select_all_tuple_helper(size_t pending, Fun&& f)
: pending(std::make_shared<size_t>(pending)), f(std::forward<Fun>(f)) { : pending(std::make_shared<size_t>(pending)), f(std::forward<Fun>(f)) {
results.reserve(pending); results.reserve(pending);
} }
...@@ -89,56 +89,32 @@ struct fan_in_responses_tuple_helper { ...@@ -89,56 +89,32 @@ struct fan_in_responses_tuple_helper {
}; };
template <class F, class = typename detail::get_callable_trait<F>::arg_types> template <class F, class = typename detail::get_callable_trait<F>::arg_types>
struct select_fan_in_responses_helper; struct select_select_all_helper;
template <class F, class... Ts> template <class F, class... Ts>
struct select_fan_in_responses_helper< struct select_select_all_helper<
F, detail::type_list<std::vector<std::tuple<Ts...>>>> { F, detail::type_list<std::vector<std::tuple<Ts...>>>> {
using type = fan_in_responses_tuple_helper<F, Ts...>; using type = select_all_tuple_helper<F, Ts...>;
}; };
template <class F, class T> template <class F, class T>
struct select_fan_in_responses_helper<F, detail::type_list<std::vector<T>>> { struct select_select_all_helper<F, detail::type_list<std::vector<T>>> {
using type = fan_in_responses_helper<F, T>; using type = select_all_helper<F, T>;
}; };
template <class F> template <class F>
using fan_in_responses_helper_t = using select_all_helper_t = typename select_select_all_helper<F>::type;
typename select_fan_in_responses_helper<F>::type;
// TODO: Replace with a lambda when switching to C++17 (move g into lambda).
template <class G>
class fan_in_responses_error_handler {
public:
template <class Fun>
fan_in_responses_error_handler(Fun&& fun, std::shared_ptr<size_t> pending)
: handler(std::forward<Fun>(fun)), pending(std::move(pending)) {
// nop
}
void operator()(error& err) {
CAF_LOG_TRACE(CAF_ARG2("pending", *pending));
if (*pending > 0) {
*pending = 0;
handler(err);
}
}
private:
G handler;
std::shared_ptr<size_t> pending;
};
} // namespace caf::detail } // namespace caf::detail
namespace caf::policy { namespace caf::policy {
/// Enables a `response_handle` to fan-in multiple responses into a single /// Enables a `response_handle` to fan-in all responses messages into a single
/// result (a `vector` of individual values) for the client. /// result (a `vector` that stores all received results).
/// @relates mixin::requester /// @relates mixin::requester
/// @relates response_handle /// @relates response_handle
template <class ResponseType> template <class ResponseType>
class fan_in_responses { class select_all {
public: public:
static constexpr bool is_trivial = false; static constexpr bool is_trivial = false;
...@@ -147,17 +123,18 @@ public: ...@@ -147,17 +123,18 @@ public:
using message_id_list = std::vector<message_id>; using message_id_list = std::vector<message_id>;
template <class Fun> template <class Fun>
using type_checker = detail::type_checker< using type_checker
response_type, detail::fan_in_responses_helper_t<detail::decay_t<Fun>>>; = detail::type_checker<response_type,
detail::select_all_helper_t<detail::decay_t<Fun>>>;
explicit fan_in_responses(message_id_list ids) : ids_(std::move(ids)) { explicit select_all(message_id_list ids) : ids_(std::move(ids)) {
CAF_ASSERT(ids_.size() CAF_ASSERT(ids_.size()
<= static_cast<size_t>(std::numeric_limits<int>::max())); <= static_cast<size_t>(std::numeric_limits<int>::max()));
} }
fan_in_responses(fan_in_responses&&) noexcept = default; select_all(select_all&&) noexcept = default;
fan_in_responses& operator=(fan_in_responses&&) noexcept = default; select_all& operator=(select_all&&) noexcept = default;
template <class Self, class F, class OnError> template <class Self, class F, class OnError>
void await(Self* self, F&& f, OnError&& g) const { void await(Self* self, F&& f, OnError&& g) const {
...@@ -178,7 +155,7 @@ public: ...@@ -178,7 +155,7 @@ public:
template <class Self, class F, class G> template <class Self, class F, class G>
void receive(Self* self, F&& f, G&& g) const { void receive(Self* self, F&& f, G&& g) const {
CAF_LOG_TRACE(CAF_ARG(ids_)); CAF_LOG_TRACE(CAF_ARG(ids_));
using helper_type = detail::fan_in_responses_helper_t<detail::decay_t<F>>; using helper_type = detail::select_all_helper_t<detail::decay_t<F>>;
helper_type helper{ids_.size(), std::forward<F>(f)}; helper_type helper{ids_.size(), std::forward<F>(f)};
auto error_handler = [&](error& err) { auto error_handler = [&](error& err) {
if (*helper.pending > 0) { if (*helper.pending > 0) {
...@@ -202,13 +179,20 @@ private: ...@@ -202,13 +179,20 @@ private:
template <class F, class OnError> template <class F, class OnError>
behavior make_behavior(F&& f, OnError&& g) const { behavior make_behavior(F&& f, OnError&& g) const {
using namespace detail; using namespace detail;
using helper_type = fan_in_responses_helper_t<decay_t<F>>; using helper_type = select_all_helper_t<decay_t<F>>;
using error_handler_type = fan_in_responses_error_handler<decay_t<OnError>>;
helper_type helper{ids_.size(), std::move(f)}; helper_type helper{ids_.size(), std::move(f)};
error_handler_type err_helper{std::forward<OnError>(g), helper.pending}; auto pending = helper.pending;
auto error_handler = [pending{std::move(pending)},
g{std::forward<OnError>(g)}](error& err) mutable {
CAF_LOG_TRACE(CAF_ARG2("pending", *pending));
if (*pending > 0) {
*pending = 0;
g(err);
}
};
return { return {
std::move(helper), std::move(helper),
std::move(err_helper), std::move(error_handler),
}; };
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <cstddef>
#include <memory>
#include "caf/behavior.hpp"
#include "caf/config.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/detail/typed_actor_util.hpp"
#include "caf/logger.hpp"
#include "caf/sec.hpp"
namespace caf::detail {
template <class F, class = typename get_callable_trait<F>::arg_types>
struct select_any_factory;
template <class F, class... Ts>
struct select_any_factory<F, type_list<Ts...>> {
template <class Fun>
static auto make(std::shared_ptr<size_t> pending, Fun&& fun) {
return [pending, f{std::forward<Fun>(fun)}](Ts... xs) mutable {
CAF_LOG_TRACE(CAF_ARG2("pending", *pending));
if (*pending > 0) {
f(xs...);
*pending = 0;
}
};
}
};
} // namespace caf::detail
namespace caf::policy {
/// Enables a `response_handle` to pick the first arriving response, ignoring
/// all other results.
/// @relates mixin::requester
/// @relates response_handle
template <class ResponseType>
class select_any {
public:
static constexpr bool is_trivial = false;
using response_type = ResponseType;
using message_id_list = std::vector<message_id>;
template <class Fun>
using type_checker
= detail::type_checker<response_type, detail::decay_t<Fun>>;
explicit select_any(message_id_list ids) : ids_(std::move(ids)) {
CAF_ASSERT(ids_.size()
<= static_cast<size_t>(std::numeric_limits<int>::max()));
}
template <class Self, class F, class OnError>
void await(Self* self, F&& f, OnError&& g) const {
CAF_LOG_TRACE(CAF_ARG(ids_));
auto bhvr = make_behavior(std::forward<F>(f), std::forward<OnError>(g));
for (auto id : ids_)
self->add_awaited_response_handler(id, bhvr);
}
template <class Self, class F, class OnError>
void then(Self* self, F&& f, OnError&& g) const {
CAF_LOG_TRACE(CAF_ARG(ids_));
auto bhvr = make_behavior(std::forward<F>(f), std::forward<OnError>(g));
for (auto id : ids_)
self->add_multiplexed_response_handler(id, bhvr);
}
template <class Self, class F, class G>
void receive(Self* self, F&& f, G&& g) const {
CAF_LOG_TRACE(CAF_ARG(ids_));
using factory = detail::select_any_factory<std::decay_t<F>>;
auto pending = std::make_shared<size_t>(ids_.size());
auto fw = factory::make(pending, std::forward<F>(f));
auto gw = make_error_handler(std::move(pending), std::forward<G>(g));
for (auto id : ids_) {
typename Self::accept_one_cond rc;
auto fcopy = fw;
auto gcopy = gw;
self->varargs_receive(rc, id, fcopy, gcopy);
}
}
const message_id_list& ids() const noexcept {
return ids_;
}
private:
template <class OnError>
auto make_error_handler(std::shared_ptr<size_t> p, OnError&& g) const {
return [p{std::move(p)}, g{std::forward<OnError>(g)}](error&) mutable {
if (*p == 0) {
// nop
} else if (*p == 1) {
auto err = make_error(sec::all_requests_failed);
g(err);
} else {
--*p;
}
};
}
template <class F, class OnError>
behavior make_behavior(F&& f, OnError&& g) const {
using factory = detail::select_any_factory<std::decay_t<F>>;
auto pending = std::make_shared<size_t>(ids_.size());
auto result_handler = factory::make(pending, std::forward<F>(f));
return {
std::move(result_handler),
make_error_handler(std::move(pending), std::forward<OnError>(g)),
};
}
message_id_list ids_;
};
} // namespace caf::policy
...@@ -128,10 +128,23 @@ enum class sec : uint8_t { ...@@ -128,10 +128,23 @@ enum class sec : uint8_t {
socket_operation_failed = 45, socket_operation_failed = 45,
/// A resource is temporarily unavailable or would block. /// A resource is temporarily unavailable or would block.
unavailable_or_would_block, unavailable_or_would_block,
/// Connection refused because of incompatible CAF versions.
incompatible_versions,
/// Connection refused because of incompatible application IDs.
incompatible_application_ids,
/// The middleman received a malformed BASP message from another node.
malformed_basp_message,
/// The middleman closed a connection because it failed to serialize or
/// deserialize a payload.
serializing_basp_payload_failed,
/// The middleman closed a connection to itself or an already connected node.
redundant_connection,
/// Resolving a path on a remote node failed. /// Resolving a path on a remote node failed.
remote_lookup_failed, remote_lookup_failed,
/// Serialization failed because actor_system::tracing_context is null. /// Serialization failed because actor_system::tracing_context is null.
no_tracing_context, no_tracing_context,
/// No request produced a valid result.
all_requests_failed,
}; };
/// @relates sec /// @relates sec
......
...@@ -27,6 +27,7 @@ namespace caf::detail { ...@@ -27,6 +27,7 @@ namespace caf::detail {
void append_percent_encoded(std::string& str, string_view x, bool is_path) { void append_percent_encoded(std::string& str, string_view x, bool is_path) {
for (auto ch : x) for (auto ch : x)
switch (ch) { switch (ch) {
case ':':
case '/': case '/':
if (is_path) { if (is_path) {
str += ch; str += ch;
...@@ -34,7 +35,6 @@ void append_percent_encoded(std::string& str, string_view x, bool is_path) { ...@@ -34,7 +35,6 @@ void append_percent_encoded(std::string& str, string_view x, bool is_path) {
} }
CAF_ANNOTATE_FALLTHROUGH; CAF_ANNOTATE_FALLTHROUGH;
case ' ': case ' ':
case ':':
case '?': case '?':
case '#': case '#':
case '[': case '[':
......
// clang-format off
// DO NOT EDIT: this file is auto-generated by caf-generate-enum-strings.
// Run the target update-enum-strings if this file is out of sync.
#include "caf/exit_reason.hpp"
#include <string>
namespace caf {
std::string to_string(exit_reason x) {
switch(x) {
default:
return "???";
case exit_reason::normal:
return "normal";
case exit_reason::unhandled_exception:
return "unhandled_exception";
case exit_reason::unknown:
return "unknown";
case exit_reason::out_of_workers:
return "out_of_workers";
case exit_reason::user_shutdown:
return "user_shutdown";
case exit_reason::kill:
return "kill";
case exit_reason::remote_link_unreachable:
return "remote_link_unreachable";
case exit_reason::unreachable:
return "unreachable";
};
}
} // namespace caf
// clang-format off
// DO NOT EDIT: this file is auto-generated by caf-generate-enum-strings.
// Run the target update-enum-strings if this file is out of sync.
#include "caf/intrusive/inbox_result.hpp"
#include <string>
namespace caf {
namespace intrusive {
std::string to_string(inbox_result x) {
switch(x) {
default:
return "???";
case inbox_result::success:
return "success";
case inbox_result::unblocked_reader:
return "unblocked_reader";
case inbox_result::queue_closed:
return "queue_closed";
};
}
} // namespace intrusive
} // namespace caf
// clang-format off
// DO NOT EDIT: this file is auto-generated by caf-generate-enum-strings.
// Run the target update-enum-strings if this file is out of sync.
#include "caf/intrusive/task_result.hpp"
#include <string>
namespace caf {
namespace intrusive {
std::string to_string(task_result x) {
switch(x) {
default:
return "???";
case task_result::resume:
return "resume";
case task_result::skip:
return "skip";
case task_result::stop:
return "stop";
case task_result::stop_all:
return "stop_all";
};
}
} // namespace intrusive
} // namespace caf
// clang-format off
// DO NOT EDIT: this file is auto-generated by caf-generate-enum-strings.
// Run the target update-enum-strings if this file is out of sync.
#include "caf/invoke_message_result.hpp"
#include <string>
namespace caf {
std::string to_string(invoke_message_result x) {
switch(x) {
default:
return "???";
case invoke_message_result::consumed:
return "consumed";
case invoke_message_result::skipped:
return "skipped";
case invoke_message_result::dropped:
return "dropped";
};
}
} // namespace caf
// clang-format off
// DO NOT EDIT: this file is auto-generated by caf-generate-enum-strings.
// Run the target update-enum-strings if this file is out of sync.
#include "caf/message_priority.hpp"
#include <string>
namespace caf {
std::string to_string(message_priority x) {
switch(x) {
default:
return "???";
case message_priority::high:
return "high";
case message_priority::normal:
return "normal";
};
}
} // namespace caf
// clang-format off
// DO NOT EDIT: this file is auto-generated by caf-generate-enum-strings.
// Run the target update-enum-strings if this file is out of sync.
#include "caf/pec.hpp"
#include <string>
namespace caf {
std::string to_string(pec x) {
switch(x) {
default:
return "???";
case pec::success:
return "success";
case pec::trailing_character:
return "trailing_character";
case pec::unexpected_eof:
return "unexpected_eof";
case pec::unexpected_character:
return "unexpected_character";
case pec::timespan_overflow:
return "timespan_overflow";
case pec::fractional_timespan:
return "fractional_timespan";
case pec::too_many_characters:
return "too_many_characters";
case pec::illegal_escape_sequence:
return "illegal_escape_sequence";
case pec::unexpected_newline:
return "unexpected_newline";
case pec::integer_overflow:
return "integer_overflow";
case pec::integer_underflow:
return "integer_underflow";
case pec::exponent_underflow:
return "exponent_underflow";
case pec::exponent_overflow:
return "exponent_overflow";
case pec::type_mismatch:
return "type_mismatch";
case pec::not_an_option:
return "not_an_option";
case pec::illegal_argument:
return "illegal_argument";
case pec::missing_argument:
return "missing_argument";
case pec::illegal_category:
return "illegal_category";
case pec::invalid_field_name:
return "invalid_field_name";
case pec::repeated_field_name:
return "repeated_field_name";
case pec::missing_field:
return "missing_field";
case pec::invalid_range_expression:
return "invalid_range_expression";
};
}
} // namespace caf
// clang-format off
// DO NOT EDIT: this file is auto-generated by caf-generate-enum-strings.
// Run the target update-enum-strings if this file is out of sync.
#include "caf/sec.hpp"
#include <string>
namespace caf {
std::string to_string(sec x) {
switch(x) {
default:
return "???";
case sec::none:
return "none";
case sec::unexpected_message:
return "unexpected_message";
case sec::unexpected_response:
return "unexpected_response";
case sec::request_receiver_down:
return "request_receiver_down";
case sec::request_timeout:
return "request_timeout";
case sec::no_such_group_module:
return "no_such_group_module";
case sec::no_actor_published_at_port:
return "no_actor_published_at_port";
case sec::unexpected_actor_messaging_interface:
return "unexpected_actor_messaging_interface";
case sec::state_not_serializable:
return "state_not_serializable";
case sec::unsupported_sys_key:
return "unsupported_sys_key";
case sec::unsupported_sys_message:
return "unsupported_sys_message";
case sec::disconnect_during_handshake:
return "disconnect_during_handshake";
case sec::cannot_forward_to_invalid_actor:
return "cannot_forward_to_invalid_actor";
case sec::no_route_to_receiving_node:
return "no_route_to_receiving_node";
case sec::failed_to_assign_scribe_from_handle:
return "failed_to_assign_scribe_from_handle";
case sec::failed_to_assign_doorman_from_handle:
return "failed_to_assign_doorman_from_handle";
case sec::cannot_close_invalid_port:
return "cannot_close_invalid_port";
case sec::cannot_connect_to_node:
return "cannot_connect_to_node";
case sec::cannot_open_port:
return "cannot_open_port";
case sec::network_syscall_failed:
return "network_syscall_failed";
case sec::invalid_argument:
return "invalid_argument";
case sec::invalid_protocol_family:
return "invalid_protocol_family";
case sec::cannot_publish_invalid_actor:
return "cannot_publish_invalid_actor";
case sec::cannot_spawn_actor_from_arguments:
return "cannot_spawn_actor_from_arguments";
case sec::end_of_stream:
return "end_of_stream";
case sec::no_context:
return "no_context";
case sec::unknown_type:
return "unknown_type";
case sec::no_proxy_registry:
return "no_proxy_registry";
case sec::runtime_error:
return "runtime_error";
case sec::remote_linking_failed:
return "remote_linking_failed";
case sec::cannot_add_upstream:
return "cannot_add_upstream";
case sec::upstream_already_exists:
return "upstream_already_exists";
case sec::invalid_upstream:
return "invalid_upstream";
case sec::cannot_add_downstream:
return "cannot_add_downstream";
case sec::downstream_already_exists:
return "downstream_already_exists";
case sec::invalid_downstream:
return "invalid_downstream";
case sec::no_downstream_stages_defined:
return "no_downstream_stages_defined";
case sec::stream_init_failed:
return "stream_init_failed";
case sec::invalid_stream_state:
return "invalid_stream_state";
case sec::unhandled_stream_error:
return "unhandled_stream_error";
case sec::bad_function_call:
return "bad_function_call";
case sec::feature_disabled:
return "feature_disabled";
case sec::cannot_open_file:
return "cannot_open_file";
case sec::socket_invalid:
return "socket_invalid";
case sec::socket_disconnected:
return "socket_disconnected";
case sec::socket_operation_failed:
return "socket_operation_failed";
case sec::unavailable_or_would_block:
return "unavailable_or_would_block";
case sec::incompatible_versions:
return "incompatible_versions";
case sec::incompatible_application_ids:
return "incompatible_application_ids";
case sec::malformed_basp_message:
return "malformed_basp_message";
case sec::serializing_basp_payload_failed:
return "serializing_basp_payload_failed";
case sec::redundant_connection:
return "redundant_connection";
case sec::remote_lookup_failed:
return "remote_lookup_failed";
case sec::no_tracing_context:
return "no_tracing_context";
case sec::all_requests_failed:
return "all_requests_failed";
};
}
} // namespace caf
// clang-format off
// DO NOT EDIT: this file is auto-generated by caf-generate-enum-strings.
// Run the target update-enum-strings if this file is out of sync.
#include "caf/stream_priority.hpp"
#include <string>
namespace caf {
std::string to_string(stream_priority x) {
switch(x) {
default:
return "???";
case stream_priority::very_high:
return "very_high";
case stream_priority::high:
return "high";
case stream_priority::normal:
return "normal";
case stream_priority::low:
return "low";
case stream_priority::very_low:
return "very_low";
};
}
} // namespace caf
...@@ -25,7 +25,7 @@ ...@@ -25,7 +25,7 @@
#include <numeric> #include <numeric>
#include "caf/event_based_actor.hpp" #include "caf/event_based_actor.hpp"
#include "caf/policy/fan_in_responses.hpp" #include "caf/policy/select_all.hpp"
using namespace caf; using namespace caf;
...@@ -154,7 +154,7 @@ CAF_TEST(delegated request with integer result) { ...@@ -154,7 +154,7 @@ CAF_TEST(delegated request with integer result) {
} }
CAF_TEST(requesters support fan_out_request) { CAF_TEST(requesters support fan_out_request) {
using policy::fan_in_responses; using policy::select_all;
std::vector<adding_server_type> workers{ std::vector<adding_server_type> workers{
make_server([](int x, int y) { return x + y; }), make_server([](int x, int y) { return x + y; }),
make_server([](int x, int y) { return x + y; }), make_server([](int x, int y) { return x + y; }),
...@@ -163,7 +163,7 @@ CAF_TEST(requesters support fan_out_request) { ...@@ -163,7 +163,7 @@ CAF_TEST(requesters support fan_out_request) {
run(); run();
auto sum = std::make_shared<int>(0); auto sum = std::make_shared<int>(0);
auto client = sys.spawn([=](event_based_actor* self) { auto client = sys.spawn([=](event_based_actor* self) {
self->fan_out_request<fan_in_responses>(workers, infinite, 1, 2) self->fan_out_request<select_all>(workers, infinite, 1, 2)
.then([=](std::vector<int> results) { .then([=](std::vector<int> results) {
for (auto result : results) for (auto result : results)
CAF_CHECK_EQUAL(result, 3); CAF_CHECK_EQUAL(result, 3);
......
...@@ -16,9 +16,9 @@ ...@@ -16,9 +16,9 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#define CAF_SUITE policy.fan_in_responses #define CAF_SUITE policy.select_all
#include "caf/policy/fan_in_responses.hpp" #include "caf/policy/select_all.hpp"
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
...@@ -28,7 +28,7 @@ ...@@ -28,7 +28,7 @@
#include "caf/event_based_actor.hpp" #include "caf/event_based_actor.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
using caf::policy::fan_in_responses; using caf::policy::select_all;
using namespace caf; using namespace caf;
...@@ -63,9 +63,9 @@ struct fixture : test_coordinator_fixture<> { ...@@ -63,9 +63,9 @@ struct fixture : test_coordinator_fixture<> {
CAF_MESSAGE("subtest: " message); \ CAF_MESSAGE("subtest: " message); \
for (int subtest_dummy = 0; subtest_dummy < 1; ++subtest_dummy) for (int subtest_dummy = 0; subtest_dummy < 1; ++subtest_dummy)
CAF_TEST_FIXTURE_SCOPE(fan_in_responses_tests, fixture) CAF_TEST_FIXTURE_SCOPE(select_all_tests, fixture)
CAF_TEST(fan_in_responses combines two integer results into one vector) { CAF_TEST(select_all combines two integer results into one vector) {
using int_list = std::vector<int>; using int_list = std::vector<int>;
auto f = [](int x, int y) { return x + y; }; auto f = [](int x, int y) { return x + y; };
auto server1 = make_server(f); auto server1 = make_server(f);
...@@ -74,7 +74,7 @@ CAF_TEST(fan_in_responses combines two integer results into one vector) { ...@@ -74,7 +74,7 @@ CAF_TEST(fan_in_responses combines two integer results into one vector) {
SUBTEST("vector of int") { SUBTEST("vector of int") {
auto r1 = self->request(server1, infinite, 1, 2); auto r1 = self->request(server1, infinite, 1, 2);
auto r2 = self->request(server2, infinite, 2, 3); auto r2 = self->request(server2, infinite, 2, 3);
fan_in_responses<detail::type_list<int>> merge{{r1.id(), r2.id()}}; select_all<detail::type_list<int>> merge{{r1.id(), r2.id()}};
run(); run();
merge.receive( merge.receive(
self.ptr(), self.ptr(),
...@@ -88,7 +88,7 @@ CAF_TEST(fan_in_responses combines two integer results into one vector) { ...@@ -88,7 +88,7 @@ CAF_TEST(fan_in_responses combines two integer results into one vector) {
using std::make_tuple; using std::make_tuple;
auto r1 = self->request(server1, infinite, 1, 2); auto r1 = self->request(server1, infinite, 1, 2);
auto r2 = self->request(server2, infinite, 2, 3); auto r2 = self->request(server2, infinite, 2, 3);
fan_in_responses<detail::type_list<int>> merge{{r1.id(), r2.id()}}; select_all<detail::type_list<int>> merge{{r1.id(), r2.id()}};
run(); run();
using results_vector = std::vector<std::tuple<int>>; using results_vector = std::vector<std::tuple<int>>;
merge.receive( merge.receive(
...@@ -106,7 +106,7 @@ CAF_TEST(fan_in_responses combines two integer results into one vector) { ...@@ -106,7 +106,7 @@ CAF_TEST(fan_in_responses combines two integer results into one vector) {
auto client = sys.spawn([=, &results](event_based_actor* client_ptr) { auto client = sys.spawn([=, &results](event_based_actor* client_ptr) {
auto r1 = client_ptr->request(server1, infinite, 1, 2); auto r1 = client_ptr->request(server1, infinite, 1, 2);
auto r2 = client_ptr->request(server2, infinite, 2, 3); auto r2 = client_ptr->request(server2, infinite, 2, 3);
fan_in_responses<detail::type_list<int>> merge{{r1.id(), r2.id()}}; select_all<detail::type_list<int>> merge{{r1.id(), r2.id()}};
merge.then( merge.then(
client_ptr, [&results](int_list xs) { results = std::move(xs); }, client_ptr, [&results](int_list xs) { results = std::move(xs); },
make_error_handler()); make_error_handler());
...@@ -124,7 +124,7 @@ CAF_TEST(fan_in_responses combines two integer results into one vector) { ...@@ -124,7 +124,7 @@ CAF_TEST(fan_in_responses combines two integer results into one vector) {
auto client = sys.spawn([=, &results](event_based_actor* client_ptr) { auto client = sys.spawn([=, &results](event_based_actor* client_ptr) {
auto r1 = client_ptr->request(server1, infinite, 1, 2); auto r1 = client_ptr->request(server1, infinite, 1, 2);
auto r2 = client_ptr->request(server2, infinite, 2, 3); auto r2 = client_ptr->request(server2, infinite, 2, 3);
fan_in_responses<detail::type_list<int>> merge{{r1.id(), r2.id()}}; select_all<detail::type_list<int>> merge{{r1.id(), r2.id()}};
merge.await( merge.await(
client_ptr, [&results](int_list xs) { results = std::move(xs); }, client_ptr, [&results](int_list xs) { results = std::move(xs); },
make_error_handler()); make_error_handler());
...@@ -141,7 +141,7 @@ CAF_TEST(fan_in_responses combines two integer results into one vector) { ...@@ -141,7 +141,7 @@ CAF_TEST(fan_in_responses combines two integer results into one vector) {
} }
} }
CAF_TEST(fan_in_responses calls the error handler at most once) { CAF_TEST(select_all calls the error handler at most once) {
using int_list = std::vector<int>; using int_list = std::vector<int>;
auto f = [](int, int) -> result<int> { return sec::invalid_argument; }; auto f = [](int, int) -> result<int> { return sec::invalid_argument; };
auto server1 = make_server(f); auto server1 = make_server(f);
...@@ -149,7 +149,7 @@ CAF_TEST(fan_in_responses calls the error handler at most once) { ...@@ -149,7 +149,7 @@ CAF_TEST(fan_in_responses calls the error handler at most once) {
SUBTEST("request.receive") { SUBTEST("request.receive") {
auto r1 = self->request(server1, infinite, 1, 2); auto r1 = self->request(server1, infinite, 1, 2);
auto r2 = self->request(server2, infinite, 2, 3); auto r2 = self->request(server2, infinite, 2, 3);
fan_in_responses<detail::type_list<int>> merge{{r1.id(), r2.id()}}; select_all<detail::type_list<int>> merge{{r1.id(), r2.id()}};
run(); run();
size_t errors = 0; size_t errors = 0;
merge.receive( merge.receive(
...@@ -163,7 +163,7 @@ CAF_TEST(fan_in_responses calls the error handler at most once) { ...@@ -163,7 +163,7 @@ CAF_TEST(fan_in_responses calls the error handler at most once) {
auto client = sys.spawn([=, &errors](event_based_actor* client_ptr) { auto client = sys.spawn([=, &errors](event_based_actor* client_ptr) {
auto r1 = client_ptr->request(server1, infinite, 1, 2); auto r1 = client_ptr->request(server1, infinite, 1, 2);
auto r2 = client_ptr->request(server2, infinite, 2, 3); auto r2 = client_ptr->request(server2, infinite, 2, 3);
fan_in_responses<detail::type_list<int>> merge{{r1.id(), r2.id()}}; select_all<detail::type_list<int>> merge{{r1.id(), r2.id()}};
merge.then( merge.then(
client_ptr, client_ptr,
[](int_list) { CAF_FAIL("fan-in policy called the result handler"); }, [](int_list) { CAF_FAIL("fan-in policy called the result handler"); },
...@@ -181,7 +181,7 @@ CAF_TEST(fan_in_responses calls the error handler at most once) { ...@@ -181,7 +181,7 @@ CAF_TEST(fan_in_responses calls the error handler at most once) {
auto client = sys.spawn([=, &errors](event_based_actor* client_ptr) { auto client = sys.spawn([=, &errors](event_based_actor* client_ptr) {
auto r1 = client_ptr->request(server1, infinite, 1, 2); auto r1 = client_ptr->request(server1, infinite, 1, 2);
auto r2 = client_ptr->request(server2, infinite, 2, 3); auto r2 = client_ptr->request(server2, infinite, 2, 3);
fan_in_responses<detail::type_list<int>> merge{{r1.id(), r2.id()}}; select_all<detail::type_list<int>> merge{{r1.id(), r2.id()}};
merge.await( merge.await(
client_ptr, client_ptr,
[](int_list) { CAF_FAIL("fan-in policy called the result handler"); }, [](int_list) { CAF_FAIL("fan-in policy called the result handler"); },
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 policy.select_any
#include "caf/policy/select_any.hpp"
#include "caf/test/dsl.hpp"
#include "caf/actor_system.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/sec.hpp"
using caf::policy::select_any;
using namespace caf;
namespace {
struct fixture : test_coordinator_fixture<> {
template <class F>
actor make_server(F f) {
auto init = [f]() -> behavior {
return {
[f](int x, int y) { return f(x, y); },
};
};
return sys.spawn(init);
}
auto make_error_handler() {
return [this](const error& err) {
CAF_FAIL("unexpected error: " << sys.render(err));
};
}
auto make_counting_error_handler(size_t* count) {
return [count](const error&) { *count += 1; };
}
};
} // namespace
#define SUBTEST(message) \
run(); \
CAF_MESSAGE("subtest: " message); \
for (int subtest_dummy = 0; subtest_dummy < 1; ++subtest_dummy)
CAF_TEST_FIXTURE_SCOPE(select_any_tests, fixture)
CAF_TEST(select_any picks the first arriving integer) {
auto f = [](int x, int y) { return x + y; };
auto server1 = make_server(f);
auto server2 = make_server(f);
SUBTEST("request.receive") {
SUBTEST("single integer") {
auto r1 = self->request(server1, infinite, 1, 2);
auto r2 = self->request(server2, infinite, 2, 3);
select_any<detail::type_list<int>> choose{{r1.id(), r2.id()}};
run();
choose.receive(
self.ptr(), [](int result) { CAF_CHECK_EQUAL(result, 3); },
make_error_handler());
}
}
SUBTEST("request.then") {
int result = 0;
auto client = sys.spawn([=, &result](event_based_actor* client_ptr) {
auto r1 = client_ptr->request(server1, infinite, 1, 2);
auto r2 = client_ptr->request(server2, infinite, 2, 3);
select_any<detail::type_list<int>> choose{{r1.id(), r2.id()}};
choose.then(
client_ptr, [&result](int x) { result = x; }, make_error_handler());
});
run_once();
expect((int, int), from(client).to(server1).with(1, 2));
expect((int, int), from(client).to(server2).with(2, 3));
expect((int), from(server1).to(client).with(3));
expect((int), from(server2).to(client).with(5));
CAF_MESSAGE("request.then picks the first arriving result");
CAF_CHECK_EQUAL(result, 3);
}
SUBTEST("request.await") {
int result = 0;
auto client = sys.spawn([=, &result](event_based_actor* client_ptr) {
auto r1 = client_ptr->request(server1, infinite, 1, 2);
auto r2 = client_ptr->request(server2, infinite, 2, 3);
select_any<detail::type_list<int>> choose{{r1.id(), r2.id()}};
choose.await(
client_ptr, [&result](int x) { result = x; }, make_error_handler());
});
run_once();
expect((int, int), from(client).to(server1).with(1, 2));
expect((int, int), from(client).to(server2).with(2, 3));
// TODO: DSL (mailbox.peek) cannot deal with receivers that skip messages.
// expect((int), from(server1).to(client).with(3));
// expect((int), from(server2).to(client).with(5));
run();
CAF_MESSAGE("request.await froces responses into reverse request order");
CAF_CHECK_EQUAL(result, 5);
}
}
CAF_TEST(select_any calls the error handler at most once) {
auto f = [](int, int) -> result<int> { return sec::invalid_argument; };
auto server1 = make_server(f);
auto server2 = make_server(f);
SUBTEST("request.receive") {
auto r1 = self->request(server1, infinite, 1, 2);
auto r2 = self->request(server2, infinite, 2, 3);
select_any<detail::type_list<int>> choose{{r1.id(), r2.id()}};
run();
size_t errors = 0;
choose.receive(
self.ptr(),
[](int) { CAF_FAIL("fan-in policy called the result handler"); },
make_counting_error_handler(&errors));
CAF_CHECK_EQUAL(errors, 1u);
}
SUBTEST("request.then") {
size_t errors = 0;
auto client = sys.spawn([=, &errors](event_based_actor* client_ptr) {
auto r1 = client_ptr->request(server1, infinite, 1, 2);
auto r2 = client_ptr->request(server2, infinite, 2, 3);
select_any<detail::type_list<int>> choose{{r1.id(), r2.id()}};
choose.then(
client_ptr,
[](int) { CAF_FAIL("fan-in policy called the result handler"); },
make_counting_error_handler(&errors));
});
run_once();
expect((int, int), from(client).to(server1).with(1, 2));
expect((int, int), from(client).to(server2).with(2, 3));
expect((error), from(server1).to(client).with(sec::invalid_argument));
expect((error), from(server2).to(client).with(sec::invalid_argument));
CAF_CHECK_EQUAL(errors, 1u);
}
SUBTEST("request.await") {
size_t errors = 0;
auto client = sys.spawn([=, &errors](event_based_actor* client_ptr) {
auto r1 = client_ptr->request(server1, infinite, 1, 2);
auto r2 = client_ptr->request(server2, infinite, 2, 3);
select_any<detail::type_list<int>> choose{{r1.id(), r2.id()}};
choose.await(
client_ptr,
[](int) { CAF_FAIL("fan-in policy called the result handler"); },
make_counting_error_handler(&errors));
});
run_once();
expect((int, int), from(client).to(server1).with(1, 2));
expect((int, int), from(client).to(server2).with(2, 3));
// TODO: DSL (mailbox.peek) cannot deal with receivers that skip messages.
// expect((int), from(server1).to(client).with(3));
// expect((int), from(server2).to(client).with(5));
run();
CAF_CHECK_EQUAL(errors, 1u);
}
}
CAF_TEST_FIXTURE_SCOPE_END()
...@@ -284,6 +284,7 @@ CAF_TEST(from string) { ...@@ -284,6 +284,7 @@ CAF_TEST(from string) {
// all combinations of components // all combinations of components
ROUNDTRIP("http:file"); ROUNDTRIP("http:file");
ROUNDTRIP("http:foo-bar"); ROUNDTRIP("http:foo-bar");
ROUNDTRIP("http:foo:bar");
ROUNDTRIP("http:file?a=1&b=2"); ROUNDTRIP("http:file?a=1&b=2");
ROUNDTRIP("http:file#42"); ROUNDTRIP("http:file#42");
ROUNDTRIP("http:file?a=1&b=2#42"); ROUNDTRIP("http:file?a=1&b=2#42");
...@@ -354,12 +355,14 @@ CAF_TEST(from string) { ...@@ -354,12 +355,14 @@ CAF_TEST(from string) {
ROUNDTRIP("http://me@[::1]:80/file?a=1&b=2#42"); ROUNDTRIP("http://me@[::1]:80/file?a=1&b=2#42");
// percent encoding // percent encoding
ROUNDTRIP("hi%20there://it%27s@me%21/file%201#%5B42%5D"); ROUNDTRIP("hi%20there://it%27s@me%21/file%201#%5B42%5D");
ROUNDTRIP("file://localhost/tmp/test/test.{%3A04d}.exr");
} }
#undef ROUNDTRIP #undef ROUNDTRIP
CAF_TEST(empty components) { CAF_TEST(empty components) {
CAF_CHECK_EQUAL("foo:/"_u, "foo:/"); CAF_CHECK_EQUAL("foo:/"_u, "foo:/");
CAF_CHECK_EQUAL("foo:///"_u, "foo:/");
CAF_CHECK_EQUAL("foo:/#"_u, "foo:/"); CAF_CHECK_EQUAL("foo:/#"_u, "foo:/");
CAF_CHECK_EQUAL("foo:/?"_u, "foo:/"); CAF_CHECK_EQUAL("foo:/?"_u, "foo:/");
CAF_CHECK_EQUAL("foo:/?#"_u, "foo:/"); CAF_CHECK_EQUAL("foo:/?#"_u, "foo:/");
......
...@@ -2,21 +2,23 @@ ...@@ -2,21 +2,23 @@
file(GLOB_RECURSE CAF_IO_HEADERS "caf/*.hpp") file(GLOB_RECURSE CAF_IO_HEADERS "caf/*.hpp")
# -- auto generate to_string for enum types ------------------------------------ # -- add consistency checks for enum to_string implementations -----------------
enum_to_string("caf/io/basp/message_type.hpp" "message_type_to_string.cpp") add_enum_consistency_check("caf/io/basp/message_type.hpp"
enum_to_string("caf/io/network/operation.hpp" "operation_to_string.cpp") "src/io/basp/message_type_strings.cpp")
add_enum_consistency_check("caf/io/network/operation.hpp"
"src/io/network/operation_strings.cpp")
# -- list cpp files ------------------------------------------------------------ # -- list cpp files ------------------------------------------------------------
set(CAF_IO_SOURCES set(CAF_IO_SOURCES
"${CMAKE_BINARY_DIR}/message_type_to_string.cpp"
"${CMAKE_BINARY_DIR}/operation_to_string.cpp"
src/detail/socket_guard.cpp src/detail/socket_guard.cpp
src/io/abstract_broker.cpp src/io/abstract_broker.cpp
src/io/basp/header.cpp src/io/basp/header.cpp
src/io/basp/instance.cpp src/io/basp/instance.cpp
src/io/basp/message_queue.cpp src/io/basp/message_queue.cpp
src/io/basp/message_type_strings.cpp
src/io/basp/routing_table.cpp src/io/basp/routing_table.cpp
src/io/basp/worker.cpp src/io/basp/worker.cpp
src/io/basp_broker.cpp src/io/basp_broker.cpp
...@@ -40,6 +42,7 @@ set(CAF_IO_SOURCES ...@@ -40,6 +42,7 @@ set(CAF_IO_SOURCES
src/io/network/manager.cpp src/io/network/manager.cpp
src/io/network/multiplexer.cpp src/io/network/multiplexer.cpp
src/io/network/native_socket.cpp src/io/network/native_socket.cpp
src/io/network/operation_strings.cpp
src/io/network/pipe_reader.cpp src/io/network/pipe_reader.cpp
src/io/network/protocol.cpp src/io/network/protocol.cpp
src/io/network/receive_buffer.cpp src/io/network/receive_buffer.cpp
......
...@@ -18,11 +18,15 @@ ...@@ -18,11 +18,15 @@
#pragma once #pragma once
#include "caf/detail/io_export.hpp"
#include "caf/sec.hpp"
namespace caf::io::basp { namespace caf::io::basp {
/// @addtogroup BASP /// @addtogroup BASP
/// Denotes the state of a connection between to BASP nodes. /// Denotes the state of a connection between two BASP nodes. Overlaps with
/// `sec` (these states get converted to an error by the BASP instance).
enum connection_state { enum connection_state {
/// Indicates that a connection is established and this node is /// Indicates that a connection is established and this node is
/// waiting for the next BASP header. /// waiting for the next BASP header.
...@@ -31,16 +35,55 @@ enum connection_state { ...@@ -31,16 +35,55 @@ enum connection_state {
/// and is waiting for the data. /// and is waiting for the data.
await_payload, await_payload,
/// Indicates that this connection no longer exists. /// Indicates that this connection no longer exists.
close_connection close_connection,
/// See `sec::incompatible_versions`.
incompatible_versions,
/// See `sec::incompatible_application_ids`.
incompatible_application_ids,
/// See `sec::malformed_basp_message`.
malformed_basp_message,
/// See `sec::serializing_basp_payload_failed`.
serializing_basp_payload_failed,
/// See `sec::redundant_connection`.
redundant_connection,
/// See `sec::no_route_to_receiving_node`.
no_route_to_receiving_node,
}; };
/// Returns whether the connection state requries a shutdown of the socket
/// connection.
/// @relates connection_state
inline bool requires_shutdown(connection_state x) {
// Any enum value other than await_header (0) and await_payload (1) signal the
// BASP broker to shutdown the connection.
return static_cast<int>(x) > 1;
}
/// Converts the connection state to a system error code if it holds one of the
/// overlapping values. Otherwise returns `sec::none`.
/// @relates connection_state /// @relates connection_state
inline std::string to_string(connection_state x) { inline sec to_sec(connection_state x) {
return x == await_header switch (x) {
? "await_header" default:
: (x == await_payload ? "await_payload" : "close_connection"); return sec::none;
case incompatible_versions:
return sec::incompatible_versions;
case incompatible_application_ids:
return sec::incompatible_application_ids;
case malformed_basp_message:
return sec::malformed_basp_message;
case serializing_basp_payload_failed:
return sec::serializing_basp_payload_failed;
case redundant_connection:
return sec::redundant_connection;
case no_route_to_receiving_node:
return sec::no_route_to_receiving_node;
}
} }
/// @relates connection_state
CAF_IO_EXPORT std::string to_string(connection_state x);
/// @} /// @}
} // namespace caf::io::basp } // namespace caf::io::basp
...@@ -221,8 +221,8 @@ public: ...@@ -221,8 +221,8 @@ public:
return system().config(); return system().config();
} }
bool handle(execution_unit* ctx, connection_handle hdl, header& hdr, connection_state handle(execution_unit* ctx, connection_handle hdl,
byte_buffer* payload); header& hdr, byte_buffer* payload);
private: private:
void forward(execution_unit* ctx, const node_id& dest_node, const header& hdr, void forward(execution_unit* ctx, const node_id& dest_node, const header& hdr,
......
...@@ -111,7 +111,7 @@ public: ...@@ -111,7 +111,7 @@ public:
void set_context(connection_handle hdl); void set_context(connection_handle hdl);
/// Cleans up any state for `hdl`. /// Cleans up any state for `hdl`.
void connection_cleanup(connection_handle hdl); void connection_cleanup(connection_handle hdl, sec code);
/// Sends a basp::down_message message to a remote node. /// Sends a basp::down_message message to a remote node.
void send_basp_down_message(const node_id& nid, actor_id aid, error err); void send_basp_down_message(const node_id& nid, actor_id aid, error err);
......
...@@ -51,10 +51,10 @@ connection_state instance::handle(execution_unit* ctx, new_data_msg& dm, ...@@ -51,10 +51,10 @@ connection_state instance::handle(execution_unit* ctx, new_data_msg& dm,
header& hdr, bool is_payload) { header& hdr, bool is_payload) {
CAF_LOG_TRACE(CAF_ARG(dm) << CAF_ARG(is_payload)); CAF_LOG_TRACE(CAF_ARG(dm) << CAF_ARG(is_payload));
// function object providing cleanup code on errors // function object providing cleanup code on errors
auto err = [&]() -> connection_state { auto err = [&](connection_state code) {
if (auto nid = tbl_.erase_direct(dm.handle)) if (auto nid = tbl_.erase_direct(dm.handle))
callee_.purge_state(nid); callee_.purge_state(nid);
return close_connection; return code;
}; };
byte_buffer* payload = nullptr; byte_buffer* payload = nullptr;
if (is_payload) { if (is_payload) {
...@@ -62,14 +62,14 @@ connection_state instance::handle(execution_unit* ctx, new_data_msg& dm, ...@@ -62,14 +62,14 @@ connection_state instance::handle(execution_unit* ctx, new_data_msg& dm,
if (payload->size() != hdr.payload_len) { if (payload->size() != hdr.payload_len) {
CAF_LOG_WARNING("received invalid payload, expected" CAF_LOG_WARNING("received invalid payload, expected"
<< hdr.payload_len << "bytes, got" << payload->size()); << hdr.payload_len << "bytes, got" << payload->size());
return err(); return err(malformed_basp_message);
} }
} else { } else {
binary_deserializer bd{ctx, dm.buf}; binary_deserializer bd{ctx, dm.buf};
auto e = bd(hdr); auto e = bd(hdr);
if (e || !valid(hdr)) { if (e || !valid(hdr)) {
CAF_LOG_WARNING("received invalid header:" << CAF_ARG(hdr)); CAF_LOG_WARNING("received invalid header:" << CAF_ARG(hdr));
return err(); return err(malformed_basp_message);
} }
if (hdr.payload_len > 0) { if (hdr.payload_len > 0) {
CAF_LOG_DEBUG("await payload before processing further"); CAF_LOG_DEBUG("await payload before processing further");
...@@ -77,9 +77,7 @@ connection_state instance::handle(execution_unit* ctx, new_data_msg& dm, ...@@ -77,9 +77,7 @@ connection_state instance::handle(execution_unit* ctx, new_data_msg& dm,
} }
} }
CAF_LOG_DEBUG(CAF_ARG(hdr)); CAF_LOG_DEBUG(CAF_ARG(hdr));
if (!handle(ctx, dm.handle, hdr, payload)) return handle(ctx, dm.handle, hdr, payload);
return err();
return await_header;
} }
void instance::handle_heartbeat(execution_unit* ctx) { void instance::handle_heartbeat(execution_unit* ctx) {
...@@ -286,18 +284,18 @@ void instance::write_heartbeat(execution_unit* ctx, byte_buffer& buf) { ...@@ -286,18 +284,18 @@ void instance::write_heartbeat(execution_unit* ctx, byte_buffer& buf) {
write(ctx, buf, hdr); write(ctx, buf, hdr);
} }
bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, connection_state instance::handle(execution_unit* ctx, connection_handle hdl,
byte_buffer* payload) { header& hdr, byte_buffer* payload) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(hdr)); CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(hdr));
// Check payload validity. // Check payload validity.
if (payload == nullptr) { if (payload == nullptr) {
if (hdr.payload_len != 0) { if (hdr.payload_len != 0) {
CAF_LOG_WARNING("invalid payload"); CAF_LOG_WARNING("missing payload");
return false; return malformed_basp_message;
} }
} else if (hdr.payload_len != payload->size()) { } else if (hdr.payload_len != payload->size()) {
CAF_LOG_WARNING("invalid payload"); CAF_LOG_WARNING("actual payload size differs from advertised size");
return false; return malformed_basp_message;
} }
// Dispatch by message type. // Dispatch by message type.
switch (hdr.operation) { switch (hdr.operation) {
...@@ -311,7 +309,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -311,7 +309,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if (auto err = bd(source_node, app_ids, aid, sigs)) { if (auto err = bd(source_node, app_ids, aid, sigs)) {
CAF_LOG_WARNING("unable to deserialize payload of server handshake:" CAF_LOG_WARNING("unable to deserialize payload of server handshake:"
<< ctx->system().render(err)); << ctx->system().render(err));
return false; return serializing_basp_payload_failed;
} }
// Check the application ID. // Check the application ID.
auto whitelist = get_or(config(), "middleman.app-identifiers", auto whitelist = get_or(config(), "middleman.app-identifiers",
...@@ -321,20 +319,20 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -321,20 +319,20 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if (i == app_ids.end()) { if (i == app_ids.end()) {
CAF_LOG_WARNING("refuse to connect to server due to app ID mismatch:" CAF_LOG_WARNING("refuse to connect to server due to app ID mismatch:"
<< CAF_ARG(app_ids) << CAF_ARG(whitelist)); << CAF_ARG(app_ids) << CAF_ARG(whitelist));
return false; return incompatible_application_ids;
} }
// Close connection to ourselves immediately after sending client HS. // Close connection to ourselves immediately after sending client HS.
if (source_node == this_node_) { if (source_node == this_node_) {
CAF_LOG_DEBUG("close connection to self immediately"); CAF_LOG_DEBUG("close connection to self immediately");
callee_.finalize_handshake(source_node, aid, sigs); callee_.finalize_handshake(source_node, aid, sigs);
return false; return redundant_connection;
} }
// Close this connection if we already have a direct connection. // Close this connection if we already have a direct connection.
if (tbl_.lookup_direct(source_node)) { if (tbl_.lookup_direct(source_node)) {
CAF_LOG_DEBUG( CAF_LOG_DEBUG(
"close redundant direct connection:" << CAF_ARG(source_node)); "close redundant direct connection:" << CAF_ARG(source_node));
callee_.finalize_handshake(source_node, aid, sigs); callee_.finalize_handshake(source_node, aid, sigs);
return false; return redundant_connection;
} }
// Add direct route to this node and remove any indirect entry. // Add direct route to this node and remove any indirect entry.
CAF_LOG_DEBUG("new direct connection:" << CAF_ARG(source_node)); CAF_LOG_DEBUG("new direct connection:" << CAF_ARG(source_node));
...@@ -344,7 +342,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -344,7 +342,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
auto path = tbl_.lookup(source_node); auto path = tbl_.lookup(source_node);
if (!path) { if (!path) {
CAF_LOG_ERROR("no route to host after server handshake"); CAF_LOG_ERROR("no route to host after server handshake");
return false; return no_route_to_receiving_node;
} }
write_client_handshake(ctx, callee_.get_buffer(path->hdl)); write_client_handshake(ctx, callee_.get_buffer(path->hdl));
callee_.learned_new_node_directly(source_node, was_indirect); callee_.learned_new_node_directly(source_node, was_indirect);
...@@ -359,7 +357,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -359,7 +357,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if (auto err = bd(source_node)) { if (auto err = bd(source_node)) {
CAF_LOG_WARNING("unable to deserialize payload of client handshake:" CAF_LOG_WARNING("unable to deserialize payload of client handshake:"
<< ctx->system().render(err)); << ctx->system().render(err));
return false; return serializing_basp_payload_failed;
} }
// Drop repeated handshakes. // Drop repeated handshakes.
if (tbl_.lookup_direct(source_node)) { if (tbl_.lookup_direct(source_node)) {
...@@ -383,11 +381,11 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -383,11 +381,11 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
CAF_LOG_WARNING( CAF_LOG_WARNING(
"unable to deserialize source and destination for routed message:" "unable to deserialize source and destination for routed message:"
<< ctx->system().render(err)); << ctx->system().render(err));
return false; return serializing_basp_payload_failed;
} }
if (dest_node != this_node_) { if (dest_node != this_node_) {
forward(ctx, dest_node, hdr, *payload); forward(ctx, dest_node, hdr, *payload);
return true; return await_header;
} }
auto last_hop = tbl_.lookup_direct(hdl); auto last_hop = tbl_.lookup_direct(hdl);
if (source_node != none && source_node != this_node_ if (source_node != none && source_node != this_node_
...@@ -441,7 +439,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -441,7 +439,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if (auto err = bd(source_node, dest_node)) { if (auto err = bd(source_node, dest_node)) {
CAF_LOG_WARNING("unable to deserialize payload of monitor message:" CAF_LOG_WARNING("unable to deserialize payload of monitor message:"
<< ctx->system().render(err)); << ctx->system().render(err));
return false; return serializing_basp_payload_failed;
} }
if (dest_node == this_node_) if (dest_node == this_node_)
callee_.proxy_announced(source_node, hdr.dest_actor); callee_.proxy_announced(source_node, hdr.dest_actor);
...@@ -458,7 +456,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -458,7 +456,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if (auto err = bd(source_node, dest_node, fail_state)) { if (auto err = bd(source_node, dest_node, fail_state)) {
CAF_LOG_WARNING("unable to deserialize payload of down message:" CAF_LOG_WARNING("unable to deserialize payload of down message:"
<< ctx->system().render(err)); << ctx->system().render(err));
return false; return serializing_basp_payload_failed;
} }
if (dest_node == this_node_) { if (dest_node == this_node_) {
// Delay this message to make sure we don't skip in-flight messages. // Delay this message to make sure we don't skip in-flight messages.
...@@ -481,10 +479,10 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -481,10 +479,10 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
} }
default: { default: {
CAF_LOG_ERROR("invalid operation"); CAF_LOG_ERROR("invalid operation");
return false; return malformed_basp_message;
} }
} }
return true; return await_header;
} }
void instance::forward(execution_unit* ctx, const node_id& dest_node, void instance::forward(execution_unit* ctx, const node_id& dest_node,
......
// clang-format off
// DO NOT EDIT: this file is auto-generated by caf-generate-enum-strings.
// Run the target update-enum-strings if this file is out of sync.
#include "caf/io/basp/message_type.hpp"
#include <string>
namespace caf {
namespace io {
namespace basp {
std::string to_string(message_type x) {
switch(x) {
default:
return "???";
case message_type::server_handshake:
return "server_handshake";
case message_type::client_handshake:
return "client_handshake";
case message_type::direct_message:
return "direct_message";
case message_type::routed_message:
return "routed_message";
case message_type::monitor_message:
return "monitor_message";
case message_type::down_message:
return "down_message";
case message_type::heartbeat:
return "heartbeat";
};
}
} // namespace basp
} // namespace io
} // namespace caf
...@@ -127,8 +127,8 @@ behavior basp_broker::make_behavior() { ...@@ -127,8 +127,8 @@ behavior basp_broker::make_behavior() {
auto& ctx = *this_context; auto& ctx = *this_context;
auto next = instance.handle(context(), msg, ctx.hdr, auto next = instance.handle(context(), msg, ctx.hdr,
ctx.cstate == basp::await_payload); ctx.cstate == basp::await_payload);
if (next == basp::close_connection) { if (requires_shutdown(next)) {
connection_cleanup(msg.handle); connection_cleanup(msg.handle, to_sec(next));
close(msg.handle); close(msg.handle);
return; return;
} }
...@@ -224,7 +224,9 @@ behavior basp_broker::make_behavior() { ...@@ -224,7 +224,9 @@ behavior basp_broker::make_behavior() {
msg.handle)); msg.handle));
}, },
// received from the message handler above for connection_closed_msg // received from the message handler above for connection_closed_msg
[=](delete_atom, connection_handle hdl) { connection_cleanup(hdl); }, [=](delete_atom, connection_handle hdl) {
connection_cleanup(hdl, sec::none);
},
// received from underlying broker implementation // received from underlying broker implementation
[=](const acceptor_closed_msg& msg) { [=](const acceptor_closed_msg& msg) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
...@@ -558,8 +560,8 @@ void basp_broker::set_context(connection_handle hdl) { ...@@ -558,8 +560,8 @@ void basp_broker::set_context(connection_handle hdl) {
t_last_hop = &i->second.id; t_last_hop = &i->second.id;
} }
void basp_broker::connection_cleanup(connection_handle hdl) { void basp_broker::connection_cleanup(connection_handle hdl, sec code) {
CAF_LOG_TRACE(CAF_ARG(hdl)); CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(code));
// Remove handle from the routing table and clean up any node-specific state // Remove handle from the routing table and clean up any node-specific state
// we might still have. // we might still have.
if (auto nid = instance.tbl().erase_direct(hdl)) if (auto nid = instance.tbl().erase_direct(hdl))
...@@ -571,8 +573,9 @@ void basp_broker::connection_cleanup(connection_handle hdl) { ...@@ -571,8 +573,9 @@ void basp_broker::connection_cleanup(connection_handle hdl) {
auto& ref = i->second; auto& ref = i->second;
CAF_ASSERT(i->first == ref.hdl); CAF_ASSERT(i->first == ref.hdl);
if (ref.callback) { if (ref.callback) {
CAF_LOG_DEBUG("connection closed during handshake"); CAF_LOG_DEBUG("connection closed during handshake:" << CAF_ARG(code));
ref.callback->deliver(sec::disconnect_during_handshake); auto x = code != sec::none ? code : sec::disconnect_during_handshake;
ref.callback->deliver(x);
} }
ctx.erase(i); ctx.erase(i);
} }
......
// clang-format off
// DO NOT EDIT: this file is auto-generated by caf-generate-enum-strings.
// Run the target update-enum-strings if this file is out of sync.
#include "caf/io/network/operation.hpp"
#include <string>
namespace caf {
namespace io {
namespace network {
std::string to_string(operation x) {
switch(x) {
default:
return "???";
case operation::read:
return "read";
case operation::write:
return "write";
case operation::propagate_error:
return "propagate_error";
};
}
} // namespace network
} // namespace io
} // namespace caf
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