Commit 9f16ab2e authored by Dominik Charousset's avatar Dominik Charousset

Maintenance & coding style

parent 3a77d2f7
......@@ -59,33 +59,34 @@ enum class enqueue_result {
/**
* @brief An intrusive, thread-safe queue implementation.
* @note For implementation details see
* http://libcaf.blogspot.com/2011/04/mailbox-part-1.html
* http://libcppa.blogspot.com/2011/04/mailbox-part-1.html
*/
template <class T, class Delete = std::default_delete<T>>
class single_reader_queue {
public:
using value_type = T;
using pointer = value_type*;
/**
* @warning call only from the reader (owner)
*/
pointer try_pop() { return take_head(); }
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 {
auto loop = [&]() -> bool {
while (p) {
if (f(*p)) {
if (last == nullptr)
if (last == nullptr) {
m_head = p->next;
else
} else {
last = p->next;
}
m_delete(p);
return true;
} else {
......@@ -94,7 +95,6 @@ class single_reader_queue {
}
}
return false;
};
if (!loop()) {
// last points to the tail now
......@@ -114,7 +114,6 @@ class single_reader_queue {
}
}
// returns true if the queue was empty
enqueue_result enqueue(pointer new_element) {
pointer e = m_stack.load();
for (;;) {
......@@ -125,42 +124,45 @@ class single_reader_queue {
}
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;
return (e == reader_blocked_dummy()) ? enqueue_result::unblocked_reader
: enqueue_result::success;
}
}
}
/**
* @brief Queries whether there is new data to read.
* @pre m_stack.load() != reader_blocked_dummy()
* @pre !closed()
*/
inline bool can_fetch_more() {
bool can_fetch_more() {
auto ptr = m_stack.load();
CAF_REQUIRE(ptr != nullptr);
CAF_REQUIRE(!closed());
return !is_dummy(ptr);
}
/**
* @warning call only from the reader (owner)
* @warning Call only from the reader (owner).
*/
inline bool empty() {
CAF_REQUIRE(m_stack.load() != nullptr);
bool empty() {
CAF_REQUIRE(!closed());
return (!m_head && is_dummy(m_stack.load()));
}
inline bool closed() { return m_stack.load() == nullptr; }
bool closed() {
return m_stack.load() == nullptr;
}
inline bool blocked() { return m_stack == reader_blocked_dummy(); }
bool blocked() {
return m_stack.load() == reader_blocked_dummy();
}
/**
* @brief Tries to set this queue from state @p empty to state @p blocked.
* @returns @p true if the state change was successful or if the mailbox
* was already blocked, otherwise @p false.
* @note This function does never fail spuriously.
* was already blocked, otherwise @p false.
* @note This function never fails spuriously.
*/
inline bool try_block() {
bool try_block() {
auto e = stack_empty_dummy();
bool res = m_stack.compare_exchange_strong(e, reader_blocked_dummy());
// return true in case queue was already blocked
......@@ -170,41 +172,52 @@ class single_reader_queue {
/**
* @brief Tries to set this queue from state @p blocked to state @p empty.
* @returns @p true if the state change was successful, otherwise @p false.
* @note This function does never fail spuriously.
* @note This function never fails spuriously.
*/
inline bool try_unblock() {
bool try_unblock() {
auto e = reader_blocked_dummy();
return m_stack.compare_exchange_strong(e, stack_empty_dummy());
}
/**
* @warning call only from the reader (owner)
* @brief Closes this queue and deletes all remaining elements.
* @warning Call only from the reader (owner).
*/
// closes this queue deletes all remaining elements
inline void close() {
void close() {
clear_cached_elements();
if (fetch_new_data(nullptr)) clear_cached_elements();
if (fetch_new_data(nullptr)) {
clear_cached_elements();
}
}
// 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.
* @warning Call only from the reader (owner).
*/
template <class F>
inline void close(const F& f) {
void close(const F& f) {
clear_cached_elements(f);
if (fetch_new_data(nullptr)) clear_cached_elements(f);
if (fetch_new_data(nullptr)) {
clear_cached_elements(f);
}
}
inline single_reader_queue() : m_head(nullptr) {
single_reader_queue() : m_head(nullptr) {
m_stack = stack_empty_dummy();
}
inline void clear() {
void clear() {
if (!closed()) {
clear_cached_elements();
if (fetch_new_data()) clear_cached_elements();
if (fetch_new_data()) {
clear_cached_elements();
}
}
}
~single_reader_queue() { clear(); }
~single_reader_queue() {
clear();
}
/**************************************************************************
* support for synchronized access *
......@@ -229,26 +242,6 @@ class single_reader_queue {
CAF_CRITICAL("invalid result of enqueue()");
}
template <class Mutex, class CondVar, class TimePoint>
pointer synchronized_try_pop(Mutex& mtx, CondVar& cv,
const TimePoint& abs_time) {
auto res = try_pop();
if (!res && synchronized_await(mtx, cv, abs_time)) {
res = try_pop();
}
return res;
}
template <class Mutex, class CondVar>
pointer synchronized_pop(Mutex& mtx, CondVar& cv) {
auto res = try_pop();
if (!res) {
synchronized_await(mtx, cv);
res = try_pop();
}
return res;
}
template <class Mutex, class CondVar>
void synchronized_await(Mutex& mtx, CondVar& cv) {
CAF_REQUIRE(!closed());
......@@ -275,7 +268,6 @@ class single_reader_queue {
}
private:
// exposed to "outside" access
std::atomic<pointer> m_stack;
......@@ -314,7 +306,7 @@ class single_reader_queue {
return false;
}
inline 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()) {
......@@ -343,22 +335,21 @@ class single_reader_queue {
}
}
inline pointer stack_empty_dummy() {
pointer stack_empty_dummy() {
// we are *never* going to dereference the returned pointer;
// it is only used as indicator wheter this queue is closed or not
return reinterpret_cast<pointer>(this);
}
inline pointer reader_blocked_dummy() {
pointer reader_blocked_dummy() {
// we are not going to dereference this pointer either
return reinterpret_cast<pointer>(reinterpret_cast<intptr_t>(this) +
sizeof(void*));
return reinterpret_cast<pointer>(reinterpret_cast<intptr_t>(this)
+ sizeof(void*));
}
inline bool is_dummy(pointer ptr) {
bool is_dummy(pointer ptr) {
return ptr == stack_empty_dummy() || ptr == reader_blocked_dummy();
}
};
} // namespace detail
......
......@@ -50,7 +50,7 @@ class no_scheduling {
template <class Actor>
void enqueue(Actor* self, const actor_addr& sender, message_id mid,
message& msg, execution_unit*) {
message& msg, execution_unit*) {
auto ptr = self->new_mailbox_element(sender, mid, std::move(msg));
// returns false if mailbox has been closed
if (!self->mailbox().synchronized_enqueue(m_mtx, m_cv, ptr)) {
......
......@@ -72,6 +72,24 @@ void splice(std::string& str, const std::string& glue, T&& arg, Ts&&... args) {
splice(str, glue, std::forward<Ts>(args)...);
}
template <size_t WhatSize, size_t WithSize>
void replace_all(std::string& str,
const char (&what)[WhatSize],
const char (&with)[WithSize]) {
// end(what) - 1 points to the null-terminator
auto next = [&](std::string::iterator pos) -> std::string::iterator{
return std::search(pos, str.end(), std::begin(what), std::end(what) - 1);
};
auto i = next(std::begin(str));
while (i != std::end(str)) {
auto before = static_cast<size_t>(std::distance(std::begin(str), i));
str.replace(i, i + WhatSize - 1, with);
// i became invalidated -> use new iterator pointing
// to the first character after the replaced text
i = next(std::begin(str) + before + (WithSize - 1));
}
}
/**
* @brief Compares two values by using @p operator== unless two floating
* point numbers are compared. In the latter case, the function
......
......@@ -130,7 +130,7 @@ void local_actor::forward_message(const actor& dest, message_priority prio) {
}
void local_actor::send_tuple(message_priority prio, const channel& dest,
message what) {
message what) {
if (!dest) return;
message_id id;
if (prio == message_priority::high) id = id.with_high_priority();
......
......@@ -30,6 +30,8 @@
#include <sys/types.h>
#endif
#include "caf/string_algorithms.hpp"
#include "caf/all.hpp"
#include "caf/actor_proxy.hpp"
......@@ -43,27 +45,15 @@ namespace {
__thread actor_id t_self_id;
template <size_t RawSize>
void replace_all(std::string& str, const char (&before)[RawSize],
const char* after) {
// end(before) - 1 points to the null-terminator
auto i = std::search(std::begin(str), std::end(str), std::begin(before),
std::end(before) - 1);
while (i != std::end(str)) {
str.replace(i, i + RawSize - 1, after);
i = std::search(std::begin(str), std::end(str), std::begin(before),
std::end(before) - 1);
}
}
constexpr struct pop_aid_log_event_t {
constexpr pop_aid_log_event_t() {}
constexpr pop_aid_log_event_t() {
// nop
}
} pop_aid_log_event;
struct log_event {
log_event* next;
std::string msg;
};
#ifndef CAF_LOG_LEVEL
......@@ -73,12 +63,10 @@ struct log_event {
#endif
class logging_impl : public logging {
public:
void initialize() {
const char* log_level_lookup_table[] = {"ERROR", "WARN", "INFO",
"DEBUG", "TRACE"};
"DEBUG", "TRACE"};
m_thread = std::thread([this] { (*this)(); });
std::string msg = "ENTRY log level = ";
msg += log_level_lookup_table[global_log_level];
......@@ -99,18 +87,23 @@ class logging_impl : public logging {
std::fstream out(fname.str().c_str(), std::ios::out | std::ios::app);
std::unique_ptr<log_event> event;
for (;;) {
event.reset(m_queue.synchronized_pop(m_queue_mtx, m_queue_cv));
// make sure we have data to read
m_queue.synchronized_await(m_queue_mtx, m_queue_cv);
// read & process event
event.reset(m_queue.try_pop());
CAF_REQUIRE(event != nullptr);
if (event->msg.empty()) {
out.close();
return;
} else
} else {
out << event->msg << std::flush;
}
}
}
void log(const char* level, const char* c_class_name,
const char* function_name, const char* c_full_file_name,
int line_num, const std::string& msg) override {
const char* function_name, const char* c_full_file_name,
int line_num, const std::string& msg) override {
std::string class_name = c_class_name;
replace_all(class_name, "::", ".");
replace_all(class_name, "(anonymous namespace)", "$anon$");
......@@ -119,51 +112,53 @@ class logging_impl : public logging {
auto ri = find(full_file_name.rbegin(), full_file_name.rend(), '/');
if (ri != full_file_name.rend()) {
auto i = ri.base();
if (i == full_file_name.end())
if (i == full_file_name.end()) {
file_name = std::move(full_file_name);
else
} else {
file_name = std::string(i, full_file_name.end());
} else
}
} else {
file_name = std::move(full_file_name);
}
std::ostringstream line;
line << time(0) << " " << level << " "
<< "actor" << t_self_id << " " << std::this_thread::get_id() << " "
<< class_name << " " << function_name << " " << file_name << ":"
<< line_num << " " << msg << std::endl;
<< "actor" << t_self_id << " " << std::this_thread::get_id() << " "
<< class_name << " " << function_name << " " << file_name << ":"
<< line_num << " " << msg << std::endl;
m_queue.synchronized_enqueue(m_queue_mtx, m_queue_cv,
new log_event{nullptr, line.str()});
new log_event{nullptr, line.str()});
}
private:
std::thread m_thread;
std::mutex m_queue_mtx;
std::condition_variable m_queue_cv;
detail::single_reader_queue<log_event> m_queue;
};
} // namespace <anonymous>
logging::trace_helper::trace_helper(std::string class_name,
const char* fun_name, const char* file_name,
int line_num, const std::string& msg)
: m_class(std::move(class_name))
, m_fun_name(fun_name)
, m_file_name(file_name)
, m_line_num(line_num) {
const char* fun_name, const char* file_name,
int line_num, const std::string& msg)
: m_class(std::move(class_name)), m_fun_name(fun_name),
m_file_name(file_name), m_line_num(line_num) {
singletons::get_logger()->log("TRACE", m_class.c_str(), fun_name, file_name,
line_num, "ENTRY " + msg);
line_num, "ENTRY " + msg);
}
logging::trace_helper::~trace_helper() {
singletons::get_logger()->log("TRACE", m_class.c_str(), m_fun_name,
m_file_name, m_line_num, "EXIT");
m_file_name, m_line_num, "EXIT");
}
logging::~logging() {}
logging::~logging() {
// nop
}
logging* logging::create_singleton() { return new logging_impl; }
logging* logging::create_singleton() {
return new logging_impl;
}
actor_id logging::set_aid(actor_id aid) {
actor_id prev = t_self_id;
......
......@@ -43,6 +43,11 @@
#include "caf/detail/logging.hpp"
#include "caf/detail/proper_actor.hpp"
#include "caf/actor_ostream.hpp"
namespace caf {
namespace scheduler {
......@@ -108,10 +113,12 @@ class timer_actor final : public detail::proper_actor<blocking_actor,
on(atom("_Send"), arg_match) >> [&](const duration& d,
actor_addr& from, channel& to,
message_id mid, message& tup) {
aout(this) << "new timeout requested!\n";
insert_dmsg(messages, d, std::move(from),
std::move(to), mid, std::move(tup));
},
[&](const exit_msg&) {
aout(this) << "DONE!\n";
done = true;
},
others() >> [&] {
......@@ -129,6 +136,7 @@ class timer_actor final : public detail::proper_actor<blocking_actor,
// handle timeouts (send messages)
auto it = messages.begin();
while (it != messages.end() && (it->first) <= tout) {
aout(this) << "deliver timeout!\n";
deliver(it->second);
messages.erase(it);
it = messages.begin();
......
......@@ -26,6 +26,8 @@
#include <stdexcept>
#include <algorithm>
#include "caf/string_algorithms.hpp"
#include "caf/atom.hpp"
#include "caf/actor.hpp"
#include "caf/channel.hpp"
......@@ -44,18 +46,18 @@
#ifdef DEBUG_PARSER
namespace {
size_t s_indentation = 0;
size_t s_indentation = 0;
} // namespace <anonymous>
# define PARSER_INIT(message) \
std::cout << std::string(s_indentation, ' ') << ">>> " << message \
<< std::endl; \
s_indentation += 2; \
# define PARSER_INIT(message) \
std::cout << std::string(s_indentation, ' ') << ">>> " << message \
<< std::endl; \
s_indentation += 2; \
auto ____sg = caf::detail::make_scope_guard([] { s_indentation -= 2; })
# define PARSER_OUT(condition, message) \
if (condition) { \
std::cout << std::string(s_indentation, ' ') << "### " << message \
<< std::endl; \
} \
# define PARSER_OUT(condition, message) \
if (condition) { \
std::cout << std::string(s_indentation, ' ') << "### " << message \
<< std::endl; \
} \
static_cast<void>(0)
#else
# define PARSER_INIT(unused) static_cast<void>(0)
......@@ -73,48 +75,45 @@ struct platform_int_mapping {
const char* name;
size_t size;
bool is_signed;
};
// WARNING: this list is sorted and searched with std::lower_bound;
// keep ordered when adding elements!
platform_int_mapping platform_dependent_sizes[] = {
{"char", sizeof(char), true },
{"char16_t", sizeof(char16_t), true },
{"char32_t", sizeof(char32_t), true },
{"int", sizeof(int), true },
{"long", sizeof(long), true },
{"long int", sizeof(long int), true },
{"long long", sizeof(long long), true },
{"short", sizeof(short), true },
{"short int", sizeof(short int), true },
{"signed char", sizeof(signed char), true },
{"signed int", sizeof(signed int), true },
{"signed long", sizeof(signed long), true },
{"signed long int", sizeof(signed long int), true },
{"signed long long", sizeof(signed long long), true },
{"signed short", sizeof(signed short), true },
{"signed short int", sizeof(signed short int), true },
{"unsigned char", sizeof(unsigned char), false},
{"unsigned int", sizeof(unsigned int), false},
{"unsigned long", sizeof(unsigned long), false},
// keep ordered when adding elements!
constexpr platform_int_mapping platform_dependent_sizes[] = {
{"char", sizeof(char), true },
{"char16_t", sizeof(char16_t), true },
{"char32_t", sizeof(char32_t), true },
{"int", sizeof(int), true },
{"long", sizeof(long), true },
{"long int", sizeof(long int), true },
{"long long", sizeof(long long), true },
{"short", sizeof(short), true },
{"short int", sizeof(short int), true },
{"signed char", sizeof(signed char), true },
{"signed int", sizeof(signed int), true },
{"signed long", sizeof(signed long), true },
{"signed long int", sizeof(signed long int), true },
{"signed long long", sizeof(signed long long), true },
{"signed short", sizeof(signed short), true },
{"signed short int", sizeof(signed short int), true },
{"unsigned char", sizeof(unsigned char), false},
{"unsigned int", sizeof(unsigned int), false},
{"unsigned long", sizeof(unsigned long), false},
{"unsigned long int", sizeof(unsigned long int), false},
{"unsigned long long", sizeof(unsigned long long), false},
{"unsigned short", sizeof(unsigned short), false},
{"unsigned short", sizeof(unsigned short), false},
{"unsigned short int", sizeof(unsigned short int), false}
};
string map2decorated(string&& name) {
auto cmp = [](const platform_int_mapping& pim, const string& str) {
return strcmp(pim.name, str.c_str()) < 0;
};
auto e = end(platform_dependent_sizes);
auto i = lower_bound(begin(platform_dependent_sizes), e, name, cmp);
if (i != e && i->name == name) {
PARSER_OUT(true,
name << " => "
<< mapped_int_names[i->size][i->is_signed ? 1 : 0]);
PARSER_OUT(true, name << " => "
<< mapped_int_names[i->size][i->is_signed ? 1 : 0]);
return mapped_int_names[i->size][i->is_signed ? 1 : 0];
}
#ifdef DEBUG_PARSER
......@@ -127,21 +126,23 @@ string map2decorated(string&& name) {
}
class parse_tree {
public:
string compile(bool parent_invoked = false) {
string result;
propagate_flags();
if (!parent_invoked) {
if (m_volatile) result += "volatile ";
if (m_const) result += "const ";
if (m_volatile) {
result += "volatile ";
}
if (m_const) {
result += "const ";
}
}
if (has_children()) {
string sub_result = m_children.front().compile(true);
for (auto i = m_children.begin() + 1; i != m_children.end(); ++i) {
sub_result += "::";
sub_result += i->compile(true);
string sub_result;
for (auto& child : m_children) {
if (!sub_result.empty()) sub_result += "::";
sub_result += child.compile(true);
}
result += map2decorated(std::move(sub_result));
} else {
......@@ -303,7 +304,9 @@ class parse_tree {
} else if (token == "class" || token == "struct") {
// ignored (created by visual c++ compilers)
} else if (!token.empty()) {
if (!result.m_name.empty()) result.m_name += " ";
if (!result.m_name.empty()) {
result.m_name += " ";
}
result.m_name += token;
}
}
......@@ -323,7 +326,6 @@ class parse_tree {
}
private:
void propagate_flags() {
for (auto& c : m_children) {
c.propagate_flags();
......@@ -335,8 +337,9 @@ class parse_tree {
}
}
parse_tree() : m_const(false), m_pointer(false), m_volatile(false)
, m_lvalue_ref(false), m_rvalue_ref(false) {
parse_tree()
: m_const(false), m_pointer(false), m_volatile(false),
m_lvalue_ref(false), m_rvalue_ref(false) {
// nop
}
......@@ -349,7 +352,6 @@ class parse_tree {
string m_name;
vector<parse_tree> m_children;
vector<parse_tree> m_template_parameters;
};
template <class Iterator>
......@@ -379,26 +381,15 @@ vector<parse_tree> parse_tree::parse_tpl_args(Iterator first, Iterator last) {
return result;
}
template <size_t RawSize>
void replace_all(string& str, const char (&before)[RawSize],
const char* after) {
// end(before) - 1 points to the null-terminator
auto i = search(begin(str), end(str), begin(before), end(before) - 1);
while (i != end(str)) {
str.replace(i, i + RawSize - 1, after);
i = search(begin(str), end(str), begin(before), end(before) - 1);
}
}
const char s_rawan[] = "anonymous namespace";
const char s_an[] = "$";
const char raw_anonymous_namespace[] = "anonymous namespace";
const char unified_anonymous_namespace[] = "$";
} // namespace <anonymous>
std::string to_uniform_name(const std::string& dname) {
auto r = parse_tree::parse(begin(dname), end(dname)).compile();
// replace compiler-dependent "anonmyous namespace" with "@_"
replace_all(r, s_rawan, s_an);
replace_all(r, raw_anonymous_namespace, unified_anonymous_namespace);
return r.c_str();
}
......
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