local_actor_ptr m_mailman;

std::thread m_mailman_thread;

local_actor_ptr m_post_office;

std::thread m_post_office_thread;

int pipe_fd[2];

void start() { // override

if (pipe(pipe_fd) != 0) {

CPPA_CRITICAL("cannot create pipe");

}

// create actors

m_post_office.reset(new thread_mapped_actor);

m_mailman.reset(new thread_mapped_actor);

// store some data in local variables for lambdas

int pipe_fd0 = pipe_fd[0];

auto po_ptr = m_post_office;

auto mm_ptr = m_mailman;

// start threads

m_post_office_thread = std::thread([pipe_fd0, po_ptr]() {

scoped_self_setter sss{po_ptr.get()};

post_office_loop(pipe_fd0);

});

m_mailman_thread = std::thread([mm_ptr]() {

scoped_self_setter sss{mm_ptr.get()};

mailman_loop();

});

}

void stop() { // override

m_mailman->enqueue(nullptr, make_any_tuple(atom("DONE")));

m_mailman_thread.join();

// wait until mailman is done; post_office closes all sockets

std::atomic_thread_fence(std::memory_order_seq_cst);

send_to_post_office(po_message{atom("DONE"), -1, 0});

m_post_office_thread.join();

close(pipe_fd[0]);

close(pipe_fd[0]);

}

void send_to_post_office(const po_message& msg) {

if (write(pipe_fd[1], &msg, sizeof(po_message)) != sizeof(po_message)) {

CPPA_CRITICAL("cannot write to pipe");

}

}

void send_to_post_office(any_tuple msg) {

m_post_office->enqueue(nullptr, std::move(msg));

}

void send_to_mailman(any_tuple msg) {

m_mailman->enqueue(nullptr, std::move(msg));

}