Commit 71144fb7 authored by neverlord's avatar neverlord

actor_proxy bugfix

parent b88aaefe
...@@ -135,7 +135,7 @@ ...@@ -135,7 +135,7 @@
</data> </data>
<data> <data>
<variable>ProjectExplorer.Project.Updater.EnvironmentId</variable> <variable>ProjectExplorer.Project.Updater.EnvironmentId</variable>
<value type="QString">{23902c37-f07e-47cd-bb19-c366b9f708db}</value> <value type="QString">{07fcd197-092d-45a0-8500-3be614e6ae31}</value>
</data> </data>
<data> <data>
<variable>ProjectExplorer.Project.Updater.FileVersion</variable> <variable>ProjectExplorer.Project.Updater.FileVersion</variable>
......
...@@ -34,8 +34,7 @@ actor_proxy_ptr actor_proxy_cache::get(const key_tuple& key) ...@@ -34,8 +34,7 @@ actor_proxy_ptr actor_proxy_cache::get(const key_tuple& key)
return i->second; return i->second;
} }
actor_proxy_ptr result(new actor_proxy(std::get<0>(key), get_pinfo(key))); actor_proxy_ptr result(new actor_proxy(std::get<0>(key), get_pinfo(key)));
result->enqueue(message(result, nullptr, make_tuple(atom(":Monitor"), result->enqueue(message(result, nullptr, atom(":Monitor")));
result)));
add(result); add(result);
return result; return result;
} }
......
...@@ -2,6 +2,7 @@ ...@@ -2,6 +2,7 @@
#include <sstream> #include <sstream>
#include <algorithm> #include <algorithm>
#include "cppa/atom.hpp"
#include "cppa/object.hpp" #include "cppa/object.hpp"
#include "cppa/to_string.hpp" #include "cppa/to_string.hpp"
#include "cppa/serializer.hpp" #include "cppa/serializer.hpp"
...@@ -49,9 +50,9 @@ class string_serializer : public serializer ...@@ -49,9 +50,9 @@ class string_serializer : public serializer
}; };
int m_open_objects;
bool m_after_value; bool m_after_value;
bool m_obj_just_opened; bool m_obj_just_opened;
std::stack<std::string> m_open_objects;
inline void clear() inline void clear()
{ {
...@@ -70,15 +71,14 @@ class string_serializer : public serializer ...@@ -70,15 +71,14 @@ class string_serializer : public serializer
public: public:
string_serializer(std::ostream& mout) string_serializer(std::ostream& mout)
: out(mout), m_open_objects(0) : out(mout), m_after_value(false), m_obj_just_opened(false)
, m_after_value(false), m_obj_just_opened(false)
{ {
} }
void begin_object(const std::string& type_name) void begin_object(const std::string& type_name)
{ {
clear(); clear();
++m_open_objects; m_open_objects.push(type_name);
out << type_name;// << " ( "; out << type_name;// << " ( ";
m_obj_just_opened = true; m_obj_just_opened = true;
} }
...@@ -86,6 +86,7 @@ class string_serializer : public serializer ...@@ -86,6 +86,7 @@ class string_serializer : public serializer
{ {
if (m_obj_just_opened) if (m_obj_just_opened)
{ {
// no open brackets to close
m_obj_just_opened = false; m_obj_just_opened = false;
} }
else else
...@@ -93,6 +94,10 @@ class string_serializer : public serializer ...@@ -93,6 +94,10 @@ class string_serializer : public serializer
out << (m_after_value ? " )" : ")"); out << (m_after_value ? " )" : ")");
} }
m_after_value = true; m_after_value = true;
if (!m_open_objects.empty())
{
m_open_objects.pop();
}
} }
void begin_sequence(size_t) void begin_sequence(size_t)
...@@ -109,7 +114,24 @@ class string_serializer : public serializer ...@@ -109,7 +114,24 @@ class string_serializer : public serializer
void write_value(const primitive_variant& value) void write_value(const primitive_variant& value)
{ {
clear(); clear();
if (m_open_objects.empty())
{
throw std::runtime_error("write_value(): m_open_objects.empty()");
}
if (m_open_objects.top() == "@atom")
{
if (value.ptype() != pt_uint64)
{
throw std::runtime_error("expected uint64 value after @atom");
}
// write atoms as strings instead of integer values
auto av = static_cast<atom_value>(get<std::uint64_t>(value));
(pt_writer(out))(to_string(av));
}
else
{
value.apply(pt_writer(out)); value.apply(pt_writer(out));
}
m_after_value = true; m_after_value = true;
} }
...@@ -132,8 +154,9 @@ class string_deserializer : public deserializer ...@@ -132,8 +154,9 @@ class string_deserializer : public deserializer
std::string m_str; std::string m_str;
std::string::iterator m_pos; std::string::iterator m_pos;
size_t m_obj_count; //size_t m_obj_count;
std::stack<bool> m_obj_had_left_parenthesis; std::stack<bool> m_obj_had_left_parenthesis;
std::stack<std::string> m_open_objects;
void skip_space_and_comma() void skip_space_and_comma()
{ {
...@@ -190,11 +213,11 @@ class string_deserializer : public deserializer ...@@ -190,11 +213,11 @@ class string_deserializer : public deserializer
void integrity_check() void integrity_check()
{ {
if (m_obj_had_left_parenthesis.empty()) if (m_open_objects.empty() || m_obj_had_left_parenthesis.empty())
{ {
throw_malformed("missing begin_object()"); throw_malformed("missing begin_object()");
} }
else if (m_obj_had_left_parenthesis.top() == false) if (m_obj_had_left_parenthesis.top() == false)
{ {
throw_malformed("expected left parenthesis after " throw_malformed("expected left parenthesis after "
"begin_object call or void value"); "begin_object call or void value");
...@@ -206,13 +229,11 @@ class string_deserializer : public deserializer ...@@ -206,13 +229,11 @@ class string_deserializer : public deserializer
string_deserializer(const std::string& str) : m_str(str) string_deserializer(const std::string& str) : m_str(str)
{ {
m_pos = m_str.begin(); m_pos = m_str.begin();
m_obj_count = 0;
} }
string_deserializer(std::string&& str) : m_str(std::move(str)) string_deserializer(std::string&& str) : m_str(std::move(str))
{ {
m_pos = m_str.begin(); m_pos = m_str.begin();
m_obj_count = 0;
} }
std::string seek_object() std::string seek_object()
...@@ -236,9 +257,10 @@ class string_deserializer : public deserializer ...@@ -236,9 +257,10 @@ class string_deserializer : public deserializer
return result; return result;
} }
void begin_object(const std::string&) void begin_object(const std::string& type_name)
{ {
++m_obj_count; m_open_objects.push(type_name);
//++m_obj_count;
skip_space_and_comma(); skip_space_and_comma();
m_obj_had_left_parenthesis.push(try_consume('(')); m_obj_had_left_parenthesis.push(try_consume('('));
//consume('('); //consume('(');
...@@ -258,7 +280,12 @@ class string_deserializer : public deserializer ...@@ -258,7 +280,12 @@ class string_deserializer : public deserializer
} }
m_obj_had_left_parenthesis.pop(); m_obj_had_left_parenthesis.pop();
} }
if (--m_obj_count == 0) if (m_open_objects.empty())
{
throw std::runtime_error("no object to end");
}
m_open_objects.pop();
if (m_open_objects.empty())
{ {
skip_space_and_comma(); skip_space_and_comma();
if (m_pos != m_str.end()) if (m_pos != m_str.end())
...@@ -302,6 +329,19 @@ class string_deserializer : public deserializer ...@@ -302,6 +329,19 @@ class string_deserializer : public deserializer
primitive_variant read_value(primitive_type ptype) primitive_variant read_value(primitive_type ptype)
{ {
integrity_check(); integrity_check();
if (m_open_objects.top() == "@atom")
{
if (ptype != pt_uint64)
{
throw_malformed("expected read of pt_uint64 after @atom");
}
auto str_val = get<std::string>(read_value(pt_u8string));
if (str_val.size() > 10)
{
throw_malformed("atom string size > 10");
}
return detail::atom_val(str_val.c_str());
}
skip_space_and_comma(); skip_space_and_comma();
std::string::iterator substr_end; std::string::iterator substr_end;
auto find_if_cond = [] (char c) -> bool auto find_if_cond = [] (char c) -> bool
......
...@@ -257,12 +257,18 @@ class remote_observer : public attachable ...@@ -257,12 +257,18 @@ class remote_observer : public attachable
void detach(std::uint32_t reason) void detach(std::uint32_t reason)
{ {
actor_ptr self_ptr = self(); actor_ptr self_ptr = self();
message msg(self_ptr, self_ptr, make_tuple(atom(":KillProxy"), reason)); message msg(self_ptr, self_ptr, atom(":KillProxy"), reason);
s_mailman_queue().push_back(new mailman_job(peer, msg)); s_mailman_queue().push_back(new mailman_job(peer, msg));
} }
}; };
template<typename T>
T& operator<<(T& o, const process_information& pinfo)
{
return (o << pinfo.process_id << "@" << pinfo.node_id_as_string());
}
// handles *all* outgoing messages // handles *all* outgoing messages
void mailman_loop() void mailman_loop()
{ {
...@@ -280,23 +286,6 @@ void mailman_loop() ...@@ -280,23 +286,6 @@ void mailman_loop()
{ {
mailman_send_job& sjob = job->send_job(); mailman_send_job& sjob = job->send_job();
const message& out_msg = sjob.original_message; const message& out_msg = sjob.original_message;
/*
// keep track about link states of local actors
// (remove link states between local and remote actors if needed)
if (match<atom(":Exit"), std::uint32_t>(out_msg.content()))
{
auto sender = out_msg.sender();
if (pself == sender->parent_process())
{
// local to remote (local actor just died)
//sjob.client->unlink_from(sender);
}
else
{
// remote to remote (ignored)
}
}
*/
// forward message to receiver peer // forward message to receiver peer
auto peer_element = peers.find(*(sjob.target_peer)); auto peer_element = peers.find(*(sjob.target_peer));
if (peer_element != peers.end()) if (peer_element != peers.end())
...@@ -315,6 +304,7 @@ cout << "--> " << to_string(out_msg) << endl; ...@@ -315,6 +304,7 @@ cout << "--> " << to_string(out_msg) << endl;
if (sent == -1) if (sent == -1)
{ {
// peer unreachable // peer unreachable
cout << "peer " << *(sjob.target_peer) << " unreachable" << endl;
peers.erase(*(sjob.target_peer)); peers.erase(*(sjob.target_peer));
} }
} }
...@@ -374,22 +364,16 @@ void read_from_socket(native_socket_t sfd, void* buf, size_t buf_size) ...@@ -374,22 +364,16 @@ void read_from_socket(native_socket_t sfd, void* buf, size_t buf_size)
while (urres < left); while (urres < left);
} }
template<typename T>
T& operator<<(T& o, const process_information& pinfo)
{
return (o << pinfo.process_id << "@" << pinfo.node_id_as_string());
}
// handles *one* socket / peer // handles *one* socket / peer
void post_office_loop(native_socket_t socket_fd, void post_office_loop(native_socket_t socket_fd,
process_information_ptr peer, process_information_ptr peer,
actor_proxy_ptr aptr) actor_proxy_ptr aptr)
{ {
cout << "--> post_office_loop; self() = " //cout << "--> post_office_loop; self() = "
<< process_information::get() // << process_information::get()
<< ", peer = " // << ", peer = "
<< *peer // << *peer
<< endl; // << endl;
if (aptr) detail::get_actor_proxy_cache().add(aptr); if (aptr) detail::get_actor_proxy_cache().add(aptr);
message msg; message msg;
std::uint32_t rsize; std::uint32_t rsize;
...@@ -417,7 +401,7 @@ cout << "--> post_office_loop; self() = " ...@@ -417,7 +401,7 @@ cout << "--> post_office_loop; self() = "
buf = new char[buf_allocated]; buf = new char[buf_allocated];
} }
buf_size = rsize; buf_size = rsize;
cout << "[" << pinfo << "] " << "received " << rsize << " bytes" << endl; //cout << "[" << pinfo << "] read " << rsize << " bytes" << endl;
read_from_socket(socket_fd, buf, buf_size); read_from_socket(socket_fd, buf, buf_size);
binary_deserializer bd(buf, buf_size); binary_deserializer bd(buf, buf_size);
meta_msg->deserialize(&msg, &bd); meta_msg->deserialize(&msg, &bd);
...@@ -430,8 +414,8 @@ cout << "<-- " << to_string(msg) << endl; ...@@ -430,8 +414,8 @@ cout << "<-- " << to_string(msg) << endl;
actor_ptr sender = msg.sender(); actor_ptr sender = msg.sender();
if (sender->parent_process() == pinfo) if (sender->parent_process() == pinfo)
{ {
cout << pinfo.process_id << "@" << pinfo.node_id_as_string() cout << pinfo << " ':Monitor'; actor id = "
<< " :Monitor; actor id = " << sender->id() << endl; << sender->id() << endl;
// local actor? // local actor?
// this message was send from a proxy // this message was send from a proxy
sender->attach(new remote_observer(peer)); sender->attach(new remote_observer(peer));
...@@ -453,7 +437,8 @@ cout << pinfo.process_id << "@" << pinfo.node_id_as_string() ...@@ -453,7 +437,8 @@ cout << pinfo.process_id << "@" << pinfo.node_id_as_string()
<< detail::to_uniform_name(typeid(e)) << ": " << detail::to_uniform_name(typeid(e)) << ": "
<< e.what() << endl; << e.what() << endl;
} }
cout << "<-- post_office_loop" << endl; cout << "[" << process_information::get() << "] ~post_office_loop" << endl;
//cout << "<-- post_office_loop" << endl;
} }
struct mm_worker struct mm_worker
......
...@@ -77,7 +77,6 @@ int main(int argc, char** c_argv) ...@@ -77,7 +77,6 @@ int main(int argc, char** c_argv)
else if (argv.size() == 2 && argv.front() == "test__remote_actor") else if (argv.size() == 2 && argv.front() == "test__remote_actor")
{ {
test__remote_actor(c_argv[0], true, argv); test__remote_actor(c_argv[0], true, argv);
cout << "BLABLUBB" << endl;
return 0; return 0;
} }
else else
......
...@@ -24,6 +24,7 @@ void client_part(const std::vector<std::string>& argv) ...@@ -24,6 +24,7 @@ void client_part(const std::vector<std::string>& argv)
catch (...) catch (...)
{ {
} }
await_all_others_done();
} }
} // namespace <anonymous> } // namespace <anonymous>
...@@ -56,7 +57,7 @@ size_t test__remote_actor(const char* app_path, bool is_client, ...@@ -56,7 +57,7 @@ size_t test__remote_actor(const char* app_path, bool is_client,
std::string cmd; std::string cmd;
{ {
std::ostringstream oss; std::ostringstream oss;
oss << app_path << " test__remote_actor " << port;// << " &>/dev/null"; oss << app_path << " test__remote_actor " << port << " &>/dev/null";
cmd = oss.str(); cmd = oss.str();
} }
// execute client_part() in a separate process, // execute client_part() in a separate process,
......
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