Commit 2e89b4a2 authored by Dominik Charousset's avatar Dominik Charousset

provide `aout` "stream" and use it in examples

the `aout` stream acts like `cout`, but instead of printing content
directly to the terminal, `aout` sends all strings to an actor
that has a per-sender map to prevent interleaved output
parent 9e25fec2
......@@ -851,15 +851,59 @@ actor_ptr remote_actor(network::io_stream_ptr_pair connection);
*/
void shutdown(); // note: implemented in singleton_manager.cpp
struct actor_ostream {
typedef const actor_ostream& (*fun_type)(const actor_ostream&);
constexpr actor_ostream() { }
inline const actor_ostream& write(std::string arg) const {
send(get_scheduler()->printer(), atom("add"), move(arg));
return *this;
}
inline const actor_ostream& flush() const {
send(get_scheduler()->printer(), atom("flush"));
return *this;
}
};
namespace { constexpr actor_ostream aout; }
inline const actor_ostream& operator<<(const actor_ostream& o, std::string arg) {
return o.write(move(arg));
}
inline const actor_ostream& operator<<(const actor_ostream& o, const any_tuple& arg) {
return o.write(cppa::to_string(arg));
}
template<typename T>
inline typename std::enable_if< !std::is_convertible<T,std::string>::value
&& !std::is_convertible<T,any_tuple>::value,
const actor_ostream&>::type
operator<<(const actor_ostream& o, T&& arg) {
return o.write(std::to_string(std::forward<T>(arg)));
}
inline const actor_ostream& operator<<(const actor_ostream& o, actor_ostream::fun_type f) {
return f(o);
}
} // namespace cppa
namespace std {
// allow actor_ptr to be used in hash maps
template<>
struct hash<cppa::actor_ptr> {
inline size_t operator()(const cppa::actor_ptr& ptr) const {
return (ptr) ? static_cast<size_t>(ptr->id()) : 0;
}
};
// provide convenience overlaods for aout; implemented in logging.cpp
const cppa::actor_ostream& endl(const cppa::actor_ostream& o);
const cppa::actor_ostream& flush(const cppa::actor_ostream& o);
} // namespace std
#endif // CPPA_HPP
......@@ -106,6 +106,8 @@ class scheduler {
public:
const actor_ptr& printer() const;
virtual void enqueue(scheduled_actor*) = 0;
/**
......
......@@ -3,6 +3,7 @@
* exercise using only libcppa's event-based actor implementation. *
\******************************************************************************/
#include <map>
#include <vector>
#include <chrono>
#include <sstream>
......@@ -10,10 +11,9 @@
#include "cppa/cppa.hpp"
using std::cout;
using std::endl;
using std::chrono::seconds;
using namespace std;
using namespace cppa;
// either taken by a philosopher or available
......@@ -99,13 +99,12 @@ struct philosopher : sb_actor<philosopher> {
// create message in memory to avoid interleaved
// messages on the terminal
std::ostringstream oss;
oss << name
<< " has picked up chopsticks with IDs "
<< left->id()
<< " and "
<< right->id()
<< " and starts to eat\n";
cout << oss.str();
aout << name
<< " has picked up chopsticks with IDs "
<< left->id()
<< " and "
<< right->id()
<< " and starts to eat\n";
// eat some time
delayed_send(this, seconds(5), atom("think"));
become(eating);
......@@ -158,7 +157,7 @@ struct philosopher : sb_actor<philosopher> {
send(left, atom("put"), this);
send(right, atom("put"), this);
delayed_send(this, seconds(5), atom("eat"));
cout << ( name
aout << ( name
+ " puts down his chopsticks and starts to think\n");
become(thinking);
}
......@@ -166,7 +165,7 @@ struct philosopher : sb_actor<philosopher> {
// philosophers start to think after receiving {think}
init_state = (
on(atom("think")) >> [=]() {
cout << (name + " starts to think\n");
aout << (name + " starts to think\n");
delayed_send(this, seconds(5), atom("eat"));
become(thinking);
}
......@@ -177,13 +176,13 @@ struct philosopher : sb_actor<philosopher> {
int main(int, char**) {
// create five chopsticks
cout << "chopstick ids:";
aout << "chopstick ids:";
std::vector<actor_ptr> chopsticks;
for (size_t i = 0; i < 5; ++i) {
chopsticks.push_back(spawn<chopstick>());
cout << " " << chopsticks.back()->id();
aout << " " << chopsticks.back()->id();
}
cout << endl;
aout << endl;
// a group to address all philosophers
auto dinner_club = group::anonymous();
// spawn five philosopher, each joining the Dinner Club
......
......@@ -43,97 +43,6 @@ struct math_actor : event_based_actor {
}
};
namespace {
// this actor manages the output of the program and has a per-actor cache
actor_ptr s_printer = factory::event_based([](map<actor_ptr,string>* out) {
auto flush_output = [out](const actor_ptr& s) {
auto i = out->find(s);
if (i != out->end()) {
auto& ref = i->second;
if (!ref.empty()) {
cout << move(ref) << flush;
ref = string();
}
}
};
self->become (
on(atom("add"), arg_match) >> [=](string& str) {
if (!str.empty()) {
auto s = self->last_sender();
auto i = out->find(s);
if (i == out->end()) {
i = out->insert(make_pair(s, move(str))).first;
// monitor actor to flush its output on exit
self->monitor(s);
}
auto& ref = i->second;
ref += move(str);
if (ref.back() == '\n') {
cout << move(ref) << flush;
ref = string();
}
}
},
on(atom("flush")) >> [=] {
flush_output(self->last_sender());
},
on(atom("DOWN"), any_vals) >> [=] {
auto s = self->last_sender();
flush_output(s);
out->erase(s);
},
on(atom("quit")) >> [] {
self->quit();
},
others() >> [] {
cout << "*** unexpected: "
<< to_string(self->last_dequeued())
<< endl;
}
);
}).spawn();
} // namespace <anonymous>
// an output stream sending messages to s_printer
struct actor_ostream {
typedef actor_ostream& (*fun_type)(actor_ostream&);
virtual ~actor_ostream() { }
virtual actor_ostream& write(string arg) {
send(s_printer, atom("add"), move(arg));
return *this;
}
virtual actor_ostream& flush() {
send(s_printer, atom("flush"));
return *this;
}
};
inline actor_ostream& operator<<(actor_ostream& o, string arg) {
return o.write(move(arg));
}
template<typename T>
inline typename enable_if<is_convertible<T,string>::value == false, actor_ostream&>::type
operator<<(actor_ostream& o, T&& arg) {
return o.write(to_string(std::forward<T>(arg)));
}
inline actor_ostream& operator<<(actor_ostream& o, actor_ostream::fun_type f) {
return f(o);
}
namespace { actor_ostream aout; }
inline actor_ostream& endl(actor_ostream& o) { return o.write("\n"); }
inline actor_ostream& flush(actor_ostream& o) { return o.flush(); }
inline string& ltrim(string &s) {
s.erase(s.begin(), find_if(s.begin(), s.end(), [](char c) { return !isspace(c); }));
return s;
......@@ -265,7 +174,6 @@ void client_repl(actor_ptr server, string host, uint16_t port) {
trim(line);
if (line == "quit") {
send(client, atom("quit"));
send(s_printer, atom("quit"));
return;
}
// the STL way of line.starts_with("connect")
......
......@@ -38,6 +38,7 @@
#include <sys/types.h>
#endif
#include "cppa/cppa.hpp"
#include "cppa/logging.hpp"
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
......@@ -140,3 +141,15 @@ logging* logging::instance() { return detail::singleton_manager::get_logger(); }
logging* logging::create_singleton() { return new logging_impl; }
} // namespace cppa
namespace std {
const cppa::actor_ostream& endl(const cppa::actor_ostream& o) {
return o.write("\n");
}
const cppa::actor_ostream& flush(const cppa::actor_ostream& o) {
return o.flush();
}
} // namespace std
......@@ -35,6 +35,7 @@
#include "cppa/on.hpp"
#include "cppa/self.hpp"
#include "cppa/receive.hpp"
#include "cppa/anything.hpp"
#include "cppa/to_string.hpp"
#include "cppa/scheduler.hpp"
......@@ -125,23 +126,34 @@ class scheduler_helper {
typedef intrusive_ptr<thread_mapped_actor> ptr_type;
scheduler_helper() : m_worker(new thread_mapped_actor) { }
void start() {
m_thread = std::thread(&scheduler_helper::time_emitter, m_worker);
ptr_type mtimer{new thread_mapped_actor};
ptr_type mprinter{new thread_mapped_actor};
m_timer_thread = std::thread(&scheduler_helper::timer_loop, mtimer);
m_printer_thread = std::thread(&scheduler_helper::printer_loop, mprinter);
m_timer = mtimer;
m_printer = mprinter;
}
void stop() {
m_worker->enqueue(nullptr, make_cow_tuple(atom("DIE")));
m_thread.join();
auto msg = make_any_tuple(atom("DIE"));
m_timer->enqueue(nullptr, msg);
m_printer->enqueue(nullptr, msg);
m_timer_thread.join();
m_printer_thread.join();
}
ptr_type m_worker;
std::thread m_thread;
actor_ptr m_timer;
std::thread m_timer_thread;
actor_ptr m_printer;
std::thread m_printer_thread;
private:
static void time_emitter(ptr_type m_self);
static void timer_loop(ptr_type m_self);
static void printer_loop(ptr_type m_self);
};
......@@ -159,7 +171,7 @@ void insert_dmsg(Map& storage,
storage.insert(std::make_pair(std::move(tout), std::move(dmsg)));
}
void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self) {
void scheduler_helper::timer_loop(scheduler_helper::ptr_type m_self) {
typedef detail::abstract_actor<local_actor> impl_type;
typedef std::unique_ptr<detail::recursive_queue_node,detail::disposer> queue_node_ptr;
// setup & local variables
......@@ -190,7 +202,7 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self) {
},
others() >> [&]() {
# ifdef CPPA_DEBUG
std::cerr << "scheduler_helper::time_emitter: UNKNOWN MESSAGE: "
std::cerr << "scheduler_helper::timer_loop: UNKNOWN MESSAGE: "
<< to_string(msg_ptr->msg)
<< std::endl;
# endif
......@@ -222,9 +234,65 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self) {
}
}
scheduler::scheduler() : m_helper(new scheduler_helper) {
void scheduler_helper::printer_loop(ptr_type m_self) {
self.set(m_self.get());
std::map<actor_ptr,std::string> out;
auto flush_output = [&out](const actor_ptr& s) {
auto i = out.find(s);
if (i != out.end()) {
auto& line = i->second;
if (!line.empty()) {
std::cout << line << std::flush;
line.clear();
}
}
};
auto flush_if_needed = [](std::string& str) {
if (str.back() == '\n') {
std::cout << str << std::flush;
str.clear();
}
};
bool running = true;
receive_while (gref(running)) (
on(atom("add"), arg_match) >> [&](std::string& str) {
auto s = self->last_sender();
if (!str.empty() && s != nullptr) {
auto i = out.find(s);
if (i == out.end()) {
i = out.insert(make_pair(s, move(str))).first;
// monitor actor to flush its output on exit
self->monitor(s);
flush_if_needed(i->second);
}
else {
auto& ref = i->second;
ref += move(str);
flush_if_needed(ref);
}
}
},
on(atom("flush")) >> [&] {
flush_output(self->last_sender());
},
on(atom("DOWN"), any_vals) >> [&] {
auto s = self->last_sender();
flush_output(s);
out.erase(s);
},
on(atom("DIE")) >> [&] {
running = false;
},
others() >> [] {
//cout << "*** unexpected: "
// << to_string(self->last_dequeued())
// << endl;
}
);
}
scheduler::scheduler() : m_helper(new scheduler_helper) { }
void scheduler::initialize() {
m_helper->start();
}
......@@ -239,7 +307,7 @@ scheduler::~scheduler() {
}
channel* scheduler::delayed_send_helper() {
return m_helper->m_worker.get();
return m_helper->m_timer.get();
}
void scheduler::register_converted_context(actor* what) {
......@@ -272,4 +340,9 @@ scheduler* scheduler::create_singleton() {
return new detail::thread_pool_scheduler;
}
const actor_ptr& scheduler::printer() const {
return m_helper->m_printer;
}
} // namespace cppa
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