Commit f2fb9e71 authored by Dominik Charousset's avatar Dominik Charousset

Allow redirecting aout streams

parent 5e413f56
...@@ -29,33 +29,52 @@ namespace caf { ...@@ -29,33 +29,52 @@ namespace caf {
class local_actor; class local_actor;
class scoped_actor; class scoped_actor;
/// Provides support for thread-safe output operations on character streams. The
/// stream operates on a per-actor basis and will print only complete lines or
/// when explicitly forced to flush its buffer. The stream will convert *any*
/// operation to a message received by a printer actor. This actor is a sequence
/// point that ensures output appears never interleaved.
class actor_ostream { class actor_ostream {
public: public:
using fun_type = actor_ostream& (*)(actor_ostream&); using fun_type = actor_ostream& (*)(actor_ostream&);
actor_ostream(actor_ostream&&) = default; actor_ostream(actor_ostream&&) = default;
actor_ostream(const actor_ostream&) = default; actor_ostream(const actor_ostream&) = default;
actor_ostream& operator=(actor_ostream&&) = default; actor_ostream& operator=(actor_ostream&&) = default;
actor_ostream& operator=(const actor_ostream&) = default; actor_ostream& operator=(const actor_ostream&) = default;
/// Open redirection file in append mode.
static constexpr int append = 0x01;
/// Creates a stream for `self`.
explicit actor_ostream(actor self); explicit actor_ostream(actor self);
/// Writes `arg` to the buffer allocated for the calling actor.
actor_ostream& write(std::string arg); actor_ostream& write(std::string arg);
/// Flushes the buffer allocated for the calling actor.
actor_ostream& flush(); actor_ostream& flush();
/// Redirects all further output from `src` to `file_name`.
static void redirect(const actor& src, std::string file_name, int flags = 0);
/// Redirects all further output from any actor that did not
/// redirect its output to `file_name`.
static void redirect_all(std::string file_name, int flags = 0);
/// Writes `arg` to the buffer allocated for the calling actor.
inline actor_ostream& operator<<(std::string arg) { inline actor_ostream& operator<<(std::string arg) {
return write(std::move(arg)); return write(std::move(arg));
} }
// disambiguate between conversion to string and to message /// Writes `arg` to the buffer allocated for the calling actor.
inline actor_ostream& operator<<(const char* arg) { inline actor_ostream& operator<<(const char* arg) {
return *this << std::string{arg}; return *this << std::string{arg};
} }
/// Writes `to_string(arg)` to the buffer allocated for the calling actor,
/// calling either `std::to_string` or `caf::to_string` depending on
/// the argument.
template <class T> template <class T>
inline typename std::enable_if< inline typename std::enable_if<
! std::is_convertible<T, std::string>::value, actor_ostream& ! std::is_convertible<T, std::string>::value, actor_ostream&
...@@ -64,17 +83,17 @@ public: ...@@ -64,17 +83,17 @@ public:
return write(to_string(std::forward<T>(arg))); return write(to_string(std::forward<T>(arg)));
} }
/// Apply `f` to `*this`.
inline actor_ostream& operator<<(actor_ostream::fun_type f) { inline actor_ostream& operator<<(actor_ostream::fun_type f) {
return f(*this); return f(*this);
} }
private: private:
actor self_; actor self_;
actor printer_; actor printer_;
}; };
/// Convenience factory function for creating an actor output stream.
inline actor_ostream aout(actor self) { inline actor_ostream aout(actor self) {
return actor_ostream{self}; return actor_ostream{self};
} }
......
...@@ -95,6 +95,9 @@ using forward_atom = atom_constant<atom("FORWARD")>; ...@@ -95,6 +95,9 @@ using forward_atom = atom_constant<atom("FORWARD")>;
/// Generic 'FLUSH' atom, e.g., used by `aout`. /// Generic 'FLUSH' atom, e.g., used by `aout`.
using flush_atom = atom_constant<atom("FLUSH")>; using flush_atom = atom_constant<atom("FLUSH")>;
/// Generic 'REDIRECT' atom, e.g., used by `aout`.
using redirect_atom = atom_constant<atom("REDIRECT")>;
/// Generic 'LINK' atom for link requests over network. /// Generic 'LINK' atom for link requests over network.
using link_atom = atom_constant<atom("LINK")>; using link_atom = atom_constant<atom("LINK")>;
......
...@@ -19,9 +19,11 @@ ...@@ -19,9 +19,11 @@
#include "caf/scheduler/abstract_coordinator.hpp" #include "caf/scheduler/abstract_coordinator.hpp"
#include <ios>
#include <thread> #include <thread>
#include <atomic> #include <atomic>
#include <chrono> #include <chrono>
#include <fstream>
#include <iostream> #include <iostream>
#include <condition_variable> #include <condition_variable>
...@@ -32,6 +34,7 @@ ...@@ -32,6 +34,7 @@
#include "caf/to_string.hpp" #include "caf/to_string.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/scoped_actor.hpp" #include "caf/scoped_actor.hpp"
#include "caf/actor_ostream.hpp"
#include "caf/system_messages.hpp" #include "caf/system_messages.hpp"
#include "caf/scheduler/coordinator.hpp" #include "caf/scheduler/coordinator.hpp"
...@@ -136,55 +139,171 @@ public: ...@@ -136,55 +139,171 @@ public:
} }
}; };
void printer_loop(blocking_actor* self) { using string_sink = std::function<void (std::string&&)>;
self->trap_exit(true);
std::map<actor_addr, std::string> out; // the first value is the use count, the last ostream_handle that
auto flush_output = [&out](const actor_addr& s) { // decrements it to 0 removes the ostream pointer from the map
auto i = out.find(s); using counted_sink = std::pair<size_t, string_sink>;
if (i != out.end()) {
auto& line = i->second; using sink_cache = std::map<std::string, counted_sink>;
if (! line.empty()) {
std::cout << line << std::flush; class sink_handle {
line.clear(); public:
using iterator = sink_cache::iterator;
sink_handle() : cache_(nullptr) {
// nop
}
sink_handle(sink_cache* fc, iterator iter) : cache_(fc), iter_(iter) {
if (cache_)
++iter_->second.first;
}
sink_handle(const sink_handle& other) : cache_(nullptr) {
*this = other;
}
sink_handle& operator=(const sink_handle& other) {
if (cache_ != other.cache_ || iter_ != other.iter_) {
clear();
cache_ = other.cache_;
if (cache_) {
iter_ = other.iter_;
++iter_->second.first;
} }
} }
return *this;
}
~sink_handle() {
clear();
}
explicit operator bool() {
return cache_ != nullptr;
}
string_sink& operator*() {
CAF_ASSERT(iter_->second.second != nullptr);
return iter_->second.second;
}
private:
void clear() {
if (cache_ && --iter_->second.first == 0) {
cache_->erase(iter_);
cache_ = nullptr;
}
}
sink_cache* cache_;
sink_cache::iterator iter_;
};
string_sink make_sink(const std::string& fn, int flags) {
if (fn.empty())
return nullptr;
if (fn.front() == ':') {
// "virtual file" name given, translate this to group communication
auto grp = group::get("local", fn);
return [grp, fn](std::string&& out) { anon_send(grp, fn, std::move(out)); };
}
auto append = static_cast<bool>(flags & actor_ostream::append);
auto fs = std::make_shared<std::ofstream>();
fs->open(fn, append ? std::ios_base::out | std::ios_base::app
: std::ios_base::out);
if (fs->is_open())
return [fs](std::string&& out) { *fs << out; };
std::cerr << "cannot open file: " << fn << std::endl;
return nullptr;
}
sink_handle get_sink_handle(sink_cache& fc, const std::string& fn, int flags) {
auto i = fc.find(fn);
if (i != fc.end())
return {&fc, i};
auto fs = make_sink(fn, flags);
if (fs) {
i = fc.emplace(fn, sink_cache::mapped_type{0, std::move(fs)}).first;
return {&fc, i};
}
return {};
}
void printer_loop(blocking_actor* self) {
struct actor_data {
std::string current_line;
sink_handle redirect;
actor_data(std::string&& line) : current_line(std::move(line)) {
// nop
}
}; };
auto flush_if_needed = [](std::string& str) { using data_map = std::map<actor_addr, actor_data>;
if (str.back() == '\n') { sink_cache fcache;
std::cout << str << std::flush; sink_handle global_redirect;
str.clear(); data_map data;
auto get_data = [&](const actor_addr& addr, bool insert_missing)
-> optional<actor_data&> {
if (addr == invalid_actor_addr)
return none;
auto i = data.find(addr);
if (i == data.end() && insert_missing) {
i = data.emplace(addr, std::string{}).first;
self->monitor(addr);
} }
if (i != data.end())
return i->second;
return none;
}; };
auto flush = [&](optional<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();
};
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, std::string& str) {
auto s = self->current_sender(); auto s = self->current_sender();
if (str.empty() || s == invalid_actor_addr) { if (str.empty() || s == invalid_actor_addr)
return; return;
auto d = get_data(s, true);
if (d) {
d->current_line += str;
flush(d, false);
} }
auto i = out.find(s);
if (i == out.end()) {
i = out.emplace(s, move(str)).first;
// monitor actor to flush its output on exit
self->monitor(s);
} else {
i->second += std::move(str);
}
flush_if_needed(i->second);
}, },
[&](flush_atom) { [&](flush_atom) {
flush_output(self->current_sender()); flush(get_data(self->current_sender(), false), true);
}, },
[&](const down_msg& dm) { [&](const down_msg& dm) {
flush_output(dm.source); flush(get_data(dm.source, false), true);
out.erase(dm.source); data.erase(dm.source);
}, },
[&](const exit_msg&) { [&](const exit_msg&) {
running = false; running = false;
}, },
[&](redirect_atom, const std::string& fn, int flag) {
global_redirect = get_sink_handle(fcache, fn, flag);
},
[&](redirect_atom, const actor_addr& src, const std::string& fn, int flag) {
auto d = get_data(src, true);
if (d)
d->redirect = get_sink_handle(fcache, fn, flag);
},
others >> [&] { others >> [&] {
std::cerr << "*** unexpected: " << to_string(self->current_message()) std::cerr << "*** unexpected: "
<< std::endl; << to_string(self->current_message()) << std::endl;
} }
); );
} }
......
...@@ -43,6 +43,16 @@ actor_ostream& actor_ostream::flush() { ...@@ -43,6 +43,16 @@ actor_ostream& actor_ostream::flush() {
return *this; return *this;
} }
void actor_ostream::redirect(const actor& src, std::string f, int flags) {
send_as(src, detail::singletons::get_scheduling_coordinator()->printer(),
redirect_atom::value, src.address(), std::move(f), flags);
}
void actor_ostream::redirect_all(std::string f, int flags) {
anon_send(detail::singletons::get_scheduling_coordinator()->printer(),
redirect_atom::value, std::move(f), flags);
}
} // namespace caf } // namespace caf
namespace std { namespace std {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* 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/config.hpp"
#define CAF_SUITE aout
#include "caf/test/unit_test.hpp"
#include "caf/all.hpp"
using namespace caf;
using std::endl;
namespace {
struct fixture {
~fixture() {
await_all_actors_done();
shutdown();
}
};
constexpr const char* global_redirect = ":test";
constexpr const char* local_redirect = ":test2";
constexpr const char* chatty_line = "hi there! :)";
constexpr const char* chattier_line = "hello there, fellow friend! :)";
void chatty_actor(event_based_actor* self) {
aout(self) << chatty_line << endl;
}
void chattier_actor(event_based_actor* self, const std::string& fn) {
aout(self) << chatty_line << endl;
actor_ostream::redirect(self, fn);
aout(self) << chattier_line << endl;
}
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(aout_tests, fixture)
CAF_TEST(global_redirect) {
scoped_actor self;
self->join(group::get("local", global_redirect));
actor_ostream::redirect_all(global_redirect);
spawn(chatty_actor);
self->receive(
[](const std::string& virtual_file, const std::string& line) {
CAF_CHECK_EQUAL(virtual_file, ":test");
CAF_CHECK_EQUAL(line, chatty_line);
}
);
self->await_all_other_actors_done();
CAF_CHECK_EQUAL(self->mailbox().count(), 0);
}
CAF_TEST(global_and_local_redirect) {
scoped_actor self;
self->join(group::get("local", global_redirect));
self->join(group::get("local", local_redirect));
actor_ostream::redirect_all(global_redirect);
spawn(chatty_actor);
spawn(chattier_actor, local_redirect);
int i = 0;
self->receive_for(i, 2)(
on(global_redirect, arg_match) >> [](std::string& line) {
line.pop_back(); // drop '\n'
CAF_CHECK_EQUAL(line, chatty_line);
}
);
self->receive(
on(local_redirect, arg_match) >> [](std::string& line) {
line.pop_back(); // drop '\n'
CAF_CHECK_EQUAL(line, chattier_line);
}
);
self->await_all_other_actors_done();
CAF_CHECK_EQUAL(self->mailbox().count(), 0);
}
CAF_TEST_FIXTURE_SCOPE_END()
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