Commit 8e7a1b40 authored by Dominik Charousset's avatar Dominik Charousset

Reformatted according to new coding style

parent 9b367680
......@@ -13,7 +13,6 @@ using namespace caf;
using std::endl;
int main() {
std::srand(static_cast<unsigned>(std::time(0)));
for (int i = 1; i <= 50; ++i) {
spawn<blocking_api>([i](blocking_actor* self) {
aout(self) << "Hi there! This is actor nr. "
......
......@@ -25,7 +25,6 @@
#include <memory>
#include <utility>
#include <typeinfo>
#include <iostream>
#include "caf/config.hpp"
#include "caf/ref_counted.hpp"
......
#ifndef PROPER_ACTOR_HPP
#define PROPER_ACTOR_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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 LICENCE_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_PROPER_ACTOR_HPP
#define CAF_DETAIL_PROPER_ACTOR_HPP
#include <type_traits>
......@@ -349,4 +368,4 @@ class proper_actor<Base, Policies, true>
} // namespace detail
} // namespace caf
#endif // PROPER_ACTOR_HPP
#endif // CAF_DETAIL_PROPER_ACTOR_HPP
......@@ -35,7 +35,6 @@ namespace detail {
* @brief Denotes in which state queue and reader are after an enqueue.
*/
enum class enqueue_result {
/**
* @brief Indicates that the enqueue operation succeeded and
* the reader is ready to receive the data.
......@@ -53,7 +52,6 @@ enum class enqueue_result {
* queue has been closed by the reader.
*/
queue_closed
};
/**
......@@ -68,53 +66,19 @@ class single_reader_queue {
using pointer = value_type*;
/**
* @warning call only from the reader (owner)
* @brief Tries to dequeue a new element from the mailbox.
* @warning Call only from the reader (owner).
*/
pointer try_pop() {
return take_head();
}
template <class UnaryPredicate>
void remove_if(UnaryPredicate f) {
pointer head = m_head;
pointer last = nullptr;
pointer p = m_head;
auto loop = [&]() -> bool {
while (p) {
if (f(*p)) {
if (last == nullptr) {
m_head = p->next;
} else {
last = p->next;
}
m_delete(p);
return true;
} else {
last = p;
p = p->next;
}
}
return false;
};
if (!loop()) {
// last points to the tail now
auto old_tail = last;
m_head = nullptr; // fetch_new_data assumes cached list to be empty
if (fetch_new_data()) {
last = nullptr;
p = m_head; // let p point to the first newly fetched element
loop();
// restore cached list
if (head) {
old_tail->next = m_head;
m_head = head;
}
} else
m_head = head;
}
}
/**
* @brief Tries to enqueue a new element to the mailbox.
* @warning Call only from the reader (owner).
*/
enqueue_result enqueue(pointer new_element) {
CAF_REQUIRE(new_element != nullptr);
pointer e = m_stack.load();
for (;;) {
if (!e) {
......@@ -122,36 +86,50 @@ class single_reader_queue {
m_delete(new_element);
return enqueue_result::queue_closed;
}
// a dummy is never part of a non-empty list
new_element->next = is_dummy(e) ? nullptr : e;
if (m_stack.compare_exchange_weak(e, new_element)) {
return (e == reader_blocked_dummy()) ? enqueue_result::unblocked_reader
: enqueue_result::success;
if (m_stack.compare_exchange_strong(e, new_element)) {
return (e == reader_blocked_dummy()) ? enqueue_result::unblocked_reader
: enqueue_result::success;
}
// continue with new value of e
}
}
/**
* @brief Queries whether there is new data to read.
* @brief Queries whether there is new data to read, i.e., whether the next
* call to {@link try_pop} would succeeed.
* @pre !closed()
*/
bool can_fetch_more() {
if (m_head != nullptr) {
return true;
}
auto ptr = m_stack.load();
CAF_REQUIRE(!closed());
CAF_REQUIRE(ptr != nullptr);
return !is_dummy(ptr);
}
/**
* @brief Queries whether this queue is empty.
* @warning Call only from the reader (owner).
*/
bool empty() {
CAF_REQUIRE(!closed());
return (!m_head && is_dummy(m_stack.load()));
return m_head == nullptr && is_dummy(m_stack.load());
}
/**
* @brief Queries whether this has been closed.
*/
bool closed() {
return m_stack.load() == nullptr;
}
/**
* @brief Queries whether this has been marked as blocked, i.e., the
* owner of the list is waiting for new data.
*/
bool blocked() {
return m_stack.load() == reader_blocked_dummy();
}
......@@ -165,6 +143,7 @@ class single_reader_queue {
bool try_block() {
auto e = stack_empty_dummy();
bool res = m_stack.compare_exchange_strong(e, reader_blocked_dummy());
CAF_REQUIRE(e != nullptr);
// return true in case queue was already blocked
return res || e == reader_blocked_dummy();
}
......@@ -191,7 +170,8 @@ class single_reader_queue {
}
/**
* @brief Closes this queue and applies f to all remaining elements before deleting.
* @brief Closes this queue and applies f to all remaining
* elements before deleting them.
* @warning Call only from the reader (owner).
*/
template <class F>
......@@ -225,9 +205,9 @@ class single_reader_queue {
template <class Mutex, class CondVar>
bool synchronized_enqueue(Mutex& mtx, CondVar& cv, pointer new_element) {
std::unique_lock<Mutex> guard(mtx);
switch (enqueue(new_element)) {
case enqueue_result::unblocked_reader: {
std::unique_lock<Mutex> guard(mtx);
cv.notify_one();
return true;
}
......@@ -244,18 +224,20 @@ class single_reader_queue {
template <class Mutex, class CondVar>
void synchronized_await(Mutex& mtx, CondVar& cv) {
std::unique_lock<Mutex> guard(mtx);
CAF_REQUIRE(!closed());
if (try_block()) {
while (blocked()) cv.wait(guard);
if (!can_fetch_more() && try_block()) {
std::unique_lock<Mutex> guard(mtx);
while (blocked()) {
cv.wait(guard);
}
}
}
template <class Mutex, class CondVar, class TimePoint>
bool synchronized_await(Mutex& mtx, CondVar& cv, const TimePoint& timeout) {
std::unique_lock<Mutex> guard(mtx);
CAF_REQUIRE(!closed());
if (try_block()) {
if (!can_fetch_more() && try_block()) {
std::unique_lock<Mutex> guard(mtx);
while (blocked()) {
if (cv.wait_until(guard, timeout) == std::cv_status::timeout) {
// if we're unable to set the queue from blocked to empty,
......@@ -278,7 +260,7 @@ class single_reader_queue {
// atomically sets m_stack back and enqueues all elements to the cache
bool fetch_new_data(pointer end_ptr) {
CAF_REQUIRE(m_head == nullptr);
CAF_REQUIRE(!end_ptr || end_ptr == stack_empty_dummy());
CAF_REQUIRE(end_ptr == nullptr || end_ptr == stack_empty_dummy());
pointer e = m_stack.load();
// must not be called on a closed queue
CAF_REQUIRE(e != nullptr);
......@@ -306,7 +288,9 @@ class single_reader_queue {
return false;
}
bool fetch_new_data() { return fetch_new_data(stack_empty_dummy()); }
bool fetch_new_data() {
return fetch_new_data(stack_empty_dummy());
}
pointer take_head() {
if (m_head != nullptr || fetch_new_data()) {
......
......@@ -119,21 +119,17 @@ class event_based_resume {
d->push_to_cache(std::move(ptr));
}
} else {
CAF_LOG_DEBUG(
"no more element in mailbox; "
"going to block");
CAF_LOG_DEBUG("no more element in mailbox; going to block");
if (d->mailbox().try_block()) {
return resumable::resume_later;
}
// else: try again
CAF_LOG_DEBUG("try_block() interrupted by new message");
}
}
}
catch (actor_exited& what) {
CAF_LOG_INFO(
"actor died because of exception: actor_exited, "
"reason = "
<< what.reason());
CAF_LOG_INFO("actor died because of exception: actor_exited, "
"reason = " << what.reason());
if (d->exit_reason() == exit_reason::not_exited) {
d->quit(what.reason());
}
......
......@@ -21,7 +21,6 @@
#define CAF_POLICY_INVOKE_POLICY_HPP
#include <memory>
#include <iostream>
#include <type_traits>
#include "caf/none.hpp"
......
......@@ -52,13 +52,21 @@ class not_prioritizing {
m_cache.push_back(std::move(ptr));
}
inline cache_iterator cache_begin() { return m_cache.begin(); }
inline cache_iterator cache_begin() {
return m_cache.begin();
}
inline cache_iterator cache_end() { return m_cache.end(); }
inline cache_iterator cache_end() {
return m_cache.end();
}
inline void cache_erase(cache_iterator iter) { m_cache.erase(iter); }
inline void cache_erase(cache_iterator iter) {
m_cache.erase(iter);
}
inline bool cache_empty() const { return m_cache.empty(); }
inline bool cache_empty() const {
return m_cache.empty();
}
inline unique_mailbox_element_pointer cache_take_first() {
auto tmp = std::move(m_cache.front());
......
......@@ -21,7 +21,6 @@
#define CAF_POLICY_PRIORITIZING_HPP
#include <list>
#include <iostream>
#include "caf/mailbox_element.hpp"
#include "caf/message_priority.hpp"
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the Boost Software License, Version 1.0. See *
* accompanying file LICENSE or copy at http://www.boost.org/LICENSE_1_0.txt *
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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 LICENCE_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/io/spawn_io.hpp"
......@@ -23,4 +24,4 @@ namespace caf {
// import io::spawn_io for backward compatbility reasons
using io::spawn_io;
} // namespace caf
\ No newline at end of file
} // namespace caf
......@@ -18,7 +18,6 @@
******************************************************************************/
#include <utility>
#include <iostream>
#include "caf/locks.hpp"
......
......@@ -22,7 +22,6 @@
#include <cstring>
#include <sstream>
#include <iterator>
#include <iostream>
#include <exception>
#include <stdexcept>
#include <type_traits>
......
......@@ -20,7 +20,6 @@
#include <set>
#include <mutex>
#include <sstream>
#include <iostream>
#include <stdexcept>
#include <condition_variable>
......
......@@ -21,8 +21,12 @@
namespace caf {
resumable::resumable() : m_hidden(true) {}
resumable::resumable() : m_hidden(true) {
// nop
}
resumable::~resumable() {}
resumable::~resumable() {
// nop
}
} // namespace caf
......@@ -32,6 +32,8 @@
#include "caf/scoped_actor.hpp"
#include "caf/system_messages.hpp"
#include "caf/actor_ostream.hpp"
#include "caf/policy/fork_join.hpp"
#include "caf/policy/no_resume.hpp"
#include "caf/policy/no_scheduling.hpp"
......@@ -86,10 +88,8 @@ inline void insert_dmsg(Map& storage, const duration& d, Ts&&... vs) {
}
class timer_actor final : public detail::proper_actor<blocking_actor,
timer_actor_policies> {
timer_actor_policies> {
public:
inline unique_mailbox_element_pointer dequeue() {
await_data();
return next_message();
......@@ -103,6 +103,24 @@ class timer_actor final : public detail::proper_actor<blocking_actor,
}
void act() override {
trap_exit(true);
bool i_am_done = false;
receive_while([&]{ return !i_am_done; })(
on(atom("_Send"), arg_match) >> [&](const duration& d,
actor_addr& from, channel& to,
message_id mid, message& tup) {
to->enqueue(from, mid, std::move(tup), nullptr);
},
[&](const exit_msg&) {
i_am_done = true;
},
others() >> [&] {
std::cerr << "coordinator::timer_loop: UNKNOWN MESSAGE: " << std::endl;
}
);
return;
// setup & local variables
bool done = false;
unique_mailbox_element_pointer msg_ptr;
......@@ -111,8 +129,8 @@ class timer_actor final : public detail::proper_actor<blocking_actor,
// message handling rules
message_handler mfun{
on(atom("_Send"), arg_match) >> [&](const duration& d,
actor_addr& from, channel& to,
message_id mid, message& tup) {
actor_addr& from, channel& to,
message_id mid, message& tup) {
insert_dmsg(messages, d, std::move(from),
std::move(to), mid, std::move(tup));
},
......@@ -213,12 +231,14 @@ void printer_loop(blocking_actor* self) {
******************************************************************************/
class shutdown_helper : public resumable {
public:
void attach_to_scheduler() override {
// nop
}
void attach_to_scheduler() override { }
void detach_from_scheduler() override { }
void detach_from_scheduler() override {
// nop
}
resumable::resume_result resume(execution_unit* ptr) {
CAF_LOG_DEBUG("shutdown_helper::resume => shutdown worker");
......@@ -230,14 +250,15 @@ class shutdown_helper : public resumable {
return resumable::shutdown_execution_unit;
}
shutdown_helper() : last_worker(nullptr) { }
shutdown_helper() : last_worker(nullptr) {
// nop
}
~shutdown_helper();
std::mutex mtx;
std::condition_variable cv;
abstract_worker* last_worker;
};
shutdown_helper::~shutdown_helper() {
......
......@@ -18,7 +18,6 @@
******************************************************************************/
#include <atomic>
#include <iostream>
#include "caf/message.hpp"
#include "caf/scheduler.hpp"
......
......@@ -21,7 +21,6 @@
#include <cctype>
#include <sstream>
#include <iomanip>
#include <iostream>
#include <algorithm>
#include "caf/string_algorithms.hpp"
......
......@@ -45,6 +45,7 @@
//#define DEBUG_PARSER
#ifdef DEBUG_PARSER
# include <iostream>
namespace {
size_t s_indentation = 0;
} // namespace <anonymous>
......
/******************************************************************************\
* *
* ____ _ _ _ *
* | __ ) ___ ___ ___| |_ / \ ___| |_ ___ _ __ *
* | _ \ / _ \ / _ \/ __| __| / _ \ / __| __/ _ \| '__| *
* | |_) | (_) | (_) \__ \ |_ _ / ___ \ (__| || (_) | | *
* |____/ \___/ \___/|___/\__(_)_/ \_\___|\__\___/|_| *
* *
* *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the Boost Software License, Version 1.0. See *
* accompanying file LICENSE or copy at http://www.boost.org/LICENSE_1_0.txt *
\ ******************************************************************************/
#include <iostream>
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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 LICENCE_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/none.hpp"
......@@ -32,9 +32,6 @@
#include "caf/detail/actor_registry.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
using std::cout;
using std::endl;
namespace caf {
namespace io {
......@@ -42,7 +39,9 @@ broker::servant::~servant() {
// nop
}
broker::servant::servant(broker* ptr) : m_disconnected(false), m_broker(ptr) { }
broker::servant::servant(broker* ptr) : m_disconnected(false), m_broker(ptr) {
// nop
}
void broker::servant::set_broker(broker* new_broker) {
if (!m_disconnected) m_broker = new_broker;
......@@ -143,9 +142,7 @@ void broker::doorman::io_failure(network::operation op) {
}
class broker::continuation {
public:
continuation(broker_ptr ptr, actor_addr from, message_id mid, message&& msg)
: m_self(std::move(ptr)), m_from(from)
, m_mid(mid), m_data(std::move(msg)) { }
......@@ -157,17 +154,14 @@ class broker::continuation {
}
private:
broker_ptr m_self;
actor_addr m_from;
message_id m_mid;
message m_data;
};
void broker::invoke_message(const actor_addr& sender,
message_id mid,
message& msg) {
message_id mid, message& msg) {
CAF_LOG_TRACE(CAF_TARG(msg, to_string));
m_running = true;
auto sg = detail::make_scope_guard([=] {
......@@ -268,12 +262,10 @@ void broker::write(connection_handle hdl, size_t bs, const void* buf) {
out.insert(out.end(), first, last);
}
void broker::enqueue(const actor_addr& sender,
message_id mid,
message msg,
execution_unit*) {
middleman::instance()->run_later(continuation{this, sender, mid,
std::move(msg)});
void broker::enqueue(const actor_addr& sender, message_id mid,
message msg, execution_unit*) {
middleman::instance()->run_later(continuation{this, sender,
mid, std::move(msg)});
}
bool broker::initialized() const {
......@@ -281,8 +273,8 @@ bool broker::initialized() const {
}
broker::broker()
: m_initialized(false), m_hidden(true)
, m_running(false), m_mm(*middleman::instance()) {
: m_initialized(false), m_hidden(true),
m_running(false), m_mm(*middleman::instance()) {
// nop
}
......@@ -318,7 +310,7 @@ void broker::launch(bool is_hidden, execution_unit*) {
}
);
self->enqueue(invalid_actor_addr, message_id::invalid,
make_message(atom("INITMSG")), nullptr);
make_message(atom("INITMSG")), nullptr);
}
void broker::configure_read(connection_handle hdl, receive_policy::config cfg) {
......
......@@ -22,7 +22,6 @@
#include <memory>
#include <cstring>
#include <sstream>
#include <iostream>
#include <stdexcept>
#include "caf/on.hpp"
......
/******************************************************************************\
* *
* ____ _ _ _ *
* | __ ) ___ ___ ___| |_ / \ ___| |_ ___ _ __ *
* | _ \ / _ \ / _ \/ __| __| / _ \ / __| __/ _ \| '__| *
* | |_) | (_) | (_) \__ \ |_ _ / ___ \ (__| || (_) | | *
* |____/ \___/ \___/|___/\__(_)_/ \_\___|\__\___/|_| *
* *
* *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the Boost Software License, Version 1.0. See *
* accompanying file LICENSE or copy at http://www.boost.org/LICENSE_1_0.txt *
\ ******************************************************************************/
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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 LICENCE_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/send.hpp"
#include "caf/to_string.hpp"
......
......@@ -157,6 +157,8 @@ The following table summarizes the changes made to the API.
\hline
\lstinline^any_tuple => message^ & This type is only being used to pass a message from one actor to another. Hence, \lstinline^message^ is the logical name. \\
\hline
\lstinline^partial_function => ^ \lstinline^message_handler^ & Technically, it still is a partial function, but wanted to emphasize its use case in the library. \\
\hline
\lstinline^cow_tuple => X^ & We want to provide a streamlined, simple API. Shipping a full tuple abstraction with the library does not fit into this philosophy. The removal of \lstinline^cow_tuple^ implies the removal of related functions such as \lstinline^tuple_cast^. \\
\hline
\lstinline^cow_ptr => X^ & This pointer class is an implementation detail of \lstinline^message^ and should not live in the global namespace in the first place. It also had the wrong name, because it is intrusive. \\
......
No preview for this file type
......@@ -106,18 +106,6 @@ int main() {
CAF_CHECK(announce1 == announce3);
CAF_CHECK(announce1 == announce4);
CAF_CHECK_EQUAL(announce1->name(), "$::foo");
{
/*
//bar.create_object();
auto obj1 = uniform_typeid<foo>()->create();
auto obj2(obj1);
CAF_CHECK(obj1 == obj2);
get_ref<foo>(obj1).value = 42;
CAF_CHECK(obj1 != obj2);
CAF_CHECK_EQUAL(get<foo>(obj1).value, 42);
CAF_CHECK_EQUAL(get<foo>(obj2).value, 0);
*/
}
{
auto uti = uniform_typeid<atom_value>();
CAF_CHECK(uti != nullptr);
......@@ -161,12 +149,12 @@ int main() {
io::middleman::instance();
// ok, check whether middleman announces its types correctly
check_types(append(expected,
"caf::io::accept_handle",
"caf::io::acceptor_closed_msg",
"caf::io::connection_handle",
"caf::io::connection_closed_msg",
"caf::io::new_connection_msg",
"caf::io::new_data_msg"));
"caf::io::accept_handle",
"caf::io::acceptor_closed_msg",
"caf::io::connection_handle",
"caf::io::connection_closed_msg",
"caf::io::new_connection_msg",
"caf::io::new_data_msg"));
}
return CAF_TEST_RESULT();
}
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