Commit 99d1af2b authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'issue/1155'

Close #1155.
parents 639ac2ed e1308bfa
......@@ -105,7 +105,7 @@ if(CAF_ENABLE_TESTING)
target_sources(${target} PRIVATE
"${CMAKE_CURRENT_SOURCE_DIR}/test/${suitePath}.cpp")
add_test(NAME ${suiteName}
COMMAND ${target} -r300 -n -v5 -s"^${suiteName}$")
COMMAND ${target} -r300 -n -v5 -s "^${suiteName}$")
endforeach()
endfunction()
endif()
......
......@@ -94,7 +94,7 @@ config = [
'ASAN_OPTIONS=detect_leaks=1',
],
]],
// One extra debug build with static libraries and UBSanitizer.
// One extra debug build with static libs, UBSan and hardening flags.
['fedora-32', [
numCores: 4,
tags: ['docker', 'UBSanitizer'],
......@@ -104,7 +104,7 @@ config = [
'CAF_SANITIZERS:STRING=address,undefined',
],
extraBuildEnv: [
'CXXFLAGS=-fno-sanitize-recover=undefined',
'CXXFLAGS=-fno-sanitize-recover=undefined -D_GLIBCXX_DEBUG',
'LDFLAGS=-fno-sanitize-recover=undefined',
],
]],
......
......@@ -88,7 +88,7 @@ public:
template <class... Us>
std::enable_if_t<(std::is_constructible<Ts, Us>::value && ...)>
deliver(Us... xs) {
promise_.deliver(make_message(Ts{std::forward<Us>(xs)}...));
promise_.deliver(Ts{std::forward<Us>(xs)}...);
}
/// Satisfies the promise by sending an empty response message.
......
......@@ -16,15 +16,16 @@ namespace caf {
namespace {
bool requires_response(const strong_actor_ptr& source,
const mailbox_element::forwarding_stack& stages,
message_id mid) {
return !mid.is_response() && !mid.is_answered()
&& (source || !stages.empty());
bool requires_response(message_id mid) {
return !mid.is_response() && !mid.is_answered();
}
bool requires_response(const mailbox_element& src) {
return requires_response(src.sender, src.stages, src.mid);
return requires_response(src.mid);
}
bool has_response_receiver(const mailbox_element& src) {
return src.sender || !src.stages.empty();
}
} // namespace
......@@ -37,7 +38,7 @@ response_promise::response_promise(local_actor* self, strong_actor_ptr source,
// Form an invalid request promise when initialized from a response ID, since
// we always drop messages in this case. Also don't create promises for
// anonymous messages since there's nowhere to send the message to anyway.
if (requires_response(source, stages, mid)) {
if (requires_response(mid)) {
state_ = make_counted<state>();
state_->self = self;
state_->source.swap(source);
......@@ -118,7 +119,8 @@ void response_promise::deliver() {
void response_promise::respond_to(local_actor* self, mailbox_element* request,
message& response) {
if (request && requires_response(*request)) {
if (request && requires_response(*request)
&& has_response_receiver(*request)) {
state tmp;
tmp.self = self;
tmp.source.swap(request->sender);
......@@ -131,7 +133,8 @@ void response_promise::respond_to(local_actor* self, mailbox_element* request,
void response_promise::respond_to(local_actor* self, mailbox_element* request,
error& response) {
if (request && requires_response(*request)) {
if (request && requires_response(*request)
&& has_response_receiver(*request)) {
state tmp;
tmp.self = self;
tmp.source.swap(request->sender);
......@@ -159,17 +162,14 @@ void response_promise::state::deliver_impl(message msg) {
CAF_LOG_TRACE(CAF_ARG(msg));
if (msg.empty() && id.is_async()) {
CAF_LOG_DEBUG("drop response: empty response to asynchronous input");
} else {
if (stages.empty()) {
detail::profiled_send(self, self->ctrl(), source, id.response_id(),
forwarding_stack{}, self->context(),
std::move(msg));
} else {
auto next = std::move(stages.back());
stages.pop_back();
detail::profiled_send(self, std::move(source), next, id,
std::move(stages), self->context(), std::move(msg));
}
} else if (!stages.empty()) {
auto next = std::move(stages.back());
stages.pop_back();
detail::profiled_send(self, std::move(source), next, id, std::move(stages),
self->context(), std::move(msg));
} else if (source != nullptr) {
detail::profiled_send(self, self->ctrl(), source, id.response_id(),
forwarding_stack{}, self->context(), std::move(msg));
}
cancel();
}
......
......@@ -11,6 +11,7 @@
#include <fstream>
#include <ios>
#include <iostream>
#include <memory>
#include <thread>
#include <unordered_map>
......@@ -38,95 +39,46 @@ namespace {
using string_sink = std::function<void(std::string&&)>;
// the first value is the use count, the last ostream_handle that
// decrements it to 0 removes the ostream pointer from the map
using counted_sink = std::pair<size_t, string_sink>;
using string_sink_ptr = std::shared_ptr<string_sink>;
using sink_cache = std::map<std::string, counted_sink>;
class sink_handle {
public:
using iterator = sink_cache::iterator;
sink_handle() : cache_(nullptr) {
// nop
}
sink_handle(sink_cache* fc, iterator iter) : cache_(fc), iter_(iter) {
if (cache_ != nullptr)
++iter_->second.first;
}
sink_handle(const sink_handle& other) : cache_(nullptr) {
*this = other;
}
sink_handle& operator=(const sink_handle& other) {
if (cache_ != other.cache_ || iter_ != other.iter_) {
clear();
cache_ = other.cache_;
if (cache_ != nullptr) {
iter_ = other.iter_;
++iter_->second.first;
}
}
return *this;
}
~sink_handle() {
clear();
}
explicit operator bool() const {
return cache_ != nullptr;
}
string_sink& operator*() {
CAF_ASSERT(iter_->second.second != nullptr);
return iter_->second.second;
}
private:
void clear() {
if (cache_ != nullptr && --iter_->second.first == 0) {
cache_->erase(iter_);
cache_ = nullptr;
}
}
sink_cache* cache_;
sink_cache::iterator iter_;
};
using sink_cache = std::map<std::string, string_sink_ptr>;
string_sink make_sink(actor_system& sys, const std::string& fn, int flags) {
if (fn.empty())
if (fn.empty()) {
return nullptr;
if (fn.front() == ':') {
} else if (fn.front() == ':') {
// "virtual file" name given, translate this to group communication
auto grp = sys.groups().get_local(fn);
return [grp, fn](std::string&& out) { anon_send(grp, fn, std::move(out)); };
} else {
auto append = (flags & actor_ostream::append) != 0;
auto fs = std::make_shared<std::ofstream>();
fs->open(fn, append ? std::ios_base::out | std::ios_base::app
: std::ios_base::out);
if (fs->is_open()) {
return [fs](std::string&& out) { *fs << out; };
} else {
std::cerr << "cannot open file: " << fn << std::endl;
return nullptr;
}
}
auto append = (flags & actor_ostream::append) != 0;
auto fs = std::make_shared<std::ofstream>();
fs->open(fn, append ? std::ios_base::out | std::ios_base::app
: std::ios_base::out);
if (fs->is_open())
return [fs](std::string&& out) { *fs << out; };
std::cerr << "cannot open file: " << fn << std::endl;
return nullptr;
}
sink_handle get_sink_handle(actor_system& sys, sink_cache& fc,
const std::string& fn, int flags) {
auto i = fc.find(fn);
if (i != fc.end())
return {&fc, i};
auto fs = make_sink(sys, fn, flags);
if (fs) {
i = fc.emplace(fn, sink_cache::mapped_type{0, std::move(fs)}).first;
return {&fc, i};
string_sink_ptr get_or_add_sink_ptr(actor_system& sys, sink_cache& fc,
const std::string& fn, int flags) {
if (auto i = fc.find(fn); i != fc.end()) {
return i->second;
} else if (auto fs = make_sink(sys, fn, flags)) {
if (fs) {
auto ptr = std::make_shared<string_sink>(std::move(fs));
fc.emplace(fn, ptr);
return ptr;
} else {
return nullptr;
}
} else {
return nullptr;
}
return {};
}
class printer_actor : public blocking_actor {
......@@ -138,14 +90,14 @@ public:
void act() override {
struct actor_data {
std::string current_line;
sink_handle redirect;
string_sink_ptr redirect;
actor_data() {
// nop
}
};
using data_map = std::unordered_map<actor_id, actor_data>;
sink_cache fcache;
sink_handle global_redirect;
string_sink_ptr global_redirect;
data_map data;
auto get_data = [&](actor_id addr, bool insert_missing) -> actor_data* {
if (addr == invalid_actor_id)
......@@ -191,12 +143,12 @@ public:
}
},
[&](redirect_atom, const std::string& fn, int flag) {
global_redirect = get_sink_handle(system(), fcache, fn, flag);
global_redirect = get_or_add_sink_ptr(system(), fcache, fn, flag);
},
[&](redirect_atom, actor_id aid, const std::string& fn, int flag) {
auto d = get_data(aid, true);
if (d != nullptr)
d->redirect = get_sink_handle(system(), fcache, fn, flag);
d->redirect = get_or_add_sink_ptr(system(), fcache, fn, flag);
},
[&](exit_msg& em) {
fail_state(std::move(em.reason));
......
......@@ -59,13 +59,11 @@ void replace_all(std::string& str, string_view what, string_view with) {
};
auto i = next(str.begin());
while (i != str.end()) {
auto before = std::distance(str.begin(), i);
CAF_ASSERT(before >= 0);
auto ws = static_cast<decltype(before)>(what.size());
str.replace(i, i + ws, with.begin(), with.end());
auto before = static_cast<size_t>(std::distance(str.begin(), i));
str.replace(i, i + what.size(), with.begin(), with.end());
// Iterator i became invalidated -> use new iterator pointing to the first
// character after the replaced text.
i = next(str.begin() + before + ws);
i = next(str.begin() + before + with.size());
}
}
......
......@@ -152,8 +152,10 @@ CAF_TEST(file input overrides defaults but CLI args always win) {
// Checks whether both a synced variable and the corresponding entry in
// content(cfg) are equal to `value`.
#define CHECK_SYNCED(var, value) \
#define CHECK_SYNCED(var, ...) \
do { \
using ref_value_type = std::decay_t<decltype(var)>; \
ref_value_type value{__VA_ARGS__}; \
CAF_CHECK_EQUAL(var, value); \
if (auto maybe_val = get_as<decltype(var)>(cfg, #var)) { \
CAF_CHECK_EQUAL(*maybe_val, value); \
......@@ -206,14 +208,12 @@ CAF_TEST(integers and integer containers options) {
CHECK_SYNCED(some_int, 42);
CHECK_SYNCED(some_other_int, 23);
CHECK_TEXT_ONLY(int, yet_another_int, 123);
CHECK_SYNCED(some_int_list, int_list({1, 2, 3}));
CHECK_SYNCED(some_int_list_list, int_list_list({{1, 2, 3}, {4, 5, 6}}));
CHECK_SYNCED(some_int_map, int_map({{{"a", 1}, {"b", 2}, {"c", 3}}}));
CHECK_SYNCED(some_int_list_map,
int_list_map({{{"a", {1, 2, 3}}, {"b", {4, 5, 6}}}}));
CHECK_SYNCED(some_int_map_list,
int_map_list({{{"a", 1}, {"b", 2}, {"c", 3}},
{{"d", 4}, {"e", 5}, {"f", 6}}}));
CHECK_SYNCED(some_int_list, 1, 2, 3);
CHECK_SYNCED(some_int_list_list, {1, 2, 3}, {4, 5, 6});
CHECK_SYNCED(some_int_map, {{"a", 1}, {"b", 2}, {"c", 3}});
CHECK_SYNCED(some_int_list_map, {{"a", {1, 2, 3}}, {"b", {4, 5, 6}}});
CHECK_SYNCED(some_int_map_list, {{"a", 1}, {"b", 2}, {"c", 3}},
{{"d", 4}, {"e", 5}, {"f", 6}});
}
CAF_TEST(basic and basic containers options) {
......@@ -279,22 +279,20 @@ CAF_TEST(basic and basic containers options) {
CHECK_SYNCED(some_uri, "foo:bar"_u);
CHECK_SYNCED(some_string, "string"s);
CAF_MESSAGE("check list types");
CHECK_SYNCED(some_int_list, int_list({1, 2, 3}));
CHECK_SYNCED(some_bool_list, bool_list({false, true}));
CHECK_SYNCED(some_double_list, double_list({1., 2., 3.}));
CHECK_SYNCED(some_timespan_list, timespan_list({123_ms, 234_ms, 345_ms}));
CHECK_SYNCED(some_uri_list, uri_list({"foo:a"_u, "foo:b"_u, "foo:c"_u}));
CHECK_SYNCED(some_string_list, string_list({"a", "b", "c"}));
CHECK_SYNCED(some_int_list, 1, 2, 3);
CHECK_SYNCED(some_bool_list, false, true);
CHECK_SYNCED(some_double_list, 1., 2., 3.);
CHECK_SYNCED(some_timespan_list, 123_ms, 234_ms, 345_ms);
CHECK_SYNCED(some_uri_list, "foo:a"_u, "foo:b"_u, "foo:c"_u);
CHECK_SYNCED(some_string_list, "a", "b", "c");
CAF_MESSAGE("check dictionary types");
CHECK_SYNCED(some_int_map, int_map({{"a", 1}, {"b", 2}, {"c", 3}}));
CHECK_SYNCED(some_bool_map, bool_map({{"a", true}, {"b", false}}));
CHECK_SYNCED(some_double_map, double_map({{"a", 1.}, {"b", 2.}, {"c", 3.}}));
CHECK_SYNCED(some_timespan_map,
timespan_map({{"a", 123_ms}, {"b", 234_ms}, {"c", 345_ms}}));
CHECK_SYNCED(some_uri_map,
uri_map({{"a", "foo:a"_u}, {"b", "foo:b"_u}, {"c", "foo:c"_u}}));
CHECK_SYNCED(some_string_map,
string_map({{"a", "1"}, {"b", "2"}, {"c", "3"}}));
CHECK_SYNCED(some_int_map, {"a", 1}, {"b", 2}, {"c", 3});
CHECK_SYNCED(some_bool_map, {"a", true}, {"b", false});
CHECK_SYNCED(some_double_map, {"a", 1.}, {"b", 2.}, {"c", 3.});
CHECK_SYNCED(some_timespan_map, {"a", 123_ms}, {"b", 234_ms}, {"c", 345_ms});
CHECK_SYNCED(some_uri_map, {"a", "foo:a"_u}, {"b", "foo:b"_u},
{"c", "foo:c"_u});
CHECK_SYNCED(some_string_map, {"a", "1"}, {"b", "2"}, {"c", "3"});
}
SCENARIO("config files allow both nested and dot-separated values") {
......
......@@ -20,17 +20,17 @@ public:
behavior make_behavior() override {
return {
[=](delete_atom) { quit(exit_reason::user_shutdown); },
[=](delete_atom) {
CAF_MESSAGE("testee received delete");
quit(exit_reason::user_shutdown);
},
};
}
};
class spawner : public event_based_actor {
public:
spawner(actor_config& cfg)
: event_based_actor(cfg),
downs_(0),
testee_(spawn<testee, monitored>(this)) {
spawner(actor_config& cfg) : event_based_actor(cfg), downs_(0) {
set_down_handler([=](down_msg& msg) {
CAF_CHECK_EQUAL(msg.reason, exit_reason::user_shutdown);
CAF_CHECK_EQUAL(msg.source, testee_.address());
......@@ -40,6 +40,7 @@ public:
}
behavior make_behavior() override {
testee_ = spawn<testee, monitored>(this);
return {
[=](ok_atom, const error& reason) {
CAF_CHECK_EQUAL(reason, exit_reason::user_shutdown);
......@@ -47,12 +48,15 @@ public:
quit(reason);
}
},
[=](delete_atom x) { return delegate(testee_, x); },
[=](delete_atom x) {
CAF_MESSAGE("spawner received delete");
return delegate(testee_, x);
},
};
}
void on_exit() override {
destroy(testee_);
testee_ = nullptr;
}
private:
......@@ -62,8 +66,11 @@ private:
} // namespace
BEGIN_FIXTURE_SCOPE(test_coordinator_fixture<>)
CAF_TEST(constructor_attach) {
actor_system_config cfg;
actor_system system{cfg};
anon_send(system.spawn<spawner>(), delete_atom_v);
anon_send(sys.spawn<spawner>(), delete_atom_v);
run();
}
END_FIXTURE_SCOPE()
......@@ -63,13 +63,13 @@ CAF_TEST(reserve) {
CAF_CHECK_EQUAL(a.size(), 0ul);
CAF_CHECK_EQUAL(a.capacity(), 1024ul);
CAF_CHECK(a.data() != nullptr);
CAF_CHECK_EQUAL(a.begin(), a.end());
CAF_CHECK(a.begin() == a.end());
CAF_CHECK(a.empty());
a.reserve(512);
CAF_CHECK_EQUAL(a.size(), 0ul);
CAF_CHECK_EQUAL(a.capacity(), 1024ul);
CAF_CHECK(a.data() != nullptr);
CAF_CHECK_EQUAL(a.begin(), a.end());
CAF_CHECK(a.begin() == a.end());
CAF_CHECK(a.empty());
}
......
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