Move send in message because this is where it is supposed to be

This commit is contained in:
Julien CLEMENT 2020-05-07 21:27:14 +02:00
parent 1109049a39
commit 28584759b8
7 changed files with 66 additions and 62 deletions

View File

@ -143,11 +143,11 @@ Each legislator could initiate a ballot by choosing its number, decree and quoru
To maintain $B1(\beta)$, each ballot needs to receive a unique number. A legislator can take notes in his ledger to remember of which ballot he already initiated to avoid picking the same number twice. To avoid two different legislators from initiating ballots with the same number, the set of possible ballot number is partitionned among the legislators. There are multiple way of doing this, in our implementation we give ballot numbers to legislators following a round-robin basis.\\
If we have 3 legislators (A, B and C), legislator A will have ballot numbers 0, 3, 6 ...\\
legislator B will have 1, 4, 5 ... and legislator C will have 2, 5, 8 ...
legislator B will have 1, 4, 7 ... and legislator C will have 2, 5, 8 ...
To maintain $B2(\beta)$ we can simply choose the quorum as any majority set among the legislators.\\
$B3(\beta)$
$B3(\beta)$ requires that the legislator initiating the ballot knows the decree every legislators in the quorum voted for. To do this will have to exchange messages
\subsection{The complete protocol}

View File

@ -49,54 +49,4 @@ namespace paxos
event_register.unregister_ew(this);
}
static shared_socket connect_to_ip_port(std::string ip, std::string port)
{
/* prepare hints structure */
misc::AddrInfoHint hints;
hints.family(AF_UNSPEC);
hints.socktype(SOCK_STREAM);
misc::AddrInfo result = misc::getaddrinfo(ip.c_str(), port.c_str(), hints);
for (auto it : result)
{
try
{
auto vhost_socket = std::make_shared<DefaultSocket>
(it.ai_family, it.ai_socktype, it.ai_protocol);
vhost_socket->connect(it.ai_addr, it.ai_addrlen);
vhost_socket->fcntl_set_O_NONBLOCK();
return vhost_socket;
}
catch (const std::exception& e)
{
continue;
}
}
throw ConnectionFailed();
}
void SendEW::send_message(Message message, shared_legislator legislator)
{
std::string ip = legislator->config_.ip;
std::string port = legislator->config_.port;
std::string name = legislator->config_.name;
message.add_header("receiver", name);
shared_socket socket;
try
{
socket = connect_to_ip_port(ip, port);
}
catch (const ConnectionFailed& e)
{
log("Could not connect to " + name, red);
return;
}
shared_connection connection = std::make_shared<Connection>(socket);
message.fill_buffer(connection->get_buffer_out());
event_register.register_event<SendEW>(connection);
}
}

View File

@ -26,8 +26,6 @@ namespace paxos
*/
void operator()() final;
static void send_message(Message message, shared_legislator legislator);
private:
shared_connection connection_;
};

View File

@ -2,7 +2,6 @@
#include "legislator.hh"
#include "misc/logger.hh"
#include "message/message.hh"
#include "events/register.hh"
#include "events/send.hh"
#include "vote.hh"
@ -45,7 +44,7 @@ namespace paxos
message.add_header("sender", self->config_.name);
for (auto legislator : legislators)
SendEW::send_message(message, legislator.second);
message.send(legislator.second);
}
void Legislator::receive_next_ballot(Message message)
@ -80,7 +79,7 @@ namespace paxos
Message message;
message.set_method("HigherBallot");
message.add_header("ballot", std::to_string(ballot));
SendEW::send_message(message, legislators[receiver]);
message.send(legislators[receiver]);
}
void Legislator::receive_higher_ballot(Message message)
@ -124,7 +123,7 @@ namespace paxos
message.add_header("vote_ballot_id", vote_ballot_id_string);
message.add_header("decree", decree_string);
message.add_header("sender", self->config_.name);
SendEW::send_message(message, legislators[sender]);
message.send(legislators[sender]);
}
@ -208,7 +207,7 @@ namespace paxos
for (auto legislator_vote_pair : quorum_previous_votes)
{
std::string legislator = legislator_vote_pair.first;
SendEW::send_message(message, legislators[legislator]);
message.send(legislators[legislator]);
}
}
@ -254,7 +253,7 @@ namespace paxos
message.add_header("ballot", std::to_string(ballot));
message.add_header("sender", self->config_.name);
message.add_header("decree", std::to_string(decree.decree));
SendEW::send_message(message, legislators[receiver]);
message.send(legislators[receiver]);
}
void Legislator::receive_voted(Message message)
@ -297,7 +296,7 @@ namespace paxos
message.add_header("decree", std::to_string(decree.decree));
for (auto legislator : legislators)
SendEW::send_message(message, legislator.second);
message.send(legislator.second);
}
void Legislator::receive_success(Message message)

View File

@ -1,7 +1,6 @@
#pragma once
#include "config/config.hh"
#include "ledger.hh"
#include "message/message.hh"
#include "vote.hh"
#include "law/decree.hh"
#include <unordered_map>
@ -9,6 +8,7 @@
namespace paxos
{
class Message;
class Legislator
{
public:

View File

@ -4,6 +4,11 @@
#include "error/parsing-error.hh"
#include "message.hh"
#include "misc/addrinfo/addrinfo.hh"
#include "error/connection-failed.hh"
#include "events/send.hh"
#include "misc/logger.hh"
#include "events/register.hh"
namespace paxos
@ -182,4 +187,53 @@ namespace paxos
return true;
}
static shared_socket connect_to_ip_port(std::string ip, std::string port)
{
/* prepare hints structure */
misc::AddrInfoHint hints;
hints.family(AF_UNSPEC);
hints.socktype(SOCK_STREAM);
misc::AddrInfo result = misc::getaddrinfo(ip.c_str(), port.c_str(), hints);
for (auto it : result)
{
try
{
auto vhost_socket = std::make_shared<DefaultSocket>
(it.ai_family, it.ai_socktype, it.ai_protocol);
vhost_socket->connect(it.ai_addr, it.ai_addrlen);
vhost_socket->fcntl_set_O_NONBLOCK();
return vhost_socket;
}
catch (const std::exception& e)
{
continue;
}
}
throw ConnectionFailed();
}
void Message::send(shared_legislator receiver)
{
std::string ip = receiver->config_.ip;
std::string port = receiver->config_.port;
std::string name = receiver->config_.name;
add_header("receiver", name);
shared_socket socket;
try
{
socket = connect_to_ip_port(ip, port);
}
catch (const ConnectionFailed& e)
{
log("Could not connect to " + name, red);
return;
}
shared_connection connection = std::make_shared<Connection>(socket);
fill_buffer(connection->get_buffer_out());
event_register.register_event<SendEW>(connection);
}
}

View File

@ -2,6 +2,7 @@
#include <unordered_map>
#include "misc/buffer.hh"
#include "legislator/legislator.hh"
namespace paxos
{
@ -64,6 +65,8 @@ namespace paxos
void fill_buffer(misc::Buffer& buffer);
void send(shared_legislator receiver);
protected:
std::string method_;