Commit b403c75e authored by Dominik Charousset's avatar Dominik Charousset

Add compiler flag for opting into type ID checking

When enabled, the flag adds a static assertion in `make_message` that
makes sure each type has a message ID assigned to it.
parent f102a1b1
......@@ -514,10 +514,22 @@ endif()
# all projects need the headers of the core components
include_directories("${CAF_INCLUDE_DIRS}")
# -- unit tests setup ----------------------------------------------------------
################################################################################
# add targets #
################################################################################
if(NOT CAF_NO_UNIT_TESTS)
enable_testing()
function(caf_add_test_suites target)
foreach(suiteName ${ARGN})
string(REPLACE "." "/" suitePath ${suiteName})
target_sources(${target} PRIVATE
"${CMAKE_CURRENT_SOURCE_DIR}/test/${suitePath}.cpp")
add_test(NAME ${suiteName}
COMMAND ${target} -r300 -n -v5 -s"^${suiteName}$")
endforeach()
endfunction()
endif()
# -- add targets ---------------------------------------------------------------
macro(add_caf_lib name)
string(TOUPPER ${name} upper_name)
......@@ -653,48 +665,6 @@ add_optional_caf_binaries(examples)
# build tools if not being told otherwise
add_optional_caf_binaries(tools)
################################################################################
# unit tests setup #
################################################################################
if(NOT CAF_NO_UNIT_TESTS)
# setup unit test binary
add_executable(caf-test
libcaf_test/src/caf-test.cpp
libcaf_test/caf/test/unit_test.hpp
libcaf_test/caf/test/unit_test_impl.hpp
${CAF_ALL_UNIT_TESTS})
target_link_libraries(caf-test
${CAF_EXTRA_LDFLAGS}
${CAF_LIBRARIES}
${PTHREAD_LIBRARIES})
add_custom_target(all_unit_tests)
add_dependencies(caf-test all_unit_tests)
# enumerate all test suites.
foreach(test ${CAF_ALL_UNIT_TESTS})
file(STRINGS ${test} contents)
foreach(line ${contents})
if ("${line}" MATCHES "CAF_SUITE (.*)")
string(REGEX REPLACE ".* CAF_SUITE (.*)" "\\1" suite ${line})
list(APPEND suites ${suite})
endif()
endforeach()
endforeach()
list(REMOVE_DUPLICATES suites)
# creates one CMake test per test suite.
macro (make_test suite)
string(REPLACE " " "_" test_name ${suite})
set(caf_test ${EXECUTABLE_OUTPUT_PATH}/caf-test)
add_test(${test_name} ${caf_test} -r 300 -n -v 5 -s "${suite}" ${ARGN})
endmacro ()
list(LENGTH suites num_suites)
message(STATUS "Found ${num_suites} test suites")
foreach(suite ${suites})
make_test("${suite}")
endforeach ()
endif()
# -- Setup for building manual and API documentation ---------------------------
if(NOT caf_is_subproject)
......
......@@ -7,6 +7,7 @@ defaultReleaseBuildFlags = [
'CAF_ENABLE_RUNTIME_CHECKS:BOOL=yes',
'CAF_NO_OPENCL:BOOL=yes',
'CAF_INSTALL_UNIT_TESTS:BOOL=yes',
'CAF_ENABLE_TYPE_ID_CHECKS:BOOL=yes',
]
// Default CMake flags for debug builds.
......@@ -14,6 +15,7 @@ defaultDebugBuildFlags = defaultReleaseBuildFlags + [
'CAF_ENABLE_ADDRESS_SANITIZER:BOOL=yes',
'CAF_LOG_LEVEL:STRING=TRACE',
'CAF_ENABLE_ACTOR_PROFILER:BOOL=yes',
'CAF_ENABLE_TYPE_ID_CHECKS:BOOL=yes',
]
// Configures the behavior of our stages.
......@@ -25,7 +27,6 @@ config = [
'build',
'style',
'tests',
'coverage',
],
// Our build matrix. Keys are the operating system labels and values are build configurations.
buildMatrix: [
......@@ -36,7 +37,6 @@ config = [
['Linux', [
builds: ['debug'],
tools: ['gcc8'],
extraSteps: ['coverageReport'],
]],
['Linux', [
builds: ['debug'],
......@@ -92,20 +92,6 @@ config = [
],
],
],
// Configures what binary the coverage report uses and what paths to exclude.
coverage: [
binary: 'build/bin/caf-test',
relativeExcludePaths: [
'examples',
'tools',
'libcaf_test',
'libcaf_core/test',
'libcaf_io/test',
'libcaf_openssl/test',
'libcaf_opencl',
'libcaf_core/caf/scheduler/profiled_coordinator.hpp',
],
],
]
// Declarative pipeline for triggering all stages.
......
......@@ -32,3 +32,5 @@
#cmakedefine CAF_NO_EXCEPTIONS
#cmakedefine CAF_ENABLE_ACTOR_PROFILER
#cmakedefine CAF_ENABLE_TYPE_ID_CHECKS
......@@ -213,3 +213,126 @@ install(DIRECTORY "${CMAKE_CURRENT_SOURCE_DIR}/caf"
DESTINATION ${CMAKE_INSTALL_INCLUDEDIR}
FILES_MATCHING PATTERN "*.hpp"
)
# -- build unit tests ----------------------------------------------------------
if(CAF_NO_UNIT_TESTS)
return()
endif()
add_executable(caf-core-test test/core-test.cpp)
target_include_directories(caf-core-test PRIVATE
"${CMAKE_CURRENT_SOURCE_DIR}/test")
if (CAF_BUILD_STATIC_ONLY OR CAF_BUILD_STATIC)
target_link_libraries(caf-core-test libcaf_core_static ${CAF_EXTRA_LDFLAGS})
else()
target_link_libraries(caf-core-test libcaf_core_shared ${CAF_EXTRA_LDFLAGS})
endif()
caf_add_test_suites(caf-core-test
actor_clock
actor_factory
actor_lifetime
actor_pool
actor_profiler
actor_registry
actor_system_config
actor_termination
aout
behavior
blocking_actor
broadcast_downstream_manager
byte
composition
config_option
config_option_set
config_value
config_value_adaptor
constructor_attach
continuous_streaming
cow_tuple
decorator.sequencer
deep_to_string
detached_actors
detail.bounds_checker
detail.ini_consumer
detail.limited_vector
detail.parse
detail.parser.read_bool
detail.parser.read_floating_point
detail.parser.read_ini
detail.parser.read_number
detail.parser.read_number_or_timespan
detail.parser.read_signed_integer
detail.parser.read_string
detail.parser.read_timespan
detail.parser.read_unsigned_integer
detail.ringbuffer
detail.ripemd_160
detail.serialized_size
detail.tick_emitter
detail.unique_function
detail.unordered_flat_map
dictionary
dynamic_spawn
expected
function_view
fused_downstream_manager
handles
inspector
intrusive.drr_cached_queue
intrusive.drr_queue
intrusive.fifo_inbox
intrusive.lifo_inbox
intrusive.task_queue
intrusive.wdrr_dynamic_multiplexed_queue
intrusive.wdrr_fixed_multiplexed_queue
intrusive_ptr
ipv4_address
ipv4_endpoint
ipv4_subnet
ipv6_address
ipv6_endpoint
ipv6_subnet
local_group
logger
mailbox_element
make_config_value_field
message
message_id
message_lifetime
metaprogramming
mixin.requester
mixin.sender
mock_streaming_classes
native_streaming_classes
node_id
optional
or_else
pipeline_streaming
policy.categorized
request_timeout
result
selective_streaming
serialization
settings
simple_timeout
span
stateful_actor
string_algorithms
string_view
sum_type
thread_hook
typed_behavior
typed_response_promise
typed_spawn
unit
uri
variant
)
if(NOT CAF_NO_EXCEPTIONS)
caf_add_test_suites(caf-core-test custom_exception_handler)
endif()
......@@ -168,12 +168,7 @@ public:
/// Adds message type `T` with runtime type info `name`.
template <class T>
actor_system_config& add_message_type(std::string name) {
static_assert(std::is_empty<T>::value
|| std::is_same<T, actor>::value // silence add_actor_type err
|| is_typed_actor<T>::value
|| (std::is_default_constructible<T>::value
&& std::is_copy_constructible<T>::value),
"T must provide default and copy constructors");
assert_message_type_eligible<T>();
std::string stream_name = "stream<";
stream_name += name;
stream_name += ">";
......@@ -421,6 +416,17 @@ protected:
config_option_set custom_options_;
private:
template <class T>
static void assert_message_type_eligible() {
static_assert(
std::is_empty<T>::value
|| std::is_same<T, actor>::value // silence add_actor_type err
|| is_typed_actor<T>::value
|| (std::is_default_constructible<T>::value
&& std::is_copy_constructible<T>::value),
"T must provide default and copy constructors");
}
template <class T>
void add_message_type_impl(std::string name) {
type_names_by_rtti.emplace(std::type_index(typeid(T)), name);
......@@ -435,7 +441,9 @@ private:
template <long I, long... Is>
void add_message_types(detail::int_list<I, Is...>) {
add_message_type<typename type_by_id<I>::type>(type_name_by_id<I>::value);
using type = typename type_by_id<I>::type;
assert_message_type_eligible<type>();
add_message_type_impl<type>(type_name_by_id<I>::value);
add_message_types(detail::int_list<Is...>{});
}
......
......@@ -22,106 +22,144 @@
#include <type_traits>
#include "caf/actor_traits.hpp"
#include "caf/fwd.hpp"
#include "caf/detail/squashed_int.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
namespace caf {
namespace detail {
template <class T,
bool IsDyn = std::is_base_of<dynamically_typed_actor_base, T>::value,
bool IsStat = std::is_base_of<statically_typed_actor_base, T>::value>
struct implicit_actor_conversions {
constexpr int integral_type_flag = 0x01;
constexpr int error_type_flag = 0x02;
constexpr int dynamically_typed_actor_flag = 0x04;
constexpr int statically_typed_actor_flag = 0x08;
template <bool, int>
struct conversion_flag {
static constexpr int value = 0;
};
template <int Value>
struct conversion_flag<true, Value> {
static constexpr int value = Value;
};
template <class T>
struct implicit_conversion_oracle {
static constexpr int value
= conversion_flag<std::is_integral<T>::value, integral_type_flag>::value
| conversion_flag<std::is_convertible<T, error>::value,
error_type_flag>::value
| conversion_flag<std::is_base_of<dynamically_typed_actor_base, T>::value,
dynamically_typed_actor_flag>::value
| conversion_flag<std::is_base_of<statically_typed_actor_base, T>::value,
statically_typed_actor_flag>::value;
};
template <class T, int = implicit_conversion_oracle<T>::value>
struct implicit_conversions;
template <class T>
struct implicit_conversions<T, 0> {
using type = T;
};
template <>
struct implicit_conversions<bool, integral_type_flag> {
using type = bool;
};
template <class T>
struct implicit_conversions<T, integral_type_flag> {
using type = squashed_int_t<T>;
};
template <class T>
struct implicit_actor_conversions<T, true, false> {
struct implicit_conversions<T, dynamically_typed_actor_flag> {
using type = actor;
};
template <class T>
struct implicit_actor_conversions<T, false, true> {
struct implicit_conversions<T, statically_typed_actor_flag> {
using type =
typename detail::tl_apply<
typename T::signatures,
typed_actor
>::type;
typename detail::tl_apply<typename T::signatures, typed_actor>::type;
};
template <>
struct implicit_actor_conversions<actor_control_block, false, false> {
struct implicit_conversions<actor_control_block*, 0> {
using type = strong_actor_ptr;
};
template <class T>
struct implicit_conversions {
using type =
typename std::conditional<
std::is_convertible<T, error>::value,
error,
T
>::type;
struct implicit_conversions<T, error_type_flag> {
using type = error;
};
template <class T>
struct implicit_conversions<T*> : implicit_actor_conversions<T> {};
struct implicit_conversions<T*, 0> {
static constexpr int oracle = implicit_conversion_oracle<T>::value;
static constexpr int is_actor_mask = dynamically_typed_actor_flag
| statically_typed_actor_flag;
static_assert((oracle & is_actor_mask) != 0,
"messages must not contain pointers");
using type = typename implicit_conversions<T, oracle>::type;
};
template <>
struct implicit_conversions<char*> {
struct implicit_conversions<char*, 0> {
using type = std::string;
};
template <size_t N>
struct implicit_conversions<char[N]>
struct implicit_conversions<char[N], 0>
: implicit_conversions<char*> {};
template <>
struct implicit_conversions<const char*>
struct implicit_conversions<const char*, 0>
: implicit_conversions<char*> {};
template <size_t N>
struct implicit_conversions<const char[N]>
struct implicit_conversions<const char[N], 0>
: implicit_conversions<char*> {};
template <>
struct implicit_conversions<char16_t*> {
struct implicit_conversions<char16_t*, 0> {
using type = std::u16string;
};
template <size_t N>
struct implicit_conversions<char16_t[N]>
struct implicit_conversions<char16_t[N], 0>
: implicit_conversions<char16_t*> {};
template <>
struct implicit_conversions<const char16_t*>
struct implicit_conversions<const char16_t*, 0>
: implicit_conversions<char16_t*> {};
template <size_t N>
struct implicit_conversions<const char16_t[N]>
struct implicit_conversions<const char16_t[N], 0>
: implicit_conversions<char16_t*> {};
template <>
struct implicit_conversions<char32_t*> {
struct implicit_conversions<char32_t*, 0> {
using type = std::u16string;
};
template <size_t N>
struct implicit_conversions<char32_t[N]>
struct implicit_conversions<char32_t[N], 0>
: implicit_conversions<char32_t*> {};
template <>
struct implicit_conversions<const char32_t*>
struct implicit_conversions<const char32_t*, 0>
: implicit_conversions<char32_t*> {};
template <size_t N>
struct implicit_conversions<const char32_t[N]>
struct implicit_conversions<const char32_t[N], 0>
: implicit_conversions<char32_t*> {};
template <>
struct implicit_conversions<scoped_actor> {
struct implicit_conversions<scoped_actor, 0> {
using type = actor;
};
......@@ -145,4 +183,3 @@ using strip_and_convert_t = typename strip_and_convert<T>::type;
} // namespace detail
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <cstddef>
#include <type_traits>
namespace caf {
namespace detail {
template <class T, size_t = sizeof(T)>
std::true_type is_complete_impl(T*);
std::false_type is_complete_impl(...);
/// Checks whether T is a complete type. For example, passing a forward
/// declaration or undefined template specialization evaluates to `false`.
template <class T>
struct is_complete
{
static constexpr bool value
= decltype(is_complete_impl(std::declval<T*>()))::value;
};
} // namespace detail
} // namespace caf
......@@ -220,8 +220,9 @@ public:
}
template <class InputIterator>
void insert(iterator pos, InputIterator first, InputIterator last) {
CAF_ASSERT(first <= last);
iterator insert(iterator pos, InputIterator first, InputIterator last) {
if (first == last)
return pos;
auto num_elements = static_cast<size_t>(std::distance(first, last));
if ((size() + num_elements) > MaxSize) {
CAF_RAISE_ERROR("limited_vector::insert: too much elements");
......@@ -238,6 +239,13 @@ public:
// insert new elements
std::copy(first, last, pos);
}
// Iterator to the first element inserted.
return pos + 1;
}
iterator insert(iterator pos, value_type value) {
T tmp[] = {value};
return insert(pos, std::begin(tmp), std::end(tmp));
}
private:
......
......@@ -23,13 +23,11 @@
#include <utility>
#include "caf/atom.hpp"
#include "caf/detail/comparable.hpp"
#include "caf/fwd.hpp"
#include "caf/none.hpp"
#include "caf/meta/omittable_if_empty.hpp"
#include "caf/meta/type_name.hpp"
#include "caf/detail/comparable.hpp"
#include "caf/none.hpp"
namespace caf {
......
......@@ -22,11 +22,13 @@
#include <sstream>
#include <type_traits>
#include "caf/message.hpp"
#include "caf/allowed_unsafe_message_type.hpp"
#include "caf/detail/build_config.hpp"
#include "caf/detail/is_complete.hpp"
#include "caf/detail/tuple_vals.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/message.hpp"
#include "caf/type_id.hpp"
namespace caf {
......@@ -52,13 +54,21 @@ struct unbox_message_element<actor_control_block*, 0> {
using type = strong_actor_ptr;
};
///
/// @private
template <class T>
struct is_serializable_or_whitelisted {
static constexpr bool value = detail::is_serializable<T>::value
|| allowed_unsafe_message_type<T>::value;
};
/// @private
template <class T>
struct has_type_id {
static constexpr bool value = detail::is_complete<type_id<T>>::value
|| is_atom_constant<T>::value
|| allowed_unsafe_message_type<T>::value;
};
/// Returns a new `message` containing the values `(x, xs...)`.
/// @relates message
template <class T, class... Ts>
......@@ -86,6 +96,12 @@ make_message(T&& x, Ts&&... xs) {
"you can whitelist individual types by "
"specializing `caf::allowed_unsafe_message_type<T>` "
"or using the macro CAF_ALLOW_UNSAFE_MESSAGE_TYPE");
#ifdef CAF_ENABLE_TYPE_ID_CHECKS
static_assert(tl_forall<stored_types, has_type_id>::value,
"at least one type has no type ID: please assign type IDs "
"to all of your types via CAF_ADD_TYPE_ID (this check was "
"enabled via CAF_ENABLE_TYPE_ID_CHECKS)");
#endif
using storage = typename tl_apply<stored_types, tuple_vals>::type;
auto ptr = make_counted<storage>(std::forward<T>(x), std::forward<Ts>(xs)...);
return message{detail::message_data::cow_ptr{std::move(ptr)}};
......@@ -118,4 +134,3 @@ message make_message_from_tuple(std::tuple<Ts...> xs) {
}
} // namespace caf
......@@ -183,10 +183,12 @@ CAF_BEGIN_TYPE_ID_BLOCK(core_module, 0)
CAF_ADD_TYPE_ID(core_module, (caf::actor))
CAF_ADD_TYPE_ID(core_module, (caf::actor_addr))
CAF_ADD_TYPE_ID(core_module, (caf::atom_value))
CAF_ADD_TYPE_ID(core_module, (caf::config_value))
CAF_ADD_TYPE_ID(core_module, (caf::dictionary<caf::config_value>) )
CAF_ADD_TYPE_ID(core_module, (caf::down_msg))
CAF_ADD_TYPE_ID(core_module, (caf::downstream_msg))
CAF_ADD_TYPE_ID(core_module, (caf::duration))
CAF_ADD_TYPE_ID(core_module, (caf::error))
CAF_ADD_TYPE_ID(core_module, (caf::exit_msg))
CAF_ADD_TYPE_ID(core_module, (caf::exit_reason))
......
......@@ -16,10 +16,9 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/config.hpp"
#define CAF_SUITE actor_pool
#include "caf/test/unit_test.hpp"
#include "core-test.hpp"
#include "caf/all.hpp"
......@@ -42,13 +41,10 @@ public:
behavior make_behavior() override {
auto nested = exit_handler_;
set_exit_handler([=](scheduled_actor* self, exit_msg& em) {
nested(self, em);
});
set_exit_handler(
[=](scheduled_actor* self, exit_msg& em) { nested(self, em); });
return {
[](int x, int y) {
return x + y;
}
[](int32_t x, int32_t y) { return x + y; },
};
}
};
......@@ -56,17 +52,20 @@ public:
struct fixture {
// allows us to check s_dtors after dtor of actor_system
actor_system_config cfg;
union { actor_system system; };
union { scoped_execution_unit context; };
union {
actor_system system;
};
union {
scoped_execution_unit context;
};
std::function<actor ()> spawn_worker;
std::function<actor()> spawn_worker;
fixture() {
cfg.add_message_types<id_block::core_test>();
new (&system) actor_system(cfg);
new (&context) scoped_execution_unit(&system);
spawn_worker = [&] {
return system.spawn<worker>();
};
spawn_worker = [&] { return system.spawn<worker>(); };
}
~fixture() {
......@@ -91,28 +90,28 @@ CAF_TEST(round_robin_actor_pool) {
actor_pool::round_robin());
self->send(pool, sys_atom::value, put_atom::value, spawn_worker());
std::vector<actor> workers;
for (int i = 0; i < 6; ++i) {
self->request(pool, infinite, i, i).receive(
[&](int res) {
CAF_CHECK_EQUAL(res, i + i);
auto sender = actor_cast<strong_actor_ptr>(self->current_sender());
CAF_REQUIRE(sender);
workers.push_back(actor_cast<actor>(std::move(sender)));
},
handle_err
);
for (int32_t i = 0; i < 6; ++i) {
self->request(pool, infinite, i, i)
.receive(
[&](int32_t res) {
CAF_CHECK_EQUAL(res, i + i);
auto sender = actor_cast<strong_actor_ptr>(self->current_sender());
CAF_REQUIRE(sender);
workers.push_back(actor_cast<actor>(std::move(sender)));
},
handle_err);
}
CAF_CHECK_EQUAL(workers.size(), 6u);
CAF_CHECK(std::unique(workers.begin(), workers.end()) == workers.end());
self->request(pool, infinite, sys_atom::value, get_atom::value).receive(
[&](std::vector<actor>& ws) {
std::sort(workers.begin(), workers.end());
std::sort(ws.begin(), ws.end());
CAF_REQUIRE_EQUAL(workers.size(), ws.size());
CAF_CHECK(std::equal(workers.begin(), workers.end(), ws.begin()));
},
handle_err
);
self->request(pool, infinite, sys_atom::value, get_atom::value)
.receive(
[&](std::vector<actor>& ws) {
std::sort(workers.begin(), workers.end());
std::sort(ws.begin(), ws.end());
CAF_REQUIRE_EQUAL(workers.size(), ws.size());
CAF_CHECK(std::equal(workers.begin(), workers.end(), ws.begin()));
},
handle_err);
CAF_MESSAGE("await last worker");
anon_send_exit(workers.back(), exit_reason::user_shutdown);
self->wait_for(workers.back());
......@@ -122,19 +121,19 @@ CAF_TEST(round_robin_actor_pool) {
bool success = false;
size_t i = 0;
while (!success && ++i <= 10) {
self->request(pool, infinite, sys_atom::value, get_atom::value).receive(
[&](std::vector<actor>& ws) {
success = workers.size() == ws.size();
if (success) {
std::sort(ws.begin(), ws.end());
CAF_CHECK_EQUAL(workers, ws);
} else {
// wait a bit until polling again
std::this_thread::sleep_for(std::chrono::milliseconds(5));
}
},
handle_err
);
self->request(pool, infinite, sys_atom::value, get_atom::value)
.receive(
[&](std::vector<actor>& ws) {
success = workers.size() == ws.size();
if (success) {
std::sort(ws.begin(), ws.end());
CAF_CHECK_EQUAL(workers, ws);
} else {
// wait a bit until polling again
std::this_thread::sleep_for(std::chrono::milliseconds(5));
}
},
handle_err);
}
CAF_REQUIRE_EQUAL(success, true);
CAF_MESSAGE("about to send exit to workers");
......@@ -152,32 +151,23 @@ CAF_TEST(broadcast_actor_pool) {
auto pool = actor_pool::make(&context, 5, spawn5, actor_pool::broadcast());
CAF_CHECK_EQUAL(system.registry().running(), 32u);
self->send(pool, 1, 2);
std::vector<int> results;
int i = 0;
self->receive_for(i, 25)(
[&](int res) {
results.push_back(res);
},
after(std::chrono::milliseconds(250)) >> [] {
CAF_ERROR("didn't receive a result");
}
);
std::vector<int32_t> results;
int32_t i = 0;
self->receive_for(i, 25)([&](int32_t res) { results.push_back(res); },
after(std::chrono::milliseconds(250)) >>
[] { CAF_ERROR("didn't receive a result"); });
CAF_CHECK_EQUAL(results.size(), 25u);
CAF_CHECK(std::all_of(results.begin(), results.end(),
[](int res) { return res == 3; }));
[](int32_t res) { return res == 3; }));
self->send_exit(pool, exit_reason::user_shutdown);
}
CAF_TEST(random_actor_pool) {
scoped_actor self{system};
auto pool = actor_pool::make(&context, 5, spawn_worker, actor_pool::random());
for (int i = 0; i < 5; ++i) {
self->request(pool, std::chrono::milliseconds(250), 1, 2).receive(
[&](int res) {
CAF_CHECK_EQUAL(res, 3);
},
handle_err
);
for (int32_t i = 0; i < 5; ++i) {
self->request(pool, std::chrono::milliseconds(250), 1, 2)
.receive([&](int32_t res) { CAF_CHECK_EQUAL(res, 3); }, handle_err);
}
self->send_exit(pool, exit_reason::user_shutdown);
}
......@@ -186,10 +176,7 @@ CAF_TEST(split_join_actor_pool) {
auto spawn_split_worker = [&] {
return system.spawn<lazy_init>([]() -> behavior {
return {
[](size_t pos, const std::vector<int>& xs) {
return xs[pos];
}
};
[](size_t pos, const std::vector<int32_t>& xs) { return xs[pos]; }};
});
};
auto split_fun = [](std::vector<std::pair<actor, message>>& xs, message& y) {
......@@ -197,27 +184,18 @@ CAF_TEST(split_join_actor_pool) {
xs[i].second = make_message(i) + y;
}
};
auto join_fun = [](int& res, message& msg) {
msg.apply([&](int x) {
res += x;
});
auto join_fun = [](int32_t& res, message& msg) {
msg.apply([&](int32_t x) { res += x; });
};
scoped_actor self{system};
CAF_MESSAGE("create actor pool");
auto pool = actor_pool::make(&context, 5, spawn_split_worker,
actor_pool::split_join<int>(join_fun, split_fun));
self->request(pool, infinite, std::vector<int>{1, 2, 3, 4, 5}).receive(
[&](int res) {
CAF_CHECK_EQUAL(res, 15);
},
handle_err
);
self->request(pool, infinite, std::vector<int>{6, 7, 8, 9, 10}).receive(
[&](int res) {
CAF_CHECK_EQUAL(res, 40);
},
handle_err
);
auto pool
= actor_pool::make(&context, 5, spawn_split_worker,
actor_pool::split_join<int32_t>(join_fun, split_fun));
self->request(pool, infinite, std::vector<int32_t>{1, 2, 3, 4, 5})
.receive([&](int32_t res) { CAF_CHECK_EQUAL(res, 15); }, handle_err);
self->request(pool, infinite, std::vector<int32_t>{6, 7, 8, 9, 10})
.receive([&](int32_t res) { CAF_CHECK_EQUAL(res, 40); }, handle_err);
self->send_exit(pool, exit_reason::user_shutdown);
}
......
......@@ -19,23 +19,22 @@
#define CAF_SUITE broadcast_downstream_manager
#include "core-test.hpp"
#include <cstdint>
#include <memory>
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/broadcast_downstream_manager.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/test/unit_test.hpp"
#include "caf/scheduled_actor.hpp"
using namespace caf;
namespace {
using bcast_manager = broadcast_downstream_manager<int>;
using bcast_manager = broadcast_downstream_manager<int32_t>;
// Mocks just enough of a stream manager to serve our entity.
class mock_stream_manager : public stream_manager {
......@@ -130,7 +129,7 @@ public:
return static_cast<size_t>((*i)->open_credit);
}
void new_round(int num, bool force_emit) {
void new_round(int32_t num, bool force_emit) {
for (auto& ptr : paths)
ptr->open_credit += num;
if (force_emit)
......@@ -165,9 +164,15 @@ struct not_empty_t {};
constexpr auto some = not_empty_t{};
struct config : actor_system_config {
config() {
add_message_types<id_block::core_test>();
}
};
// Provides the setup with alice, bob, and carl.
struct fixture {
actor_system_config cfg;
config cfg;
actor_system sys;
......@@ -205,7 +210,7 @@ struct fixture {
// nop
}
using batch_type = std::vector<int>;
using batch_type = std::vector<int32_t>;
using batches_type = std::vector<batch_type>;
......@@ -224,7 +229,7 @@ struct fixture {
return result;
}
batch_type make_batch(int first, int last) {
batch_type make_batch(int32_t first, int32_t last) {
batch_type result;
result.resize(static_cast<size_t>((last + 1) - first));
std::iota(result.begin(), result.end(), first);
......@@ -351,7 +356,7 @@ CAF_TEST(one_path_force) {
// Give alice 100 elements to send and a path to bob with desired batch size
// of 10.
alice.add_path_to(bob, 10);
for (int i = 1; i <= 100; ++i)
for (int32_t i = 1; i <= 100; ++i)
alice.mgr.out().push(i);
// Give 3 credit (less than 10).
AFTER ENTITY alice TRIED FORCE_SENDING 3 ELEMENTS {
......@@ -390,7 +395,7 @@ CAF_TEST(one_path_without_force) {
// Give alice 100 elements to send and a path to bob with desired batch size
// of 10.
alice.add_path_to(bob, 10);
for (int i = 1; i <= 100; ++i)
for (int32_t i = 1; i <= 100; ++i)
alice.mgr.out().push(i);
// Give 3 credit (less than 10).
AFTER ENTITY alice TRIED SENDING 3 ELEMENTS {
......@@ -430,7 +435,7 @@ CAF_TEST(two_paths_different_sizes_force) {
// 10, and a path to carl with desired batch size of 7.
alice.add_path_to(bob, 10);
alice.add_path_to(carl, 7);
for (int i = 1; i <= 100; ++i)
for (int32_t i = 1; i <= 100; ++i)
alice.mgr.out().push(i);
// Give 3 credit (less than 10).
AFTER ENTITY alice TRIED FORCE_SENDING 3 ELEMENTS {
......@@ -481,7 +486,7 @@ CAF_TEST(two_paths_different_sizes_without_force) {
// 10, and a path to carl with desired batch size of 7.
alice.add_path_to(bob, 10);
alice.add_path_to(carl, 7);
for (int i = 1; i <= 100; ++i)
for (int32_t i = 1; i <= 100; ++i)
alice.mgr.out().push(i);
// Give 3 credit (less than 10).
AFTER ENTITY alice TRIED SENDING 3 ELEMENTS {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE composable_behavior
#include "caf/composable_behavior.hpp"
#include "caf/test/dsl.hpp"
#include "caf/atom.hpp"
#include "caf/attach_stream_sink.hpp"
#include "caf/attach_stream_source.hpp"
#include "caf/attach_stream_stage.hpp"
#include "caf/typed_actor.hpp"
#define ERROR_HANDLER [&](error& err) { CAF_FAIL(system.render(err)); }
using namespace caf;
namespace {
// -- composable behaviors using primitive data types and streams --------------
using i3_actor = typed_actor<replies_to<int, int, int>::with<int>>;
using d_actor = typed_actor<replies_to<double>::with<double, double>>;
using source_actor = typed_actor<replies_to<open_atom>::with<stream<int>>>;
using stage_actor = typed_actor<replies_to<stream<int>>::with<stream<int>>>;
using sink_actor = typed_actor<reacts_to<stream<int>>>;
using foo_actor = i3_actor::extend_with<d_actor>;
class foo_actor_state : public composable_behavior<foo_actor> {
public:
result<int> operator()(int x, int y, int z) override {
return x + y + z;
}
result<double, double> operator()(double x) override {
return {x, x};
}
};
class i3_actor_state : public composable_behavior<i3_actor> {
public:
result<int> operator()(int x, int y, int z) override {
return x + y + z;
}
};
class d_actor_state : public composable_behavior<d_actor> {
public:
result<double, double> operator()(double x) override {
return {x, x};
}
};
class i3_actor_state2 : public composable_behavior<i3_actor> {
public:
result<int> operator()(int x, int y, int z) override {
return x * (y * z);
}
};
// checks whether CAF resolves "diamonds" properly by inheriting
// from two behaviors that both implement i3_actor
struct foo_actor_state2
: composed_behavior<i3_actor_state2, i3_actor_state, d_actor_state> {
result<int> operator()(int x, int y, int z) override {
return x - y - z;
}
};
class source_actor_state : public composable_behavior<source_actor> {
public:
result<stream<int>> operator()(open_atom) override {
return attach_stream_source(
self, [](size_t& counter) { counter = 0; },
[](size_t& counter, downstream<int>& out, size_t hint) {
auto n = std::min(static_cast<size_t>(100 - counter), hint);
for (size_t i = 0; i < n; ++i)
out.push(counter++);
},
[](const size_t& counter) { return counter < 100; });
}
};
class stage_actor_state : public composable_behavior<stage_actor> {
public:
result<stream<int>> operator()(stream<int> in) override {
return attach_stream_stage(
self, in,
[](unit_t&) {
// nop
},
[](unit_t&, downstream<int>& out, int x) {
if (x % 2 == 0)
out.push(x);
});
}
};
class sink_actor_state : public composable_behavior<sink_actor> {
public:
std::vector<int> buf;
result<void> operator()(stream<int> in) override {
attach_stream_sink(
self, in,
[](unit_t&) {
// nop
},
[=](unit_t&, int x) { buf.emplace_back(x); });
return unit;
}
};
// -- composable behaviors using param<T> arguments ----------------------------
std::atomic<long> counting_strings_created;
std::atomic<long> counting_strings_moved;
std::atomic<long> counting_strings_destroyed;
// counts how many instances where created
struct counting_string {
public:
counting_string() {
++counting_strings_created;
}
counting_string(const char* cstr) : str_(cstr) {
++counting_strings_created;
}
counting_string(const counting_string& x) : str_(x.str_) {
++counting_strings_created;
}
counting_string(counting_string&& x) : str_(std::move(x.str_)) {
++counting_strings_created;
++counting_strings_moved;
}
~counting_string() {
++counting_strings_destroyed;
}
counting_string& operator=(const char* cstr) {
str_ = cstr;
return *this;
}
const std::string& str() const {
return str_;
}
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f,
counting_string& x) {
return f(x.str_);
}
private:
std::string str_;
};
bool operator==(const counting_string& x, const counting_string& y) {
return x.str() == y.str();
}
bool operator==(const counting_string& x, const char* y) {
return x.str() == y;
}
std::string to_string(const counting_string& ref) {
return ref.str();
}
} // namespace
namespace std {
template <>
struct hash<counting_string> {
size_t operator()(const counting_string& ref) const {
hash<string> f;
return f(ref.str());
}
};
} // namespace std
namespace {
using add_atom = atom_constant<atom("add")>;
using get_name_atom = atom_constant<atom("getName")>;
using ping_atom = atom_constant<atom("ping")>;
using pong_atom = atom_constant<atom("pong")>;
// "base" interface
using named_actor = typed_actor<
replies_to<get_name_atom>::with<counting_string>,
replies_to<ping_atom>::with<pong_atom>>;
// a simple dictionary
using dict = named_actor::extend<
replies_to<get_atom, counting_string>::with<counting_string>,
replies_to<put_atom, counting_string, counting_string>::with<void>>;
class dict_state : public composable_behavior<dict> {
public:
result<counting_string> operator()(get_name_atom) override {
return "dictionary";
}
result<pong_atom> operator()(ping_atom) override {
return pong_atom::value;
}
result<counting_string> operator()(get_atom,
param<counting_string> key) override {
auto i = values_.find(key.get());
if (i == values_.end())
return "";
return i->second;
}
result<void> operator()(put_atom, param<counting_string> key,
param<counting_string> value) override {
if (values_.count(key.get()) != 0)
return unit;
values_.emplace(key.move(), value.move());
return unit;
}
protected:
std::unordered_map<counting_string, counting_string> values_;
};
using delayed_testee_actor = typed_actor<
reacts_to<int>, replies_to<bool>::with<int>, reacts_to<std::string>>;
class delayed_testee : public composable_behavior<delayed_testee_actor> {
public:
result<void> operator()(int x) override {
CAF_CHECK_EQUAL(x, 42);
delayed_anon_send(self, std::chrono::milliseconds(10), true);
return unit;
}
result<int> operator()(bool x) override {
CAF_CHECK_EQUAL(x, true);
self->delayed_send(self, std::chrono::milliseconds(10), "hello");
return 0;
}
result<void> operator()(param<std::string> x) override {
CAF_CHECK_EQUAL(x.get(), "hello");
return unit;
}
};
struct config : actor_system_config {
config() {
using foo_actor_impl = composable_behavior_based_actor<foo_actor_state>;
add_actor_type<foo_actor_impl>("foo_actor");
}
};
struct fixture : test_coordinator_fixture<config> {
// nop
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(composable_behaviors_tests, fixture)
CAF_TEST(composition) {
CAF_MESSAGE("test foo_actor_state");
auto f1 = sys.spawn<foo_actor_state>();
inject((int, int, int), from(self).to(f1).with(1, 2, 4));
expect((int), from(f1).to(self).with(7));
inject((double), from(self).to(f1).with(42.0));
expect((double, double), from(f1).to(self).with(42.0, 42.0));
CAF_MESSAGE("test composed_behavior<i3_actor_state, d_actor_state>");
f1 = sys.spawn<composed_behavior<i3_actor_state, d_actor_state>>();
inject((int, int, int), from(self).to(f1).with(1, 2, 4));
expect((int), from(f1).to(self).with(7));
inject((double), from(self).to(f1).with(42.0));
expect((double, double), from(f1).to(self).with(42.0, 42.0));
CAF_MESSAGE("test composed_behavior<i3_actor_state2, d_actor_state>");
f1 = sys.spawn<composed_behavior<i3_actor_state2, d_actor_state>>();
inject((int, int, int), from(self).to(f1).with(1, 2, 4));
expect((int), from(f1).to(self).with(8));
inject((double), from(self).to(f1).with(42.0));
expect((double, double), from(f1).to(self).with(42.0, 42.0));
CAF_MESSAGE("test foo_actor_state2");
f1 = sys.spawn<foo_actor_state2>();
inject((int, int, int), from(self).to(f1).with(1, 2, 4));
expect((int), from(f1).to(self).with(-5));
inject((double), from(self).to(f1).with(42.0));
expect((double, double), from(f1).to(self).with(42.0, 42.0));
}
CAF_TEST(param_detaching) {
auto dict = actor_cast<actor>(sys.spawn<dict_state>());
// Using CAF is the key to success!
counting_string key = "CAF";
counting_string value = "success";
CAF_CHECK_EQUAL(counting_strings_created.load(), 2);
CAF_CHECK_EQUAL(counting_strings_moved.load(), 0);
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 0);
// Wrap two strings into messages.
auto put_msg = make_message(put_atom::value, key, value);
auto get_msg = make_message(get_atom::value, key);
CAF_CHECK_EQUAL(counting_strings_created.load(), 5);
CAF_CHECK_EQUAL(counting_strings_moved.load(), 0);
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 0);
// Send put message to dictionary.
self->send(dict, put_msg);
sched.run();
// The handler of put_atom calls .move() on key and value, both causing to
// detach + move into the map.
CAF_CHECK_EQUAL(counting_strings_created.load(), 9);
CAF_CHECK_EQUAL(counting_strings_moved.load(), 2);
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 2);
// Send put message to dictionary again.
self->send(dict, put_msg);
sched.run();
// The handler checks whether key already exists -> no copies.
CAF_CHECK_EQUAL(counting_strings_created.load(), 9);
CAF_CHECK_EQUAL(counting_strings_moved.load(), 2);
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 2);
// Alter our initial put, this time moving it to the dictionary.
put_msg.get_mutable_as<counting_string>(1) = "neverlord";
put_msg.get_mutable_as<counting_string>(2) = "CAF";
// Send new put message to dictionary.
self->send(dict, std::move(put_msg));
sched.run();
// The handler of put_atom calls .move() on key and value, but no detaching
// occurs this time (unique access) -> move into the map.
CAF_CHECK_EQUAL(counting_strings_created.load(), 11);
CAF_CHECK_EQUAL(counting_strings_moved.load(), 4);
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 4);
// Finally, check for original key.
self->send(dict, std::move(get_msg));
sched.run();
self->receive([&](const counting_string& str) {
// We receive a copy of the value, which is copied out of the map and then
// moved into the result message; the string from our get_msg is destroyed.
CAF_CHECK_EQUAL(counting_strings_created.load(), 13);
CAF_CHECK_EQUAL(counting_strings_moved.load(), 5);
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 6);
CAF_CHECK_EQUAL(str, "success");
});
// Temporary of our handler is destroyed.
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 7);
self->send_exit(dict, exit_reason::user_shutdown);
sched.run();
dict = nullptr;
// Only `key` and `value` from this scope remain.
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 11);
}
CAF_TEST(delayed_sends) {
auto testee = self->spawn<delayed_testee>();
inject((int), from(self).to(testee).with(42));
disallow((bool), from(_).to(testee));
sched.trigger_timeouts();
expect((bool), from(_).to(testee));
disallow((std::string), from(testee).to(testee).with("hello"));
sched.trigger_timeouts();
expect((std::string), from(testee).to(testee).with("hello"));
}
CAF_TEST(dynamic_spawning) {
auto testee = unbox(sys.spawn<foo_actor>("foo_actor", make_message()));
inject((int, int, int), from(self).to(testee).with(1, 2, 4));
expect((int), from(testee).to(self).with(7));
inject((double), from(self).to(testee).with(42.0));
expect((double, double), from(testee).to(self).with(42.0, 42.0));
}
CAF_TEST(streaming) {
auto src = sys.spawn<source_actor_state>();
auto stg = sys.spawn<stage_actor_state>();
auto snk = sys.spawn<sink_actor_state>();
using src_to_stg = typed_actor<replies_to<open_atom>::with<stream<int>>>;
using stg_to_snk = typed_actor<reacts_to<stream<int>>>;
static_assert(std::is_same<decltype(stg * src), src_to_stg>::value,
"stg * src produces the wrong type");
static_assert(std::is_same<decltype(snk * stg), stg_to_snk>::value,
"stg * src produces the wrong type");
auto pipeline = snk * stg * src;
self->send(pipeline, open_atom::value);
run();
using sink_actor = composable_behavior_based_actor<sink_actor_state>;
auto& st = deref<sink_actor>(snk).state;
CAF_CHECK_EQUAL(st.buf.size(), 50u);
auto is_even = [](int x) { return x % 2 == 0; };
CAF_CHECK(std::all_of(st.buf.begin(), st.buf.end(), is_even));
anon_send_exit(src, exit_reason::user_shutdown);
anon_send_exit(stg, exit_reason::user_shutdown);
anon_send_exit(snk, exit_reason::user_shutdown);
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -18,7 +18,7 @@
#define CAF_SUITE continuous_streaming
#include "caf/test/dsl.hpp"
#include "core-test.hpp"
#include <memory>
#include <numeric>
......@@ -37,79 +37,83 @@ using namespace caf;
namespace {
/// Returns the sum of natural numbers up until `n`, i.e., 1 + 2 + ... + n.
int sum(int n) {
int32_t sum(int32_t n) {
return (n * (n + 1)) / 2;
}
TESTEE_SETUP();
TESTEE_STATE(file_reader) {
std::vector<int> buf;
std::vector<int32_t> buf;
};
VARARGS_TESTEE(file_reader, size_t buf_size) {
return {[=](string& fname) -> result<stream<int>, string> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return attach_stream_source(
self,
// forward file name in handshake to next stage
std::forward_as_tuple(std::move(fname)),
// initialize state
[=](unit_t&) {
auto& xs = self->state.buf;
xs.resize(buf_size);
std::iota(xs.begin(), xs.end(), 1);
},
// get next element
[=](unit_t&, downstream<int>& out, size_t num) {
auto& xs = self->state.buf;
CAF_MESSAGE("push " << num << " messages downstream");
auto n = std::min(num, xs.size());
for (size_t i = 0; i < n; ++i)
out.push(xs[i]);
xs.erase(xs.begin(), xs.begin() + static_cast<ptrdiff_t>(n));
},
// check whether we reached the end
[=](const unit_t&) {
if (self->state.buf.empty()) {
CAF_MESSAGE(self->name() << " is done");
return true;
}
return false;
});
}};
return {
[=](string& fname) -> result<stream<int32_t>, string> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return attach_stream_source(
self,
// forward file name in handshake to next stage
std::forward_as_tuple(std::move(fname)),
// initialize state
[=](unit_t&) {
auto& xs = self->state.buf;
xs.resize(buf_size);
std::iota(xs.begin(), xs.end(), 1);
},
// get next element
[=](unit_t&, downstream<int32_t>& out, size_t num) {
auto& xs = self->state.buf;
CAF_MESSAGE("push " << num << " messages downstream");
auto n = std::min(num, xs.size());
for (size_t i = 0; i < n; ++i)
out.push(xs[i]);
xs.erase(xs.begin(), xs.begin() + static_cast<ptrdiff_t>(n));
},
// check whether we reached the end
[=](const unit_t&) {
if (self->state.buf.empty()) {
CAF_MESSAGE(self->name() << " is done");
return true;
}
return false;
});
},
};
}
TESTEE_STATE(sum_up) {
int x = 0;
int32_t x = 0;
};
TESTEE(sum_up) {
return {[=](stream<int>& in, const string& fname) {
CAF_CHECK_EQUAL(fname, "numbers.txt");
using int_ptr = int*;
return attach_stream_sink(
self,
// input stream
in,
// initialize state
[=](int_ptr& x) { x = &self->state.x; },
// processing step
[](int_ptr& x, int y) { *x += y; },
// cleanup
[=](int_ptr&, const error&) {
CAF_MESSAGE(self->name() << " is done");
});
},
[=](join_atom atm, actor src) {
CAF_MESSAGE(self->name() << " joins a stream");
self->send(self * src, atm);
}};
return {
[=](stream<int32_t>& in, const string& fname) {
CAF_CHECK_EQUAL(fname, "numbers.txt");
using int_ptr = int32_t*;
return attach_stream_sink(
self,
// input stream
in,
// initialize state
[=](int_ptr& x) { x = &self->state.x; },
// processing step
[](int_ptr& x, int32_t y) { *x += y; },
// cleanup
[=](int_ptr&, const error&) {
CAF_MESSAGE(self->name() << " is done");
});
},
[=](join_atom atm, actor src) {
CAF_MESSAGE(self->name() << " joins a stream");
self->send(self * src, atm);
},
};
}
TESTEE_STATE(stream_multiplexer) {
stream_stage_ptr<int, broadcast_downstream_manager<int>> stage;
stream_stage_ptr<int32_t, broadcast_downstream_manager<int32_t>> stage;
};
TESTEE(stream_multiplexer) {
......@@ -120,7 +124,7 @@ TESTEE(stream_multiplexer) {
// nop
},
// processing step
[](unit_t&, downstream<int>& out, int x) { out.push(x); },
[](unit_t&, downstream<int32_t>& out, int32_t x) { out.push(x); },
// cleanup
[=](unit_t&, const error&) { CAF_MESSAGE(self->name() << " is done"); });
return {
......@@ -129,18 +133,24 @@ TESTEE(stream_multiplexer) {
return self->state.stage->add_outbound_path(
std::make_tuple("numbers.txt"));
},
[=](const stream<int>& in, std::string& fname) {
[=](const stream<int32_t>& in, std::string& fname) {
CAF_CHECK_EQUAL(fname, "numbers.txt");
return self->state.stage->add_inbound_path(in);
},
[=](close_atom, int sink_index) {
[=](close_atom, int32_t sink_index) {
auto& out = self->state.stage->out();
out.close(out.path_slots().at(static_cast<size_t>(sink_index)));
},
};
}
using fixture = test_coordinator_fixture<>;
struct config : actor_system_config {
config() {
add_message_types<id_block::core_test>();
}
};
using fixture = test_coordinator_fixture<config>;
} // namespace
......@@ -224,7 +234,7 @@ CAF_TEST(closing_downstreams_before_end_of_stream) {
auto sink1_result = sum(next_pending - 1);
CAF_MESSAGE("gracefully close sink 1, next pending: " << next_pending);
self->send(stg, close_atom::value, 0);
expect((atom_value, int), from(self).to(stg));
expect((atom_value, int32_t), from(self).to(stg));
CAF_MESSAGE("ship remaining elements");
run();
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 1u);
......
#include "core-test.hpp"
#include "caf/test/unit_test_impl.hpp"
fail_on_copy::fail_on_copy(const fail_on_copy&) {
CAF_FAIL("fail_on_copy: copy constructor called");
}
fail_on_copy& fail_on_copy::operator=(const fail_on_copy&) {
CAF_FAIL("fail_on_copy: copy assign operator called");
}
size_t i32_wrapper::instances = 0;
size_t i64_wrapper::instances = 0;
void test_empty_non_pod::foo() {
// nop
}
test_empty_non_pod::~test_empty_non_pod() {
// nop
}
namespace {
const char* test_enum_strings[] = {
"a",
"b",
"c",
};
} // namespace
std::string to_string(test_enum x) {
return test_enum_strings[static_cast<uint32_t>(x)];
}
#include "caf/fwd.hpp"
#include "caf/test/dsl.hpp"
#include "caf/type_id.hpp"
#include "caf/typed_actor.hpp"
#include <cstdint>
#include <numeric>
#include <string>
#include <utility>
// -- forward declarations for all unit test suites ----------------------------
using float_actor = caf::typed_actor<caf::reacts_to<float>>;
using int_actor = caf::typed_actor<caf::replies_to<int32_t>::with<int32_t>>;
using foo_actor
= caf::typed_actor<caf::replies_to<int32_t, int32_t, int32_t>::with<int32_t>,
caf::replies_to<double>::with<double, double>>;
// A simple POD type.
struct dummy_struct {
int a;
std::string b;
friend bool operator==(const dummy_struct& x, const dummy_struct& y) {
return x.a == y.a && x.b == y.b;
}
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, dummy_struct& x) {
return f(caf::meta::type_name("dummy_struct"), x.a, x.b);
}
// An empty type.
struct dummy_tag_type {};
constexpr bool operator==(dummy_tag_type, dummy_tag_type) {
return true;
}
struct fail_on_copy {
int value;
fail_on_copy() : value(0) {
// nop
}
explicit fail_on_copy(int x) : value(x) {
// nop
}
fail_on_copy(fail_on_copy&&) = default;
fail_on_copy& operator=(fail_on_copy&&) = default;
fail_on_copy(const fail_on_copy&);
fail_on_copy& operator=(const fail_on_copy&);
template <class Inspector>
friend typename Inspector::result_type
inspect(Inspector& f, fail_on_copy& x) {
return f(x.value);
}
};
struct i32_wrapper {
static size_t instances;
int32_t value;
i32_wrapper() : value(0) {
++instances;
}
~i32_wrapper() {
--instances;
}
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f, i32_wrapper& x) {
return f(x.value);
}
};
struct i64_wrapper {
static size_t instances;
int64_t value;
i64_wrapper() : value(0) {
++instances;
}
~i64_wrapper() {
--instances;
}
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f, i64_wrapper& x) {
return f(x.value);
}
};
struct my_request {
int32_t a;
int32_t b;
friend bool operator==(const my_request& x, const my_request& y) {
return std::tie(x.a, x.b) == std::tie(y.a, y.b);
}
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, my_request& x) {
return f(x.a, x.b);
}
struct raw_struct {
std::string str;
friend bool operator==(const raw_struct& lhs, const raw_struct& rhs) {
return lhs.str == rhs.str;
}
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, raw_struct& x) {
return f(x.str);
}
struct s1 {
int value[3] = {10, 20, 30};
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, s1& x) {
return f(x.value);
}
struct s2 {
int value[4][2] = {{1, 10}, {2, 20}, {3, 30}, {4, 40}};
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, s2& x) {
return f(x.value);
}
struct s3 {
std::array<int, 4> value;
s3() {
std::iota(value.begin(), value.end(), 1);
}
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, s3& x) {
return f(x.value);
}
struct test_array {
int32_t value[4];
int32_t value2[2][4];
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, test_array& x) {
return f(x.value, x.value2);
}
struct test_empty_non_pod {
test_empty_non_pod() = default;
test_empty_non_pod(const test_empty_non_pod&) = default;
test_empty_non_pod& operator=(const test_empty_non_pod&) = default;
virtual void foo();
virtual ~test_empty_non_pod();
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, test_empty_non_pod&) {
return f();
}
enum class test_enum : int32_t {
a,
b,
c,
};
std::string to_string(test_enum x);
struct test_data {
int32_t i32;
int64_t i64;
float f32;
double f64;
caf::timestamp ts;
test_enum te;
std::string str;
friend bool operator==(const test_data& x, const test_data& y) {
return std::tie(x.i32, x.i64, x.f32, x.f64, x.ts, x.te, x.str)
== std::tie(y.i32, y.i64, y.f32, y.f64, y.ts, y.te, y.str);
}
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, test_data& x) {
return f(caf::meta::type_name("test_data"), x.i32, x.i64, x.f32, x.f64, x.ts,
x.te, x.str);
}
enum class dummy_enum_class : short { foo, bar };
inline std::string to_string(dummy_enum_class x) {
return x == dummy_enum_class::foo ? "foo" : "bar";
}
enum class level { all, trace, debug, warning, error };
enum dummy_enum { de_foo, de_bar };
// -- type IDs for for all unit test suites ------------------------------------
#define ADD_TYPE_ID(type) CAF_ADD_TYPE_ID(core_test, type)
#define ADD_ATOM(atom_name, atom_text) \
CAF_ADD_ATOM(core_test, core_test_atoms, atom_name, atom_text)
CAF_BEGIN_TYPE_ID_BLOCK(core_test, caf::first_custom_type_id)
ADD_TYPE_ID((caf::stream<int32_t>) )
ADD_TYPE_ID((caf::stream<std::string>) )
ADD_TYPE_ID((caf::stream<std::pair<level, std::string>>) )
ADD_TYPE_ID((dummy_enum))
ADD_TYPE_ID((dummy_enum_class))
ADD_TYPE_ID((dummy_struct))
ADD_TYPE_ID((dummy_tag_type))
ADD_TYPE_ID((fail_on_copy))
ADD_TYPE_ID((float_actor))
ADD_TYPE_ID((foo_actor))
ADD_TYPE_ID((i32_wrapper))
ADD_TYPE_ID((i64_wrapper))
ADD_TYPE_ID((int_actor))
ADD_TYPE_ID((level))
ADD_TYPE_ID((my_request))
ADD_TYPE_ID((raw_struct))
ADD_TYPE_ID((s1))
ADD_TYPE_ID((s2))
ADD_TYPE_ID((s3))
ADD_TYPE_ID((std::map<int32_t, int32_t>) )
ADD_TYPE_ID((std::map<std::string, std::u16string>) )
ADD_TYPE_ID((std::pair<level, std::string>) )
ADD_TYPE_ID((std::tuple<int32_t, int32_t, int32_t>) )
ADD_TYPE_ID((std::tuple<std::string, int32_t, uint32_t>) )
ADD_TYPE_ID((std::vector<bool>) )
ADD_TYPE_ID((std::vector<int32_t>) )
ADD_TYPE_ID((std::vector<std::pair<level, std::string>>) )
ADD_TYPE_ID((std::vector<std::string>) )
ADD_TYPE_ID((test_array))
ADD_TYPE_ID((test_empty_non_pod))
ADD_TYPE_ID((test_enum))
ADD_ATOM(abc_atom, "abc")
ADD_ATOM(get_state_atom, "get_state")
ADD_ATOM(name_atom, "name")
CAF_END_TYPE_ID_BLOCK(core_test)
using namespace core_test_atoms;
#undef ADD_TYPE_ID
#undef ADD_ATOM
......@@ -24,7 +24,7 @@
#include "caf/all.hpp"
#define ERROR_HANDLER [&](error& err) { CAF_FAIL(system.render(err)); }
#define ERROR_HANDLER [&](error& err) { CAF_FAIL(err); }
using namespace caf;
......@@ -42,7 +42,9 @@ using second_stage = typed_actor<replies_to<double, double>::with<double>>;
first_stage::behavior_type typed_first_stage() {
return {
[](int i) { return std::make_tuple(i * 2.0, i * 4.0); },
[](int i) -> result<double, double> {
return {i * 2.0, i * 4.0};
},
};
}
......
......@@ -88,11 +88,11 @@ CAF_TEST(shutdown_delayed_send_loop) {
CAF_MESSAGE("does sys shut down after spawning a detached actor that used "
"a delayed send loop and was interrupted via exit message?");
auto f = [](event_based_actor* self) -> behavior {
self->send(self, std::chrono::milliseconds(1), ok_atom::value);
self->delayed_send(self, std::chrono::milliseconds(1), ok_atom::value);
return {
[=](ok_atom) {
self->send(self, std::chrono::milliseconds(1), ok_atom::value);
}
self->delayed_send(self, std::chrono::milliseconds(1), ok_atom::value);
},
};
};
auto a = sys.spawn<detached>(f);
......
......@@ -20,7 +20,7 @@
#include "caf/fused_downstream_manager.hpp"
#include "caf/test/dsl.hpp"
#include "core-test.hpp"
#include <memory>
#include <numeric>
......@@ -250,11 +250,11 @@ TESTEE(stream_multiplexer) {
struct config : actor_system_config {
config() {
add_message_type<std::deque<std::string>>("deque<string>");
add_message_types<id_block::core_test>();
}
};
using fixture = test_coordinator_fixture<>;
using fixture = test_coordinator_fixture<config>;
} // namespace
......
......@@ -16,26 +16,24 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE message_operations
#include "core-test.hpp"
#include <iostream>
#include <map>
#include <vector>
#include <string>
#include <numeric>
#include <iostream>
#include <set>
#include <string>
#include <unordered_set>
#include "caf/config.hpp"
#define CAF_SUITE message_operations
#include "caf/test/unit_test.hpp"
#include <vector>
#include "caf/all.hpp"
using std::map;
using std::string;
using std::tuple;
using std::vector;
using std::make_tuple;
using namespace caf;
......@@ -185,45 +183,11 @@ CAF_TEST(concat) {
CAF_CHECK_EQUAL(to_string(message::concat(m3, message{}, m1, m2)), to_string(m4));
}
namespace {
struct s1 {
int value[3] = {10, 20, 30};
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, s1& x) {
return f(x.value);
}
struct s2 {
int value[4][2] = {{1, 10}, {2, 20}, {3, 30}, {4, 40}};
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, s2& x) {
return f(x.value);
}
struct s3 {
std::array<int, 4> value;
s3() {
std::iota(value.begin(), value.end(), 1);
}
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, s3& x) {
return f(x.value);
}
template <class... Ts>
std::string msg_as_string(Ts&&... xs) {
return to_string(make_message(std::forward<Ts>(xs)...));
}
} // namespace
CAF_TEST(compare_custom_types) {
s2 tmp;
tmp.value[0][1] = 100;
......@@ -237,7 +201,7 @@ CAF_TEST(empty_to_string) {
}
CAF_TEST(integers_to_string) {
using ivec = vector<int>;
using ivec = vector<int32_t>;
CAF_CHECK_EQUAL(msg_as_string(1, 2, 3), "(1, 2, 3)");
CAF_CHECK_EQUAL(msg_as_string(ivec{1, 2, 3}), "([1, 2, 3])");
CAF_CHECK_EQUAL(msg_as_string(ivec{1, 2}, 3, 4, ivec{5, 6, 7}),
......@@ -259,15 +223,16 @@ CAF_TEST(strings_to_string) {
}
CAF_TEST(maps_to_string) {
map<int, int> m1{{1, 10}, {2, 20}, {3, 30}};
map<int32_t, int32_t> m1{{1, 10}, {2, 20}, {3, 30}};
auto msg1 = make_message(move(m1));
CAF_CHECK_EQUAL(to_string(msg1), "({1 = 10, 2 = 20, 3 = 30})");
}
CAF_TEST(tuples_to_string) {
auto msg1 = make_message(make_tuple(1, 2, 3), 4, 5);
auto msg1 = make_message(tuple<int32_t, int32_t, int32_t>{1, 2, 3}, 4, 5);
CAF_CHECK_EQUAL(to_string(msg1), "((1, 2, 3), 4, 5)");
auto msg2 = make_message(make_tuple(string{"one"}, 2, uint32_t{3}), 4, true);
using msg2_type = tuple<string, int32_t, uint32_t>;
auto msg2 = make_message(msg2_type{"one", 2, 3}, 4, true);
CAF_CHECK_EQUAL(to_string(msg2), "((\"one\", 2, 3), 4, true)");
}
......
......@@ -16,10 +16,9 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/config.hpp"
#define CAF_SUITE message_lifetime
#include "caf/test/unit_test.hpp"
#include "core-test.hpp"
#include <atomic>
#include <iostream>
......@@ -81,35 +80,16 @@ private:
message msg_;
};
struct fixture {
actor_system_config cfg;
actor_system system{cfg};
};
struct fail_on_copy {
int value;
fail_on_copy(int x = 0) : value(x) {
// nop
}
fail_on_copy(fail_on_copy&&) = default;
fail_on_copy& operator=(fail_on_copy&&) = default;
fail_on_copy(const fail_on_copy&) {
CAF_FAIL("fail_on_copy: copy constructor called");
}
fail_on_copy& operator=(const fail_on_copy&) {
CAF_FAIL("fail_on_copy: copy assign operator called");
struct config : actor_system_config {
config() {
add_message_types<id_block::core_test>();
}
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, fail_on_copy& x) {
return f(x.value);
}
struct fixture {
config cfg;
actor_system system{cfg};
};
} // namespace
......
......@@ -81,6 +81,13 @@
#include "caf/detail/stream_stage_impl.hpp"
#include "caf/detail/tick_emitter.hpp"
CAF_BEGIN_TYPE_ID_BLOCK(native_streaming_classes, first_custom_type_id)
CAF_ADD_TYPE_ID(native_streaming_classes, (caf::stream<int32_t>) )
CAF_ADD_TYPE_ID(native_streaming_classes, (std::vector<int32_t>) )
CAF_END_TYPE_ID_BLOCK(native_streaming_classes)
using std::vector;
using namespace caf;
......@@ -228,9 +235,9 @@ public:
return nullptr;
}
void start_streaming(entity& ref, int num_messages) {
void start_streaming(entity& ref, int32_t num_messages) {
CAF_REQUIRE_NOT_EQUAL(num_messages, 0);
using downstream_manager = broadcast_downstream_manager<int>;
using downstream_manager = broadcast_downstream_manager<int32_t>;
struct driver final : public stream_source_driver<downstream_manager> {
public:
driver(int32_t sentinel) : x_(0), sentinel_(sentinel) {
......@@ -540,6 +547,7 @@ struct fixture {
caf::test::engine::argv()))
CAF_FAIL("parsing the config failed: " << to_string(err));
cfg.set("scheduler.policy", caf::atom("testing"));
cfg.add_message_types<id_block::native_streaming_classes>();
return cfg;
}
......
......@@ -30,6 +30,13 @@
#include "caf/event_based_actor.hpp"
#include "caf/stateful_actor.hpp"
CAF_BEGIN_TYPE_ID_BLOCK(pipeline_streaming, first_custom_type_id)
CAF_ADD_TYPE_ID(pipeline_streaming, (caf::stream<int32_t>) )
CAF_ADD_TYPE_ID(pipeline_streaming, (std::vector<int32_t>) )
CAF_END_TYPE_ID_BLOCK(pipeline_streaming)
using std::string;
using namespace caf;
......@@ -83,16 +90,16 @@ TESTEE_STATE(infinite_source) {
TESTEE(infinite_source) {
return {
[=](string& fname) -> result<stream<int>> {
[=](string& fname) -> result<stream<int32_t>> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return attach_stream_source(
self, [](int& x) { x = 0; },
[](int& x, downstream<int>& out, size_t num) {
self, [](int32_t& x) { x = 0; },
[](int32_t& x, downstream<int32_t>& out, size_t num) {
for (size_t i = 0; i < num; ++i)
out.push(x++);
},
[](const int&) { return false; }, fin<int>(self));
[](const int32_t&) { return false; }, fin<int>(self));
},
};
}
......@@ -103,7 +110,7 @@ TESTEE_STATE(file_reader) {
VARARGS_TESTEE(file_reader, size_t buf_size) {
return {
[=](string& fname) -> result<stream<int>> {
[=](string& fname) -> result<stream<int32_t>> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return attach_stream_source(self, init(buf_size), push_from_buf,
......@@ -119,14 +126,14 @@ VARARGS_TESTEE(file_reader, size_t buf_size) {
}
TESTEE_STATE(sum_up) {
int x = 0;
int fin_called = 0;
int32_t x = 0;
int32_t fin_called = 0;
};
TESTEE(sum_up) {
using intptr = int*;
using intptr = int32_t*;
return {
[=](stream<int>& in) {
[=](stream<int32_t>& in) {
return attach_stream_sink(
self,
// input stream
......@@ -134,22 +141,22 @@ TESTEE(sum_up) {
// initialize state
[=](intptr& x) { x = &self->state.x; },
// processing step
[](intptr& x, int y) { *x += y; }, fin<intptr>(self));
[](intptr& x, int32_t y) { *x += y; }, fin<intptr>(self));
},
};
}
TESTEE_STATE(delayed_sum_up) {
int x = 0;
int fin_called = 0;
int32_t x = 0;
int32_t fin_called = 0;
};
TESTEE(delayed_sum_up) {
using intptr = int*;
using intptr = int32_t*;
self->set_default_handler(skip);
return {
[=](ok_atom) {
self->become([=](stream<int>& in) {
self->become([=](stream<int32_t>& in) {
self->set_default_handler(print_and_drop);
return attach_stream_sink(
self,
......@@ -158,7 +165,7 @@ TESTEE(delayed_sum_up) {
// initialize state
[=](intptr& x) { x = &self->state.x; },
// processing step
[](intptr& x, int y) { *x += y; },
[](intptr& x, int32_t y) { *x += y; },
// cleanup
fin<intptr>(self));
});
......@@ -167,26 +174,26 @@ TESTEE(delayed_sum_up) {
}
TESTEE_STATE(broken_sink) {
int fin_called = 0;
int32_t fin_called = 0;
};
TESTEE(broken_sink) {
CAF_IGNORE_UNUSED(self);
return {
[=](stream<int>&, const actor&) {
[=](stream<int32_t>&, const actor&) {
// nop
},
};
}
TESTEE_STATE(filter) {
int fin_called = 0;
int32_t fin_called = 0;
};
TESTEE(filter) {
CAF_IGNORE_UNUSED(self);
return {
[=](stream<int>& in) {
[=](stream<int32_t>& in) {
return attach_stream_stage(
self,
// input stream
......@@ -196,7 +203,7 @@ TESTEE(filter) {
// nop
},
// processing step
[](unit_t&, downstream<int>& out, int x) {
[](unit_t&, downstream<int32_t>& out, int32_t x) {
if ((x & 0x01) != 0)
out.push(x);
},
......@@ -207,13 +214,13 @@ TESTEE(filter) {
}
TESTEE_STATE(doubler) {
int fin_called = 0;
int32_t fin_called = 0;
};
TESTEE(doubler) {
CAF_IGNORE_UNUSED(self);
return {
[=](stream<int>& in) {
[=](stream<int32_t>& in) {
return attach_stream_stage(
self,
// input stream
......@@ -223,14 +230,20 @@ TESTEE(doubler) {
// nop
},
// processing step
[](unit_t&, downstream<int>& out, int x) { out.push(x * 2); },
[](unit_t&, downstream<int32_t>& out, int32_t x) { out.push(x * 2); },
// cleanup
fin<unit_t>(self));
},
};
}
struct fixture : test_coordinator_fixture<> {
struct config : actor_system_config {
config() {
add_message_types<id_block::pipeline_streaming>();
}
};
struct fixture : test_coordinator_fixture<config> {
void tick() {
advance_time(cfg.stream_credit_round_interval);
}
......
......@@ -68,14 +68,14 @@ struct consumer {
intrusive::task_result operator()(const Key&, const Queue&,
const mailbox_element& x) {
if (!x.content().match_elements<int>())
CAF_FAIL("unexepected message: " << x.content());
CAF_FAIL("unexpected message: " << x.content());
ints.emplace_back(x.content().get_as<int>(0));
return intrusive::task_result::resume;
}
template <class Key, class Queue, class... Ts>
intrusive::task_result operator()(const Key&, const Queue&, const Ts&...) {
CAF_FAIL("unexepected message type"); // << typeid(Ts).name());
CAF_FAIL("unexpected message type"); // << typeid(Ts).name());
return intrusive::task_result::resume;
}
};
......
......@@ -33,6 +33,21 @@
#include "caf/event_based_actor.hpp"
#include "caf/stateful_actor.hpp"
namespace {
enum class level;
} // namespace
CAF_BEGIN_TYPE_ID_BLOCK(selective_streaming, first_custom_type_id)
CAF_ADD_TYPE_ID(selective_streaming, (caf::stream<std::pair<level, std::string>>) )
CAF_ADD_TYPE_ID(selective_streaming, (level))
CAF_ADD_TYPE_ID(selective_streaming, (std::pair<level, std::string>) )
CAF_ADD_TYPE_ID(selective_streaming, (std::vector<std::pair<level, std::string>>) )
CAF_END_TYPE_ID_BLOCK(selective_streaming)
using std::string;
using namespace caf;
......@@ -73,34 +88,36 @@ buf make_log(level lvl) {
TESTEE_SETUP();
TESTEE(log_producer) {
return {[=](level lvl) -> result<stream<value_type>> {
auto res = attach_stream_source(
self,
// initialize state
[=](buf& xs) { xs = make_log(lvl); },
// get next element
[](buf& xs, downstream<value_type>& out, size_t num) {
CAF_MESSAGE("push " << num << " messages downstream");
auto n = std::min(num, xs.size());
for (size_t i = 0; i < n; ++i)
out.push(xs[i]);
xs.erase(xs.begin(), xs.begin() + static_cast<ptrdiff_t>(n));
},
// check whether we reached the end
[=](const buf& xs) {
if (xs.empty()) {
CAF_MESSAGE(self->name() << " is done");
return true;
}
return false;
},
unit, policy::arg<manager_type>::value);
auto& out = res.ptr()->out();
static_assert(std::is_same<decltype(out), manager_type&>::value,
"source has wrong manager_type type");
out.set_filter(res.outbound_slot(), lvl);
return res;
}};
return {
[=](level lvl) -> result<stream<value_type>> {
auto res = attach_stream_source(
self,
// initialize state
[=](buf& xs) { xs = make_log(lvl); },
// get next element
[](buf& xs, downstream<value_type>& out, size_t num) {
CAF_MESSAGE("push " << num << " messages downstream");
auto n = std::min(num, xs.size());
for (size_t i = 0; i < n; ++i)
out.push(xs[i]);
xs.erase(xs.begin(), xs.begin() + static_cast<ptrdiff_t>(n));
},
// check whether we reached the end
[=](const buf& xs) {
if (xs.empty()) {
CAF_MESSAGE(self->name() << " is done");
return true;
}
return false;
},
unit, policy::arg<manager_type>::value);
auto& out = res.ptr()->out();
static_assert(std::is_same<decltype(out), manager_type&>::value,
"source has wrong manager_type type");
out.set_filter(res.outbound_slot(), lvl);
return res;
},
};
}
TESTEE_STATE(log_dispatcher) {
......@@ -158,7 +175,7 @@ TESTEE(log_consumer) {
struct config : actor_system_config {
config() {
add_message_type<value_type>("value_type");
add_message_types<id_block::selective_streaming>();
}
};
......
......@@ -68,6 +68,25 @@
#include "caf/detail/safe_equal.hpp"
#include "caf/detail/type_traits.hpp"
namespace {
enum class test_enum : uint32_t;
struct raw_struct;
struct test_array;
struct test_empty_non_pod;
} // namespace
CAF_BEGIN_TYPE_ID_BLOCK(serialization, first_custom_type_id)
CAF_ADD_TYPE_ID(serialization, (raw_struct))
CAF_ADD_TYPE_ID(serialization, (std::vector<bool>))
CAF_ADD_TYPE_ID(serialization, (test_array))
CAF_ADD_TYPE_ID(serialization, (test_empty_non_pod))
CAF_ADD_TYPE_ID(serialization, (test_enum))
CAF_END_TYPE_ID_BLOCK(serialization)
using namespace std;
using namespace caf;
using caf::detail::type_erased_value_impl;
......@@ -135,11 +154,8 @@ typename Inspector::result_type inspect(Inspector& f, test_empty_non_pod&) {
class config : public actor_system_config {
public:
config() {
add_message_type<test_enum>("test_enum");
add_message_type<raw_struct>("raw_struct");
add_message_type<test_array>("test_array");
add_message_type<test_empty_non_pod>("test_empty_non_pod");
add_message_type<std::vector<bool>>("bool_vector");
puts("add_message_types");
add_message_types<id_block::serialization>();
}
};
......
......@@ -31,6 +31,39 @@
# define ERROR_HANDLER [&](error& err) { CAF_FAIL(system.render(err)); }
namespace {
struct get_state_msg;
struct my_request;
using server_type = caf::typed_actor<caf::replies_to<my_request>::with<bool>>;
using event_testee_type
= caf::typed_actor<caf::replies_to<get_state_msg>::with<std::string>,
caf::replies_to<std::string>::with<void>,
caf::replies_to<float>::with<void>,
caf::replies_to<int32_t>::with<int32_t>>;
using string_actor
= caf::typed_actor<caf::replies_to<std::string>::with<std::string>>;
using int_actor = caf::typed_actor<caf::replies_to<int32_t>::with<int32_t>>;
using float_actor = caf::typed_actor<caf::reacts_to<float>>;
} // namespace
CAF_BEGIN_TYPE_ID_BLOCK(typed_spawn, first_custom_type_id)
CAF_ADD_TYPE_ID(typed_spawn, (event_testee_type))
CAF_ADD_TYPE_ID(typed_spawn, (float_actor))
CAF_ADD_TYPE_ID(typed_spawn, (get_state_msg))
CAF_ADD_TYPE_ID(typed_spawn, (int_actor))
CAF_ADD_TYPE_ID(typed_spawn, (my_request))
CAF_ADD_TYPE_ID(typed_spawn, (string_actor))
CAF_END_TYPE_ID_BLOCK(typed_spawn)
using std::string;
using namespace caf;
......@@ -48,7 +81,7 @@ error make_error(mock_errc x) {
}
// check invariants of type system
using dummy1 = typed_actor<reacts_to<int, int>,
using dummy1 = typed_actor<reacts_to<int32_t, int32_t>,
replies_to<double>::with<double>>;
using dummy2 = dummy1::extend<reacts_to<ok_atom>>;
......@@ -56,8 +89,8 @@ using dummy2 = dummy1::extend<reacts_to<ok_atom>>;
static_assert(std::is_convertible<dummy2, dummy1>::value,
"handle not assignable to narrower definition");
using dummy3 = typed_actor<reacts_to<float, int>>;
using dummy4 = typed_actor<replies_to<int>::with<double>>;
using dummy3 = typed_actor<reacts_to<float, int32_t>>;
using dummy4 = typed_actor<replies_to<int32_t>::with<double>>;
using dummy5 = dummy4::extend_with<dummy3>;
static_assert(std::is_convertible<dummy5, dummy3>::value,
......@@ -71,8 +104,8 @@ static_assert(std::is_convertible<dummy5, dummy4>::value,
******************************************************************************/
struct my_request {
int a;
int b;
int32_t a;
int32_t b;
};
template <class Inspector>
......@@ -80,8 +113,6 @@ typename Inspector::result_type inspect(Inspector& f, my_request& x) {
return f(x.a, x.b);
}
using server_type = typed_actor<replies_to<my_request>::with<bool>>;
server_type::behavior_type typed_server1() {
return {
[](const my_request& req) { return req.a == req.b; },
......@@ -121,10 +152,6 @@ void client(event_based_actor* self, const actor& parent,
struct get_state_msg {};
using event_testee_type = typed_actor<
replies_to<get_state_msg>::with<string>, replies_to<string>::with<void>,
replies_to<float>::with<void>, replies_to<int>::with<int>>;
class event_testee : public event_testee_type::base {
public:
event_testee(actor_config& cfg) : event_testee_type::base(cfg) {
......@@ -170,8 +197,6 @@ public:
* simple 'forwarding' chain *
******************************************************************************/
using string_actor = typed_actor<replies_to<string>::with<string>>;
string_actor::behavior_type string_reverter() {
return {
[](string& str) -> string {
......@@ -222,10 +247,6 @@ maybe_string_delegator(maybe_string_actor::pointer self,
* sending typed actor handles *
******************************************************************************/
using int_actor = typed_actor<replies_to<int>::with<int>>;
using float_actor = typed_actor<reacts_to<float>>;
int_actor::behavior_type int_fun() {
return {
[](int i) { return i * i; },
......@@ -286,7 +307,7 @@ struct fixture {
scoped_actor self;
static actor_system_config& init(actor_system_config& cfg) {
cfg.add_message_type<get_state_msg>("get_state_msg");
cfg.add_message_types<id_block::typed_spawn>();
cfg.parse(test::engine::argc(), test::engine::argv());
return cfg;
}
......
......@@ -90,3 +90,41 @@ endif ()
install(DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/caf
DESTINATION ${CMAKE_INSTALL_INCLUDEDIR}
FILES_MATCHING PATTERN "*.hpp")
# -- build unit tests ----------------------------------------------------------
if(CAF_NO_UNIT_TESTS)
return()
endif()
add_executable(caf-io-test test/io-test.cpp)
target_include_directories(caf-io-test PRIVATE
"${CMAKE_CURRENT_SOURCE_DIR}/test")
if (CAF_BUILD_STATIC_ONLY OR CAF_BUILD_STATIC)
target_link_libraries(caf-io-test
libcaf_io_static
libcaf_core_static
${CAF_EXTRA_LDFLAGS})
else()
target_link_libraries(caf-io-test
libcaf_io_shared
libcaf_core_shared
${CAF_EXTRA_LDFLAGS})
endif()
caf_add_test_suites(caf-io-test
io.basp.message_queue
io.basp_broker
io.broker
io.http_broker
io.network.default_multiplexer
io.network.ip_endpoint
io.receive_buffer
io.remote_actor
io.remote_group
io.remote_spawn
io.unpublish
io.worker
)
......@@ -58,8 +58,3 @@ using doorman_ptr = intrusive_ptr<doorman>;
} // namespace io
} // namespace caf
// Allows the `middleman_actor` to create a `doorman` and then send it to the
// BASP broker.
CAF_ALLOW_UNSAFE_MESSAGE_TYPE(caf::io::doorman_ptr)
......@@ -18,6 +18,18 @@
#pragma once
#include <map>
#include <string>
#include <vector>
#include "caf/allowed_unsafe_message_type.hpp"
#include "caf/type_id.hpp"
// Unfortunately required, because we cannot add a forward declaration for the
// enum protol::network that we need for assigning a type ID to
// io::network::address_listing.
#include "caf/io/network/protocol.hpp"
namespace caf {
// -- templates from the parent namespace necessary for defining aliases -------
......@@ -33,14 +45,30 @@ class typed_broker;
// -- classes ------------------------------------------------------------------
class scribe;
class abstract_broker;
class accept_handle;
class basp_broker;
class broker;
class connection_handle;
class datagram_servant;
class doorman;
class middleman;
class basp_broker;
class receive_policy;
class abstract_broker;
class datagram_servant;
class scribe;
// -- structs ------------------------------------------------------------------
struct acceptor_closed_msg;
struct acceptor_passivated_msg;
struct connection_closed_msg;
struct connection_passivated_msg;
struct data_transferred_msg;
struct datagram_sent_msg;
struct datagram_servant_closed_msg;
struct datagram_servant_passivated_msg;
struct new_connection_msg;
struct new_data_msg;
struct new_datagram_msg;
// -- aliases ------------------------------------------------------------------
......@@ -52,11 +80,43 @@ using datagram_servant_ptr = intrusive_ptr<datagram_servant>;
namespace network {
class multiplexer;
class default_multiplexer;
class multiplexer;
class receive_buffer;
using address_listing = std::map<protocol::network, std::vector<std::string>>;
} // namespace network
} // namespace io
} // namespace caf
CAF_BEGIN_TYPE_ID_BLOCK(io_module, detail::io_module_begin)
CAF_ADD_TYPE_ID(io_module, (caf::io::accept_handle))
CAF_ADD_TYPE_ID(io_module, (caf::io::acceptor_closed_msg))
CAF_ADD_TYPE_ID(io_module, (caf::io::acceptor_passivated_msg))
CAF_ADD_TYPE_ID(io_module, (caf::io::connection_closed_msg))
CAF_ADD_TYPE_ID(io_module, (caf::io::connection_handle))
CAF_ADD_TYPE_ID(io_module, (caf::io::connection_passivated_msg))
CAF_ADD_TYPE_ID(io_module, (caf::io::data_transferred_msg))
CAF_ADD_TYPE_ID(io_module, (caf::io::datagram_sent_msg))
CAF_ADD_TYPE_ID(io_module, (caf::io::datagram_servant_closed_msg))
CAF_ADD_TYPE_ID(io_module, (caf::io::datagram_servant_passivated_msg))
CAF_ADD_TYPE_ID(io_module, (caf::io::datagram_servant_ptr))
CAF_ADD_TYPE_ID(io_module, (caf::io::doorman_ptr))
CAF_ADD_TYPE_ID(io_module, (caf::io::network::address_listing))
CAF_ADD_TYPE_ID(io_module, (caf::io::network::protocol))
CAF_ADD_TYPE_ID(io_module, (caf::io::network::receive_buffer))
CAF_ADD_TYPE_ID(io_module, (caf::io::new_connection_msg))
CAF_ADD_TYPE_ID(io_module, (caf::io::new_data_msg))
CAF_ADD_TYPE_ID(io_module, (caf::io::new_datagram_msg))
CAF_ADD_TYPE_ID(io_module, (caf::io::scribe_ptr))
CAF_END_TYPE_ID_BLOCK(io_module)
CAF_ALLOW_UNSAFE_MESSAGE_TYPE(caf::io::doorman_ptr)
CAF_ALLOW_UNSAFE_MESSAGE_TYPE(caf::io::scribe_ptr)
static_assert(caf::id_block::io_module::end == caf::detail::io_module_end,
"caf::id_block::io_module::end != caf::detail::io_module_end");
......@@ -69,8 +69,3 @@ using scribe_ptr = intrusive_ptr<scribe>;
} // namespace io
} // namespace caf
// Allows the `middleman_actor` to create a `scribe` and then send it to the
// BASP broker.
CAF_ALLOW_UNSAFE_MESSAGE_TYPE(caf::io::scribe_ptr)
......@@ -37,6 +37,9 @@
#include "caf/sec.hpp"
#include "caf/send.hpp"
CAF_ALLOW_UNSAFE_MESSAGE_TYPE(caf::actor_clock::duration_type)
CAF_ALLOW_UNSAFE_MESSAGE_TYPE(caf::actor_clock::time_point)
namespace {
#ifdef CAF_MSVC
......
#include "io-test.hpp"
#include "caf/test/unit_test_impl.hpp"
#include "caf/test/io_dsl.hpp"
using calculator = caf::typed_actor<
caf::replies_to<caf::add_atom, int32_t, int32_t>::with<int32_t>,
caf::replies_to<caf::sub_atom, int32_t, int32_t>::with<int32_t>>;
CAF_BEGIN_TYPE_ID_BLOCK(io_test, caf::first_custom_type_id)
CAF_ADD_TYPE_ID(io_test, (calculator))
CAF_END_TYPE_ID_BLOCK(io_test)
......@@ -18,9 +18,7 @@
#define CAF_SUITE io.remote_spawn
#include "caf/config.hpp"
#include "caf/test/io_dsl.hpp"
#include "io-test.hpp"
#include <cstring>
#include <functional>
......@@ -36,17 +34,11 @@ using namespace caf;
namespace {
using add_atom = atom_constant<atom("add")>;
using sub_atom = atom_constant<atom("sub")>;
using calculator = typed_actor<replies_to<add_atom, int, int>::with<int>,
replies_to<sub_atom, int, int>::with<int>>;
// function-based, dynamically typed, event-based API
behavior calculator_fun(event_based_actor*) {
return {
[](add_atom, int a, int b) { return a + b; },
[](sub_atom, int a, int b) { return a - b; },
[](add_atom, int32_t a, int32_t b) { return a + b; },
[](sub_atom, int32_t a, int32_t b) { return a - b; },
};
}
......@@ -58,8 +50,8 @@ public:
behavior make_behavior() override {
return {
[](add_atom, int a, int b) { return a + b; },
[](sub_atom, int a, int b) { return a - b; },
[](add_atom, int32_t a, int32_t b) { return a + b; },
[](sub_atom, int32_t a, int32_t b) { return a - b; },
};
}
};
......@@ -67,14 +59,15 @@ public:
// function-based, statically typed, event-based API
calculator::behavior_type typed_calculator_fun() {
return {
[](add_atom, int a, int b) { return a + b; },
[](sub_atom, int a, int b) { return a - b; },
[](add_atom, int32_t a, int32_t b) { return a + b; },
[](sub_atom, int32_t a, int32_t b) { return a - b; },
};
}
struct config : actor_system_config {
config() {
load<io::middleman>();
add_message_types<id_block::io_test>();
add_actor_type<calculator_class>("calculator-class");
add_actor_type("calculator", calculator_fun);
add_actor_type("typed_calculator", typed_calculator_fun);
......
......@@ -25,6 +25,8 @@ namespace opencl {
class nd_range {
public:
nd_range() = default;
nd_range(const opencl::dim_vec& dimensions,
const opencl::dim_vec& offsets = {},
const opencl::dim_vec& local_dimensions = {})
......@@ -61,6 +63,11 @@ public:
return local_dims_;
}
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f, nd_range& x) {
return f(x.dims_, x.offset_, x.local_dims_);
}
private:
opencl::dim_vec dims_;
opencl::dim_vec offset_;
......@@ -69,4 +76,3 @@ private:
} // namespace opencl
} // namespace caf
......@@ -27,6 +27,18 @@
#include "caf/opencl/all.hpp"
template <size_t Size>
class square_matrix;
constexpr size_t matrix_size = 8;
CAF_BEGIN_TYPE_ID_BLOCK(proper_matrix, first_custom_type_id)
CAF_ADD_TYPE_ID(proper_matrix, (square_matrix<matrix_size>) )
CAF_ADD_TYPE_ID(proper_matrix, (std::vector<float>) )
CAF_END_TYPE_ID_BLOCK(proper_matrix)
using namespace caf;
using namespace caf::opencl;
......@@ -41,7 +53,6 @@ namespace {
using fvec = vector<float>;
constexpr size_t matrix_size = 8;
constexpr const char* kernel_name = "matrix_mult";
// opencl kernel, multiplies matrix1 and matrix2
......@@ -196,9 +207,7 @@ int main() {
// matrix_type ist not a simple type,
// it must be annouced to libcaf
actor_system_config cfg;
cfg.load<opencl::manager>()
.add_message_type<fvec>("float_vector")
.add_message_type<matrix_type>("square_matrix");
cfg.load<opencl::manager>().add_message_types<id_block::proper_matrix>();
actor_system system{cfg};
system.spawn(multiplier);
system.await_all_actors_done();
......
......@@ -26,6 +26,14 @@
#include "caf/all.hpp"
#include "caf/opencl/all.hpp"
CAF_BEGIN_TYPE_ID_BLOCK(scan, first_custom_type_id)
CAF_ADD_TYPE_ID(scan, (caf::opencl::dim_vec))
CAF_ADD_TYPE_ID(scan, (caf::opencl::nd_range))
CAF_ADD_TYPE_ID(scan, (std::vector<uint32_t>))
CAF_END_TYPE_ID_BLOCK(scan)
using namespace caf;
using namespace caf::opencl;
......@@ -39,7 +47,7 @@ using caf::detail::limited_vector;
namespace {
using uval = unsigned;
using uval = uint32_t;
using uvec = std::vector<uval>;
using uref = mem_ref<uval>;
......@@ -178,8 +186,7 @@ T round_up(T numToRound, T multiple) {
int main() {
actor_system_config cfg;
cfg.load<opencl::manager>()
.add_message_type<uvec>("uint_vector");
cfg.load<opencl::manager>().add_message_types<id_block::scan>();
actor_system system{cfg};
cout << "Calculating exclusive scan of '" << problem_size
<< "' values." << endl;
......
......@@ -24,6 +24,13 @@
#include "caf/all.hpp"
#include "caf/opencl/all.hpp"
CAF_BEGIN_TYPE_ID_BLOCK(simple_matrix, first_custom_type_id)
CAF_ADD_TYPE_ID(simple_matrix, (std::vector<float>) )
CAF_END_TYPE_ID_BLOCK(simple_matrix)
using namespace std;
using namespace caf;
using namespace caf::opencl;
......@@ -109,8 +116,7 @@ void multiplier(event_based_actor* self) {
int main() {
actor_system_config cfg;
cfg.load<opencl::manager>()
.add_message_type<fvec>("float_vector");
cfg.load<opencl::manager>().add_message_types<id_block::simple_matrix>();
actor_system system{cfg};
system.spawn(multiplier);
system.await_all_actors_done();
......
#define CAF_SUITE opencl
#include "caf/test/unit_test.hpp"
#include <vector>
#include <iomanip>
#include <cassert>
#include <iostream>
#include <algorithm>
#include "caf/all.hpp"
#include "caf/system_messages.hpp"
#include "caf/opencl/all.hpp"
using namespace caf;
using namespace caf::opencl;
using std::cout;
using std::endl;
using std::string;
using std::vector;
using caf::detail::tl_at;
using caf::detail::tl_head;
using caf::detail::type_list;
using caf::detail::limited_vector;
namespace {
using ivec = vector<int>;
using iref = mem_ref<int>;
using dims = opencl::dim_vec;
constexpr size_t matrix_size = 4;
constexpr size_t array_size = 32;
constexpr size_t problem_size = 1024;
constexpr const char* kn_matrix = "matrix_square";
constexpr const char* kn_compiler_flag = "compiler_flag";
constexpr const char* kn_reduce = "reduce";
constexpr const char* kn_const = "const_mod";
constexpr const char* kn_inout = "times_two";
constexpr const char* kn_scratch = "use_scratch";
constexpr const char* kn_local = "use_local";
constexpr const char* kn_order = "test_order";
constexpr const char* kn_private = "use_private";
constexpr const char* kn_varying = "varying";
constexpr const char* compiler_flag = "-D CAF_OPENCL_TEST_FLAG";
constexpr const char* kernel_source = R"__(
kernel void matrix_square(global const int* restrict matrix,
global int* restrict output) {
size_t size = get_global_size(0); // == get_global_size_(1);
size_t x = get_global_id(0);
size_t y = get_global_id(1);
int result = 0;
for (size_t idx = 0; idx < size; ++idx) {
result += matrix[idx + y * size] * matrix[x + idx * size];
}
output[x + y * size] = result;
}
// http://developer.amd.com/resources/documentation-articles/
// articles-whitepapers/opencl-optimization-case-study-simple-reductions
kernel void reduce(global const int* restrict buffer,
global int* restrict result) {
local int scratch[512];
int local_index = get_local_id(0);
scratch[local_index] = buffer[get_global_id(0)];
barrier(CLK_LOCAL_MEM_FENCE);
for(int offset = get_local_size(0) / 2; offset > 0; offset = offset / 2) {
if (local_index < offset) {
int other = scratch[local_index + offset];
int mine = scratch[local_index];
scratch[local_index] = (mine < other) ? mine : other;
}
barrier(CLK_LOCAL_MEM_FENCE);
}
if (local_index == 0)
result[get_group_id(0)] = scratch[0];
}
kernel void const_mod(constant int* restrict input,
global int* restrict output) {
size_t idx = get_global_id(0);
output[idx] = input[0];
}
kernel void times_two(global int* restrict values) {
size_t idx = get_global_id(0);
values[idx] = values[idx] * 2;
}
kernel void use_scratch(global int* restrict values,
global int* restrict buf) {
size_t idx = get_global_id(0);
buf[idx] = values[idx];
buf[idx] += values[idx];
values[idx] = buf[idx];
}
inline void prefix_sum(local int* restrict data, size_t len, size_t lids) {
size_t lid = get_local_id(0);
size_t inc = 2;
// reduce
while (inc <= len) {
int j = inc >> 1;
for (int i = (j - 1) + (lid * inc); (i + inc) < len; i += (lids * inc))
data[i + j] = data[i] + data[i + j];
inc = inc << 1;
barrier(CLK_LOCAL_MEM_FENCE);
}
// downsweep
data[len - 1] = 0;
barrier(CLK_LOCAL_MEM_FENCE);
while (inc >= 2) {
int j = inc >> 1;
for (int i = (j - 1) + (lid * inc); (i + j) <= len; i += (lids * inc)) {
uint tmp = data[i + j];
data[i + j] = data[i] + data[i + j];
data[i] = tmp;
}
inc = inc >> 1;
barrier(CLK_LOCAL_MEM_FENCE);
}
}
kernel void use_local(global int* restrict values,
local int* restrict buf) {
size_t lid = get_local_id(0);
size_t gid = get_group_id(0);
size_t gs = get_local_size(0);
buf[lid] = values[gid * gs + lid];
barrier(CLK_LOCAL_MEM_FENCE);
prefix_sum(buf, gs, gs);
barrier(CLK_LOCAL_MEM_FENCE);
values[gid * gs + lid] = buf[lid];
}
kernel void test_order(local int* buf,
global int* restrict values) {
size_t lid = get_local_id(0);
size_t gid = get_group_id(0);
size_t gs = get_local_size(0);
buf[lid] = values[gid * gs + lid];
barrier(CLK_LOCAL_MEM_FENCE);
prefix_sum(buf, gs, gs);
barrier(CLK_LOCAL_MEM_FENCE);
values[gid * gs + lid] = buf[lid];
}
kernel void use_private(global int* restrict buf,
private int val) {
buf[get_global_id(0)] += val;
}
kernel void varying(global const int* restrict in1,
global int* restrict out1,
global const int* restrict in2,
global int* restrict out2) {
size_t idx = get_global_id(0);
out1[idx] = in1[idx];
out2[idx] = in2[idx];
}
)__";
#ifndef CAF_NO_EXCEPTIONS
constexpr const char* kernel_source_error = R"__(
kernel void missing(global int*) {
size_t semicolon_missing
}
)__";
#endif // CAF_NO_EXCEPTIONS
constexpr const char* kernel_source_compiler_flag = R"__(
kernel void compiler_flag(global const int* restrict input,
global int* restrict output) {
size_t x = get_global_id(0);
# ifdef CAF_OPENCL_TEST_FLAG
output[x] = input[x];
# else
output[x] = 0;
# endif
}
)__";
} // namespace
template<size_t Size>
class square_matrix {
public:
using value_type = ivec::value_type;
static constexpr size_t num_elements = Size * Size;
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f,
square_matrix& x) {
return f(meta::type_name("square_matrix"), x.data_);
}
square_matrix(square_matrix&&) = default;
square_matrix(const square_matrix&) = default;
square_matrix& operator=(square_matrix&&) = default;
square_matrix& operator=(const square_matrix&) = default;
square_matrix() : data_(num_elements) {
// nop
}
explicit square_matrix(ivec d) : data_(move(d)) {
assert(data_.size() == num_elements);
}
int& operator()(size_t column, size_t row) {
return data_[column + row * Size];
}
const int& operator()(size_t column, size_t row) const {
return data_[column + row * Size];
}
using const_iterator = typename ivec::const_iterator;
const_iterator begin() const {
return data_.begin();
}
const_iterator end() const {
return data_.end();
}
ivec& data() {
return data_;
}
const ivec& data() const {
return data_;
}
void data(ivec new_data) {
data_ = move(new_data);
}
private:
ivec data_;
};
template <class T>
vector<T> make_iota_vector(size_t num_elements) {
vector<T> result;
result.resize(num_elements);
iota(result.begin(), result.end(), T{0});
return result;
}
template <size_t Size>
square_matrix<Size> make_iota_matrix() {
square_matrix<Size> result;
iota(result.data().begin(), result.data().end(), 0);
return result;
}
template<size_t Size>
bool operator==(const square_matrix<Size>& lhs,
const square_matrix<Size>& rhs) {
return lhs.data() == rhs.data();
}
template<size_t Size>
bool operator!=(const square_matrix<Size>& lhs,
const square_matrix<Size>& rhs) {
return !(lhs == rhs);
}
using matrix_type = square_matrix<matrix_size>;
template <class T>
void check_vector_results(const string& description,
const vector<T>& expected,
const vector<T>& result) {
auto cond = (expected == result);
CAF_CHECK(cond);
if (!cond) {
CAF_ERROR(description << " failed.");
cout << "Expected: " << endl;
for (size_t i = 0; i < expected.size(); ++i) {
cout << " " << expected[i];
}
cout << endl << "Received: " << endl;
for (size_t i = 0; i < result.size(); ++i) {
cout << " " << result[i];
}
cout << endl;
cout << "Size: " << expected.size() << " vs. " << result.size() << endl;
cout << "Differ at: " << endl;
bool same = true;
for (size_t i = 0; i < std::min(expected.size(), result.size()); ++i) {
if (expected[i] != result[i]) {
cout << "[" << i << "] " << expected[i] << " != " << result[i] << endl;
same = false;
}
}
if (same) {
cout << "... nowhere." << endl;
}
}
}
template <class T>
void check_mref_results(const string& description,
const vector<T>& expected,
mem_ref<T>& result) {
auto exp_res = result.data();
CAF_REQUIRE(exp_res);
auto res = *exp_res;
auto cond = (expected == res);
CAF_CHECK(cond);
if (!cond) {
CAF_ERROR(description << " failed.");
cout << "Expected: " << endl;
for (size_t i = 0; i < expected.size(); ++i) {
cout << " " << expected[i];
}
cout << endl << "Received: " << endl;
for (size_t i = 0; i < res.size(); ++i) {
cout << " " << res[i];
}
cout << endl;
}
}
void test_opencl(actor_system& sys) {
auto& mngr = sys.opencl_manager();
auto opt = mngr.find_device_if([](const device_ptr){ return true; });
CAF_REQUIRE(opt);
auto dev = *opt;
auto prog = mngr.create_program(kernel_source, "", dev);
scoped_actor self{sys};
auto wrong_msg = [&](message_view& x) -> result<message> {
CAF_ERROR("unexpected message" << x.content().stringify());
return sec::unexpected_message;
};
const ivec expected1{ 56, 62, 68, 74,
152, 174, 196, 218,
248, 286, 324, 362,
344, 398, 452, 506};
auto w1 = mngr.spawn(prog, kn_matrix,
opencl::nd_range{dims{matrix_size, matrix_size}},
opencl::in<int>{}, opencl::out<int>{});
self->send(w1, make_iota_vector<int>(matrix_size * matrix_size));
self->receive (
[&](const ivec& result) {
check_vector_results("Simple matrix multiplication using vectors"
" (kernel wrapped in program)",
expected1, result);
}, others >> wrong_msg
);
opencl::nd_range range2{dims{matrix_size, matrix_size}};
// Pass kernel directly to the actor
auto w2 = mngr.spawn(kernel_source, kn_matrix, range2,
opencl::in<int>{}, opencl::out<int>{});
self->send(w2, make_iota_vector<int>(matrix_size * matrix_size));
self->receive (
[&](const ivec& result) {
check_vector_results("Simple matrix multiplication using vectors"
" (kernel passed directly)",
expected1, result);
}, others >> wrong_msg
);
const matrix_type expected2(move(expected1));
auto map_arg = [](message& msg) -> optional<message> {
return msg.apply(
[](matrix_type& mx) {
return make_message(move(mx.data()));
}
);
};
auto map_res = [](ivec result) -> message {
return make_message(matrix_type{move(result)});
};
opencl::nd_range range3{dims{matrix_size, matrix_size}};
// let the runtime choose the device
auto w3 = mngr.spawn(mngr.create_program(kernel_source), kn_matrix, range3,
map_arg, map_res,
opencl::in<int>{}, opencl::out<int>{});
self->send(w3, make_iota_matrix<matrix_size>());
self->receive (
[&](const matrix_type& result) {
check_vector_results("Matrix multiplication with user defined type "
"(kernel wrapped in program)",
expected2.data(), result.data());
}, others >> wrong_msg
);
opencl::nd_range range4{dims{matrix_size, matrix_size}};
auto w4 = mngr.spawn(prog, kn_matrix, range4,
map_arg, map_res,
opencl::in<int>{}, opencl::out<int>{});
self->send(w4, make_iota_matrix<matrix_size>());
self->receive (
[&](const matrix_type& result) {
check_vector_results("Matrix multiplication with user defined type",
expected2.data(), result.data());
}, others >> wrong_msg
);
#ifndef CAF_NO_EXCEPTIONS
CAF_MESSAGE("Expecting exception (compiling invalid kernel, "
"semicolon is missing).");
try {
/* auto expected_error = */ mngr.create_program(kernel_source_error);
} catch (const std::exception& exc) {
CAF_MESSAGE("got: " << exc.what());
}
#endif // CAF_NO_EXCEPTIONS
// create program with opencl compiler flags
auto prog5 = mngr.create_program(kernel_source_compiler_flag, compiler_flag);
opencl::nd_range range5{dims{array_size}};
auto w5 = mngr.spawn(prog5, kn_compiler_flag, range5,
opencl::in<int>{}, opencl::out<int>{});
self->send(w5, make_iota_vector<int>(array_size));
auto expected3 = make_iota_vector<int>(array_size);
self->receive (
[&](const ivec& result) {
check_vector_results("Passing compiler flags", expected3, result);
}, others >> wrong_msg
);
// test for manuel return size selection (max workgroup size 1d)
auto max_wg_size = std::min(dev->max_work_item_sizes()[0], size_t{512});
auto reduce_buffer_size = static_cast<size_t>(max_wg_size) * 8;
auto reduce_local_size = static_cast<size_t>(max_wg_size);
auto reduce_work_groups = reduce_buffer_size / reduce_local_size;
auto reduce_global_size = reduce_buffer_size;
auto reduce_result_size = reduce_work_groups;
ivec arr6(reduce_buffer_size);
int n = static_cast<int>(arr6.capacity());
generate(arr6.begin(), arr6.end(), [&]{ return --n; });
opencl::nd_range range6{dims{reduce_global_size},
dims{ }, // no offset
dims{reduce_local_size}};
auto result_size_6 = [reduce_result_size](const ivec&) {
return reduce_result_size;
};
auto w6 = mngr.spawn(prog, kn_reduce, range6,
opencl::in<int>{}, opencl::out<int>{result_size_6});
self->send(w6, move(arr6));
auto wg_size_as_int = static_cast<int>(max_wg_size);
ivec expected4{wg_size_as_int * 7, wg_size_as_int * 6, wg_size_as_int * 5,
wg_size_as_int * 4, wg_size_as_int * 3, wg_size_as_int * 2,
wg_size_as_int , 0};
self->receive(
[&](const ivec& result) {
check_vector_results("Passing size for the output", expected4, result);
}, others >> wrong_msg
);
// calculator function for getting the size of the output
auto result_size_7 = [](const ivec&) {
return problem_size;
};
// constant memory arguments
const ivec arr7{static_cast<int>(problem_size)};
auto w7 = mngr.spawn(kernel_source, kn_const,
opencl::nd_range{dims{problem_size}},
opencl::in<int>{},
opencl::out<int>{result_size_7});
self->send(w7, move(arr7));
ivec expected5(problem_size);
fill(begin(expected5), end(expected5), static_cast<int>(problem_size));
self->receive(
[&](const ivec& result) {
check_vector_results("Using const input argument", expected5, result);
}, others >> wrong_msg
);
}
void test_arguments(actor_system& sys) {
auto& mngr = sys.opencl_manager();
auto opt = mngr.find_device_if([](const device_ptr){ return true; });
CAF_REQUIRE(opt);
auto dev = *opt;
scoped_actor self{sys};
auto wrong_msg = [&](message_view& x) -> result<message> {
CAF_ERROR("unexpected message" << x.content().stringify());
return sec::unexpected_message;
};
const ivec expected1{ 56, 62, 68, 74, 152, 174, 196, 218,
248, 286, 324, 362, 344, 398, 452, 506};
auto w1 = mngr.spawn(mngr.create_program(kernel_source, "", dev), kn_matrix,
opencl::nd_range{dims{matrix_size, matrix_size}},
opencl::in<int>{}, opencl::out<int>{});
self->send(w1, make_iota_vector<int>(matrix_size * matrix_size));
self->receive (
[&](const ivec& result) {
check_vector_results("arguments: from in to out", expected1, result);
}, others >> wrong_msg
);
ivec input9 = make_iota_vector<int>(problem_size);
ivec expected9{input9};
for_each(begin(expected9), end(expected9), [](int& val){ val *= 2; });
auto w9 = mngr.spawn(kernel_source, kn_inout,
nd_range{dims{problem_size}},
opencl::in_out<int>{});
self->send(w9, move(input9));
self->receive(
[&](const ivec& result) {
check_vector_results("Testing in_out arugment", expected9, result);
}, others >> wrong_msg
);
ivec input10 = make_iota_vector<int>(problem_size);
ivec expected10{input10};
for_each(begin(expected10), end(expected10), [](int& val){ val *= 2; });
auto result_size_10 = [=](const ivec& input) { return input.size(); };
auto w10 = mngr.spawn(kernel_source, kn_scratch,
nd_range{dims{problem_size}},
opencl::in_out<int>{},
opencl::scratch<int>{result_size_10});
self->send(w10, move(input10));
self->receive(
[&](const ivec& result) {
check_vector_results("Testing buffer arugment", expected10, result);
}, others >> wrong_msg
);
// test local
size_t la_global = 256;
size_t la_local = la_global / 2;
ivec input_local = make_iota_vector<int>(la_global);
ivec expected_local{input_local};
auto last = 0;
for (size_t i = 0; i < la_global; ++i) {
if (i == la_local) last = 0;
auto tmp = expected_local[i];
expected_local[i] = last;
last += tmp;
}
auto work_local = mngr.spawn(kernel_source, kn_local,
nd_range{dims{la_global}, {}, dims{la_local}},
opencl::in_out<int>{},
opencl::local<int>{la_local});
self->send(work_local, std::move(input_local));
self->receive(
[&](const ivec& result) {
check_vector_results("Testing local arugment", expected_local, result);
}
);
// Same test, different argument order
input_local = make_iota_vector<int>(la_global);
work_local = mngr.spawn(kernel_source, kn_order,
nd_range{dims{la_global}, {}, dims{la_local}},
opencl::local<int>{la_local},
opencl::in_out<int>{});
self->send(work_local, std::move(input_local));
self->receive(
[&](const ivec& result) {
check_vector_results("Testing local arugment", expected_local, result);
}
);
// Test private argument
ivec input_private = make_iota_vector<int>(problem_size);
int val_private = 42;
ivec expected_private{input_private};
for_each(begin(expected_private), end(expected_private),
[val_private](int& val){ val += val_private; });
auto worker_private = mngr.spawn(kernel_source, kn_private,
nd_range{dims{problem_size}},
opencl::in_out<int>{},
opencl::priv<int>{val_private});
self->send(worker_private, std::move(input_private));
self->receive(
[&](const ivec& result) {
check_vector_results("Testing private arugment", expected_private,
result);
}
);
}
CAF_TEST(opencl_basics) {
actor_system_config cfg;
cfg.load<opencl::manager>()
.add_message_type<ivec>("int_vector")
.add_message_type<matrix_type>("square_matrix");
actor_system system{cfg};
test_opencl(system);
system.await_all_actors_done();
}
CAF_TEST(opencl_arguments) {
actor_system_config cfg;
cfg.load<opencl::manager>()
.add_message_type<ivec>("int_vector")
.add_message_type<matrix_type>("square_matrix");
actor_system system{cfg};
test_arguments(system);
system.await_all_actors_done();
}
CAF_TEST(opencl_mem_refs) {
actor_system_config cfg;
cfg.load<opencl::manager>();
actor_system system{cfg};
auto& mngr = system.opencl_manager();
auto opt = mngr.find_device(0);
CAF_REQUIRE(opt);
auto dev = *opt;
// global arguments
vector<uint32_t> input{1, 2, 3, 4};
auto buf_1 = dev->global_argument(input, buffer_type::input_output);
CAF_CHECK_EQUAL(buf_1.size(), input.size());
auto res_1 = buf_1.data();
CAF_CHECK(res_1);
CAF_CHECK_EQUAL(res_1->size(), input.size());
check_vector_results("Testing mem_ref", input, *res_1);
auto res_2 = buf_1.data(2ul);
CAF_CHECK(res_2);
CAF_CHECK_EQUAL(res_2->size(), 2ul);
CAF_CHECK_EQUAL((*res_2)[0], input[0]);
CAF_CHECK_EQUAL((*res_2)[1], input[1]);
vector<uint32_t> new_input{1,2,3,4,5};
buf_1 = dev->global_argument(new_input, buffer_type::input_output);
CAF_CHECK_EQUAL(buf_1.size(), new_input.size());
auto res_3 = buf_1.data();
CAF_CHECK(res_3);
mem_ref<uint32_t> buf_2{std::move(buf_1)};
CAF_CHECK_EQUAL(buf_2.size(), new_input.size());
auto res_4 = buf_2.data();
CAF_CHECK(res_4);
buf_2.reset();
auto res_5 = buf_2.data();
CAF_CHECK(!res_5);
}
CAF_TEST(opencl_argument_info) {
using base_t = int;
using in_arg_t = ::type_list<opencl::in<base_t>>;
using in_arg_info_t = typename cl_arg_info_list<in_arg_t>::type;
using in_arg_wrap_t = typename ::tl_head<in_arg_info_t>::type;
static_assert(in_arg_wrap_t::in_pos == 0, "In-index for `in` wrong.");
static_assert(in_arg_wrap_t::out_pos == -1, "Out-index for `in` wrong.");
using out_arg_t = ::type_list<opencl::out<base_t>>;
using out_arg_info_t = typename cl_arg_info_list<out_arg_t>::type;
using out_arg_wrap_t = typename ::tl_head<out_arg_info_t>::type;
static_assert(out_arg_wrap_t::in_pos == -1, "In-index for `out` wrong.");
static_assert(out_arg_wrap_t::out_pos == 0, "Out-index for `out` wrong.");
using io_arg_t = ::type_list<opencl::in_out<base_t>>;
using io_arg_info_t = typename cl_arg_info_list<io_arg_t>::type;
using io_arg_wrap_t = typename ::tl_head<io_arg_info_t>::type;
static_assert(io_arg_wrap_t::in_pos == 0, "In-index for `in_out` wrong.");
static_assert(io_arg_wrap_t::out_pos == 0, "Out-index for `in_out` wrong.");
using arg_list_t = ::type_list<opencl::in<base_t>,
opencl::out<base_t>,
opencl::local<base_t>,
opencl::in_out<base_t>,
opencl::priv<base_t>,
opencl::priv<base_t, val>>;
using arg_info_list_t = typename cl_arg_info_list<arg_list_t>::type;
using arg_info_0_t = typename ::tl_at<arg_info_list_t,0>::type;
static_assert(arg_info_0_t::in_pos == 0, "In-index for `in` wrong.");
static_assert(arg_info_0_t::out_pos == -1, "Out-index for `in` wrong.");
using arg_info_1_t = typename ::tl_at<arg_info_list_t,1>::type;
static_assert(arg_info_1_t::in_pos == -1, "In-index for `out` wrong.");
static_assert(arg_info_1_t::out_pos == 0, "Out-index for `out` wrong.");
using arg_info_2_t = typename ::tl_at<arg_info_list_t,2>::type;
static_assert(arg_info_2_t::in_pos == -1, "In-index for `local` wrong.");
static_assert(arg_info_2_t::out_pos == -1, "Out-index for `local` wrong.");
using arg_info_3_t = typename ::tl_at<arg_info_list_t,3>::type;
static_assert(arg_info_3_t::in_pos == 1, "In-index for `in_out` wrong.");
static_assert(arg_info_3_t::out_pos == 1, "Out-index for `in_out` wrong.");
using arg_info_4_t = typename ::tl_at<arg_info_list_t,4>::type;
static_assert(arg_info_4_t::in_pos == -1, "In-index for `priv` wrong.");
static_assert(arg_info_4_t::out_pos == -1, "Out-index for `priv` wrong.");
using arg_info_5_t = typename ::tl_at<arg_info_list_t,5>::type;
static_assert(arg_info_5_t::in_pos == 2, "In-index for `priv` wrong.");
static_assert(arg_info_5_t::out_pos == -1, "Out-index for `priv` wrong.");
// gives the test some output.
CAF_CHECK_EQUAL(true, true);
}
void test_in_val_out_val(actor_system& sys) {
CAF_MESSAGE("Testing in: val -> out: val ");
auto& mngr = sys.opencl_manager();
auto opt = mngr.find_device(0);
CAF_REQUIRE(opt);
auto dev = *opt;
auto prog = mngr.create_program(kernel_source, "", dev);
scoped_actor self{sys};
auto wrong_msg = [&](message_view& x) -> result<message> {
CAF_ERROR("unexpected message" << x.content().stringify());
return sec::unexpected_message;
};
const ivec res1{ 56, 62, 68, 74, 152, 174, 196, 218,
248, 286, 324, 362, 344, 398, 452, 506};
auto conf = opencl::nd_range{dims{matrix_size, matrix_size}};
auto w1 = mngr.spawn(prog, kn_matrix, conf, in<int>{}, out<int>{});
self->send(w1, make_iota_vector<int>(matrix_size * matrix_size));
self->receive([&](const ivec& result) {
check_vector_results("Simple matrix multiplication using vectors"
" (kernel wrapped in program)", res1, result);
}, others >> wrong_msg);
// Pass kernel directly to the actor
auto w2 = mngr.spawn(kernel_source, kn_matrix, conf,
in<int>{}, out<int>{});
self->send(w2, make_iota_vector<int>(matrix_size * matrix_size));
self->receive([&](const ivec& result) {
check_vector_results("Simple matrix multiplication using vectors"
" (kernel passed directly)", res1, result);
}, others >> wrong_msg);
// Wrap message in user-defined type and use mapping functions
const matrix_type res2(move(res1));
auto map_arg = [](message& msg) -> optional<message> {
return msg.apply([](matrix_type& mx) {
return make_message(move(mx.data()));
});
};
auto map_res = [](ivec result) -> message {
return make_message(matrix_type{move(result)});
};
auto w3 = mngr.spawn(prog, kn_matrix, conf, map_arg, map_res,
in<int, val>{}, out<int, val>{});
self->send(w3, make_iota_matrix<matrix_size>());
self->receive([&](const matrix_type& result) {
check_vector_results("Matrix multiplication with user defined type "
"(kernel wrapped in program)",
res2.data(), result.data());
}, others >> wrong_msg);
// create program with opencl compiler flags
auto prog2 = mngr.create_program(kernel_source_compiler_flag, compiler_flag);
nd_range range2{dims{array_size}};
auto w4 = mngr.spawn(prog2, kn_compiler_flag, range2,
in<int>{}, out<int>{});
self->send(w4, make_iota_vector<int>(array_size));
auto res3 = make_iota_vector<int>(array_size);
self->receive([&](const ivec& result) {
check_vector_results("Passing compiler flags", res3, result);
}, others >> wrong_msg);
// test for manuel return size selection (max workgroup size 1d)
auto max_wg_size = std::min(dev->max_work_item_sizes()[0], size_t{512});
auto reduce_buffer_size = static_cast<size_t>(max_wg_size) * 8;
auto reduce_local_size = static_cast<size_t>(max_wg_size);
auto reduce_work_groups = reduce_buffer_size / reduce_local_size;
auto reduce_global_size = reduce_buffer_size;
auto reduce_result_size = reduce_work_groups;
ivec input(reduce_buffer_size);
int n = static_cast<int>(input.capacity());
generate(input.begin(), input.end(), [&]{ return --n; });
nd_range range3{dims{reduce_global_size}, dims{}, dims{reduce_local_size}};
auto res_size = [&](const ivec&) { return reduce_result_size; };
auto w5 = mngr.spawn(prog, kn_reduce, range3,
in<int>{}, out<int>{res_size});
self->send(w5, move(input));
auto wg_size_as_int = static_cast<int>(max_wg_size);
ivec res4{
wg_size_as_int * 7, wg_size_as_int * 6, wg_size_as_int * 5,
wg_size_as_int * 4, wg_size_as_int * 3, wg_size_as_int * 2,
wg_size_as_int , 0
};
self->receive([&](const ivec& result) {
check_vector_results("Passing size for the output", res4, result);
}, others >> wrong_msg);
// calculator function for getting the size of the output
auto res_size2 = [](const ivec&) { return problem_size; };
// constant memory arguments
const ivec input2{static_cast<int>(problem_size)};
auto w6 = mngr.spawn(kernel_source, kn_const,
nd_range{dims{problem_size}},
in<int>{}, out<int>{res_size2});
self->send(w6, move(input2));
ivec res5(problem_size);
fill(begin(res5), end(res5), static_cast<int>(problem_size));
self->receive([&](const ivec& result) {
check_vector_results("Using const input argument", res5, result);
}, others >> wrong_msg);
}
void test_in_val_out_mref(actor_system& sys) {
CAF_MESSAGE("Testing in: val -> out: mref");
// setup
auto& mngr = sys.opencl_manager();
auto opt = mngr.find_device(0);
CAF_REQUIRE(opt);
auto dev = *opt;
auto prog = mngr.create_program(kernel_source, "", dev);
scoped_actor self{sys};
auto wrong_msg = [&](message_view& x) -> result<message> {
CAF_ERROR("unexpected message" << x.content().stringify());
return sec::unexpected_message;
};
// tests
const ivec res1{ 56, 62, 68, 74, 152, 174, 196, 218,
248, 286, 324, 362, 344, 398, 452, 506};
auto range = opencl::nd_range{dims{matrix_size, matrix_size}};
auto w1 = mngr.spawn(prog, kn_matrix, range, in<int>{}, out<int, mref>{});
self->send(w1, make_iota_vector<int>(matrix_size * matrix_size));
self->receive([&](iref& result) {
check_mref_results("Simple matrix multiplication using vectors"
" (kernel wrapped in program)", res1, result);
}, others >> wrong_msg);
// Pass kernel directly to the actor
auto w2 = mngr.spawn(kernel_source, kn_matrix, range,
in<int>{}, out<int, mref>{});
self->send(w2, make_iota_vector<int>(matrix_size * matrix_size));
self->receive([&](iref& result) {
check_mref_results("Simple matrix multiplication using vectors"
" (kernel passed directly)", res1, result);
}, others >> wrong_msg);
// test for manuel return size selection (max workgroup size 1d)
auto max_wg_size = std::min(dev->max_work_item_sizes()[0], size_t{512});
auto reduce_buffer_size = static_cast<size_t>(max_wg_size) * 8;
auto reduce_local_size = static_cast<size_t>(max_wg_size);
auto reduce_work_groups = reduce_buffer_size / reduce_local_size;
auto reduce_global_size = reduce_buffer_size;
auto reduce_result_size = reduce_work_groups;
ivec input(reduce_buffer_size);
int n = static_cast<int>(input.capacity());
generate(input.begin(), input.end(), [&]{ return --n; });
nd_range range3{dims{reduce_global_size}, dims{}, dims{reduce_local_size}};
auto res_size = [&](const ivec&) { return reduce_result_size; };
auto w5 = mngr.spawn(prog, kn_reduce, range3,
in<int>{}, out<int, mref>{res_size});
self->send(w5, move(input));
auto wg_size_as_int = static_cast<int>(max_wg_size);
ivec res4{wg_size_as_int * 7, wg_size_as_int * 6, wg_size_as_int * 5,
wg_size_as_int * 4, wg_size_as_int * 3, wg_size_as_int * 2,
wg_size_as_int * 1, wg_size_as_int * 0};
self->receive([&](iref& result) {
check_mref_results("Passing size for the output", res4, result);
}, others >> wrong_msg);
}
void test_in_mref_out_val(actor_system& sys) {
CAF_MESSAGE("Testing in: mref -> out: val ");
// setup
auto& mngr = sys.opencl_manager();
auto opt = mngr.find_device(0);
CAF_REQUIRE(opt);
auto dev = *opt;
auto prog = mngr.create_program(kernel_source, "", dev);
scoped_actor self{sys};
auto wrong_msg = [&](message_view& x) -> result<message> {
CAF_ERROR("unexpected message" << x.content().stringify());
return sec::unexpected_message;
};
// tests
const ivec res1{ 56, 62, 68, 74, 152, 174, 196, 218,
248, 286, 324, 362, 344, 398, 452, 506};
auto range = opencl::nd_range{dims{matrix_size, matrix_size}};
auto w1 = mngr.spawn(prog, kn_matrix, range, in<int, mref>{}, out<int>{});
auto matrix1 = make_iota_vector<int>(matrix_size * matrix_size);
auto input1 = dev->global_argument(matrix1);
self->send(w1, input1);
self->receive([&](const ivec& result) {
check_vector_results("Simple matrix multiplication using vectors"
" (kernel wrapped in program)", res1, result);
}, others >> wrong_msg);
// Pass kernel directly to the actor
auto w2 = mngr.spawn(kernel_source, kn_matrix, range,
in<int, mref>{}, out<int, val>{});
self->send(w2, input1);
self->receive([&](const ivec& result) {
check_vector_results("Simple matrix multiplication using vectors"
" (kernel passed directly)", res1, result);
}, others >> wrong_msg);
// test for manuel return size selection (max workgroup size 1d)
auto max_wg_size = std::min(dev->max_work_item_sizes()[0], size_t{512});
auto reduce_buffer_size = static_cast<size_t>(max_wg_size) * 8;
auto reduce_local_size = static_cast<size_t>(max_wg_size);
auto reduce_work_groups = reduce_buffer_size / reduce_local_size;
auto reduce_global_size = reduce_buffer_size;
auto reduce_result_size = reduce_work_groups;
ivec values(reduce_buffer_size);
int n = static_cast<int>(values.capacity());
generate(values.begin(), values.end(), [&]{ return --n; });
nd_range range3{dims{reduce_global_size}, dims{}, dims{reduce_local_size}};
auto res_size = [&](const iref&) { return reduce_result_size; };
auto w5 = mngr.spawn(prog, kn_reduce, range3,
in<int, mref>{}, out<int>{res_size});
auto input2 = dev->global_argument(values);
self->send(w5, input2);
auto multiplier = static_cast<int>(max_wg_size);
ivec res4{multiplier * 7, multiplier * 6, multiplier * 5,
multiplier * 4, multiplier * 3, multiplier * 2,
multiplier * 1, multiplier * 0};
self->receive([&](const ivec& result) {
check_vector_results("Passing size for the output", res4, result);
}, others >> wrong_msg);
}
void test_in_mref_out_mref(actor_system& sys) {
CAF_MESSAGE("Testing in: mref -> out: mref");
// setup
auto& mngr = sys.opencl_manager();
auto opt = mngr.find_device(0);
CAF_REQUIRE(opt);
auto dev = *opt;
auto prog = mngr.create_program(kernel_source, "", dev);
scoped_actor self{sys};
auto wrong_msg = [&](message_view& x) -> result<message> {
CAF_ERROR("unexpected message" << x.content().stringify());
return sec::unexpected_message;
};
// tests
const ivec res1{ 56, 62, 68, 74, 152, 174, 196, 218,
248, 286, 324, 362, 344, 398, 452, 506};
auto range = opencl::nd_range{dims{matrix_size, matrix_size}};
auto w1 = mngr.spawn(prog, kn_matrix, range,
in<int, mref>{}, out<int, mref>{});
auto matrix1 = make_iota_vector<int>(matrix_size * matrix_size);
auto input1 = dev->global_argument(matrix1);
self->send(w1, input1);
self->receive([&](iref& result) {
check_mref_results("Simple matrix multiplication using vectors"
" (kernel wrapped in program)", res1, result);
}, others >> wrong_msg);
// Pass kernel directly to the actor
auto w2 = mngr.spawn(kernel_source, kn_matrix, range,
in<int, mref>{}, out<int, mref>{});
self->send(w2, input1);
self->receive([&](iref& result) {
check_mref_results("Simple matrix multiplication using vectors"
" (kernel passed directly)", res1, result);
}, others >> wrong_msg);
// test for manuel return size selection (max workgroup size 1d)
auto max_wg_size = std::min(dev->max_work_item_sizes()[0], size_t{512});
auto reduce_buffer_size = static_cast<size_t>(max_wg_size) * 8;
auto reduce_local_size = static_cast<size_t>(max_wg_size);
auto reduce_work_groups = reduce_buffer_size / reduce_local_size;
auto reduce_global_size = reduce_buffer_size;
auto reduce_result_size = reduce_work_groups;
ivec values(reduce_buffer_size);
int n = static_cast<int>(values.capacity());
generate(values.begin(), values.end(), [&]{ return --n; });
nd_range range3{dims{reduce_global_size}, dims{}, dims{reduce_local_size}};
auto res_size = [&](const iref&) { return reduce_result_size; };
auto w5 = mngr.spawn(prog, kn_reduce, range3,
in<int, mref>{}, out<int, mref>{res_size});
auto input2 = dev->global_argument(values);
self->send(w5, input2);
auto multiplier = static_cast<int>(max_wg_size);
ivec res4{multiplier * 7, multiplier * 6, multiplier * 5,
multiplier * 4, multiplier * 3, multiplier * 2,
multiplier , 0};
self->receive([&](iref& result) {
check_mref_results("Passing size for the output", res4, result);
}, others >> wrong_msg);
}
void test_varying_arguments(actor_system& sys) {
CAF_MESSAGE("Testing varying argument order "
"(Might fail on some integrated GPUs)");
// setup
auto& mngr = sys.opencl_manager();
auto opt = mngr.find_device(0);
CAF_REQUIRE(opt);
auto dev = *opt;
auto prog = mngr.create_program(kernel_source, "", dev);
scoped_actor self{sys};
auto wrong_msg = [&](message_view& x) -> result<message> {
CAF_ERROR("unexpected message" << x.content().stringify());
return sec::unexpected_message;
};
// tests
size_t size = 23;
nd_range range{dims{size}};
auto input1 = make_iota_vector<int>(size);
auto input2 = dev->global_argument(input1);
auto w1 = mngr.spawn(prog, kn_varying, range,
in<int>{}, out<int>{}, in<int>{}, out<int>{});
self->send(w1, input1, input1);
self->receive([&](const ivec& res1, const ivec& res2) {
check_vector_results("Varying args (vec only), output 1", input1, res1);
check_vector_results("Varying args (vec only), output 2", input1, res2);
}, others >> wrong_msg);
auto w2 = mngr.spawn(prog, kn_varying, range,
in<int,mref>{}, out<int>{},
in<int>{}, out<int,mref>{});
self->send(w2, input2, input1);
self->receive([&](const ivec& res1, iref& res2) {
check_vector_results("Varying args (vec), output 1", input1, res1);
check_mref_results("Varying args (ref), output 2", input1, res2);
}, others >> wrong_msg);
}
void test_inout(actor_system& sys) {
CAF_MESSAGE("Testing in_out arguments");
// setup
auto& mngr = sys.opencl_manager();
auto opt = mngr.find_device(0);
CAF_REQUIRE(opt);
auto dev = *opt;
auto prog = mngr.create_program(kernel_source, "", dev);
scoped_actor self{sys};
auto wrong_msg = [&](message_view& x) -> result<message> {
CAF_ERROR("unexpected message" << x.content().stringify());
return sec::unexpected_message;
};
// tests
ivec input = make_iota_vector<int>(problem_size);
auto input2 = dev->global_argument(input);
auto input3 = dev->global_argument(input);
ivec res{input};
for_each(begin(res), end(res), [](int& val){ val *= 2; });
auto range = nd_range{dims{problem_size}};
auto w1 = mngr.spawn(kernel_source, kn_inout, range,
in_out<int,val,val>{});
self->send(w1, input);
self->receive([&](const ivec& result) {
check_vector_results("Testing in_out (val -> val)", res, result);
}, others >> wrong_msg);
auto w2 = mngr.spawn(kernel_source, kn_inout, range,
in_out<int,val,mref>{});
self->send(w2, input);
self->receive([&](iref& result) {
check_mref_results("Testing in_out (val -> mref)", res, result);
}, others >> wrong_msg);
auto w3 = mngr.spawn(kernel_source, kn_inout, range,
in_out<int,mref,val>{});
self->send(w3, input2);
self->receive([&](const ivec& result) {
check_vector_results("Testing in_out (mref -> val)", res, result);
}, others >> wrong_msg);
auto w4 = mngr.spawn(kernel_source, kn_inout, range,
in_out<int,mref,mref>{});
self->send(w4, input3);
self->receive([&](iref& result) {
check_mref_results("Testing in_out (mref -> mref)", res, result);
}, others >> wrong_msg);
}
void test_priv(actor_system& sys) {
CAF_MESSAGE("Testing priv argument");
// setup
auto& mngr = sys.opencl_manager();
auto opt = mngr.find_device(0);
CAF_REQUIRE(opt);
auto dev = *opt;
auto prog = mngr.create_program(kernel_source, "", dev);
scoped_actor self{sys};
auto wrong_msg = [&](message_view& x) -> result<message> {
CAF_ERROR("unexpected message" << x.content().stringify());
return sec::unexpected_message;
};
// tests
nd_range range{dims{problem_size}};
ivec input = make_iota_vector<int>(problem_size);
int value = 42;
ivec res{input};
for_each(begin(res), end(res), [&](int& val){ val += value; });
auto w1 = mngr.spawn(kernel_source, kn_private, range,
in_out<int>{}, priv<int>{value});
self->send(w1, input);
self->receive([&](const ivec& result) {
check_vector_results("Testing hidden private arugment", res, result);
}, others >> wrong_msg);
auto w2 = mngr.spawn(kernel_source, kn_private, range,
in_out<int>{}, priv<int,val>{});
self->send(w2, input, value);
self->receive([&](const ivec& result) {
check_vector_results("Testing val private arugment", res, result);
}, others >> wrong_msg);
}
void test_local(actor_system& sys) {
CAF_MESSAGE("Testing local argument");
// setup
auto& mngr = sys.opencl_manager();
auto opt = mngr.find_device(0);
CAF_REQUIRE(opt);
auto dev = *opt;
auto prog = mngr.create_program(kernel_source, "", dev);
scoped_actor self{sys};
auto wrong_msg = [&](message_view& x) -> result<message> {
CAF_ERROR("unexpected message" << x.content().stringify());
return sec::unexpected_message;
};
// tests
size_t global_size = 256;
size_t local_size = global_size / 2;
ivec res = make_iota_vector<int>(global_size);
auto last = 0;
for (size_t i = 0; i < global_size; ++i) {
if (i == local_size) last = 0;
auto tmp = res[i];
res[i] = last;
last += tmp;
}
auto range = nd_range{dims{global_size}, {}, dims{local_size}};
auto w = mngr.spawn(kernel_source, kn_local, range,
in_out<int>{}, local<int>{local_size});
self->send(w, make_iota_vector<int>(global_size));
self->receive([&](const ivec& result) {
check_vector_results("Testing local arugment", res, result);
}, others >> wrong_msg);
// Same test, different argument order
w = mngr.spawn(kernel_source, kn_order, range,
local<int>{local_size}, in_out<int>{});
self->send(w, make_iota_vector<int>(global_size));
self->receive([&](const ivec& result) {
check_vector_results("Testing local arugment", res, result);
}, others >> wrong_msg);
}
CAF_TEST(actor_facade) {
actor_system_config cfg;
cfg.load<opencl::manager>()
.add_message_type<ivec>("int_vector")
.add_message_type<matrix_type>("square_matrix");
actor_system system{cfg};
test_in_val_out_val(system);
test_in_val_out_mref(system);
test_in_mref_out_val(system);
test_in_mref_out_mref(system);
test_varying_arguments(system);
test_inout(system);
test_priv(system);
test_local(system);
system.await_all_actors_done();
}
......@@ -64,3 +64,32 @@ if(NOT WIN32)
install(DIRECTORY caf/ DESTINATION ${CMAKE_INSTALL_INCLUDEDIR}/caf FILES_MATCHING PATTERN "*.hpp")
endif()
# -- build unit tests ----------------------------------------------------------
if(CAF_NO_UNIT_TESTS)
return()
endif()
add_executable(caf-openssl-test test/openssl-test.cpp)
target_include_directories(caf-openssl-test PRIVATE
"${CMAKE_CURRENT_SOURCE_DIR}/test")
if (CAF_BUILD_STATIC_ONLY OR CAF_BUILD_STATIC)
target_link_libraries(caf-openssl-test
libcaf_openssl_static
libcaf_io_static
libcaf_core_static
${CAF_EXTRA_LDFLAGS})
else()
target_link_libraries(caf-openssl-test
libcaf_openssl_shared
libcaf_io_shared
libcaf_core_shared
${CAF_EXTRA_LDFLAGS})
endif()
caf_add_test_suites(caf-openssl-test
openssl.authentication
openssl.remote_actor
)
#include "openssl-test.hpp"
#include "caf/test/unit_test_impl.hpp"
#include "caf/test/dsl.hpp"
#include <cstdint>
CAF_BEGIN_TYPE_ID_BLOCK(openssl_test, caf::first_custom_type_id)
CAF_ADD_TYPE_ID(openssl_test, (std::vector<int32_t>) )
CAF_END_TYPE_ID_BLOCK(openssl_test)
......@@ -20,7 +20,7 @@
#include "caf/openssl/all.hpp"
#include "caf/test/dsl.hpp"
#include "openssl-test.hpp"
#include "caf/config.hpp"
......@@ -54,11 +54,10 @@ public:
config() {
load<io::middleman>();
load<openssl::manager>();
add_message_type<std::vector<int>>("std::vector<int>");
actor_system_config::parse(test::engine::argc(), test::engine::argv());
set("middleman.manual-multiplexing", true);
set("middleman.attach-utility-actors", true);
set("scheduler.policy", atom("testing"));
set("scheduler.policy", "testing");
}
static std::string data_dir() {
......
......@@ -16,14 +16,12 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/config.hpp"
#define CAF_SUITE openssl.dynamic_remote_actor
#include <signal.h>
#define CAF_SUITE openssl_dynamic_remote_actor
#include "caf/test/dsl.hpp"
#include "openssl-test.hpp"
#include <algorithm>
#include <signal.h>
#include <sstream>
#include <utility>
#include <vector>
......@@ -43,7 +41,7 @@ public:
config() {
load<io::middleman>();
load<openssl::manager>();
add_message_type<std::vector<int>>("std::vector<int>");
add_message_types<id_block::openssl_test>();
actor_system_config::parse(test::engine::argc(), test::engine::argv());
// Setting the "max consecutive reads" to 1 is highly likely to cause
// OpenSSL to buffer data internally and report "pending" data after a read
......@@ -67,7 +65,7 @@ struct fixture {
behavior make_pong_behavior() {
return {
[](int val) -> int {
[](int32_t val) -> int32_t {
++val;
CAF_MESSAGE("pong with " << val);
return val;
......@@ -79,7 +77,7 @@ behavior make_ping_behavior(event_based_actor* self, const actor& pong) {
CAF_MESSAGE("ping with " << 0);
self->send(pong, 0);
return {
[=](int val) -> int {
[=](int32_t val) -> int32_t {
if (val == 3) {
CAF_MESSAGE("ping with exit");
self->send_exit(self->current_sender(), exit_reason::user_shutdown);
......@@ -92,7 +90,7 @@ behavior make_ping_behavior(event_based_actor* self, const actor& pong) {
};
}
std::string to_string(const std::vector<int>& vec) {
std::string to_string(const std::vector<int32_t>& vec) {
std::ostringstream os;
for (size_t i = 0; i + 1 < vec.size(); ++i)
os << vec[i] << ", ";
......@@ -102,7 +100,7 @@ std::string to_string(const std::vector<int>& vec) {
behavior make_sort_behavior() {
return {
[](std::vector<int>& vec) -> std::vector<int> {
[](std::vector<int32_t>& vec) -> std::vector<int32_t> {
CAF_MESSAGE("sorter received: " << to_string(vec));
std::sort(vec.begin(), vec.end());
CAF_MESSAGE("sorter sent: " << to_string(vec));
......@@ -113,12 +111,12 @@ behavior make_sort_behavior() {
behavior make_sort_requester_behavior(event_based_actor* self,
const actor& sorter) {
self->send(sorter, std::vector<int>{5, 4, 3, 2, 1});
self->send(sorter, std::vector<int32_t>{5, 4, 3, 2, 1});
return {
[=](const std::vector<int>& vec) {
[=](const std::vector<int32_t>& vec) {
CAF_MESSAGE("sort requester received: " << to_string(vec));
for (size_t i = 1; i < vec.size(); ++i)
CAF_CHECK_EQUAL(static_cast<int>(i), vec[i - 1]);
CAF_CHECK_EQUAL(static_cast<int32_t>(i), vec[i - 1]);
self->send_exit(sorter, exit_reason::user_shutdown);
self->quit();
},
......@@ -127,7 +125,7 @@ behavior make_sort_requester_behavior(event_based_actor* self,
behavior fragile_mirror(event_based_actor* self) {
return {
[=](int i) {
[=](int32_t i) {
self->quit(exit_reason::user_shutdown);
return i;
},
......@@ -139,7 +137,7 @@ behavior linking_actor(event_based_actor* self, const actor& buddy) {
self->link_to(buddy);
self->send(buddy, 42);
return {
[](int i) { CAF_CHECK_EQUAL(i, 42); },
[](int32_t i) { CAF_CHECK_EQUAL(i, 42); },
};
}
......
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