From ede1ec450051a6da9260542cc8e2f5cb0569e6ee Mon Sep 17 00:00:00 2001 From: Michael Pivato Date: Fri, 1 Mar 2019 15:44:58 +1030 Subject: [PATCH] Add ability to propogate result of vote and making a vote. --- .../CentralisedDecision/ballotvoter.py | 42 +++++-------------- .../CentralisedDecision/commander.py | 11 ++++- DecisionSystem/messages.py | 12 +++++- 3 files changed, 30 insertions(+), 35 deletions(-) diff --git a/DecisionSystem/CentralisedDecision/ballotvoter.py b/DecisionSystem/CentralisedDecision/ballotvoter.py index b5b2e47..deb9f65 100644 --- a/DecisionSystem/CentralisedDecision/ballotvoter.py +++ b/DecisionSystem/CentralisedDecision/ballotvoter.py @@ -1,33 +1,19 @@ import json -from DecisionSystem.messages import ConnectSwarm, SubmitVote, Message, deserialise, RequestVote +from DecisionSystem.messages import ConnectSwarm, SubmitVote, Message, deserialise, RequestVote, ClientVoteRequest, VoteResult from multiprocessing import Pool from messenger import Messenger class BallotVoter: - def __init__(self, on_vote, messenger: Messenger): - - # Load config file - cfg = None - with open('config.json') as json_config: - cfg = json.load(json_config) - self._cfg = cfg - + def __init__(self, on_vote, handle_agreement, messenger: Messenger): self.messenger = messenger self.messenger.add_message_callback(self.on_message) self.messenger.add_connect(self.on_connect) self.on_vote = on_vote + self.handle_agreement = handle_agreement def on_connect(self, rc): print("Connected with result code " + str(rc)) - # Should subscribes be handled by the messenger? - #self.client.subscribe('swarm1/voters', qos=1) - - # Need to set a will as well to broadcast on unexpected disconnection, so commander - # knows it is no longer part of the set of voters. - # Leaving this until core centralised system is working. - #will_message = {"type": "UDisconnect"} - # Tell commander we are now connected. self.send_connect() @@ -41,12 +27,8 @@ class BallotVoter: self.submit_vote() elif messageD.type == "listening": self.send_connect() - elif messageD.type == "disconnectedcommander": - # Elect new leader... - # Might just use raft's method for electing a leader, - # by doing a random timeout then choosing itself as the candidate, - # and collects who is part of the swarm. This may need to be another class... - pass + elif messageD.type == VoteResult.type: + self.handle_agreement(messageD.data["vote"]) def submit_vote(self): v = self.on_vote() @@ -59,15 +41,11 @@ class BallotVoter: self.messenger.broadcast_message(self.messenger.swarm, vote.serialise()) print('published vote') - def submit_vote_multicore(self): - ''' - Uses multiple processes to work on multiple frames at the same time - to take an average for this voter, or if it could not find a hand - in one frame but can in another. - ''' - pass - def send_connect(self): # Send a connected message to let any commanders know that # it is available. - self.messenger.broadcast_message(self.messenger.swarm, ConnectSwarm(self.messenger.id).serialise()) \ No newline at end of file + self.messenger.broadcast_message(self.messenger.swarm, ConnectSwarm(self.messenger.id).serialise()) + + def request_vote(self): + """Sends a request to the leader to start collecting votes.""" + self.messenger.broadcast_message(self.messenger.swarm, ClientVoteRequest(self.messenger.id).serialise()) \ No newline at end of file diff --git a/DecisionSystem/CentralisedDecision/commander.py b/DecisionSystem/CentralisedDecision/commander.py index 7ecf669..d84e7cf 100644 --- a/DecisionSystem/CentralisedDecision/commander.py +++ b/DecisionSystem/CentralisedDecision/commander.py @@ -1,5 +1,5 @@ import time -from DecisionSystem.messages import Message, CommanderWill, RequestVote, GetSwarmParticipants, deserialise +from DecisionSystem.messages import Message, CommanderWill, RequestVote, GetSwarmParticipants, deserialise, ClientVoteRequest, VoteResult import json import numpy as np @@ -62,6 +62,7 @@ class Commander: print("published message") time.sleep(self.timeout) self._taking_votes = False + # TODO: Work out how to broadcast votes back to the swarm, maybe using raft? return self.make_decision() def on_message(self, message): @@ -90,6 +91,9 @@ class Commander: # Only add vote to list if the client has not already voted. if messageD.sender not in self._votes: self._votes[messageD.sender] = int(messageD.data["vote"]) + elif messageD.type == ClientVoteRequest().type: + # received a request to get votes/consensus. + self.get_votes() elif messageD.type == "disconnected": print("Voter disconnected :(") @@ -109,4 +113,7 @@ class Commander: # self.client.publish("swarm1/voters", CommanderWill(self.client._client_id).serialise()) def on_disconnect(self, rc): - pass \ No newline at end of file + pass + + def propogate_result(self, result): + self._messenger.broadcast_message(self._messenger.swarm, ) \ No newline at end of file diff --git a/DecisionSystem/messages.py b/DecisionSystem/messages.py index 49e61d6..5d47435 100644 --- a/DecisionSystem/messages.py +++ b/DecisionSystem/messages.py @@ -88,4 +88,14 @@ class SubmitVote(Message): self._data["vote"] = value class GetSwarmParticipants(Message): - _type = "listening" \ No newline at end of file + _type = "listening" + +class VoteResult(Message): + _type = "voteresult" + + def __init__(self, vote, sender='', data={}): + super().__init__(sender=sender, data=data) + self._data["vote"] = vote + +class ClientVoteRequest(Message): + _type = "clientvoterequest" \ No newline at end of file