Commit bdb7d450 authored by Dominik Charousset's avatar Dominik Charousset

Maintenance and coding style

parent 9f16ab2e
...@@ -122,7 +122,9 @@ class proper_actor_base : public Policies::resume_policy::template ...@@ -122,7 +122,9 @@ class proper_actor_base : public Policies::resume_policy::template
awaited_response); awaited_response);
} }
inline bool hidden() const { return this->m_hidden; } inline bool hidden() const {
return this->m_hidden;
}
void cleanup(uint32_t reason) override { void cleanup(uint32_t reason) override {
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(reason));
...@@ -167,9 +169,8 @@ class proper_actor_base : public Policies::resume_policy::template ...@@ -167,9 +169,8 @@ class proper_actor_base : public Policies::resume_policy::template
// this is the nonblocking version of proper_actor; it assumes that Base is // this is the nonblocking version of proper_actor; it assumes that Base is
// derived from local_actor and uses the behavior_stack_based mixin // derived from local_actor and uses the behavior_stack_based mixin
template <class Base, template <class Base, class Policies,
class Policies, bool OverrideDequeue = std::is_base_of<blocking_actor, Base>::value>
bool OverrideDequeue = std::is_base_of<blocking_actor, Base>::value>
class proper_actor class proper_actor
: public proper_actor_base<Base, proper_actor<Base, Policies, false>, : public proper_actor_base<Base, proper_actor<Base, Policies, false>,
Policies> { Policies> {
...@@ -179,10 +180,10 @@ class proper_actor ...@@ -179,10 +180,10 @@ class proper_actor
static_assert(std::is_base_of<local_actor, Base>::value, static_assert(std::is_base_of<local_actor, Base>::value,
"Base is not derived from local_actor"); "Base is not derived from local_actor");
template <class... Ts> template <class... Ts>
proper_actor(Ts&&... args) proper_actor(Ts&&... args) : super(std::forward<Ts>(args)...) {
: super(std::forward<Ts>(args)...) {} // nop
}
// required by event_based_resume::mixin::resume // required by event_based_resume::mixin::resume
...@@ -245,7 +246,6 @@ class proper_actor<Base, Policies, true> ...@@ -245,7 +246,6 @@ class proper_actor<Base, Policies, true>
if (!tmp_vec.empty()) { if (!tmp_vec.empty()) {
this->cache_prepend(tmp_vec.begin(), tmp_vec.end()); this->cache_prepend(tmp_vec.begin(), tmp_vec.end());
} }
}; };
while (!this->cache_empty()) { while (!this->cache_empty()) {
auto tmp = this->cache_take_first(); auto tmp = this->cache_take_first();
...@@ -270,15 +270,17 @@ class proper_actor<Base, Policies, true> ...@@ -270,15 +270,17 @@ class proper_actor<Base, Policies, true>
if (has_timeout) { if (has_timeout) {
auto e = pending_timeouts.end(); auto e = pending_timeouts.end();
auto i = std::find(pending_timeouts.begin(), e, timeout_id); auto i = std::find(pending_timeouts.begin(), e, timeout_id);
if (i != e) pending_timeouts.erase(i); if (i != e) {
pending_timeouts.erase(i);
}
} }
}); });
// read incoming messages // read incoming messages
for (;;) { for (;;) {
auto msg = this->next_message(); auto msg = this->next_message();
if (!msg) if (!msg) {
this->await_ready(); this->await_ready();
else { } else {
if (this->invoke_message(msg, bhvr, mid)) { if (this->invoke_message(msg, bhvr, mid)) {
// we're done // we're done
return; return;
...@@ -300,8 +302,9 @@ class proper_actor<Base, Policies, true> ...@@ -300,8 +302,9 @@ class proper_actor<Base, Policies, true>
this->m_host); this->m_host);
// auto e = this->new_mailbox_element(this, std::move(msg)); // auto e = this->new_mailbox_element(this, std::move(msg));
// this->m_mailbox.enqueue(e); // this->m_mailbox.enqueue(e);
} else } else {
this->delayed_send_tuple(this, d, std::move(msg)); this->delayed_send_tuple(this, d, std::move(msg));
}
m_pending_timeouts.push_back(tid); m_pending_timeouts.push_back(tid);
return tid; return tid;
} }
......
...@@ -220,14 +220,14 @@ class single_reader_queue { ...@@ -220,14 +220,14 @@ class single_reader_queue {
} }
/************************************************************************** /**************************************************************************
* support for synchronized access * * support for synchronized access *
**************************************************************************/ **************************************************************************/
template <class Mutex, class CondVar> template <class Mutex, class CondVar>
bool synchronized_enqueue(Mutex& mtx, CondVar& cv, pointer new_element) { bool synchronized_enqueue(Mutex& mtx, CondVar& cv, pointer new_element) {
std::unique_lock<Mutex> guard(mtx);
switch (enqueue(new_element)) { switch (enqueue(new_element)) {
case enqueue_result::unblocked_reader: { case enqueue_result::unblocked_reader: {
std::unique_lock<Mutex> guard(mtx);
cv.notify_one(); cv.notify_one();
return true; return true;
} }
...@@ -244,18 +244,18 @@ class single_reader_queue { ...@@ -244,18 +244,18 @@ class single_reader_queue {
template <class Mutex, class CondVar> template <class Mutex, class CondVar>
void synchronized_await(Mutex& mtx, CondVar& cv) { void synchronized_await(Mutex& mtx, CondVar& cv) {
std::unique_lock<Mutex> guard(mtx);
CAF_REQUIRE(!closed()); CAF_REQUIRE(!closed());
if (try_block()) { if (try_block()) {
std::unique_lock<Mutex> guard(mtx);
while (blocked()) cv.wait(guard); while (blocked()) cv.wait(guard);
} }
} }
template <class Mutex, class CondVar, class TimePoint> template <class Mutex, class CondVar, class TimePoint>
bool synchronized_await(Mutex& mtx, CondVar& cv, const TimePoint& timeout) { bool synchronized_await(Mutex& mtx, CondVar& cv, const TimePoint& timeout) {
std::unique_lock<Mutex> guard(mtx);
CAF_REQUIRE(!closed()); CAF_REQUIRE(!closed());
if (try_block()) { if (try_block()) {
std::unique_lock<Mutex> guard(mtx);
while (blocked()) { while (blocked()) {
if (cv.wait_until(guard, timeout) == std::cv_status::timeout) { if (cv.wait_until(guard, timeout) == std::cv_status::timeout) {
// if we're unable to set the queue from blocked to empty, // if we're unable to set the queue from blocked to empty,
......
...@@ -11,30 +11,28 @@ ...@@ -11,30 +11,28 @@
namespace caf { namespace caf {
namespace policy { namespace policy {
// this policy simply forwards calls to @p await_data to the scheduling
// policy and throws an exception whenever @p resume is called;
// it intentionally works only with the no_scheduling policy
class no_resume { class no_resume {
public: public:
template <class Base, class Derived> template <class Base, class Derived>
struct mixin : Base { struct mixin : Base {
template <class... Ts> template <class... Ts>
mixin(Ts&&... args) mixin(Ts&&... args) : Base(std::forward<Ts>(args)...), m_hidden(true) {
: Base(std::forward<Ts>(args)...), m_hidden(true) {} // nop
}
inline void attach_to_scheduler() { this->ref(); } void attach_to_scheduler() {
this->ref();
}
inline void detach_from_scheduler() { this->deref(); } void detach_from_scheduler() {
this->deref();
}
inline resumable::resume_result resume(execution_unit*) { resumable::resume_result resume(execution_unit*) {
auto done_cb = [=](uint32_t reason) { auto done_cb = [=](uint32_t reason) {
this->planned_exit_reason(reason); this->planned_exit_reason(reason);
this->on_exit(); this->on_exit();
this->cleanup(reason); this->cleanup(reason);
}; };
try { try {
this->act(); this->act();
...@@ -50,14 +48,12 @@ class no_resume { ...@@ -50,14 +48,12 @@ class no_resume {
} }
bool m_hidden; bool m_hidden;
}; };
template <class Actor> template <class Actor>
void await_ready(Actor* self) { void await_ready(Actor* self) {
self->await_data(); self->await_data();
} }
}; };
} // namespace policy } // namespace policy
......
...@@ -37,15 +37,17 @@ ...@@ -37,15 +37,17 @@
#include "caf/detail/sync_request_bouncer.hpp" #include "caf/detail/sync_request_bouncer.hpp"
#include "caf/detail/single_reader_queue.hpp" #include "caf/detail/single_reader_queue.hpp"
#include "caf/actor_ostream.hpp"
namespace caf { namespace caf {
namespace policy { namespace policy {
class no_scheduling { class no_scheduling {
using lock_type = std::unique_lock<std::mutex>; using lock_type = std::unique_lock<std::mutex>;
public: public:
using timeout_type = std::chrono::high_resolution_clock::time_point; using timeout_type = std::chrono::high_resolution_clock::time_point;
template <class Actor> template <class Actor>
...@@ -69,22 +71,21 @@ class no_scheduling { ...@@ -69,22 +71,21 @@ class no_scheduling {
intrusive_ptr<Actor> mself{self}; intrusive_ptr<Actor> mself{self};
self->attach_to_scheduler(); self->attach_to_scheduler();
std::thread([=] { std::thread([=] {
CAF_PUSH_AID(mself->id()); CAF_PUSH_AID(mself->id());
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
for (;;) { for (;;) {
if (mself->resume(nullptr) == resumable::done) { if (mself->resume(nullptr) == resumable::done) {
return; return;
} }
// await new data before resuming actor // await new data before resuming actor
await_data(mself.get()); await_data(mself.get());
CAF_REQUIRE(self->mailbox().blocked() == false); CAF_REQUIRE(self->mailbox().blocked() == false);
} }
self->detach_from_scheduler(); self->detach_from_scheduler();
}).detach(); }).detach();
} }
// await_data is being called from no_scheduling (only) // await_data is being called from no_resume (only)
template <class Actor> template <class Actor>
void await_data(Actor* self) { void await_data(Actor* self) {
if (self->has_next_message()) return; if (self->has_next_message()) return;
...@@ -100,10 +101,8 @@ class no_scheduling { ...@@ -100,10 +101,8 @@ class no_scheduling {
} }
private: private:
std::mutex m_mtx; std::mutex m_mtx;
std::condition_variable m_cv; std::condition_variable m_cv;
}; };
} // namespace policy } // namespace policy
......
...@@ -113,12 +113,10 @@ class timer_actor final : public detail::proper_actor<blocking_actor, ...@@ -113,12 +113,10 @@ class timer_actor final : public detail::proper_actor<blocking_actor,
on(atom("_Send"), arg_match) >> [&](const duration& d, on(atom("_Send"), arg_match) >> [&](const duration& d,
actor_addr& from, channel& to, actor_addr& from, channel& to,
message_id mid, message& tup) { message_id mid, message& tup) {
aout(this) << "new timeout requested!\n";
insert_dmsg(messages, d, std::move(from), insert_dmsg(messages, d, std::move(from),
std::move(to), mid, std::move(tup)); std::move(to), mid, std::move(tup));
}, },
[&](const exit_msg&) { [&](const exit_msg&) {
aout(this) << "DONE!\n";
done = true; done = true;
}, },
others() >> [&] { others() >> [&] {
...@@ -136,7 +134,6 @@ aout(this) << "DONE!\n"; ...@@ -136,7 +134,6 @@ aout(this) << "DONE!\n";
// handle timeouts (send messages) // handle timeouts (send messages)
auto it = messages.begin(); auto it = messages.begin();
while (it != messages.end() && (it->first) <= tout) { while (it != messages.end() && (it->first) <= tout) {
aout(this) << "deliver timeout!\n";
deliver(it->second); deliver(it->second);
messages.erase(it); messages.erase(it);
it = messages.begin(); it = messages.begin();
......
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