From 7956fb8b67e6f10760431cbe77db2fcf33d5e9e0 Mon Sep 17 00:00:00 2001 From: Marcin Chrzanowski Date: Fri, 10 Jan 2020 20:14:43 +0100 Subject: Send state updates after gossip --- .../mimuw/cloudatlas/agent/modules/GossipGirl.java | 42 ++++++++ .../cloudatlas/agent/modules/GossipGirlState.java | 114 ++++++++++++++++++--- 2 files changed, 139 insertions(+), 17 deletions(-) (limited to 'src/main/java/pl/edu/mimuw/cloudatlas/agent/modules') diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirl.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirl.java index 73aef8b..e7bd227 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirl.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirl.java @@ -1,5 +1,6 @@ package pl.edu.mimuw.cloudatlas.agent.modules; +import java.util.AbstractMap.SimpleImmutableEntry; import java.util.HashMap; import java.util.Map; import java.util.Map.Entry; @@ -14,6 +15,8 @@ import pl.edu.mimuw.cloudatlas.agent.messages.QueryMessage; import pl.edu.mimuw.cloudatlas.agent.messages.ResponseMessage; import pl.edu.mimuw.cloudatlas.agent.messages.StateMessage; import pl.edu.mimuw.cloudatlas.agent.messages.UDUPMessage; +import pl.edu.mimuw.cloudatlas.agent.messages.UpdateAttributesMessage; +import pl.edu.mimuw.cloudatlas.agent.messages.UpdateQueriesMessage; import pl.edu.mimuw.cloudatlas.model.Attribute; import pl.edu.mimuw.cloudatlas.model.AttributesMap; import pl.edu.mimuw.cloudatlas.model.PathName; @@ -38,6 +41,12 @@ public class GossipGirl extends Module { case NO_CO_TAM: handleNoCoTam((NoCoTamMessage) message); break; + case ATTRIBUTES: + handleAttributes((AttributesMessage) message); + break; + case QUERY: + handleQuery((QueryMessage) message); + break; default: throw new InvalidMessageType("This type of message cannot be handled by GossipGirl"); } @@ -103,4 +112,37 @@ public class GossipGirl extends Module { System.out.println("ERROR: GossipGirl got state for a nonexistent gossip"); } } + + private void handleAttributes(AttributesMessage message) throws InterruptedException { + GossipGirlState state = gossipStates.get(message.getReceiverGossipId()); + if (state != null) { + state.gotAttributesFor(message.getPath()); + UpdateAttributesMessage updateMessage = new UpdateAttributesMessage("", 0, message.getPath().toString(), message.getAttributes()); + sendMessage(updateMessage); + if (state.state == GossipGirlState.State.FINISHED) { + gossipStates.remove(message.getReceiverGossipId()); + } + } else { + System.out.println("ERROR: GossipGirl got attributes for a nonexistent gossip"); + } + } + + private void handleQuery(QueryMessage message) throws InterruptedException { + GossipGirlState state = gossipStates.get(message.getReceiverGossipId()); + if (state != null) { + state.gotQuery(message.getName()); + Map> queries = new HashMap(); + queries.put( + message.getName(), + new SimpleImmutableEntry(message.getQuery(), state.getTheirQueryTimestamp(message.getName())) + ); + UpdateQueriesMessage updateMessage = new UpdateQueriesMessage("", 0, queries); + sendMessage(updateMessage); + if (state.state == GossipGirlState.State.FINISHED) { + gossipStates.remove(message.getReceiverGossipId()); + } + } else { + System.out.println("ERROR: GossipGirl got query for a nonexistent gossip"); + } + } } diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirlState.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirlState.java index df17957..8b0711e 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirlState.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirlState.java @@ -2,11 +2,13 @@ package pl.edu.mimuw.cloudatlas.agent.modules; import java.util.AbstractMap.SimpleImmutableEntry; import java.util.HashMap; +import java.util.HashSet; import java.util.Iterator; import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Map.Entry; +import java.util.Set; import pl.edu.mimuw.cloudatlas.agent.messages.NoCoTamMessage; import pl.edu.mimuw.cloudatlas.model.Attribute; @@ -26,6 +28,7 @@ public class GossipGirlState { WAIT_FOR_NO_CO_TAM, WAIT_FOR_FIRST_INFO, WAIT_FOR_INFO, + FINISHED, ERROR } public PathName ourPath; @@ -42,6 +45,10 @@ public class GossipGirlState { public ValueTime noCoTamSendReceiveTimestamp; private Map theirZoneTimestamps; private Map theirQueryTimestamps; + private List zonesToSend; + private List queriesToSend; + private Set waitingForZones; + private Set waitingForQueries; public GossipGirlState(long gossipId, PathName ourPath, ValueContact theirContact, boolean initiating) { this.gossipId = gossipId; @@ -91,6 +98,9 @@ public class GossipGirlState { theirQueryTimestamps = message.getQueryTimestamps(); hejkaSendTimestamp = message.getHejkaSendTimestamp(); hejkaReceiveTimestamp = message.getHejkaReceiveTimestamp(); + setZonesToSend(); + setQueriesToSend(); + setWaitingFor(); state = State.SEND_INFO; break; default: @@ -99,6 +109,25 @@ public class GossipGirlState { } } + private void setWaitingFor() { + setWaitingForZones(); + setWaitingForQueries(); + } + + private void setWaitingForZones() { + waitingForZones = new HashSet(theirZoneTimestamps.keySet()); + for (PathName path : zonesToSend) { + waitingForZones.remove(path); + } + } + + private void setWaitingForQueries() { + waitingForQueries = new HashSet(theirQueryTimestamps.keySet()); + for (Attribute name : queriesToSend) { + waitingForQueries.remove(name); + } + } + public Map getZoneTimestampsToSend() { Map timestamps = new HashMap(); collectZoneTimestamps(timestamps, hierarchy, theirContact.getName()); @@ -114,35 +143,48 @@ public class GossipGirlState { return queryTimestamps; } - public List getZMIsToSend() { - List zmis = new LinkedList(); + public void setZonesToSend() { + zonesToSend = new LinkedList(); for (Entry timestampedPath : getZoneTimestampsToSend().entrySet()) { ValueTime theirTimestamp = theirZoneTimestamps.get(timestampedPath.getKey()); if (theirTimestamp == null || ValueUtils.valueLower(theirTimestamp, timestampedPath.getValue())) { - System.out.println("going to send " + timestampedPath.getKey().toString()); - try { - zmis.add(hierarchy.findDescendant(timestampedPath.getKey())); - } catch (ZMI.NoSuchZoneException e) { - System.out.println("ERROR: didn't find a zone we wanted to send in getZMIsToSend"); - } + zonesToSend.add(timestampedPath.getKey()); } } - return zmis; } - public List> getQueriesToSend() { - List> queryList = new LinkedList(); + public void setQueriesToSend() { + queriesToSend = new LinkedList(); for (Entry timestampedQuery : getQueryTimestampsToSend().entrySet()) { ValueTime theirTimestamp = theirQueryTimestamps.get(timestampedQuery.getKey()); if (theirTimestamp == null || ValueUtils.valueLower(theirTimestamp, timestampedQuery.getValue())) { - queryList.add( - new SimpleImmutableEntry( - timestampedQuery.getKey(), - queries.get(timestampedQuery.getKey()).getKey() - ) - ); + queriesToSend.add(timestampedQuery.getKey()); } } + } + + public List getZMIsToSend() { + List zmis = new LinkedList(); + for (PathName path : zonesToSend) { + try { + zmis.add(hierarchy.findDescendant(path)); + } catch (ZMI.NoSuchZoneException e) { + System.out.println("ERROR: didn't find a zone we wanted to send in getZMIsToSend"); + } + } + return zmis; + } + + public List> getQueriesToSend() { + List> queryList = new LinkedList(); + for (Attribute name : queriesToSend) { + queryList.add( + new SimpleImmutableEntry( + name, + queries.get(name).getKey() + ) + ); + } return queryList; } @@ -194,4 +236,42 @@ public class GossipGirlState { state = State.ERROR; } } + + public void gotAttributesFor(PathName path) { + switch (state) { + case WAIT_FOR_INFO: + if (!waitingForZones.remove(path)) { + System.out.println("DEBUG: got zone attributes we weren't expecting"); + } + if (waitingForZones.isEmpty() && waitingForQueries.isEmpty()) { + System.out.println("INFO: done waiting for info"); + state = state.FINISHED; + } + break; + default: + System.out.println("ERROR: got attributes when not expected"); + state = State.ERROR; + } + } + + public void gotQuery(Attribute name) { + switch (state) { + case WAIT_FOR_INFO: + if (!waitingForQueries.remove(name)) { + System.out.println("DEBUG: got query we weren't expecting"); + } + if (waitingForZones.isEmpty() && waitingForQueries.isEmpty()) { + System.out.println("INFO: done waiting for info"); + state = state.FINISHED; + } + break; + default: + System.out.println("ERROR: got query when not expected"); + state = State.ERROR; + } + } + + public ValueTime getTheirQueryTimestamp(Attribute name) { + return theirQueryTimestamps.get(name); + } } -- cgit v1.2.3 From 89af38da0b95445180440e85bb8248ab910ef9f8 Mon Sep 17 00:00:00 2001 From: Marcin Chrzanowski Date: Fri, 10 Jan 2020 22:58:32 +0100 Subject: Send NoCoTam when responder gets state --- .../mimuw/cloudatlas/agent/modules/GossipGirl.java | 37 ++++++++++++++++++++++ .../cloudatlas/agent/modules/GossipGirlState.java | 35 ++++++++++++++++++-- 2 files changed, 70 insertions(+), 2 deletions(-) (limited to 'src/main/java/pl/edu/mimuw/cloudatlas/agent/modules') diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirl.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirl.java index e7bd227..0c02f1f 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirl.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirl.java @@ -20,6 +20,7 @@ import pl.edu.mimuw.cloudatlas.agent.messages.UpdateQueriesMessage; import pl.edu.mimuw.cloudatlas.model.Attribute; import pl.edu.mimuw.cloudatlas.model.AttributesMap; import pl.edu.mimuw.cloudatlas.model.PathName; +import pl.edu.mimuw.cloudatlas.model.ValueContact; import pl.edu.mimuw.cloudatlas.model.ValueQuery; import pl.edu.mimuw.cloudatlas.model.ValueTime; import pl.edu.mimuw.cloudatlas.model.ZMI; @@ -38,6 +39,9 @@ public class GossipGirl extends Module { case INITIATE: initiateGossip((InitiateGossipMessage) message); break; + case HEJKA: + receiveGossip((HejkaMessage) message); + break; case NO_CO_TAM: handleNoCoTam((NoCoTamMessage) message); break; @@ -71,6 +75,23 @@ public class GossipGirl extends Module { sendMessage(getState); } + private void receiveGossip(HejkaMessage message) throws InterruptedException { + Long gossipId = nextGossipId; + nextGossipId++; + gossipStates.put(gossipId, new GossipGirlState( + gossipId, + message.getReceiverPath(), + new ValueContact(message.getSenderPath(), message.getSenderAddress()), + false + ) + ); + + gossipStates.get(gossipId).handleHejka(message); + + GetStateMessage getState = new GetStateMessage("", 0, ModuleType.GOSSIP, gossipId); + sendMessage(getState); + } + private void setState(StateMessage message) throws InterruptedException { GossipGirlState state = gossipStates.get(message.getRequestId()); if (state != null) { @@ -80,12 +101,28 @@ public class GossipGirl extends Module { "", 0, state.gossipId, + state.ourPath, + state.theirContact.getName(), state.getZoneTimestampsToSend(), state.getQueryTimestampsToSend() ); UDUPMessage udupMessage = new UDUPMessage("", 0, state.theirContact, hejka); sendMessage(udupMessage); state.sentHejka(); + } else if (state.state == GossipGirlState.State.SEND_NO_CO_TAM) { + NoCoTamMessage noCoTam = new NoCoTamMessage( + "", + 0, + state.gossipId, + state.theirGossipId, + state.getZoneTimestampsToSend(), + state.getQueryTimestampsToSend(), + state.hejkaSendTimestamp, + state.hejkaReceiveTimestamp + ); + UDUPMessage udupMessage = new UDUPMessage("", 0, state.theirContact, noCoTam); + sendMessage(udupMessage); + state.sentNoCoTam(); } } else { System.out.println("ERROR: GossipGirl got state for a nonexistent gossip"); diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirlState.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirlState.java index 8b0711e..1629914 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirlState.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirlState.java @@ -10,6 +10,7 @@ import java.util.Map; import java.util.Map.Entry; import java.util.Set; +import pl.edu.mimuw.cloudatlas.agent.messages.HejkaMessage; import pl.edu.mimuw.cloudatlas.agent.messages.NoCoTamMessage; import pl.edu.mimuw.cloudatlas.model.Attribute; import pl.edu.mimuw.cloudatlas.model.PathName; @@ -22,8 +23,10 @@ import pl.edu.mimuw.cloudatlas.model.ZMI; public class GossipGirlState { public enum State { WAIT_FOR_STATE_INITIALIZER, + APPLY_HEJKA, WAIT_FOR_STATE_RESPONDER, SEND_HEJKA, + SEND_NO_CO_TAM, SEND_INFO, WAIT_FOR_NO_CO_TAM, WAIT_FOR_FIRST_INFO, @@ -57,7 +60,7 @@ public class GossipGirlState { if (initiating) { state = State.WAIT_FOR_STATE_INITIALIZER; } else { - state = State.WAIT_FOR_STATE_RESPONDER; + state = State.APPLY_HEJKA; } } @@ -71,7 +74,7 @@ public class GossipGirlState { case WAIT_FOR_STATE_RESPONDER: this.hierarchy = hierarchy; this.queries = queries; - state = State.WAIT_FOR_FIRST_INFO; + state = State.SEND_NO_CO_TAM; break; default: System.out.println("ERROR: tried to set gossip state when not expected"); @@ -90,6 +93,34 @@ public class GossipGirlState { } } + public void sentNoCoTam() { + switch (state) { + case SEND_NO_CO_TAM: + state = state.WAIT_FOR_FIRST_INFO; + break; + default: + System.out.println("ERROR: tried to set gossip state when not expected"); + state = State.ERROR; + } + } + + public void handleHejka(HejkaMessage message) { + switch (state) { + case APPLY_HEJKA: + System.out.println("setting sender gossip id to " + message.getSenderGossipId()); + theirGossipId = message.getSenderGossipId(); + theirZoneTimestamps = message.getZoneTimestamps(); + theirQueryTimestamps = message.getQueryTimestamps(); + hejkaSendTimestamp = message.getSentTimestamp(); + hejkaReceiveTimestamp = message.getReceivedTimestamp(); + state = State.WAIT_FOR_STATE_RESPONDER; + break; + default: + System.out.println("ERROR: tried to set gossip state when not expected"); + state = State.ERROR; + } + } + public void handleNoCoTam(NoCoTamMessage message) { switch (state) { case WAIT_FOR_NO_CO_TAM: -- cgit v1.2.3 From 0fca21f2011958d709a25aa1f4d863c1d646da6e Mon Sep 17 00:00:00 2001 From: Marcin Chrzanowski Date: Sat, 11 Jan 2020 12:08:02 +0100 Subject: Send info back from responder --- .../mimuw/cloudatlas/agent/modules/GossipGirl.java | 33 +++++++++++++--------- .../cloudatlas/agent/modules/GossipGirlState.java | 25 ++++++++++++++-- 2 files changed, 42 insertions(+), 16 deletions(-) (limited to 'src/main/java/pl/edu/mimuw/cloudatlas/agent/modules') diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirl.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirl.java index 0c02f1f..dd8f0b7 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirl.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirl.java @@ -133,27 +133,34 @@ public class GossipGirl extends Module { GossipGirlState state = gossipStates.get(message.getReceiverGossipId()); if (state != null) { state.handleNoCoTam(message); - for (ZMI zmi : state.getZMIsToSend()) { - AttributesMessage attributesMessage = new AttributesMessage("", 0, zmi.getPathName(), zmi.getAttributes(), state.theirGossipId); - UDUPMessage udupMessage = new UDUPMessage("", 0, state.theirContact, attributesMessage); - sendMessage(udupMessage); - } - - for (Entry query : state.getQueriesToSend()) { - QueryMessage queryMessage = new QueryMessage("", 0, query.getKey(), query.getValue(), state.theirGossipId); - UDUPMessage udupMessage = new UDUPMessage("", 0, state.theirContact, queryMessage); - sendMessage(udupMessage); - } - state.sentInfo(); + sendInfo(state); } else { System.out.println("ERROR: GossipGirl got state for a nonexistent gossip"); } } + private void sendInfo(GossipGirlState state) throws InterruptedException { + for (ZMI zmi : state.getZMIsToSend()) { + AttributesMessage attributesMessage = new AttributesMessage("", 0, zmi.getPathName(), zmi.getAttributes(), state.theirGossipId); + UDUPMessage udupMessage = new UDUPMessage("", 0, state.theirContact, attributesMessage); + sendMessage(udupMessage); + } + + for (Entry query : state.getQueriesToSend()) { + QueryMessage queryMessage = new QueryMessage("", 0, query.getKey(), query.getValue(), state.theirGossipId); + UDUPMessage udupMessage = new UDUPMessage("", 0, state.theirContact, queryMessage); + sendMessage(udupMessage); + } + state.sentInfo(); + } + private void handleAttributes(AttributesMessage message) throws InterruptedException { GossipGirlState state = gossipStates.get(message.getReceiverGossipId()); if (state != null) { - state.gotAttributesFor(message.getPath()); + state.gotAttributes(message); + if (state.state == GossipGirlState.State.SEND_INFO) { + sendInfo(state); + } UpdateAttributesMessage updateMessage = new UpdateAttributesMessage("", 0, message.getPath().toString(), message.getAttributes()); sendMessage(updateMessage); if (state.state == GossipGirlState.State.FINISHED) { diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirlState.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirlState.java index 1629914..eafbcca 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirlState.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirlState.java @@ -10,6 +10,7 @@ import java.util.Map; import java.util.Map.Entry; import java.util.Set; +import pl.edu.mimuw.cloudatlas.agent.messages.AttributesMessage; import pl.edu.mimuw.cloudatlas.agent.messages.HejkaMessage; import pl.edu.mimuw.cloudatlas.agent.messages.NoCoTamMessage; import pl.edu.mimuw.cloudatlas.model.Attribute; @@ -28,6 +29,7 @@ public class GossipGirlState { SEND_HEJKA, SEND_NO_CO_TAM, SEND_INFO, + SEND_INFO_AND_FINISH, WAIT_FOR_NO_CO_TAM, WAIT_FOR_FIRST_INFO, WAIT_FOR_INFO, @@ -107,7 +109,6 @@ public class GossipGirlState { public void handleHejka(HejkaMessage message) { switch (state) { case APPLY_HEJKA: - System.out.println("setting sender gossip id to " + message.getSenderGossipId()); theirGossipId = message.getSenderGossipId(); theirZoneTimestamps = message.getZoneTimestamps(); theirQueryTimestamps = message.getQueryTimestamps(); @@ -262,16 +263,19 @@ public class GossipGirlState { case SEND_INFO: state = State.WAIT_FOR_INFO; break; + case SEND_INFO_AND_FINISH: + state = State.FINISHED; + break; default: System.out.println("ERROR: tried to set gossip state when not expected"); state = State.ERROR; } } - public void gotAttributesFor(PathName path) { + public void gotAttributes(AttributesMessage message) { switch (state) { case WAIT_FOR_INFO: - if (!waitingForZones.remove(path)) { + if (!waitingForZones.remove(message.getPath())) { System.out.println("DEBUG: got zone attributes we weren't expecting"); } if (waitingForZones.isEmpty() && waitingForQueries.isEmpty()) { @@ -279,6 +283,21 @@ public class GossipGirlState { state = state.FINISHED; } break; + case WAIT_FOR_FIRST_INFO: + // TODO: use offset to setup GTP + setZonesToSend(); + setQueriesToSend(); + setWaitingFor(); + state = State.SEND_INFO; + + if (!waitingForZones.remove(message.getPath())) { + System.out.println("DEBUG: got zone attributes we weren't expecting"); + } + if (waitingForZones.isEmpty() && waitingForQueries.isEmpty()) { + System.out.println("INFO: done waiting for info"); + state = state.SEND_INFO_AND_FINISH; + } + break; default: System.out.println("ERROR: got attributes when not expected"); state = State.ERROR; -- cgit v1.2.3