Commit 6d67d620 authored by Dominik Charousset's avatar Dominik Charousset

Implement new actor_clock type

parent a50563f2
...@@ -18,6 +18,7 @@ set (LIBCAF_CORE_SRCS ...@@ -18,6 +18,7 @@ set (LIBCAF_CORE_SRCS
src/abstract_group.cpp src/abstract_group.cpp
src/actor.cpp src/actor.cpp
src/actor_addr.cpp src/actor_addr.cpp
src/actor_clock.cpp
src/actor_companion.cpp src/actor_companion.cpp
src/actor_config.cpp src/actor_config.cpp
src/actor_control_block.cpp src/actor_control_block.cpp
...@@ -89,6 +90,7 @@ set (LIBCAF_CORE_SRCS ...@@ -89,6 +90,7 @@ set (LIBCAF_CORE_SRCS
src/sequencer.cpp src/sequencer.cpp
src/serializer.cpp src/serializer.cpp
src/shared_spinlock.cpp src/shared_spinlock.cpp
src/simple_actor_clock.cpp
src/skip.cpp src/skip.cpp
src/splitter.cpp src/splitter.cpp
src/stream.cpp src/stream.cpp
...@@ -106,6 +108,7 @@ set (LIBCAF_CORE_SRCS ...@@ -106,6 +108,7 @@ set (LIBCAF_CORE_SRCS
src/sync_request_bouncer.cpp src/sync_request_bouncer.cpp
src/term.cpp src/term.cpp
src/terminal_stream_scatterer.cpp src/terminal_stream_scatterer.cpp
src/test_actor_clock.cpp
src/test_coordinator.cpp src/test_coordinator.cpp
src/timestamp.cpp src/timestamp.cpp
src/try_match.cpp src/try_match.cpp
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#ifndef CAF_ACTOR_CLOCK
#define CAF_ACTOR_CLOCK
#include <chrono>
#include "caf/fwd.hpp"
namespace caf {
/// A monotonic clock for scheduling timeouts and delayed messages.
class actor_clock {
public:
// -- member types -----------------------------------------------------------
/// Underlying clock type.
using clock_type = std::chrono::steady_clock;
/// Discrete point in time.
using time_point = typename clock_type::time_point;
/// Difference between two points in time.
using duration_type = typename clock_type::duration;
// -- constructors, destructors, and assignment operators --------------------
virtual ~actor_clock();
// -- observers --------------------------------------------------------------
/// Returns the current wall-clock time.
virtual time_point now() const noexcept;
/// Returns the difference between `t0` and `t1`, allowing the clock to
/// return any arbitrary value depending on the measurement that took place.
virtual duration_type difference(atom_value measurement, time_point t0,
time_point t1) const noexcept;
/// Schedules a `timeout_msg` for `self` at time point `t`, overriding any
/// previous receive timeout.
virtual void set_receive_timeout(time_point t, abstract_actor* self,
uint32_t id) = 0;
/// Schedules a `sec::request_timeout` for `self` at time point `t`.
virtual void set_request_timeout(time_point t, abstract_actor* self,
message_id id) = 0;
/// Cancels a pending receive timeout.
virtual void cancel_receive_timeout(abstract_actor* self) = 0;
/// Cancels the pending request timeout for `id`.
virtual void cancel_request_timeout(abstract_actor* self, message_id id) = 0;
/// Cancels all timeouts for `self`.
virtual void cancel_timeouts(abstract_actor* self) = 0;
/// Schedules an arbitrary message to `receiver` for time point `t`.
virtual void schedule_message(time_point t, strong_actor_ptr receiver,
mailbox_element_ptr content) = 0;
/// Schedules an arbitrary message to `target` for time point `t`.
virtual void schedule_message(time_point t, group target,
strong_actor_ptr sender, message content) = 0;
};
} // namespace caf
#endif // CAF_ACTOR_CLOCK
...@@ -51,6 +51,7 @@ ...@@ -51,6 +51,7 @@
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
#include "caf/replies_to.hpp" #include "caf/replies_to.hpp"
#include "caf/serializer.hpp" #include "caf/serializer.hpp"
#include "caf/actor_clock.hpp"
#include "caf/actor_proxy.hpp" #include "caf/actor_proxy.hpp"
#include "caf/exit_reason.hpp" #include "caf/exit_reason.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#ifndef CAF_DETAIL_SIMPLE_ACTOR_CLOCK_HPP
#define CAF_DETAIL_SIMPLE_ACTOR_CLOCK_HPP
#include <map>
#include "caf/actor_clock.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/group.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/message.hpp"
#include "caf/message_id.hpp"
#include "caf/variant.hpp"
namespace caf {
namespace detail {
class simple_actor_clock : public actor_clock {
public:
// -- member types -----------------------------------------------------------
/// Request for a `timeout_msg`.
struct receive_timeout {
strong_actor_ptr self;
uint32_t id;
};
/// Request for a `sec::request_timeout` error.
struct request_timeout {
strong_actor_ptr self;
message_id id;
};
/// Request for sending a message to an actor at a later time.
struct actor_msg {
strong_actor_ptr receiver;
mailbox_element_ptr content;
};
/// Request for sending a message to a group at a later time.
struct group_msg {
group target;
strong_actor_ptr sender;
message content;
};
using value_type = variant<receive_timeout, request_timeout,
actor_msg, group_msg>;
using map_type = std::multimap<time_point, value_type>;
using secondary_map = std::multimap<abstract_actor*, map_type::iterator>;
struct receive_predicate {
bool operator()(const secondary_map::value_type& x) const noexcept;
};
struct request_predicate {
message_id id;
bool operator()(const secondary_map::value_type& x) const noexcept;
};
struct visitor {
simple_actor_clock* thisptr;
void operator()(receive_timeout& x);
void operator()(request_timeout& x);
void operator()(actor_msg& x);
void operator()(group_msg& x);
};
void set_receive_timeout(time_point t, abstract_actor* self,
uint32_t id) override;
void set_request_timeout(time_point t, abstract_actor* self,
message_id id) override;
void cancel_receive_timeout(abstract_actor* self) override;
void cancel_request_timeout(abstract_actor* self, message_id id) override;
void cancel_timeouts(abstract_actor* self) override;
void schedule_message(time_point t, strong_actor_ptr receiver,
mailbox_element_ptr content) override;
void schedule_message(time_point t, group target, strong_actor_ptr sender,
message content) override;
inline const map_type& schedule() const {
return schedule_;
}
inline const secondary_map& actor_lookup() const {
return actor_lookup_;
}
protected:
template <class Predicate>
secondary_map::iterator lookup(abstract_actor* self, Predicate pred) {
auto e = actor_lookup_.end();
auto range = actor_lookup_.equal_range(self);
if (range.first == range.second)
return e;
auto i = std::find_if(range.first, range.second, pred);
return i != range.second ? i : e;
}
template <class Predicate>
void cancel(abstract_actor* self, Predicate pred) {
auto i = lookup(self, pred);
if (i != actor_lookup_.end()) {
schedule_.erase(i->second);
actor_lookup_.erase(i);
}
}
template <class Predicate>
void drop_lookup(abstract_actor* self, Predicate pred) {
auto i = lookup(self, pred);
if (i != actor_lookup_.end())
actor_lookup_.erase(i);
}
map_type schedule_;
secondary_map actor_lookup_;
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_SIMPLE_ACTOR_CLOCK_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#ifndef CAF_DETAIL_TEST_ACTOR_CLOCK_HPP
#define CAF_DETAIL_TEST_ACTOR_CLOCK_HPP
#include "caf/detail/simple_actor_clock.hpp"
namespace caf {
namespace detail {
class test_actor_clock : public simple_actor_clock {
public:
time_point current_time;
time_point now() const noexcept override;
void advance_time(duration_type x);
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_TEST_ACTOR_CLOCK_HPP
...@@ -74,6 +74,7 @@ class actor_addr; ...@@ -74,6 +74,7 @@ class actor_addr;
class actor_pool; class actor_pool;
class message_id; class message_id;
class serializer; class serializer;
class actor_clock;
class actor_proxy; class actor_proxy;
class local_actor; class local_actor;
class ref_counted; class ref_counted;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#include "caf/actor_clock.hpp"
namespace caf {
// -- constructors, destructors, and assignment operators ----------------------
actor_clock::~actor_clock() {
// nop
}
// -- observers ----------------------------------------------------------------
actor_clock::time_point actor_clock::now() const noexcept {
return clock_type::now();
}
actor_clock::duration_type
actor_clock::difference(atom_value, time_point t0,
time_point t1) const noexcept {
return t1 - t0;
}
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#include "caf/detail/simple_actor_clock.hpp"
#include "caf/actor_cast.hpp"
#include "caf/sec.hpp"
#include "caf/system_messages.hpp"
namespace caf {
namespace detail {
bool simple_actor_clock::receive_predicate::
operator()(const secondary_map::value_type& x) const noexcept {
return holds_alternative<receive_timeout>(x.second->second);
}
bool simple_actor_clock::request_predicate::
operator()(const secondary_map::value_type& x) const noexcept {
if (holds_alternative<request_timeout>(x.second->second)) {
auto& rt = get<request_timeout>(x.second->second);
return rt.id == id;
}
return false;
}
void simple_actor_clock::visitor::operator()(receive_timeout& x) {
CAF_ASSERT(x.self != nullptr);
x.self->get()->eq_impl(message_id::make(), x.self, nullptr,
timeout_msg{x.id});
receive_predicate pred;
thisptr->drop_lookup(x.self->get(), pred);
}
void simple_actor_clock::visitor::operator()(request_timeout& x) {
CAF_ASSERT(x.self != nullptr);
x.self->get()->eq_impl(x.id, x.self, nullptr, sec::request_timeout);
request_predicate pred{x.id};
thisptr->drop_lookup(x.self->get(), pred);
}
void simple_actor_clock::visitor::operator()(actor_msg& x) {
x.receiver->enqueue(std::move(x.content), nullptr);
}
void simple_actor_clock::visitor::operator()(group_msg& x) {
x.target->eq_impl(message_id::make(), std::move(x.sender), nullptr,
std::move(x.content));
}
void simple_actor_clock::set_receive_timeout(time_point t, abstract_actor* self,
uint32_t id) {
receive_predicate pred;
auto i = lookup(self, pred);
auto sptr = actor_cast<strong_actor_ptr>(self);
if (i != actor_lookup_.end()) {
schedule_.erase(i->second);
i->second = schedule_.emplace(t, receive_timeout{std::move(sptr), id});
} else {
auto j = schedule_.emplace(t, receive_timeout{std::move(sptr), id});
actor_lookup_.emplace(self, j);
}
}
void simple_actor_clock::set_request_timeout(time_point t, abstract_actor* self,
message_id id) {
request_predicate pred{id};
auto i = lookup(self, pred);
auto sptr = actor_cast<strong_actor_ptr>(self);
if (i != actor_lookup_.end()) {
schedule_.erase(i->second);
i->second = schedule_.emplace(t, request_timeout{std::move(sptr), id});
} else {
auto j = schedule_.emplace(t, request_timeout{std::move(sptr), id});
actor_lookup_.emplace(self, j);
}
}
void simple_actor_clock::cancel_receive_timeout(abstract_actor* self) {
receive_predicate pred;
cancel(self, pred);
}
void simple_actor_clock::cancel_request_timeout(abstract_actor* self,
message_id id) {
request_predicate pred{id};
cancel(self, pred);
}
void simple_actor_clock::cancel_timeouts(abstract_actor* self) {
auto range = actor_lookup_.equal_range(self);
if (range.first == range.second)
return;
for (auto i = range.first; i != range.second; ++i)
schedule_.erase(i->second);
actor_lookup_.erase(range.first, range.second);
}
void simple_actor_clock::schedule_message(time_point t,
strong_actor_ptr receiver,
mailbox_element_ptr content) {
schedule_.emplace(t, actor_msg{std::move(receiver), std::move(content)});
}
void simple_actor_clock::schedule_message(time_point t, group target,
strong_actor_ptr sender,
message content) {
schedule_.emplace(
t, group_msg{std::move(target), std::move(sender), std::move(content)});
}
} // namespace detail
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#include "caf/detail/test_actor_clock.hpp"
namespace caf {
namespace detail {
test_actor_clock::time_point test_actor_clock::now() const noexcept {
return current_time;
}
void test_actor_clock::advance_time(duration_type x) {
visitor f{this};
current_time += x;
auto i = schedule_.begin();
while (i != schedule_.end() && i->first <= current_time) {
visit(f, i->second);
i = schedule_.erase(i);
}
}
} // namespace detail
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#include "caf/config.hpp"
#define CAF_SUITE actor_clock
#include "caf/test/dsl.hpp"
#include <chrono>
#include <memory>
#include "caf/all.hpp"
#include "caf/detail/test_actor_clock.hpp"
#include "caf/raw_event_based_actor.hpp"
using namespace caf;
namespace {
using std::chrono::seconds;
struct testee_state {
uint32_t timeout_id = 41;
};
behavior testee(stateful_actor<testee_state, raw_event_based_actor>* self,
detail::test_actor_clock* t) {
return {
[=](ok_atom) {
auto n = t->now() + seconds(10);
self->state.timeout_id += 1;
t->set_receive_timeout(n, self, self->state.timeout_id);
},
[=](add_atom) {
auto n = t->now() + seconds(10);
self->state.timeout_id += 1;
auto mid = message_id::make(self->state.timeout_id).response_id();
t->set_request_timeout(n, self, mid);
},
[](const timeout_msg&) {
// nop
},
[](const error&) {
// nop
},
[](const std::string&) {
// nop
},
[=](group& grp) {
self->join(grp);
},
[=](exit_msg& x) {
self->quit(x.reason);
}
};
}
struct fixture : test_coordinator_fixture<> {
detail::test_actor_clock t;
actor aut;
fixture() : aut(sys.spawn<lazy_init>(testee, &t)) {
// nop
}
};
struct tid {
uint32_t value;
};
inline bool operator==(const timeout_msg& x, const tid& y) {
return x.timeout_id == y.value;
}
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(timer_tests, fixture)
CAF_TEST(single_receive_timeout) {
// Have AUT call t.set_receive_timeout().
self->send(aut, ok_atom::value);
expect((ok_atom), from(self).to(aut).with(_));
CAF_CHECK_EQUAL(t.schedule().size(), 1u);
CAF_CHECK_EQUAL(t.actor_lookup().size(), 1u);
// Advance time to send timeout message.
t.advance_time(seconds(10));
CAF_CHECK_EQUAL(t.schedule().size(), 0u);
CAF_CHECK_EQUAL(t.actor_lookup().size(), 0u);
// Have AUT receive the timeout.
expect((timeout_msg), from(aut).to(aut).with(tid{42}));
}
CAF_TEST(override_receive_timeout) {
// Have AUT call t.set_receive_timeout().
self->send(aut, ok_atom::value);
expect((ok_atom), from(self).to(aut).with(_));
CAF_CHECK_EQUAL(t.schedule().size(), 1u);
CAF_CHECK_EQUAL(t.actor_lookup().size(), 1u);
// Have AUT call t.set_timeout() again.
self->send(aut, ok_atom::value);
expect((ok_atom), from(self).to(aut).with(_));
CAF_CHECK_EQUAL(t.schedule().size(), 1u);
CAF_CHECK_EQUAL(t.actor_lookup().size(), 1u);
// Advance time to send timeout message.
t.advance_time(seconds(10));
CAF_CHECK_EQUAL(t.schedule().size(), 0u);
CAF_CHECK_EQUAL(t.actor_lookup().size(), 0u);
// Have AUT receive the timeout.
expect((timeout_msg), from(aut).to(aut).with(tid{43}));
}
CAF_TEST(single_request_timeout) {
// Have AUT call t.set_request_timeout().
self->send(aut, add_atom::value);
expect((add_atom), from(self).to(aut).with(_));
CAF_CHECK_EQUAL(t.schedule().size(), 1u);
CAF_CHECK_EQUAL(t.actor_lookup().size(), 1u);
// Advance time to send timeout message.
t.advance_time(seconds(10));
CAF_CHECK_EQUAL(t.schedule().size(), 0u);
CAF_CHECK_EQUAL(t.actor_lookup().size(), 0u);
// Have AUT receive the timeout.
expect((error), from(aut).to(aut).with(sec::request_timeout));
}
CAF_TEST(mixed_receive_and_request_timeouts) {
// Have AUT call t.set_receive_timeout().
self->send(aut, ok_atom::value);
expect((ok_atom), from(self).to(aut).with(_));
CAF_CHECK_EQUAL(t.schedule().size(), 1u);
CAF_CHECK_EQUAL(t.actor_lookup().size(), 1u);
// Cause the request timeout to arrive later.
t.advance_time(seconds(5));
// Have AUT call t.set_request_timeout().
self->send(aut, add_atom::value);
expect((add_atom), from(self).to(aut).with(_));
CAF_CHECK_EQUAL(t.schedule().size(), 2u);
CAF_CHECK_EQUAL(t.actor_lookup().size(), 2u);
// Advance time to send receive timeout message.
t.advance_time(seconds(5));
CAF_CHECK_EQUAL(t.schedule().size(), 1u);
CAF_CHECK_EQUAL(t.actor_lookup().size(), 1u);
// Have AUT receive the timeout.
expect((timeout_msg), from(aut).to(aut).with(tid{42}));
// Advance time to send request timeout message.
t.advance_time(seconds(10));
CAF_CHECK_EQUAL(t.schedule().size(), 0u);
CAF_CHECK_EQUAL(t.actor_lookup().size(), 0u);
// Have AUT receive the timeout.
expect((error), from(aut).to(aut).with(sec::request_timeout));
}
CAF_TEST(delay_actor_message) {
// Schedule a message for now + 10s.
auto n = t.now() + seconds(10);
auto autptr = actor_cast<strong_actor_ptr>(aut);
t.schedule_message(n, autptr,
make_mailbox_element(autptr, message_id::make(),
no_stages, "foo"));
CAF_CHECK_EQUAL(t.schedule().size(), 1u);
CAF_CHECK_EQUAL(t.actor_lookup().size(), 0u);
// Advance time to send the message.
t.advance_time(seconds(10));
CAF_CHECK_EQUAL(t.schedule().size(), 0u);
CAF_CHECK_EQUAL(t.actor_lookup().size(), 0u);
// Have AUT receive the message.
expect((std::string), from(aut).to(aut).with("foo"));
}
CAF_TEST(delay_group_message) {
// Have AUT join the group.
auto grp = sys.groups().anonymous();
self->send(aut, grp);
expect((group), from(self).to(aut).with(_));
// Schedule a message for now + 10s.
auto n = t.now() + seconds(10);
auto autptr = actor_cast<strong_actor_ptr>(aut);
t.schedule_message(n, std::move(grp), autptr, make_message("foo"));
CAF_CHECK_EQUAL(t.schedule().size(), 1u);
CAF_CHECK_EQUAL(t.actor_lookup().size(), 0u);
// Advance time to send the message.
t.advance_time(seconds(10));
CAF_CHECK_EQUAL(t.schedule().size(), 0u);
CAF_CHECK_EQUAL(t.actor_lookup().size(), 0u);
// Have AUT receive the message.
expect((std::string), from(aut).to(aut).with("foo"));
// Kill AUT (necessary because the group keeps a reference around).
self->send_exit(aut, exit_reason::kill);
expect((exit_msg), from(self).to(aut).with(_));
}
CAF_TEST_FIXTURE_SCOPE_END()
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