Commit 3132985f authored by neverlord's avatar neverlord

use buffers

parent c390beaa
...@@ -73,8 +73,8 @@ ...@@ -73,8 +73,8 @@
<< cppa::process_information::get()->process_id() \ << cppa::process_information::get()->process_id() \
<< "] " << arg << std::endl << "] " << arg << std::endl
#undef DEBUG //#undef DEBUG
#define DEBUG(unused) ((void) 0) //#define DEBUG(unused) ((void) 0)
using std::cout; using std::cout;
using std::cerr; using std::cerr;
...@@ -176,18 +176,28 @@ class po_peer ...@@ -176,18 +176,28 @@ class po_peer
send2po(atom("RM_PEER"), m_socket); send2po(atom("RM_PEER"), m_socket);
}); });
DEBUG("po_peer loop started"); DEBUG("po_peer loop started");
std::uint32_t msg_size; buffer<512, (16 * 1024 * 1024)> m_buf;
std::vector<char> msg_buf; m_buf.reset(m_state == wait_for_process_info ? sizeof(std::uint32_t) + process_information::node_id_size
: sizeof(std::uint32_t));
for (;;) for (;;)
{ {
while(m_buf.ready() == false)
{
if (m_buf.append_from(m_socket, 0) == false)
{
DEBUG("cannot read from socket");
return;
}
}
switch (m_state) switch (m_state)
{ {
case wait_for_process_info: case wait_for_process_info:
{ {
std::uint32_t process_id; std::uint32_t process_id;
process_information::node_id_type node_id; process_information::node_id_type node_id;
recv(m_socket, &process_id, sizeof(std::uint32_t), 0); memcpy(&process_id, m_buf.data(), sizeof(std::uint32_t));
recv(m_socket, node_id.data(), node_id.size(), 0); memcpy(node_id.data(), m_buf.data() + sizeof(std::uint32_t),
process_information::node_id_size);
m_peer.reset(new process_information(process_id, node_id)); m_peer.reset(new process_information(process_id, node_id));
// inform mailman about new peer // inform mailman about new peer
nm->send_to_mailman(make_any_tuple(m_socket, m_peer)); nm->send_to_mailman(make_any_tuple(m_socket, m_peer));
...@@ -196,28 +206,21 @@ class po_peer ...@@ -196,28 +206,21 @@ class po_peer
<< m_peer->process_id() << m_peer->process_id()
<< "@" << "@"
<< to_string(m_peer->node_id())); << to_string(m_peer->node_id()));
// fall through and try to read more from socket m_buf.reset(sizeof(std::uint32_t));
break;
} }
case wait_for_msg_size: case wait_for_msg_size:
{ {
recv(m_socket, &msg_size, sizeof(std::uint32_t), 0); std::uint32_t msg_size;
if (msg_size > (16 * 1024 * 1024)) memcpy(&msg_size, m_buf.data(), sizeof(std::uint32_t));
{ m_buf.reset(msg_size);
// maximum of 16MB
return;
}
if (msg_buf.size() < msg_size)
{
msg_buf.resize(msg_size);
}
m_state = read_message; m_state = read_message;
// fall through and try to read more from socket break;
} }
case read_message: case read_message:
{ {
recv(m_socket, msg_buf.data(), msg_size, 0);
addressed_message msg; addressed_message msg;
binary_deserializer bd(msg_buf.data(), msg_size); binary_deserializer bd(m_buf.data(), m_buf.size());
try try
{ {
meta_msg->deserialize(&msg, &bd); meta_msg->deserialize(&msg, &bd);
...@@ -287,6 +290,7 @@ class po_peer ...@@ -287,6 +290,7 @@ class po_peer
} }
} }
); );
m_buf.reset(sizeof(std::uint32_t));
m_state = wait_for_msg_size; m_state = wait_for_msg_size;
break; break;
} }
......
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