Commit ba1f2c35 authored by Dominik Charousset's avatar Dominik Charousset

Merge tag '0.18.6' into release/0.18

parents d19dd46d bfa0f83d
......@@ -6,12 +6,12 @@ RUN yum update -y \
&& yum update -y \
&& yum install -y \
cmake3 \
devtoolset-7 \
devtoolset-7-libasan-devel \
devtoolset-7-libubsan-devel \
devtoolset-8 \
devtoolset-8-libasan-devel \
devtoolset-8-libubsan-devel \
git \
make \
openssl-devel \
&& yum clean all
ENV CXX=/opt/rh/devtoolset-7/root/usr/bin/g++
ENV CXX=/opt/rh/devtoolset-8/root/usr/bin/g++
......@@ -5,10 +5,45 @@ is based on [Keep a Changelog](https://keepachangelog.com).
## [Unreleased]
## [0.18.6]
### Added
- When adding CAF with exceptions enabled (default), the unit test framework now
offers new check macros:
- `CAF_CHECK_NOTHROW(expr)`
- `CAF_CHECK_THROWS_AS(expr, type)`
- `CAF_CHECK_THROWS_WITH(expr, str)`
- `CAF_CHECK_THROWS_WITH_AS(expr, str, type)`
### Fixed
- The DSL function `run_until` miscounted the number of executed events, also
causing `run_once` to report a wrong value. Both functions now return the
correct result.
- Using `allow(...).with(...)` in unit tests without a matching message crashed
the program. By adding a missing NULL-check, `allow` is now always safe to
use.
- Passing a response promise to a run-delayed continuation could result in a
heap-use-after-free if the actor terminates before the action runs. The
destructor of the promise now checks for this case.
- Fix OpenSSL 3.0 warnings when building the OpenSSL module by switching to
newer EC-curve API.
- When working with settings, `put`, `put_missing`, `get_if`, etc. now
gracefully handle the `global` category when explicitly using it.
- Messages created from a `message_builder` did not call the destructors for
their values, potentially causing memory leaks (#1321).
### Changed
- Since support of Qt 5 expired, we have ported the Qt examples to version 6.
Hence, building the Qt examples now requires Qt in version 6.
- When compiling CAF with exceptions enabled (default), `REQUIRE*` macros,
`expect` and `disallow` no longer call `abort()`. Instead, they throw an
exception that only stops the current test instead of stopping the entire test
program.
- Reporting of several unit test macros has been improved to show correct line
numbers and provide better diagnostic of test errors.
## [0.18.5] - 2021-07-16
......@@ -751,7 +786,8 @@ is based on [Keep a Changelog](https://keepachangelog.com).
- Setting the log level to `quiet` now properly suppresses any log output.
- Configuring colored terminal output should now print colored output.
[Unreleased]: https://github.com/actor-framework/actor-framework/compare/0.18.5...master
[Unreleased]: https://github.com/actor-framework/actor-framework/compare/0.18.6...master
[0.18.6]: https://github.com/actor-framework/actor-framework/releases/0.18.6
[0.18.5]: https://github.com/actor-framework/actor-framework/releases/0.18.5
[0.18.4]: https://github.com/actor-framework/actor-framework/releases/0.18.4
[0.18.3]: https://github.com/actor-framework/actor-framework/releases/0.18.3
......
......@@ -25,7 +25,6 @@ option(CAF_ENABLE_CURL_EXAMPLES "Build examples with libcurl" OFF)
option(CAF_ENABLE_PROTOBUF_EXAMPLES "Build examples with Google Protobuf" OFF)
option(CAF_ENABLE_QT6_EXAMPLES "Build examples with the Qt6 framework" OFF)
option(CAF_ENABLE_RUNTIME_CHECKS "Build CAF with extra runtime assertions" OFF)
option(CAF_ENABLE_UTILITY_TARGETS "Include targets like consistency-check" OFF)
option(CAF_ENABLE_ACTOR_PROFILER "Enable experimental profiler API" OFF)
# -- CAF options that are on by default ----------------------------------------
......@@ -93,6 +92,8 @@ if(CMAKE_CXX_COMPILER_ID MATCHES "Clang" OR CMAKE_CXX_COMPILER_ID MATCHES "GNU")
target_compile_options(caf_internal INTERFACE
-Wno-missing-field-initializers)
endif()
elseif(MSVC)
target_compile_options(caf_internal INTERFACE /EHsc)
endif()
# -- unit testing setup / caf_add_test_suites function ------------------------
......@@ -204,51 +205,24 @@ if(NOT TARGET uninstall)
COMMAND ${CMAKE_COMMAND} -P ${CMAKE_CURRENT_BINARY_DIR}/uninstall.cmake)
endif()
# -- utility targets -----------------------------------------------------------
if(CAF_ENABLE_UTILITY_TARGETS)
add_executable(caf-generate-enum-strings
EXCLUDE_FROM_ALL
cmake/caf-generate-enum-strings.cpp)
target_link_libraries(caf-generate-enum-strings PRIVATE CAF::internal)
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(caf_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}"
COMMAND caf-generate-enum-strings "${input}" "${output}"
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()
else()
function(caf_add_enum_consistency_check hpp_file cpp_file)
# nop
endfunction()
endif()
# -- utility functions ---------------------------------------------------------
# generates the implementation file for the enum that contains to_string,
# from_string and from_integer
function(caf_add_enum_type target enum_name)
string(REPLACE "." "/" path "${enum_name}")
set(hpp_file "${CMAKE_CURRENT_SOURCE_DIR}/caf/${path}.hpp")
set(cpp_file "${CMAKE_CURRENT_BINARY_DIR}/src/${path}_strings.cpp")
set(gen_file "${PROJECT_SOURCE_DIR}/cmake/caf-generate-enum-strings.cmake")
add_custom_command(OUTPUT "${cpp_file}"
COMMAND ${CMAKE_COMMAND}
"-DINPUT_FILE=${hpp_file}"
"-DOUTPUT_FILE=${cpp_file}"
-P "${gen_file}"
DEPENDS "${hpp_file}" "${gen_file}")
target_sources(${target} PRIVATE "${cpp_file}")
endfunction()
function(caf_export_and_install_lib component)
add_library(CAF::${component} ALIAS libcaf_${component})
string(TOUPPER "CAF_${component}_EXPORT" export_macro_name)
......@@ -328,8 +302,7 @@ endfunction()
# ...
# )
function(caf_add_component name)
set(varargs DEPENDENCIES HEADERS SOURCES TEST_SOURCES TEST_SUITES
ENUM_CONSISTENCY_CHECKS)
set(varargs DEPENDENCIES HEADERS SOURCES TEST_SOURCES TEST_SUITES ENUM_TYPES)
cmake_parse_arguments(CAF_ADD_COMPONENT "" "" "${varargs}" ${ARGN})
if(NOT CAF_ADD_COMPONENT_HEADERS)
message(FATAL_ERROR "Cannot add CAF component without at least one header.")
......@@ -343,9 +316,9 @@ function(caf_add_component name)
endif()
endforeach()
set(pub_lib_target "libcaf_${name}")
set(obj_lib_target "libcaf_${name}_obj")
set(tst_bin_target "caf-${name}-test")
if(CAF_ENABLE_TESTING AND CAF_ADD_COMPONENT_TEST_SOURCES)
set(obj_lib_target "libcaf_${name}_obj")
set(tst_bin_target "caf-${name}-test")
set(targets ${pub_lib_target} ${obj_lib_target} ${tst_bin_target})
add_library(${obj_lib_target} OBJECT
${CAF_ADD_COMPONENT_HEADERS} ${CAF_ADD_COMPONENT_SOURCES})
......@@ -372,6 +345,15 @@ function(caf_add_component name)
set_property(TARGET ${pub_lib_target} PROPERTY POSITION_INDEPENDENT_CODE ON)
endif()
target_link_libraries(${pub_lib_target} ${CAF_ADD_COMPONENT_DEPENDENCIES})
if(CAF_ADD_COMPONENT_ENUM_TYPES)
foreach(enum_name ${CAF_ADD_COMPONENT_ENUM_TYPES})
if(obj_lib_target)
caf_add_enum_type(${obj_lib_target} ${enum_name})
else()
caf_add_enum_type(${pub_lib_target} ${enum_name})
endif()
endforeach()
endif()
foreach(target ${targets})
target_compile_definitions(${target} PRIVATE "libcaf_${name}_EXPORTS")
target_include_directories(${target} PRIVATE
......@@ -384,13 +366,6 @@ function(caf_add_component name)
endif()
endforeach()
caf_export_and_install_lib(${name})
if(CAF_ADD_COMPONENT_ENUM_CONSISTENCY_CHECKS)
foreach(enum_name ${CAF_ADD_COMPONENT_ENUM_CONSISTENCY_CHECKS})
string(REPLACE "." "/" path "${enum_name}")
caf_add_enum_consistency_check("caf/${path}.hpp"
"src/${path}_strings.cpp")
endforeach()
endif()
endfunction()
# -- build all components the user asked for -----------------------------------
......
......@@ -167,26 +167,6 @@ pipeline {
runClangFormat(config)
}
}
stage('Check Consistency') {
agent { label 'unix' }
steps {
deleteDir()
unstash('sources')
dir('sources') {
cmakeBuild([
buildDir: 'build',
installation: 'cmake in search path',
sourceDir: '.',
cmakeArgs: '-DCAF_ENABLE_IO_MODULE:BOOL=OFF ' +
'-DCAF_ENABLE_UTILITY_TARGETS:BOOL=ON',
steps: [[
args: '--target consistency-check',
withCmake: true,
]],
])
}
}
}
stage('Build') {
steps {
buildParallel(config)
......
# seeks the beginning of the enum while also keeping track of namespaces
macro(seek_enum_mode)
if(line MATCHES "^namespace ")
string(REGEX REPLACE "^namespace ([a-zA-Z0-9:_]+).*$" "\\1" tmp "${line}")
list(APPEND namespaces "${tmp}")
elseif(line MATCHES "^enum ")
if(NOT namespaces)
message(FATAL_ERROR "enum found outside of a namespace")
endif()
if(line MATCHES "^enum class ")
set(is_enum_class TRUE)
endif()
string(REGEX REPLACE "^enum (class )?([0-9a-zA-Z_]+).*$" "\\2" enum_name "${line}")
set(mode "read_values")
endif()
endmacro()
# scans the input for enum values
macro(read_values_mode)
if(line MATCHES "^}")
set(mode "generate")
elseif(line MATCHES "^[0-9a-zA-Z_]+")
string(REGEX REPLACE "^([0-9a-zA-Z_]+).*$" "\\1" tmp "${line}")
list(APPEND enum_values "${tmp}")
endif()
endmacro()
macro(write_enum_file)
if(is_enum_class)
set(label_prefix "${enum_name}::")
else()
set(label_prefix "")
endif()
string(REPLACE ";" "::" namespace_str "${namespaces}")
string(REPLACE "::" "/" namespace_path "${namespace_str}")
set(out "")
# File header and includes.
list(APPEND out
"#include \"caf/config.hpp\"\n"
"#include \"caf/string_view.hpp\"\n"
"\n"
"CAF_PUSH_DEPRECATED_WARNING\n"
"\n"
"#include \"${namespace_path}/${enum_name}.hpp\"\n"
"\n"
"#include <string>\n"
"\n"
"namespace ${namespace_str} {\n"
"\n")
# Generate to_string implementation.
list(APPEND out
"std::string to_string(${enum_name} x) {\n"
" switch (x) {\n"
" default:\n"
" return \"???\"\;\n")
foreach(label IN LISTS enum_values)
list(APPEND out
" case ${label_prefix}${label}:\n"
" return \"${namespace_str}::${enum_name}::${label}\"\;\n")
endforeach()
list(APPEND out
" }\n"
"}\n"
"\n")
# Generate from_string implementation.
list(APPEND out
"bool from_string(string_view in, ${enum_name}& out) {\n")
foreach(label IN LISTS enum_values)
list(APPEND out
" if (in == \"${namespace_str}::${enum_name}::${label}\") {\n"
" out = ${label_prefix}${label}\;\n"
" return true\;\n"
" }\n")
endforeach()
list(APPEND out
" return false\;\n"
"}\n"
"\n")
# Generate from_integer implementation.
list(APPEND out
"bool from_integer(std::underlying_type_t<${enum_name}> in,\n"
" ${enum_name}& out) {\n"
" auto result = static_cast<${enum_name}>(in)\;\n"
" switch (result) {\n"
" default:\n"
" return false\;\n")
foreach(label IN LISTS enum_values)
list(APPEND out
" case ${label_prefix}${label}:\n")
endforeach()
list(APPEND out
" out = result\;\n"
" return true\;\n"
" }\n"
"}\n"
"\n")
# File footer.
list(APPEND out
"} // namespace ${namespace_str}\n"
"\n")
file(WRITE "${OUTPUT_FILE}" ${out})
endmacro()
function(main)
set(namespaces "")
set(enum_name "")
set(enum_values "")
set(is_enum_class FALSE)
file(STRINGS "${INPUT_FILE}" lines)
set(mode "seek_enum")
foreach(line IN LISTS lines)
string(REGEX REPLACE "^(.+)(//.*)?" "\\1" line "${line}")
string(STRIP "${line}" line)
if(mode STREQUAL seek_enum)
seek_enum_mode()
elseif(mode STREQUAL read_values)
read_values_mode()
else()
# reached the end
if(NOT enum_name)
message(FATAL_ERROR "No enum found in input file")
endif()
if(NOT enum_values)
message(FATAL_ERROR "No enum values found for ${enum_name}")
endif()
write_enum_file()
return()
endif()
endforeach()
endfunction()
# - check parameters and run ---------------------------------------------------
if(NOT INPUT_FILE)
message(FATAL_ERROR "Variable INPUT_FILE undefined")
endif()
if(NOT OUTPUT_FILE)
message(FATAL_ERROR "Variable OUTPUT_FILE undefined")
endif()
main()
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include <algorithm>
#include <cstring>
#include <fstream>
#include <iostream>
#include <string>
#include <vector>
using std::cerr;
using std::find;
using std::find_if;
using std::string;
using std::vector;
void split(const string& str, const string& separator, vector<string>& dest) {
size_t current, previous = 0;
current = str.find(separator);
while (current != std::string::npos) {
dest.emplace_back(str.substr(previous, current - previous));
previous = current + separator.size();
current = str.find(separator, previous);
}
dest.push_back(str.substr(previous, current - previous));
}
void trim(string& str) {
auto not_space = [](char c) { return isspace(c) == 0; };
str.erase(str.begin(), find_if(str.begin(), str.end(), not_space));
str.erase(find_if(str.rbegin(), str.rend(), not_space).base(), str.end());
}
template <size_t N>
bool starts_with(const string& str, const char (&prefix)[N]) {
return str.compare(0, N - 1, prefix) == 0;
}
template <size_t N>
void drop_prefix(string& str, const char (&prefix)[N]) {
if (str.compare(0, N - 1, prefix) == 0)
str.erase(str.begin(), str.begin() + (N - 1));
}
void keep_alnum(string& str) {
auto not_alnum = [](char c) { return isalnum(c) == 0 && c != '_'; };
str.erase(find_if(str.begin(), str.end(), not_alnum), str.end());
}
int main(int argc, char** argv) {
if (argc != 3) {
cerr << "wrong number of arguments.\n"
<< "usage: " << argv[0] << " input-file output-file\n";
return EXIT_FAILURE;
}
std::ifstream in{argv[1]};
if (!in) {
cerr << "unable to open input file: " << argv[1] << '\n';
return EXIT_FAILURE;
}
vector<string> namespaces;
string enum_name;
string line;
bool is_enum_class = false;
// Locate the beginning of the enum.
for (;;) {
if (!getline(in, line)) {
cerr << "unable to locate enum in file: " << argv[1] << '\n';
return EXIT_FAILURE;
}
trim(line);
if (starts_with(line, "enum ")) {
drop_prefix(line, "enum ");
if (starts_with(line, "class ")) {
is_enum_class = true;
drop_prefix(line, "class ");
}
trim(line);
keep_alnum(line);
enum_name = line;
break;
}
if (starts_with(line, "namespace ")) {
if (line.back() == '{')
line.pop_back();
line.erase(line.begin(), find(line.begin(), line.end(), ' '));
trim(line);
split(line, "::", namespaces);
}
}
// Sanity checking.
if (namespaces.empty()) {
cerr << "enum found outside of a namespace\n";
return EXIT_FAILURE;
}
auto full_namespace = namespaces[0];
for (size_t index = 1; index < namespaces.size(); ++index) {
full_namespace += "::";
full_namespace += namespaces[index];
}
auto full_namespace_prefix = full_namespace + "::";
if (enum_name.empty()) {
cerr << "empty enum name found\n";
return EXIT_FAILURE;
}
std::string case_label_prefix;
if (is_enum_class)
case_label_prefix = enum_name + "::";
// Read until hitting the closing '}'.
std::vector<std::string> enum_values;
for (;;) {
if (!getline(in, line)) {
cerr << "unable to read enum values\n";
return EXIT_FAILURE;
}
trim(line);
if (line.empty())
continue;
if (line[0] == '}')
break;
if (line[0] != '/') {
keep_alnum(line);
enum_values.emplace_back(enum_name + "::" + line);
}
}
// Generate output file.
std::ofstream out{argv[2]};
if (!out) {
cerr << "unable to open output file: " << argv[2] << '\n';
return EXIT_FAILURE;
}
// Print file header.
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 \"caf/config.hpp\"\n"
"#include \"caf/string_view.hpp\"\n\n"
"CAF_PUSH_DEPRECATED_WARNING\n\n"
<< "#include \"" << namespaces[0];
for (size_t i = 1; i < namespaces.size(); ++i)
out << '/' << namespaces[i];
out << '/' << enum_name << ".hpp\"\n\n"
<< "#include <string>\n\n"
<< "namespace " << full_namespace << " {\n";
out << '\n';
// Generate to_string implementation.
out << "std::string to_string(" << enum_name << " x) {\n"
<< " switch(x) {\n"
<< " default:\n"
<< " return \"???\";\n";
for (auto& val : enum_values)
out << " case " << val << ":\n"
<< " return \"" << full_namespace_prefix << val << "\";\n";
out << " };\n"
<< "}\n\n";
// Generate from_string implementation.
out << "bool from_string(string_view in, " << enum_name << "& out) {\n ";
for (auto& val : enum_values)
out << "if (in == \"" << full_namespace_prefix << val << "\") {\n"
<< " out = " << val << ";\n"
<< " return true;\n"
<< " } else ";
out << "{\n"
<< " return false;\n"
<< " }\n"
<< "}\n\n";
// Generate from_integer implementation.
out << "bool from_integer(std::underlying_type_t<" << enum_name << "> in,\n"
<< " " << enum_name << "& out) {\n"
<< " auto result = static_cast<" << enum_name << ">(in);\n"
<< " switch(result) {\n"
<< " default:\n"
<< " return false;\n";
for (auto& val : enum_values)
out << " case " << val << ":\n";
out << " out = result;\n"
<< " return true;\n"
<< " };\n"
<< "}\n\n";
// Done. Print file footer and exit.
out << "} // namespace " << full_namespace << '\n';
out << "\nCAF_POP_WARNINGS\n";
return EXIT_SUCCESS;
}
......@@ -18,6 +18,9 @@ endfunction()
add_core_example(. aout)
add_core_example(. hello_world)
# configuration API
add_core_example(config read-json)
# basic message passing primitives
add_core_example(message_passing calculator)
add_core_example(message_passing cell)
......
// Illustrates how to read custom data types from JSON files.
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/caf_main.hpp"
#include "caf/json_reader.hpp"
#include "caf/type_id.hpp"
#include <fstream>
#include <iostream>
#include <string>
#include <string_view>
constexpr std::string_view example_input = R"([
{
"id": 1,
"name": "John Doe"
},
{
"id": 2,
"name": "Jane Doe",
"email": "jane@doe.com"
}
])";
struct user {
uint32_t id;
std::string name;
std::optional<std::string> email;
};
template <class Inspector>
bool inspect(Inspector& f, user& x) {
return f.object(x).fields(f.field("id", x.id), f.field("name", x.name),
f.field("email", x.email));
}
using user_list = std::vector<user>;
CAF_BEGIN_TYPE_ID_BLOCK(example_app, caf::first_custom_type_id)
CAF_ADD_TYPE_ID(example_app, (user))
CAF_ADD_TYPE_ID(example_app, (user_list))
CAF_END_TYPE_ID_BLOCK(example_app)
int caf_main(caf::actor_system& sys) {
// Get file path from config (positional argument).
auto& cfg = sys.config();
if (cfg.remainder.size() != 1) {
std::cerr << "*** expected one positional argument: path to a JSON file\n";
return EXIT_FAILURE;
}
auto& file_path = cfg.remainder[0];
// Read file into a string.
std::ifstream input{file_path};
if (!input) {
std::cerr << "*** unable to open input file '" << file_path << "'\n";
return EXIT_FAILURE;
}
std::string json{std::istreambuf_iterator<char>{input},
std::istreambuf_iterator<char>{}};
// Parse the JSON-formatted text.
caf::json_reader reader;
if (!reader.load(json)) {
std::cerr << "*** failed to parse JSON input: "
<< to_string(reader.get_error()) << '\n';
return EXIT_FAILURE;
}
// Deserialize our user list from the parsed JSON.
user_list users;
if (!reader.apply(users)) {
std::cerr
<< "*** failed to deserialize the user list: "
<< to_string(reader.get_error())
<< "\n\nNote: expected a JSON list of user objects. For example:\n"
<< example_input << '\n';
return EXIT_FAILURE;
}
// Print the list in "CAF format".
std::cout << "Entries loaded from file:\n";
for (auto& entry : users)
std::cout << "- " << caf::deep_to_string(entry) << '\n';
return EXIT_SUCCESS;
}
CAF_MAIN(caf::id_block::example_app)
......@@ -52,7 +52,7 @@ caf_add_component(
${LIBCAF_CORE_OPTIONAL_DEPENDENCIES}
PRIVATE
CAF::internal
ENUM_CONSISTENCY_CHECKS
ENUM_TYPES
exit_reason
intrusive.inbox_result
intrusive.task_result
......@@ -67,6 +67,7 @@ caf_add_component(
src/abstract_actor.cpp
src/abstract_channel.cpp
src/abstract_group.cpp
src/action.cpp
src/actor.cpp
src/actor_addr.cpp
src/actor_clock.cpp
......@@ -126,7 +127,6 @@ caf_add_component(
src/detail/serialized_size.cpp
src/detail/set_thread_name.cpp
src/detail/shared_spinlock.cpp
src/detail/simple_actor_clock.cpp
src/detail/size_based_credit_controller.cpp
src/detail/stringification_inspector.cpp
src/detail/sync_request_bouncer.cpp
......@@ -135,12 +135,12 @@ caf_add_component(
src/detail/tick_emitter.cpp
src/detail/token_based_credit_controller.cpp
src/detail/type_id_list_builder.cpp
src/disposable.cpp
src/downstream_manager.cpp
src/downstream_manager_base.cpp
src/error.cpp
src/event_based_actor.cpp
src/execution_unit.cpp
src/exit_reason_strings.cpp
src/forwarding_actor_proxy.cpp
src/group.cpp
src/group_manager.cpp
......@@ -148,9 +148,6 @@ caf_add_component(
src/hash/sha1.cpp
src/inbound_path.cpp
src/init_global_meta_objects.cpp
src/intrusive/inbox_result_strings.cpp
src/intrusive/task_result_strings.cpp
src/invoke_message_result_strings.cpp
src/ipv4_address.cpp
src/ipv4_endpoint.cpp
src/ipv4_subnet.cpp
......@@ -164,15 +161,12 @@ caf_add_component(
src/logger.cpp
src/mailbox_element.cpp
src/make_config_option.cpp
src/memory_managed.cpp
src/message.cpp
src/message_builder.cpp
src/message_handler.cpp
src/message_priority_strings.cpp
src/monitorable_actor.cpp
src/node_id.cpp
src/outbound_path.cpp
src/pec_strings.cpp
src/policy/downstream_messages.cpp
src/policy/unprofiled.cpp
src/policy/work_sharing.cpp
......@@ -189,22 +183,20 @@ caf_add_component(
src/scheduler/test_coordinator.cpp
src/scoped_actor.cpp
src/scoped_execution_unit.cpp
src/sec_strings.cpp
src/serializer.cpp
src/settings.cpp
src/skip.cpp
src/stream_aborter.cpp
src/stream_manager.cpp
src/stream_priority_strings.cpp
src/string_algorithms.cpp
src/string_view.cpp
src/telemetry/collector/prometheus.cpp
src/telemetry/importer/process.cpp
src/telemetry/label.cpp
src/telemetry/label_view.cpp
src/telemetry/metric.cpp
src/telemetry/metric_family.cpp
src/telemetry/metric_registry.cpp
src/telemetry/importer/process.cpp
src/term.cpp
src/thread_hook.cpp
src/timestamp.cpp
......@@ -219,6 +211,7 @@ caf_add_component(
test/core-test.cpp
test/nasty.cpp
TEST_SUITES
action
actor_clock
actor_factory
actor_lifetime
......@@ -277,6 +270,7 @@ caf_add_component(
detail.unique_function
detail.unordered_flat_map
dictionary
dsl
dynamic_spawn
error
expected
......
......@@ -52,12 +52,20 @@ public:
/// implementation is required to call `super::destroy()` at the end.
virtual void on_destroy();
void enqueue(strong_actor_ptr sender, message_id mid, message msg,
bool enqueue(strong_actor_ptr sender, message_id mid, message msg,
execution_unit* host) override;
/// Enqueues a new message wrapped in a `mailbox_element` to the actor.
/// This `enqueue` variant allows to define forwarding chains.
virtual void enqueue(mailbox_element_ptr what, execution_unit* host) = 0;
/// @returns `true` if the message has added to the mailbox, `false`
/// otherwise. In the latter case, the actor terminated and the
/// message has been dropped. Once this function returns `false`, it
/// returns `false` for all future invocations.
/// @note The returned value is purely informational and may be used to
/// discard actor handles early. Messages may still get dropped later
/// even if this function returns `true`. In particular when dealing
/// with remote actors.
virtual bool enqueue(mailbox_element_ptr what, execution_unit* host) = 0;
/// Attaches `ptr` to this actor. The actor will call `ptr->detach(...)` on
/// exit, or immediately if it already finished execution.
......@@ -103,15 +111,14 @@ public:
virtual mailbox_element* peek_at_next_mailbox_element();
template <class... Ts>
void eq_impl(message_id mid, strong_actor_ptr sender, execution_unit* ctx,
bool eq_impl(message_id mid, strong_actor_ptr sender, execution_unit* ctx,
Ts&&... xs) {
enqueue(make_mailbox_element(std::move(sender), mid, {},
std::forward<Ts>(xs)...),
ctx);
return enqueue(make_mailbox_element(std::move(sender), mid, {},
std::forward<Ts>(xs)...),
ctx);
}
// flags storing runtime information used by ...
static constexpr int has_timeout_flag = 0x0004; // single_timeout
static constexpr int is_registered_flag = 0x0008; // (several actors)
static constexpr int is_initialized_flag = 0x0010; // event-based actors
static constexpr int is_blocking_flag = 0x0020; // blocking_actor
......
......@@ -23,7 +23,11 @@ public:
virtual ~abstract_channel();
/// Enqueues a new message without forwarding stack to the channel.
virtual void enqueue(strong_actor_ptr sender, message_id mid, message content,
/// @returns `true` if the message has been dispatches successful, `false`
/// otherwise. In the latter case, the channel has been closed and
/// the message has been dropped. Once this function returns `false`,
/// it returns `false` for all future invocations.
virtual bool enqueue(strong_actor_ptr sender, message_id mid, message content,
execution_unit* host = nullptr)
= 0;
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/allowed_unsafe_message_type.hpp"
#include "caf/config.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/disposable.hpp"
#include "caf/make_counted.hpp"
#include "caf/ref_counted.hpp"
#include <atomic>
namespace caf {
/// A functional interface similar to `std::function<void()>` with dispose
/// semantics.
class CAF_CORE_EXPORT action {
public:
// -- member types -----------------------------------------------------------
/// Describes the current state of an `action`.
enum class state {
disposed, /// The action may no longer run.
scheduled, /// The action is scheduled for execution.
invoked, /// The action fired and needs rescheduling before running again.
waiting, /// The action waits for reschedule but didn't run yet.
};
/// Describes the result of an attempted state transition.
enum class transition {
success, /// Transition completed as expected.
disposed, /// No transition since the action has been disposed.
failure, /// No transition since preconditions did not hold.
};
/// Internal interface of `action`.
class impl : public disposable::impl {
public:
virtual transition reschedule() = 0;
virtual transition run() = 0;
virtual state current_state() const noexcept = 0;
friend void intrusive_ptr_add_ref(const impl* ptr) noexcept {
ptr->ref_disposable();
}
friend void intrusive_ptr_release(const impl* ptr) noexcept {
ptr->deref_disposable();
}
};
using impl_ptr = intrusive_ptr<impl>;
// -- constructors, destructors, and assignment operators --------------------
explicit action(impl_ptr ptr) noexcept : pimpl_(std::move(ptr)) {
// nop
}
action() noexcept = default;
action(action&&) noexcept = default;
action(const action&) noexcept = default;
action& operator=(action&&) noexcept = default;
action& operator=(const action&) noexcept = default;
// -- observers --------------------------------------------------------------
[[nodiscard]] bool disposed() const {
return pimpl_->current_state() == state::disposed;
}
[[nodiscard]] bool scheduled() const {
return pimpl_->current_state() == state::scheduled;
}
[[nodiscard]] bool invoked() const {
return pimpl_->current_state() == state::invoked;
}
// -- mutators ---------------------------------------------------------------
/// Tries to transition from `scheduled` to `invoked`, running the body of the
/// internal function object as a side effect on success.
/// @return whether the transition took place.
transition run();
/// Cancel the action if it has not been invoked yet.
void dispose() {
pimpl_->dispose();
}
/// Tries setting the state from `invoked` back to `scheduled`.
/// @return whether the transition took place.
transition reschedule() {
return pimpl_->reschedule();
}
// -- conversion -------------------------------------------------------------
/// Returns a smart pointer to the implementation.
[[nodiscard]] disposable as_disposable() && noexcept {
return disposable{std::move(pimpl_)};
}
/// Returns a smart pointer to the implementation.
[[nodiscard]] disposable as_disposable() const& noexcept {
return disposable{pimpl_};
}
/// Returns a pointer to the implementation.
[[nodiscard]] impl* ptr() const noexcept {
return pimpl_.get();
}
/// Returns a smart pointer to the implementation.
[[nodiscard]] impl_ptr&& as_intrusive_ptr() && noexcept {
return std::move(pimpl_);
}
/// Returns a smart pointer to the implementation.
[[nodiscard]] impl_ptr as_intrusive_ptr() const& noexcept {
return pimpl_;
}
private:
impl_ptr pimpl_;
};
} // namespace caf
namespace caf::detail {
template <class F>
struct default_action_impl : ref_counted, action::impl {
std::atomic<action::state> state_;
F f_;
default_action_impl(F fn, action::state init_state)
: state_(init_state), f_(std::move(fn)) {
// nop
}
void dispose() override {
state_ = action::state::disposed;
}
bool disposed() const noexcept override {
return state_.load() == action::state::disposed;
}
action::state current_state() const noexcept override {
return state_.load();
}
action::transition reschedule() override {
auto st = action::state::invoked;
for (;;) {
if (state_.compare_exchange_strong(st, action::state::scheduled))
return action::transition::success;
switch (st) {
case action::state::invoked:
case action::state::waiting:
break; // Try again.
case action::state::disposed:
return action::transition::disposed;
default:
return action::transition::failure;
}
}
}
action::transition run() override {
auto st = state_.load();
switch (st) {
case action::state::scheduled:
f_();
// No retry. If this action has been disposed while running, we stay
// in the state 'disposed'. We assume that only one thread may try to
// transition from scheduled to invoked, while other threads may only
// dispose the action.
if (state_.compare_exchange_strong(st, action::state::invoked)) {
return action::transition::success;
} else {
CAF_ASSERT(st == action::state::disposed);
return action::transition::disposed;
}
case action::state::disposed:
return action::transition::disposed;
default:
return action::transition::failure;
}
}
void ref_disposable() const noexcept override {
ref();
}
void deref_disposable() const noexcept override {
deref();
}
friend void intrusive_ptr_add_ref(const default_action_impl* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const default_action_impl* ptr) noexcept {
ptr->deref();
}
};
} // namespace caf::detail
namespace caf {
/// Convenience function for creating @ref action objects from a function
/// object.
/// @param f The body for the action.
/// @param init_state either `action::state::scheduled` or
/// `action::state::waiting`.
template <class F>
action make_action(F f, action::state init_state = action::state::scheduled) {
CAF_ASSERT(init_state == action::state::scheduled
|| init_state == action::state::waiting);
using impl_t = detail::default_action_impl<F>;
return action{make_counted<impl_t>(std::move(f), init_state)};
}
} // namespace caf
CAF_ALLOW_UNSAFE_MESSAGE_TYPE(caf::action)
......@@ -26,53 +26,93 @@ public:
/// Time interval.
using duration_type = typename clock_type::duration;
/// Configures how the clock responds to a stalling actor when trying to
/// schedule a periodic action.
enum class stall_policy {
fail, /// Causes the clock to dispose an action send an error to the actor.
skip, /// Causes the clock to skip scheduled runs without emitting errors.
};
// -- constructors, destructors, and assignment operators --------------------
virtual ~actor_clock();
// -- observers --------------------------------------------------------------
// -- scheduling -------------------------------------------------------------
/// Returns the current wall-clock time.
virtual time_point now() const noexcept;
/// Schedules a `timeout_msg` for `self` at time point `t`, overriding any
/// previous receive timeout.
virtual void set_ordinary_timeout(time_point t, abstract_actor* self,
std::string type, uint64_t id)
= 0;
/// Schedules a `timeout_msg` for `self` at time point `t`.
virtual void set_multi_timeout(time_point t, abstract_actor* self,
std::string type, uint64_t id)
/// Schedules an action for execution.
/// @param f The action to schedule.
/// @note The action runs on the thread of the clock worker and thus must
/// complete within a very short time in order to not delay other work.
disposable schedule(action f);
/// Schedules an action for execution at a later time.
/// @param t The local time at which the action should run.
/// @param f The action to schedule.
/// @note The action runs on the thread of the clock worker and thus must
/// complete within a very short time in order to not delay other work.
disposable schedule(time_point t, action f);
/// Schedules an action for periodic execution.
/// @param first_run The local time at which the action should run initially.
/// @param f The action to schedule.
/// @param period The time to wait between two runs. A non-positive period
/// disables periodic execution.
/// @note The action runs on the thread of the clock worker and thus must
/// complete within a very short time in order to not delay other work.
virtual disposable
schedule_periodically(time_point first_run, action f, duration_type period)
= 0;
/// Schedules a `sec::request_timeout` for `self` at time point `t`.
virtual void
set_request_timeout(time_point t, abstract_actor* self, message_id id)
= 0;
/// Cancels a pending receive timeout.
virtual void cancel_ordinary_timeout(abstract_actor* self, std::string type)
= 0;
/// Schedules an action for execution by an actor at a later time.
/// @param t The local time at which the action should get enqueued to the
/// mailbox of the target.
/// @param f The action to schedule.
/// @param target The actor that should run the action.
disposable schedule(time_point t, action f, strong_actor_ptr target);
/// Schedules an action for periodic execution by an actor.
/// @param first_run The local time at which the action should get enqueued to
/// the mailbox of the target for the first time.
/// @param f The action to schedule.
/// @param target The actor that should run the action.
/// @param period The time to wait between two messages to the actor.
/// @param policy The strategy for dealing with a stalling target.
disposable schedule_periodically(time_point first_run, action f,
strong_actor_ptr target,
duration_type period, stall_policy policy);
/// Schedules an action for execution by an actor at a later time.
/// @param target The actor that should run the action.
/// @param f The action to schedule.
/// @param t The local time at which the action should get enqueued to the
/// mailbox of the target.
disposable schedule(time_point t, action f, weak_actor_ptr target);
/// Schedules an action for periodic execution by an actor.
/// @param first_run The local time at which the action should get enqueued to
/// the mailbox of the target for the first time.
/// @param f The action to schedule.
/// @param target The actor that should run the action.
/// @param period The time to wait between two messages to the actor.
/// @param policy The strategy for dealing with a stalling target.
disposable schedule_periodically(time_point first_run, action f,
weak_actor_ptr target, duration_type period,
stall_policy policy);
/// Cancels the pending request timeout for `id`.
virtual void cancel_request_timeout(abstract_actor* self, message_id id) = 0;
/// Cancels all timeouts for `self`.
virtual void cancel_timeouts(abstract_actor* self) = 0;
/// Schedules an arbitrary message to `receiver` for time point `t`.
disposable schedule_message(time_point t, strong_actor_ptr receiver,
mailbox_element_ptr content);
/// Schedules an arbitrary message to `receiver` for time point `t`.
virtual void schedule_message(time_point t, strong_actor_ptr receiver,
mailbox_element_ptr content)
= 0;
disposable schedule_message(time_point t, weak_actor_ptr receiver,
mailbox_element_ptr content);
/// Schedules an arbitrary message to `target` for time point `t`.
virtual void schedule_message(time_point t, group target,
strong_actor_ptr sender, message content)
= 0;
/// Cancels all timeouts and scheduled messages.
virtual void cancel_all() = 0;
disposable schedule_message(time_point t, group target,
strong_actor_ptr sender, message content);
};
} // namespace caf
......@@ -62,9 +62,9 @@ public:
// -- overridden functions ---------------------------------------------------
void enqueue(mailbox_element_ptr ptr, execution_unit* host) override;
bool enqueue(mailbox_element_ptr ptr, execution_unit* host) override;
void enqueue(strong_actor_ptr src, message_id mid, message content,
bool enqueue(strong_actor_ptr src, message_id mid, message content,
execution_unit* eu) override;
void launch(execution_unit* eu, bool lazy, bool hide) override;
......
......@@ -115,10 +115,10 @@ public:
return nid;
}
void enqueue(strong_actor_ptr sender, message_id mid, message content,
bool enqueue(strong_actor_ptr sender, message_id mid, message content,
execution_unit* host);
void enqueue(mailbox_element_ptr what, execution_unit* host);
bool enqueue(mailbox_element_ptr what, execution_unit* host);
/// @endcond
};
......
......@@ -85,7 +85,7 @@ public:
static actor
make(execution_unit* eu, size_t num_workers, const factory& fac, policy pol);
void enqueue(mailbox_element_ptr what, execution_unit* eu) override;
bool enqueue(mailbox_element_ptr what, execution_unit* eu) override;
actor_pool(actor_config& cfg);
......
......@@ -298,10 +298,6 @@ public:
/// Returns the system-wide actor registry.
actor_registry& registry();
/// Returns a string representation for `err`.
[[deprecated("please use to_string() on the error")]] std::string
render(const error& x) const;
/// Returns the system-wide group manager.
group_manager& groups();
......@@ -574,10 +570,9 @@ public:
infer_handle_from_class_t<C> spawn_impl(actor_config& cfg, Ts&&... xs) {
static_assert(is_unbound(Os),
"top-level spawns cannot have monitor or link flag");
// TODO: use `if constexpr` when switching to C++17
if (has_detach_flag(Os) || std::is_base_of<blocking_actor, C>::value)
if constexpr (has_detach_flag(Os) || std::is_base_of_v<blocking_actor, C>)
cfg.flags |= abstract_actor::is_detached_flag;
if (has_hide_flag(Os))
if constexpr (has_hide_flag(Os))
cfg.flags |= abstract_actor::is_hidden_flag;
if (cfg.host == nullptr)
cfg.host = dummy_execution_unit();
......
......@@ -99,9 +99,6 @@ public:
/// provides a config file path on the command line.
error parse(string_list args);
[[deprecated("set the config_file_path member instead")]] error
parse(string_list args, const char* config_file_cstr);
/// Parses the CLI options `{argc, argv}` and `config` as configuration file.
error parse(int argc, char** argv, std::istream& config);
......@@ -110,9 +107,6 @@ public:
/// provides a config file path on the command line.
error parse(int argc, char** argv);
[[deprecated("set the config_file_path member instead")]] error
parse(int argc, char** argv, const char* config_file_cstr);
/// Allows other nodes to spawn actors created by `fun`
/// dynamically by using `name` as identifier.
/// @experimental
......@@ -252,11 +246,6 @@ public:
int (*slave_mode_fun)(actor_system&, const actor_system_config&);
// -- default error rendering functions --------------------------------------
[[deprecated("please use to_string() on the error")]] static std::string
render(const error& err);
// -- config file parsing ----------------------------------------------------
/// Tries to open `filename` and parses its content as CAF config file.
......
......@@ -59,7 +59,6 @@
#include "caf/logger.hpp"
#include "caf/make_config_option.hpp"
#include "caf/may_have_timeout.hpp"
#include "caf/memory_managed.hpp"
#include "caf/message.hpp"
#include "caf/message_builder.hpp"
#include "caf/message_handler.hpp"
......@@ -105,12 +104,6 @@
#include "caf/decorator/sequencer.hpp"
#include "caf/meta/annotation.hpp"
#include "caf/meta/load_callback.hpp"
#include "caf/meta/omittable_if_empty.hpp"
#include "caf/meta/save_callback.hpp"
#include "caf/meta/type_name.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/scheduler/test_coordinator.hpp"
......
......@@ -220,7 +220,7 @@ public:
// -- overridden functions of abstract_actor ---------------------------------
void enqueue(mailbox_element_ptr, execution_unit*) override;
bool enqueue(mailbox_element_ptr, execution_unit*) override;
mailbox_element* peek_at_next_mailbox_element() override;
......
......@@ -264,8 +264,7 @@ private:
// Don't enqueue new data into a closing path.
if (!x.second->closing) {
// Push data from the global buffer to path buffers.
// TODO: replace with `if constexpr` when switching to C++17
if (std::is_same<select_type, detail::select_all>::value) {
if constexpr (std::is_same_v<select_type, detail::select_all>) {
st.buf.insert(st.buf.end(), chunk.begin(), chunk.end());
} else {
for (auto& piece : chunk)
......
......@@ -22,7 +22,7 @@
/// Denotes version of CAF in the format {MAJOR}{MINOR}{PATCH},
/// whereas each number is a two-digit decimal number without
/// leading zeros (e.g. 900 is version 0.9.0).
#define CAF_VERSION 1805
#define CAF_VERSION 1806
/// Defined to the major version number of CAF.
#define CAF_MAJOR_VERSION (CAF_VERSION / 10000)
......@@ -191,8 +191,13 @@
# endif
#elif defined(__FreeBSD__)
# define CAF_BSD
# define CAF_FREE_BSD
#elif defined(__NetBSD__)
# define CAF_BSD
# define CAF_NET_BSD
#elif defined(__OpenBSD__)
# define CAF_BSD
# define CAF_OPEN_BSD
#elif defined(__CYGWIN__)
# define CAF_CYGWIN
#elif defined(WIN32) || defined(_WIN32)
......@@ -201,7 +206,7 @@
# error Platform and/or compiler not supported
#endif
#if defined(CAF_MACOS) || defined(CAF_LINUX) || defined(CAF_BSD) \
|| defined(CAF_CYGWIN)
|| defined(CAF_CYGWIN) || defined(CAF_NET_BSD)
# define CAF_POSIX
#endif
......
......@@ -81,11 +81,6 @@ public:
/// - Store the converted value unless this option is stateless.
error sync(config_value& x) const;
[[deprecated("use sync instead")]] error store(const config_value& x) const;
[[deprecated("use sync instead")]] expected<config_value>
parse(string_view input) const;
/// Returns a human-readable representation of this option's expected type.
string_view type_name() const noexcept;
......@@ -95,10 +90,6 @@ public:
/// Returns whether the category is optional for CLI options.
bool has_flat_cli_name() const noexcept;
/// @private
// TODO: remove with CAF 0.17
optional<config_value> get() const;
private:
string_view buf_slice(size_t from, size_t to) const noexcept;
......
......@@ -38,12 +38,6 @@ public:
config_option_adder&
add_neg(bool& ref, string_view name, string_view description);
[[deprecated("use timespan options instead")]] config_option_adder&
add_us(size_t& ref, string_view name, string_view description);
[[deprecated("use timespan options instead")]] config_option_adder&
add_ms(size_t& ref, string_view name, string_view description);
private:
// -- properties -------------------------------------------------------------
......
......@@ -560,76 +560,29 @@ auto get_or(const config_value& x, Fallback&& fallback) {
// -- SumType-like access ------------------------------------------------------
template <class T>
[[deprecated("use get_as or get_or instead")]] optional<T>
legacy_get_if(const config_value* x) {
if (auto val = get_as<T>(*x))
return {std::move(*val)};
else
return {};
}
template <class T>
template <class T, class = std::enable_if_t<detail::is_config_value_type_v<T>>>
auto get_if(const config_value* x) {
if constexpr (detail::is_config_value_type_v<T>) {
return get_if<T>(x->get_data_ptr());
} else {
return legacy_get_if<T>(x);
}
return get_if<T>(x->get_data_ptr());
}
template <class T>
template <class T, class = std::enable_if_t<detail::is_config_value_type_v<T>>>
auto get_if(config_value* x) {
if constexpr (detail::is_config_value_type_v<T>) {
return get_if<T>(x->get_data_ptr());
} else {
return legacy_get_if<T>(x);
}
}
template <class T>
[[deprecated("use get_as or get_or instead")]] T
legacy_get(const config_value& x) {
auto val = get_as<T>(x);
if (!val)
CAF_RAISE_ERROR("legacy_get: conversion failed");
return std::move(*val);
return get_if<T>(x->get_data_ptr());
}
template <class T>
template <class T, class = std::enable_if_t<detail::is_config_value_type_v<T>>>
decltype(auto) get(const config_value& x) {
if constexpr (detail::is_config_value_type_v<T>) {
return get<T>(x.get_data());
} else {
return legacy_get<T>(x);
}
return get<T>(x.get_data());
}
template <class T>
template <class T, class = std::enable_if_t<detail::is_config_value_type_v<T>>>
decltype(auto) get(config_value& x) {
if constexpr (detail::is_config_value_type_v<T>) {
return get<T>(x.get_data());
} else {
return legacy_get<T>(x);
}
return get<T>(x.get_data());
}
template <class T>
[[deprecated("use get_as or get_or instead")]] bool
legacy_holds_alternative(const config_value& x) {
if (auto val = get_as<T>(x))
return true;
else
return false;
}
template <class T>
template <class T, class = std::enable_if_t<detail::is_config_value_type_v<T>>>
auto holds_alternative(const config_value& x) {
if constexpr (detail::is_config_value_type_v<T>) {
return holds_alternative<T>(x.get_data());
} else {
return legacy_holds_alternative<T>(x);
}
return holds_alternative<T>(x.get_data());
}
// -- comparison operator overloads --------------------------------------------
......
......@@ -28,7 +28,7 @@ public:
// non-system messages are processed and then forwarded;
// system messages are handled and consumed on the spot;
// in either case, the processing is done synchronously
void enqueue(mailbox_element_ptr what, execution_unit* context) override;
bool enqueue(mailbox_element_ptr what, execution_unit* context) override;
message_types_set message_types() const override;
......
......@@ -29,7 +29,7 @@ public:
// non-system messages are processed and then forwarded;
// system messages are handled and consumed on the spot;
// in either case, the processing is done synchronously
void enqueue(mailbox_element_ptr what, execution_unit* context) override;
bool enqueue(mailbox_element_ptr what, execution_unit* context) override;
message_types_set message_types() const override;
......
......@@ -31,10 +31,6 @@ constexpr auto max_batch_delay = timespan{1'000'000};
/// This strategy makes no dynamic adjustment or sampling.
constexpr auto credit_policy = string_view{"size-based"};
[[deprecated("this parameter no longer has any effect")]] //
constexpr auto credit_round_interval
= max_batch_delay;
} // namespace caf::defaults::stream
namespace caf::defaults::stream::size_policy {
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/type_traits.hpp"
#include "caf/disposable.hpp"
namespace caf::detail {
template <class Signature>
struct dispose_on_call_t;
template <class R, class... Ts>
struct dispose_on_call_t<R(Ts...)> {
template <class F>
auto operator()(disposable resource, F f) {
return [resource{std::move(resource)}, f{std::move(f)}](Ts... xs) mutable {
resource.dispose();
return f(xs...);
};
}
};
/// Returns a decorator for the function object `f` that calls
/// `resource.dispose()` before invoking `f`.
template <class F>
auto dispose_on_call(disposable resource, F f) {
using sig = typename get_callable_trait_t<F>::fun_sig;
dispose_on_call_t<sig> factory;
return factory(std::move(resource), std::move(f));
}
} // namespace caf::detail
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/byte_span.hpp"
#include "caf/detail/base64.hpp"
#include "caf/string_view.hpp"
#include <string>
namespace caf::detail {
[[deprecated("use base64::encode instead")]] inline std::string
encode_base64(string_view str) {
return base64::encode(str);
}
[[deprecated("use base64::encode instead")]] inline std::string
encode_base64(const_byte_span bytes) {
return base64::encode(bytes);
}
} // namespace caf::detail
......@@ -40,7 +40,7 @@ public:
void unsubscribe(const actor_control_block* who) override;
// Locally enqueued message, forwarded via worker_.
void enqueue(strong_actor_ptr sender, message_id mid, message content,
bool enqueue(strong_actor_ptr sender, message_id mid, message content,
execution_unit* host) override;
void stop() override;
......
......@@ -54,7 +54,7 @@ public:
~impl() override;
void enqueue(strong_actor_ptr sender, message_id mid, message content,
bool enqueue(strong_actor_ptr sender, message_id mid, message content,
execution_unit* host) override;
bool subscribe(strong_actor_ptr who) override;
......
......@@ -7,6 +7,7 @@
#include <algorithm>
#include <cstddef>
#include "caf/allowed_unsafe_message_type.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp"
......@@ -16,6 +17,7 @@
#include "caf/detail/stringification_inspector.hpp"
#include "caf/inspector_access.hpp"
#include "caf/serializer.hpp"
#include "caf/type_id.hpp"
namespace caf::detail::default_function {
......@@ -56,9 +58,14 @@ bool load(deserializer& source, void* ptr) {
template <class T>
void stringify(std::string& buf, const void* ptr) {
stringification_inspector f{buf};
auto unused = f.apply(*static_cast<const T*>(ptr));
static_cast<void>(unused);
if constexpr (is_allowed_unsafe_message_type_v<T>) {
auto tn = type_name_v<T>;
buf.insert(buf.end(), tn.begin(), tn.end());
} else {
stringification_inspector f{buf};
auto unused = f.apply(*static_cast<const T*>(ptr));
static_cast<void>(unused);
}
}
} // namespace caf::detail::default_function
......
......@@ -17,6 +17,10 @@ namespace caf::detail {
/// later.
class CAF_CORE_EXPORT message_builder_element {
public:
message_builder_element() = default;
message_builder_element(const message_builder_element&) = delete;
message_builder_element& operator=(const message_builder_element&) = delete;
virtual ~message_builder_element();
/// Uses placement new to create a copy of the wrapped value at given memory
......@@ -34,10 +38,24 @@ public:
template <class T>
class message_builder_element_impl : public message_builder_element {
public:
message_builder_element_impl(T value) : value_(std::move(value)) {
template <class Value>
explicit message_builder_element_impl(Value&& value)
: value_(std::forward<Value>(value)) {
// nop
}
message_builder_element_impl() = delete;
message_builder_element_impl(message_builder_element_impl&&) = delete;
message_builder_element_impl(const message_builder_element_impl&) = delete;
message_builder_element_impl& operator=(message_builder_element_impl&&)
= delete;
message_builder_element_impl& operator=(const message_builder_element_impl&)
= delete;
byte* copy_init(byte* storage) const override {
new (storage) T(value_);
return storage + padded_size_v<T>;
......@@ -54,10 +72,4 @@ private:
using message_builder_element_ptr = std::unique_ptr<message_builder_element>;
template <class T>
auto make_message_builder_element(T&& x) {
using impl = message_builder_element_impl<std::decay_t<T>>;
return message_builder_element_ptr{new impl(std::forward<T>(x))};
}
} // namespace caf::detail
......@@ -94,6 +94,11 @@ public:
cv_empty_.notify_all();
}
template <class... Us>
void emplace_back(Us&&... xs) {
push_back(T{std::forward<Us>(xs)...});
}
bool empty() const noexcept {
return rd_pos_ == wr_pos_;
}
......
This diff is collapsed.
......@@ -4,22 +4,45 @@
#pragma once
#include <algorithm>
#include <map>
#include "caf/action.hpp"
#include "caf/actor_clock.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/simple_actor_clock.hpp"
namespace caf::detail {
class CAF_CORE_EXPORT test_actor_clock : public simple_actor_clock {
class CAF_CORE_EXPORT test_actor_clock : public actor_clock {
public:
time_point current_time;
// -- member types -----------------------------------------------------------
struct schedule_entry {
action f;
duration_type period;
};
using schedule_map = std::multimap<time_point, schedule_entry>;
// -- constructors, destructors, and assignment operators --------------------
test_actor_clock();
// -- overrides --------------------------------------------------------------
time_point now() const noexcept override;
disposable schedule_periodically(time_point first_run, action f,
duration_type period) override;
// -- testing DSL API --------------------------------------------------------
/// Returns whether the actor clock has at least one pending timeout.
bool has_pending_timeout() const {
return !schedule_.empty();
auto not_disposed = [](const auto& kvp) {
return !kvp.second.f.disposed();
};
return std::any_of(schedule.begin(), schedule.end(), not_disposed);
}
/// Triggers the next pending timeout regardless of its timestamp. Sets
......@@ -37,6 +60,15 @@ public:
/// Advances the time by `x` and dispatches timeouts and delayed messages.
/// @returns The number of triggered timeouts.
size_t advance_time(duration_type x);
// -- member variables -------------------------------------------------------
time_point current_time;
schedule_map schedule;
private:
bool try_trigger_once();
};
} // namespace caf::detail
......@@ -4,21 +4,20 @@
#pragma once
#include <array>
#include <atomic>
#include <condition_variable>
#include <memory>
#include <mutex>
#include <string>
#include <thread>
#include <vector>
#include "caf/abstract_actor.hpp"
#include "caf/action.hpp"
#include "caf/actor_clock.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/ringbuffer.hpp"
#include "caf/detail/simple_actor_clock.hpp"
#include "caf/fwd.hpp"
namespace caf::detail {
class CAF_CORE_EXPORT thread_safe_actor_clock : public simple_actor_clock {
class CAF_CORE_EXPORT thread_safe_actor_clock : public actor_clock {
public:
// -- constants --------------------------------------------------------------
......@@ -26,45 +25,47 @@ public:
// -- member types -----------------------------------------------------------
using super = simple_actor_clock;
/// Stores actions along with their scheduling period.
struct schedule_entry {
time_point t;
action f;
duration_type period;
};
// -- member functions -------------------------------------------------------
/// @relates schedule_entry
using schedule_entry_ptr = std::unique_ptr<schedule_entry>;
void set_ordinary_timeout(time_point t, abstract_actor* self,
std::string type, uint64_t id) override;
// -- constructors, destructors, and assignment operators --------------------
void set_request_timeout(time_point t, abstract_actor* self,
message_id id) override;
thread_safe_actor_clock();
void set_multi_timeout(time_point t, abstract_actor* self, std::string type,
uint64_t id) override;
// -- overrides --------------------------------------------------------------
void cancel_ordinary_timeout(abstract_actor* self, std::string type) override;
disposable schedule_periodically(time_point first_run, action f,
duration_type period) override;
void cancel_request_timeout(abstract_actor* self, message_id id) override;
// -- thread management ------------------------------------------------------
void cancel_timeouts(abstract_actor* self) override;
void start_dispatch_loop(caf::actor_system& sys);
void schedule_message(time_point t, strong_actor_ptr receiver,
mailbox_element_ptr content) override;
void stop_dispatch_loop();
void schedule_message(time_point t, group target, strong_actor_ptr sender,
message content) override;
void cancel_all() override;
private:
void run();
void run_dispatch_loop();
// -- member variables -------------------------------------------------------
void cancel_dispatch_loop();
/// Communication to the dispatcher thread.
detail::ringbuffer<schedule_entry_ptr, buffer_size> queue_;
private:
void push(event* ptr);
/// Handle to the dispatcher thread.
std::thread dispatcher_;
/// Receives timer events from other threads.
detail::ringbuffer<unique_event_ptr, buffer_size> queue_;
/// Internal data of the dispatcher.
bool running_ = true;
/// Locally caches events for processing.
std::array<unique_event_ptr, buffer_size> events_;
/// Internal data of the dispatcher.
std::vector<schedule_entry_ptr> tbl_;
};
} // namespace caf::detail
......@@ -164,9 +164,7 @@ public:
iterator insert(const_iterator hint, value_type x) {
auto i = find(x.first);
return i == end()
? xs_.insert(gcc48_iterator_workaround(hint), std::move(x))
: i;
return i == end() ? xs_.insert(hint, std::move(x)) : i;
}
template <class InputIterator>
......@@ -188,12 +186,11 @@ public:
// -- removal ----------------------------------------------------------------
iterator erase(const_iterator i) {
return xs_.erase(gcc48_iterator_workaround(i));
return xs_.erase(i);
}
iterator erase(const_iterator first, const_iterator last) {
return xs_.erase(gcc48_iterator_workaround(first),
gcc48_iterator_workaround(last));
return xs_.erase(first, last);
}
size_type erase(const key_type& x) {
......@@ -249,33 +246,6 @@ public:
}
private:
// GCC < 4.9 has a broken STL: vector::erase accepts iterator instead of
// const_iterator.
// TODO: remove when dropping support for GCC 4.8.
template <class Iter>
struct is_valid_erase_iter {
template <class U>
static auto sfinae(U* x)
-> decltype(std::declval<vector_type&>().erase(*x), std::true_type{});
template <class U>
static auto sfinae(...) -> std::false_type;
static constexpr bool value = decltype(sfinae<Iter>(nullptr))::value;
};
template <class I, class E = enable_if_t<!is_valid_erase_iter<I>::value>>
iterator gcc48_iterator_workaround(I i) {
auto j = begin();
std::advance(j, std::distance(cbegin(), i));
return j;
}
template <class I, class E = enable_if_t<is_valid_erase_iter<I>::value>>
const_iterator gcc48_iterator_workaround(I i) {
return i;
}
vector_type xs_;
};
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <vector>
#include "caf/detail/core_export.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/ref_counted.hpp"
namespace caf {
/// Represents a disposable resource.
class CAF_CORE_EXPORT disposable {
public:
/// Internal implementation class of a `disposable`.
class impl {
public:
virtual ~impl();
virtual void dispose() = 0;
virtual bool disposed() const noexcept = 0;
disposable as_disposable() noexcept;
virtual void ref_disposable() const noexcept = 0;
virtual void deref_disposable() const noexcept = 0;
friend void intrusive_ptr_add_ref(const impl* ptr) noexcept {
ptr->ref_disposable();
}
friend void intrusive_ptr_release(const impl* ptr) noexcept {
ptr->deref_disposable();
}
};
explicit disposable(intrusive_ptr<impl> pimpl) noexcept
: pimpl_(std::move(pimpl)) {
// nop
}
disposable() noexcept = default;
disposable(disposable&&) noexcept = default;
disposable(const disposable&) noexcept = default;
disposable& operator=(disposable&&) noexcept = default;
disposable& operator=(const disposable&) noexcept = default;
/// Combines multiple disposables into a single disposable. The new disposable
/// is disposed if all of its elements are disposed. Disposing the composite
/// disposes all elements individually.
static disposable make_composite(std::vector<disposable> entries);
/// Disposes the resource. Calling `dispose()` on a disposed resource is a
/// no-op.
void dispose() {
if (pimpl_) {
pimpl_->dispose();
pimpl_ = nullptr;
}
}
/// Returns whether the resource has been disposed.
[[nodiscard]] bool disposed() const noexcept {
return pimpl_ ? pimpl_->disposed() : true;
}
/// Returns whether this handle still points to a resource.
[[nodiscard]] bool valid() const noexcept {
return pimpl_ != nullptr;
}
/// Returns `valid()`;
explicit operator bool() const noexcept {
return valid();
}
/// Returns `!valid()`;
bool operator!() const noexcept {
return !valid();
}
/// Returns a pointer to the implementation.
[[nodiscard]] impl* ptr() const noexcept {
return pimpl_.get();
}
/// Returns a smart pointer to the implementation.
[[nodiscard]] intrusive_ptr<impl>&& as_intrusive_ptr() && noexcept {
return std::move(pimpl_);
}
/// Returns a smart pointer to the implementation.
[[nodiscard]] intrusive_ptr<impl> as_intrusive_ptr() const& noexcept {
return pimpl_;
}
/// Exchanges the content of this handle with `other`.
void swap(disposable& other) {
pimpl_.swap(other.pimpl_);
}
private:
intrusive_ptr<impl> pimpl_;
};
} // namespace caf
......@@ -55,8 +55,7 @@ void exec_main_load_module(actor_system_config& cfg) {
}
template <class... Ts, class F = void (*)(actor_system&)>
[[deprecated("override config_file_path in the config class instead")]] int
exec_main(F fun, int argc, char** argv, const char* config_file_name) {
int exec_main(F fun, int argc, char** argv) {
using trait = typename detail::get_callable_trait<F>::type;
using arg_types = typename trait::arg_types;
static_assert(detail::tl_size<arg_types>::value == 1
......@@ -77,13 +76,11 @@ exec_main(F fun, int argc, char** argv, const char* config_file_name) {
using helper = exec_main_helper<typename trait::arg_types>;
// Pass CLI options to config.
typename helper::config cfg;
CAF_PUSH_DEPRECATED_WARNING
if (auto err = cfg.parse(argc, argv, config_file_name)) {
if (auto err = cfg.parse(argc, argv)) {
std::cerr << "error while parsing CLI and file options: " << to_string(err)
<< std::endl;
return EXIT_FAILURE;
}
CAF_POP_WARNINGS
// Return immediately if a help text was printed.
if (cfg.cli_helptext_printed)
return EXIT_SUCCESS;
......@@ -108,13 +105,6 @@ exec_main(F fun, int argc, char** argv, const char* config_file_name) {
}
}
template <class... Ts, class F = void (*)(actor_system&)>
int exec_main(F fun, int argc, char** argv) {
CAF_PUSH_DEPRECATED_WARNING
return exec_main<Ts...>(std::move(fun), argc, argv, nullptr);
CAF_POP_WARNINGS
}
} // namespace caf
#define CAF_MAIN(...) \
......
......@@ -24,8 +24,6 @@ namespace caf {
enum class exit_reason : uint8_t {
/// Indicates that an actor finished execution without error.
normal = 0,
/// Indicates that an actor died because of an unhandled exception.
unhandled_exception [[deprecated("superseded by sec::runtime_error")]],
/// Indicates that the exit reason for this actor is unknown, i.e.,
/// the actor has been terminated and no longer exists.
unknown,
......
......@@ -20,7 +20,7 @@ public:
~forwarding_actor_proxy() override;
void enqueue(mailbox_element_ptr what, execution_unit* context) override;
bool enqueue(mailbox_element_ptr what, execution_unit* context) override;
bool add_backlink(abstract_actor* x) override;
......@@ -29,7 +29,7 @@ public:
void kill_proxy(execution_unit* ctx, error rsn) override;
private:
void forward_msg(strong_actor_ptr sender, message_id mid, message msg,
bool forward_msg(strong_actor_ptr sender, message_id mid, message msg,
const forwarding_stack* fwd = nullptr);
mutable detail::shared_spinlock broker_mtx_;
......
......@@ -127,18 +127,12 @@ public:
template <class U>
U& get() {
static constexpr auto i = detail::tl_index_of<param_list, U>::value;
return std::get<i>(nested_);
// TODO: replace with this line when switching to C++14
// return std::get<U>(substreams_);
return std::get<U>(nested_);
}
template <class U>
const U& get() const {
static constexpr auto i = detail::tl_index_of<param_list, U>::value;
return std::get<i>(nested_);
// TODO: replace with this line when switching to C++14
// return std::get<U>(substreams_);
return std::get<U>(nested_);
}
/// Requires a previous call to `add_path` for given slot.
......
......@@ -75,6 +75,7 @@ template <class, class...> class outbound_stream_slot;
class [[nodiscard]] error;
class abstract_actor;
class abstract_group;
class action;
class actor;
class actor_addr;
class actor_clock;
......@@ -96,6 +97,7 @@ class config_option_adder;
class config_option_set;
class config_value;
class deserializer;
class disposable;
class downstream_manager;
class downstream_manager_base;
class event_based_actor;
......@@ -159,7 +161,6 @@ struct none_t;
struct open_stream_msg;
struct prohibit_top_level_spawn_marker;
struct stream_slots;
struct timeout_msg;
struct unit_t;
struct upstream_msg;
struct upstream_msg_ack_batch;
......
......@@ -57,26 +57,6 @@ constexpr auto always_true = always_true_t{};
template <class>
constexpr bool assertion_failed_v = false;
// TODO: remove with CAF 0.19
template <class T, class Inspector, class Obj>
class has_static_apply {
private:
template <class U>
static auto sfinae(Inspector* f, Obj* x) -> decltype(U::apply(*f, *x));
template <class U>
static auto sfinae(...) -> std::false_type;
using sfinae_type = decltype(sfinae<T>(nullptr, nullptr));
public:
static constexpr bool value
= !std::is_same<sfinae_type, std::false_type>::value;
};
template <class T, class Inspector, class Obj>
constexpr bool has_static_apply_v = has_static_apply<T, Inspector, Obj>::value;
// -- loading ------------------------------------------------------------------
// Converts a setter that returns void, error or bool to a sync function object
......@@ -105,17 +85,8 @@ auto bind_setter(Inspector& f, Set& set, ValueType& tmp) {
}
}
// TODO: remove with CAF 0.19
template <class Inspector, class T>
[[deprecated("please provide apply instead of apply_object/apply_value")]] //
std::enable_if_t<!has_static_apply_v<inspector_access<T>, Inspector, T>, bool>
load(Inspector& f, T& x, inspector_access_type::specialization) {
return inspector_access<T>::apply_value(f, x);
}
template <class Inspector, class T>
std::enable_if_t<has_static_apply_v<inspector_access<T>, Inspector, T>, bool>
load(Inspector& f, T& x, inspector_access_type::specialization) {
bool load(Inspector& f, T& x, inspector_access_type::specialization) {
return inspector_access<T>::apply(f, x);
}
......@@ -206,17 +177,8 @@ bool load_field(Inspector& f, string_view field_name, T& x, IsValid& is_valid,
// -- saving -------------------------------------------------------------------
// TODO: remove with CAF 0.19
template <class Inspector, class T>
[[deprecated("please provide apply instead of apply_object/apply_value")]] //
std::enable_if_t<!has_static_apply_v<inspector_access<T>, Inspector, T>, bool>
save(Inspector& f, T& x, inspector_access_type::specialization) {
return inspector_access<T>::apply_value(f, x);
}
template <class Inspector, class T>
std::enable_if_t<has_static_apply_v<inspector_access<T>, Inspector, T>, bool>
save(Inspector& f, T& x, inspector_access_type::specialization) {
bool save(Inspector& f, T& x, inspector_access_type::specialization) {
return inspector_access<T>::apply(f, x);
}
......@@ -394,18 +356,6 @@ struct optional_inspector_access {
return f.object(x).fields(f.field("value", x));
}
template <class Inspector>
[[deprecated("use apply instead")]] static bool
apply_object(Inspector& f, container_type& x) {
return apply(f, x);
}
template <class Inspector>
[[deprecated("use apply instead")]] static bool
apply_value(Inspector& f, container_type& x) {
return apply(f, x);
}
template <class Inspector>
static bool
save_field(Inspector& f, string_view field_name, container_type& x) {
......@@ -478,8 +428,6 @@ struct inspector_access<std::unique_ptr<error::data>>
// -- inspection support for std::byte -----------------------------------------
#ifdef __cpp_lib_byte
template <>
struct inspector_access<std::byte> : inspector_access_base<std::byte> {
template <class Inspector>
......@@ -491,8 +439,6 @@ struct inspector_access<std::byte> : inspector_access_base<std::byte> {
}
};
#endif
// -- inspection support for variant<Ts...> ------------------------------------
template <class T>
......@@ -554,18 +500,6 @@ struct variant_inspector_access {
return f.object(x).fields(f.field("value", x));
}
template <class Inspector>
[[deprecated("use apply instead")]] static bool
apply_object(Inspector& f, T& x) {
return apply(f, x);
}
template <class Inspector>
[[deprecated("use apply instead")]] static bool
apply_value(Inspector& f, T& x) {
return apply(f, x);
}
template <class Inspector>
static bool save_field(Inspector& f, string_view field_name, value_type& x) {
auto g = [&f](auto& y) { return detail::save(f, y); };
......@@ -760,18 +694,6 @@ struct inspector_access<std::chrono::duration<Rep, Period>>
return f.apply(get, set);
}
}
template <class Inspector>
[[deprecated("use apply instead")]] static bool
apply_object(Inspector& f, value_type& x) {
return apply(f, x);
}
template <class Inspector>
[[deprecated("use apply instead")]] static bool
apply_value(Inspector& f, value_type& x) {
return apply(f, x);
}
};
template <class Duration>
......@@ -802,35 +724,6 @@ struct inspector_access<
return f.apply(get, set);
}
}
template <class Inspector>
[[deprecated("use apply instead")]] static bool
apply_object(Inspector& f, value_type& x) {
return apply(f, x);
}
template <class Inspector>
[[deprecated("use apply instead")]] static bool
apply_value(Inspector& f, value_type& x) {
return apply(f, x);
}
};
// -- deprecated API -----------------------------------------------------------
template <class T>
struct default_inspector_access : inspector_access_base<T> {
template <class Inspector>
[[deprecated("call f.apply(x) instead")]] static bool
apply_object(Inspector& f, T& x) {
return f.apply(x);
}
template <class Inspector>
[[deprecated("call f.apply(x) instead")]] static bool
apply_value(Inspector& f, T& x) {
return f.apply(x);
}
};
} // namespace caf
......@@ -21,26 +21,10 @@ constexpr bool is_error_code_enum_v = is_error_code_enum<T>::value;
} // namespace caf
// TODO: the overload with two arguments exists only for backwards compatibility
// with CAF 0.17. Remove with CAF 0.19.
#define CAF_ERROR_CODE_ENUM_1(type_name) \
#define CAF_ERROR_CODE_ENUM(type_name) \
namespace caf { \
template <> \
struct is_error_code_enum<type_name> { \
static constexpr bool value = true; \
}; \
}
#define CAF_ERROR_CODE_ENUM_2(type_name, unused) \
CAF_ERROR_CODE_ENUM_1(type_name)
#ifdef CAF_MSVC
# define CAF_ERROR_CODE_ENUM(...) \
CAF_PP_CAT(CAF_PP_OVERLOAD(CAF_ERROR_CODE_ENUM_, \
__VA_ARGS__)(__VA_ARGS__), \
CAF_PP_EMPTY())
#else
# define CAF_ERROR_CODE_ENUM(...) \
CAF_PP_OVERLOAD(CAF_ERROR_CODE_ENUM_, __VA_ARGS__)(__VA_ARGS__)
#endif
......@@ -21,10 +21,6 @@ namespace caf {
/// for the DSL.
class CAF_CORE_EXPORT load_inspector {
public:
// -- member types -----------------------------------------------------------
using result_type [[deprecated("inspectors always return bool")]] = bool;
// -- constants --------------------------------------------------------------
/// Enables dispatching on the inspector type.
......
......@@ -144,36 +144,6 @@ public:
}
}
// -- deprecated API: remove with CAF 0.19 -----------------------------------
template <class T>
[[deprecated("auto-conversion to underlying type is unsafe, add inspect")]] //
std::enable_if_t<std::is_enum<T>::value, bool>
opaque_value(T& val) {
auto tmp = std::underlying_type_t<T>{0};
if (dref().value(tmp)) {
val = static_cast<T>(tmp);
return true;
} else {
return false;
}
}
template <class T>
[[deprecated("use apply instead")]] bool apply_object(T& x) {
return detail::save(dref(), x);
}
template <class... Ts>
[[deprecated("use apply instead")]] bool apply_objects(Ts&... xs) {
return (apply(xs) && ...);
}
template <class T>
[[deprecated("use apply instead")]] bool apply_value(T& x) {
return detail::save(dref(), x);
}
private:
Subtype* dptr() {
return static_cast<Subtype*>(this);
......
......@@ -110,7 +110,7 @@ public:
/// Requests a new timeout for `mid`.
/// @pre `mid.is_request()`
void request_response_timeout(timespan d, message_id mid);
disposable request_response_timeout(timespan d, message_id mid);
// -- spawn functions --------------------------------------------------------
......@@ -357,17 +357,6 @@ public:
return make_response_promise<response_promise>();
}
template <class... Ts>
[[deprecated("simply return the result from the message handler")]] //
detail::response_promise_t<std::decay_t<Ts>...>
response(Ts&&... xs) {
if (current_element_) {
response_promise::respond_to(this, current_element_,
make_message(std::forward<Ts>(xs)...));
}
return {};
}
const char* name() const override;
/// Serializes the state of this actor to `sink`. This function is
......
......@@ -73,14 +73,4 @@ CAF_CORE_EXPORT config_option
make_negated_config_option(bool& storage, string_view category,
string_view name, string_view description);
// Reads timespans, but stores an integer representing microsecond resolution.
[[deprecated("use timespan options instead")]] CAF_CORE_EXPORT config_option
make_us_resolution_config_option(size_t& storage, string_view category,
string_view name, string_view description);
// Reads timespans, but stores an integer representing millisecond resolution.
[[deprecated("use timespan options instead")]] CAF_CORE_EXPORT config_option
make_ms_resolution_config_option(size_t& storage, string_view category,
string_view name, string_view description);
} // namespace caf
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/core_export.hpp"
namespace caf {
class [[deprecated]] memory_managed;
class CAF_CORE_EXPORT memory_managed {
public:
virtual void request_deletion(bool decremented_rc) const noexcept;
protected:
virtual ~memory_managed();
};
} // namespace caf
......@@ -45,12 +45,12 @@ public:
/// Adds `x` to the elements of the buffer.
template <class T>
message_builder& append(T&& x) {
using namespace detail;
using value_type = strip_and_convert_t<T>;
static_assert(sendable<value_type>);
storage_size_ += padded_size_v<value_type>;
using value_type = detail::strip_and_convert_t<T>;
static_assert(detail::sendable<value_type>);
using wrapper_type = detail::message_builder_element_impl<value_type>;
storage_size_ += detail::padded_size_v<value_type>;
types_.push_back(type_id_v<value_type>);
elements_.emplace_back(make_message_builder_element(std::forward<T>(x)));
elements_.emplace_back(std::make_unique<wrapper_type>(std::forward<T>(x)));
return *this;
}
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <type_traits>
namespace caf::meta {
struct annotation {
constexpr annotation() {
// nop
}
};
template <class T>
struct is_annotation {
static constexpr bool value = std::is_base_of<annotation, T>::value;
};
template <class T>
struct is_annotation<T&> : is_annotation<T> {};
template <class T>
struct is_annotation<const T&> : is_annotation<T> {};
template <class T>
struct is_annotation<T&&> : is_annotation<T> {};
template <class T>
[[deprecated]] constexpr bool is_annotation_v = is_annotation<T>::value;
} // namespace caf::meta
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/meta/annotation.hpp"
namespace caf::meta {
struct hex_formatted_t : annotation {
constexpr hex_formatted_t() {
// nop
}
};
[[deprecated]] constexpr hex_formatted_t hex_formatted() {
return {};
}
} // namespace caf::meta
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <type_traits>
#include <utility>
#include "caf/meta/annotation.hpp"
namespace caf::meta {
template <class F>
struct load_callback_t : annotation {
load_callback_t(F&& f) : fun(f) {
// nop
}
load_callback_t(load_callback_t&&) = default;
load_callback_t(const load_callback_t&) = default;
F fun;
};
template <class T>
struct is_load_callback : std::false_type {};
template <class F>
struct is_load_callback<load_callback_t<F>> : std::true_type {};
template <class F>
constexpr bool is_load_callback_v = is_load_callback<F>::value;
template <class F>
[[deprecated]] load_callback_t<F> load_callback(F fun) {
return {std::move(fun)};
}
} // namespace caf::meta
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/meta/annotation.hpp"
namespace caf::meta {
struct omittable_t : annotation {
constexpr omittable_t() {
// nop
}
};
[[deprecated]] constexpr omittable_t omittable() {
return {};
}
} // namespace caf::meta
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/meta/annotation.hpp"
namespace caf::meta {
struct omittable_if_empty_t : annotation {
constexpr omittable_if_empty_t() {
// nop
}
};
[[deprecated]] constexpr omittable_if_empty_t omittable_if_empty() {
return {};
}
} // namespace caf::meta
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/meta/annotation.hpp"
namespace caf::meta {
struct omittable_if_none_t : annotation {
constexpr omittable_if_none_t() {
// nop
}
};
[[deprecated]] constexpr omittable_if_none_t omittable_if_none() {
return {};
}
} // namespace caf::meta
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <type_traits>
#include "caf/meta/annotation.hpp"
namespace caf::meta {
template <class F>
struct save_callback_t : annotation {
save_callback_t(F&& f) : fun(f) {
// nop
}
save_callback_t(save_callback_t&&) = default;
save_callback_t(const save_callback_t&) = default;
F fun;
};
template <class T>
struct is_save_callback : std::false_type {};
template <class F>
struct is_save_callback<save_callback_t<F>> : std::true_type {};
template <class F>
constexpr bool is_save_callback_v = is_save_callback<F>::value;
template <class F>
[[deprecated]] save_callback_t<F> save_callback(F fun) {
return {std::move(fun)};
}
} // namespace caf::meta
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/meta/annotation.hpp"
namespace caf::meta {
struct type_name_t : annotation {
constexpr type_name_t(const char* cstr) : value(cstr) {
// nop
}
const char* value;
};
[[deprecated]] type_name_t constexpr type_name(const char* cstr) {
return {cstr};
}
} // namespace caf::meta
......@@ -11,6 +11,7 @@
#include "caf/actor.hpp"
#include "caf/check_typed_input.hpp"
#include "caf/detail/profiled_send.hpp"
#include "caf/disposable.hpp"
#include "caf/fwd.hpp"
#include "caf/message.hpp"
#include "caf/message_id.hpp"
......@@ -54,10 +55,11 @@ public:
"receiver does not accept given message");
auto self = static_cast<Subtype*>(this);
auto req_id = self->new_request_id(P);
auto pending_msg = disposable{};
if (dest) {
detail::profiled_send(self, self->ctrl(), dest, req_id, {},
self->context(), std::forward<Ts>(xs)...);
self->request_response_timeout(timeout, req_id);
pending_msg = self->request_response_timeout(timeout, req_id);
} else {
self->eq_impl(req_id.response_id(), self->ctrl(), self->context(),
make_error(sec::invalid_argument));
......@@ -68,7 +70,7 @@ public:
detail::implicit_conversions_t<detail::decay_t<Ts>>...>;
using handle_type
= response_handle<Subtype, policy::single_response<response_type>>;
return handle_type{self, req_id.response_id()};
return handle_type{self, req_id.response_id(), std::move(pending_msg)};
}
/// Sends `{xs...}` to each actor in the range `destinations` as a synchronous
......@@ -106,13 +108,16 @@ public:
auto dptr = static_cast<Subtype*>(this);
std::vector<message_id> ids;
ids.reserve(destinations.size());
std::vector<disposable> pending_msgs;
pending_msgs.reserve(destinations.size());
for (const auto& dest : destinations) {
if (!dest)
continue;
auto req_id = dptr->new_request_id(Prio);
dest->eq_impl(req_id, dptr->ctrl(), dptr->context(),
std::forward<Ts>(xs)...);
dptr->request_response_timeout(timeout, req_id);
pending_msgs.emplace_back(
dptr->request_response_timeout(timeout, req_id));
ids.emplace_back(req_id.response_id());
}
if (ids.empty()) {
......@@ -125,7 +130,8 @@ public:
= response_type_t<typename handle_type::signatures,
detail::implicit_conversions_t<detail::decay_t<Ts>>...>;
using result_type = response_handle<Subtype, MergePolicy<response_type>>;
return result_type{dptr, std::move(ids)};
return result_type{dptr, std::move(ids),
disposable::make_composite(std::move(pending_msgs))};
}
};
......
......@@ -15,60 +15,54 @@
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/detail/typed_actor_util.hpp"
#include "caf/disposable.hpp"
#include "caf/error.hpp"
#include "caf/logger.hpp"
#include "caf/message_id.hpp"
namespace caf::detail {
template <class F, class T>
struct select_all_helper {
std::vector<T> results;
std::shared_ptr<size_t> pending;
F f;
void operator()(T& x) {
CAF_LOG_TRACE(CAF_ARG2("pending", *pending));
if (*pending > 0) {
results.emplace_back(std::move(x));
if (--*pending == 0)
f(std::move(results));
}
}
template <class Fun>
select_all_helper(size_t pending, Fun&& f)
: pending(std::make_shared<size_t>(pending)), f(std::forward<Fun>(f)) {
results.reserve(pending);
}
template <class... Ts>
struct select_all_helper_value_oracle {
using type = std::tuple<Ts...>;
};
auto wrap() {
return [this](T& x) { (*this)(x); };
}
template <class T>
struct select_all_helper_value_oracle<T> {
using type = T;
};
template <class... Ts>
using select_all_helper_value_t =
typename select_all_helper_value_oracle<Ts...>::type;
template <class F, class... Ts>
struct select_all_tuple_helper {
using value_type = std::tuple<Ts...>;
struct select_all_helper {
using value_type = select_all_helper_value_t<Ts...>;
std::vector<value_type> results;
std::shared_ptr<size_t> pending;
disposable timeouts;
F f;
template <class Fun>
select_all_helper(size_t pending, disposable timeouts, Fun&& f)
: pending(std::make_shared<size_t>(pending)),
timeouts(std::move(timeouts)),
f(std::forward<Fun>(f)) {
results.reserve(pending);
}
void operator()(Ts&... xs) {
CAF_LOG_TRACE(CAF_ARG2("pending", *pending));
if (*pending > 0) {
results.emplace_back(std::move(xs)...);
if (--*pending == 0)
if (--*pending == 0) {
timeouts.dispose();
f(std::move(results));
}
}
}
template <class Fun>
select_all_tuple_helper(size_t pending, Fun&& f)
: pending(std::make_shared<size_t>(pending)), f(std::forward<Fun>(f)) {
results.reserve(pending);
}
auto wrap() {
return [this](Ts&... xs) { (*this)(xs...); };
}
......@@ -80,7 +74,7 @@ struct select_select_all_helper;
template <class F, class... Ts>
struct select_select_all_helper<
F, detail::type_list<std::vector<std::tuple<Ts...>>>> {
using type = select_all_tuple_helper<F, Ts...>;
using type = select_all_helper<F, Ts...>;
};
template <class F, class T>
......@@ -113,7 +107,8 @@ public:
= detail::type_checker<response_type,
detail::select_all_helper_t<detail::decay_t<Fun>>>;
explicit select_all(message_id_list ids) : ids_(std::move(ids)) {
explicit select_all(message_id_list ids, disposable pending_timeouts)
: ids_(std::move(ids)), pending_timeouts_(std::move(pending_timeouts)) {
CAF_ASSERT(ids_.size()
<= static_cast<size_t>(std::numeric_limits<int>::max()));
}
......@@ -123,7 +118,7 @@ public:
select_all& operator=(select_all&&) noexcept = default;
template <class Self, class F, class OnError>
void await(Self* self, F&& f, OnError&& g) const {
void await(Self* self, F&& f, OnError&& g) {
CAF_LOG_TRACE(CAF_ARG(ids_));
auto bhvr = make_behavior(std::forward<F>(f), std::forward<OnError>(g));
for (auto id : ids_)
......@@ -131,7 +126,7 @@ public:
}
template <class Self, class F, class OnError>
void then(Self* self, F&& f, OnError&& g) const {
void then(Self* self, F&& f, OnError&& g) {
CAF_LOG_TRACE(CAF_ARG(ids_));
auto bhvr = make_behavior(std::forward<F>(f), std::forward<OnError>(g));
for (auto id : ids_)
......@@ -139,12 +134,13 @@ public:
}
template <class Self, class F, class G>
void receive(Self* self, F&& f, G&& g) const {
void receive(Self* self, F&& f, G&& g) {
CAF_LOG_TRACE(CAF_ARG(ids_));
using helper_type = detail::select_all_helper_t<detail::decay_t<F>>;
helper_type helper{ids_.size(), std::forward<F>(f)};
auto error_handler = [&](error& err) {
helper_type helper{ids_.size(), pending_timeouts_, std::forward<F>(f)};
auto error_handler = [&](error& err) mutable {
if (*helper.pending > 0) {
pending_timeouts_.dispose();
*helper.pending = 0;
helper.results.clear();
g(err);
......@@ -161,17 +157,23 @@ public:
return ids_;
}
disposable pending_timeouts() {
return pending_timeouts_;
}
private:
template <class F, class OnError>
behavior make_behavior(F&& f, OnError&& g) const {
behavior make_behavior(F&& f, OnError&& g) {
using namespace detail;
using helper_type = select_all_helper_t<decay_t<F>>;
helper_type helper{ids_.size(), std::move(f)};
helper_type helper{ids_.size(), pending_timeouts_, std::move(f)};
auto pending = helper.pending;
auto error_handler = [pending{std::move(pending)},
timeouts{pending_timeouts_},
g{std::forward<OnError>(g)}](error& err) mutable {
CAF_LOG_TRACE(CAF_ARG2("pending", *pending));
if (*pending > 0) {
timeouts.dispose();
*pending = 0;
g(err);
}
......@@ -183,6 +185,7 @@ private:
}
message_id_list ids_;
disposable pending_timeouts_;
};
} // namespace caf::policy
......@@ -12,6 +12,7 @@
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/detail/typed_actor_util.hpp"
#include "caf/disposable.hpp"
#include "caf/logger.hpp"
#include "caf/sec.hpp"
......@@ -23,10 +24,14 @@ 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 {
static auto
make(std::shared_ptr<size_t> pending, disposable timeouts, Fun f) {
using std::move;
return [pending{move(pending)}, timeouts{move(timeouts)},
f{move(f)}](Ts... xs) mutable {
CAF_LOG_TRACE(CAF_ARG2("pending", *pending));
if (*pending > 0) {
timeouts.dispose();
f(xs...);
*pending = 0;
}
......@@ -54,13 +59,14 @@ public:
using type_checker
= detail::type_checker<response_type, detail::decay_t<Fun>>;
explicit select_any(message_id_list ids) : ids_(std::move(ids)) {
explicit select_any(message_id_list ids, disposable pending_timeouts)
: ids_(std::move(ids)), pending_timeouts_(std::move(pending_timeouts)) {
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 {
void await(Self* self, F&& f, OnError&& g) {
CAF_LOG_TRACE(CAF_ARG(ids_));
auto bhvr = make_behavior(std::forward<F>(f), std::forward<OnError>(g));
for (auto id : ids_)
......@@ -68,7 +74,7 @@ public:
}
template <class Self, class F, class OnError>
void then(Self* self, F&& f, OnError&& g) const {
void then(Self* self, F&& f, OnError&& g) {
CAF_LOG_TRACE(CAF_ARG(ids_));
auto bhvr = make_behavior(std::forward<F>(f), std::forward<OnError>(g));
for (auto id : ids_)
......@@ -76,11 +82,11 @@ public:
}
template <class Self, class F, class G>
void receive(Self* self, F&& f, G&& g) const {
void receive(Self* self, F&& f, G&& g) {
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 fw = factory::make(pending, pending_timeouts_, 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;
......@@ -94,13 +100,19 @@ public:
return ids_;
}
disposable pending_timeouts() {
return pending_timeouts_;
}
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 {
auto make_error_handler(std::shared_ptr<size_t> p, OnError&& g) {
return [p{std::move(p)}, timeouts{pending_timeouts_},
g{std::forward<OnError>(g)}](error&) mutable {
if (*p == 0) {
// nop
} else if (*p == 1) {
timeouts.dispose();
auto err = make_error(sec::all_requests_failed);
g(err);
} else {
......@@ -110,17 +122,17 @@ private:
}
template <class F, class OnError>
behavior make_behavior(F&& f, OnError&& g) const {
behavior make_behavior(F&& f, OnError&& g) {
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)),
};
auto result_handler = factory::make(pending, pending_timeouts_,
std::forward<F>(f));
return {std::move(result_handler),
make_error_handler(std::move(pending), std::forward<OnError>(g))};
}
message_id_list ids_;
disposable pending_timeouts_;
};
} // namespace caf::policy
......@@ -6,9 +6,11 @@
#include "caf/behavior.hpp"
#include "caf/config.hpp"
#include "caf/detail/dispose_on_call.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/detail/typed_actor_util.hpp"
#include "caf/disposable.hpp"
#include "caf/error.hpp"
#include "caf/message_id.hpp"
......@@ -26,7 +28,8 @@ public:
template <class Fun>
using type_checker = detail::type_checker<response_type, Fun>;
explicit single_response(message_id mid) noexcept : mid_(mid) {
explicit single_response(message_id mid, disposable pending_timeout) noexcept
: mid_(mid), pending_timeout_(std::move(pending_timeout)) {
// nop
}
......@@ -35,30 +38,41 @@ public:
single_response& operator=(single_response&&) noexcept = default;
template <class Self, class F, class OnError>
void await(Self* self, F&& f, OnError&& g) const {
behavior bhvr{std::forward<F>(f), std::forward<OnError>(g)};
void await(Self* self, F&& f, OnError&& g) {
using detail::dispose_on_call;
behavior bhvr{dispose_on_call(pending_timeout_, std::forward<F>(f)),
dispose_on_call(pending_timeout_, std::forward<OnError>(g))};
self->add_awaited_response_handler(mid_, std::move(bhvr));
}
template <class Self, class F, class OnError>
void then(Self* self, F&& f, OnError&& g) const {
behavior bhvr{std::forward<F>(f), std::forward<OnError>(g)};
void then(Self* self, F&& f, OnError&& g) {
using detail::dispose_on_call;
behavior bhvr{dispose_on_call(pending_timeout_, std::forward<F>(f)),
dispose_on_call(pending_timeout_, std::forward<OnError>(g))};
self->add_multiplexed_response_handler(mid_, std::move(bhvr));
}
template <class Self, class F, class OnError>
void receive(Self* self, F&& f, OnError&& g) const {
void receive(Self* self, F&& f, OnError&& g) {
using detail::dispose_on_call;
typename Self::accept_one_cond rc;
self->varargs_receive(rc, mid_, std::forward<F>(f),
std::forward<OnError>(g));
self->varargs_receive(
rc, mid_, dispose_on_call(pending_timeout_, std::forward<F>(f)),
dispose_on_call(pending_timeout_, std::forward<OnError>(g)));
}
message_id id() const noexcept {
return mid_;
}
disposable pending_timeouts() {
return pending_timeout_;
}
private:
message_id mid_;
disposable pending_timeout_;
};
} // namespace caf::policy
......@@ -41,12 +41,13 @@ public:
};
template <class Coordinator>
void enqueue(Coordinator* self, resumable* job) {
bool enqueue(Coordinator* self, resumable* job) {
queue_type l;
l.push_back(job);
std::unique_lock<std::mutex> guard(d(self).lock);
d(self).queue.splice(d(self).queue.end(), l);
d(self).cv.notify_one();
return true;
}
template <class Coordinator>
......
......@@ -55,7 +55,7 @@ public:
// -- non-blocking API -------------------------------------------------------
template <class T = traits, class F, class OnError>
detail::enable_if_t<T::is_non_blocking> await(F f, OnError g) const {
detail::enable_if_t<T::is_non_blocking> await(F f, OnError g) {
static_assert(detail::has_add_awaited_response_handler_v<ActorType>,
"this actor type does not support awaiting responses, "
"try using .then instead");
......@@ -74,13 +74,13 @@ public:
template <class T = traits, class F>
detail::enable_if_t<detail::has_call_error_handler_v<ActorType> //
&& T::is_non_blocking>
await(F f) const {
await(F f) {
auto self = self_;
await(std::move(f), [self](error& err) { self->call_error_handler(err); });
}
template <class T = traits, class F, class OnError>
detail::enable_if_t<T::is_non_blocking> then(F f, OnError g) const {
detail::enable_if_t<T::is_non_blocking> then(F f, OnError g) {
static_assert(detail::has_add_multiplexed_response_handler_v<ActorType>,
"this actor type does not support multiplexed responses, "
"try using .await instead");
......@@ -99,7 +99,7 @@ public:
template <class T = traits, class F>
detail::enable_if_t<detail::has_call_error_handler_v<ActorType> //
&& T::is_non_blocking>
then(F f) const {
then(F f) {
auto self = self_;
then(std::move(f), [self](error& err) { self->call_error_handler(err); });
}
......@@ -152,6 +152,10 @@ public:
return self_;
}
policy_type& policy() noexcept {
return policy_;
}
private:
/// Points to the parent actor.
actor_type* self_;
......
......@@ -48,11 +48,6 @@ public:
response_promise& operator=(const response_promise&) = default;
[[deprecated("use the default constructor instead")]] response_promise(none_t)
: response_promise() {
// nop
}
// -- properties -------------------------------------------------------------
/// Returns whether this response promise replies to an asynchronous message.
......@@ -202,7 +197,7 @@ private:
void delegate_impl(abstract_actor* receiver, message msg);
mutable size_t ref_count = 1;
local_actor* self;
weak_actor_ptr weak_self;
strong_actor_ptr source;
forwarding_stack stages;
message_id id;
......
......@@ -21,10 +21,6 @@ namespace caf {
/// for the DSL.
class CAF_CORE_EXPORT save_inspector {
public:
// -- member types -----------------------------------------------------------
using result_type [[deprecated("inspectors always return bool")]] = bool;
// -- constants --------------------------------------------------------------
/// Enables dispatching on the inspector type.
......
......@@ -97,30 +97,6 @@ public:
return detail::save(dref(), get());
}
// -- deprecated API: remove with CAF 0.19 -----------------------------------
template <class T>
[[deprecated("auto-conversion to underlying type is unsafe, add inspect")]] //
std::enable_if_t<std::is_enum<T>::value, bool>
opaque_value(T val) {
return dref().value(static_cast<std::underlying_type_t<T>>(val));
}
template <class T>
[[deprecated("use apply instead")]] bool apply_object(const T& x) {
return apply(x);
}
template <class... Ts>
[[deprecated("use apply instead")]] bool apply_objects(const Ts&... xs) {
return (apply(xs) && ...);
}
template <class T>
[[deprecated("use apply instead")]] bool apply_value(const T& x) {
return apply(x);
}
private:
Subtype* dptr() {
return static_cast<Subtype*>(this);
......
......@@ -15,6 +15,7 @@
#include <type_traits>
#include <unordered_map>
#include "caf/action.hpp"
#include "caf/actor_traits.hpp"
#include "caf/detail/behavior_stack.hpp"
#include "caf/detail/core_export.hpp"
......@@ -217,7 +218,7 @@ public:
using abstract_actor::enqueue;
void enqueue(mailbox_element_ptr ptr, execution_unit* eu) override;
bool enqueue(mailbox_element_ptr ptr, execution_unit* eu) override;
mailbox_element* peek_at_next_mailbox_element() override;
......@@ -396,20 +397,11 @@ public:
// -- timeout management -----------------------------------------------------
/// Requests a new timeout and returns its ID.
uint64_t set_receive_timeout(actor_clock::time_point x);
/// Requests a new timeout for the current behavior.
void set_receive_timeout();
/// Requests a new timeout for the current behavior and returns its ID.
uint64_t set_receive_timeout();
/// Resets the timeout if `timeout_id` is the active timeout.
void reset_receive_timeout(uint64_t timeout_id);
/// Returns whether `timeout_id` is currently active.
bool is_active_receive_timeout(uint64_t tid) const;
/// Requests a new timeout and returns its ID.
uint64_t set_stream_timeout(actor_clock::time_point x);
/// Requests a new timeout.
void set_stream_timeout(actor_clock::time_point x);
// -- message processing -----------------------------------------------------
......@@ -574,10 +566,46 @@ public:
call_handler(error_handler_, this, err);
}
// -- timeout management -----------------------------------------------------
// -- scheduling actions -----------------------------------------------------
/// Runs `what` asynchronously at some point after `when`.
/// @param when The local time until the actor waits before invoking the
/// action. Due to scheduling delays, there will always be some
/// additional wait time. Passing the current time or a past time
/// immediately schedules the action for execution.
/// @param what The action to invoke after waiting on the timeout.
/// @returns A @ref disposable that allows the actor to cancel the action.
template <class Duration, class F>
disposable run_scheduled(
std::chrono::time_point<std::chrono::system_clock, Duration> when, F what) {
using std::chrono::time_point_cast;
return run_scheduled(time_point_cast<timespan>(when),
make_action(what, action::state::waiting));
}
/// Requests a new timeout and returns its ID.
uint64_t set_timeout(std::string type, actor_clock::time_point x);
/// @copydoc run_scheduled
template <class Duration, class F>
disposable
run_scheduled(std::chrono::time_point<actor_clock::clock_type, Duration> when,
F what) {
using std::chrono::time_point_cast;
using duration_t = actor_clock::duration_type;
return run_scheduled(time_point_cast<duration_t>(when),
make_action(what, action::state::waiting));
}
/// Runs `what` asynchronously after the `delay`.
/// @param delay Minimum amount of time that actor waits before invoking the
/// action. Due to scheduling delays, there will always be some
/// additional wait time.
/// @param what The action to invoke after the delay.
/// @returns A @ref disposable that allows the actor to cancel the action.
template <class Rep, class Period, class F>
disposable run_delayed(std::chrono::duration<Rep, Period> delay, F what) {
using std::chrono::duration_cast;
return run_delayed(duration_cast<timespan>(delay),
make_action(what, action::state::waiting));
}
// -- stream processing ------------------------------------------------------
......@@ -654,8 +682,8 @@ protected:
/// Stores user-defined callbacks for message handling.
detail::behavior_stack bhvr_stack_;
/// Identifies the timeout messages we are currently waiting for.
uint64_t timeout_id_;
/// Allows us to cancel our current in-flight timeout.
disposable pending_timeout_;
/// Stores callbacks for awaited responses.
std::forward_list<pending_response> awaited_responses_;
......@@ -723,6 +751,10 @@ private:
return body();
}
}
disposable run_scheduled(timestamp when, action what);
disposable run_scheduled(actor_clock::time_point when, action what);
disposable run_delayed(timespan delay, action what);
};
} // namespace caf
......@@ -58,16 +58,13 @@ protected:
// Start all workers.
for (auto& w : workers_)
w->start();
// Launch an additional background thread for dispatching timeouts and
// delayed messages.
timer_ = system().launch_thread("caf.clock",
[this] { clock_.run_dispatch_loop(); });
// Run remaining startup code.
clock_.start_dispatch_loop(system());
super::start();
}
void stop() override {
// shutdown workers
// Shutdown workers.
class shutdown_helper : public resumable, public ref_counted {
public:
resumable::resume_result resume(execution_unit* ptr, size_t) override {
......@@ -91,18 +88,18 @@ protected:
std::condition_variable cv;
execution_unit* last_worker;
};
// use a set to keep track of remaining workers
// Use a set to keep track of remaining workers.
shutdown_helper sh;
std::set<worker_type*> alive_workers;
auto num = num_workers();
for (size_t i = 0; i < num; ++i) {
alive_workers.insert(worker_by_id(i));
sh.ref(); // make sure reference count is high enough
sh.ref(); // Make sure reference count is high enough.
}
while (!alive_workers.empty()) {
(*alive_workers.begin())->external_enqueue(&sh);
// since jobs can be stolen, we cannot assume that we have
// actually shut down the worker we've enqueued sh to
// Since jobs can be stolen, we cannot assume that we have actually shut
// down the worker we've enqueued sh to.
{ // lifetime scope of guard
std::unique_lock<std::mutex> guard(sh.mtx);
sh.cv.wait(guard, [&] { return sh.last_worker != nullptr; });
......@@ -110,20 +107,19 @@ protected:
alive_workers.erase(static_cast<worker_type*>(sh.last_worker));
sh.last_worker = nullptr;
}
// shutdown utility actors
// Shutdown utility actors.
stop_actors();
// wait until all workers are done
// Wait until all workers are done.
for (auto& w : workers_) {
w->get_thread().join();
}
// run cleanup code for each resumable
// Run cleanup code for each resumable.
auto f = &abstract_coordinator::cleanup_and_release;
for (auto& w : workers_)
policy_.foreach_resumable(w.get(), f);
policy_.foreach_central_resumable(this, f);
// stop timer thread
clock_.cancel_dispatch_loop();
timer_.join();
// Stop timer thread.
clock_.stop_dispatch_loop();
}
void enqueue(resumable* ptr) override {
......
......@@ -46,14 +46,18 @@ public:
}
/// Peeks into the mailbox of `next_job<scheduled_actor>()`.
template <class T>
const T& peek() {
template <class... Ts>
decltype(auto) peek() {
auto ptr = next_job<scheduled_actor>().mailbox().peek();
CAF_ASSERT(ptr != nullptr);
auto view = make_typed_message_view<T>(ptr->content());
if (!view)
CAF_RAISE_ERROR("Mailbox element does not match T.");
return get<0>(view);
if (auto view = make_const_typed_message_view<Ts...>(ptr->payload)) {
if constexpr (sizeof...(Ts) == 1)
return get<0>(view);
else
return to_tuple(view);
} else {
CAF_RAISE_ERROR("Mailbox element does not match.");
}
}
/// Puts `x` at the front of the queue unless it cannot be found in the queue.
......
......@@ -163,6 +163,9 @@ enum class sec : uint8_t {
broken_promise,
/// Disconnected from a BASP node after reaching the connection timeout.
connection_timeout,
/// Signals that an actor fell behind a periodic action trigger. After raising
/// this error, an @ref actor_clock stops scheduling the action.
action_reschedule_failed,
};
// --(rst-sec-end)--
......
......@@ -81,12 +81,7 @@ expected<T> get_as(const settings& xs, string_view name) {
}
/// @private
CAF_CORE_EXPORT config_value& put_impl(settings& dict,
const std::vector<string_view>& path,
config_value& value);
/// @private
CAF_CORE_EXPORT config_value& put_impl(settings& dict, string_view key,
CAF_CORE_EXPORT config_value& put_impl(settings& dict, string_view name,
config_value& value);
/// Converts `value` to a `config_value` and assigns it to `key`.
......
......@@ -71,8 +71,6 @@ public:
const char*>::value) {
return State::name;
}
} else {
non_static_name_member(state.name);
}
}
return Base::name();
......@@ -84,14 +82,6 @@ public:
/// its reference count drops to zero.
State state;
};
template <class T>
[[deprecated("non-static 'State::name' members have no effect since 0.18")]]
// This function only exists to raise a deprecated warning.
static void
non_static_name_member(const T&) {
// nop
}
};
} // namespace caf
......
......@@ -7,6 +7,7 @@
#include <cstdint>
#include "caf/detail/comparable.hpp"
#include "caf/fwd.hpp"
namespace caf {
......
......@@ -106,22 +106,6 @@ bool inspect(Inspector& f, node_down_msg& x) {
f.field("reason", x.reason));
}
/// Signalizes a timeout event.
/// @note This message is handled implicitly by the runtime system.
struct timeout_msg {
/// Type of the timeout (usually either "receive" or "cycle").
std::string type;
/// Actor-specific timeout ID.
uint64_t timeout_id;
};
/// @relates timeout_msg
template <class Inspector>
bool inspect(Inspector& f, timeout_msg& x) {
return f.object(x).fields(f.field("type", x.type),
f.field("timeout_id", x.timeout_id));
}
/// Demands the receiver to open a new stream from the sender to the receiver.
struct open_stream_msg {
/// Reserved slot on the source.
......
......@@ -78,12 +78,6 @@ constexpr type_id_t first_custom_type_id = 200;
template <class T>
constexpr bool has_type_id_v = detail::is_complete<type_id<T>>;
// TODO: remove with CAF 0.19 (this exists for compatibility with CAF 0.17).
template <class T>
struct has_type_id {
static constexpr bool value = detail::is_complete<type_id<T>>;
};
/// Returns `type_name_v<T>` if available, "anonymous" otherwise.
template <class T>
string_view type_name_or_anonymous() {
......@@ -381,6 +375,7 @@ CAF_BEGIN_TYPE_ID_BLOCK(core_module, 0)
// -- CAF types
CAF_ADD_TYPE_ID(core_module, (caf::action))
CAF_ADD_TYPE_ID(core_module, (caf::actor))
CAF_ADD_TYPE_ID(core_module, (caf::actor_addr))
CAF_ADD_TYPE_ID(core_module, (caf::byte_buffer))
......@@ -413,7 +408,6 @@ CAF_BEGIN_TYPE_ID_BLOCK(core_module, 0)
CAF_ADD_TYPE_ID(core_module, (caf::sec))
CAF_ADD_TYPE_ID(core_module, (caf::stream_slots))
CAF_ADD_TYPE_ID(core_module, (caf::strong_actor_ptr))
CAF_ADD_TYPE_ID(core_module, (caf::timeout_msg))
CAF_ADD_TYPE_ID(core_module, (caf::timespan))
CAF_ADD_TYPE_ID(core_module, (caf::timestamp))
CAF_ADD_TYPE_ID(core_module, (caf::unit_t))
......
......@@ -89,10 +89,6 @@ public:
template <class State>
using stateful_impl = stateful_actor<State, impl>;
template <class State>
using stateful_base [[deprecated("use stateful_impl instead")]]
= stateful_actor<State, base>;
/// Convenience alias for `stateful_impl<State>*`.
template <class State>
using stateful_pointer = stateful_impl<State>*;
......
......@@ -230,7 +230,7 @@ public:
return self_->new_request_id(mp);
}
void request_response_timeout(timespan d, message_id mid) {
disposable request_response_timeout(timespan d, message_id mid) {
return self_->request_response_timeout(d, mid);
}
......@@ -238,13 +238,6 @@ public:
return self_->make_response_promise();
}
template <class... Ts,
class R = typename detail::make_response_promise_helper<
typename std::decay<Ts>::type...>::type>
R response(Ts&&... xs) {
return self_->response(std::forward<Ts>(xs)...);
}
template <class... Ts>
void eq_impl(Ts&&... xs) {
self_->eq_impl(std::forward<Ts>(xs)...);
......
......@@ -39,12 +39,6 @@ public:
typed_response_promise& operator=(const typed_response_promise&) = default;
[[deprecated("use the default constructor instead")]] //
typed_response_promise(none_t x)
: promise_(x) {
// nop
}
// -- properties -------------------------------------------------------------
/// @copydoc response_promise::async
......@@ -77,11 +71,6 @@ public:
return promise_.id();
}
[[deprecated("use the typed_response_promise directly")]]
operator response_promise&() {
return promise_;
}
// -- delivery ---------------------------------------------------------------
/// Satisfies the promise by sending a non-error response message.
......
......@@ -41,9 +41,9 @@ void abstract_actor::on_destroy() {
// nop
}
void abstract_actor::enqueue(strong_actor_ptr sender, message_id mid,
bool abstract_actor::enqueue(strong_actor_ptr sender, message_id mid,
message msg, execution_unit* host) {
enqueue(make_mailbox_element(sender, mid, {}, std::move(msg)), host);
return enqueue(make_mailbox_element(sender, mid, {}, std::move(msg)), host);
}
abstract_actor::abstract_actor(actor_config& cfg)
......
......@@ -2,16 +2,16 @@
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/memory_managed.hpp"
#include "caf/action.hpp"
namespace caf {
#include "caf/logger.hpp"
memory_managed::~memory_managed() {
// nop
}
namespace caf {
void memory_managed::request_deletion(bool) const noexcept {
delete this;
action::transition action::run() {
CAF_LOG_TRACE("");
CAF_ASSERT(pimpl_ != nullptr);
return pimpl_->run();
}
} // namespace caf
......@@ -4,18 +4,219 @@
#include "caf/actor_clock.hpp"
#include "caf/action.hpp"
#include "caf/actor_cast.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/disposable.hpp"
#include "caf/group.hpp"
#include "caf/sec.hpp"
namespace caf {
// -- private utility ----------------------------------------------------------
namespace {
// Unlike the regular action implementation, this one is *not* thread-safe!
// Only the clock itself may access this.
template <class WorkerPtr>
class action_decorator : public ref_counted, public action::impl {
public:
using state = action::state;
using transition = action::transition;
action_decorator(action::impl_ptr decorated, WorkerPtr worker,
actor_clock::stall_policy policy)
: decorated_(std::move(decorated)),
worker_(std::move(worker)),
policy_(policy) {
CAF_ASSERT(decorated_ != nullptr);
CAF_ASSERT(worker_ != nullptr);
}
void dispose() override {
if (decorated_) {
decorated_->dispose();
decorated_ = nullptr;
}
if (worker_)
worker_ = nullptr;
}
bool disposed() const noexcept override {
return decorated_ ? decorated_->disposed() : true;
}
void ref_disposable() const noexcept override {
ref();
}
void deref_disposable() const noexcept override {
deref();
}
transition reschedule() override {
// Always succeeds since we implicitly reschedule in do_run.
return transition::success;
}
transition run() override {
CAF_ASSERT(decorated_ != nullptr);
CAF_ASSERT(worker_ != nullptr);
if constexpr (std::is_same_v<WorkerPtr, weak_actor_ptr>) {
if (auto ptr = actor_cast<strong_actor_ptr>(worker_)) {
return do_run(ptr);
} else {
dispose();
return transition::disposed;
}
} else {
return do_run(worker_);
}
}
state current_state() const noexcept override {
return decorated_ ? decorated_->current_state() : action::state::disposed;
}
friend void intrusive_ptr_add_ref(const action_decorator* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const action_decorator* ptr) noexcept {
ptr->deref();
}
private:
transition do_run(strong_actor_ptr& ptr) {
switch (decorated_->reschedule()) {
case transition::disposed:
decorated_ = nullptr;
worker_ = nullptr;
return transition::disposed;
case transition::success:
if (ptr->enqueue(nullptr, make_message_id(),
make_message(action{decorated_}), nullptr)) {
return transition::success;
} else {
dispose();
return transition::disposed;
}
default:
if (policy_ == actor_clock::stall_policy::fail) {
ptr->enqueue(nullptr, make_message_id(),
make_message(make_error(sec::action_reschedule_failed)),
nullptr);
dispose();
return transition::failure;
} else {
return transition::success;
}
}
}
action::impl_ptr decorated_;
WorkerPtr worker_;
actor_clock::stall_policy policy_;
};
template <class WorkerPtr>
action decorate(action f, WorkerPtr worker, actor_clock::stall_policy policy) {
CAF_ASSERT(f.ptr() != nullptr);
using impl_t = action_decorator<WorkerPtr>;
auto ptr = make_counted<impl_t>(std::move(f).as_intrusive_ptr(),
std::move(worker), policy);
return action{std::move(ptr)};
}
} // namespace
// -- constructors, destructors, and assignment operators ----------------------
actor_clock::~actor_clock() {
// nop
}
// -- observers ----------------------------------------------------------------
// -- scheduling ---------------------------------------------------------------
actor_clock::time_point actor_clock::now() const noexcept {
return clock_type::now();
}
disposable actor_clock::schedule(action f) {
return schedule_periodically(time_point{duration_type{0}}, std::move(f),
duration_type{0});
}
disposable actor_clock::schedule(time_point t, action f) {
return schedule_periodically(t, std::move(f), duration_type{0});
}
disposable actor_clock::schedule(time_point t, action f,
strong_actor_ptr worker) {
return schedule_periodically(t, std::move(f), std::move(worker),
duration_type{0}, stall_policy::skip);
}
disposable actor_clock::schedule_periodically(time_point first_run, action f,
strong_actor_ptr worker,
duration_type period,
stall_policy policy) {
auto res = f.as_disposable();
auto g = decorate(std::move(f), std::move(worker), policy);
schedule_periodically(first_run, std::move(g), period);
return res;
}
disposable actor_clock::schedule(time_point t, action f,
weak_actor_ptr worker) {
return schedule_periodically(t, std::move(f), std::move(worker),
duration_type{0}, stall_policy::skip);
}
disposable actor_clock::schedule_periodically(time_point first_run, action f,
weak_actor_ptr worker,
duration_type period,
stall_policy policy) {
auto res = f.as_disposable();
auto g = decorate(std::move(f), std::move(worker), policy);
schedule_periodically(first_run, std::move(g), period);
return res;
}
disposable actor_clock::schedule_message(time_point t,
strong_actor_ptr receiver,
mailbox_element_ptr content) {
auto f = make_action(
[rptr{std::move(receiver)}, cptr{std::move(content)}]() mutable {
rptr->enqueue(std::move(cptr), nullptr);
});
schedule(t, f);
return std::move(f).as_disposable();
}
disposable actor_clock::schedule_message(time_point t, weak_actor_ptr receiver,
mailbox_element_ptr content) {
auto f = make_action(
[rptr{std::move(receiver)}, cptr{std::move(content)}]() mutable {
if (auto ptr = actor_cast<strong_actor_ptr>(rptr))
ptr->enqueue(std::move(cptr), nullptr);
});
schedule(t, f);
return std::move(f).as_disposable();
}
disposable actor_clock::schedule_message(time_point t, group target,
strong_actor_ptr sender,
message content) {
auto f = make_action([=]() mutable {
if (auto dst = target->get())
dst->enqueue(std::move(sender), make_message_id(), std::move(content),
nullptr);
});
schedule(t, f);
return std::move(f).as_disposable();
}
} // namespace caf
......@@ -24,17 +24,21 @@ void actor_companion::on_exit(on_exit_handler handler) {
on_exit_ = std::move(handler);
}
void actor_companion::enqueue(mailbox_element_ptr ptr, execution_unit*) {
bool actor_companion::enqueue(mailbox_element_ptr ptr, execution_unit*) {
CAF_ASSERT(ptr);
shared_lock<lock_type> guard(lock_);
if (on_enqueue_)
if (on_enqueue_) {
on_enqueue_(std::move(ptr));
return true;
} else {
return false;
}
}
void actor_companion::enqueue(strong_actor_ptr src, message_id mid,
bool actor_companion::enqueue(strong_actor_ptr src, message_id mid,
message content, execution_unit* eu) {
auto ptr = make_mailbox_element(std::move(src), mid, {}, std::move(content));
enqueue(std::move(ptr), eu);
return enqueue(std::move(ptr), eu);
}
void actor_companion::launch(execution_unit*, bool, bool hide) {
......
......@@ -17,14 +17,14 @@ actor_addr actor_control_block::address() {
return {this, true};
}
void actor_control_block::enqueue(strong_actor_ptr sender, message_id mid,
bool actor_control_block::enqueue(strong_actor_ptr sender, message_id mid,
message content, execution_unit* host) {
get()->enqueue(std::move(sender), mid, std::move(content), host);
return get()->enqueue(std::move(sender), mid, std::move(content), host);
}
void actor_control_block::enqueue(mailbox_element_ptr what,
bool actor_control_block::enqueue(mailbox_element_ptr what,
execution_unit* host) {
get()->enqueue(std::move(what), host);
return get()->enqueue(std::move(what), host);
}
bool intrusive_ptr_upgrade_weak(actor_control_block* x) {
......
......@@ -101,11 +101,12 @@ actor actor_pool::make(execution_unit* eu, size_t num_workers,
return res;
}
void actor_pool::enqueue(mailbox_element_ptr what, execution_unit* eu) {
bool actor_pool::enqueue(mailbox_element_ptr what, execution_unit* eu) {
upgrade_lock<detail::shared_spinlock> guard{workers_mtx_};
if (filter(guard, what->sender, what->mid, what->payload, eu))
return;
return false;
policy_(home_system(), guard, workers_, what, eu);
return true;
}
actor_pool::actor_pool(actor_config& cfg)
......
......@@ -419,10 +419,6 @@ actor_registry& actor_system::registry() {
return registry_;
}
std::string actor_system::render(const error& x) const {
return to_string(x);
}
group_manager& actor_system::groups() {
return groups_;
}
......
......@@ -183,23 +183,6 @@ error actor_system_config::parse(int argc, char** argv) {
return parse(std::move(args));
}
error actor_system_config::parse(int argc, char** argv,
const char* config_file_cstr) {
if (config_file_cstr == nullptr) {
return parse(argc, argv);
} else {
string_list args;
if (argc > 0) {
program_name = argv[0];
if (argc > 1)
args.assign(argv + 1, argv + argc);
}
CAF_PUSH_DEPRECATED_WARNING
return parse(std::move(args), config_file_cstr);
CAF_POP_WARNINGS
}
}
error actor_system_config::parse(int argc, char** argv, std::istream& conf) {
string_list args;
if (argc > 0) {
......@@ -376,25 +359,6 @@ error actor_system_config::parse(string_list args) {
}
}
error actor_system_config::parse(string_list args,
const char* config_file_cstr) {
if (config_file_cstr == nullptr) {
return parse(std::move(args));
} else if (auto&& [err, path] = extract_config_file_path(args); !err) {
std::ifstream conf;
if (!path.empty()) {
conf.open(path);
} else {
conf.open(config_file_cstr);
if (conf.is_open())
set("global.config-file", config_file_cstr);
}
return parse(std::move(args), conf);
} else {
return err;
}
}
actor_system_config& actor_system_config::add_actor_factory(std::string name,
actor_factory fun) {
actor_factories.emplace(std::move(name), std::move(fun));
......@@ -420,10 +384,6 @@ actor_system_config& actor_system_config::set_impl(string_view name,
return *this;
}
std::string actor_system_config::render(const error& x) {
return to_string(x);
}
expected<settings>
actor_system_config::parse_config_file(const char* filename) {
config_option_set dummy;
......
......@@ -50,7 +50,7 @@ blocking_actor::~blocking_actor() {
// avoid weak-vtables warning
}
void blocking_actor::enqueue(mailbox_element_ptr ptr, execution_unit*) {
bool blocking_actor::enqueue(mailbox_element_ptr ptr, execution_unit*) {
CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(getf(is_blocking_flag));
CAF_LOG_TRACE(CAF_ARG(*ptr));
......@@ -72,8 +72,10 @@ void blocking_actor::enqueue(mailbox_element_ptr ptr, execution_unit*) {
detail::sync_request_bouncer srb{exit_reason()};
srb(src, mid);
}
return false;
} else {
CAF_LOG_ACCEPT_EVENT(false);
return true;
}
}
......
......@@ -118,11 +118,6 @@ error config_option::sync(config_value& x) const {
return meta_->sync(value_, x);
}
error config_option::store(const config_value& x) const {
auto cpy = x;
return sync(cpy);
}
string_view config_option::type_name() const noexcept {
return meta_->type_name;
}
......@@ -135,20 +130,6 @@ bool config_option::has_flat_cli_name() const noexcept {
return buf_[0] == '?' || category() == "global";
}
expected<config_value> config_option::parse(string_view input) const {
config_value val{input};
if (auto err = sync(val))
return {std::move(err)};
else
return {std::move(val)};
}
optional<config_value> config_option::get() const {
if (value_ != nullptr && meta_->get != nullptr)
return meta_->get(value_);
return none;
}
string_view config_option::buf_slice(size_t from, size_t to) const noexcept {
CAF_ASSERT(from <= to);
return {buf_.get() + from, to - from};
......
......@@ -24,18 +24,6 @@ config_option_adder& config_option_adder::add_neg(bool& ref, string_view name,
name, description));
}
config_option_adder& config_option_adder::add_us(size_t& ref, string_view name,
string_view description) {
return add_impl(make_us_resolution_config_option(ref, category_,
name, description));
}
config_option_adder& config_option_adder::add_ms(size_t& ref, string_view name,
string_view description) {
return add_impl(make_ms_resolution_config_option(ref, category_,
name, description));
}
config_option_adder& config_option_adder::add_impl(config_option&& opt) {
xs_.add(std::move(opt));
return *this;
......
......@@ -31,13 +31,13 @@ sequencer::sequencer(strong_actor_ptr f, strong_actor_ptr g,
}
}
void sequencer::enqueue(mailbox_element_ptr what, execution_unit* context) {
bool sequencer::enqueue(mailbox_element_ptr what, execution_unit* context) {
auto down_msg_handler = [&](down_msg& dm) {
// quit if either `f` or `g` are no longer available
cleanup(std::move(dm.reason), context);
};
if (handle_system_message(*what, context, false, down_msg_handler))
return;
return true;
strong_actor_ptr f;
strong_actor_ptr g;
error err;
......@@ -49,13 +49,13 @@ void sequencer::enqueue(mailbox_element_ptr what, execution_unit* context) {
if (!f) {
// f and g are invalid only after the sequencer terminated
bounce(what, err);
return;
return false;
}
// process and forward the non-system message;
// store `f` as the next stage in the forwarding chain
what->stages.push_back(std::move(f));
// forward modified message to `g`
g->enqueue(std::move(what), context);
return g->enqueue(std::move(what), context);
}
sequencer::message_types_set sequencer::message_types() const {
......
......@@ -45,6 +45,32 @@ std::string get_root_uuid() {
} // namespace detail
} // namespace caf
#elif defined(CAF_IOS) || defined(CAF_ANDROID) || defined(CAF_NET_BSD)
// Return a randomly-generated UUID on mobile devices or NetBSD (requires root
// access to get UUID from disk).
# include <random>
namespace caf {
namespace detail {
std::string get_root_uuid() {
std::random_device rd;
std::uniform_int_distribution<int> dist(0, 15);
std::string uuid = uuid_format;
for (auto& c : uuid) {
if (c != '-') {
auto n = dist(rd);
c = static_cast<char>((n < 10) ? n + '0' : (n - 10) + 'A');
}
}
return uuid;
}
} // namespace detail
} // namespace caf
#elif defined(CAF_LINUX) || defined(CAF_BSD) || defined(CAF_CYGWIN)
# include <algorithm>
......@@ -192,29 +218,4 @@ std::string get_root_uuid() {
} // namespace detail
} // namespace caf
#elif defined(CAF_IOS) || defined(CAF_ANDROID)
// return a randomly-generated UUID on mobile devices
# include <random>
namespace caf {
namespace detail {
std::string get_root_uuid() {
std::random_device rd;
std::uniform_int_distribution<int> dist(0, 15);
std::string uuid = uuid_format;
for (auto& c : uuid) {
if (c != '-') {
auto n = dist(rd);
c = static_cast<char>((n < 10) ? n + '0' : (n - 10) + 'A');
}
}
return uuid;
}
} // namespace detail
} // namespace caf
#endif // CAF_WINDOWS
......@@ -89,7 +89,7 @@ void group_tunnel::unsubscribe(const actor_control_block* who) {
});
}
void group_tunnel::enqueue(strong_actor_ptr sender, message_id mid,
bool group_tunnel::enqueue(strong_actor_ptr sender, message_id mid,
message content, execution_unit* host) {
CAF_LOG_TRACE(CAF_ARG(sender) << CAF_ARG(content));
std::unique_lock<std::mutex> guard{mtx_};
......@@ -100,6 +100,7 @@ void group_tunnel::enqueue(strong_actor_ptr sender, message_id mid,
auto wrapped = make_message(sys_atom_v, forward_atom_v, std::move(content));
cached_messages_.emplace_back(std::move(sender), mid, std::move(wrapped));
}
return true;
}
void group_tunnel::stop() {
......
......@@ -66,11 +66,12 @@ local_group_module::impl::~impl() {
// nop
}
void local_group_module::impl::enqueue(strong_actor_ptr sender, message_id mid,
bool local_group_module::impl::enqueue(strong_actor_ptr sender, message_id mid,
message content, execution_unit* host) {
std::unique_lock<std::mutex> guard{mtx_};
for (auto subscriber : subscribers_)
subscriber->enqueue(sender, mid, content, host);
return true;
}
bool local_group_module::impl::subscribe(strong_actor_ptr who) {
......
......@@ -6,7 +6,11 @@
#include "caf/config.hpp"
#if defined(CAF_LINUX) || defined(CAF_MACOS)
#if defined(CAF_LINUX) || defined(CAF_MACOS) || defined(CAF_BSD)
# define CAF_HAS_CXX_ABI
#endif
#ifdef CAF_HAS_CXX_ABI
# include <cxxabi.h>
# include <sys/types.h>
# include <unistd.h>
......@@ -43,7 +47,7 @@ void prettify_type_name(std::string& class_name) {
}
void prettify_type_name(std::string& class_name, const char* c_class_name) {
#if defined(CAF_LINUX) || defined(CAF_MACOS)
#ifdef CAF_HAS_CXX_ABI
int stat = 0;
std::unique_ptr<char, decltype(free)*> real_class_name{nullptr, free};
auto tmp = abi::__cxa_demangle(c_class_name, nullptr, nullptr, &stat);
......
......@@ -12,7 +12,7 @@
#if defined(CAF_LINUX)
# include <sys/prctl.h>
#elif defined(CAF_BSD)
#elif defined(CAF_BSD) && !defined(CAF_NET_BSD)
# include <pthread_np.h>
#endif // defined(...)
......@@ -32,6 +32,8 @@ void set_thread_name(const char* name) {
pthread_setname_np(name);
# elif defined(CAF_LINUX)
prctl(PR_SET_NAME, name, 0, 0, 0);
# elif defined(CAF_NET_BSD)
pthread_setname_np(pthread_self(), name, NULL);
# elif defined(CAF_BSD)
pthread_set_name_np(pthread_self(), name);
# endif // defined(...)
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/detail/simple_actor_clock.hpp"
#include "caf/actor_cast.hpp"
#include "caf/sec.hpp"
#include "caf/system_messages.hpp"
namespace caf::detail {
simple_actor_clock::event::~event() {
// nop
}
void simple_actor_clock::set_ordinary_timeout(time_point t,
abstract_actor* self,
std::string type, uint64_t id) {
new_schedule_entry<ordinary_timeout>(t, self->ctrl(), type, id);
}
void simple_actor_clock::set_multi_timeout(time_point t, abstract_actor* self,
std::string type, uint64_t id) {
new_schedule_entry<multi_timeout>(t, self->ctrl(), type, id);
}
void simple_actor_clock::set_request_timeout(time_point t, abstract_actor* self,
message_id id) {
new_schedule_entry<request_timeout>(t, self->ctrl(), id);
}
void simple_actor_clock::cancel_ordinary_timeout(abstract_actor* self,
std::string type) {
ordinary_timeout_cancellation tmp{self->id(), std::move(type)};
handle(tmp);
}
void simple_actor_clock::cancel_request_timeout(abstract_actor* self,
message_id id) {
request_timeout_cancellation tmp{self->id(), id};
handle(tmp);
}
void simple_actor_clock::cancel_timeouts(abstract_actor* self) {
auto range = actor_lookup_.equal_range(self->id());
if (range.first == range.second)
return;
for (auto i = range.first; i != range.second; ++i)
schedule_.erase(i->second);
actor_lookup_.erase(range.first, range.second);
}
void simple_actor_clock::schedule_message(time_point t,
strong_actor_ptr receiver,
mailbox_element_ptr content) {
new_schedule_entry<actor_msg>(t, std::move(receiver), std::move(content));
}
void simple_actor_clock::schedule_message(time_point t, group target,
strong_actor_ptr sender,
message content) {
new_schedule_entry<group_msg>(t, std::move(target), std::move(sender),
std::move(content));
}
void simple_actor_clock::cancel_all() {
actor_lookup_.clear();
schedule_.clear();
}
void simple_actor_clock::ship(delayed_event& x) {
switch (x.subtype) {
case ordinary_timeout_type: {
auto& dref = static_cast<ordinary_timeout&>(x);
auto& self = dref.self;
self->get()->eq_impl(make_message_id(), self, nullptr,
timeout_msg{dref.type, dref.id});
break;
}
case multi_timeout_type: {
auto& dref = static_cast<multi_timeout&>(x);
auto& self = dref.self;
self->get()->eq_impl(make_message_id(), self, nullptr,
timeout_msg{dref.type, dref.id});
break;
}
case request_timeout_type: {
auto& dref = static_cast<request_timeout&>(x);
auto& self = dref.self;
self->get()->eq_impl(dref.id, self, nullptr, sec::request_timeout);
break;
}
case actor_msg_type: {
auto& dref = static_cast<actor_msg&>(x);
dref.receiver->enqueue(std::move(dref.content), nullptr);
break;
}
case group_msg_type: {
auto& dref = static_cast<group_msg&>(x);
auto dst = dref.target->get();
if (dst)
dst->enqueue(std::move(dref.sender), make_message_id(),
std::move(dref.content), nullptr);
break;
}
default:
break;
}
}
void simple_actor_clock::handle(const ordinary_timeout_cancellation& x) {
auto pred = [&](const actor_lookup_map::value_type& kvp) {
auto& y = *kvp.second->second;
return y.subtype == ordinary_timeout_type
&& x.type == static_cast<const ordinary_timeout&>(y).type;
};
cancel(x.aid, pred);
}
void simple_actor_clock::handle(const multi_timeout_cancellation& x) {
auto pred = [&](const actor_lookup_map::value_type& kvp) {
auto& y = *kvp.second->second;
if (y.subtype != multi_timeout_type)
return false;
auto& dref = static_cast<const multi_timeout&>(y);
return x.type == dref.type && x.id == dref.id;
};
cancel(x.aid, pred);
}
void simple_actor_clock::handle(const request_timeout_cancellation& x) {
auto pred = [&](const actor_lookup_map::value_type& kvp) {
auto& y = *kvp.second->second;
return y.subtype == request_timeout_type
&& x.id == static_cast<const request_timeout&>(y).id;
};
cancel(x.aid, pred);
}
void simple_actor_clock::handle(const timeouts_cancellation& x) {
auto range = actor_lookup_.equal_range(x.aid);
if (range.first == range.second)
return;
for (auto i = range.first; i != range.second; ++i)
schedule_.erase(i->second);
actor_lookup_.erase(range.first, range.second);
}
size_t simple_actor_clock::trigger_expired_timeouts() {
size_t result = 0;
auto t = now();
auto i = schedule_.begin();
auto e = schedule_.end();
while (i != e && i->first <= t) {
auto ptr = std::move(i->second);
auto backlink = ptr->backlink;
if (backlink != actor_lookup_.end())
actor_lookup_.erase(backlink);
i = schedule_.erase(i);
ship(*ptr);
++result;
}
return result;
}
void simple_actor_clock::add_schedule_entry(
time_point t, std::unique_ptr<ordinary_timeout> x) {
auto aid = x->self->id();
auto type = x->type;
auto pred = [&](const actor_lookup_map::value_type& kvp) {
auto& y = *kvp.second->second;
return y.subtype == ordinary_timeout_type
&& static_cast<const ordinary_timeout&>(y).type == type;
};
auto i = lookup(aid, pred);
if (i != actor_lookup_.end()) {
schedule_.erase(i->second);
i->second = schedule_.emplace(t, std::move(x));
} else {
auto j = schedule_.emplace(t, std::move(x));
i = actor_lookup_.emplace(aid, j);
}
i->second->second->backlink = i;
}
} // namespace caf::detail
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment