Commit b86b6c30 authored by Dominik Charousset's avatar Dominik Charousset

Fix crash when using aout in state dtors

parent 5572f4fd
...@@ -156,15 +156,16 @@ public: ...@@ -156,15 +156,16 @@ public:
}; };
// flags storing runtime information used by ... // flags storing runtime information used by ...
static constexpr int trap_exit_flag = 0x01; // local_actor static constexpr int trap_exit_flag = 0x0001; // local_actor
static constexpr int has_timeout_flag = 0x02; // single_timeout static constexpr int has_timeout_flag = 0x0002; // single_timeout
static constexpr int is_registered_flag = 0x04; // (several actors) static constexpr int is_registered_flag = 0x0004; // (several actors)
static constexpr int is_initialized_flag = 0x08; // event-based actors static constexpr int is_initialized_flag = 0x0008; // event-based actors
static constexpr int is_blocking_flag = 0x10; // blocking_actor static constexpr int is_blocking_flag = 0x0010; // blocking_actor
static constexpr int is_detached_flag = 0x20; // local_actor static constexpr int is_detached_flag = 0x0020; // local_actor
static constexpr int is_priority_aware_flag = 0x40; // local_actor static constexpr int is_priority_aware_flag = 0x0040; // local_actor
static constexpr int is_serializable_flag = 0x40; // local_actor static constexpr int is_serializable_flag = 0x0040; // local_actor
static constexpr int is_migrated_from_flag = 0x80; // local_actor static constexpr int is_migrated_from_flag = 0x0080; // local_actor
static constexpr int has_used_aout_flag = 0x0100; // local_actor
inline void set_flag(bool enable_flag, int mask) { inline void set_flag(bool enable_flag, int mask) {
auto x = flags(); auto x = flags();
......
...@@ -41,15 +41,15 @@ public: ...@@ -41,15 +41,15 @@ public:
virtual void enqueue(strong_actor_ptr sender, message_id mid, message content, virtual void enqueue(strong_actor_ptr sender, message_id mid, message content,
execution_unit* host = nullptr) = 0; execution_unit* host = nullptr) = 0;
static constexpr int is_abstract_actor_flag = 0x100000; static constexpr int is_abstract_actor_flag = 0x01000000;
static constexpr int is_abstract_group_flag = 0x200000; static constexpr int is_abstract_group_flag = 0x02000000;
static constexpr int is_actor_bind_decorator_flag = 0x400000; static constexpr int is_actor_bind_decorator_flag = 0x04000000;
static constexpr int is_actor_dot_decorator_flag = 0x800000; static constexpr int is_actor_dot_decorator_flag = 0x08000000;
static constexpr int is_actor_decorator_mask = 0xC00000; static constexpr int is_actor_decorator_mask = 0x0C000000;
inline bool is_abstract_actor() const { inline bool is_abstract_actor() const {
return static_cast<bool>(flags() & is_abstract_actor_flag); return static_cast<bool>(flags() & is_abstract_actor_flag);
......
...@@ -83,7 +83,9 @@ public: ...@@ -83,7 +83,9 @@ public:
} }
private: private:
actor self_; void init(abstract_actor*);
actor_id self_;
actor printer_; actor printer_;
}; };
......
...@@ -25,6 +25,7 @@ ...@@ -25,6 +25,7 @@
#include <chrono> #include <chrono>
#include <fstream> #include <fstream>
#include <iostream> #include <iostream>
#include <unordered_map>
#include <condition_variable> #include <condition_variable>
#include "caf/send.hpp" #include "caf/send.hpp"
...@@ -239,24 +240,21 @@ void printer_loop(blocking_actor* self) { ...@@ -239,24 +240,21 @@ void printer_loop(blocking_actor* self) {
// nop // nop
} }
}; };
using data_map = std::map<actor_addr, actor_data>; using data_map = std::unordered_map<actor_id, actor_data>;
sink_cache fcache; sink_cache fcache;
sink_handle global_redirect; sink_handle global_redirect;
data_map data; data_map data;
auto get_data = [&](const actor_addr& addr, bool insert_missing) auto get_data = [&](actor_id addr, bool insert_missing) -> actor_data* {
-> optional<actor_data&> { if (addr == invalid_actor_id)
if (addr == invalid_actor_addr) return nullptr;
return none;
auto i = data.find(addr); auto i = data.find(addr);
if (i == data.end() && insert_missing) { if (i == data.end() && insert_missing)
i = data.emplace(addr, actor_data{}).first; i = data.emplace(addr, actor_data{}).first;
self->monitor(addr);
}
if (i != data.end()) if (i != data.end())
return i->second; return &(i->second);
return none; return nullptr;
}; };
auto flush = [&](optional<actor_data&> what, bool forced) { auto flush = [&](actor_data* what, bool forced) {
if (! what) if (! what)
return; return;
auto& line = what->current_line; auto& line = what->current_line;
...@@ -273,33 +271,35 @@ void printer_loop(blocking_actor* self) { ...@@ -273,33 +271,35 @@ void printer_loop(blocking_actor* self) {
self->trap_exit(true); self->trap_exit(true);
bool running = true; bool running = true;
self->receive_while([&] { return running; })( self->receive_while([&] { return running; })(
[&](add_atom, std::string& str) { [&](add_atom, actor_id aid, std::string& str) {
auto s = self->current_sender(); if (str.empty() || aid == invalid_actor_id)
if (str.empty() || s == invalid_actor_addr)
return; return;
auto d = get_data(s, true); auto d = get_data(aid, true);
if (d) { if (d) {
d->current_line += str; d->current_line += str;
flush(d, false); flush(d, false);
} }
}, },
[&](flush_atom) { [&](flush_atom, actor_id aid) {
flush(get_data(self->current_sender(), false), true); flush(get_data(aid, false), true);
}, },
[&](const down_msg& dm) { [&](delete_atom, actor_id aid) {
flush(get_data(dm.source, false), true); auto data_ptr = get_data(aid, false);
data.erase(dm.source); if (data_ptr) {
}, flush(data_ptr, true);
[&](const exit_msg&) { data.erase(aid);
running = false; }
}, },
[&](redirect_atom, const std::string& fn, int flag) { [&](redirect_atom, const std::string& fn, int flag) {
global_redirect = get_sink_handle(self->system(), fcache, fn, flag); global_redirect = get_sink_handle(self->system(), fcache, fn, flag);
}, },
[&](redirect_atom, const actor_addr& src, const std::string& fn, int flag) { [&](redirect_atom, actor_id aid, const std::string& fn, int flag) {
auto d = get_data(src, true); auto d = get_data(aid, true);
if (d) if (d)
d->redirect = get_sink_handle(self->system(), fcache, fn, flag); d->redirect = get_sink_handle(self->system(), fcache, fn, flag);
},
[&](const exit_msg&) {
running = false;
} }
); );
} }
......
...@@ -22,43 +22,60 @@ ...@@ -22,43 +22,60 @@
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/scoped_actor.hpp" #include "caf/scoped_actor.hpp"
#include "caf/abstract_actor.hpp" #include "caf/abstract_actor.hpp"
#include "caf/default_attachable.hpp"
#include "caf/scheduler/abstract_coordinator.hpp" #include "caf/scheduler/abstract_coordinator.hpp"
namespace caf { namespace caf {
actor_ostream::actor_ostream(local_actor* self) actor_ostream::actor_ostream(local_actor* self)
: self_(actor_cast<actor>(self)), : self_(self->id()),
printer_(self->home_system().scheduler().printer()) { printer_(self->home_system().scheduler().printer()) {
// nop init(self);
} }
actor_ostream::actor_ostream(scoped_actor& self) actor_ostream::actor_ostream(scoped_actor& self)
: self_(actor_cast<actor>(self)), : self_(self->id()),
printer_(self->home_system().scheduler().printer()) { printer_(self->home_system().scheduler().printer()) {
// nop init(actor_cast<abstract_actor*>(self));
} }
actor_ostream& actor_ostream::write(std::string arg) { actor_ostream& actor_ostream::write(std::string arg) {
send_as(self_, printer_, add_atom::value, std::move(arg)); printer_->enqueue(mailbox_element::make_joint(nullptr, message_id::make(), {},
add_atom::value, self_,
std::move(arg)),
nullptr);
return *this; return *this;
} }
actor_ostream& actor_ostream::flush() { actor_ostream& actor_ostream::flush() {
send_as(self_, printer_, flush_atom::value); printer_->enqueue(mailbox_element::make_joint(nullptr, message_id::make(), {},
flush_atom::value, self_),
nullptr);
return *this; return *this;
} }
void actor_ostream::redirect(abstract_actor* self, std::string fn, int flags) { void actor_ostream::redirect(abstract_actor* self, std::string fn, int flags) {
if (! self) if (! self)
return; return;
send_as(actor_cast<actor>(self), self->home_system().scheduler().printer(), auto pr = self->home_system().scheduler().printer();
redirect_atom::value, self->address(), std::move(fn), flags); pr->enqueue(mailbox_element::make_joint(nullptr, message_id::make(), {},
redirect_atom::value, self->id(),
std::move(fn), flags),
nullptr);
} }
void actor_ostream::redirect_all(actor_system& sys, std::string fn, int flags) { void actor_ostream::redirect_all(actor_system& sys, std::string fn, int flags) {
anon_send(sys.scheduler().printer(), auto pr = sys.scheduler().printer();
redirect_atom::value, std::move(fn), flags); pr->enqueue(mailbox_element::make_joint(nullptr, message_id::make(), {},
redirect_atom::value,
std::move(fn), flags),
nullptr);
}
void actor_ostream::init(abstract_actor* self) {
if (! self->get_flag(abstract_actor::has_used_aout_flag))
self->set_flag(true, abstract_actor::has_used_aout_flag);
} }
actor_ostream aout(local_actor* self) { actor_ostream aout(local_actor* self) {
......
...@@ -27,6 +27,8 @@ ...@@ -27,6 +27,8 @@
#include "caf/system_messages.hpp" #include "caf/system_messages.hpp"
#include "caf/default_attachable.hpp" #include "caf/default_attachable.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
namespace caf { namespace caf {
const char* monitorable_actor::name() const { const char* monitorable_actor::name() const {
...@@ -73,6 +75,13 @@ void monitorable_actor::cleanup(exit_reason reason, execution_unit* host) { ...@@ -73,6 +75,13 @@ void monitorable_actor::cleanup(exit_reason reason, execution_unit* host) {
// send exit messages // send exit messages
for (attachable* i = head.get(); i != nullptr; i = i->next.get()) for (attachable* i = head.get(); i != nullptr; i = i->next.get())
i->actor_exited(reason, host); i->actor_exited(reason, host);
// tell printer to purge its state for us if we ever used aout()
if (get_flag(abstract_actor::has_used_aout_flag)) {
auto pr = home_system().scheduler().printer();
pr->enqueue(mailbox_element::make_joint(nullptr, message_id::make(), {},
delete_atom::value, id()),
nullptr);
}
} }
monitorable_actor::monitorable_actor(actor_config& cfg) monitorable_actor::monitorable_actor(actor_config& cfg)
......
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