Commit 9860e195 authored by Dominik Charousset's avatar Dominik Charousset

SE-0001: Restructure logging and add cav-vec tool

parent 7e027c3d
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#ifndef CAF_ACTOR_CONFIG_HPP #ifndef CAF_ACTOR_CONFIG_HPP
#define CAF_ACTOR_CONFIG_HPP #define CAF_ACTOR_CONFIG_HPP
#include <string>
#include <functional> #include <functional>
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
...@@ -28,6 +29,7 @@ ...@@ -28,6 +29,7 @@
namespace caf { namespace caf {
/// Stores spawn-time flags and groups.
class actor_config { class actor_config {
public: public:
execution_unit* host; execution_unit* host;
...@@ -35,12 +37,7 @@ public: ...@@ -35,12 +37,7 @@ public:
input_range<const group>* groups; input_range<const group>* groups;
std::function<behavior (local_actor*)> init_fun; std::function<behavior (local_actor*)> init_fun;
explicit actor_config(execution_unit* ptr = nullptr) explicit actor_config(execution_unit* ptr = nullptr);
: host(ptr),
flags(abstract_channel::is_abstract_actor_flag),
groups(nullptr) {
// nop
}
inline actor_config& add_flag(int x) { inline actor_config& add_flag(int x) {
flags |= x; flags |= x;
...@@ -48,6 +45,9 @@ public: ...@@ -48,6 +45,9 @@ public:
} }
}; };
/// @relates actor_config
std::string to_string(const actor_config& x);
} // namespace caf } // namespace caf
#endif // CAF_ACTOR_CONFIG_HPP #endif // CAF_ACTOR_CONFIG_HPP
...@@ -122,8 +122,9 @@ std::string get_rtti_from_mpi(const uniform_type_info_map& types) { ...@@ -122,8 +122,9 @@ std::string get_rtti_from_mpi(const uniform_type_info_map& types) {
/// such as a middleman. /// such as a middleman.
class actor_system { class actor_system {
public: public:
friend class abstract_actor; friend class logger;
friend class io::middleman; friend class io::middleman;
friend class abstract_actor;
actor_system() = delete; actor_system() = delete;
actor_system(const actor_system&) = delete; actor_system(const actor_system&) = delete;
...@@ -407,8 +408,8 @@ public: ...@@ -407,8 +408,8 @@ public:
template <spawn_options Os = no_spawn_options, class F, class... Ts> template <spawn_options Os = no_spawn_options, class F, class... Ts>
infer_handle_from_fun_t<F> infer_handle_from_fun_t<F>
spawn_in_group(const group& grp, F fun, Ts&&... xs) { spawn_in_group(const group& grp, F fun, Ts&&... xs) {
return spawn_fun_in_groups<Os>({grp}, std::move(fun), return spawn_in_groups<Os>({grp}, std::move(fun),
std::forward<Ts>(xs)...); std::forward<Ts>(xs)...);
} }
/// Returns a new class-based actor subscribed to all groups in `gs`. /// Returns a new class-based actor subscribed to all groups in `gs`.
...@@ -492,10 +493,9 @@ private: ...@@ -492,10 +493,9 @@ private:
cfg.flags |= abstract_actor::is_detached_flag; cfg.flags |= abstract_actor::is_detached_flag;
if (!cfg.host) if (!cfg.host)
cfg.host = dummy_execution_unit(); cfg.host = dummy_execution_unit();
CAF_SET_LOGGER_SYS(this);
auto res = make_actor<C>(next_actor_id(), node(), this, auto res = make_actor<C>(next_actor_id(), node(), this,
cfg, std::forward<Ts>(xs)...); cfg, std::forward<Ts>(xs)...);
CAF_SET_LOGGER_SYS(this);
CAF_LOG_DEBUG("spawned actor:" << CAF_ARG(res.id()));
CAF_PUSH_AID(res->id()); CAF_PUSH_AID(res->id());
auto ptr = static_cast<C*>(actor_cast<abstract_actor*>(res)); auto ptr = static_cast<C*>(actor_cast<abstract_actor*>(res));
ptr->launch(cfg.host, has_lazy_init_flag(Os), has_hide_flag(Os)); ptr->launch(cfg.host, has_lazy_init_flag(Os), has_hide_flag(Os));
...@@ -505,7 +505,7 @@ private: ...@@ -505,7 +505,7 @@ private:
std::atomic<size_t> ids_; std::atomic<size_t> ids_;
uniform_type_info_map types_; uniform_type_info_map types_;
node_id node_; node_id node_;
caf::logger logger_; intrusive_ptr<caf::logger> logger_;
actor_registry registry_; actor_registry registry_;
group_manager groups_; group_manager groups_;
module_array modules_; module_array modules_;
...@@ -519,6 +519,9 @@ private: ...@@ -519,6 +519,9 @@ private:
mutable std::mutex detached_mtx; mutable std::mutex detached_mtx;
mutable std::condition_variable detached_cv; mutable std::condition_variable detached_cv;
actor_system_config& cfg_; actor_system_config& cfg_;
std::mutex logger_dtor_mtx_;
std::condition_variable logger_dtor_cv_;
volatile bool logger_dtor_done_;
}; };
} // namespace caf } // namespace caf
......
...@@ -102,7 +102,8 @@ public: ...@@ -102,7 +102,8 @@ public:
/// Picks up user-defined `to_string` functions. /// Picks up user-defined `to_string` functions.
template <class T> template <class T>
enable_if_tt<has_to_string<T>> consume(T& x) { enable_if_t<!std::is_pointer<T>::value && has_to_string<T>::value>
consume(T& x) {
result_ += to_string(x); result_ += to_string(x);
} }
......
...@@ -185,7 +185,7 @@ public: ...@@ -185,7 +185,7 @@ public:
/// Sends an exit message to `dest`. /// Sends an exit message to `dest`.
template <class ActorHandle> template <class ActorHandle>
void send_exit(const ActorHandle& dest, error reason) { void send_exit(const ActorHandle& dest, error reason) {
dest->eq_impl(message_id::make(), nullptr, context(), dest->eq_impl(message_id::make(), ctrl(), context(),
exit_msg{address(), std::move(reason)}); exit_msg{address(), std::move(reason)});
} }
......
...@@ -31,6 +31,7 @@ ...@@ -31,6 +31,7 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/unifyn.hpp" #include "caf/unifyn.hpp"
#include "caf/ref_counted.hpp"
#include "caf/abstract_actor.hpp" #include "caf/abstract_actor.hpp"
#include "caf/deep_to_string.hpp" #include "caf/deep_to_string.hpp"
...@@ -57,7 +58,7 @@ namespace caf { ...@@ -57,7 +58,7 @@ namespace caf {
/// Centrally logs events from all actors in an actor system. To enable /// Centrally logs events from all actors in an actor system. To enable
/// logging in your application, you need to define `CAF_LOG_LEVEL`. Per /// logging in your application, you need to define `CAF_LOG_LEVEL`. Per
/// default, the logger generates log4j compatible output. /// default, the logger generates log4j compatible output.
class logger { class logger : public ref_counted {
public: public:
friend class actor_system; friend class actor_system;
...@@ -96,9 +97,7 @@ public: ...@@ -96,9 +97,7 @@ public:
line_builder& operator<<(const T& x) { line_builder& operator<<(const T& x) {
if (!str_.empty()) if (!str_.empty())
str_ += " "; str_ += " ";
std::stringstream ss; str_ += deep_to_string(x);
ss << x;
str_ += ss.str();
behind_arg_ = false; behind_arg_ = false;
return *this; return *this;
} }
...@@ -170,6 +169,10 @@ private: ...@@ -170,6 +169,10 @@ private:
void stop(); void stop();
void log_prefix(std::ostream& out, int level, const char* component,
const std::string& class_name, const char* function_name,
const char* file_name, int line_num);
actor_system& system_; actor_system& system_;
int level_; int level_;
detail::shared_spinlock aids_lock_; detail::shared_spinlock aids_lock_;
...@@ -322,4 +325,41 @@ inline caf::actor_id caf_set_aid_dummy() { return 0; } ...@@ -322,4 +325,41 @@ inline caf::actor_id caf_set_aid_dummy() { return 0; }
#endif // CAF_LOG_LEVEL #endif // CAF_LOG_LEVEL
// -- Standardized CAF events according to SE-0001.
#define CAF_LOG_SPAWN_EVENT(aid, aargs) \
CAF_LOG_DEBUG("SPAWN ; ID =" << aid \
<< "; ARGS =" << deep_to_string(aargs).c_str())
#define CAF_LOG_INIT_EVENT(aName, aLazy, aHide) \
CAF_LOG_DEBUG("INIT ; NAME =" << aName << "; LAZY =" << aLazy \
<< "; HIDDEN =" << aHide)
/// Logs
#define CAF_LOG_SEND_EVENT(ptr) \
CAF_LOG_DEBUG("SEND ; TO =" \
<< deep_to_string(strong_actor_ptr{this->ctrl()}).c_str() \
<< "; FROM =" << deep_to_string(ptr->sender).c_str() \
<< "; STAGES =" << deep_to_string(ptr->stages).c_str() \
<< "; CONTENT =" << deep_to_string(ptr->content()).c_str())
#define CAF_LOG_RECEIVE_EVENT(ptr) \
CAF_LOG_DEBUG("RECEIVE ; FROM =" \
<< deep_to_string(ptr->sender).c_str() \
<< "; STAGES =" << deep_to_string(ptr->stages).c_str() \
<< "; CONTENT =" << deep_to_string(ptr->content()).c_str())
#define CAF_LOG_REJECT_EVENT() CAF_LOG_DEBUG("REJECT")
#define CAF_LOG_ACCEPT_EVENT() CAF_LOG_DEBUG("ACCEPT")
#define CAF_LOG_DROP_EVENT() CAF_LOG_DEBUG("DROP")
#define CAF_LOG_SKIP_EVENT() CAF_LOG_DEBUG("SKIP")
#define CAF_LOG_FINALIZE_EVENT() CAF_LOG_DEBUG("FINALIZE")
#define CAF_LOG_TERMINATE_EVENT(rsn) \
CAF_LOG_DEBUG("TERMINATE ; REASON =" << deep_to_string(rsn).c_str());
#endif // CAF_LOGGER_HPP #endif // CAF_LOGGER_HPP
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include <type_traits> #include <type_traits>
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/logger.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/infer_handle.hpp" #include "caf/infer_handle.hpp"
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
...@@ -33,6 +34,7 @@ namespace caf { ...@@ -33,6 +34,7 @@ namespace caf {
template <class T, class R = infer_handle_from_class_t<T>, class... Ts> template <class T, class R = infer_handle_from_class_t<T>, class... Ts>
R make_actor(actor_id aid, node_id nid, actor_system* sys, Ts&&... xs) { R make_actor(actor_id aid, node_id nid, actor_system* sys, Ts&&... xs) {
CAF_LOG_SPAWN_EVENT(aid, std::forward_as_tuple(xs...));
auto ptr = new actor_storage<T>(aid, std::move(nid), sys, auto ptr = new actor_storage<T>(aid, std::move(nid), sys,
std::forward<Ts>(xs)...); std::forward<Ts>(xs)...);
return {&(ptr->ctrl), false}; return {&(ptr->ctrl), false};
......
...@@ -70,12 +70,10 @@ protected: ...@@ -70,12 +70,10 @@ protected:
} }
void stop() override { void stop() override {
CAF_LOG_TRACE("");
// shutdown workers // shutdown workers
class shutdown_helper : public resumable, public ref_counted { class shutdown_helper : public resumable, public ref_counted {
public: public:
resumable::resume_result resume(execution_unit* ptr, size_t) override { resumable::resume_result resume(execution_unit* ptr, size_t) override {
CAF_LOG_DEBUG("shutdown_helper::resume => shutdown worker");
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
std::unique_lock<std::mutex> guard(mtx); std::unique_lock<std::mutex> guard(mtx);
last_worker = ptr; last_worker = ptr;
...@@ -104,7 +102,6 @@ protected: ...@@ -104,7 +102,6 @@ protected:
alive_workers.insert(worker_by_id(i)); alive_workers.insert(worker_by_id(i));
sh.ref(); // make sure reference count is high enough sh.ref(); // make sure reference count is high enough
} }
CAF_LOG_DEBUG("enqueue shutdown_helper into each worker");
while (!alive_workers.empty()) { while (!alive_workers.empty()) {
(*alive_workers.begin())->external_enqueue(&sh); (*alive_workers.begin())->external_enqueue(&sh);
// since jobs can be stolen, we cannot assume that we have // since jobs can be stolen, we cannot assume that we have
......
...@@ -55,7 +55,6 @@ public: ...@@ -55,7 +55,6 @@ public:
CAF_ASSERT(this_thread_.get_id() == std::thread::id{}); CAF_ASSERT(this_thread_.get_id() == std::thread::id{});
auto this_worker = this; auto this_worker = this;
this_thread_ = std::thread{[this_worker] { this_thread_ = std::thread{[this_worker] {
CAF_LOG_TRACE(CAF_ARG(this_worker->id()));
this_worker->run(); this_worker->run();
}}; }};
} }
...@@ -67,7 +66,6 @@ public: ...@@ -67,7 +66,6 @@ public:
/// source, i.e., from any other thread. /// source, i.e., from any other thread.
void external_enqueue(job_ptr job) { void external_enqueue(job_ptr job) {
CAF_ASSERT(job != nullptr); CAF_ASSERT(job != nullptr);
CAF_LOG_TRACE(CAF_ARG(id()) << CAF_ARG(id_of(job)));
policy_.external_enqueue(this, job); policy_.external_enqueue(this, job);
} }
...@@ -76,7 +74,6 @@ public: ...@@ -76,7 +74,6 @@ public:
/// @warning Must not be called from other threads. /// @warning Must not be called from other threads.
void exec_later(job_ptr job) override { void exec_later(job_ptr job) override {
CAF_ASSERT(job != nullptr); CAF_ASSERT(job != nullptr);
CAF_LOG_TRACE(CAF_ARG(id()) << CAF_ARG(id_of(job)));
policy_.internal_enqueue(this, job); policy_.internal_enqueue(this, job);
} }
...@@ -109,13 +106,11 @@ public: ...@@ -109,13 +106,11 @@ public:
private: private:
void run() { void run() {
CAF_SET_LOGGER_SYS(&system()); CAF_SET_LOGGER_SYS(&system());
CAF_LOG_TRACE(CAF_ARG(id_));
// scheduling loop // scheduling loop
for (;;) { for (;;) {
auto job = policy_.dequeue(this); auto job = policy_.dequeue(this);
CAF_ASSERT(job != nullptr); CAF_ASSERT(job != nullptr);
CAF_ASSERT(job->subtype() != resumable::io_actor); CAF_ASSERT(job->subtype() != resumable::io_actor);
CAF_LOG_DEBUG("resume actor:" << CAF_ARG(id_of(job)));
CAF_PUSH_AID_FROM_PTR(dynamic_cast<abstract_actor*>(job)); CAF_PUSH_AID_FROM_PTR(dynamic_cast<abstract_actor*>(job));
policy_.before_resume(this, job); policy_.before_resume(this, job);
auto res = job->resume(this, max_throughput_); auto res = job->resume(this, max_throughput_);
......
...@@ -83,6 +83,7 @@ void anon_send(const Dest& dest, Ts&&... xs) { ...@@ -83,6 +83,7 @@ void anon_send(const Dest& dest, Ts&&... xs) {
/// Anonymously sends `dest` an exit message. /// Anonymously sends `dest` an exit message.
template <class Dest> template <class Dest>
void anon_send_exit(const Dest& dest, exit_reason reason) { void anon_send_exit(const Dest& dest, exit_reason reason) {
CAF_LOG_TRACE(CAF_ARG(dest) << CAF_ARG(reason));
if (dest) if (dest)
dest->enqueue(nullptr, message_id::make(), dest->enqueue(nullptr, message_id::make(),
make_message(exit_msg{dest->address(), reason}), nullptr); make_message(exit_msg{dest->address(), reason}), nullptr);
......
...@@ -131,7 +131,16 @@ void replace_all(std::string& str, ...@@ -131,7 +131,16 @@ void replace_all(std::string& str,
template<size_t S> template<size_t S>
bool starts_with(const std::string& str, const char (&prefix)[S]) { bool starts_with(const std::string& str, const char (&prefix)[S]) {
return str.compare(0, S, prefix) == 0; return str.compare(0, S - 1, prefix) == 0;
}
template <size_t S>
bool ends_with(const std::string& str, const char (&suffix)[S]) {
auto n = str.size();
auto m = S - 1;
if (n >= m)
return str.compare(n - m, m, suffix) == 0;
return false;
} }
template <class T> template <class T>
......
...@@ -127,6 +127,10 @@ public: ...@@ -127,6 +127,10 @@ public:
} }
} }
} }
const char* name() const override {
return "timer_actor";
}
}; };
using string_sink = std::function<void (std::string&&)>; using string_sink = std::function<void (std::string&&)>;
...@@ -222,77 +226,88 @@ sink_handle get_sink_handle(actor_system& sys, sink_cache& fc, ...@@ -222,77 +226,88 @@ sink_handle get_sink_handle(actor_system& sys, sink_cache& fc,
return {}; return {};
} }
void printer_loop(blocking_actor* self) { class printer_actor : public blocking_actor {
struct actor_data { public:
std::string current_line; printer_actor(actor_config& cfg) : blocking_actor(cfg) {
sink_handle redirect; // nop
actor_data() { }
// nop
} void act() override {
}; struct actor_data {
using data_map = std::unordered_map<actor_id, actor_data>; std::string current_line;
sink_cache fcache; sink_handle redirect;
sink_handle global_redirect; actor_data() {
data_map data; // nop
auto get_data = [&](actor_id addr, bool insert_missing) -> actor_data* { }
if (addr == invalid_actor_id) };
using data_map = std::unordered_map<actor_id, actor_data>;
sink_cache fcache;
sink_handle global_redirect;
data_map data;
auto get_data = [&](actor_id addr, bool insert_missing) -> actor_data* {
if (addr == invalid_actor_id)
return nullptr;
auto i = data.find(addr);
if (i == data.end() && insert_missing)
i = data.emplace(addr, actor_data{}).first;
if (i != data.end())
return &(i->second);
return nullptr; return nullptr;
auto i = data.find(addr); };
if (i == data.end() && insert_missing) auto flush = [&](actor_data* what, bool forced) {
i = data.emplace(addr, actor_data{}).first; if (!what)
if (i != data.end())
return &(i->second);
return nullptr;
};
auto flush = [&](actor_data* what, bool forced) {
if (!what)
return;
auto& line = what->current_line;
if (line.empty() || (line.back() != '\n' && !forced))
return;
if (what->redirect)
(*what->redirect)(std::move(line));
else if (global_redirect)
(*global_redirect)(std::move(line));
else
std::cout << line << std::flush;
line.clear();
};
bool done = false;
self->do_receive(
[&](add_atom, actor_id aid, std::string& str) {
if (str.empty() || aid == invalid_actor_id)
return; return;
auto d = get_data(aid, true); auto& line = what->current_line;
if (d) { if (line.empty() || (line.back() != '\n' && !forced))
d->current_line += str; return;
flush(d, false); if (what->redirect)
} (*what->redirect)(std::move(line));
}, else if (global_redirect)
[&](flush_atom, actor_id aid) { (*global_redirect)(std::move(line));
flush(get_data(aid, false), true); else
}, std::cout << line << std::flush;
[&](delete_atom, actor_id aid) { line.clear();
auto data_ptr = get_data(aid, false); };
if (data_ptr) { bool done = false;
flush(data_ptr, true); do_receive(
data.erase(aid); [&](add_atom, actor_id aid, std::string& str) {
if (str.empty() || aid == invalid_actor_id)
return;
auto d = get_data(aid, true);
if (d) {
d->current_line += str;
flush(d, false);
}
},
[&](flush_atom, actor_id aid) {
flush(get_data(aid, false), true);
},
[&](delete_atom, actor_id aid) {
auto data_ptr = get_data(aid, false);
if (data_ptr) {
flush(data_ptr, true);
data.erase(aid);
}
},
[&](redirect_atom, const std::string& fn, int flag) {
global_redirect = get_sink_handle(system(), fcache, fn, flag);
},
[&](redirect_atom, actor_id aid, const std::string& fn, int flag) {
auto d = get_data(aid, true);
if (d)
d->redirect = get_sink_handle(system(), fcache, fn, flag);
},
[&](exit_msg& em) {
fail_state(std::move(em.reason));
done = true;
} }
}, ).until([&] { return done; });
[&](redirect_atom, const std::string& fn, int flag) { }
global_redirect = get_sink_handle(self->system(), fcache, fn, flag);
}, const char* name() const override {
[&](redirect_atom, actor_id aid, const std::string& fn, int flag) { return "printer_actor";
auto d = get_data(aid, true); }
if (d) };
d->redirect = get_sink_handle(self->system(), fcache, fn, flag);
},
[&](exit_msg& em) {
self->fail_state(std::move(em.reason));
done = true;
}
).until([&] { return done; });
}
} // namespace <anonymous> } // namespace <anonymous>
...@@ -308,7 +323,7 @@ void abstract_coordinator::start() { ...@@ -308,7 +323,7 @@ void abstract_coordinator::start() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// launch utility actors // launch utility actors
timer_ = actor_cast<strong_actor_ptr>(system_.spawn<timer_actor, hidden + detached>()); timer_ = actor_cast<strong_actor_ptr>(system_.spawn<timer_actor, hidden + detached>());
printer_ = actor_cast<strong_actor_ptr>(system_.spawn<hidden + detached>(printer_loop)); printer_ = actor_cast<strong_actor_ptr>(system_.spawn<printer_actor, hidden + detached>());
} }
void abstract_coordinator::init(actor_system_config& cfg) { void abstract_coordinator::init(actor_system_config& cfg) {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* 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 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_config.hpp"
namespace caf {
actor_config::actor_config(execution_unit* ptr)
: host(ptr),
flags(abstract_channel::is_abstract_actor_flag),
groups(nullptr) {
// nop
}
std::string to_string(const actor_config&) {
// TODO: print flags and groups
return "actor_config";
}
} // namespace caf
...@@ -52,7 +52,7 @@ struct kvstate { ...@@ -52,7 +52,7 @@ struct kvstate {
} }
}; };
const char* kvstate::name = "caf.config_server"; const char* kvstate::name = "config_server";
behavior config_serv_impl(stateful_actor<kvstate>* self) { behavior config_serv_impl(stateful_actor<kvstate>* self) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
...@@ -144,7 +144,13 @@ behavior config_serv_impl(stateful_actor<kvstate>* self) { ...@@ -144,7 +144,13 @@ behavior config_serv_impl(stateful_actor<kvstate>* self) {
}; };
} }
behavior spawn_serv_impl(event_based_actor* self) { struct spawn_serv_state {
static const char* name;
};
const char* spawn_serv_state::name = "spawn_server";
behavior spawn_serv_impl(stateful_actor<spawn_serv_state>* self) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
return { return {
[=](spawn_atom, const std::string& name, [=](spawn_atom, const std::string& name,
...@@ -178,14 +184,15 @@ actor_system::module::~module() { ...@@ -178,14 +184,15 @@ actor_system::module::~module() {
actor_system::actor_system(actor_system_config& cfg) actor_system::actor_system(actor_system_config& cfg)
: ids_(0), : ids_(0),
types_(*this), types_(*this),
logger_(*this), logger_(new caf::logger(*this), false),
registry_(*this), registry_(*this),
groups_(*this), groups_(*this),
middleman_(nullptr), middleman_(nullptr),
dummy_execution_unit_(this), dummy_execution_unit_(this),
await_actors_before_shutdown_(true), await_actors_before_shutdown_(true),
detached(0), detached(0),
cfg_(cfg) { cfg_(cfg),
logger_dtor_done_(false) {
CAF_SET_LOGGER_SYS(this); CAF_SET_LOGGER_SYS(this);
for (auto& f : cfg.module_factories) { for (auto& f : cfg.module_factories) {
auto mod_ptr = f(*this); auto mod_ptr = f(*this);
...@@ -242,8 +249,6 @@ actor_system::actor_system(actor_system_config& cfg) ...@@ -242,8 +249,6 @@ actor_system::actor_system(actor_system_config& cfg)
if (mod) if (mod)
mod->init(cfg); mod->init(cfg);
groups_.init(cfg); groups_.init(cfg);
// start logger before spawning actors (since that uses the logger)
logger_.start();
// spawn config and spawn servers (lazily to not access the scheduler yet) // spawn config and spawn servers (lazily to not access the scheduler yet)
static constexpr auto Flags = hidden + lazy_init; static constexpr auto Flags = hidden + lazy_init;
spawn_serv_ = actor_cast<strong_actor_ptr>(spawn<Flags>(spawn_serv_impl)); spawn_serv_ = actor_cast<strong_actor_ptr>(spawn<Flags>(spawn_serv_impl));
...@@ -256,9 +261,11 @@ actor_system::actor_system(actor_system_config& cfg) ...@@ -256,9 +261,11 @@ actor_system::actor_system(actor_system_config& cfg)
if (mod) if (mod)
mod->start(); mod->start();
groups_.start(); groups_.start();
logger_->start();
} }
actor_system::~actor_system() { actor_system::~actor_system() {
CAF_LOG_DEBUG("shutdown actor system");
if (await_actors_before_shutdown_) if (await_actors_before_shutdown_)
await_all_actors_done(); await_all_actors_done();
// shutdown system-level servers // shutdown system-level servers
...@@ -277,8 +284,12 @@ actor_system::~actor_system() { ...@@ -277,8 +284,12 @@ actor_system::~actor_system() {
(*i)->stop(); (*i)->stop();
await_detached_threads(); await_detached_threads();
registry_.stop(); registry_.stop();
logger_.stop(); // reset logger and wait until dtor was called
CAF_SET_LOGGER_SYS(nullptr); CAF_SET_LOGGER_SYS(nullptr);
logger_.reset();
std::unique_lock<std::mutex> guard{logger_dtor_mtx_};
while (!logger_dtor_done_)
logger_dtor_cv_.wait(guard);
} }
/// Returns the host-local identifier for this system. /// Returns the host-local identifier for this system.
...@@ -293,7 +304,7 @@ scheduler::abstract_coordinator& actor_system::scheduler() { ...@@ -293,7 +304,7 @@ scheduler::abstract_coordinator& actor_system::scheduler() {
} }
caf::logger& actor_system::logger() { caf::logger& actor_system::logger() {
return logger_; return *logger_;
} }
actor_registry& actor_system::registry() { actor_registry& actor_system::registry() {
......
...@@ -59,14 +59,21 @@ blocking_actor::~blocking_actor() { ...@@ -59,14 +59,21 @@ blocking_actor::~blocking_actor() {
} }
void blocking_actor::enqueue(mailbox_element_ptr ptr, execution_unit*) { void blocking_actor::enqueue(mailbox_element_ptr ptr, execution_unit*) {
CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(getf(is_blocking_flag));
CAF_LOG_TRACE(CAF_ARG(*ptr));
CAF_LOG_SEND_EVENT(ptr);
auto mid = ptr->mid; auto mid = ptr->mid;
auto src = ptr->sender; auto src = ptr->sender;
// returns false if mailbox has been closed // returns false if mailbox has been closed
if (!mailbox().synchronized_enqueue(mtx_, cv_, ptr.release())) { if (!mailbox().synchronized_enqueue(mtx_, cv_, ptr.release())) {
CAF_LOG_REJECT_EVENT();
if (mid.is_request()) { if (mid.is_request()) {
detail::sync_request_bouncer srb{exit_reason()}; detail::sync_request_bouncer srb{exit_reason()};
srb(src, mid); srb(src, mid);
} }
} else {
CAF_LOG_ACCEPT_EVENT();
} }
} }
...@@ -75,6 +82,7 @@ const char* blocking_actor::name() const { ...@@ -75,6 +82,7 @@ const char* blocking_actor::name() const {
} }
void blocking_actor::launch(execution_unit*, bool, bool hide) { void blocking_actor::launch(execution_unit*, bool, bool hide) {
CAF_LOG_INIT_EVENT(name(), false, hide);
CAF_LOG_TRACE(CAF_ARG(hide)); CAF_LOG_TRACE(CAF_ARG(hide));
CAF_ASSERT(getf(is_blocking_flag)); CAF_ASSERT(getf(is_blocking_flag));
if (!hide) if (!hide)
...@@ -85,6 +93,8 @@ void blocking_actor::launch(execution_unit*, bool, bool hide) { ...@@ -85,6 +93,8 @@ void blocking_actor::launch(execution_unit*, bool, bool hide) {
auto this_ptr = ptr->get(); auto this_ptr = ptr->get();
CAF_ASSERT(dynamic_cast<blocking_actor*>(this_ptr) != 0); CAF_ASSERT(dynamic_cast<blocking_actor*>(this_ptr) != 0);
auto self = static_cast<blocking_actor*>(this_ptr); auto self = static_cast<blocking_actor*>(this_ptr);
CAF_SET_LOGGER_SYS(ptr->home_system);
CAF_PUSH_AID_FROM_PTR(self);
error rsn; error rsn;
# ifndef CAF_NO_EXCEPTIONS # ifndef CAF_NO_EXCEPTIONS
try { try {
...@@ -332,18 +342,21 @@ void blocking_actor::receive_impl(receive_cond& rcc, ...@@ -332,18 +342,21 @@ void blocking_actor::receive_impl(receive_cond& rcc,
skipped = false; skipped = false;
timed_out = false; timed_out = false;
auto& x = seq.value(); auto& x = seq.value();
CAF_LOG_RECEIVE_EVENT((&x));
// skip messages that don't match our message ID // skip messages that don't match our message ID
if ((mid.valid() && mid != x.mid) if ((mid.valid() && mid != x.mid)
|| (!mid.valid() && x.mid.is_response())) { || (!mid.valid() && x.mid.is_response())) {
skipped = true; skipped = true;
CAF_LOG_SKIP_EVENT();
} else { } else {
// blocking actors can use nested receives => restore current_element_ // blocking actors can use nested receives => restore current_element_
auto prev_element = current_element_; auto prev_element = current_element_;
current_element_ = &x; current_element_ = &x;
switch (bhvr.nested(visitor, x.content())) { switch (bhvr.nested(visitor, x.content())) {
case match_case::skip: case match_case::skip:
skipped = true; skipped = true;
break; CAF_LOG_SKIP_EVENT();
break;
default: default:
break; break;
case match_case::no_match: { case match_case::no_match: {
...@@ -353,6 +366,7 @@ void blocking_actor::receive_impl(receive_cond& rcc, ...@@ -353,6 +366,7 @@ void blocking_actor::receive_impl(receive_cond& rcc,
// get a match on the second (error) handler // get a match on the second (error) handler
if (sres.flag != rt_skip) { if (sres.flag != rt_skip) {
visitor.visit(sres); visitor.visit(sres);
CAF_LOG_FINALIZE_EVENT();
} else if (mid.valid()) { } else if (mid.valid()) {
// invoke again with an unexpected_response error // invoke again with an unexpected_response error
auto& old = *current_element_; auto& old = *current_element_;
...@@ -362,8 +376,10 @@ void blocking_actor::receive_impl(receive_cond& rcc, ...@@ -362,8 +376,10 @@ void blocking_actor::receive_impl(receive_cond& rcc,
std::move(old.stages), err}; std::move(old.stages), err};
current_element_ = &tmp; current_element_ = &tmp;
bhvr.nested(tmp.content()); bhvr.nested(tmp.content());
CAF_LOG_FINALIZE_EVENT();
} else { } else {
skipped = true; skipped = true;
CAF_LOG_SKIP_EVENT();
} }
} }
} }
......
...@@ -54,11 +54,7 @@ local_actor::~local_actor() { ...@@ -54,11 +54,7 @@ local_actor::~local_actor() {
} }
void local_actor::on_destroy() { void local_actor::on_destroy() {
// disable logging from this point on, because on_destroy can CAF_PUSH_AID_FROM_PTR(this);
// be called after the logger is already destroyed;
// alternatively, we would have to use a reference-counted,
// heap-allocated logger
CAF_SET_LOGGER_SYS(nullptr);
if (!getf(is_cleaned_up_flag)) { if (!getf(is_cleaned_up_flag)) {
on_exit(); on_exit();
cleanup(exit_reason::unreachable, nullptr); cleanup(exit_reason::unreachable, nullptr);
...@@ -190,6 +186,7 @@ bool local_actor::cleanup(error&& fail_state, execution_unit* host) { ...@@ -190,6 +186,7 @@ bool local_actor::cleanup(error&& fail_state, execution_unit* host) {
// tell registry we're done // tell registry we're done
unregister_from_system(); unregister_from_system();
monitorable_actor::cleanup(std::move(fail_state), host); monitorable_actor::cleanup(std::move(fail_state), host);
CAF_LOG_TERMINATE_EVENT(fail_state)
return true; return true;
} }
......
...@@ -58,29 +58,54 @@ constexpr const char* log_level_name[] = { ...@@ -58,29 +58,54 @@ constexpr const char* log_level_name[] = {
}; };
#ifdef CAF_LOG_LEVEL #ifdef CAF_LOG_LEVEL
static_assert(CAF_LOG_LEVEL >= 0 && CAF_LOG_LEVEL <= 4, static_assert(CAF_LOG_LEVEL >= 0 && CAF_LOG_LEVEL <= 4,
"assertion: 0 <= CAF_LOG_LEVEL <= 4"); "assertion: 0 <= CAF_LOG_LEVEL <= 4");
#ifdef CAF_MSVC #ifdef CAF_MSVC
thread_local
#else
__thread
#endif
actor_system* current_logger_system_ptr = nullptr;
inline actor_system* current_logger_system() { thread_local intrusive_ptr<logger> current_logger;
return current_logger_system_ptr;
}
inline void current_logger_system(actor_system* x) { inline void set_current_logger(logger* x) {
current_logger_system_ptr = x; current_logger.reset(x);
} }
inline logger* get_current_logger() { inline logger* get_current_logger() {
auto sys = current_logger_system(); return current_logger.get();
return sys ? &sys->logger() : nullptr; }
#else // CAF_MSVC
pthread_key_t s_key;
pthread_once_t s_key_once = PTHREAD_ONCE_INIT;
void logger_ptr_destructor(void* ptr) {
if (ptr) {
intrusive_ptr_release(reinterpret_cast<logger*>(ptr));
}
}
void make_logger_ptr() {
pthread_key_create(&s_key, logger_ptr_destructor);
}
void set_current_logger(logger* x) {
pthread_once(&s_key_once, make_logger_ptr);
logger_ptr_destructor(pthread_getspecific(s_key));
if (x)
intrusive_ptr_add_ref(x);
pthread_setspecific(s_key, x);
} }
#else
logger* get_current_logger() {
pthread_once(&s_key_once, make_logger_ptr);
return reinterpret_cast<logger*>(pthread_getspecific(s_key));
}
#endif // CAF_MSVC
#else // CAF_LOG_LEVEL
inline void current_logger_system(actor_system*) { inline void current_logger_system(actor_system*) {
// nop // nop
} }
...@@ -88,7 +113,7 @@ inline void current_logger_system(actor_system*) { ...@@ -88,7 +113,7 @@ inline void current_logger_system(actor_system*) {
inline logger* get_current_logger() { inline logger* get_current_logger() {
return nullptr; return nullptr;
} }
#endif #endif // CAF_LOG_LEVEL
void prettify_type_name(std::string& class_name) { void prettify_type_name(std::string& class_name) {
//replace_all(class_name, " ", ""); //replace_all(class_name, " ", "");
...@@ -108,6 +133,8 @@ void prettify_type_name(std::string& class_name) { ...@@ -108,6 +133,8 @@ void prettify_type_name(std::string& class_name) {
}; };
char prefix1[] = "caf.detail.embedded<"; char prefix1[] = "caf.detail.embedded<";
strip_magic(prefix1, prefix1 + (sizeof(prefix1) - 1)); strip_magic(prefix1, prefix1 + (sizeof(prefix1) - 1));
// finally, replace any whitespace with %20
replace_all(class_name, " ", "%20");
} }
void prettify_type_name(std::string& class_name, const char* c_class_name) { void prettify_type_name(std::string& class_name, const char* c_class_name) {
...@@ -222,13 +249,10 @@ actor_id logger::thread_local_aid(actor_id aid) { ...@@ -222,13 +249,10 @@ actor_id logger::thread_local_aid(actor_id aid) {
return 0; // was empty before return 0; // was empty before
} }
void logger::log(int level, const char* component, void logger::log_prefix(std::ostream& out, int level, const char* component,
const std::string& class_name, const char* function_name, const std::string& class_name,
const char* c_full_file_name, int line_num, const char* function_name, const char* c_full_file_name,
const std::string& msg) { int line_num) {
CAF_ASSERT(level >= 0 && level <= 4);
if (level > level_)
return;
std::string file_name; std::string file_name;
std::string full_file_name = c_full_file_name; std::string full_file_name = c_full_file_name;
auto ri = find(full_file_name.rbegin(), full_file_name.rend(), '/'); auto ri = find(full_file_name.rbegin(), full_file_name.rend(), '/');
...@@ -242,17 +266,32 @@ void logger::log(int level, const char* component, ...@@ -242,17 +266,32 @@ void logger::log(int level, const char* component,
file_name = std::move(full_file_name); file_name = std::move(full_file_name);
} }
std::ostringstream prefix; std::ostringstream prefix;
prefix << timestamp_to_string(make_timestamp()) << " " << component << " " out << timestamp_to_string(make_timestamp()) << " " << component << " "
<< log_level_name[level] << " " << log_level_name[level] << " "
<< "actor" << thread_local_aid() << " " << std::this_thread::get_id() << "actor" << thread_local_aid() << " " << std::this_thread::get_id()
<< " " << class_name << " " << function_name << " " << file_name << ":" << " " << class_name << " " << function_name << " " << file_name << ":"
<< line_num; << line_num;
}
void logger::log(int level, const char* component,
const std::string& class_name, const char* function_name,
const char* c_full_file_name, int line_num,
const std::string& msg) {
CAF_ASSERT(level >= 0 && level <= 4);
if (level > level_)
return;
std::ostringstream prefix;
log_prefix(prefix, level, component, class_name, function_name,
c_full_file_name, line_num);
queue_.synchronized_enqueue(queue_mtx_, queue_cv_, queue_.synchronized_enqueue(queue_mtx_, queue_cv_,
new event{level, component, prefix.str(), msg}); new event{level, component, prefix.str(), msg});
} }
void logger::set_current_actor_system(actor_system* x) { void logger::set_current_actor_system(actor_system* x) {
current_logger_system(x); if (x)
set_current_logger(&x->logger());
else
set_current_logger(nullptr);
} }
logger* logger::current_logger() { logger* logger::current_logger() {
...@@ -270,7 +309,11 @@ void logger::log_static(int level, const char* component, ...@@ -270,7 +309,11 @@ void logger::log_static(int level, const char* component,
} }
logger::~logger() { logger::~logger() {
// nop stop();
// tell system our dtor is done
std::unique_lock<std::mutex> guard{system_.logger_dtor_mtx_};
system_.logger_dtor_done_ = true;
system_.logger_dtor_cv_.notify_one();
} }
logger::logger(actor_system& sys) : system_(sys) { logger::logger(actor_system& sys) : system_(sys) {
...@@ -299,6 +342,42 @@ void logger::run() { ...@@ -299,6 +342,42 @@ void logger::run() {
f.replace(i, i + sizeof(node) - 1, nid); f.replace(i, i + sizeof(node) - 1, nid);
} }
std::fstream file(f, std::ios::out | std::ios::app); std::fstream file(f, std::ios::out | std::ios::app);
if (!file) {
std::cerr << "unable to open log file " << f << std::endl;
return;
}
// log first entry
auto lvl_atom = system_.config().logger_verbosity;
switch (static_cast<uint64_t>(lvl_atom)) {
case error_log_lvl_atom::uint_value():
level_ = CAF_LOG_LEVEL_ERROR;
break;
case warning_log_lvl_atom::uint_value():
level_ = CAF_LOG_LEVEL_WARNING;
break;
case info_log_lvl_atom::uint_value():
level_ = CAF_LOG_LEVEL_INFO;
break;
case debug_log_lvl_atom::uint_value():
level_ = CAF_LOG_LEVEL_DEBUG;
break;
case trace_log_lvl_atom::uint_value():
level_ = CAF_LOG_LEVEL_TRACE;
break;
default: {
constexpr atom_value level_names[] = {
error_log_lvl_atom::value, warning_log_lvl_atom::value,
info_log_lvl_atom::value, debug_log_lvl_atom::value,
trace_log_lvl_atom::value};
lvl_atom = level_names[CAF_LOG_LEVEL];
level_ = CAF_LOG_LEVEL;
}
}
log_prefix(file, CAF_LOG_LEVEL_INFO, "caf", "caf.logger", "start",
__FILE__, __LINE__);
file << " level = " << to_string(lvl_atom) << ", node = " << to_string(system_.node())
<< std::endl;
// receive log entries from other threads and actors
std::unique_ptr<event> ptr; std::unique_ptr<event> ptr;
for (;;) { for (;;) {
// make sure we have data to read // make sure we have data to read
...@@ -306,10 +385,9 @@ void logger::run() { ...@@ -306,10 +385,9 @@ void logger::run() {
// read & process event // read & process event
ptr.reset(queue_.try_pop()); ptr.reset(queue_.try_pop());
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
if (ptr->msg.empty()) { // empty message means: shut down
file.close(); if (ptr->msg.empty())
return; break;
}
file << ptr->prefix << ' ' << ptr->msg << std::endl; file << ptr->prefix << ' ' << ptr->msg << std::endl;
// TODO: once we've phased out GCC 4.8, we can upgarde this to a regex. // TODO: once we've phased out GCC 4.8, we can upgarde this to a regex.
if (!system_.config().logger_filter.empty() if (!system_.config().logger_filter.empty()
...@@ -340,44 +418,17 @@ void logger::run() { ...@@ -340,44 +418,17 @@ void logger::run() {
std::clog << ptr->msg << color(reset) << std::endl; std::clog << ptr->msg << color(reset) << std::endl;
} }
} }
log_prefix(file, CAF_LOG_LEVEL_INFO, "caf", "caf.logger", "stop",
__FILE__, __LINE__);
file << " EOF" << std::endl;
file.close();
} }
void logger::start() { void logger::start() {
#if defined(CAF_LOG_LEVEL) #if defined(CAF_LOG_LEVEL)
auto lvl_atom = system_.config().logger_verbosity; if (system_.config().logger_verbosity == quiet_log_lvl_atom::value)
switch (static_cast<uint64_t>(lvl_atom)) { return;
case quiet_log_lvl_atom::uint_value():
return;
case error_log_lvl_atom::uint_value():
level_ = CAF_LOG_LEVEL_ERROR;
break;
case warning_log_lvl_atom::uint_value():
level_ = CAF_LOG_LEVEL_WARNING;
break;
case info_log_lvl_atom::uint_value():
level_ = CAF_LOG_LEVEL_INFO;
break;
case debug_log_lvl_atom::uint_value():
level_ = CAF_LOG_LEVEL_DEBUG;
break;
case trace_log_lvl_atom::uint_value():
level_ = CAF_LOG_LEVEL_TRACE;
break;
default: {
constexpr atom_value level_names[] = {
error_log_lvl_atom::value, warning_log_lvl_atom::value,
info_log_lvl_atom::value, debug_log_lvl_atom::value,
trace_log_lvl_atom::value};
lvl_atom = level_names[CAF_LOG_LEVEL];
level_ = CAF_LOG_LEVEL;
}
}
thread_ = std::thread{[this] { this->run(); }}; thread_ = std::thread{[this] { this->run(); }};
std::string msg = "ENTRY level = ";
msg += to_string(lvl_atom);
msg += ", node = ";
msg += to_string(system_.node());
log(CAF_LOG_LEVEL_INFO, "caf", "caf::logger", "run", __FILE__, __LINE__, msg);
#endif #endif
} }
...@@ -385,8 +436,6 @@ void logger::stop() { ...@@ -385,8 +436,6 @@ void logger::stop() {
#if defined(CAF_LOG_LEVEL) #if defined(CAF_LOG_LEVEL)
if (!thread_.joinable()) if (!thread_.joinable())
return; return;
log(CAF_LOG_LEVEL_INFO, "caf", "caf::logger", "run", __FILE__, __LINE__,
"EXIT");
// an empty string means: shut down // an empty string means: shut down
queue_.synchronized_enqueue(queue_mtx_, queue_cv_, new event); queue_.synchronized_enqueue(queue_mtx_, queue_cv_, new event);
thread_.join(); thread_.join();
......
...@@ -35,6 +35,7 @@ private_thread::private_thread(scheduled_actor* self) ...@@ -35,6 +35,7 @@ private_thread::private_thread(scheduled_actor* self)
void private_thread::run() { void private_thread::run() {
auto job = const_cast<scheduled_actor*>(self_); auto job = const_cast<scheduled_actor*>(self_);
CAF_SET_LOGGER_SYS(&job->system());
CAF_PUSH_AID(job->id()); CAF_PUSH_AID(job->id());
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
scoped_execution_unit ctx{&job->system()}; scoped_execution_unit ctx{&job->system()};
......
...@@ -118,16 +118,16 @@ scheduled_actor::~scheduled_actor() { ...@@ -118,16 +118,16 @@ scheduled_actor::~scheduled_actor() {
// -- overridden functions of abstract_actor ----------------------------------- // -- overridden functions of abstract_actor -----------------------------------
void scheduled_actor::enqueue(mailbox_element_ptr ptr, void scheduled_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) {
execution_unit* eu) {
CAF_PUSH_AID(id());
CAF_LOG_TRACE(CAF_ARG(*ptr));
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(!getf(is_blocking_flag)); CAF_ASSERT(!getf(is_blocking_flag));
CAF_LOG_TRACE(CAF_ARG(*ptr));
CAF_LOG_SEND_EVENT(ptr);
auto mid = ptr->mid; auto mid = ptr->mid;
auto sender = ptr->sender; auto sender = ptr->sender;
switch (mailbox().enqueue(ptr.release())) { switch (mailbox().enqueue(ptr.release())) {
case detail::enqueue_result::unblocked_reader: { case detail::enqueue_result::unblocked_reader: {
CAF_LOG_ACCEPT_EVENT();
// add a reference count to this actor and re-schedule it // add a reference count to this actor and re-schedule it
intrusive_ptr_add_ref(ctrl()); intrusive_ptr_add_ref(ctrl());
if (getf(is_detached_flag)) { if (getf(is_detached_flag)) {
...@@ -142,6 +142,7 @@ void scheduled_actor::enqueue(mailbox_element_ptr ptr, ...@@ -142,6 +142,7 @@ void scheduled_actor::enqueue(mailbox_element_ptr ptr,
break; break;
} }
case detail::enqueue_result::queue_closed: { case detail::enqueue_result::queue_closed: {
CAF_LOG_REJECT_EVENT();
if (mid.is_request()) { if (mid.is_request()) {
detail::sync_request_bouncer f{exit_reason()}; detail::sync_request_bouncer f{exit_reason()};
f(sender, mid); f(sender, mid);
...@@ -150,6 +151,7 @@ void scheduled_actor::enqueue(mailbox_element_ptr ptr, ...@@ -150,6 +151,7 @@ void scheduled_actor::enqueue(mailbox_element_ptr ptr,
} }
case detail::enqueue_result::success: case detail::enqueue_result::success:
// enqueued to a running actors' mailbox; nothing to do // enqueued to a running actors' mailbox; nothing to do
CAF_LOG_ACCEPT_EVENT();
break; break;
} }
} }
...@@ -161,6 +163,7 @@ const char* scheduled_actor::name() const { ...@@ -161,6 +163,7 @@ const char* scheduled_actor::name() const {
} }
void scheduled_actor::launch(execution_unit* eu, bool lazy, bool hide) { void scheduled_actor::launch(execution_unit* eu, bool lazy, bool hide) {
CAF_LOG_INIT_EVENT(name(), lazy, hide);
CAF_LOG_TRACE(CAF_ARG(lazy) << CAF_ARG(hide)); CAF_LOG_TRACE(CAF_ARG(lazy) << CAF_ARG(hide));
CAF_ASSERT(!getf(is_blocking_flag)); CAF_ASSERT(!getf(is_blocking_flag));
if (!hide) if (!hide)
...@@ -376,6 +379,7 @@ scheduled_actor::categorize(mailbox_element& x) { ...@@ -376,6 +379,7 @@ scheduled_actor::categorize(mailbox_element& x) {
invoke_message_result scheduled_actor::consume(mailbox_element& x) { invoke_message_result scheduled_actor::consume(mailbox_element& x) {
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
current_element_ = &x; current_element_ = &x;
CAF_LOG_RECEIVE_EVENT(current_element_);
// short-circuit awaited responses // short-circuit awaited responses
if (!awaited_responses_.empty()) { if (!awaited_responses_.empty()) {
auto& pr = awaited_responses_.front(); auto& pr = awaited_responses_.front();
......
...@@ -49,22 +49,19 @@ scoped_actor::scoped_actor(actor_system& sys, bool hide) : context_(&sys) { ...@@ -49,22 +49,19 @@ scoped_actor::scoped_actor(actor_system& sys, bool hide) : context_(&sys) {
actor_config cfg{&context_}; actor_config cfg{&context_};
self_ = make_actor<impl, strong_actor_ptr>(sys.next_actor_id(), sys.node(), self_ = make_actor<impl, strong_actor_ptr>(sys.next_actor_id(), sys.node(),
&sys, cfg); &sys, cfg);
if (!hide) prev_ = CAF_SET_AID(self_->id());
prev_ = CAF_SET_AID(self_->id()); CAF_LOG_INIT_EVENT("scoped_actor", false, hide);
CAF_LOG_TRACE(CAF_ARG(hide));
if (!hide) if (!hide)
ptr()->register_at_system(); ptr()->register_at_system();
} }
scoped_actor::~scoped_actor() { scoped_actor::~scoped_actor() {
CAF_LOG_TRACE("");
if (!self_) if (!self_)
return; return;
auto x = ptr(); auto x = ptr();
if (x->getf(abstract_actor::is_registered_flag))
CAF_SET_AID(prev_);
if (!x->getf(abstract_actor::is_terminated_flag)) if (!x->getf(abstract_actor::is_terminated_flag))
x->cleanup(exit_reason::normal, &context_); x->cleanup(exit_reason::normal, &context_);
CAF_SET_AID(prev_);
} }
blocking_actor* scoped_actor::ptr() const { blocking_actor* scoped_actor::ptr() const {
......
...@@ -47,6 +47,7 @@ void abstract_broker::enqueue(mailbox_element_ptr ptr, execution_unit*) { ...@@ -47,6 +47,7 @@ void abstract_broker::enqueue(mailbox_element_ptr ptr, execution_unit*) {
} }
void abstract_broker::launch(execution_unit* eu, bool is_lazy, bool is_hidden) { void abstract_broker::launch(execution_unit* eu, bool is_lazy, bool is_hidden) {
CAF_LOG_INIT_EVENT(name(), is_lazy, is_hidden);
CAF_ASSERT(eu != nullptr); CAF_ASSERT(eu != nullptr);
CAF_ASSERT(eu == &backend()); CAF_ASSERT(eu == &backend());
// add implicit reference count held by middleman/multiplexer // add implicit reference count held by middleman/multiplexer
......
...@@ -978,7 +978,6 @@ event_handler::event_handler(default_multiplexer& dm, native_socket sockfd) ...@@ -978,7 +978,6 @@ event_handler::event_handler(default_multiplexer& dm, native_socket sockfd)
fd_(sockfd), fd_(sockfd),
read_channel_closed_(false), read_channel_closed_(false),
backend_(dm) { backend_(dm) {
CAF_LOG_TRACE(CAF_ARG(sockfd));
set_fd_flags(); set_fd_flags();
} }
......
...@@ -27,3 +27,5 @@ if(WIN32) ...@@ -27,3 +27,5 @@ if(WIN32)
else() else()
add(caf-run) add(caf-run)
endif() endif()
add(caf-vec)
#include <cctype>
#include <string>
#include <vector>
#include <utility>
#include <cassert>
#include <fstream>
#include <iostream>
#include "caf/all.hpp"
using std::cout;
using std::cerr;
using std::endl;
using std::string;
using namespace caf;
using thread_id = std::string;
using vector_timestamp = std::vector<size_t>;
// -- convenience functions for strings
// removes leading and trailing whitespaces
void trim(string& s) {
auto not_space = [](char c) { return !isspace(c); };
// trim left
s.erase(s.begin(), find_if(s.begin(), s.end(), not_space));
// trim right
s.erase(find_if(s.rbegin(), s.rend(), not_space).base(), s.end());
}
// -- convenience functions for I/O streams
using istream_fun = std::function<std::istream& (std::istream&)>;
std::istream& skip_whitespaces(std::istream& in) {
while (in.peek() == ' ')
in.get();
return in;
}
std::istream& skip_to_next_line(std::istream& in) {
in.ignore(std::numeric_limits<std::streamsize>::max(), '\n');
return in;
}
std::istream& skip_word(std::istream& in) {
skip_whitespaces(in);
auto nonspace = [](char x) { return isprint(x) && !isspace(x); };
while (nonspace(in.peek()))
in.get();
return in;
}
struct line_reader {
std::string& line;
char delim;
};
std::istream& operator>>(std::istream& in, line_reader x) {
std::getline(in, x.line, x.delim);
trim(x.line);
return in;
}
line_reader rd_line(std::string& line, char delim = '\n') {
return {line, delim};
}
struct istream_char_consumer {
const char* what;
size_t count;
};
std::istream& operator>>(std::istream& in, istream_char_consumer x) {
if (!in)
return in;
// ignore leading whitespaces
skip_whitespaces(in);
// ignore trailing '\0'
for (size_t i = 0; i < x.count; ++i) {
//cout << "in: " << (char) in.peek() << ", x: " << x.what[i] << endl;
if (in.get() != x.what[i]) {
in.setstate(std::ios::failbit);
break;
}
}
return in;
}
template <size_t S>
istream_char_consumer consume(const char (&what)[S]) {
return {what, S - 1};
}
// -- convenience functions for vector timestamps
vector_timestamp& merge(vector_timestamp& x, const vector_timestamp& y) {
assert(x.size() == y.size());
for (size_t i = 0; i < x.size(); ++i)
x[i] = std::max(x[i], y[i]);
return x;
}
constexpr const char* log_level_name[] = {"ERROR", "WARN", "INFO",
"DEBUG", "TRACE", "?????"};
enum class log_level { error, warn, info, debug, trace, invalid };
std::ostream& operator<<(std::ostream& out, const log_level& lvl) {
return out << log_level_name[static_cast<size_t>(lvl)];
}
std::istream& operator>>(std::istream& in, log_level& lvl) {
std::string tmp;
in >> tmp;
auto pred = [&](const char* cstr) {
return cstr == tmp;
};
auto b = std::begin(log_level_name);
auto e = std::end(log_level_name);
auto i = std::find_if(b, e, pred);
if (i == e)
lvl = log_level::invalid;
else
lvl = static_cast<log_level>(std::distance(b, i));
return in;
}
/// The ID of entities as used in a logfile. If the logger field is "actor0"
/// then this line represents a thread. Otherwise, the thread field is ignored.
struct logger_id {
/// Content of the [LOGGER] field (0 if logger is a thread).
actor_id aid;
/// Content of the [THREAD] field.
string tid;
};
bool operator<(const logger_id& x, const logger_id& y) {
return x.aid == 0 && y.aid == 0 ? x.tid < y.tid : x.aid < y.aid;
}
std::istream& operator>>(std::istream& in, logger_id& x) {
return in >> consume("actor") >> x.aid >> skip_whitespaces >> x.tid;
}
std::istream& operator>>(std::istream& in, node_id& x) {
in >> skip_whitespaces;
if (in.peek() == 'i') {
x = node_id{};
return in >> consume("invalid-node");
}
string node_hex_id;
uint32_t pid;
if (in >> rd_line(node_hex_id, '#') >> pid) {
x = node_id{pid, node_hex_id};
}
return in;
}
/// The ID of a mailbox in a logfile. Parsed from `<actor>@<node>` entries.
struct mailbox_id {
/// Actor ID of the receiver.
actor_id aid;
/// Node ID of the receiver.
node_id nid;
};
std::string to_string(const mailbox_id& x) {
auto res = std::to_string(x.aid);
res += '@';
res += to_string(x.nid);
return res;
}
std::istream& operator>>(std::istream& in, mailbox_id& x) {
// format is <actor>@<node>
return in >> x.aid >> consume("@") >> x.nid;
}
std::ostream& operator<<(std::ostream& out, const mailbox_id& x) {
return out << x.aid << '@' << to_string(x.nid);
}
/// An entity in our distributed system, i.e., either an actor or a thread.
struct entity {
/// The ID of this entity if it is an actor, otherwise 0.
actor_id aid;
/// The ID of this entity if it is a thread, otherwise empty.
thread_id tid;
/// The ID of the node this entity is running at.
node_id nid;
/// The ID of this node in the vector clock.
size_t vid;
/// Marks system-level actors to enable filtering.
bool hidden;
/// A human-redable name, e.g., "actor42" or "thread23".
string pretty_name;
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, entity& x) {
return f(meta::type_name("entity"), x.aid, x.tid, x.nid, x.vid, x.hidden,
x.pretty_name);
}
mailbox_id to_mailbox_id(const entity& x) {
if (x.aid == 0)
throw std::runtime_error("threads do not have a mailbox ID");
return {x.aid, x.nid};
}
logger_id to_logger_id(const entity& x) {
return {x.aid, x.tid};
}
/// Sorts entities by `nid` first, then places threads before actors
/// and finally compares `aid` or `tid`.
bool operator<(const entity& x, const entity& y) {
// We sort by node ID first.
auto cres = x.nid.compare(y.nid);
if (cres != 0)
return cres < 0;
return (x.aid == 0 && y.aid == 0) ? x.tid < y.tid : x.aid < y.aid;
}
/// Set of `entity` sorted in ascending order by node ID, actor ID,
/// and thread ID (in that order).
using entity_set = std::set<entity>;
class entity_set_range {
public:
using iterator = entity_set::const_iterator;
entity_set_range() = default;
entity_set_range(const entity_set_range&) = default;
entity_set_range& operator=(const entity_set_range&) = default;
iterator begin() const {
return begin_;
}
iterator end() const {
return end_;
}
protected:
iterator begin_;
iterator end_;
};
struct node_cmp_t {
bool operator()(const entity& x, const node_id& y) const {
return x.nid < y;
};
bool operator()(const node_id& x, const entity& y) const {
return x < y.nid;
};
};
constexpr node_cmp_t node_cmp = node_cmp_t{};
/// Range within an `entity_set` containing all entities for a given actor.
struct actor_cmp_t {
bool operator()(const entity& x, actor_id y) const {
return x.aid < y;
};
bool operator()(actor_id x, const entity& y) const {
return x < y.aid;
};
};
constexpr actor_cmp_t actor_cmp = actor_cmp_t{};
class node_range : public entity_set_range {
public:
node_range(const entity_set& xs, const node_id& y) {
// get range for the node
using namespace std;
begin_ = lower_bound(xs.begin(), xs.end(), y, node_cmp);
end_ = upper_bound(begin_, xs.end(), y, node_cmp);
}
node_range(const node_range&) = default;
node_range& operator=(const node_range&) = default;
const node_id& node() const {
return node_;
}
private:
node_id node_;
};
/// Range within an `entity_set` containing all entities for a given node.
class thread_range : public entity_set_range {
public:
thread_range(const node_range& xs) : node_(xs.node()) {
actor_id dummy = 0;
// get range for the node
using namespace std;
begin_ = xs.begin();
end_ = upper_bound(begin_, xs.end(), dummy, actor_cmp);
}
thread_range(const thread_range&) = default;
thread_range& operator=(const thread_range&) = default;
const node_id& node() const {
return node_;
}
private:
node_id node_;
};
const entity* get(const thread_range& xs, const thread_id& y) {
// only compares thread ID
auto thread_cmp = [](const entity& lhs, thread_id rhs) {
return lhs.tid < rhs;
};
// range [xs.first, xs.second) is sortd by thread ID
using namespace std;
auto i = lower_bound(xs.begin(), xs.end(), y, thread_cmp);
if (i->tid == y)
return &(*i);
return nullptr;
}
const entity* get(const node_range& xs, const thread_id& y) {
thread_range subrange{xs};
return get(subrange, y);
}
/// Returns the entity for `y` from the node range `xs`.
const entity* get(const node_range& xs, actor_id y) {
if (y == 0)
return nullptr;
// only compares actor ID
auto actor_cmp = [](const entity& lhs, actor_id rhs) {
return lhs.aid < rhs;
};
// range [xs.first, xs.second) is sortd by actor ID
using namespace std;
auto i = lower_bound(xs.begin(), xs.end(), y, actor_cmp);
if (i->aid == y)
return &(*i);
return nullptr;
}
const entity* get(const node_range& xs, const logger_id& y) {
return y.aid > 0 ? get(xs, y.aid) : get(xs, y.tid);
}
/// A single entry in a logfile.
struct log_entry {
/// A UNIX timestamp.
int64_t timestamp;
/// Identifies the logging component, e.g., "caf".
string component;
/// Severity level of this entry.
log_level level;
/// ID of the logging entitiy.
logger_id id;
/// Context information about currently active class.
string class_name;
/// Context information about currently executed function.
string function_name;
/// Context information about currently executed source file.
string file_name;
/// Context information about currently executed source line.
int32_t line_number;
/// Description of the log entry.
string message;
};
/// Stores a log event along with context information.
struct enhanced_log_entry {
/// The original log entry without context information.
const log_entry& data;
/// The actual ID of the logging entity.
const entity& id;
/// Current vector time as seen by `id`.
vector_timestamp& vstamp;
/// JSON representation of `vstamp`.
string json_vstamp;
};
/// CAF events according to SE-0001.
enum class se_type {
spawn,
init,
send,
reject,
receive,
drop,
skip,
finalize,
terminate,
none
};
string to_string(se_type x) {
const char* tbl[] = {"spawn", "init", "send", "reject", "receive",
"drop", "skip", "finalize", "terminate", "none"};
return tbl[static_cast<int>(x)];
}
/// An SE-0001 event, see http://actor-framework.github.io/rfcs/
struct se_event {
const entity* source;
vector_timestamp vstamp;
se_type type;
std::map<string, string> fields;
};
string to_string(const se_event& x) {
string res;
res += "node{";
res += to_string(*x.source);
res += ", ";
res += deep_to_string(x.vstamp);
res += ", ";
res += to_string(x.type);
res += ", ";
res += deep_to_string(x.fields);
res += "}";
return res;
}
CAF_ALLOW_UNSAFE_MESSAGE_TYPE(se_event)
bool field_key_compare(const std::pair<const std::string, std::string>& x,
const std::string& y) {
return x.first == y;
}
#define ATM_CASE(name, value) \
case static_cast<uint64_t>(atom(name)): \
y.type = se_type::value
#define CHECK_FIELDS(...) \
{ \
std::set<std::string> keys{__VA_ARGS__}; \
if (y.fields.size() != keys.size()) \
return sec::invalid_argument; \
if (!std::equal(y.fields.begin(), y.fields.end(), keys.begin(), \
field_key_compare)) \
return sec::invalid_argument; \
} \
static_cast<void>(0)
#define CHECK_NO_FIELDS() \
if (y.fields.size() > 0) \
return sec::invalid_argument;
expected<se_event> parse_event(const enhanced_log_entry& x) {
se_event y{&x.id, x.vstamp, se_type::none, {}};
std::istringstream in{x.data.message};
string type;
if (!(in >> type))
return sec::invalid_argument;
string field_name;
string field_content;
in >> consume(";");
while (in >> field_name >> consume("=") >> rd_line(field_content, ';'))
y.fields.emplace(std::move(field_name), std::move(field_content));
switch (static_cast<uint64_t>(atom_from_string(type))) {
default:
return sec::invalid_argument;
ATM_CASE("SPAWN", spawn);
CHECK_FIELDS("ID", "ARGS");
break;
ATM_CASE("INIT", init);
CHECK_FIELDS("NAME", "LAZY", "HIDDEN");
break;
ATM_CASE("SEND", send);
CHECK_FIELDS("TO", "FROM", "STAGES", "CONTENT");
break;
ATM_CASE("REJECT", reject);
CHECK_NO_FIELDS();
break;
ATM_CASE("RECEIVE", receive);
CHECK_FIELDS("FROM", "STAGES", "CONTENT");
// insert TO field to allow comparing SEND and RECEIVE events easily
y.fields.emplace("TO", to_string(to_mailbox_id(x.id)));
break;
ATM_CASE("DROP", drop);
CHECK_NO_FIELDS();
break;
ATM_CASE("SKIP", skip);
CHECK_NO_FIELDS();
break;
ATM_CASE("FINALIZE", finalize);
CHECK_NO_FIELDS();
break;
ATM_CASE("TERMINATE", terminate);
CHECK_FIELDS("REASON");
break;
}
return {std::move(y)};
}
bool matches(const se_event& x, const se_event& y) {
switch (x.type) {
default:
return false;
case se_type::receive:
if (y.type == se_type::send) {
// TODO
//return x.receiver == y.receiver && x.sender == y.sender
// && x.stages == y.stages && x.content == y.content;
}
return false;
case se_type::init:
if (y.type == se_type::spawn) {
// TODO: return x.id == y.id
}
return false;
}
}
std::ostream& operator<<(std::ostream& out, const enhanced_log_entry& x) {
return out << x.json_vstamp << ' ' << x.data.timestamp << ' '
<< x.data.component << ' ' << x.data.level << ' '
<< x.id.pretty_name << ' ' << x.data.class_name << ' '
<< x.data.function_name << ' '
<< x.data.file_name << ':' << x.data.line_number << ' '
<< x.data.message;
}
std::istream& operator>>(std::istream& in, log_entry& x) {
in >> x.timestamp >> x.component >> x.level
>> consume("actor") >> x.id.aid >> x.id.tid
>> x.class_name >> x.function_name
>> skip_whitespaces >> rd_line(x.file_name, ':')
>> x.line_number >> skip_whitespaces >> rd_line(x.message);
if (x.level == log_level::invalid)
in.setstate(std::ios::failbit);
return in;
}
struct logger_id_meta_data {
bool hidden;
string pretty_name;
};
/// Stores all log entities and their node ID.
struct first_pass_result {
/// Node ID used in the parsed file.
node_id this_node;
/// Entities of the parsed file. The value is `true` if an entity is
/// hidden, otherwise `false`.
std::map<logger_id, logger_id_meta_data> entities;
};
enum verbosity_level {
silent,
informative,
noisy
};
expected<first_pass_result> first_pass(blocking_actor* self, std::istream& in,
verbosity_level vl) {
first_pass_result res;
// read first line to extract the node ID of local actors
// _ caf INFO actor0 _ caf.logger start _:_ level = _, node = NODE
if (!(in >> skip_word >> consume("caf") >> consume("INFO")
>> consume("actor0") >> skip_word >> consume("caf.logger")
>> consume("start") >> skip_word
>> consume("level =") >> skip_word >> consume("node = ")
>> res.this_node >> skip_to_next_line)) {
cerr << "*** malformed log file, expect the first line to contain "
<< "an INFO entry of the logger" << endl;
return sec::invalid_argument;
}
if (vl >= verbosity_level::informative)
aout(self) << "found node " << res.this_node << endl;
logger_id id;
string message;
while (in >> skip_word >> skip_word >> skip_word >> id
>> skip_word >> skip_word >> skip_word >> rd_line(message)) {
// store in map
auto i = res.entities.emplace(id, logger_id_meta_data{false, "actor"}).first;
if (starts_with(message, "INIT ; NAME = ")) {
std::istringstream iss{message};
iss >> consume("INIT ; NAME = ") >> rd_line(i->second.pretty_name, ';');
if (ends_with(message, "HIDDEN = true"))
i->second.hidden = true;
}
}
if (vl >= verbosity_level::informative)
aout(self) << "found " << res.entities.size() << " entities for node "
<< res.this_node << endl;
return res;
}
const string& get(const std::map<string, string>& xs, const string& x) {
auto i = xs.find(x);
if (i != xs.end())
return i->second;
throw std::runtime_error("key not found");
}
void second_pass(blocking_actor* self, const group& grp,
const entity_set& entities, const node_id& nid,
const std::vector<string>& json_names, std::istream& in,
std::ostream& out, std::mutex& out_mtx,
bool drop_hidden_actors, verbosity_level vl) {
assert(entities.size() == json_names.size());
node_range local_entities{entities, nid};
if (local_entities.begin() == local_entities.end())
return;
// state for each local entity
struct state_t {
const entity& eid;
vector_timestamp clock;
};
std::map<logger_id, state_t> local_entities_state;
for (auto& x : local_entities) {
vector_timestamp vzero;
vzero.resize(entities.size());
local_entities_state.emplace(logger_id{x.aid, x.tid},
state_t{x, std::move(vzero)});
}
// lambda for accessing state via logger ID
auto state = [&](const logger_id& x) -> state_t& {
auto i = local_entities_state.find(x);
if (i != local_entities_state.end())
return i->second;
throw std::runtime_error("logger ID not found");
};
// additional state for second pass
size_t line = 0;
log_entry plain_entry;
std::vector<se_event> in_flight_messages;
std::vector<se_event> in_flight_spawns;
// maps scoped actor IDs to their parent ID
std::map<logger_id, logger_id> scoped_actors;
// lambda for broadcasting events that could cross node boundary
auto bcast = [&](const se_event& x) {
if (vl >= verbosity_level::noisy)
aout(self) << "broadcast event from " << nid
<< ": " << deep_to_string(x) << endl;
if (self)
self->send(grp, x);
};
// fetch message from another node via the group
auto fetch_message = [&](const std::map<string, string>& fields)
-> se_event& {
// TODO: this receive unconditionally waits on a message,
// i.e., is a potential deadlock
if (vl >= verbosity_level::noisy)
aout(self) << "wait for send from another node matching fields "
<< deep_to_string(fields) << endl;
se_event* res = nullptr;
self->receive_while([&] { return res == nullptr; })(
[&](const se_event& x) {
switch (x.type) {
default:
break;
case se_type::send:
in_flight_messages.emplace_back(x);
if (x.fields == fields)
res = &in_flight_messages.back();
break;
}
}
);
return *res;
};
// second pass
while (in >> plain_entry) {
++line;
// increment local time
auto& st = state(plain_entry.id);
// do not produce log output for silent actors but still track messages
// through those actors, because they might be forwarding messages
bool silent = drop_hidden_actors && st.eid.hidden;
if (!silent)
st.clock[st.eid.vid] += 1;
// generate enhanced entry (with incomplete JSON timestamp for now)
enhanced_log_entry entry{plain_entry, st.eid, st.clock, string{}};
// check whether entry contains an SE-0001 event
auto tmp = parse_event(entry);
if (tmp) {
auto& event = *tmp;
switch (event.type) {
default:
break;
case se_type::send:
bcast(event);
in_flight_messages.emplace_back(std::move(event));
break;
case se_type::receive: {
auto pred = [&](const se_event& x) {
assert(x.type == se_type::send);
return event.fields == x.fields;
};
auto e = in_flight_messages.end();
auto i = std::find_if(in_flight_messages.begin(), e, pred);
if (i != e) {
merge(st.clock, i->vstamp);
} else {
merge(st.clock, fetch_message(event.fields).vstamp);
}
break;
}
case se_type::spawn:
in_flight_spawns.emplace_back(std::move(event));
break;
case se_type::init: {
auto id_field = std::to_string(st.eid.aid);
auto pred = [&](const se_event& x) {
assert(x.type == se_type::spawn);
return get(x.fields, "ID") == id_field;
};
auto e = in_flight_spawns.end();
auto i = std::find_if(in_flight_spawns.begin(), e, pred);
if (i != e) {
merge(st.clock, i->vstamp);
// keep book on scoped actors since their terminate
// event propagates back to the parent
if (get(event.fields, "NAME") == "scoped_actor")
scoped_actors.emplace(plain_entry.id, to_logger_id(*i->source));
in_flight_spawns.erase(i);
} else {
std::cerr << "*** cannot match init event to a previous spawn"
<< endl;
//merge(st.clock, fetch_message(se_type::spawn, pred).vstamp);
}
break;
}
case se_type::terminate:
auto i = scoped_actors.find(plain_entry.id);
if (i != scoped_actors.end()) {
// merge timestamp with parent to capture happens-before relation
auto& parent_state = state(i->second);
merge(parent_state.clock, st.clock);
scoped_actors.erase(i);
}
break;
}
}
// create ShiViz compatible JSON-formatted vector timestamp
std::ostringstream oss;
oss << '{';
bool need_comma = false;
for (size_t i = 0; i < st.clock.size(); ++i) {
auto x = st.clock[i];
if (x > 0) {
if (need_comma)
oss << ',';
else
need_comma = true;
oss << '"' << json_names[i] << '"' << ':' << x;
}
}
oss << '}';
entry.json_vstamp = oss.str();
// print entry to output file
if (!silent) {
std::lock_guard<std::mutex> guard{out_mtx};
out << entry << '\n';
}
}
}
struct config : public actor_system_config {
string output_file;
bool include_hidden_actors = false;
size_t verbosity = 0;
config() {
opt_group{custom_options_, "global"}
.add(output_file, "output-file,o", "Path for the output file")
.add(include_hidden_actors, "include-hidden-actors,i",
"Include hidden (system-level) actors")
.add(verbosity, "verbosity,v", "Debug output (from 0 to 2)");
}
};
// two pass parser for CAF log files that enhances logs with vector
// clock timestamps
void caf_main(actor_system& sys, const config& cfg) {
using namespace std;
if (cfg.output_file.empty()) {
cerr << "*** no output file specified" << endl;
return;
}
verbosity_level vl;
switch (cfg.verbosity) {
case 0:
vl = verbosity_level::silent;
break;
case 1:
vl = verbosity_level::informative;
break;
default:
vl = verbosity_level::noisy;
}
// open output file
std::ofstream out{cfg.output_file};
if (!out) {
cerr << "unable to open output file: " << cfg.output_file << endl;
return;
}
using file_path = string;
static constexpr size_t irsize = sizeof(file_path) + sizeof(std::ifstream)
+ sizeof(first_pass_result);
struct intermediate_res {
file_path fname;
std::ifstream fstream;
first_pass_result res;
char pad[irsize >= CAF_CACHE_LINE_SIZE ? 1 : CAF_CACHE_LINE_SIZE - irsize];
intermediate_res() = default;
intermediate_res(intermediate_res&&) = default;
intermediate_res& operator=(intermediate_res&&) = default;
intermediate_res(file_path fp, std::ifstream&& fs, first_pass_result&& fr)
: fname(std::move(fp)),
fstream(std::move(fs)),
res(std::move(fr)) {
// nop
}
};
// do a first pass on all files to extract node IDs and entities
vector<intermediate_res> intermediate_results;
intermediate_results.resize(cfg.args_remainder.size());
for (size_t i = 0; i < cfg.args_remainder.size(); ++i) {
auto& file = cfg.args_remainder.get_as<string>(i);
auto ptr = &intermediate_results[i];
ptr->fname = file;
ptr->fstream.open(file);
if (!ptr->fstream) {
cerr << "could not open file: " << file << endl;
continue;
}
sys.spawn([ptr, vl](blocking_actor* self) {
auto& f = ptr->fstream;
auto res = first_pass(self, f, vl);
if (res) {
// rewind stream and push intermediate results
f.clear();
f.seekg(0);
ptr->res = std::move(*res);
}
});
}
sys.await_all_actors_done();
// post-process collected entity IDs before second pass
entity_set entities;
std::vector<string> entity_names;
auto sort_pred = [](const intermediate_res& x, const intermediate_res& y) {
return x.res.this_node < y.res.this_node;
};
std::map<string, size_t> pretty_actor_names;
size_t thread_count = 0;
// make sure we insert in sorted order into the entities set
std::sort(intermediate_results.begin(), intermediate_results.end(),
sort_pred);
for (auto& ir : intermediate_results) {
auto node_as_string = to_string(ir.res.this_node);
for (auto& kvp : ir.res.entities) {
string pretty_name;
// make each (pretty) actor and thread name unique
auto& pn = kvp.second.pretty_name;
if (kvp.first.aid != 0)
pretty_name = pn + std::to_string(++pretty_actor_names[pn]);
//"actor" + std::to_string(kvp.first.aid);
else
pretty_name = "thread" + std::to_string(++thread_count);
/*
pretty_name += '@';
pretty_name += node_as_string;
*/
auto vid = entities.size(); // position in the vector timestamp
entity_names.emplace_back(pretty_name);
entities.emplace(entity{kvp.first.aid, kvp.first.tid, ir.res.this_node,
vid, kvp.second.hidden, std::move(pretty_name)});
}
}
// check whether entities set is in the right order
auto vid_cmp = [](const entity& x, const entity& y) {
return x.vid < y.vid;
};
if (!std::is_sorted(entities.begin(), entities.end(), vid_cmp)) {
cerr << "*** ERROR: entity set not sorted by vector timestamp ID:\n"
<< deep_to_string(entities) << endl;
return;
}
// do a second pass for all log files
// first line is the regex to parse the remainder of the file
out << "(?<clock>\\S+) (?<timestamp>\\d+) (?<component>\\S+) "
<< "(?<level>\\S+) (?<host>\\S+) (?<class>\\S+) (?<function>\\S+) "
<< "(?<file>\\S+):(?<line>\\d+) (?<event>.+)"
<< endl;
// second line is the separator for multiple runs
out << endl;
std::mutex out_mtx;
auto grp = sys.groups().anonymous();
for (auto& fpr : intermediate_results) {
sys.spawn_in_group(grp, [&](blocking_actor* self) {
second_pass(self, grp, entities, fpr.res.this_node, entity_names,
fpr.fstream, out, out_mtx, !cfg.include_hidden_actors, vl);
});
}
sys.await_all_actors_done();
}
CAF_MAIN()
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