From 28584759b808e7d1838802b237178132e4ec7cc1 Mon Sep 17 00:00:00 2001 From: Julien Clement Date: Thu, 7 May 2020 21:27:14 +0200 Subject: [PATCH] Move send in message because this is where it is supposed to be --- latex/src/main.tex | 4 +-- src/events/send.cc | 50 --------------------------------- src/events/send.hh | 2 -- src/legislator/legislator.cc | 13 ++++----- src/legislator/legislator.hh | 2 +- src/message/message.cc | 54 ++++++++++++++++++++++++++++++++++++ src/message/message.hh | 3 ++ 7 files changed, 66 insertions(+), 62 deletions(-) diff --git a/latex/src/main.tex b/latex/src/main.tex index 156ad90..3c3f31c 100644 --- a/latex/src/main.tex +++ b/latex/src/main.tex @@ -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} diff --git a/src/events/send.cc b/src/events/send.cc index 2afefdd..9be059e 100644 --- a/src/events/send.cc +++ b/src/events/send.cc @@ -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 - (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(socket); - message.fill_buffer(connection->get_buffer_out()); - - event_register.register_event(connection); - } } diff --git a/src/events/send.hh b/src/events/send.hh index 2f695ba..4b0e6c6 100644 --- a/src/events/send.hh +++ b/src/events/send.hh @@ -26,8 +26,6 @@ namespace paxos */ void operator()() final; - static void send_message(Message message, shared_legislator legislator); - private: shared_connection connection_; }; diff --git a/src/legislator/legislator.cc b/src/legislator/legislator.cc index a21e8bd..c77022d 100644 --- a/src/legislator/legislator.cc +++ b/src/legislator/legislator.cc @@ -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) diff --git a/src/legislator/legislator.hh b/src/legislator/legislator.hh index d9afaeb..32fbd4c 100644 --- a/src/legislator/legislator.hh +++ b/src/legislator/legislator.hh @@ -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 @@ -9,6 +8,7 @@ namespace paxos { + class Message; class Legislator { public: diff --git a/src/message/message.cc b/src/message/message.cc index 4b84997..a4ed652 100644 --- a/src/message/message.cc +++ b/src/message/message.cc @@ -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 + (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(socket); + fill_buffer(connection->get_buffer_out()); + + event_register.register_event(connection); + } + } diff --git a/src/message/message.hh b/src/message/message.hh index f876e87..6797f00 100644 --- a/src/message/message.hh +++ b/src/message/message.hh @@ -2,6 +2,7 @@ #include #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_;